additional_tests.py

#!/usr/bin/env python3
"""
Additional BitTorrent Protocol Compliance Tests
"""

import asyncio
import aiohttp
import json
import time
import hashlib
import random
import bencode
import os
from dotenv import load_dotenv

# Load environment variables
load_dotenv('/app/frontend/.env')
BASE_URL = os.getenv('REACT_APP_BACKEND_URL', 'http://localhost:8001')
API_BASE = f"{BASE_URL}/api"

async def test_malformed_requests():
    """Test error handling for malformed requests"""
    print("🔍 Testing malformed request handling...")
    
    async with aiohttp.ClientSession() as session:
        # Test missing required parameters
        test_cases = [
            {"name": "Missing info_hash", "params": {"peer_id": "test", "port": 6881}},
            {"name": "Missing peer_id", "params": {"info_hash": "a" * 40, "port": 6881}},
            {"name": "Missing port", "params": {"info_hash": "a" * 40, "peer_id": "test"}},
            {"name": "Invalid port", "params": {"info_hash": "a" * 40, "peer_id": "test", "port": "invalid"}},
            {"name": "Invalid info_hash length", "params": {"info_hash": "short", "peer_id": "test", "port": 6881}},
        ]
        
        for test_case in test_cases:
            try:
                async with session.get(f"{API_BASE}/announce", params=test_case["params"]) as response:
                    if response.status == 500:
                        content = await response.read()
                        try:
                            decoded = bencode.decode(content)
                            if "failure reason" in decoded:
                                print(f"✅ {test_case['name']}: Proper error handling")
                            else:
                                print(f"❌ {test_case['name']}: No failure reason in response")
                        except:
                            print(f"❌ {test_case['name']}: Invalid bencoded error response")
                    else:
                        print(f"❌ {test_case['name']}: Expected error but got status {response.status}")
            except Exception as e:
                print(f"❌ {test_case['name']}: Exception - {e}")

async def test_peer_aggregation():
    """Test cross-protocol peer aggregation"""
    print("\n🔗 Testing cross-protocol peer aggregation...")
    
    info_hash = hashlib.sha1(b"test_aggregation").hexdigest()
    
    async with aiohttp.ClientSession() as session:
        # Add peers via HTTP
        http_peers = []
        for i in range(3):
            peer_id = f"-HTTP-{i:012d}"
            port = 6881 + i
            params = {
                'info_hash': info_hash,
                'peer_id': peer_id,
                'port': port,
                'uploaded': 0,
                'downloaded': 0,
                'left': 1000000,
                'event': 'started'
            }
            
            async with session.get(f"{API_BASE}/announce", params=params) as response:
                if response.status == 200:
                    http_peers.append({"peer_id": peer_id, "port": port})
        
        # Check swarm stats
        async with session.get(f"{API_BASE}/swarms") as response:
            if response.status == 200:
                data = await response.json()
                swarms = data.get('swarms', [])
                target_swarm = next((s for s in swarms if s['info_hash'] == info_hash), None)
                
                if target_swarm:
                    print(f"✅ Peer aggregation: {target_swarm['total_peers']} peers in swarm")
                    print(f"   Seeders: {target_swarm['seeders']}, Leechers: {target_swarm['leechers']}")
                else:
                    print("❌ Peer aggregation: Swarm not found")

async def test_bep3_compliance():
    """Test BEP-3 (BitTorrent Protocol) compliance"""
    print("\n📋 Testing BEP-3 compliance...")
    
    info_hash = hashlib.sha1(b"bep3_test").hexdigest()
    peer_id = "-BEP3-" + "0" * 14
    
    async with aiohttp.ClientSession() as session:
        params = {
            'info_hash': info_hash,
            'peer_id': peer_id,
            'port': 6881,
            'uploaded': 0,
            'downloaded': 0,
            'left': 1073741824,
            'event': 'started',
            'numwant': 50,
            'compact': 1  # Request compact format
        }
        
        async with session.get(f"{API_BASE}/announce", params=params) as response:
            if response.status == 200:
                content = await response.read()
                try:
                    decoded = bencode.decode(content)
                    
                    # Check BEP-3 required fields
                    required_fields = ['interval', 'complete', 'incomplete']
                    optional_fields = ['min interval', 'peers', 'tracker id']
                    
                    missing_required = [f for f in required_fields if f not in decoded]
                    present_optional = [f for f in optional_fields if f in decoded]
                    
                    if not missing_required:
                        print(f"✅ BEP-3 compliance: All required fields present")
                        print(f"   Optional fields: {present_optional}")
                        
                        # Check interval values
                        interval = decoded.get('interval', 0)
                        min_interval = decoded.get('min interval', 0)
                        
                        if interval > 0:
                            print(f"✅ Valid interval: {interval} seconds")
                        else:
                            print(f"❌ Invalid interval: {interval}")
                            
                        if min_interval > 0:
                            print(f"✅ Valid min interval: {min_interval} seconds")
                    else:
                        print(f"❌ BEP-3 compliance: Missing required fields: {missing_required}")
                        
                except Exception as e:
                    print(f"❌ BEP-3 compliance: Failed to decode response - {e}")
            else:
                print(f"❌ BEP-3 compliance: HTTP {response.status}")

async def test_performance_under_load():
    """Test performance under concurrent load"""
    print("\n⚡ Testing performance under load...")
    
    info_hash = hashlib.sha1(b"load_test").hexdigest()
    
    async def make_request(session, peer_num):
        peer_id = f"-LOAD-{peer_num:010d}"
        port = 6881 + (peer_num % 1000)
        
        params = {
            'info_hash': info_hash,
            'peer_id': peer_id,
            'port': port,
            'uploaded': random.randint(0, 1000000),
            'downloaded': random.randint(0, 1000000),
            'left': random.randint(0, 1000000),
            'event': random.choice(['started', 'completed', 'stopped'])
        }
        
        start_time = time.time()
        try:
            async with session.get(f"{API_BASE}/announce", params=params) as response:
                response_time = time.time() - start_time
                return response.status == 200, response_time
        except:
            return False, time.time() - start_time
    
    async with aiohttp.ClientSession() as session:
        # Test with 50 concurrent requests
        start_time = time.time()
        tasks = [make_request(session, i) for i in range(50)]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        total_time = time.time() - start_time
        
        successful = sum(1 for result in results if isinstance(result, tuple) and result[0])
        response_times = [result[1] for result in results if isinstance(result, tuple)]
        
        if response_times:
            avg_response_time = sum(response_times) / len(response_times)
            max_response_time = max(response_times)
            
            print(f"✅ Load test: {successful}/50 requests successful")
            print(f"   Total time: {total_time:.3f}s")
            print(f"   Avg response time: {avg_response_time:.3f}s")
            print(f"   Max response time: {max_response_time:.3f}s")
            print(f"   Requests/second: {50/total_time:.1f}")
        else:
            print("❌ Load test: No successful requests")

async def main():
    """Run additional tests"""
    print("🧪 Running Additional BitTorrent Protocol Tests")
    print("=" * 60)
    
    await test_malformed_requests()
    await test_peer_aggregation()
    await test_bep3_compliance()
    await test_performance_under_load()
    
    print("\n" + "=" * 60)
    print("✅ Additional tests completed")

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