""" Stress Testing Module for Arthur. Validates system performance under load. """ import asyncio import time import logging import statistics from typing import Optional, List from dataclasses import dataclass, field from datetime import datetime, timezone logger = logging.getLogger("ArthurStress") @dataclass class RequestResult: """Result of a single request.""" request_id: int success: bool duration_ms: int error: Optional[str] = None @dataclass class StressTestResult: """Result of a stress test run.""" total_requests: int successful_requests: int failed_requests: int # Timing metrics (milliseconds) avg_duration_ms: float min_duration_ms: int max_duration_ms: int p50_duration_ms: float p95_duration_ms: float p99_duration_ms: float # Throughput requests_per_second: float total_duration_sec: float # Errors errors: List[str] = field(default_factory=list) # Test configuration concurrent_requests: int = 0 test_name: str = "" class StressTester: """ Stress testing for Arthur Agent. Tests: - Concurrent request handling - Response latency under load - Resource utilization - Rate limiter behavior """ def __init__(self): """Initialize stress tester.""" self._results: List[RequestResult] = [] async def run_load_test( self, test_func, num_requests: int = 10, concurrent: int = 5, delay_between_ms: int = 0 ) -> StressTestResult: """ Run a load test. Args: test_func: Async function to test num_requests: Total number of requests concurrent: Maximum concurrent requests delay_between_ms: Delay between request batches Returns: StressTestResult with metrics """ self._results = [] start_time = time.time() # Create semaphore for concurrency control semaphore = asyncio.Semaphore(concurrent) async def limited_request(request_id: int): async with semaphore: return await self._execute_request(request_id, test_func) # Run all requests tasks = [ limited_request(i) for i in range(num_requests) ] await asyncio.gather(*tasks) end_time = time.time() total_duration = end_time - start_time return self._calculate_metrics( total_duration=total_duration, concurrent=concurrent ) async def _execute_request( self, request_id: int, test_func ) -> RequestResult: """Execute a single test request.""" start = time.time() try: await test_func() duration_ms = int((time.time() - start) * 1000) result = RequestResult( request_id=request_id, success=True, duration_ms=duration_ms ) except Exception as e: duration_ms = int((time.time() - start) * 1000) result = RequestResult( request_id=request_id, success=False, duration_ms=duration_ms, error=str(e) ) self._results.append(result) return result def _calculate_metrics( self, total_duration: float, concurrent: int ) -> StressTestResult: """Calculate metrics from results.""" successful = [r for r in self._results if r.success] failed = [r for r in self._results if not r.success] durations = [r.duration_ms for r in self._results] if not durations: return StressTestResult( total_requests=0, successful_requests=0, failed_requests=0, avg_duration_ms=0, min_duration_ms=0, max_duration_ms=0, p50_duration_ms=0, p95_duration_ms=0, p99_duration_ms=0, requests_per_second=0, total_duration_sec=total_duration, concurrent_requests=concurrent ) sorted_durations = sorted(durations) return StressTestResult( total_requests=len(self._results), successful_requests=len(successful), failed_requests=len(failed), avg_duration_ms=statistics.mean(durations), min_duration_ms=min(durations), max_duration_ms=max(durations), p50_duration_ms=self._percentile(sorted_durations, 50), p95_duration_ms=self._percentile(sorted_durations, 95), p99_duration_ms=self._percentile(sorted_durations, 99), requests_per_second=len(self._results) / total_duration if total_duration > 0 else 0, total_duration_sec=total_duration, concurrent_requests=concurrent, errors=[r.error for r in failed if r.error] ) def _percentile(self, sorted_data: List[int], percentile: int) -> float: """Calculate percentile from sorted data.""" if not sorted_data: return 0 index = (len(sorted_data) - 1) * percentile / 100 lower = int(index) upper = min(lower + 1, len(sorted_data) - 1) weight = index - lower return sorted_data[lower] * (1 - weight) + sorted_data[upper] * weight def format_report(self, result: StressTestResult) -> str: """Format a human-readable report.""" lines = [ "=" * 60, f" STRESS TEST REPORT - {result.test_name or 'Unnamed'}", "=" * 60, "", f" Total Requests: {result.total_requests}", f" Successful: {result.successful_requests} ({result.successful_requests/result.total_requests*100:.1f}%)" if result.total_requests > 0 else "", f" Failed: {result.failed_requests}", f" Concurrent: {result.concurrent_requests}", "", " LATENCY (ms):", f" Average: {result.avg_duration_ms:.1f}", f" Min: {result.min_duration_ms}", f" Max: {result.max_duration_ms}", f" P50: {result.p50_duration_ms:.1f}", f" P95: {result.p95_duration_ms:.1f}", f" P99: {result.p99_duration_ms:.1f}", "", " THROUGHPUT:", f" Requests/sec: {result.requests_per_second:.2f}", f" Total time: {result.total_duration_sec:.2f}s", "", ] if result.errors: lines.append(" ERRORS:") for error in result.errors[:5]: # Show first 5 errors lines.append(f" - {error[:50]}") lines.append("=" * 60) return "\n".join(lines) async def run_dispatcher_stress_test( num_requests: int = 10, concurrent: int = 5 ) -> StressTestResult: """ Run stress test on the dispatcher. Args: num_requests: Total requests to send concurrent: Maximum concurrent requests Returns: StressTestResult """ from src.agents import get_dispatcher tester = StressTester() dispatcher = get_dispatcher() request_counter = [0] async def test_request(): request_counter[0] += 1 ticket_id = f"STRESS-{request_counter[0]:04d}" # Simulate a ticket processing await dispatcher.dispatch( ticket_id=ticket_id, sender_email="stress@oestepan.com.br", subject="[STRESS TEST] Teste de carga", body="Este é um teste automatizado de stress do sistema." ) result = await tester.run_load_test( test_func=test_request, num_requests=num_requests, concurrent=concurrent ) result.test_name = "Dispatcher Stress Test" # Print report print(tester.format_report(result)) return result async def run_rate_limiter_stress_test( num_requests: int = 20, concurrent: int = 10 ) -> StressTestResult: """ Run stress test on rate limiter. Args: num_requests: Total requests concurrent: Concurrent requests Returns: StressTestResult """ from src.agents import get_rate_limiter tester = StressTester() limiter = get_rate_limiter() async def test_request(): result = await limiter.check_limit("stress-tenant") if result.allowed: await asyncio.sleep(0.01) # Simulate work await limiter.release("stress-tenant") else: raise Exception(f"Rate limited: {result.reason}") result = await tester.run_load_test( test_func=test_request, num_requests=num_requests, concurrent=concurrent ) result.test_name = "Rate Limiter Stress Test" print(tester.format_report(result)) return result