scripts/test_clients.py

#!/usr/bin/env python3
"""

Client SDK testing script for FastAPI Video Generation Backend.



This script tests the generated client SDKs to ensure they work correctly

with the API endpoints. It performs basic functionality tests and validates

the client-server communication.

"""

import os
import sys
import json
import time
import asyncio
import requests
import subprocess
from pathlib import Path
from typing import Dict, List, Optional, Any
import tempfile
import logging

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


class ClientTester:
    """Test suite for generated client SDKs."""
    
    def __init__(self, api_url: str = "http://localhost:8000", clients_dir: str = "generated_clients"):
        self.api_url = api_url.rstrip("/")
        self.clients_dir = Path(clients_dir)
        self.test_results: Dict[str, Dict[str, Any]] = {}
        
    def check_api_availability(self) -> bool:
        """Check if the API server is running and accessible."""
        try:
            logger.info(f"Checking API availability at {self.api_url}")
            response = requests.get(f"{self.api_url}/health", timeout=10)
            response.raise_for_status()
            logger.info("API server is available")
            return True
        except requests.RequestException as e:
            logger.error(f"API server is not available: {e}")
            return False
    
    def test_openapi_spec(self) -> bool:
        """Test if OpenAPI specification is accessible."""
        try:
            logger.info("Testing OpenAPI specification accessibility")
            response = requests.get(f"{self.api_url}/openapi.json", timeout=10)
            response.raise_for_status()
            
            spec = response.json()
            
            # Validate basic OpenAPI structure
            required_fields = ["openapi", "info", "paths"]
            for field in required_fields:
                if field not in spec:
                    logger.error(f"OpenAPI spec missing required field: {field}")
                    return False
            
            logger.info("OpenAPI specification is valid")
            return True
            
        except requests.RequestException as e:
            logger.error(f"Failed to fetch OpenAPI specification: {e}")
            return False
        except json.JSONDecodeError as e:
            logger.error(f"Invalid JSON in OpenAPI specification: {e}")
            return False
    
    def test_typescript_client(self) -> Dict[str, Any]:
        """Test TypeScript client."""
        client_dir = self.clients_dir / "typescript"
        result = {
            "language": "typescript",
            "exists": False,
            "builds": False,
            "has_examples": False,
            "has_docs": False,
            "errors": []
        }
        
        try:
            if not client_dir.exists():
                result["errors"].append("Client directory does not exist")
                return result
            
            result["exists"] = True
            
            # Check for essential files
            essential_files = ["package.json", "README.md"]
            for file in essential_files:
                if not (client_dir / file).exists():
                    result["errors"].append(f"Missing {file}")
            
            # Check for example file
            if (client_dir / "example.ts").exists():
                result["has_examples"] = True
            
            # Check for documentation
            if (client_dir / "README.md").exists():
                result["has_docs"] = True
            
            # Try to build the client
            if (client_dir / "package.json").exists():
                logger.info("Testing TypeScript client build")
                
                # Install dependencies
                install_result = subprocess.run(
                    ["npm", "install"],
                    cwd=client_dir,
                    capture_output=True,
                    text=True,
                    timeout=120
                )
                
                if install_result.returncode != 0:
                    result["errors"].append(f"npm install failed: {install_result.stderr}")
                else:
                    # Try to build
                    build_result = subprocess.run(
                        ["npm", "run", "build"],
                        cwd=client_dir,
                        capture_output=True,
                        text=True,
                        timeout=60
                    )
                    
                    if build_result.returncode == 0:
                        result["builds"] = True
                    else:
                        result["errors"].append(f"Build failed: {build_result.stderr}")
            
        except subprocess.TimeoutExpired:
            result["errors"].append("Build timeout")
        except Exception as e:
            result["errors"].append(f"Unexpected error: {str(e)}")
        
        return result
    
    def test_python_client(self) -> Dict[str, Any]:
        """Test Python client."""
        client_dir = self.clients_dir / "python"
        result = {
            "language": "python",
            "exists": False,
            "installs": False,
            "imports": False,
            "has_examples": False,
            "has_docs": False,
            "errors": []
        }
        
        try:
            if not client_dir.exists():
                result["errors"].append("Client directory does not exist")
                return result
            
            result["exists"] = True
            
            # Check for essential files
            essential_files = ["setup.py", "README.md"]
            for file in essential_files:
                if not (client_dir / file).exists():
                    result["errors"].append(f"Missing {file}")
            
            # Check for example file
            if (client_dir / "example.py").exists():
                result["has_examples"] = True
            
            # Check for documentation
            if (client_dir / "README.md").exists():
                result["has_docs"] = True
            
            # Try to install the client in development mode
            if (client_dir / "setup.py").exists():
                logger.info("Testing Python client installation")
                
                install_result = subprocess.run(
                    [sys.executable, "setup.py", "develop", "--user"],
                    cwd=client_dir,
                    capture_output=True,
                    text=True,
                    timeout=120
                )
                
                if install_result.returncode == 0:
                    result["installs"] = True
                    
                    # Try to import the client
                    try:
                        import importlib.util
                        
                        # Find the main module
                        main_module_path = None
                        for py_file in client_dir.rglob("*.py"):
                            if "__init__.py" in py_file.name:
                                main_module_path = py_file
                                break
                        
                        if main_module_path:
                            spec = importlib.util.spec_from_file_location("test_client", main_module_path)
                            if spec and spec.loader:
                                module = importlib.util.module_from_spec(spec)
                                spec.loader.exec_module(module)
                                result["imports"] = True
                        
                    except Exception as e:
                        result["errors"].append(f"Import failed: {str(e)}")
                else:
                    result["errors"].append(f"Installation failed: {install_result.stderr}")
            
        except subprocess.TimeoutExpired:
            result["errors"].append("Installation timeout")
        except Exception as e:
            result["errors"].append(f"Unexpected error: {str(e)}")
        
        return result
    
    def test_java_client(self) -> Dict[str, Any]:
        """Test Java client."""
        client_dir = self.clients_dir / "java"
        result = {
            "language": "java",
            "exists": False,
            "compiles": False,
            "has_examples": False,
            "has_docs": False,
            "errors": []
        }
        
        try:
            if not client_dir.exists():
                result["errors"].append("Client directory does not exist")
                return result
            
            result["exists"] = True
            
            # Check for essential files
            essential_files = ["pom.xml", "README.md"]
            for file in essential_files:
                if not (client_dir / file).exists():
                    result["errors"].append(f"Missing {file}")
            
            # Check for example file
            if (client_dir / "example.java").exists():
                result["has_examples"] = True
            
            # Check for documentation
            if (client_dir / "README.md").exists():
                result["has_docs"] = True
            
            # Try to compile the client (if Maven is available)
            if (client_dir / "pom.xml").exists():
                logger.info("Testing Java client compilation")
                
                # Check if Maven is available
                maven_check = subprocess.run(
                    ["mvn", "--version"],
                    capture_output=True,
                    text=True,
                    timeout=10
                )
                
                if maven_check.returncode == 0:
                    compile_result = subprocess.run(
                        ["mvn", "compile"],
                        cwd=client_dir,
                        capture_output=True,
                        text=True,
                        timeout=180
                    )
                    
                    if compile_result.returncode == 0:
                        result["compiles"] = True
                    else:
                        result["errors"].append(f"Compilation failed: {compile_result.stderr}")
                else:
                    result["errors"].append("Maven not available for compilation test")
            
        except subprocess.TimeoutExpired:
            result["errors"].append("Compilation timeout")
        except Exception as e:
            result["errors"].append(f"Unexpected error: {str(e)}")
        
        return result
    
    def test_client_structure(self, language: str) -> Dict[str, Any]:
        """Test the structure of a generated client."""
        client_dir = self.clients_dir / language
        result = {
            "language": language,
            "exists": False,
            "structure_valid": False,
            "has_api_files": False,
            "has_model_files": False,
            "has_docs": False,
            "file_count": 0,
            "errors": []
        }
        
        try:
            if not client_dir.exists():
                result["errors"].append("Client directory does not exist")
                return result
            
            result["exists"] = True
            
            # Count files
            all_files = list(client_dir.rglob("*"))
            result["file_count"] = len([f for f in all_files if f.is_file()])
            
            # Check for API files
            api_files = list(client_dir.rglob("*api*"))
            if api_files:
                result["has_api_files"] = True
            
            # Check for model files
            model_files = list(client_dir.rglob("*model*"))
            if model_files:
                result["has_model_files"] = True
            
            # Check for documentation
            doc_files = list(client_dir.rglob("README*")) + list(client_dir.rglob("*.md"))
            if doc_files:
                result["has_docs"] = True
            
            # Basic structure validation
            if result["has_api_files"] and result["has_model_files"]:
                result["structure_valid"] = True
            
        except Exception as e:
            result["errors"].append(f"Unexpected error: {str(e)}")
        
        return result
    
    def run_functional_tests(self) -> Dict[str, Any]:
        """Run functional tests against the API using generated clients."""
        result = {
            "api_reachable": False,
            "openapi_valid": False,
            "endpoints_tested": 0,
            "endpoints_passed": 0,
            "errors": []
        }
        
        try:
            # Test API availability
            result["api_reachable"] = self.check_api_availability()
            
            # Test OpenAPI spec
            result["openapi_valid"] = self.test_openapi_spec()
            
            if not result["api_reachable"]:
                result["errors"].append("API not reachable - skipping functional tests")
                return result
            
            # Test basic endpoints
            endpoints_to_test = [
                {"method": "GET", "path": "/health", "expected_status": 200},
                {"method": "GET", "path": "/", "expected_status": 200},
                {"method": "GET", "path": "/openapi.json", "expected_status": 200},
            ]
            
            for endpoint in endpoints_to_test:
                try:
                    result["endpoints_tested"] += 1
                    
                    response = requests.request(
                        endpoint["method"],
                        f"{self.api_url}{endpoint['path']}",
                        timeout=10
                    )
                    
                    if response.status_code == endpoint["expected_status"]:
                        result["endpoints_passed"] += 1
                    else:
                        result["errors"].append(
                            f"{endpoint['method']} {endpoint['path']} returned {response.status_code}, "
                            f"expected {endpoint['expected_status']}"
                        )
                        
                except requests.RequestException as e:
                    result["errors"].append(f"Failed to test {endpoint['path']}: {str(e)}")
            
        except Exception as e:
            result["errors"].append(f"Functional test error: {str(e)}")
        
        return result
    
    def run_all_tests(self) -> Dict[str, Any]:
        """Run all client tests."""
        logger.info("Starting comprehensive client testing")
        
        # Test API functionality
        functional_results = self.run_functional_tests()
        
        # Test individual clients
        client_results = {}
        
        # Test TypeScript client
        if (self.clients_dir / "typescript").exists():
            logger.info("Testing TypeScript client")
            client_results["typescript"] = self.test_typescript_client()
        
        # Test Python client
        if (self.clients_dir / "python").exists():
            logger.info("Testing Python client")
            client_results["python"] = self.test_python_client()
        
        # Test Java client
        if (self.clients_dir / "java").exists():
            logger.info("Testing Java client")
            client_results["java"] = self.test_java_client()
        
        # Test structure for all clients
        structure_results = {}
        for client_dir in self.clients_dir.iterdir():
            if client_dir.is_dir():
                language = client_dir.name
                logger.info(f"Testing {language} client structure")
                structure_results[language] = self.test_client_structure(language)
        
        return {
            "functional_tests": functional_results,
            "client_tests": client_results,
            "structure_tests": structure_results,
            "summary": self._generate_summary(functional_results, client_results, structure_results)
        }
    
    def _generate_summary(self, functional: Dict, clients: Dict, structures: Dict) -> Dict[str, Any]:
        """Generate test summary."""
        total_clients = len(structures)
        working_clients = 0
        
        for language, result in clients.items():
            if language == "typescript" and result.get("builds", False):
                working_clients += 1
            elif language == "python" and result.get("imports", False):
                working_clients += 1
            elif language == "java" and result.get("compiles", False):
                working_clients += 1
        
        return {
            "total_clients_found": total_clients,
            "clients_tested": len(clients),
            "working_clients": working_clients,
            "api_functional": functional.get("api_reachable", False),
            "openapi_valid": functional.get("openapi_valid", False),
            "endpoints_success_rate": (
                functional.get("endpoints_passed", 0) / max(functional.get("endpoints_tested", 1), 1) * 100
            )
        }
    
    def print_results(self, results: Dict[str, Any]) -> None:
        """Print test results in a formatted way."""
        print("\n" + "="*60)
        print("CLIENT SDK TEST RESULTS")
        print("="*60)
        
        # Print summary
        summary = results["summary"]
        print(f"\nSUMMARY:")
        print(f"  Total clients found: {summary['total_clients_found']}")
        print(f"  Clients tested: {summary['clients_tested']}")
        print(f"  Working clients: {summary['working_clients']}")
        print(f"  API functional: {'✅' if summary['api_functional'] else '❌'}")
        print(f"  OpenAPI valid: {'✅' if summary['openapi_valid'] else '❌'}")
        print(f"  Endpoints success rate: {summary['endpoints_success_rate']:.1f}%")
        
        # Print functional test results
        print(f"\nFUNCTIONAL TESTS:")
        functional = results["functional_tests"]
        print(f"  API reachable: {'✅' if functional['api_reachable'] else '❌'}")
        print(f"  OpenAPI valid: {'✅' if functional['openapi_valid'] else '❌'}")
        print(f"  Endpoints tested: {functional['endpoints_tested']}")
        print(f"  Endpoints passed: {functional['endpoints_passed']}")
        
        if functional["errors"]:
            print("  Errors:")
            for error in functional["errors"]:
                print(f"    - {error}")
        
        # Print client test results
        print(f"\nCLIENT TESTS:")
        for language, result in results["client_tests"].items():
            print(f"\n  {language.upper()}:")
            print(f"    Exists: {'✅' if result['exists'] else '❌'}")
            
            if language == "typescript":
                print(f"    Builds: {'✅' if result['builds'] else '❌'}")
            elif language == "python":
                print(f"    Installs: {'✅' if result['installs'] else '❌'}")
                print(f"    Imports: {'✅' if result['imports'] else '❌'}")
            elif language == "java":
                print(f"    Compiles: {'✅' if result['compiles'] else '❌'}")
            
            print(f"    Has examples: {'✅' if result['has_examples'] else '❌'}")
            print(f"    Has docs: {'✅' if result['has_docs'] else '❌'}")
            
            if result["errors"]:
                print("    Errors:")
                for error in result["errors"]:
                    print(f"      - {error}")
        
        # Print structure test results
        print(f"\nSTRUCTURE TESTS:")
        for language, result in results["structure_tests"].items():
            print(f"\n  {language.upper()}:")
            print(f"    Exists: {'✅' if result['exists'] else '❌'}")
            print(f"    Structure valid: {'✅' if result['structure_valid'] else '❌'}")
            print(f"    Has API files: {'✅' if result['has_api_files'] else '❌'}")
            print(f"    Has model files: {'✅' if result['has_model_files'] else '❌'}")
            print(f"    Has docs: {'✅' if result['has_docs'] else '❌'}")
            print(f"    File count: {result['file_count']}")
            
            if result["errors"]:
                print("    Errors:")
                for error in result["errors"]:
                    print(f"      - {error}")
        
        print("\n" + "="*60)


def main():
    """Main function for command-line usage."""
    import argparse
    
    parser = argparse.ArgumentParser(description="Test generated client SDKs")
    parser.add_argument(
        "--api-url",
        default="http://localhost:8000",
        help="Base URL of the API (default: http://localhost:8000)"
    )
    parser.add_argument(
        "--clients-dir",
        default="generated_clients",
        help="Directory containing generated clients (default: generated_clients)"
    )
    parser.add_argument(
        "--output-file",
        help="Save test results to JSON file"
    )
    parser.add_argument(
        "--verbose",
        action="store_true",
        help="Enable verbose logging"
    )
    
    args = parser.parse_args()
    
    if args.verbose:
        logging.getLogger().setLevel(logging.DEBUG)
    
    # Create tester instance
    tester = ClientTester(args.api_url, args.clients_dir)
    
    # Run tests
    results = tester.run_all_tests()
    
    # Print results
    tester.print_results(results)
    
    # Save results to file if requested
    if args.output_file:
        with open(args.output_file, 'w') as f:
            json.dump(results, f, indent=2, default=str)
        logger.info(f"Test results saved to {args.output_file}")
    
    # Exit with error code if tests failed
    summary = results["summary"]
    if not summary["api_functional"] or summary["working_clients"] == 0:
        sys.exit(1)


if __name__ == "__main__":
    main()