test_enhanced.py

#!/usr/bin/env python3
"""
Modern Test Suite for LinkedIn Profile Analyzer
Tests all components: modern scraping, agents, and frontend integration
"""

import os
import sys
import json
import time
from datetime import datetime
from typing import Dict, List

# Add current directory to path for imports
sys.path.append(os.path.dirname(os.path.abspath(__file__)))

from dotenv import load_dotenv
load_dotenv()

# Import our modern modules
from agent_modern import ModernLinkedInAgent, generate_profile_summary_and_facts_single_step
from scraper_modern import scrape_linkedin_profile_modern, test_modern_scraper
from cache import get as cache_get, set as cache_set, clear_cache

class ModernTestSuite:
    """Comprehensive test suite for the modern LinkedIn analyzer"""
    
    def __init__(self):
        self.results = []
        self.start_time = None
        
    def log_test(self, test_name: str, success: bool, message: str = "", data: Dict = None):
        """Log test results"""
        result = {
            "test": test_name,
            "success": success,
            "message": message,
            "timestamp": datetime.now().isoformat(),
            "data": data
        }
        self.results.append(result)
        
        status = "✅ PASS" if success else "❌ FAIL"
        print(f"{status} {test_name}: {message}")
        
    def run_all_tests(self):
        """Run all tests in the suite"""
        print("🧪 Starting Modern LinkedIn Analyzer Test Suite")
        print("=" * 60)
        self.start_time = time.time()
        
        # Test 1: Environment Setup
        self.test_environment()
        
        # Test 2: Cache System
        self.test_cache_system()
        
        # Test 3: Modern Scraper
        self.test_modern_scraper()
        
        # Test 4: LinkedIn URL Tool
        self.test_linkedin_url_tool()
        
        # Test 5: Agent Analysis
        self.test_agent_analysis()
        
        # Test 6: Full Integration
        self.test_full_integration()
        
        # Test 7: Performance
        self.test_performance()
        
        # Test 8: Error Handling
        self.test_error_handling()
        
        # Generate report
        self.generate_report()
        
    def test_environment(self):
        """Test environment setup and API keys"""
        try:
            # Check required environment variables
            required_vars = ["OPENAI_API_KEY", "TAVILY_API_KEY"]
            missing_vars = []
            
            for var in required_vars:
                if not os.environ.get(var):
                    missing_vars.append(var)
            
            if missing_vars:
                self.log_test(
                    "Environment Setup",
                    False,
                    f"Missing environment variables: {', '.join(missing_vars)}"
                )
            else:
                self.log_test(
                    "Environment Setup",
                    True,
                    "All required environment variables are set"
                )
                
        except Exception as e:
            self.log_test("Environment Setup", False, f"Error: {str(e)}")
    
    def test_cache_system(self):
        """Test caching functionality"""
        try:
            # Test cache operations
            test_key = "test_profile_url"
            test_data = {"name": "Test User", "timestamp": datetime.now().isoformat()}
            
            # Clear any existing cache
            clear_cache()
            
            # Test cache miss
            cached_data, age, from_cache = cache_get(test_key)
            if cached_data is None:
                self.log_test("Cache Miss", True, "Cache correctly returns None for non-existent key")
            else:
                self.log_test("Cache Miss", False, "Cache should return None for non-existent key")
                return
            
            # Test cache set
            cache_set(test_key, test_data)
            
            # Test cache hit
            cached_data, age, from_cache = cache_get(test_key)
            if cached_data and cached_data.get("name") == "Test User":
                self.log_test("Cache Hit", True, f"Cache correctly retrieved data (age: {age} min)")
            else:
                self.log_test("Cache Hit", False, "Cache failed to retrieve stored data")
                
        except Exception as e:
            self.log_test("Cache System", False, f"Error: {str(e)}")
    
    def test_modern_scraper(self):
        """Test modern scraper functionality"""
        try:
            test_url = "https://www.linkedin.com/in/williamhgates/"
            
            print("🔍 Testing modern scraper...")
            result = scrape_linkedin_profile_modern(test_url)
            
            if result and isinstance(result, dict):
                scraping_info = result.get("_scraping_info", {})
                method = scraping_info.get("method", "unknown")
                success = scraping_info.get("success", False)
                
                if success:
                    self.log_test(
                        "Modern Scraper",
                        True,
                        f"Successfully scraped using method: {method}",
                        {"method": method, "name": result.get("full_name", "N/A")}
                    )
                else:
                    self.log_test(
                        "Modern Scraper",
                        False,
                        f"Scraping failed with method: {method}"
                    )
            else:
                self.log_test("Modern Scraper", False, "Invalid response format")
                
        except Exception as e:
            self.log_test("Modern Scraper", False, f"Error: {str(e)}")
    
    def test_linkedin_url_tool(self):
        """Test LinkedIn URL search tool"""
        try:
            from linkedin_url import linkedin_url
            
            test_name = "Satya Nadella"
            
            print(f"🔍 Testing LinkedIn URL search for: {test_name}")
            result = linkedin_url.func(test_name)
            
            if result and isinstance(result, str):
                if "linkedin.com" in result.lower():
                    self.log_test(
                        "LinkedIn URL Tool",
                        True,
                        f"Found LinkedIn URL: {result[:50]}...",
                        {"url": result}
                    )
                else:
                    self.log_test(
                        "LinkedIn URL Tool",
                        False,
                        "No LinkedIn URL found in search results"
                    )
            else:
                self.log_test("LinkedIn URL Tool", False, "Invalid search response")
                
        except Exception as e:
            self.log_test("LinkedIn URL Tool", False, f"Error: {str(e)}")
    
    def test_agent_analysis(self):
        """Test AI agent analysis"""
        try:
            test_name = "Tim Cook"
            
            print(f"🤖 Testing agent analysis for: {test_name}")
            
            # Test modern agent
            agent = ModernLinkedInAgent()
            result = agent.analyze_profile(test_name)
            
            if result and isinstance(result, dict):
                full_name = result.get("full_name", "")
                has_summary = bool(result.get("summary", ""))
                has_facts = bool(result.get("interesting_facts", []))
                
                self.log_test(
                    "Agent Analysis",
                    True,
                    f"Analyzed: {full_name}, Summary: {has_summary}, Facts: {has_facts}",
                    {"name": full_name, "has_summary": has_summary, "has_facts": has_facts}
                )
            else:
                self.log_test("Agent Analysis", False, "Invalid agent response")
                
        except Exception as e:
            self.log_test("Agent Analysis", False, f"Error: {str(e)}")
    
    def test_full_integration(self):
        """Test full integration with all components"""
        try:
            test_cases = [
                "Elon Musk",
                "Sundar Pichai",
                "Reid Hoffman"
            ]
            
            successful_tests = 0
            
            for test_case in test_cases:
                try:
                    print(f"🔄 Full integration test: {test_case}")
                    
                    # Use the main analysis function
                    result_json = generate_profile_summary_and_facts_single_step(test_case)
                    result = json.loads(result_json) if isinstance(result_json, str) else result_json
                    
                    if result and isinstance(result, dict) and not result.get("error"):
                        successful_tests += 1
                        
                except Exception as e:
                    print(f"   ❌ Failed: {str(e)}")
            
            success_rate = (successful_tests / len(test_cases)) * 100
            
            self.log_test(
                "Full Integration",
                successful_tests > 0,
                f"Success rate: {success_rate:.1f}% ({successful_tests}/{len(test_cases)})",
                {"success_rate": success_rate, "successful_tests": successful_tests}
            )
            
        except Exception as e:
            self.log_test("Full Integration", False, f"Error: {str(e)}")
    
    def test_performance(self):
        """Test performance metrics"""
        try:
            test_name = "Reid Hoffman"
            
            print(f"⚡ Performance test for: {test_name}")
            
            start_time = time.time()
            
            # Run analysis
            result_json = generate_profile_summary_and_facts_single_step(test_name)
            result = json.loads(result_json) if isinstance(result_json, str) else result_json
            
            end_time = time.time()
            duration = end_time - start_time
            
            # Performance thresholds
            excellent_threshold = 30  # seconds
            good_threshold = 60      # seconds
            
            if duration < excellent_threshold:
                performance_rating = "Excellent"
            elif duration < good_threshold:
                performance_rating = "Good"
            else:
                performance_rating = "Needs Improvement"
            
            self.log_test(
                "Performance",
                duration < good_threshold,
                f"Analysis completed in {duration:.2f}s - {performance_rating}",
                {"duration": duration, "rating": performance_rating}
            )
            
        except Exception as e:
            self.log_test("Performance", False, f"Error: {str(e)}")
    
    def test_error_handling(self):
        """Test error handling with invalid inputs"""
        try:
            test_cases = [
                "",  # Empty string
                "NonExistentPerson12345",  # Non-existent person
                "https://invalid-url.com",  # Invalid URL
            ]
            
            handled_errors = 0
            
            for test_case in test_cases:
                try:
                    print(f"🔄 Error handling test: '{test_case}'")
                    
                    result_json = generate_profile_summary_and_facts_single_step(test_case)
                    result = json.loads(result_json) if isinstance(result_json, str) else result_json
                    
                    # Should handle gracefully without crashing
                    if result and isinstance(result, dict):
                        handled_errors += 1
                        
                except Exception as e:
                    print(f"   ❌ Unhandled error: {str(e)}")
            
            success_rate = (handled_errors / len(test_cases)) * 100
            
            self.log_test(
                "Error Handling",
                handled_errors == len(test_cases),
                f"Handled {handled_errors}/{len(test_cases)} error cases gracefully",
                {"success_rate": success_rate}
            )
            
        except Exception as e:
            self.log_test("Error Handling", False, f"Error: {str(e)}")
    
    def generate_report(self):
        """Generate comprehensive test report"""
        total_time = time.time() - self.start_time
        
        print("\n" + "=" * 60)
        print("📊 MODERN TEST SUITE REPORT")
        print("=" * 60)
        
        # Summary statistics
        total_tests = len(self.results)
        passed_tests = sum(1 for r in self.results if r["success"])
        failed_tests = total_tests - passed_tests
        success_rate = (passed_tests / total_tests) * 100 if total_tests > 0 else 0
        
        print(f"📈 Total Tests: {total_tests}")
        print(f"✅ Passed: {passed_tests}")
        print(f"❌ Failed: {failed_tests}")
        print(f"📊 Success Rate: {success_rate:.1f}%")
        print(f"⏱️  Total Time: {total_time:.2f}s")
        
        # Detailed results
        print(f"\n📋 Detailed Results:")
        print("-" * 60)
        
        for result in self.results:
            status = "✅" if result["success"] else "❌"
            print(f"{status} {result['test']}: {result['message']}")
        
        # System status
        print(f"\n🔧 System Status:")
        print("-" * 60)
        
        if failed_tests == 0:
            print("🎉 All tests passed! Your modern LinkedIn analyzer is working perfectly.")
            print("✨ Features available:")
            print("   • Modern multi-method scraping")
            print("   • AI-powered profile analysis")
            print("   • Intelligent caching system")
            print("   • Robust error handling")
        else:
            print("⚠️  Some tests failed. Check the following:")
            
            for result in self.results:
                if not result["success"]:
                    print(f"   • {result['test']}: {result['message']}")
        
        # Save report to file
        report_data = {
            "timestamp": datetime.now().isoformat(),
            "version": "modern_v1.0",
            "summary": {
                "total_tests": total_tests,
                "passed_tests": passed_tests,
                "failed_tests": failed_tests,
                "success_rate": success_rate,
                "total_time": total_time
            },
            "results": self.results
        }
        
        with open("test_report_modern.json", "w") as f:
            json.dump(report_data, f, indent=2)
        
        print(f"\n💾 Report saved to: test_report_modern.json")

def main():
    """Main test runner"""
    print("🚀 Modern LinkedIn Profile Analyzer - Test Suite")
    print("🔧 Testing modern scraping, AI agents, and integrations...")
    print()
    
    # Run the test suite
    suite = ModernTestSuite()
    suite.run_all_tests()

if __name__ == "__main__":
    main()