DocForge2/tests/test_milestone5_task_evidence.py

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()