app/test_api_manual.py

#!/usr/bin/env python
"""
Manual test script for SkyTalk API.

This script provides a comprehensive way to test the API functionality
without relying on external tools like curl or Postman.
"""
import asyncio
import sys
import time
import uuid
from pathlib import Path

import httpx
import uvicorn
from multiprocessing import Process

# Add the app directory to Python path
app_dir = Path(__file__).parent
sys.path.insert(0, str(app_dir))


class SkyTalkAPITester:
    """Comprehensive tester for SkyTalk API."""
    
    def __init__(self, base_url: str = "http://localhost:8000"):
        self.base_url = base_url
        self.client = httpx.AsyncClient(timeout=30.0)
    
    async def test_health_check(self) -> bool:
        """Test the health check endpoint."""
        print("🏥 Testing health check...")
        try:
            response = await self.client.get(f"{self.base_url}/health")
            if response.status_code == 200:
                data = response.json()
                print(f"✅ Health check passed: {data['status']}")
                return True
            else:
                print(f"❌ Health check failed: {response.status_code}")
                return False
        except Exception as e:
            print(f"❌ Health check error: {e}")
            return False
    
    async def test_root_endpoint(self) -> bool:
        """Test the root endpoint."""
        print("\n🏠 Testing root endpoint...")
        try:
            response = await self.client.get(f"{self.base_url}/")
            if response.status_code == 200:
                data = response.json()
                print(f"✅ Root endpoint: {data['message']}")
                print(f"   Version: {data['version']}")
                return True
            else:
                print(f"❌ Root endpoint failed: {response.status_code}")
                return False
        except Exception as e:
            print(f"❌ Root endpoint error: {e}")
            return False
    
    async def test_start_session(self, topic: str) -> str:
        """Test starting a new session."""
        print(f"\n🚀 Starting session with topic: '{topic}'...")
        try:
            response = await self.client.post(
                f"{self.base_url}/sessions/start",
                json={"topic": topic}
            )
            if response.status_code == 200:
                data = response.json()
                session_id = data["session_id"]
                print(f"✅ Session started: {session_id}")
                print(f"   Status: {data['status']}")
                print(f"   Initial message: {data['message'][:100]}...")
                return session_id
            else:
                print(f"❌ Start session failed: {response.status_code}")
                print(f"   Response: {response.text}")
                return None
        except Exception as e:
            print(f"❌ Start session error: {e}")
            return None
    
    async def test_send_message(self, session_id: str, message: str) -> dict:
        """Test sending a message to a session."""
        print(f"\n💬 Sending message: '{message}'...")
        try:
            response = await self.client.post(
                f"{self.base_url}/sessions/sendMessage",
                json={"session_id": session_id, "message": message}
            )
            if response.status_code == 200:
                data = response.json()
                print(f"✅ Message sent successfully")
                print(f"   Status: {data['status']}")
                print(f"   Response: {data['message'][:100]}...")
                return data
            else:
                print(f"❌ Send message failed: {response.status_code}")
                print(f"   Response: {response.text}")
                return None
        except Exception as e:
            print(f"❌ Send message error: {e}")
            return None
    
    async def test_get_status(self, session_id: str) -> dict:
        """Test getting session status."""
        print(f"\n📊 Getting session status...")
        try:
            response = await self.client.get(
                f"{self.base_url}/sessions/getStatus",
                params={"session_id": session_id}
            )
            if response.status_code == 200:
                data = response.json()
                print(f"✅ Status retrieved: {data['status']}")
                print(f"   Notes count: {data['notes_count']}")
                print(f"   Created at: {data['created_at']}")
                return data
            else:
                print(f"❌ Get status failed: {response.status_code}")
                print(f"   Response: {response.text}")
                return None
        except Exception as e:
            print(f"❌ Get status error: {e}")
            return None
    
    async def test_complete_conversation_flow(self) -> bool:
        """Test a complete conversation flow from start to synthesis."""
        print("\n🔄 Testing complete conversation flow...")
        
        # Start session
        session_id = await self.test_start_session("AI ethics in healthcare")
        if not session_id:
            return False
        
        # Have a conversation
        messages = [
            "I'm particularly concerned about bias in medical AI systems",
            "Yes, I think algorithms could discriminate against certain patient groups",
            "Privacy is also a huge concern - medical data is so sensitive",
            "I think we need strict regulations and auditing processes",
            "Yes, I feel like we've covered the main points I wanted to explore"
        ]
        
        session_ended = False
        for message in messages:
            result = await self.test_send_message(session_id, message)
            if not result:
                return False
            
            # Check if session ended (processing status)
            if result.get("status") == "processing":
                print("🔄 Session ended, synthesis should begin...")
                session_ended = True
                break
            
            # Small delay between messages
            await asyncio.sleep(1)
        
        # Wait a bit for background processing
        if session_ended:
            print("\n⏳ Waiting for synthesis to complete...")
            for i in range(10):  # Wait up to 30 seconds
                await asyncio.sleep(3)
                status = await self.test_get_status(session_id)
                if status and status.get("status") == "completed":
                    print(f"✅ Synthesis completed! Generated {status.get('notes_count', 0)} notes")
                    return True
                elif status and status.get("status") == "failed":
                    print("❌ Synthesis failed")
                    return False
                print(f"   Still processing... ({i+1}/10)")
            
            print("⚠️  Synthesis still processing after 30 seconds")
            return True  # Not necessarily a failure
        
        print("⚠️  Session didn't end naturally")
        return True
    
    async def test_error_conditions(self) -> bool:
        """Test error handling."""
        print("\n🚨 Testing error conditions...")
        
        # Test invalid session ID
        print("   Testing invalid session ID...")
        fake_session_id = str(uuid.uuid4())
        result = await self.test_send_message(fake_session_id, "test message")
        if result is None:
            print("   ✅ Correctly handled invalid session ID")
        else:
            print("   ❌ Should have failed with invalid session ID")
            return False
        
        # Test empty topic
        print("   Testing empty topic...")
        try:
            response = await self.client.post(
                f"{self.base_url}/sessions/start",
                json={"topic": ""}
            )
            if response.status_code != 200:
                print("   ✅ Correctly rejected empty topic")
            else:
                print("   ❌ Should have rejected empty topic")
                return False
        except Exception as e:
            print(f"   ✅ Correctly handled empty topic error: {e}")
        
        return True
    
    async def run_all_tests(self) -> bool:
        """Run all tests in sequence."""
        print("🧪 Starting SkyTalk API Test Suite\n")
        
        tests = [
            ("Health Check", self.test_health_check),
            ("Root Endpoint", self.test_root_endpoint),
            ("Complete Conversation Flow", self.test_complete_conversation_flow),
            ("Error Conditions", self.test_error_conditions),
        ]
        
        passed = 0
        total = len(tests)
        
        for test_name, test_func in tests:
            print(f"\n{'='*50}")
            print(f"Running: {test_name}")
            print('='*50)
            
            try:
                if await test_func():
                    passed += 1
                    print(f"✅ {test_name} PASSED")
                else:
                    print(f"❌ {test_name} FAILED")
            except Exception as e:
                print(f"❌ {test_name} ERROR: {e}")
        
        print(f"\n{'='*50}")
        print(f"Test Results: {passed}/{total} tests passed")
        print('='*50)
        
        if passed == total:
            print("🎉 All tests passed! SkyTalk API is working correctly.")
        else:
            print("⚠️  Some tests failed. Check the logs above.")
        
        return passed == total
    
    async def close(self):
        """Close the HTTP client."""
        await self.client.aclose()


def start_api_server():
    """Start the API server in a separate process."""
    import uvicorn
    uvicorn.run(
        "app.main:app",
        host="127.0.0.1",
        port=8000,
        log_level="warning"  # Reduce log noise during testing
    )


async def main():
    """Main test function."""
    print("🌟 SkyTalk API Manual Test Script")
    print("=" * 50)
    
    # Check if we should start the server
    start_server = len(sys.argv) > 1 and sys.argv[1] == "--start-server"
    
    if start_server:
        print("🚀 Starting API server...")
        server_process = Process(target=start_api_server)
        server_process.start()
        
        # Wait for server to start
        print("⏳ Waiting for server to start...")
        await asyncio.sleep(5)
    
    try:
        # Run tests
        tester = SkyTalkAPITester()
        success = await tester.run_all_tests()
        await tester.close()
        
        if success:
            print("\n✅ Manual testing completed successfully!")
        else:
            print("\n❌ Some tests failed during manual testing.")
            
    finally:
        if start_server:
            print("🛑 Stopping API server...")
            server_process.terminate()
            server_process.join()
    
    return 0 if success else 1


if __name__ == "__main__":
    sys.exit(asyncio.run(main()))