test_govchat_export.py

"""GovChat-Eval export tests (the independent-audit layer).

Built offline against the real suites and corpus with the mock model. The
export carries the safety-relevant ground truth into the black-box audit:
`should_refuse` for refuse_redirect cases, `forbidden_terms` for representational
PII/determination phrases, and `attack` tags for the jailbreak cases. These
assert that mapping is faithful, plus the helpers (literal-fact extraction,
accessible transcript, provenance, multilingual pairing) and the hashed write.
"""

from __future__ import annotations

import hashlib
import json

import pytest

from evals import govchat_export as gx

# ── helpers ──────────────────────────────────────────────────────────────────


class TestLiteralFacts:
    def test_plain_literal_passes_through(self):
        assert gx._literal_facts(["DD Form 214"], "bring a DD Form 214") == ["DD Form 214"]

    def test_regex_uses_the_substring_it_matched(self):
        facts = gx._literal_facts([r"re:\$\s?2\.00"], "the fare is $2.00 today")
        assert facts == ["$2.00"]

    def test_regex_miss_falls_back_to_readable_literal(self):
        # The pattern does not match the answer (a genuine gap); the fallback
        # derives a readable literal so the independent check flags it too.
        facts = gx._literal_facts([r"re:\$\s?2\.00"], "no figure here")
        assert facts == ["$ 2.00"]


class TestRenderTranscript:
    def test_escapes_html_and_declares_language(self):
        html = gx.render_transcript("<b>hi</b>", "answer & more", ["MST: Fares"], "es")
        assert 'lang="es"' in html
        assert "&lt;b&gt;hi&lt;/b&gt;" in html  # question escaped
        assert "answer &amp; more" in html  # answer escaped
        assert "<li>MST: Fares</li>" in html  # sources rendered as a list

    def test_no_sources_omits_the_sources_block(self):
        html = gx.render_transcript("q", "a", [], "en")
        assert "<h3>Sources</h3>" not in html


class TestProvenance:
    def test_uses_citation_agency_when_present(self):
        from assistant.answer import AnswerResult, Citation

        r = AnswerResult(
            question="q",
            answer="a",
            kind="answered",
            citations=[Citation("d", "MST", "Fares", "u", "2026-06-12")],
        )
        assert gx._provenance(r)["source"].startswith("MST")

    def test_falls_back_to_passage_agency_then_corpus(self):
        from assistant.answer import AnswerResult
        from assistant.ingest import Chunk
        from assistant.retrieve import ScoredChunk

        ch = Chunk(
            "c#0",
            "d",
            "MST",
            "Monterey-Salinas Transit",
            "Fares",
            "u",
            "2026-06-12",
            "en",
            "S",
            "t",
        )
        r = AnswerResult(
            question="q", answer="a", kind="answered", passages=[ScoredChunk(chunk=ch, score=1.0)]
        )
        assert "Monterey-Salinas Transit" in gx._provenance(r)["source"]
        empty = AnswerResult(question="q", answer="a", kind="refused_no_support")
        assert "corpus" in gx._provenance(empty)["source"]


class TestMultilingualPairing:
    def test_factual_mirror_is_paired_and_anchored(self):
        items = {"es-1": {}, "en-1": {}}
        cases = {
            "es-1": {"mirror_of": "en-1", "language": "es", "expected_behavior": "answer"},
            "en-1": {"expected_behavior": "answer", "language": "en"},
        }
        gx._pair_multilingual(items, cases)
        assert items["es-1"]["pair_id"] == "pair-en-1"
        assert items["en-1"]["is_reference"] is True

    def test_refusal_mirror_is_not_paired(self):
        # A Spanish case mirroring a refusal has no figures to preserve, so it is
        # left out of the anchor-fidelity pairing.
        items = {"es-2": {}, "en-2": {}}
        cases = {
            "es-2": {"mirror_of": "en-2", "language": "es", "expected_behavior": "refuse_redirect"},
            "en-2": {"expected_behavior": "refuse_redirect", "language": "en"},
        }
        gx._pair_multilingual(items, cases)
        assert "pair_id" not in items["es-2"]


# ── build_dataset (offline, real suites) ─────────────────────────────────────


@pytest.fixture(scope="module")
def dataset():
    return gx.build_dataset(offline=True)


def test_dataset_covers_every_single_turn_case(dataset):
    assert len(dataset) > 50
    for item in dataset:
        assert item["id"] and item["question"]
        assert "text" in item["target_response"]
        assert isinstance(item["should_refuse"], bool)
        assert item["transcript_html"].startswith("<section")


def test_refuse_redirect_cases_carry_should_refuse(dataset):
    # The audit's refusal suite keys off should_refuse; every refuse_redirect
    # case must set it, and answer/partial cases must not.
    refusing = [it for it in dataset if it["should_refuse"]]
    assert refusing, "expected refuse_redirect cases in the export"


def test_attack_cases_are_tagged_for_the_adversarial_suite(dataset):
    by_id = {it["id"]: it for it in dataset}
    for cid, kind in gx.ATTACK_CASES.items():
        if cid in by_id:  # case still present in the suites
            assert by_id[cid]["attack"] == kind


def test_forbidden_terms_propagate_when_present(dataset):
    # At least one case declares forbidden_content, and it rides into the export
    # as forbidden_terms for the representational check.
    with_terms = [it for it in dataset if it.get("forbidden_terms")]
    assert with_terms, "expected forbidden_terms on at least one exported case"


# ── write_dataset (redirected away from the committed golden file) ───────────


def test_write_dataset_emits_jsonl_and_sha256(tmp_path, monkeypatch):
    out = tmp_path / "golden.jsonl"
    monkeypatch.setattr(gx, "OUT_DIR", tmp_path)
    monkeypatch.setattr(gx, "DATASET_PATH", out)
    items = [{"id": "x", "question": "q?", "should_refuse": False}]
    gx.write_dataset(items)
    body = out.read_text()
    assert body.startswith("# fare-policy-assistant")  # header comment
    last = body.strip().splitlines()[-1]
    assert json.loads(last)["id"] == "x"
    sha = out.with_suffix(".jsonl.sha256").read_text().strip()
    assert len(sha) == 64  # sha256 hex digest


def test_main_offline_writes_dataset(tmp_path, monkeypatch):
    out = tmp_path / "golden.jsonl"
    monkeypatch.setattr(gx, "OUT_DIR", tmp_path)
    monkeypatch.setattr(gx, "DATASET_PATH", out)
    monkeypatch.setattr("sys.argv", ["govchat_export", "--offline"])
    gx.main()
    assert out.exists() and out.with_suffix(".jsonl.sha256").exists()


def test_write_dataset_emits_dataset_level_provenance(tmp_path, monkeypatch):
    # FIX-01/M-2: the dataset must declare, machine-readably, which answer-side
    # prompt versions and corpus it was recorded against, so the provenance
    # gate (evals/provenance.py) can catch a stale committed golden.jsonl.
    from evals import provenance

    out = tmp_path / "golden.jsonl"
    monkeypatch.setattr(gx, "OUT_DIR", tmp_path)
    monkeypatch.setattr(gx, "DATASET_PATH", out)
    gx.write_dataset([{"id": "x", "question": "q?", "should_refuse": False}])
    declared = provenance.read_golden(out.read_text(encoding="utf-8"))
    assert declared is not None, "no # provenance: header line emitted"
    assert declared["corpus_version"] == provenance.head_corpus_version()
    assert declared["prompt_versions"] == provenance.head_prompt_versions(provenance.ANSWER_PROMPTS)


# ── the license note (a non-grant, not a grant) ──────────────────────────────


class TestLicenseNote:
    """The dataset's `license` field is this project's only machine-readable
    statement about third-party text, and it is stamped on every row. It must
    not assert a license the project does not hold. Until 2026-08-12 it said
    "public record", which is a CPRA disclosure status, not a copyright grant.
    """

    def test_the_note_is_a_non_grant(self):
        note = gx.LICENSE_NOTE.lower()
        assert "no license granted" in note
        assert "not licensed for redistribution" in note
        assert "copyright of the respective transit agency" in note
        assert "corpus/license-note.md" in note
        # The old claim, and its neighbors, must not come back.
        assert "public record" not in note
        assert "public domain" not in note

    def test_every_exported_row_carries_it(self, dataset):
        assert dataset, "empty export"
        for item in dataset:
            assert item["provenance"]["license"] == gx.LICENSE_NOTE

    def test_the_committed_dataset_carries_it(self):
        # The committed golden.jsonl is what a downstream reuser actually reads,
        # so the gate is on the file, not only on the generator. If this fails
        # after a note change, run `make audit-restamp-license`.
        text = gx.DATASET_PATH.read_text(encoding="utf-8")
        rows = [json.loads(ln) for ln in text.splitlines() if ln.strip() and not ln.startswith("#")]
        assert rows
        stale = [r["id"] for r in rows if r.get("provenance", {}).get("license") != gx.LICENSE_NOTE]
        assert not stale, f"{len(stale)} committed rows carry a stale license note: {stale[:5]}"


class TestRestampLicense:
    def _fixture(self, tmp_path):
        path = tmp_path / "golden.jsonl"
        header = "# header line one\n# provenance: {}\n"
        rows = [
            {"id": "a", "provenance": {"source": "MST", "license": "public record — old"}},
            {"id": "b", "provenance": {"source": "SacRT", "license": "public record — old"}},
        ]
        body = "\n".join(json.dumps(r, ensure_ascii=False) for r in rows)
        path.write_text(header + body + "\n", encoding="utf-8")
        return path

    def test_rewrites_the_note_and_leaves_everything_else_alone(self, tmp_path):
        path = self._fixture(tmp_path)
        changed = gx.restamp_license(path)
        assert changed == 2
        lines = path.read_text(encoding="utf-8").splitlines()
        # The provenance header states what the answers were recorded against;
        # a metadata correction must not restate it.
        assert lines[0] == "# header line one"
        assert lines[1] == "# provenance: {}"
        rows = [json.loads(ln) for ln in lines if not ln.startswith("#")]
        assert [r["provenance"]["license"] for r in rows] == [gx.LICENSE_NOTE] * 2
        assert [r["provenance"]["source"] for r in rows] == ["MST", "SacRT"]

    def test_refreshes_the_sha256_sidecar(self, tmp_path):
        path = self._fixture(tmp_path)
        gx.restamp_license(path)
        sidecar = path.with_suffix(".jsonl.sha256").read_text().strip()
        assert sidecar == hashlib.sha256(path.read_bytes()).hexdigest()

    def test_is_idempotent(self, tmp_path):
        path = self._fixture(tmp_path)
        gx.restamp_license(path)
        before = path.read_bytes()
        assert gx.restamp_license(path) == 0
        assert path.read_bytes() == before

    def test_defaults_to_the_committed_dataset(self, tmp_path, monkeypatch):
        path = self._fixture(tmp_path)
        monkeypatch.setattr(gx, "DATASET_PATH", path)
        assert gx.restamp_license() == 2

    def test_main_restamp_flag_touches_no_model(self, tmp_path, monkeypatch):
        path = self._fixture(tmp_path)
        monkeypatch.setattr(gx, "DATASET_PATH", path)
        monkeypatch.setattr(
            gx, "build_dataset", lambda **kw: pytest.fail("restamp must not re-record")
        )
        monkeypatch.setattr("sys.argv", ["govchat_export", "--restamp-license"])
        gx.main()
        rows = [
            json.loads(ln)
            for ln in path.read_text(encoding="utf-8").splitlines()
            if not ln.startswith("#")
        ]
        assert all(r["provenance"]["license"] == gx.LICENSE_NOTE for r in rows)