b745151d2c
- Create end-to-end tests - Add diagnose file to determine cause of failure for end-to-end tests
106 lines
3.7 KiB
Python
106 lines
3.7 KiB
Python
#!/usr/bin/env python
|
|
"""Diagnostic script to test each component independently."""
|
|
import asyncio
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
# Add the app directory to Python path
|
|
app_dir = Path(__file__).parent
|
|
sys.path.insert(0, str(app_dir))
|
|
|
|
async def diagnose():
|
|
print("🔍 SkyTalk API Diagnostics\n")
|
|
|
|
# 1. Test configuration
|
|
print("1. Testing configuration...")
|
|
try:
|
|
from app.core.config import settings
|
|
print(f" ✅ Config loaded")
|
|
print(f" - API Key set: {bool(settings.GOOGLE_API_KEY and len(settings.GOOGLE_API_KEY) > 10)}")
|
|
print(f" - Flash model: {settings.LLM_FLASH_MODEL}")
|
|
print(f" - Pro model: {settings.LLM_PRO_MODEL}")
|
|
print(f" - Embedding model: {settings.EMBEDDING_MODEL}")
|
|
except Exception as e:
|
|
print(f" ❌ Config error: {e}")
|
|
return
|
|
|
|
# 2. Test database
|
|
print("\n2. Testing database...")
|
|
try:
|
|
from app.data.database import init_db, get_session
|
|
await init_db()
|
|
print(f" ✅ Database initialized")
|
|
|
|
async with get_session() as db:
|
|
from app.data.repositories import SessionRepository
|
|
repo = SessionRepository(db)
|
|
session = await repo.create()
|
|
print(f" ✅ Can create sessions: {session.id}")
|
|
except Exception as e:
|
|
print(f" ❌ Database error: {e}")
|
|
return
|
|
|
|
# 3. Test vector service
|
|
print("\n3. Testing vector service...")
|
|
try:
|
|
from app.services.vector import VectorService
|
|
vector_service = VectorService()
|
|
print(f" ✅ Vector service initialized")
|
|
except Exception as e:
|
|
print(f" ❌ Vector service error: {e}")
|
|
return
|
|
|
|
# 4. Test interviewer agent
|
|
print("\n4. Testing interviewer agent...")
|
|
try:
|
|
from app.services.interviewer import InterviewerAgent
|
|
interviewer = InterviewerAgent(vector_service)
|
|
print(f" ✅ Interviewer agent initialized")
|
|
|
|
# Test a simple response
|
|
response, should_end = await interviewer.generate_response(
|
|
transcript=[{"role": "user", "content": "Hello"}],
|
|
context_query="greeting"
|
|
)
|
|
print(f" ✅ Can generate responses (length: {len(response)})")
|
|
except Exception as e:
|
|
print(f" ❌ Interviewer error: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
return
|
|
|
|
# 5. Test synthesizer agent
|
|
print("\n5. Testing synthesizer agent...")
|
|
try:
|
|
from app.services.synthesizer import SynthesizerAgent
|
|
synthesizer = SynthesizerAgent(vector_service)
|
|
print(f" ✅ Synthesizer agent initialized")
|
|
except Exception as e:
|
|
print(f" ❌ Synthesizer error: {e}")
|
|
return
|
|
|
|
# 6. Test session service
|
|
print("\n6. Testing session service...")
|
|
try:
|
|
from app.services.session_service import SessionService
|
|
service = SessionService()
|
|
print(f" ✅ Session service initialized")
|
|
|
|
# Test starting a session
|
|
result = await service.start_session("test diagnostics")
|
|
print(f" ✅ Can start sessions: {result['session_id']}")
|
|
print(f" - Initial message: {result['message'][:50]}...")
|
|
except Exception as e:
|
|
print(f" ❌ Session service error: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
return
|
|
|
|
print("\n✅ All components working!")
|
|
print("\nIf the API is still failing, check:")
|
|
print("1. The server logs for the actual error")
|
|
print("2. That the server was fully restarted after config changes")
|
|
print("3. That there are no port conflicts")
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(diagnose()) |