test_performance.py

#!/usr/bin/env python3
"""
Performance test script to verify caching is working
Tests multiple simultaneous requests to ensure no duplicates
"""

import asyncio
import aiohttp
import time
import sys

async def test_fundamentals(session, ticker):
    """Test fundamentals endpoint"""
    start_time = time.time()
    try:
        async with session.get(f'http://localhost:5000/api/stocks/{ticker}/fundamentals') as response:
            data = await response.json()
            end_time = time.time()
            return {
                'ticker': ticker,
                'status': response.status,
                'time': end_time - start_time,
                'cached': 'quote' in data and data['quote'] is not None
            }
    except Exception as e:
        return {
            'ticker': ticker,
            'status': 'error',
            'time': time.time() - start_time,
            'error': str(e)
        }

async def test_multiple_requests():
    """Test multiple simultaneous requests"""
    print("🚀 Testing performance with multiple simultaneous requests...")
    
    # Test popular stocks that should be cached
    tickers = ['RELIANCE', 'TCS', 'INFY', 'HDFCBANK', 'ICICIBANK']
    
    async with aiohttp.ClientSession() as session:
        # Test 1: Single requests (should be fast due to cache)
        print("\n📊 Test 1: Single requests (cached)")
        start_time = time.time()
        
        tasks = [test_fundamentals(session, ticker) for ticker in tickers]
        results = await asyncio.gather(*tasks)
        
        total_time = time.time() - start_time
        
        print(f"Total time for {len(tickers)} requests: {total_time:.2f}s")
        for result in results:
            status = "✅" if result['status'] == 200 else "❌"
            print(f"{status} {result['ticker']}: {result['time']:.2f}s (status: {result['status']})")
        
        # Test 2: Multiple requests to same ticker (should use deduplication)
        print(f"\n🔄 Test 2: Multiple requests to same ticker (deduplication test)")
        start_time = time.time()
        
        # Send 5 simultaneous requests for the same ticker
        tasks = [test_fundamentals(session, 'RELIANCE') for _ in range(5)]
        results = await asyncio.gather(*tasks)
        
        total_time = time.time() - start_time
        
        print(f"Total time for 5 simultaneous RELIANCE requests: {total_time:.2f}s")
        for i, result in enumerate(results):
            status = "✅" if result['status'] == 200 else "❌"
            print(f"{status} Request {i+1}: {result['time']:.2f}s")
        
        # Test 3: AI endpoints
        print(f"\n🤖 Test 3: AI endpoints (should be cached)")
        start_time = time.time()
        
        ai_tasks = [
            session.get('http://localhost:5000/api/ai/analyze-stock/RELIANCE'),
            session.get('http://localhost:5000/api/ai/stock-news/TCS')
        ]
        
        ai_results = await asyncio.gather(*ai_tasks, return_exceptions=True)
        total_time = time.time() - start_time
        
        print(f"Total time for AI requests: {total_time:.2f}s")
        for i, result in enumerate(ai_results):
            if isinstance(result, Exception):
                print(f"❌ AI Request {i+1}: Error - {result}")
            else:
                print(f"✅ AI Request {i+1}: {result.status}")

if __name__ == "__main__":
    try:
        asyncio.run(test_multiple_requests())
        print("\n🎉 Performance test completed!")
        print("\n💡 If all requests are fast (<1s each), caching is working properly!")
        print("💡 If duplicate requests are fast, deduplication is working!")
    except KeyboardInterrupt:
        print("\n⏹️  Test interrupted")
        sys.exit(1)
    except Exception as e:
        print(f"\n❌ Test failed: {e}")
        sys.exit(1)