| 391 | # ============================================================================= |
| 392 | |
| 393 | class TestAuditLogger: |
| 394 | |
| 395 | def test_log_and_retrieve(self, tmp_path): |
| 396 | logger = AuditLogger(str(tmp_path / "audit.jsonl")) |
| 397 | record = AuditRecord( |
| 398 | audit_id="abc123", timestamp="2025-01-01T00:00:00Z", |
| 399 | prompt_hash="hash", attempt=1, failure_mode=FailureMode.NONE, |
| 400 | latency_ms=42.0, token_count=100, passed=True, |
| 401 | strategy=RetryStrategy.SIMPLE, |
| 402 | ) |
| 403 | logger.log(record) |
| 404 | records = logger.all_records() |
| 405 | assert len(records) == 1 |
| 406 | assert records[0].audit_id == "abc123" |
| 407 | |
| 408 | def test_persists_to_file(self, tmp_path): |
| 409 | path = str(tmp_path / "audit.jsonl") |
| 410 | logger = AuditLogger(path) |
| 411 | record = AuditRecord( |
| 412 | audit_id="xyz", timestamp="2025-01-01T00:00:00Z", |
| 413 | prompt_hash="h", attempt=1, failure_mode=FailureMode.NONE, |
| 414 | latency_ms=10.0, token_count=50, passed=True, |
| 415 | strategy=RetryStrategy.SIMPLE, |
| 416 | ) |
| 417 | logger.log(record) |
| 418 | with open(path) as f: |
| 419 | line = f.readline() |
| 420 | data = json.loads(line) |
| 421 | assert data["audit_id"] == "xyz" |
| 422 | |
| 423 | def test_failure_distribution(self, tmp_path): |
| 424 | logger = AuditLogger(str(tmp_path / "audit.jsonl")) |
| 425 | for mode in [FailureMode.SCHEMA_VIOLATION, FailureMode.SCHEMA_VIOLATION, |
| 426 | FailureMode.TIMEOUT]: |
| 427 | logger.log(AuditRecord( |
| 428 | audit_id="id", timestamp="t", prompt_hash="h", |
| 429 | attempt=1, failure_mode=mode, latency_ms=1.0, |
| 430 | token_count=10, passed=False, strategy=RetryStrategy.NONE, |
| 431 | )) |
| 432 | dist = logger.failure_distribution() |
| 433 | assert dist["schema_violation"] == 2 |
| 434 | assert dist["timeout"] == 1 |
| 435 | |
| 436 | def test_pass_rate(self, tmp_path): |
| 437 | logger = AuditLogger(str(tmp_path / "audit.jsonl")) |
| 438 | for passed in [True, True, False]: |
| 439 | logger.log(AuditRecord( |
| 440 | audit_id="id", timestamp="t", prompt_hash="h", |
| 441 | attempt=1, failure_mode=FailureMode.NONE, latency_ms=1.0, |
| 442 | token_count=10, passed=passed, strategy=RetryStrategy.NONE, |
| 443 | )) |
| 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(): |
nothing calls this directly
no outgoing calls
no test coverage detected