#!/usr/bin/env python3 """ 🔥 LOAD TESTING SCRIPT ====================== Sunucuyu yoğun trafik altında test eder. 1000-4000 eşzamanlı kullanıcı simülasyonu. """ import asyncio import aiohttp import time import random import json from dataclasses import dataclass from typing import List import sys # Configuration API_URL = "http://botapi.componentmarket.org" # API_URL = "http://localhost:4000" # Local test için @dataclass class TestResult: endpoint: str status: int response_time: float success: bool error: str = None class LoadTester: def __init__(self, base_url: str): self.base_url = base_url self.results: List[TestResult] = [] self.start_time = None self.end_time = None async def make_request(self, session: aiohttp.ClientSession, endpoint: str, method: str = "GET", data: dict = None) -> TestResult: url = f"{self.base_url}{endpoint}" start = time.time() try: if method == "GET": async with session.get(url, timeout=aiohttp.ClientTimeout(total=30)) as response: await response.text() elapsed = (time.time() - start) * 1000 return TestResult(endpoint, response.status, elapsed, response.status < 400) else: async with session.post(url, json=data, timeout=aiohttp.ClientTimeout(total=30)) as response: await response.text() elapsed = (time.time() - start) * 1000 return TestResult(endpoint, response.status, elapsed, response.status < 400) except asyncio.TimeoutError: elapsed = (time.time() - start) * 1000 return TestResult(endpoint, 0, elapsed, False, "Timeout") except Exception as e: elapsed = (time.time() - start) * 1000 return TestResult(endpoint, 0, elapsed, False, str(e)[:100]) async def simulate_user_session(self, session: aiohttp.ClientSession, user_id: int): """Bir kullanıcı oturumunu simüle et""" results = [] # 1. Health check results.append(await self.make_request(session, "/api/health")) # 2. Store bots listesi results.append(await self.make_request(session, "/api/store/bots")) # 3. Random bot detayı bot_ids = ["sahibinden", "dolap", "trendyol", "hepsiburada"] bot_id = random.choice(bot_ids) results.append(await self.make_request(session, f"/api/store/bot/{bot_id}")) # 4. Heartbeat (simüle login) results.append(await self.make_request( session, "/api/user/heartbeat", "POST", {"userId": user_id, "machineId": f"machine_{user_id}", "currentPage": "launcher"} )) # 5. Update check results.append(await self.make_request( session, "/api/updates/check-delta/launcher?currentVersion=1.0.0" )) return results async def simulate_update_download(self, session: aiohttp.ClientSession, user_id: int): """Güncelleme indirme simülasyonu - daha yoğun""" results = [] # Manifest al results.append(await self.make_request(session, "/api/updates/manifest/launcher")) # Delta check results.append(await self.make_request( session, "/api/updates/check-delta/launcher?currentVersion=0.0.1" )) # Birkaç dosya indir (simülasyon) for i in range(3): results.append(await self.make_request(session, "/api/health")) await asyncio.sleep(0.1) return results async def run_concurrent_users(self, num_users: int, scenario: str = "normal"): """Belirtilen sayıda kullanıcıyı eşzamanlı çalıştır""" print(f"\n🚀 {num_users} kullanıcı başlatılıyor ({scenario} senaryo)...") self.start_time = time.time() connector = aiohttp.TCPConnector(limit=min(num_users, 500), limit_per_host=min(num_users, 300)) async with aiohttp.ClientSession(connector=connector) as session: if scenario == "normal": tasks = [self.simulate_user_session(session, i) for i in range(num_users)] elif scenario == "update": tasks = [self.simulate_update_download(session, i) for i in range(num_users)] else: # burst tasks = [self.make_request(session, "/api/health") for _ in range(num_users)] results_nested = await asyncio.gather(*tasks, return_exceptions=True) for result in results_nested: if isinstance(result, list): self.results.extend(result) elif isinstance(result, TestResult): self.results.append(result) elif isinstance(result, Exception): self.results.append(TestResult("unknown", 0, 0, False, str(result)[:100])) self.end_time = time.time() def print_report(self): """Test sonuçlarını raporla""" if not self.results: print("❌ Sonuç yok!") return total = len(self.results) successful = sum(1 for r in self.results if r.success) failed = total - successful response_times = [r.response_time for r in self.results if r.success] avg_time = sum(response_times) / len(response_times) if response_times else 0 min_time = min(response_times) if response_times else 0 max_time = max(response_times) if response_times else 0 # Percentiles sorted_times = sorted(response_times) p50 = sorted_times[int(len(sorted_times) * 0.5)] if sorted_times else 0 p95 = sorted_times[int(len(sorted_times) * 0.95)] if sorted_times else 0 p99 = sorted_times[int(len(sorted_times) * 0.99)] if sorted_times else 0 duration = self.end_time - self.start_time rps = total / duration if duration > 0 else 0 # Status code breakdown status_counts = {} for r in self.results: status_counts[r.status] = status_counts.get(r.status, 0) + 1 # Error breakdown error_counts = {} for r in self.results: if r.error: error_counts[r.error[:30]] = error_counts.get(r.error[:30], 0) + 1 print("\n" + "=" * 60) print("📊 LOAD TEST RAPORU") print("=" * 60) print(f""" 📈 GENEL İSTATİSTİKLER Toplam istek: {total:,} Başarılı: {successful:,} ({successful/total*100:.1f}%) Başarısız: {failed:,} ({failed/total*100:.1f}%) Test süresi: {duration:.2f} saniye İstek/saniye: {rps:.1f} RPS ⏱️ RESPONSE TIME Ortalama: {avg_time:.1f} ms Minimum: {min_time:.1f} ms Maximum: {max_time:.1f} ms P50 (medyan): {p50:.1f} ms P95: {p95:.1f} ms P99: {p99:.1f} ms 📊 STATUS CODE DAĞILIMI""") for status, count in sorted(status_counts.items()): status_name = { 200: "OK", 201: "Created", 400: "Bad Request", 401: "Unauthorized", 404: "Not Found", 429: "Too Many Requests", 500: "Server Error", 503: "Service Unavailable", 0: "Connection Error" }.get(status, "Unknown") print(f" {status} ({status_name}): {count:,}") if error_counts: print("\n❌ HATALAR") for error, count in sorted(error_counts.items(), key=lambda x: -x[1])[:5]: print(f" {error}: {count}") print("\n" + "=" * 60) # Sonuç değerlendirmesi if failed/total > 0.1: print("⚠️ UYARI: Hata oranı %10'un üzerinde!") elif p99 > 5000: print("⚠️ UYARI: P99 response time 5 saniyenin üzerinde!") elif avg_time > 1000: print("⚠️ UYARI: Ortalama response time 1 saniyenin üzerinde!") else: print("✅ BAŞARILI: Sunucu yükü kaldırıyor!") async def main(): print("=" * 60) print("🔥 LOAD TESTING - Bot Launcher API") print("=" * 60) print(f"🌐 Hedef: {API_URL}") tester = LoadTester(API_URL) # Test 1: Isınma - 10 kullanıcı print("\n" + "-" * 40) print("🔥 TEST 1: Isınma (10 kullanıcı)") await tester.run_concurrent_users(10, "normal") tester.print_report() # Test 2: Orta yük - 100 kullanıcı print("\n" + "-" * 40) print("🔥 TEST 2: Orta Yük (100 kullanıcı)") tester = LoadTester(API_URL) await tester.run_concurrent_users(100, "normal") tester.print_report() # Test 3: Yüksek yük - 500 kullanıcı print("\n" + "-" * 40) print("🔥 TEST 3: Yüksek Yük (500 kullanıcı)") tester = LoadTester(API_URL) await tester.run_concurrent_users(500, "normal") tester.print_report() # Test 4: Burst - 1000 anlık istek print("\n" + "-" * 40) print("🔥 TEST 4: Burst (1000 anlık istek)") tester = LoadTester(API_URL) await tester.run_concurrent_users(1000, "burst") tester.print_report() # Test 5: Güncelleme senaryosu - 100 kullanıcı print("\n" + "-" * 40) print("🔥 TEST 5: Güncelleme Senaryosu (100 kullanıcı)") tester = LoadTester(API_URL) await tester.run_concurrent_users(100, "update") tester.print_report() print("\n" + "=" * 60) print("✅ TÜM TESTLER TAMAMLANDI") print("=" * 60) if __name__ == "__main__": # Windows için event loop policy if sys.platform == 'win32': asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy()) asyncio.run(main())