|
|
|
|
|
""" |
|
|
Test script for PDF Analysis & Orchestrator deployment |
|
|
Run this to verify all components are working correctly |
|
|
""" |
|
|
|
|
|
import os |
|
|
import sys |
|
|
import asyncio |
|
|
from pathlib import Path |
|
|
|
|
|
def test_imports(): |
|
|
"""Test that all required modules can be imported""" |
|
|
print("π Testing imports...") |
|
|
|
|
|
try: |
|
|
import gradio as gr |
|
|
print("β
Gradio imported successfully") |
|
|
except ImportError as e: |
|
|
print(f"β Gradio import failed: {e}") |
|
|
return False |
|
|
|
|
|
try: |
|
|
import openai |
|
|
print("β
OpenAI imported successfully") |
|
|
except ImportError as e: |
|
|
print(f"β OpenAI import failed: {e}") |
|
|
return False |
|
|
|
|
|
try: |
|
|
import pdfplumber |
|
|
print("β
pdfplumber imported successfully") |
|
|
except ImportError as e: |
|
|
print(f"β pdfplumber import failed: {e}") |
|
|
return False |
|
|
|
|
|
try: |
|
|
import numpy as np |
|
|
print("β
NumPy imported successfully") |
|
|
except ImportError as e: |
|
|
print(f"β NumPy import failed: {e}") |
|
|
return False |
|
|
|
|
|
try: |
|
|
from reportlab.lib.pagesizes import letter |
|
|
print("β
ReportLab imported successfully") |
|
|
except ImportError as e: |
|
|
print(f"β ReportLab import failed: {e}") |
|
|
return False |
|
|
|
|
|
return True |
|
|
|
|
|
def test_config(): |
|
|
"""Test configuration loading""" |
|
|
print("\nπ§ Testing configuration...") |
|
|
|
|
|
try: |
|
|
from config import Config |
|
|
print("β
Config module imported successfully") |
|
|
|
|
|
|
|
|
print(f" - OpenAI Model: {Config.OPENAI_MODEL}") |
|
|
print(f" - Chunk Size: {Config.CHUNK_SIZE}") |
|
|
print(f" - Cache Enabled: {Config.CACHE_ENABLED}") |
|
|
print(f" - Max Upload MB: {Config.MAX_FILE_SIZE_MB}") |
|
|
|
|
|
return True |
|
|
except Exception as e: |
|
|
print(f"β Config test failed: {e}") |
|
|
return False |
|
|
|
|
|
def test_utils(): |
|
|
"""Test utility functions""" |
|
|
print("\nπ οΈ Testing utilities...") |
|
|
|
|
|
try: |
|
|
from utils import chunk_text, get_file_hash, load_pdf_text_cached |
|
|
print("β
Core utilities imported successfully") |
|
|
|
|
|
|
|
|
test_text = "This is a test document. " * 1000 |
|
|
chunks = chunk_text(test_text, 100) |
|
|
print(f" - Chunking test: {len(chunks)} chunks created") |
|
|
|
|
|
|
|
|
test_file = Path("test.txt") |
|
|
test_file.write_text("test content") |
|
|
file_hash = get_file_hash(str(test_file)) |
|
|
print(f" - File hash test: {file_hash[:8]}...") |
|
|
test_file.unlink() |
|
|
|
|
|
return True |
|
|
except Exception as e: |
|
|
print(f"β Utils test failed: {e}") |
|
|
return False |
|
|
|
|
|
def test_agents(): |
|
|
"""Test agent initialization""" |
|
|
print("\nπ€ Testing agents...") |
|
|
|
|
|
try: |
|
|
from agents import AnalysisAgent, CollaborationAgent, ConversationAgent, MasterOrchestrator |
|
|
print("β
Agent classes imported successfully") |
|
|
|
|
|
|
|
|
analysis_agent = AnalysisAgent("TestAgent", "gpt-4", 0) |
|
|
print(" - AnalysisAgent created successfully") |
|
|
|
|
|
|
|
|
agents = { |
|
|
"analysis": analysis_agent, |
|
|
"collab": CollaborationAgent("TestCollab", "gpt-4", 0), |
|
|
"conversation": ConversationAgent("TestConv", "gpt-4", 0) |
|
|
} |
|
|
orchestrator = MasterOrchestrator(agents) |
|
|
print(" - MasterOrchestrator created successfully") |
|
|
|
|
|
return True |
|
|
except Exception as e: |
|
|
print(f"β Agents test failed: {e}") |
|
|
return False |
|
|
|
|
|
def test_managers(): |
|
|
"""Test manager classes""" |
|
|
print("\nπ Testing managers...") |
|
|
|
|
|
try: |
|
|
from utils.prompts import PromptManager |
|
|
from utils.export import ExportManager |
|
|
print("β
Manager classes imported successfully") |
|
|
|
|
|
|
|
|
prompt_manager = PromptManager() |
|
|
prompts = prompt_manager.get_all_prompts() |
|
|
print(f" - PromptManager: {len(prompts)} default prompts loaded") |
|
|
|
|
|
|
|
|
export_manager = ExportManager() |
|
|
print(" - ExportManager created successfully") |
|
|
|
|
|
return True |
|
|
except Exception as e: |
|
|
print(f"β Managers test failed: {e}") |
|
|
return False |
|
|
|
|
|
def test_environment(): |
|
|
"""Test environment variables""" |
|
|
print("\nπ Testing environment...") |
|
|
|
|
|
openai_key = os.environ.get("OPENAI_API_KEY") |
|
|
if openai_key: |
|
|
print("β
OPENAI_API_KEY is set") |
|
|
print(f" - Key starts with: {openai_key[:8]}...") |
|
|
else: |
|
|
print("β οΈ OPENAI_API_KEY not set (required for full functionality)") |
|
|
|
|
|
|
|
|
env_vars = [ |
|
|
"OPENAI_MODEL", |
|
|
"CHUNK_SIZE", |
|
|
"CACHE_ENABLED", |
|
|
"ANALYSIS_MAX_UPLOAD_MB" |
|
|
] |
|
|
|
|
|
for var in env_vars: |
|
|
value = os.environ.get(var) |
|
|
if value: |
|
|
print(f" - {var}: {value}") |
|
|
else: |
|
|
print(f" - {var}: using default") |
|
|
|
|
|
return True |
|
|
|
|
|
def test_gradio_interface(): |
|
|
"""Test Gradio interface creation""" |
|
|
print("\nπ¨ Testing Gradio interface...") |
|
|
|
|
|
try: |
|
|
|
|
|
from app import demo |
|
|
print("β
Gradio interface created successfully") |
|
|
|
|
|
|
|
|
if hasattr(demo, 'blocks'): |
|
|
print(" - Interface has blocks structure") |
|
|
|
|
|
return True |
|
|
except Exception as e: |
|
|
print(f"β Gradio interface test failed: {e}") |
|
|
return False |
|
|
|
|
|
def main(): |
|
|
"""Run all tests""" |
|
|
print("π PDF Analysis & Orchestrator - Deployment Test") |
|
|
print("=" * 50) |
|
|
|
|
|
tests = [ |
|
|
test_imports, |
|
|
test_config, |
|
|
test_utils, |
|
|
test_agents, |
|
|
test_managers, |
|
|
test_environment, |
|
|
test_gradio_interface |
|
|
] |
|
|
|
|
|
passed = 0 |
|
|
total = len(tests) |
|
|
|
|
|
for test in tests: |
|
|
try: |
|
|
if test(): |
|
|
passed += 1 |
|
|
except Exception as e: |
|
|
print(f"β Test {test.__name__} failed with exception: {e}") |
|
|
|
|
|
print("\n" + "=" * 50) |
|
|
print(f"π Test Results: {passed}/{total} tests passed") |
|
|
|
|
|
if passed == total: |
|
|
print("π All tests passed! Your deployment is ready.") |
|
|
return 0 |
|
|
else: |
|
|
print("β οΈ Some tests failed. Please check the errors above.") |
|
|
return 1 |
|
|
|
|
|
if __name__ == "__main__": |
|
|
sys.exit(main()) |
|
|
|