test-runner.py

#!/usr/bin/env python3
"""
Comprehensive test runner for SuData project.
Executes all test suites and generates consolidated reports.
"""

import os
import sys
import subprocess
import json
import time
from pathlib import Path
from typing import Dict, List, Any
import argparse

class SuDataTestRunner:
    """Main test runner for SuData project."""
    
    def __init__(self, project_root: Path):
        self.project_root = project_root
        self.results = {}
        self.start_time = time.time()
    
    def run_command(self, command: str, cwd: Path = None) -> Dict[str, Any]:
        """Run a shell command and capture results."""
        print(f"Running: {command}")
        
        try:
            result = subprocess.run(
                command,
                shell=True,
                capture_output=True,
                text=True,
                cwd=cwd or self.project_root,
                timeout=300  # 5 minute timeout
            )
            
            return {
                "success": result.returncode == 0,
                "returncode": result.returncode,
                "stdout": result.stdout,
                "stderr": result.stderr,
                "command": command
            }
        except subprocess.TimeoutExpired:
            return {
                "success": False,
                "returncode": -1,
                "stdout": "",
                "stderr": "Command timed out after 5 minutes",
                "command": command
            }
        except Exception as e:
            return {
                "success": False,
                "returncode": -1,
                "stdout": "",
                "stderr": str(e),
                "command": command
            }
    
    def run_python_unit_tests(self) -> Dict[str, Any]:
        """Run unit tests for all Python services."""
        print("\n" + "="*60)
        print("RUNNING PYTHON UNIT TESTS")
        print("="*60)
        
        services = ["refinery-service", "tiktok-scraper", "youtube-scraper"]
        results = {}
        
        for service in services:
            print(f"\nTesting {service}...")
            service_path = self.project_root / "apps" / service
            
            if not service_path.exists():
                print(f"Warning: {service} directory not found")
                results[service] = {"success": False, "error": "Directory not found"}
                continue
            
            # Run pytest with coverage
            cmd = "pytest tests/ --cov=. --cov-report=json --cov-report=html --cov-report=term -v"
            result = self.run_command(cmd, service_path)
            results[service] = result
            
            # Parse coverage if available
            coverage_file = service_path / "coverage.json"
            if coverage_file.exists():
                try:
                    with open(coverage_file) as f:
                        coverage_data = json.load(f)
                        results[service]["coverage"] = coverage_data.get("totals", {}).get("percent_covered", 0)
                except Exception as e:
                    print(f"Warning: Could not parse coverage for {service}: {e}")
        
        return results
    
    def run_dashboard_tests(self) -> Dict[str, Any]:
        """Run dashboard unit and component tests."""
        print("\n" + "="*60)
        print("RUNNING DASHBOARD TESTS")
        print("="*60)
        
        dashboard_path = self.project_root / "apps" / "dashboard"
        
        if not dashboard_path.exists():
            return {"success": False, "error": "Dashboard directory not found"}
        
        # Run Vitest tests
        result = self.run_command("pnpm test", dashboard_path)
        
        # Try to get coverage information
        coverage_result = self.run_command("pnpm test:coverage", dashboard_path)
        if coverage_result["success"]:
            result["coverage_output"] = coverage_result["stdout"]
        
        return result
    
    def run_integration_tests(self) -> Dict[str, Any]:
        """Run integration tests."""
        print("\n" + "="*60)
        print("RUNNING INTEGRATION TESTS")
        print("="*60)
        
        integration_path = self.project_root / "tests" / "integration"
        
        if not integration_path.exists():
            return {"success": False, "error": "Integration tests directory not found"}
        
        cmd = "pytest tests/integration/ -v --tb=short"
        return self.run_command(cmd)
    
    def run_e2e_tests(self, headless: bool = True) -> Dict[str, Any]:
        """Run end-to-end tests with Playwright."""
        print("\n" + "="*60)
        print("RUNNING END-TO-END TESTS")
        print("="*60)
        
        # Check if Playwright is available
        playwright_check = self.run_command("pnpm playwright --version")
        if not playwright_check["success"]:
            return {"success": False, "error": "Playwright not available"}
        
        # Run E2E tests
        cmd = "pnpm playwright test"
        if not headless:
            cmd += " --headed"
        
        return self.run_command(cmd)
    
    def run_performance_tests(self) -> Dict[str, Any]:
        """Run performance tests."""
        print("\n" + "="*60)
        print("RUNNING PERFORMANCE TESTS")
        print("="*60)
        
        perf_script = self.project_root / "tests" / "performance" / "load_test.py"
        
        if not perf_script.exists():
            return {"success": False, "error": "Performance test script not found"}
        
        cmd = f"python {perf_script}"
        return self.run_command(cmd)
    
    def run_error_scenario_tests(self) -> Dict[str, Any]:
        """Run error scenario and recovery tests."""
        print("\n" + "="*60)
        print("RUNNING ERROR SCENARIO TESTS")
        print("="*60)
        
        error_path = self.project_root / "tests" / "error_scenarios"
        
        if not error_path.exists():
            return {"success": False, "error": "Error scenario tests directory not found"}
        
        cmd = "pytest tests/error_scenarios/ -v --tb=short"
        return self.run_command(cmd)
    
    def generate_report(self):
        """Generate comprehensive test report."""
        print("\n" + "="*80)
        print("TEST EXECUTION SUMMARY")
        print("="*80)
        
        total_time = time.time() - self.start_time
        total_tests = 0
        passed_tests = 0
        
        for test_type, result in self.results.items():
            print(f"\n{test_type.upper()}:")
            
            if isinstance(result, dict):
                if "success" in result:
                    status = "PASSED" if result["success"] else "FAILED"
                    print(f"  Status: {status}")
                    
                    if result["success"]:
                        passed_tests += 1
                    total_tests += 1
                    
                    if "coverage" in result:
                        print(f"  Coverage: {result['coverage']:.1f}%")
                    
                    if not result["success"] and "stderr" in result:
                        print(f"  Error: {result['stderr'][:200]}...")
                else:
                    # Handle nested results (like Python services)
                    for service, service_result in result.items():
                        if isinstance(service_result, dict) and "success" in service_result:
                            status = "PASSED" if service_result["success"] else "FAILED"
                            print(f"  {service}: {status}")
                            
                            if service_result["success"]:
                                passed_tests += 1
                            total_tests += 1
                            
                            if "coverage" in service_result:
                                print(f"    Coverage: {service_result['coverage']:.1f}%")
        
        print(f"\n{'='*80}")
        print(f"OVERALL RESULTS:")
        print(f"Total Test Suites: {total_tests}")
        print(f"Passed: {passed_tests}")
        print(f"Failed: {total_tests - passed_tests}")
        print(f"Success Rate: {(passed_tests/total_tests*100):.1f}%" if total_tests > 0 else "N/A")
        print(f"Total Execution Time: {total_time:.1f} seconds")
        
        # Save results to file
        results_file = self.project_root / "test-results" / f"test_summary_{int(time.time())}.json"
        results_file.parent.mkdir(exist_ok=True)
        
        with open(results_file, 'w') as f:
            json.dump({
                "timestamp": time.time(),
                "total_time": total_time,
                "total_tests": total_tests,
                "passed_tests": passed_tests,
                "success_rate": (passed_tests/total_tests*100) if total_tests > 0 else 0,
                "results": self.results
            }, f, indent=2)
        
        print(f"Detailed results saved to: {results_file}")
        
        return passed_tests == total_tests
    
    def run_all_tests(self, include_performance: bool = False, include_e2e: bool = True,
                     headless: bool = True):
        """Run all test suites."""
        print("Starting SuData Comprehensive Test Suite")
        print(f"Project Root: {self.project_root}")
        
        # Run unit tests
        self.results["python_unit_tests"] = self.run_python_unit_tests()
        self.results["dashboard_tests"] = self.run_dashboard_tests()
        
        # Run integration tests
        self.results["integration_tests"] = self.run_integration_tests()
        
        # Run E2E tests if requested
        if include_e2e:
            self.results["e2e_tests"] = self.run_e2e_tests(headless)
        
        # Run performance tests if requested
        if include_performance:
            self.results["performance_tests"] = self.run_performance_tests()
        
        # Run error scenario tests
        self.results["error_scenario_tests"] = self.run_error_scenario_tests()
        
        # Generate report
        success = self.generate_report()
        
        return success

def main():
    """Main entry point."""
    parser = argparse.ArgumentParser(description="SuData Comprehensive Test Runner")
    parser.add_argument("--include-performance", action="store_true",
                       help="Include performance tests (slower)")
    parser.add_argument("--skip-e2e", action="store_true",
                       help="Skip end-to-end tests")
    parser.add_argument("--headed", action="store_true",
                       help="Run E2E tests with visible browser")
    parser.add_argument("--unit-only", action="store_true",
                       help="Run only unit tests")
    parser.add_argument("--integration-only", action="store_true",
                       help="Run only integration tests")
    
    args = parser.parse_args()
    
    # Find project root
    project_root = Path(__file__).parent.parent
    
    runner = SuDataTestRunner(project_root)
    
    if args.unit_only:
        runner.results["python_unit_tests"] = runner.run_python_unit_tests()
        runner.results["dashboard_tests"] = runner.run_dashboard_tests()
    elif args.integration_only:
        runner.results["integration_tests"] = runner.run_integration_tests()
    else:
        success = runner.run_all_tests(
            include_performance=args.include_performance,
            include_e2e=not args.skip_e2e,
            headless=not args.headed
        )
        
        # Exit with error code if any tests failed
        sys.exit(0 if success else 1)
    
    runner.generate_report()

if __name__ == "__main__":
    main()