| 250 | # ============================================================================= |
| 251 | |
| 252 | class TestCircuitBreaker: |
| 253 | |
| 254 | def test_starts_closed(self): |
| 255 | cb = CircuitBreaker(failure_threshold=3) |
| 256 | assert cb.state == CircuitState.CLOSED |
| 257 | assert cb.is_open() is False |
| 258 | |
| 259 | def test_opens_after_threshold(self): |
| 260 | cb = CircuitBreaker(failure_threshold=3) |
| 261 | for _ in range(3): |
| 262 | cb.record_failure() |
| 263 | assert cb.is_open() is True |
| 264 | |
| 265 | def test_success_resets_failures(self): |
| 266 | cb = CircuitBreaker(failure_threshold=3) |
| 267 | cb.record_failure() |
| 268 | cb.record_failure() |
| 269 | cb.record_success() |
| 270 | assert cb.state == CircuitState.CLOSED |
| 271 | assert cb.is_open() is False |
| 272 | |
| 273 | def test_half_open_after_recovery(self): |
| 274 | cb = CircuitBreaker(failure_threshold=1, recovery_seconds=0.01) |
| 275 | cb.record_failure() |
| 276 | assert cb.is_open() is True |
| 277 | time.sleep(0.02) |
| 278 | assert cb.state == CircuitState.HALF_OPEN |
| 279 | |
| 280 | def test_thread_safe(self): |
| 281 | cb = CircuitBreaker(failure_threshold=100) |
| 282 | errors = [] |
| 283 | |
| 284 | def worker(): |
| 285 | try: |
| 286 | for _ in range(20): |
| 287 | cb.record_failure() |
| 288 | cb.record_success() |
| 289 | except Exception as e: |
| 290 | errors.append(e) |
| 291 | |
| 292 | threads = [threading.Thread(target=worker) for _ in range(5)] |
| 293 | for t in threads: t.start() |
| 294 | for t in threads: t.join() |
| 295 | assert len(errors) == 0 |
| 296 | |
| 297 | |
| 298 | # ============================================================================= |
nothing calls this directly
no outgoing calls
no test coverage detected