Spaces:
Sleeping
Sleeping
File size: 3,044 Bytes
5e0532d |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 |
import asyncio
import os
from app.services.orchestrator import orchestrator
from app.services.profile import profile_service
from app.services.journal import journal_service
from app.services.audit import audit_service
from app.services.bible import bible_service
# Mock user ID for testing
TEST_USER_ID = "test_user_integration_v1"
async def main():
print("=== ORA MVP Integration Test ===\n")
# 1. Setup / Cleanup
print("1. [SETUP] Clearing test data...")
# Clean up any previous test entries if needed (optional, keeping simple for now)
# 2. Profile Test
print("\n2. [PROFILE] Updating and retrieving profile...")
await profile_service.update_profile(TEST_USER_ID, {"denomination": "Integration Test", "spiritual_goals": ["study", "prayer"]})
profile = await profile_service.get_profile(TEST_USER_ID)
print(f" -> Denomination: {profile.denomination}")
print(f" -> Goals: {profile.spiritual_goals}")
assert profile.denomination == "Integration Test"
assert "study" in [g.value for g in profile.spiritual_goals]
# 3. Bible RAG Test (Mock Mode check)
print("\n3. [BIBLE] Search for 'peace'...")
# Ensure index is initialized
await bible_service.initialize_index()
results = await bible_service.search("peace")
print(f" -> Found {len(results)} results.")
if results:
print(f" -> Top result: {results[0]['reference']} - {results[0]['text'][:30]}...")
# 4. Orchestrator Flow (Simulate Chat)
print("\n4. [ORCHESTRATOR] Sending 'I feel anxious'...")
# This triggers Guardrails -> Intent -> Emotion -> Routing -> Response
# Since we don't have a real LLM running (probably), the stubbed services might fail or return mock data.
# However, we mostly want to ensure no crashes happen.
try:
response = await orchestrator.process_message(TEST_USER_ID, "I feel anxious about the integration test.")
print(f" -> ORA Response: {response['content']}")
assert response['role'] == "assistant"
except Exception as e:
print(f" -> [ERROR] Orchestrator failed: {e}")
# 5. Journal Persistence
print("\n5. [JOURNAL] Saving interaction...")
# The orchestrator usually handles this, but let's verify manual save works
entry_id = await journal_service.save_entry(TEST_USER_ID, "Integration Test Entry", ["test", "mvp"])
print(f" -> Saved Entry ID: {entry_id}")
recent = await journal_service.get_recent_entries(TEST_USER_ID, 1)
print(f" -> Retrieved: {recent[0]['content']}")
assert recent[0]['content'] == "Integration Test Entry"
# 6. Audit Log Verification
print("\n6. [AUDIT] Checking log file existence...")
if os.path.exists("logs/ora.log"):
size = os.path.getsize("logs/ora.log")
print(f" -> Log file exists. Size: {size} bytes.")
assert size > 0
else:
print(" -> [ERROR] Log file not found!")
print("\n=== Integration Test Complete ===")
if __name__ == "__main__":
asyncio.run(main())
|