from __future__ import annotations import json import subprocess import sys import tempfile import unittest from pathlib import Path from typing import cast from unittest import mock from tools.milestone5_task_evidence import ( MAX_GRAPH_INSPECTED_BYTES, MAX_SOURCE_INSPECTED_BYTES, MAX_TASK_RESPONSE_BYTES, PINNED_REAL_SOURCE_BYTES, PINNED_REAL_SOURCE_COUNT, PINNED_REAL_SOURCE_SHA256, PINNED_REAL_VERSION, SMOKE_SOURCE_COUNT, TaskEvidenceError, answer_key, build_all_task_evidence, build_real_package_evidence, build_task_evidence, fixture_tasks, installed_real_package, real_package_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_real_package_identity_is_installed_lock_pinned_and_exact(self) -> None: _, identity = installed_real_package() self.assertEqual(PINNED_REAL_VERSION, identity["lock_version"]) self.assertEqual(PINNED_REAL_VERSION, identity["installed_version"]) self.assertEqual(PINNED_REAL_SOURCE_COUNT, identity["source_count"]) self.assertEqual(PINNED_REAL_SOURCE_BYTES, identity["source_bytes"]) self.assertEqual(PINNED_REAL_SOURCE_SHA256, identity["source_tree_sha256"]) def test_real_package_tasks_are_exact_provenanced_bounded_and_read_only(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory).resolve() evidence = build_real_package_evidence(root, samples=1) copied_sources = tuple((root / "src" / "markdown_it").rglob("*.py")) self.assertTrue(copied_sources) self.assertTrue(all(path.stat().st_mode & 0o222 == 0 for path in copied_sources)) fixture = cast(dict[str, object], evidence["fixture"]) self.assertFalse(fixture["network"]) self.assertFalse(fixture["installed_source_mutation"]) self.assertFalse(fixture["external_project_mutation"]) tasks = cast(list[dict[str, object]], evidence["tasks"]) self.assertEqual( [list(expected) for _, expected in real_package_tasks()], [task["answer_key"] for task in tasks], ) for task in tasks: workflows = cast(dict[str, dict[str, object]], task["workflows"]) for workflow in workflows.values(): self.assertTrue(workflow["correct"]) result = cast(dict[str, object], workflow["result"]) self.assertTrue(result["provenance"]) self.assertLessEqual( cast(int, workflow["response_bytes"]), MAX_TASK_RESPONSE_BYTES, ) comparison = cast(dict[str, object], task["comparison"]) self.assertTrue(comparison["both_exact"]) def test_real_package_pin_change_fails_closed(self) -> None: with ( mock.patch( "tools.milestone5_task_evidence._source_tree_identity", return_value=( PINNED_REAL_SOURCE_COUNT, PINNED_REAL_SOURCE_BYTES, "0" * 64, ), ), self.assertRaises(TaskEvidenceError), ): installed_real_package() def test_combined_semantic_evidence_is_repeatable_across_both_tracks(self) -> None: hashes: list[str] = [] for _ in range(2): with tempfile.TemporaryDirectory() as directory: evidence = build_all_task_evidence( Path(directory).resolve(), generated_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_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(2, report["schema_version"]) self.assertEqual("smoke", report["mode"]) self.assertEqual( SMOKE_SOURCE_COUNT, report["tracks"]["generated_scale"]["fixture"]["source_count"], ) self.assertEqual( PINNED_REAL_SOURCE_COUNT, report["tracks"]["installed_real_package"]["fixture"]["source_count"], ) self.assertEqual(6, report["summary"]["task_count"]) if __name__ == "__main__": unittest.main()