#!/usr/bin/env python3 """ Test script for multi-level report generation and context retrieval """ import os import sys import asyncio from pathlib import Path from dotenv import load_dotenv # Add current directory to path sys.path.insert(0, str(Path(__file__).parent)) load_dotenv() async def test_context_retrieval(): """Test context retrieval functions.""" print("\n" + "=" * 60) print("Testing Context Retrieval Functions") print("=" * 60) try: from server import ( retrieve_all_module_analyses, retrieve_synthesis_analysis, retrieve_cumulative_analysis_state, retrieve_all_findings, retrieve_all_metrics, retrieve_comprehensive_report_context ) print("✅ All context retrieval functions imported") # Test with a dummy run_id test_run_id = "test_run_123" test_repository_id = "test_repo_123" test_session_id = "test_session_123" print(f"\nTesting with run_id: {test_run_id}") print(f"Repository ID: {test_repository_id}") print(f"Session ID: {test_session_id}") # Test each function print("\n1. Testing retrieve_all_module_analyses...") modules = await retrieve_all_module_analyses(test_run_id, test_repository_id) print(f" ✓ Found {len(modules)} modules") print("\n2. Testing retrieve_synthesis_analysis...") synthesis = await retrieve_synthesis_analysis(test_run_id, test_repository_id) if synthesis: print(f" ✓ Found synthesis analysis") else: print(f" ⚠️ No synthesis analysis found (expected for test)") print("\n3. Testing retrieve_cumulative_analysis_state...") state = await retrieve_cumulative_analysis_state(test_run_id, test_repository_id, test_session_id) if state: print(f" ✓ Found cumulative analysis state") else: print(f" ⚠️ No cumulative analysis state found (expected for test)") print("\n4. Testing retrieve_all_findings...") findings = await retrieve_all_findings(test_run_id) print(f" ✓ Found findings for {len(findings)} modules") print("\n5. Testing retrieve_all_metrics...") metrics = await retrieve_all_metrics(test_run_id) print(f" ✓ Found metrics for {len(metrics)} modules") print("\n6. Testing retrieve_comprehensive_report_context...") context = await retrieve_comprehensive_report_context( run_id=test_run_id, repository_id=test_repository_id, session_id=test_session_id ) print(f" ✓ Context retrieved:") print(f" - Modules: {context.get('total_modules', 0)}") print(f" - Findings: {context.get('total_findings', 0)}") print(f" - Has synthesis: {bool(context.get('synthesis_analysis'))}") print(f" - Has analysis state: {bool(context.get('analysis_state'))}") print("\n✅ All context retrieval tests passed!") return True except Exception as e: print(f"\n❌ Context retrieval test failed: {e}") import traceback traceback.print_exc() return False def test_pdf_method_exists(): """Test that the new PDF method exists.""" print("\n" + "=" * 60) print("Testing PDF Report Method") print("=" * 60) try: # Import using the same method as server.py import sys import importlib.util spec = importlib.util.spec_from_file_location("ai_analyze", "ai-analyze.py") ai_analyze_module = importlib.util.module_from_spec(spec) sys.modules["ai_analyze"] = ai_analyze_module spec.loader.exec_module(ai_analyze_module) from ai_analyze import EnhancedGitHubAnalyzer print("✅ EnhancedGitHubAnalyzer imported successfully") # Check if new method exists if hasattr(EnhancedGitHubAnalyzer, 'create_multi_level_pdf_report'): print("✅ create_multi_level_pdf_report method exists") # Check method signature import inspect sig = inspect.signature(EnhancedGitHubAnalyzer.create_multi_level_pdf_report) params = list(sig.parameters.keys()) print(f" Method parameters: {', '.join(params)}") if 'comprehensive_context' in params: print(" ✓ comprehensive_context parameter exists") if 'output_path' in params: print(" ✓ output_path parameter exists") if 'repository_id' in params: print(" ✓ repository_id parameter exists") if 'run_id' in params: print(" ✓ run_id parameter exists") return True else: print("❌ create_multi_level_pdf_report method not found") return False except Exception as e: print(f"❌ PDF method test failed: {e}") import traceback traceback.print_exc() return False def test_database_tables(): """Test that database tables exist.""" print("\n" + "=" * 60) print("Testing Database Tables") print("=" * 60) try: import psycopg2 from dotenv import load_dotenv load_dotenv() conn = psycopg2.connect( host=os.getenv('POSTGRES_HOST', 'localhost'), port=os.getenv('POSTGRES_PORT', '5432'), database=os.getenv('POSTGRES_DB', 'dev_pipeline'), user=os.getenv('POSTGRES_USER', 'pipeline_admin'), password=os.getenv('POSTGRES_PASSWORD', 'secure_pipeline_2024') ) cursor = conn.cursor() # Check each table tables_to_check = ['findings', 'metrics', 'report_sections', 'analysis_runs'] for table_name in tables_to_check: cursor.execute(f""" SELECT COUNT(*) FROM information_schema.tables WHERE table_schema = 'public' AND table_name = %s """, (table_name,)) exists = cursor.fetchone()[0] > 0 if exists: # Get row count cursor.execute(f"SELECT COUNT(*) FROM {table_name}") count = cursor.fetchone()[0] print(f"✅ Table '{table_name}' exists ({count} rows)") else: print(f"❌ Table '{table_name}' does not exist") return False cursor.close() conn.close() print("\n✅ All database tables verified!") return True except Exception as e: print(f"❌ Database test failed: {e}") import traceback traceback.print_exc() return False async def run_all_tests(): """Run all tests.""" print("\n" + "=" * 60) print("MULTI-LEVEL REPORT IMPLEMENTATION TEST SUITE") print("=" * 60) results = [] # Test 1: Database tables results.append(("Database Tables", test_database_tables())) # Test 2: PDF method exists results.append(("PDF Method", test_pdf_method_exists())) # Test 3: Context retrieval results.append(("Context Retrieval", await test_context_retrieval())) # Summary print("\n" + "=" * 60) print("TEST SUMMARY") print("=" * 60) passed = 0 failed = 0 for test_name, result in results: status = "✅ PASSED" if result else "❌ FAILED" print(f"{test_name}: {status}") if result: passed += 1 else: failed += 1 print(f"\nTotal: {passed} passed, {failed} failed out of {len(results)} tests") if failed == 0: print("\n✅ All tests passed! Implementation is ready.") return True else: print(f"\n⚠️ {failed} test(s) failed. Please review the errors above.") return False if __name__ == "__main__": success = asyncio.run(run_all_tests()) sys.exit(0 if success else 1)