Spaces:
Sleeping
Sleeping
| 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()) | |