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