#!/usr/bin/env python3 """ StreamA2026 System Test Runner Comprehensive test script that checks all critical components of the StreamA2026 Docker bundle. This script will automatically fix errors when possible. Usage: python tests/test_runner.py [options] Options: --no-fix Disable automatic error fixing --skip-frontend Skip frontend checks --skip-api Skip API tests --help Show this help message """ import os import sys import argparse from pathlib import Path # Add project root to path project_root = Path(__file__).parent.parent sys.path.insert(0, str(project_root)) from tests.utils.logger import Logger from tests.utils.docker_utils import DockerUtils from tests.utils.version_helper import VersionHelper from tests.modules.docker_checks import DockerChecks from tests.modules.database_checks import DatabaseChecks from tests.modules.api_tests import APITests from tests.modules.health_checks import HealthChecks from tests.modules.frontend_checks import FrontendChecks from tests.utils.api_analyzer import APIAnalyzer from tests.modules.permission_checks import PermissionChecks from tests.modules.env_checks import EnvironmentChecks from tests.modules.security_headers_check import SecurityHeadersCheck from tests.modules.rate_limiting_check import RateLimitingCheck from tests.modules.network_checks import NetworkChecks from tests.modules.ssl_checks import SSLChecks from tests.modules.flyway_checks import FlywayChecks from tests.modules.code_quality_checks import CodeQualityChecks from tests.utils.coverage_reporter import CoverageReporter def main(): # Load environment variables early to avoid race conditions from dotenv import load_dotenv env_file = Path(__file__).parent.parent / ".env" if env_file.exists(): load_dotenv(env_file) parser = argparse.ArgumentParser( description="StreamA2026 System Test Runner", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" Examples: python tests/test_runner.py # Run all tests with auto-fix python tests/test_runner.py --no-fix # Run tests without auto-fix python tests/test_runner.py --skip-frontend # Skip frontend checks python tests/test_runner.py --skip-api # Skip API tests python tests/test_runner.py --fresh # Reset environment to fresh installation """ ) parser.add_argument("--no-fix", action="store_true", help="Disable automatic error fixing") parser.add_argument("--skip-frontend", action="store_true", help="Skip frontend checks") parser.add_argument("--skip-api", action="store_true", help="Skip API tests") parser.add_argument("--fresh", action="store_true", help="Reset environment to fresh installation (deletes all data)") parser.add_argument("--verbose", action="store_true", default=True, help="Verbose output (default)") parser.add_argument("--generate-report", action="store_true", help="Generate comprehensive coverage report") parser.add_argument("--update-baseline", action="store_true", help="Update API endpoint baseline") args = parser.parse_args() # Initialize logger logger = Logger(verbose=args.verbose) # Print header logger.header("StreamA2026 System Test Runner") logger.info("SYSTEM", f"Project directory: {project_root}") logger.info("SYSTEM", f"Auto-fix: {'disabled' if args.no_fix else 'enabled'}") logger.info("SYSTEM", f"Fresh reset: {'enabled' if args.fresh else 'disabled'}") # Read and display current version using version helper version_helper = VersionHelper(str(project_root)) version_info = version_helper.read_version() current_version = version_info.full_version logger.info("SYSTEM", f"Current version: {version_helper.format_version_header()}") # Check if .env file exists env_file = project_root / ".env" if not env_file.exists(): logger.warning("SYSTEM", ".env file not found - using defaults") logger.info("SYSTEM", "Consider creating .env from .env.example") else: logger.success("SYSTEM", ".env file found") # Fresh reset if requested if args.fresh: logger.subheader("Fresh Environment Reset") docker_utils = DockerUtils(logger) docker_utils.set_project_dir(str(project_root)) if not docker_utils.reset_environment(logger): logger.error("DOCKER", "Failed to reset environment") logger.print_summary() sys.exit(1) # Run checks auto_fix = not args.no_fix all_passed = True # 1. Docker Checks try: docker_checks = DockerChecks(logger, str(project_root), version=current_version) if not docker_checks.run_all_checks(auto_fix=auto_fix): all_passed = False logger.error("DOCKER", "Docker checks failed - stopping further tests") logger.print_summary() sys.exit(1) except Exception as e: logger.error("DOCKER", f"Docker checks crashed: {e}") all_passed = False logger.print_summary() sys.exit(1) # Create DockerUtils instance for network and SSL checks docker_utils = DockerUtils(logger, version=current_version) docker_utils.set_project_dir(str(project_root)) # 1.5. Network Configuration Check try: network_checks = NetworkChecks(logger, docker_utils, version=current_version) if not network_checks.run_all_checks(): all_passed = False except Exception as e: logger.error("NETWORK", f"Network checks crashed: {e}") all_passed = False # 1.6. SSL/Let's Encrypt Check try: domain = os.getenv("DOMAIN") ssl_checks = SSLChecks(logger, docker_utils, domain, version=current_version) if not ssl_checks.run_all_checks(): all_passed = False except Exception as e: logger.error("SSL", f"SSL checks crashed: {e}") all_passed = False # 2. Wait for health checks try: if not docker_checks.wait_for_health(timeout=60): logger.warning("DOCKER", "Some containers failed health checks - continuing with tests") except Exception as e: logger.warning("DOCKER", f"Health check wait failed: {e}") # 3. Database Checks try: db_checks = DatabaseChecks(logger, str(project_root), version=current_version) if not db_checks.run_all_checks(auto_fix=auto_fix): all_passed = False except Exception as e: logger.error("DATABASE", f"Database checks crashed: {e}") all_passed = False # 3.5. Flyway Migration Checks try: flyway_checks = FlywayChecks(logger, str(project_root), version=current_version) if not flyway_checks.run_all_checks(): all_passed = False except Exception as e: logger.error("FLYWAY", f"Flyway checks crashed: {e}") all_passed = False # 4. Health Checks try: health_checks = HealthChecks(logger, base_url="http://localhost:8000", version=current_version) if not health_checks.run_all_checks(): all_passed = False except Exception as e: logger.error("HEALTH", f"Health checks crashed: {e}") all_passed = False # 5. API Tests if not args.skip_api: try: api_tests = APITests(logger, base_url="http://localhost:8000", project_dir=str(project_root), version=current_version) if not api_tests.run_all_tests(auto_fix=auto_fix): all_passed = False # Cleanup test user after permission checks (moved to after permission tests) # api_tests.cleanup() except Exception as e: logger.error("API", f"API tests crashed: {e}") all_passed = False # 6. Frontend Checks if not args.skip_frontend: try: frontend_checks = FrontendChecks(logger, str(project_root), version=current_version) if not frontend_checks.run_all_checks(): all_passed = False except Exception as e: logger.error("FRONTEND", f"Frontend checks crashed: {e}") all_passed = False # API analyzer check (only if frontend is checked) if not args.skip_frontend: logger.subheader("API Endpoint Analysis") api_analyzer = APIAnalyzer(str(project_root), version=current_version) api_differences = api_analyzer.compare_apis() if api_differences["missing_backend_endpoints"]: logger.error("API", "Frontend API calls that don't exist in backend:") for item in api_differences["missing_backend_endpoints"]: logger.error("API", f" {item['method']} {item['path']} ({item['file']}:{item['line']})") all_passed = False else: logger.success("API", "All frontend API calls have corresponding backend endpoints") # 7. Environment Variables Check try: env_checks = EnvironmentChecks(logger, str(project_root), version=current_version) if not env_checks.run_all_checks(): all_passed = False except Exception as e: logger.error("ENV", f"Environment checks crashed: {e}") all_passed = False # 8. Permission Checks permissions_summary = None try: permission_checks = PermissionChecks(logger, base_url="http://localhost:8000", version=current_version) if not permission_checks.run_all_checks(): all_passed = False # Get detailed permissions summary for final report try: permissions_summary = permission_checks.get_detailed_permissions_summary() except Exception as e: logger.error("PERMISSIONS", f"Failed to get permissions summary: {e}") permissions_summary = None except Exception as e: logger.error("PERMISSIONS", f"Permission checks crashed: {e}") all_passed = False permissions_summary = None # Cleanup test user after permission checks if not args.skip_api: try: api_tests.cleanup() except Exception as e: logger.warning("API", f"Failed to cleanup test user: {e}") # 9. Security Headers Check (informational only) try: security_headers_check = SecurityHeadersCheck(logger, base_url="http://localhost:8000", version=current_version) security_headers_check.run_all_checks() # Always returns True, informational only except Exception as e: logger.warning("SECURITY", f"Security headers check failed: {e}") # 10. Rate Limiting Check (informational only) try: rate_limiting_check = RateLimitingCheck(logger, base_url="http://localhost:8000", version=current_version) rate_limiting_check.run_all_checks() # Always returns True, informational only except Exception as e: logger.warning("RATE_LIMIT", f"Rate limiting check failed: {e}") # 11. Code Quality Checks try: code_quality_checks = CodeQualityChecks(logger, str(project_root), version=current_version) if not code_quality_checks.run_all_checks(): all_passed = False except Exception as e: logger.error("CODE_QUALITY", f"Code quality checks crashed: {e}") all_passed = False # 12. Generate Coverage Report (if requested) if args.generate_report: logger.subheader("Coverage Report Generation") try: coverage_reporter = CoverageReporter(str(project_root), version=current_version) # Collect coverage data api_analyzer = APIAnalyzer(str(project_root), version=current_version) # Load baseline to ensure new/removed detection works api_analyzer.load_baseline() api_coverage = api_analyzer.generate_endpoint_coverage_report() # Mark permission endpoints as intentionally unused if 'intentionally_unused_endpoints' not in api_coverage: api_coverage['intentionally_unused_endpoints'] = [] permission_endpoints = [ {'method': 'GET', 'path': '/permissions/admin'}, {'method': 'GET', 'path': '/permissions/public'}, {'method': 'GET', 'path': '/permissions'}, {'method': 'GET', 'path': '/permissions/protected'} ] for perm_endpoint in permission_endpoints: # Check if endpoint exists in intentionally_unused_endpoints exists = False for existing in api_coverage['intentionally_unused_endpoints']: if existing['method'] == perm_endpoint['method'] and existing['path'] == perm_endpoint['path']: exists = True break if not exists: api_coverage['intentionally_unused_endpoints'].append(perm_endpoint) # Remove permission endpoints from unused_endpoints if they are there if 'unused_endpoints' in api_coverage: api_coverage['unused_endpoints'] = [ ep for ep in api_coverage['unused_endpoints'] if not (ep['method'] in ['GET'] and ep['path'] in ['/permissions/admin', '/permissions/public', '/permissions', '/permissions/protected']) ] # Recalculate unused_count and unused_percentage based on updated intentionally_unused_endpoints total_backend = api_coverage.get('total_backend_endpoints', 0) intentionally_unused_count = len(api_coverage.get('intentionally_unused_endpoints', [])) unused_count = len(api_coverage.get('unused_endpoints', [])) api_coverage['unused_count'] = unused_count api_coverage['unused_percentage'] = round(unused_count / total_backend * 100, 1) if total_backend > 0 else 0 api_coverage['intentionally_unused_count'] = intentionally_unused_count api_coverage['intentionally_unused_percentage'] = round(intentionally_unused_count / total_backend * 100, 1) if total_backend > 0 else 0 model_schema_analyzer = None model_schema_coverage = None try: from tests.utils.model_schema_analyzer import ModelSchemaAnalyzer model_schema_analyzer = ModelSchemaAnalyzer(str(project_root)) model_schema_coverage = model_schema_analyzer.compare_model_schema_coverage() except: pass migration_checker = None migration_consistency = None try: from tests.utils.migration_checker import MigrationChecker migration_checker = MigrationChecker(str(project_root)) migration_consistency = migration_checker.check_migration_consistency() except: pass schema_payload_validator = None schema_payload_issues = None try: from tests.utils.schema_payload_validator import SchemaPayloadValidator schema_payload_validator = SchemaPayloadValidator(str(project_root)) schema_payload_issues = schema_payload_validator.validate_payloads() except: pass database_consistency_validator = None database_consistency_issues = None try: from tests.utils.database_consistency_validator import DatabaseConsistencyValidator database_consistency_validator = DatabaseConsistencyValidator(project_root) database_consistency_report = database_consistency_validator.generate_report() database_consistency_issues = database_consistency_report['issues'] except: pass # Generate report with permissions summary report_path = coverage_reporter.generate_comprehensive_report( api_coverage=api_coverage, model_schema_coverage=model_schema_coverage, migration_consistency=migration_consistency, schema_payload_issues=schema_payload_issues, database_consistency_issues=database_consistency_issues, permissions_summary=permissions_summary ) logger.success("REPORT", f"Coverage report generated: {report_path}") except Exception as e: logger.error("REPORT", f"Failed to generate coverage report: {e}") # 12. Update API Baseline (if requested) if args.update_baseline: logger.subheader("API Baseline Update") try: api_analyzer = APIAnalyzer(str(project_root), version=current_version) backend_endpoints = api_analyzer.analyze_backend_endpoints() if api_analyzer.save_baseline(backend_endpoints): logger.success("BASELINE", "API endpoint baseline updated successfully") else: logger.error("BASELINE", "Failed to update API baseline") except Exception as e: logger.error("BASELINE", f"Failed to update baseline: {e}") # Print detailed permissions summary if permissions_summary and "error" not in permissions_summary: logger.subheader("Detailed Permissions Summary") total_permissions = permissions_summary.get('total_permissions', permissions_summary.get('total_endpoints', 0)) logger.info("PERMISSIONS", f"Total Permissions: {total_permissions}") # Check format and display accordingly format_type = permissions_summary.get('format', 'unknown') if format_type == 'permission_definitions': # New format: permissions grouped by category by_category = permissions_summary.get('by_category', {}) for category, permissions in sorted(by_category.items()): logger.info("PERMISSIONS", f"{category}: {len(permissions)} permissions") for perm in permissions: name = perm.get('name', 'unknown') display_name = perm.get('display_name', name) endpoint_pattern = perm.get('endpoint_pattern', 'N/A') logger.info("PERMISSIONS", f" {name} ({display_name}): {endpoint_pattern}") elif format_type == 'route_analysis': # Old format: endpoints grouped by auth requirement public_endpoints = permissions_summary.get('public_endpoints', []) auth_endpoints = permissions_summary.get('auth_endpoints', []) admin_endpoints = permissions_summary.get('admin_endpoints', []) if public_endpoints: logger.info("PERMISSIONS", f"Public endpoints: {len(public_endpoints)}") for item in public_endpoints: logger.info("PERMISSIONS", f" [{item['method']}] {item['path']}") if auth_endpoints: logger.info("PERMISSIONS", f"Auth required: {len(auth_endpoints)}") for item in auth_endpoints: logger.info("PERMISSIONS", f" [{item['method']}] {item['path']}") if admin_endpoints: logger.info("PERMISSIONS", f"Admin only: {len(admin_endpoints)}") for item in admin_endpoints: logger.info("PERMISSIONS", f" [{item['method']}] {item['path']}") # Print summary logger.print_summary() # Increment version if all tests passed and AUTO_INCREMENT_VERSION is true if all_passed and not args.no_fix: auto_increment = os.getenv("AUTO_INCREMENT_VERSION", "false").lower() == "true" if auto_increment: logger.info("SYSTEM", "Incrementing version...") try: # Import version increment function sys.path.insert(0, str(project_root / "scripts")) from increment_version import increment_version new_version = increment_version() if new_version: logger.success("SYSTEM", f"Version incremented to: {new_version}") except Exception as e: logger.warning("SYSTEM", f"Failed to increment version: {e}") else: logger.info("SYSTEM", "AUTO_INCREMENT_VERSION is false - skipping version increment") # Exit with appropriate code if all_passed: sys.exit(0) else: sys.exit(1) if __name__ == "__main__": main()