| 1 | from __future__ import annotations |
| 2 | |
| 3 | import json |
| 4 | import tempfile |
| 5 | from pathlib import Path |
| 6 | |
| 7 | from scripts.observability_metrics import ( |
| 8 | METRICS_SCHEMA_VERSION, |
| 9 | AnalysisMetrics, |
| 10 | CrawlMetrics, |
| 11 | MapReduceMetrics, |
| 12 | ObservabilityLedger, |
| 13 | duration_p95, |
| 14 | emit_ledger, |
| 15 | validate_ledger, |
| 16 | ) |
| 17 | |
| 18 | |
| 19 | def sample_ledger() -> ObservabilityLedger: |
| 20 | return ObservabilityLedger( |
| 21 | schema_version=METRICS_SCHEMA_VERSION, |
| 22 | run_id="local", |
| 23 | week="2026-W21", |
| 24 | timestamp="2026-05-20T12:00:00Z", |
| 25 | crawl_metrics=[ |
| 26 | CrawlMetrics( |
| 27 | duration_seconds=12.4, |
| 28 | duration_p95_seconds=12.4, |
| 29 | duration_sample_count=1, |
| 30 | api_calls=10, |
| 31 | cache_hits=5, |
| 32 | cache_misses=10, |
| 33 | stale_cache_hits=1, |
| 34 | rate_limit_events=2, |
| 35 | secondary_rate_limit_hit=False, |
| 36 | source_type="github", |
| 37 | ) |
| 38 | ], |
| 39 | analysis_metrics=AnalysisMetrics( |
| 40 | duration_seconds=3.2, |
| 41 | token_ledger={ |
| 42 | "input_tokens": 100, |
| 43 | "output_tokens": 25, |
| 44 | "total_tokens": 125, |
| 45 | "cost_usd": 0.0, |
| 46 | }, |
| 47 | map_stages=[ |
| 48 | MapReduceMetrics( |
| 49 | stage="new_repos", |
| 50 | duration_seconds=0.3, |
| 51 | input_tokens=50, |
| 52 | output_tokens=10, |
| 53 | cost_usd=0.0, |
| 54 | status="pass", |
| 55 | gate_failure_reasons=[], |
| 56 | ) |
| 57 | ], |
| 58 | reduce_stage=MapReduceMetrics( |
| 59 | stage="reduce", |
| 60 | duration_seconds=0.8, |
| 61 | input_tokens=10, |
| 62 | output_tokens=15, |
| 63 | cost_usd=0.0, |
| 64 | status="fail", |
| 65 | gate_failure_reasons=["AI provenance metadata missing"], |
| 66 | ), |
| 67 | ), |
| 68 | environment={"pipeline": "test"}, |
| 69 | ) |
| 70 | |
| 71 | |
| 72 | def test_validate_ledger_reports_missing_required_fields() -> None: |
| 73 | errors = validate_ledger({"schema_version": METRICS_SCHEMA_VERSION}) |
| 74 | |
| 75 | assert "run_id" in errors |
| 76 | assert "crawl_metrics" in errors |
| 77 | assert "environment" in errors |
| 78 | |
| 79 | |
| 80 | def test_duration_p95_uses_high_percentile_sample() -> None: |
| 81 | assert duration_p95([]) == 0.0 |
| 82 | assert duration_p95([0.5]) == 0.5 |
| 83 | assert duration_p95([0.2, 0.4, 0.6, 0.8, 1.0]) == 1.0 |
| 84 | |
| 85 | |
| 86 | def test_validate_ledger_rejects_schema_version_mismatch() -> None: |
| 87 | payload = { |
| 88 | "schema_version": "observability_v0", |
| 89 | "run_id": "local", |
| 90 | "week": "2026-W21", |
| 91 | "timestamp": "2026-05-20T12:00:00Z", |
| 92 | "crawl_metrics": [], |
| 93 | "environment": {}, |
| 94 | } |
| 95 | |
| 96 | errors = validate_ledger(payload) |
| 97 | assert any(e.startswith("schema_version") for e in errors), ( |
| 98 | f"Expected schema_version error, got: {errors}" |
| 99 | ) |
| 100 | |
| 101 | |
| 102 | def test_emit_ledger_writes_valid_json() -> None: |
| 103 | tests_root = Path(__file__).resolve().parent |
| 104 | with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir: |
| 105 | output_path = Path(tmpdir) / "observability.json" |
| 106 | emit_ledger(sample_ledger(), output_path) |
| 107 | |
| 108 | payload = json.loads(output_path.read_text(encoding="utf-8")) |
| 109 | |
| 110 | assert payload["schema_version"] == METRICS_SCHEMA_VERSION |
| 111 | assert payload["analysis_metrics"]["reduce_stage"]["status"] == "fail" |
| 112 | assert validate_ledger(payload) == [] |
| 113 | |
| 114 | |
| 115 | def test_representative_fixture_is_valid() -> None: |
| 116 | fixture_path = ( |
| 117 | Path(__file__).resolve().parent / "fixtures" / "observability" / "2026-W21-full-run.json" |
| 118 | ) |
| 119 | payload = json.loads(fixture_path.read_text(encoding="utf-8")) |
| 120 | |
| 121 | assert validate_ledger(payload) == [] |
| 122 | assert payload["environment"]["pass_fail_counts"]["reduce_fail"] == 1 |