| import asyncio |
| import os |
| import sys |
| import httpx |
|
|
| |
| sys.path.append(os.path.join(os.getcwd(), 'backend')) |
|
|
| async def test_phase7(): |
| print("--- Phase 7 Verification: Additional Platform Integration ---") |
| |
| from core.lazy_integration_registry import load_integration |
| from enhanced_workflow_api import enhanced_workflow_api, IntelligenceAnalyzeRequest, ai_service |
| |
| |
| if ai_service: |
| async def mock_analyze(*args, **kwargs): |
| return "Mock analysis result" |
| ai_service.analyze_text = mock_analyze |
| |
| |
| platforms = ["telegram", "whatsapp", "zoom"] |
| print("\n1. Testing Route Loading & Health Endpoints...") |
| |
| for platform in platforms: |
| router = load_integration(platform) |
| if router: |
| print(f"✓ {platform.capitalize()} router loaded successfully") |
| |
| |
| |
| if hasattr(router, "prefix"): |
| print(f" Prefix: {router.prefix}") |
| else: |
| print(f"✗ Failed to load {platform} router") |
|
|
| |
| print("\n2. Testing Intelligence Suggestions...") |
| test_cases = [ |
| ("Send a telegram message to the team", "telegram"), |
| ("Message me on WhatsApp", "whatsapp"), |
| ("Start a zoom meeting", "zoom") |
| ] |
| |
| for text, expected_service in test_cases: |
| request = IntelligenceAnalyzeRequest(text=text) |
| result = await enhanced_workflow_api.analyze_workflow_intent(request) |
| |
| found = False |
| for suggestion in result.get("routing_suggestions", []): |
| if suggestion.get("primary") == expected_service: |
| print(f"✓ Found correct suggestion for '{text}': {expected_service}") |
| found = True |
| break |
| |
| if not found: |
| print(f"✗ Failed to find correct suggestion for '{text}'. Got: {[s.get('primary') for s in result.get('routing_suggestions', [])]}") |
|
|
| if __name__ == "__main__": |
| asyncio.run(test_phase7()) |
|
|
