(self)
| 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 test coverage detected