(self, tmp_path)
| 444 | assert abs(logger.pass_rate() - 2/3) < 0.01 |
| 445 | |
| 446 | def test_thread_safe_writes(self, tmp_path): |
| 447 | logger = AuditLogger(str(tmp_path / "audit.jsonl")) |
| 448 | errors = [] |
| 449 | |
| 450 | def write(): |
| 451 | try: |
| 452 | for i in range(10): |
| 453 | logger.log(AuditRecord( |
| 454 | audit_id=str(i), timestamp="t", prompt_hash="h", |
| 455 | attempt=1, failure_mode=FailureMode.NONE, latency_ms=1.0, |
| 456 | token_count=10, passed=True, strategy=RetryStrategy.NONE, |
| 457 | )) |
| 458 | except Exception as e: |
| 459 | errors.append(e) |
| 460 | |
| 461 | threads = [threading.Thread(target=write) for _ in range(5)] |
| 462 | for t in threads: t.start() |
| 463 | for t in threads: t.join() |
| 464 | assert len(errors) == 0 |
| 465 | assert len(logger.all_records()) == 50 |
| 466 | |
| 467 | |
| 468 | # ============================================================================= |
nothing calls this directly
no test coverage detected