skills/computation-audit/scripts/test_experiments.py

import argparse
from contextlib import redirect_stdout
import copy
import io
import json
import os
from pathlib import Path
import signal
import sys
import tempfile
import unittest
from unittest import mock

from run_experiment import execute, main as run_main, sha
from validate_manifest import legacy_limits, main as validate_main, validate


class ExperimentTests(unittest.TestCase):
    def setUp(self):
        self.temp = tempfile.TemporaryDirectory()
        self.addCleanup(self.temp.cleanup)
        self.root = Path(self.temp.name)
        self.contract = self.root / "contract.json"
        self.contract.write_text(json.dumps({"claim_id":"A", "mathematics":{
            "assertion_tested":"The listed integer products are even", "coefficient_domain":"Z",
            "conventions":"ordinary multiplication", "inputs":["run.py"], "bounds":{"n":[0,20]},
            "non_claims":["Not a universal proof"]},
            "execution_artifacts":["run.py"],
            "software":[{"name":"computation Python", "version":sys.version.split()[0]}]}))
        self.code = self.root / "run.py"
        self.code.write_text("assert all(n*(n+1)%2 == 0 for n in range(21))\nprint('21 cases checked')\n")

    def args(self, **extra):
        values = dict(root=self.root, contract=self.contract, output="runs/one", input=["run.py"],
                      result=[], timeout=2.0, max_output_bytes=4096, max_memory_bytes=None,
                      max_cpu_seconds=None, max_cores=None, max_threads=None,
                      argv=["--", sys.executable, "run.py"])
        values.update(extra)
        return argparse.Namespace(**values)

    def test_actual_run_records_hashes_and_no_automatic_proof(self):
        manifest, path = execute(self.args())
        self.assertEqual(manifest["run"]["status"], "completed")
        self.assertEqual(manifest["command"], [sys.executable, "run.py"])
        self.assertEqual(manifest["input_artifacts"][0]["sha256"], manifest["input_artifacts"][0]["sha256_after"])
        self.assertEqual(validate(manifest, root=self.root), [])
        self.assertTrue(path.exists())
        self.assertEqual(manifest["mathematics"]["bounds"], {"n":[0,20]})
        self.assertEqual(manifest["execution_artifacts"], ["run.py"])

    def test_failed_assertion_is_recorded_failure(self):
        self.code.write_text("assert False, 'mathematical benchmark failed'\n")
        manifest, _ = execute(self.args())
        self.assertEqual(manifest["run"]["status"], "failed")
        self.assertNotEqual(manifest["run"]["exit_status"], 0)
        self.assertEqual(validate(manifest), [])

    def test_timeout_has_manifest(self):
        self.code.write_text("import time\ntime.sleep(30)\n")
        manifest, path = execute(self.args(timeout=0.1))
        self.assertEqual(manifest["run"]["status"], "timeout")
        self.assertTrue(path.exists())
        self.assertLess(manifest["run"]["runtime_seconds"], 5)

    def test_unbounded_output_is_stopped(self):
        self.code.write_text("while True: print('x' * 1000)\n")
        manifest, _ = execute(self.args(max_output_bytes=1000))
        self.assertEqual(manifest["run"]["status"], "output-limit")
        total = sum((self.root / o["path"]).stat().st_size for o in manifest["outputs"])
        self.assertLessEqual(total, 1000)

    def test_input_mutation_is_not_completed_evidence(self):
        self.code.write_text("from pathlib import Path\nPath('run.py').write_text('changed')\n")
        manifest, _ = execute(self.args())
        self.assertEqual(manifest["run"]["status"], "inputs-changed")

    def test_launch_failure_is_recorded(self):
        manifest, _ = execute(self.args(argv=["no-such-mathbox-test-executable"]))
        self.assertEqual(manifest["run"]["status"], "launch-failed")
        self.assertEqual(manifest["run"]["exit_status"], 127)

    def test_never_overwrites_a_run(self):
        _, path = execute(self.args())
        before = path.read_bytes()
        with self.assertRaises(ValueError):
            execute(self.args())
        self.assertEqual(path.read_bytes(), before)

    def test_empty_template_rejected_as_evidence(self):
        template = json.loads((Path(__file__).parents[1] / "assets/computation-manifest.json").read_text())
        self.assertTrue(validate(template))
        self.assertEqual(validate(template, template=True), [])

    def test_tampered_logs_fail_validation(self):
        manifest, _ = execute(self.args())
        (self.root / manifest["outputs"][0]["path"]).write_text("forged success")
        self.assertTrue(validate(manifest, root=self.root))

    def test_declared_scientific_result_is_hashed_and_tampering_is_detected(self):
        self.code.write_text(
            "from pathlib import Path\n"
            "Path('runs/one/table.json').write_text('{\\\"verified_cases\\\": 21}\\n')\n"
        )
        manifest, _ = execute(self.args(result=["runs/one/table.json"]))
        self.assertEqual(manifest["run"]["status"], "completed")
        result = next(output for output in manifest["outputs"] if output.get("kind") == "result")
        self.assertEqual(result["path"], "runs/one/table.json")
        self.assertEqual(result["sha256"], sha(self.root / result["path"]))
        (self.root / result["path"]).write_text("changed\n")
        self.assertTrue(validate(manifest, root=self.root))

    def test_cli_accepts_declared_results(self):
        self.code.write_text(
            "from pathlib import Path\n"
            "Path('runs/one/result.txt').write_text('synthetic result\\n')\n"
        )
        output = io.StringIO()
        with redirect_stdout(output):
            code = run_main([
                "--root", str(self.root), "--contract", str(self.contract),
                "--output", "runs/one", "--input", "run.py",
                "--result", "runs/one/result.txt", "--", sys.executable, "run.py",
            ])
        self.assertEqual(code, 0)
        response = json.loads(output.getvalue())
        self.assertEqual(response["status"], "completed")
        manifest = json.loads(Path(response["manifest"]).read_text())
        self.assertEqual(manifest["declared_results"], ["runs/one/result.txt"])

    def test_missing_or_nonregular_declared_result_has_specific_status(self):
        missing, _ = execute(self.args(result=["runs/one/missing.json"]))
        self.assertEqual(missing["run"]["status"], "result-missing")
        self.assertEqual(missing["run"]["missing_results"], ["runs/one/missing.json"])

        self.code.write_text("from pathlib import Path\nPath('runs/two/a-directory').mkdir(parents=True)\n")
        invalid, _ = execute(self.args(output="runs/two", result=["runs/two/a-directory"]))
        self.assertEqual(invalid["run"]["status"], "result-invalid")
        self.assertEqual(invalid["run"]["invalid_results"], ["runs/two/a-directory"])

    def test_result_paths_cannot_escape_or_collide_with_runner_files(self):
        for result in (["result.json"], ["runs/one/stdout.txt"], ["runs/one/manifest.json"],
                       ["runs/one"], ["runs/one/result.json", "runs/one/result.json"]):
            with self.assertRaises(ValueError):
                execute(self.args(result=result))
        self.assertFalse((self.root / "runs").exists())

    def test_stale_or_missing_input_provenance_is_detected(self):
        manifest, _ = execute(self.args())
        malformed = copy.deepcopy(manifest)
        malformed["input_artifacts"] = [{}]
        self.assertTrue(validate(malformed))
        self.code.write_text("print('different computation')")
        self.assertTrue(validate(manifest, root=self.root))

    def test_contract_failure_has_no_run_side_effect(self):
        self.contract.write_text('{}')
        with self.assertRaises(ValueError):
            execute(self.args())
        self.assertFalse((self.root / "runs").exists())

    def test_declared_execution_artifacts_must_match_cli_pins(self):
        contract = json.loads(self.contract.read_text())
        contract["execution_artifacts"] = ["run.py", "missing-data.json"]
        self.contract.write_text(json.dumps(contract))
        with self.assertRaises(ValueError):
            execute(self.args())
        self.assertFalse((self.root / "runs").exists())

    def test_command_metacharacters_remain_literal(self):
        self.code.write_text("import sys\nassert sys.argv[1] == '$(touch injected); echo unsafe'\n")
        manifest, _ = execute(self.args(argv=[sys.executable, "run.py", "$(touch injected); echo unsafe"]))
        self.assertEqual(manifest["run"]["status"], "completed")
        self.assertFalse((self.root / "injected").exists())

    def test_escape_and_invalid_resource_limit_rejected(self):
        for changes in ({"output":"../escape"}, {"input":["../escape"]}, {"timeout":float("nan")},
                        {"max_output_bytes":0}, {"max_memory_bytes":0}, {"max_cpu_seconds":-1},
                        {"max_cores":0}, {"max_threads":0}):
            with self.assertRaises(ValueError):
                execute(self.args(**changes))
        self.assertFalse((self.root / "runs").exists())

    def test_historical_v1_shape_with_software_strings_remains_supported(self):
        archive = self.root / "archive" / "synthetic-run"
        archive.mkdir(parents=True)
        artifact = archive / "legacy-output.txt"
        artifact.write_text("12 synthetic cases checked\n")
        old = {
            "schema_version": 1,
            "claim_id": "SYNTHETIC-V1",
            "repository": {"commit": None, "dirty": False},
            "command": "python3 finite_check.py --bound 12",
            "environment": {"software": ["CPython 3.10 (standard library)"], "hardware": "unrecorded"},
            "mathematics": {
                "assertion_tested": "A finite synthetic identity holds in the listed range",
                "coefficient_domain": "exact integers",
                "conventions": "zero-based indexing",
                "inputs": ["indices 0 through 11"],
                "bounds": {"index_max": 11},
                "non_claims": ["This finite run is not a universal proof"],
            },
            "randomness": {"used": False, "generator": "", "seed": None},
            "run": {"started_at": "2025-01-01T00:00:00Z", "runtime_seconds": 0.25, "exit_status": 0},
            "outputs": [{"path": "legacy-output.txt", "sha256": sha(artifact)}],
            "checks": [{"check": "synthetic benchmark", "passed": True}],
            "result": "Finite synthetic check completed.",
            "residual_risks": ["No hashes of execution inputs were recorded."],
        }
        manifest_path = archive / "legacy-manifest.json"
        manifest_path.write_text(json.dumps(old))
        for archived_commit in (None, "", "unavailable"):
            old["repository"]["commit"] = archived_commit
            self.assertEqual(validate(old, root=self.root, manifest_path=manifest_path), [])
        old["repository"]["commit"] = None
        self.assertTrue(legacy_limits(old))
        output = io.StringIO()
        with redirect_stdout(output):
            self.assertEqual(validate_main([str(manifest_path), "--root", str(self.root)]), 0)
        self.assertIn("valid legacy version 1", output.getvalue())
        self.assertIn("does not require hashes of execution inputs", output.getvalue())
        self.assertIn("does not identify a repository revision", output.getvalue())
        self.assertIn("resolved relative to the manifest directory", output.getvalue())

    def test_ambiguous_v1_output_resolution_is_rejected(self):
        archive = self.root / "archive"
        archive.mkdir()
        root_output = self.root / "same-name.txt"
        local_output = archive / "same-name.txt"
        root_output.write_text("root interpretation\n")
        local_output.write_text("manifest interpretation\n")
        record = {
            "schema_version": 1, "claim_id": "AMBIGUOUS-V1",
            "repository": {"commit": "unavailable", "dirty": False},
            "command": "python3 synthetic.py", "environment": {"software": ["CPython 3.10"]},
            "mathematics": {"assertion_tested": "synthetic finite assertion", "coefficient_domain": "Z",
                            "conventions": "fixed", "inputs": [], "bounds": {"n": 1},
                            "non_claims": ["not universal"]},
            "randomness": {"used": False},
            "run": {"started_at": "2025-01-01T00:00:00Z", "runtime_seconds": 0.1, "exit_status": 0},
            "outputs": [{"path": "same-name.txt", "sha256": sha(root_output)}],
            "checks": [], "result": "synthetic", "residual_risks": [],
        }
        errors = validate(record, root=self.root, manifest_path=archive / "manifest.json")
        self.assertTrue(any("ambiguous legacy output path" in error for error in errors))

    def test_v1_compatibility_does_not_weaken_v2_software_or_outputs(self):
        manifest, _ = execute(self.args())
        bad_software = copy.deepcopy(manifest)
        bad_software["environment"]["software"] = ["CPython 3.10"]
        self.assertTrue(validate(bad_software))
        bad_commit = copy.deepcopy(manifest)
        bad_commit["repository"]["commit"] = None
        self.assertTrue(validate(bad_commit))
        historical = copy.deepcopy(manifest)
        historical["schema_version"] = 1
        historical.pop("input_artifacts")
        historical["outputs"] = []
        self.assertTrue(validate(historical))

    @unittest.skipUnless(
        os.name == "posix" and hasattr(os, "sched_getaffinity") and hasattr(os, "sched_setaffinity"),
        "POSIX resource and affinity limits are required",
    )
    def test_optional_resource_and_thread_caps_are_applied_and_recorded(self):
        self.code.write_text(
            "import json, os, resource\n"
            "from pathlib import Path\n"
            "record = {'as': resource.getrlimit(resource.RLIMIT_AS)[0], "
            "'cpu': resource.getrlimit(resource.RLIMIT_CPU)[0], "
            "'cores': len(os.sched_getaffinity(0)), 'threads': os.environ['OMP_NUM_THREADS']}\n"
            "Path('runs/one/caps.json').write_text(json.dumps(record))\n"
        )
        manifest, _ = execute(self.args(
            result=["runs/one/caps.json"], max_memory_bytes=1024 ** 3,
            max_cpu_seconds=2, max_cores=1, max_threads=2,
        ))
        self.assertEqual(manifest["run"]["status"], "completed")
        observed = json.loads((self.root / "runs/one/caps.json").read_text())
        limits = manifest["run"]["resource_limits"]
        self.assertEqual(observed["as"], limits["memory_bytes"])
        self.assertEqual(observed["cpu"], limits["cpu_seconds"])
        self.assertEqual(observed["cores"], limits["max_cores"])
        self.assertEqual(observed["threads"], str(limits["max_threads"]))
        self.assertEqual(validate(manifest, root=self.root), [])

    @unittest.skipUnless(
        os.name == "posix" and hasattr(signal, "SIGXCPU"),
        "POSIX CPU-limit signals are required",
    )
    def test_cpu_limit_has_a_distinct_recorded_status(self):
        self.code.write_text(
            "import os, resource, signal\n"
            "assert resource.getrlimit(resource.RLIMIT_CPU)[0] == 1\n"
            "os.kill(os.getpid(), signal.SIGXCPU)\n"
        )
        # Leave room for CI scheduling and log draining after the child exits.
        manifest, _ = execute(self.args(max_cpu_seconds=1, timeout=10))
        self.assertEqual(manifest["run"]["status"], "resource-limit")
        self.assertTrue(any("CPU" in risk for risk in manifest["residual_risks"]))
        self.assertEqual(validate(manifest, root=self.root), [])

    @unittest.skipUnless(os.name == "posix", "process group termination is POSIX-specific")
    def test_descendant_holding_log_pipe_is_bounded(self):
        self.code.write_text("import subprocess, sys\nsubprocess.Popen([sys.executable, '-c', 'import time; time.sleep(30)'])\n")
        manifest, _ = execute(self.args(timeout=0.2))
        self.assertEqual(manifest["run"]["status"], "timeout")
        self.assertLess(manifest["run"]["runtime_seconds"], 5)

    @unittest.skipUnless(os.name == "posix", "process group termination is POSIX-specific")
    def test_group_signal_only_fires_while_the_group_is_alive(self):
        real_killpg = os.killpg
        with mock.patch("run_experiment.os.killpg") as killpg:
            completed, _ = execute(self.args())
        self.assertEqual(completed["run"]["status"], "completed")
        killpg.assert_not_called()
        self.code.write_text("import time\ntime.sleep(30)\n")
        with mock.patch("run_experiment.os.killpg", side_effect=real_killpg) as killpg:
            timed_out, _ = execute(self.args(output="runs/two", timeout=0.2))
        self.assertEqual(timed_out["run"]["status"], "timeout")
        self.assertTrue(killpg.called)

    @unittest.skipUnless(os.name == "posix", "process group termination is POSIX-specific")
    def test_failed_group_signal_never_rewrites_an_executed_run(self):
        self.code.write_text("import sys, time\nsys.stderr.write('partial diagnostics\\n')\n"
                             "sys.stderr.flush()\ntime.sleep(30)\n")
        with mock.patch("run_experiment.os.killpg", side_effect=PermissionError("operation not permitted")):
            manifest, _ = execute(self.args(timeout=0.3))
        self.assertEqual(manifest["run"]["status"], "timeout")
        self.assertIn("partial diagnostics", (self.root / "runs/one/stderr.txt").read_text())
        self.assertTrue(any("termination failed" in risk for risk in manifest["residual_risks"]))
        self.assertEqual(validate(manifest, root=self.root), [])

    def test_run_requires_the_computation_software_versions(self):
        contract = json.loads(self.contract.read_text())
        del contract["software"]
        self.contract.write_text(json.dumps(contract))
        with self.assertRaises(ValueError):
            execute(self.args())
        self.assertFalse((self.root / "runs").exists())

    def test_unversioned_software_record_is_not_evidence(self):
        manifest, _ = execute(self.args())
        manifest["environment"]["software"] = [{"name": "SageMath"}]
        self.assertTrue(validate(manifest))

    def test_missing_field_errors_name_the_absent_keys(self):
        manifest, _ = execute(self.args())
        del manifest["claim_id"]
        del manifest["mathematics"]["bounds"]
        errors = "; ".join(validate(manifest))
        self.assertIn("claim_id", errors)
        self.assertIn("bounds", errors)

    def test_missing_pinned_input_fails_the_on_disk_check(self):
        manifest, _ = execute(self.args())
        record = copy.deepcopy(manifest)
        record["run"].update(status="failed", exit_status=1)
        record["input_artifacts"][0]["sha256_after"] = None
        self.assertEqual(validate(record), [])
        self.code.unlink()
        self.assertTrue(validate(record, root=self.root))


if __name__ == "__main__":
    unittest.main()