"""Tests for the coverage-registry tooling: pure logic plus a registry canary. No `e2e` marker, so these run without a proxy. They exercise the coverage math and the registry loader, and guard the checked-in registry against schema drift and duplicate ids. """ from __future__ import annotations from pathlib import Path import pytest from coverage_registry.collector import ( collect_markers, compute_coverage, render, render_json, render_loki, render_prometheus, ) from coverage_registry.registry import load_registry from coverage_registry.schema import ( GuardrailCell, LlmCell, LlmEndpoint, LoggingCell, Tier, loki_module_label, ) def _llm( cell_id: str, tier: Tier, subject_endpoint: LlmEndpoint = "chat_completions" ) -> LlmCell: return LlmCell( id=cell_id, module="llm", tier=tier, assertions=("works",), source="test", subject_endpoint=subject_endpoint, route="openai", capability="basic", streaming="nonstream", ) def test_compute_coverage_counts_covered_p0_and_gaps() -> None: cells = (_llm("llm.a", Tier.P0), _llm("llm.b", Tier.P0), _llm("llm.c", Tier.P1)) report = compute_coverage(cells, frozenset({"llm.a"})) assert (report.total, report.covered) == (3, 1) assert (report.p0_total, report.p0_covered) == (2, 1) assert report.p0_gaps == ("llm.b",) assert report.orphan_markers == () def test_orphan_marker_is_reported_not_counted() -> None: cells = (_llm("llm.a", Tier.P0),) report = compute_coverage(cells, frozenset({"llm.a", "llm.ghost"})) assert report.covered == 1 assert report.orphan_markers == ("llm.ghost",) def test_cell_claimed_only_by_a_skipped_test_is_uncovered() -> None: cells = (_llm("llm.a", Tier.P0), _llm("llm.b", Tier.P0)) report = compute_coverage( cells, frozenset({"llm.a"}), skipped_only=frozenset({"llm.b"}) ) assert (report.covered, report.p0_covered) == (1, 1) assert report.p0_gaps == ("llm.b",) assert report.skipped_markers == ("llm.b",) assert "only by skipped tests" in render(report) assert '"skipped_markers": [\n "llm.b"\n ]' in render_json(report) assert "litellm_e2e_coverage_skipped_markers 1" in render_prometheus(report) def test_skipped_marker_outside_the_registry_is_still_an_orphan() -> None: report = compute_coverage( (_llm("llm.a", Tier.P0),), frozenset(), skipped_only=frozenset({"llm.ghost"}) ) assert report.orphan_markers == ("llm.ghost",) assert report.skipped_markers == () def test_logging_and_guardrail_roll_up_into_one_module() -> None: cells = ( LoggingCell( id="logging.x", module="logging", tier=Tier.P0, assertions=("logs_spend",), source="t", event="success", exercised_on=("chat_completions",), ), GuardrailCell( id="guardrail.y", module="guardrail", tier=Tier.P1, assertions=("blocks",), source="t", hook_point="pre_call", exercised_on=("chat_completions",), ), ) report = compute_coverage(cells, frozenset()) logging_and_guardrails = next( m for m in report.modules if m.module == "Logging & Guardrails" ) assert logging_and_guardrails.total == 2 def test_llm_cells_roll_up_by_core_endpoint() -> None: cells = ( _llm("llm.chat", Tier.P0, "chat_completions"), _llm("llm.messages", Tier.P0, "messages"), _llm("llm.responses", Tier.P1, "responses"), _llm("llm.batches", Tier.P0, "batches"), _llm("llm.realtime", Tier.P1, "realtime"), ) report = compute_coverage(cells, frozenset({"llm.chat", "llm.batches"})) core = next(m for m in report.modules if m.module == "Core LLMs") non_core = next(m for m in report.modules if m.module == "Non-Core LLMs") assert (core.total, core.covered, core.p0_total, core.p0_covered) == (3, 1, 2, 1) assert ( non_core.total, non_core.covered, non_core.p0_total, non_core.p0_covered, ) == (2, 1, 1, 1) def test_text_render_uses_plain_coverage_language() -> None: report = compute_coverage( (_llm("llm.chat", Tier.P0), _llm("llm.batches", Tier.P0, "batches")), frozenset({"llm.chat"}), ) text = render(report) assert "COVERAGE" in text assert "Headline coverage: 1/2 (50.0%)" in text assert "P0 COVERED" not in text def test_json_render_exposes_module_coverage_for_grafana_jobs() -> None: report = compute_coverage( (_llm("llm.chat", Tier.P0), _llm("llm.batches", Tier.P0, "batches")), frozenset({"llm.chat"}), ) payload = render_json(report) assert '"coverage_percent": 50.0' in payload assert '"module": "Core LLMs"' in payload assert '"module": "Non-Core LLMs"' in payload def test_prometheus_render_exposes_module_coverage_timeseries() -> None: report = compute_coverage( (_llm("llm.chat", Tier.P0), _llm("llm.batches", Tier.P0, "batches")), frozenset({"llm.chat"}), ) metrics = render_prometheus(report) assert 'litellm_e2e_coverage_cells{module="Core LLMs",state="covered"} 1' in metrics assert 'litellm_e2e_coverage_percent{module="Core LLMs"} 100.000000' in metrics assert 'litellm_e2e_coverage_percent{module="Non-Core LLMs"} 0.000000' in metrics assert "litellm_e2e_coverage_orphan_markers 0" in metrics def test_loki_render_exposes_exact_stdout_lines_for_loki() -> None: report = compute_coverage( (_llm("llm.chat", Tier.P0), _llm("llm.batches", Tier.P0, "batches")), frozenset({"llm.chat"}), ) lines = render_loki(report).splitlines() assert len(lines) == 1 + len(report.modules) assert lines[0] == "COVERAGE_TOTAL percent=50.0 covered=1 total=2" assert ( lines[1] == "COVERAGE_MODULE module=core_llms percent=100.0 covered=1 total=1" ) assert ( lines[2] == "COVERAGE_MODULE module=non_core_llms percent=0.0 covered=0 total=1" ) assert [line.split("module=", 1)[1].split(" ", 1)[0] for line in lines[1:]] == [ loki_module_label(module.module) for module in report.modules ] assert all( " " not in line.split("module=", 1)[1].split(" ", 1)[0] for line in lines[1:] ) _MARKED_TESTS = ''' import pytest @pytest.mark.covers("llm.runs") def test_runs() -> None: pass @pytest.mark.skip(reason="stage red: product gap") @pytest.mark.covers("llm.skipped") def test_skipped() -> None: pass @pytest.mark.skipif(True, reason="credentials absent in this environment") @pytest.mark.covers("llm.skipif_true") def test_skipif_true() -> None: pass @pytest.mark.skipif(False, reason="credentials present in this environment") @pytest.mark.covers("llm.skipif_false") def test_skipif_false() -> None: pass @pytest.mark.skipif("True") @pytest.mark.covers("llm.skipif_string") def test_skipif_string_condition() -> None: pass @pytest.mark.covers("llm.shared") def test_shared_cell_runs() -> None: pass @pytest.mark.skip(reason="stage red: product gap") @pytest.mark.covers("llm.shared") def test_shared_cell_skipped() -> None: pass ''' _MODULE_LEVEL_SKIP = ''' import pytest pytestmark = pytest.mark.skipif(True, reason="whole module needs a session fixture") @pytest.mark.covers("llm.module_skipped") def test_module_level_skip() -> None: pass ''' def test_collection_counts_only_markers_on_tests_that_would_run( tmp_path: Path, ) -> None: """The collect-only pass is the numerator, so a test pytest would skip must not contribute its cell. A cell stays covered as long as one runnable test claims it.""" (tmp_path / "test_marked.py").write_text(_MARKED_TESTS) (tmp_path / "test_module_skip.py").write_text(_MODULE_LEVEL_SKIP) markers = collect_markers(tmp_path) assert markers.covered == frozenset( {"llm.runs", "llm.skipif_false", "llm.shared"} ) assert markers.skipped_only == frozenset( {"llm.skipped", "llm.skipif_true", "llm.skipif_string", "llm.module_skipped"} ) assert markers.collection_errors == () def test_real_registry_loads_and_ids_are_unique() -> None: cells = load_registry() ids = [c.id for c in cells] assert len(cells) > 250 assert len(ids) == len(set(ids)) assert any(c.id == "logging.prometheus.success.exports_metric" for c in cells) def test_load_registry_rejects_duplicate_ids(tmp_path: Path) -> None: row = ( "- {id: llm.dup, module: llm, tier: P0, assertions: [works], source: t, " "subject_endpoint: chat_completions, route: openai, capability: basic, streaming: nonstream}\n" ) (tmp_path / "a.yaml").write_text(row) (tmp_path / "b.yaml").write_text(row) with pytest.raises(ValueError, match="duplicate cell ids"): load_registry(tmp_path)