240 lines
9.4 KiB
Python
240 lines
9.4 KiB
Python
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()
|