165 lines
4.5 KiB
Python
165 lines
4.5 KiB
Python
"""
|
|
Tests for Deployment Module - Stress Tester.
|
|
|
|
Tests load testing functionality.
|
|
"""
|
|
|
|
import pytest
|
|
import asyncio
|
|
from src.deployment.stress_tester import (
|
|
StressTester,
|
|
StressTestResult,
|
|
RequestResult
|
|
)
|
|
|
|
|
|
class TestRequestResult:
|
|
"""Tests for RequestResult dataclass."""
|
|
|
|
def test_successful_request(self):
|
|
"""Test successful request result."""
|
|
result = RequestResult(
|
|
request_id=1,
|
|
success=True,
|
|
duration_ms=100
|
|
)
|
|
|
|
assert result.success is True
|
|
assert result.error is None
|
|
|
|
def test_failed_request(self):
|
|
"""Test failed request result."""
|
|
result = RequestResult(
|
|
request_id=1,
|
|
success=False,
|
|
duration_ms=50,
|
|
error="Timeout"
|
|
)
|
|
|
|
assert result.success is False
|
|
assert result.error == "Timeout"
|
|
|
|
|
|
class TestStressTester:
|
|
"""Tests for StressTester."""
|
|
|
|
@pytest.fixture
|
|
def tester(self):
|
|
"""Create stress tester."""
|
|
return StressTester()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_load_test_success(self, tester):
|
|
"""Test successful load test."""
|
|
counter = [0]
|
|
|
|
async def simple_test():
|
|
counter[0] += 1
|
|
await asyncio.sleep(0.01)
|
|
|
|
result = await tester.run_load_test(
|
|
test_func=simple_test,
|
|
num_requests=5,
|
|
concurrent=2
|
|
)
|
|
|
|
assert result.total_requests == 5
|
|
assert result.successful_requests == 5
|
|
assert result.failed_requests == 0
|
|
assert counter[0] == 5
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_load_test_with_failures(self, tester):
|
|
"""Test load test with some failures."""
|
|
counter = [0]
|
|
|
|
async def failing_test():
|
|
counter[0] += 1
|
|
if counter[0] % 2 == 0:
|
|
raise Exception("Simulated failure")
|
|
await asyncio.sleep(0.01)
|
|
|
|
result = await tester.run_load_test(
|
|
test_func=failing_test,
|
|
num_requests=4,
|
|
concurrent=2
|
|
)
|
|
|
|
assert result.total_requests == 4
|
|
assert result.failed_requests == 2
|
|
assert result.successful_requests == 2
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_metrics_calculation(self, tester):
|
|
"""Test that metrics are calculated correctly."""
|
|
async def fast_test():
|
|
await asyncio.sleep(0.01) # 10ms
|
|
|
|
result = await tester.run_load_test(
|
|
test_func=fast_test,
|
|
num_requests=10,
|
|
concurrent=5
|
|
)
|
|
|
|
assert result.avg_duration_ms >= 10
|
|
assert result.min_duration_ms >= 10
|
|
assert result.requests_per_second > 0
|
|
|
|
def test_percentile_calculation(self, tester):
|
|
"""Test percentile calculation."""
|
|
sorted_data = [10, 20, 30, 40, 50, 60, 70, 80, 90, 100]
|
|
|
|
p50 = tester._percentile(sorted_data, 50)
|
|
p95 = tester._percentile(sorted_data, 95)
|
|
|
|
assert 50 <= p50 <= 60
|
|
assert 90 <= p95 <= 100
|
|
|
|
def test_format_report(self, tester):
|
|
"""Test report formatting."""
|
|
result = StressTestResult(
|
|
total_requests=100,
|
|
successful_requests=95,
|
|
failed_requests=5,
|
|
avg_duration_ms=50.5,
|
|
min_duration_ms=10,
|
|
max_duration_ms=200,
|
|
p50_duration_ms=45.0,
|
|
p95_duration_ms=150.0,
|
|
p99_duration_ms=180.0,
|
|
requests_per_second=25.5,
|
|
total_duration_sec=4.0,
|
|
concurrent_requests=10,
|
|
test_name="Test Run"
|
|
)
|
|
|
|
report = tester.format_report(result)
|
|
|
|
assert "Test Run" in report
|
|
assert "100" in report
|
|
assert "95" in report
|
|
assert "50.5" in report
|
|
|
|
|
|
class TestStressTestResult:
|
|
"""Tests for StressTestResult dataclass."""
|
|
|
|
def test_result_creation(self):
|
|
"""Test creating stress test result."""
|
|
result = StressTestResult(
|
|
total_requests=10,
|
|
successful_requests=8,
|
|
failed_requests=2,
|
|
avg_duration_ms=100.0,
|
|
min_duration_ms=50,
|
|
max_duration_ms=200,
|
|
p50_duration_ms=90.0,
|
|
p95_duration_ms=180.0,
|
|
p99_duration_ms=195.0,
|
|
requests_per_second=5.0,
|
|
total_duration_sec=2.0
|
|
)
|
|
|
|
assert result.total_requests == 10
|
|
assert result.errors == []
|