Make a single HTTP request and record results.
(
config: StressTestConfig, results: StressTestResults, client: httpx.Client
)
| 119 | |
| 120 | |
| 121 | def make_request( |
| 122 | config: StressTestConfig, results: StressTestResults, client: httpx.Client |
| 123 | ): |
| 124 | """Make a single HTTP request and record results.""" |
| 125 | url = f"http://{config.host}:{config.port}/" |
| 126 | |
| 127 | try: |
| 128 | start_time = time.perf_counter() |
| 129 | response = client.get(url, timeout=config.timeout) |
| 130 | elapsed_ms = (time.perf_counter() - start_time) * 1000 |
| 131 | |
| 132 | if response.status_code == 200: |
| 133 | results.record_success(elapsed_ms) |
| 134 | else: |
| 135 | results.record_failure("http_error") |
| 136 | |
| 137 | except httpx.ConnectError: |
| 138 | results.record_failure("connection") |
| 139 | except httpx.TimeoutException: |
| 140 | results.record_failure("timeout") |
| 141 | except KeyboardInterrupt: |
| 142 | _thread.interrupt_main() |
| 143 | except Exception: |
| 144 | results.record_failure("other") |
| 145 | |
| 146 | |
| 147 | def stress_test_concurrent( |
no test coverage detected