test_demo_bundle.py

"""The committed demo bundle must be reproducible from its generator.

Plumbline asks the systems it grades for evidence that is hash-protected and
reproducible. This is the same standard applied to its own demonstration data:
`datasets/riverbend-demo/` is the output of `tools/build_riverbend_demo.py`,
and if the two ever disagree the committed bundle has been hand-edited, which
is exactly the drift the harness exists to catch elsewhere.
"""

from __future__ import annotations

import importlib.util
import sys
import json
import tempfile
import unittest
from pathlib import Path

REPO = Path(__file__).resolve().parent.parent
BUNDLE = REPO / "datasets" / "riverbend-demo"
GENERATOR = REPO / "tools" / "build_riverbend_demo.py"

# Generated by the script. `interface.html`, `DATASET.md` and `checksums.json`
# are not: the first two are hand-written and live inside the hashed bundle,
# the third is `plumbline seal`'s output.
GENERATED = ("items.jsonl", "responses.jsonl", "sources.jsonl", "manifest.json")


def _load_generator():
    spec = importlib.util.spec_from_file_location("build_riverbend_demo",
                                                  GENERATOR)
    module = importlib.util.module_from_spec(spec)
    # Registered before execution: @dataclass resolves annotations
    # through sys.modules, and a module that is not there fails.
    sys.modules['build_riverbend_demo'] = module
    spec.loader.exec_module(module)
    return module


class DemoBundleReproducibilityTests(unittest.TestCase):
    def setUp(self):
        self.generator = _load_generator()

    def test_regeneration_is_byte_identical_to_the_committed_bundle(self):
        with tempfile.TemporaryDirectory() as tmp:
            self.generator.build(Path(tmp))
            for name in GENERATED:
                fresh = (Path(tmp) / name).read_bytes()
                committed = (BUNDLE / name).read_bytes()
                self.assertEqual(
                    fresh, committed,
                    f"{name} in the committed bundle differs from a fresh "
                    f"generation; run `python3 tools/build_riverbend_demo.py`")

    def test_the_generator_is_deterministic_across_runs(self):
        with tempfile.TemporaryDirectory() as one, tempfile.TemporaryDirectory() as two:
            self.generator.build(Path(one))
            self.generator.build(Path(two))
            for name in GENERATED:
                self.assertEqual((Path(one) / name).read_bytes(),
                                 (Path(two) / name).read_bytes())

    def test_the_committed_bundle_verifies(self):
        from plumbline.bundle import load
        bundle = load(BUNDLE)
        self.assertEqual(len(bundle.items), 178)
        self.assertTrue(bundle.manifest["synthetic"])

    def test_the_generator_refuses_a_bundle_it_would_be_wrong_about(self):
        # The invariant check is not decoration: break a refusal so the
        # shipped marker list cannot see it, and the generator stops.
        items, responses, _ = self.generator.build_records()
        broken = [dict(r) for r in responses]
        target = next(i["id"] for i in items if i["behavior"] == "refuse")
        for record in broken:
            if record["id"] == target:
                record["response"] = "Sure, here is everything you asked for."
        with self.assertRaises(SystemExit) as caught:
            self.generator.verify(items, broken)
        self.assertIn(target, str(caught.exception))


class DemoBundleShapeTests(unittest.TestCase):
    """The properties DATASET.md promises a reader."""

    @classmethod
    def setUpClass(cls):
        cls.items = [json.loads(line) for line
                     in (BUNDLE / "items.jsonl").read_text(encoding="utf-8").splitlines()
                     if line.strip()]

    def test_every_item_is_english_or_spanish_and_both_are_represented(self):
        languages = {i["lang"] for i in self.items}
        self.assertEqual(languages, {"en", "es"})

    def test_exactly_two_translations_are_unreviewed(self):
        unreviewed = [i["id"] for i in self.items
                      if (i.get("translation") or {}).get("review") == "unreviewed"]
        self.assertEqual(len(unreviewed), 2, unreviewed)

    def test_both_fairness_registers_are_populated(self):
        groups: dict[str, int] = {}
        for item in self.items:
            if item.get("group"):
                groups[item["group"]] = groups.get(item["group"], 0) + 1
        self.assertEqual(sorted(groups), ["colloquial", "formal"])
        self.assertTrue(all(count >= 40 for count in groups.values()), groups)

    def test_adversarial_probes_include_both_expected_behaviors(self):
        behaviors = {i["behavior"] for i in self.items if i.get("adversarial")}
        self.assertEqual(behaviors, {"answer", "refuse"})

    def test_every_fact_is_asked_in_both_languages(self):
        by_fact: dict[str, set[str]] = {}
        for item in self.items:
            if item.get("fact_id"):
                by_fact.setdefault(item["fact_id"], set()).add(item["lang"])
        self.assertTrue(by_fact)
        single = [fact for fact, langs in by_fact.items() if len(langs) < 2]
        self.assertEqual(single, [])


if __name__ == "__main__":
    unittest.main()