Add comparative Milestone 5 task evidence

tests/test_milestone5_task_evidence.py

@@ -0,0 +1,154 @@
+from __future__ import annotations
+
+import json
+import subprocess
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from typing import cast
+
+from tools.milestone5_task_evidence import (
+    MAX_GRAPH_INSPECTED_BYTES,
+    MAX_SOURCE_INSPECTED_BYTES,
+    MAX_TASK_RESPONSE_BYTES,
+    SMOKE_SOURCE_COUNT,
+    answer_key,
+    build_task_evidence,
+    fixture_tasks,
+)
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+class Milestone5TaskEvidenceTests(unittest.TestCase):
+    def test_answer_keys_are_fixed_and_independent_of_both_workflows(self) -> None:
+        tasks = fixture_tasks(SMOKE_SOURCE_COUNT)
+
+        self.assertEqual(
+            ("evidence.component_000", "evidence.component_008"),
+            answer_key(tasks[0], SMOKE_SOURCE_COUNT),
+        )
+        self.assertEqual(
+            ("evidence.component_014", "evidence.component_015"),
+            answer_key(tasks[1], SMOKE_SOURCE_COUNT),
+        )
+        self.assertEqual(
+            (
+                "evidence.component_017",
+                "evidence.component_008",
+                "evidence.component_004",
+                "evidence.component_002",
+                "evidence.component_001",
+            ),
+            answer_key(tasks[2], SMOKE_SOURCE_COUNT),
+        )
+
+    def test_smoke_evidence_is_exact_provenanced_and_bounded(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            evidence = build_task_evidence(
+                Path(directory).resolve(),
+                source_count=SMOKE_SOURCE_COUNT,
+                samples=1,
+            )
+
+        fixture = cast(dict[str, object], evidence["fixture"])
+        self.assertFalse(fixture["external_projects"])
+        self.assertFalse(fixture["self_hosting"])
+        self.assertFalse(fixture["production_bindings"])
+        tasks = cast(list[dict[str, object]], evidence["tasks"])
+        self.assertEqual(3, len(tasks))
+        for task in tasks:
+            answer = cast(list[str], task["answer_key"])
+            workflows = cast(dict[str, dict[str, object]], task["workflows"])
+            self.assertTrue(answer)
+            for name, workflow in workflows.items():
+                with self.subTest(task=task["task"], workflow=name):
+                    self.assertTrue(workflow["correct"])
+                    result = cast(dict[str, object], workflow["result"])
+                    self.assertEqual(answer, result["answer"])
+                    self.assertTrue(result["provenance"])
+                    self.assertLessEqual(
+                        cast(int, workflow["response_bytes"]),
+                        MAX_TASK_RESPONSE_BYTES,
+                    )
+                    inspected_limit = (
+                        MAX_GRAPH_INSPECTED_BYTES
+                        if name == "graph_assisted"
+                        else MAX_SOURCE_INSPECTED_BYTES
+                    )
+                    self.assertLessEqual(
+                        cast(int, workflow["inspected_bytes"]),
+                        inspected_limit,
+                    )
+            comparison = cast(dict[str, object], task["comparison"])
+            self.assertTrue(comparison["both_exact"])
+
+    def test_semantic_evidence_is_repeatable_while_timings_remain_measurements(self) -> None:
+        hashes: list[str] = []
+        for _ in range(2):
+            with tempfile.TemporaryDirectory() as directory:
+                evidence = build_task_evidence(
+                    Path(directory).resolve(),
+                    source_count=SMOKE_SOURCE_COUNT,
+                    samples=1,
+                )
+            summary = cast(dict[str, object], evidence["summary"])
+            hashes.append(cast(str, summary["semantic_evidence_sha256"]))
+
+        self.assertEqual(hashes[0], hashes[1])
+
+    def test_source_only_scope_reflects_each_task_algorithm(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            evidence = build_task_evidence(
+                Path(directory).resolve(),
+                source_count=SMOKE_SOURCE_COUNT,
+                samples=1,
+            )
+
+        tasks = cast(list[dict[str, object]], evidence["tasks"])
+        direct = cast(
+            dict[str, object],
+            cast(dict[str, object], tasks[0]["workflows"])["source_only"],
+        )
+        impact = cast(
+            dict[str, object],
+            cast(dict[str, object], tasks[1]["workflows"])["source_only"],
+        )
+        path = cast(
+            dict[str, object],
+            cast(dict[str, object], tasks[2]["workflows"])["source_only"],
+        )
+        self.assertLess(
+            cast(int, direct["inspected_bytes"]),
+            cast(int, impact["inspected_bytes"]),
+        )
+        self.assertEqual(impact["inspected_bytes"], path["inspected_bytes"])
+
+    def test_smoke_cli_emits_the_same_machine_readable_report_it_writes(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            output = Path(directory) / "evidence.json"
+            completed = subprocess.run(
+                [
+                    sys.executable,
+                    "tools/milestone5_task_evidence.py",
+                    "--mode",
+                    "smoke",
+                    "--output",
+                    str(output),
+                ],
+                cwd=ROOT,
+                check=True,
+                capture_output=True,
+                text=True,
+            )
+            report = json.loads(completed.stdout)
+
+            self.assertEqual(completed.stdout, output.read_text(encoding="utf-8"))
+            self.assertEqual("docforge2_milestone5_representative_tasks", report["benchmark"])
+            self.assertEqual("smoke", report["mode"])
+            self.assertEqual(SMOKE_SOURCE_COUNT, report["fixture"]["source_count"])
+
+
+if __name__ == "__main__":
+    unittest.main()