tier1_api_validation.py

#!/usr/bin/env python3
"""
Tier 1 API Validation - Real API Testing
Tests the 5 Tier 1 APIs with different documentation quality levels
"""

import asyncio
import json
import os
import time
import aiohttp
import logging
from datetime import datetime
from typing import Dict, List, Optional
from dataclasses import dataclass, asdict
from dotenv import load_dotenv

# Load environment variables
load_dotenv()

# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

@dataclass
class Tier1APIResult:
    """Results from Tier 1 API testing"""
    test_id: str
    api_name: str
    quality_level: str
    
    # API response metrics
    api_response_time: float
    api_status_code: int
    authentication_success: bool
    data_received: bool
    expected_fields_present: bool
    
    # Success score
    success_score: float
    
    # Sample data
    sample_data: Dict
    
    # Error info
    error_message: Optional[str] = None

class Tier1APIValidator:
    """Validates Tier 1 APIs for documentation quality study"""
    
    def __init__(self):
        self.results = []
        
        # Tier 1 API configurations
        self.apis = {
            "weather_average": {
                "name": "OpenWeatherMap Current",
                "quality_level": "3.0",
                "url": "https://api.openweathermap.org/data/2.5/weather",
                "params": {"q": "London", "appid": os.getenv("OPENWEATHER_API_KEY")},
                "headers": {},
                "expected_fields": ["main", "weather", "name"],
                "env_key": "OPENWEATHER_API_KEY"
            },
            "weather_good": {
                "name": "WeatherAPI.com",
                "quality_level": "4.0",
                "url": "https://api.weatherapi.com/v1/current.json",
                "params": {"key": os.getenv("WEATHERAPI_KEY"), "q": "London"},
                "headers": {},
                "expected_fields": ["current", "location"],
                "env_key": "WEATHERAPI_KEY"
            },
            "news_excellent": {
                "name": "NewsAPI.org",
                "quality_level": "5.0",
                "url": "https://newsapi.org/v2/top-headlines",
                "params": {"category": "technology", "pageSize": "5"},
                "headers": {"X-API-Key": os.getenv("NEWS_API_KEY")},
                "expected_fields": ["articles", "totalResults"],
                "env_key": "NEWS_API_KEY"
            },
            "news_good": {
                "name": "Guardian API",
                "quality_level": "4.0",
                "url": "https://content.guardianapis.com/search",
                "params": {"api-key": os.getenv("GUARDIAN_API_KEY"), "q": "technology", "page-size": "5"},
                "headers": {},
                "expected_fields": ["response"],
                "env_key": "GUARDIAN_API_KEY"
            },
            "currency_excellent": {
                "name": "Fixer.io",
                "quality_level": "5.0",
                "url": "http://data.fixer.io/api/latest",
                "params": {"access_key": os.getenv("FIXER_API_KEY")},
                "headers": {},
                "expected_fields": ["success", "rates", "base"],
                "env_key": "FIXER_API_KEY"
            }
        }
    
    async def run_tier1_validation(self) -> List[Tier1APIResult]:
        """Run validation tests on all Tier 1 APIs"""
        
        logger.info("🚀 Starting Tier 1 API Validation Study")
        logger.info("=" * 60)
        
        # Check API keys
        missing_keys = self._check_api_keys()
        if missing_keys:
            logger.error(f"❌ Missing API keys: {missing_keys}")
            return []
        
        # Test each API
        test_id = 0
        async with aiohttp.ClientSession() as session:
            for api_key, api_config in self.apis.items():
                test_id += 1
                
                logger.info(f"\n🔧 Testing {api_config['name']} (Quality: {api_config['quality_level']})")
                
                result = await self._test_single_api(session, str(test_id), api_key, api_config)
                self.results.append(result)
                
                # Rate limiting delay
                await asyncio.sleep(2)
        
        # Generate summary
        self._generate_summary()
        
        # Save results
        self._save_results()
        
        return self.results
    
    def _check_api_keys(self) -> List[str]:
        """Check if all required API keys are present"""
        missing = []
        for api_key, api_config in self.apis.items():
            env_key = api_config['env_key']
            if not os.getenv(env_key) or os.getenv(env_key) == f"your_{env_key.lower()}_here":
                missing.append(env_key)
        return missing
    
    async def _test_single_api(self, session: aiohttp.ClientSession, test_id: str, 
                              api_key: str, api_config: Dict) -> Tier1APIResult:
        """Test a single API endpoint"""
        
        start_time = time.time()
        
        try:
            # Make API request
            async with session.get(
                api_config['url'],
                params=api_config['params'],
                headers=api_config['headers'],
                timeout=aiohttp.ClientTimeout(total=15)
            ) as response:
                
                response_time = time.time() - start_time
                
                # Check authentication
                authentication_success = response.status != 401
                
                if response.status == 200:
                    # Parse response
                    try:
                        data = await response.json()
                        data_received = True
                        
                        # Check expected fields
                        expected_fields_present = self._check_expected_fields(
                            data, api_config['expected_fields']
                        )
                        
                        # Calculate success score
                        success_metrics = [
                            authentication_success,
                            response.status == 200,
                            data_received,
                            expected_fields_present
                        ]
                        success_score = sum(success_metrics) / len(success_metrics)
                        
                        # Extract sample data
                        sample_data = self._extract_sample_data(data, api_key)
                        
                        logger.info(f"   ✅ SUCCESS: {response_time:.2f}s, Score: {success_score:.2f}")
                        
                        return Tier1APIResult(
                            test_id=test_id,
                            api_name=api_config['name'],
                            quality_level=api_config['quality_level'],
                            api_response_time=response_time,
                            api_status_code=response.status,
                            authentication_success=authentication_success,
                            data_received=data_received,
                            expected_fields_present=expected_fields_present,
                            success_score=success_score,
                            sample_data=sample_data
                        )
                        
                    except json.JSONDecodeError:
                        logger.error(f"   ❌ Invalid JSON response")
                        return self._create_error_result(test_id, api_config, response_time, 
                                                       response.status, "Invalid JSON response")
                
                else:
                    # Handle error responses
                    error_text = await response.text()
                    logger.error(f"   ❌ HTTP {response.status}: {error_text[:100]}")
                    
                    return self._create_error_result(test_id, api_config, response_time,
                                                   response.status, f"HTTP {response.status}")
                    
        except asyncio.TimeoutError:
            response_time = time.time() - start_time
            logger.error(f"   ❌ Timeout after {response_time:.1f}s")
            return self._create_error_result(test_id, api_config, response_time, 0, "Timeout")
            
        except Exception as e:
            response_time = time.time() - start_time
            logger.error(f"   ❌ Error: {str(e)}")
            return self._create_error_result(test_id, api_config, response_time, 0, str(e))
    
    def _check_expected_fields(self, data: Dict, expected_fields: List[str]) -> bool:
        """Check if expected fields are present in response"""
        for field in expected_fields:
            if field not in data:
                return False
        return True
    
    def _extract_sample_data(self, data: Dict, api_key: str) -> Dict:
        """Extract sample data for verification"""
        if "weather_average" in api_key:
            return {
                "city": data.get("name"),
                "temperature": data.get("main", {}).get("temp"),
                "weather": data.get("weather", [{}])[0].get("description")
            }
        elif "weather_good" in api_key:
            return {
                "city": data.get("location", {}).get("name"),
                "temperature": data.get("current", {}).get("temp_c"),
                "condition": data.get("current", {}).get("condition", {}).get("text")
            }
        elif "news_excellent" in api_key:
            articles = data.get("articles", [])
            return {
                "total_results": data.get("totalResults"),
                "first_headline": articles[0].get("title") if articles else None
            }
        elif "news_good" in api_key:
            response_data = data.get("response", {})
            results = response_data.get("results", [])
            return {
                "total_results": response_data.get("total"),
                "first_headline": results[0].get("webTitle") if results else None
            }
        elif "currency_excellent" in api_key:
            return {
                "success": data.get("success"),
                "base_currency": data.get("base"),
                "usd_rate": data.get("rates", {}).get("USD")
            }
        
        return {"raw_data_keys": list(data.keys())[:5]}
    
    def _create_error_result(self, test_id: str, api_config: Dict, response_time: float,
                           status_code: int, error_message: str) -> Tier1APIResult:
        """Create error result"""
        return Tier1APIResult(
            test_id=test_id,
            api_name=api_config['name'],
            quality_level=api_config['quality_level'],
            api_response_time=response_time,
            api_status_code=status_code,
            authentication_success=status_code != 401,
            data_received=False,
            expected_fields_present=False,
            success_score=0.0,
            sample_data={},
            error_message=error_message
        )
    
    def _generate_summary(self):
        """Generate test summary"""
        logger.info("\n" + "=" * 60)
        logger.info("📊 TIER 1 API VALIDATION SUMMARY")
        logger.info("=" * 60)
        
        successful_tests = [r for r in self.results if r.success_score > 0.5]
        failed_tests = [r for r in self.results if r.success_score <= 0.5]
        
        logger.info(f"✅ Successful APIs: {len(successful_tests)}/{len(self.results)}")
        logger.info(f"❌ Failed APIs: {len(failed_tests)}/{len(self.results)}")
        
        if successful_tests:
            logger.info(f"\n🎉 Working APIs:")
            for result in successful_tests:
                logger.info(f"   • {result.api_name} (Quality {result.quality_level}): "
                          f"{result.success_score:.2f} score, {result.api_response_time:.2f}s")
        
        if failed_tests:
            logger.info(f"\n⚠️  Failed APIs:")
            for result in failed_tests:
                logger.info(f"   • {result.api_name}: {result.error_message}")
        
        # Quality level analysis
        logger.info(f"\n📈 Quality Level Performance:")
        quality_scores = {}
        for result in self.results:
            quality = result.quality_level
            if quality not in quality_scores:
                quality_scores[quality] = []
            quality_scores[quality].append(result.success_score)
        
        for quality in sorted(quality_scores.keys()):
            avg_score = sum(quality_scores[quality]) / len(quality_scores[quality])
            logger.info(f"   • Quality {quality}: {avg_score:.2f} average score")
        
        # Check for sweet spot pattern
        if len(quality_scores) >= 3:
            excellent_score = quality_scores.get("5.0", [0])[0]
            good_scores = [score for q, scores in quality_scores.items() 
                          if q in ["4.0", "3.0"] for score in scores]
            
            if good_scores and max(good_scores) > excellent_score:
                logger.info(f"\n🎯 SWEET SPOT PATTERN DETECTED!")
                logger.info(f"   Average/Good documentation outperforming Excellent!")
                logger.info(f"   Excellent (5.0): {excellent_score:.2f}")
                logger.info(f"   Best Good/Average: {max(good_scores):.2f}")
    
    def _save_results(self):
        """Save results to JSON file"""
        results_data = {
            "timestamp": datetime.now().isoformat(),
            "test_type": "tier1_api_validation",
            "total_tests": len(self.results),
            "successful_tests": len([r for r in self.results if r.success_score > 0.5]),
            "results": [asdict(result) for result in self.results]
        }
        
        filename = f"results/tier1_api_validation_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
        os.makedirs("results", exist_ok=True)
        
        with open(filename, "w") as f:
            json.dump(results_data, f, indent=2)
        
        logger.info(f"\n💾 Results saved to: {filename}")

async def main():
    """Main function to run Tier 1 API validation"""
    
    print("🚀 Real API Validation Study - Tier 1 Testing")
    print(f"⏰ Started at: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    
    validator = Tier1APIValidator()
    results = await validator.run_tier1_validation()
    
    print(f"\n🎉 Tier 1 validation completed!")
    print(f"📊 Total tests: {len(results)}")
    print(f"✅ Successful: {len([r for r in results if r.success_score > 0.5])}")
    print(f"⏰ Completed at: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")

if __name__ == "__main__":
    asyncio.run(main())