oak/benchmarks

"""Git/Oak committed-evidence parity matrix on real binaries (QA P1 A1/A4).

Every row states the same repository situation for each arm; the arms must

agree on measured pass/fail. "Unmeasured" (None) is allowed only for a

missing instrument, never for subject-controlled state or a failing subject

binary. Ported from the reviewer's stageA/parity.py.

"""

from __future__ import annotations

import importlib.util

import os

import shutil

import subprocess

import sys

import tempfile

import unittest

from pathlib import Path

from unittest import mock

ROOT = Path(__file__).resolve().parents[1]

sys.path.insert(0, str(ROOT / "scripts"))

from oakbench import workflow_oracles as O  # noqa: E402

from oakbench.subjects import which_measured_git  # noqa: E402

GIT = which_measured_git()

OAK = shutil.which("oak")

def reader_python() -> str | None:

    # The independent reader is stdlib-only: the harness interpreter always qualifies.

    return os.environ.get("OAK_BENCH_OAK_READER_PYTHON") or sys.executable

def run(*command, cwd):

    result = subprocess.run(list(command), cwd=cwd, capture_output=True, text=True,

                            env={**os.environ, "OAK_NO_UPDATE_CHECK": "1"})

    if result.returncode:

        raise RuntimeError(f"{command}: {result.stderr}")

def commit(kind, repo):

    if kind == "git":

        run(GIT, "add", "-A", cwd=repo)

        run(GIT, "-c", "user.email=a@b", "-c", "user.name=a", "commit", "-qm", "x", "--allow-empty", cwd=repo)

    else:

        run(OAK, "commit", "--no-verify", cwd=repo)

def init(kind, repo):

    (repo / "seed.txt").write_text("seed\n")

    run(*([GIT, "init", "-q"] if kind == "git" else [OAK, "init"]), cwd=repo)

    commit(kind, repo)

def write_payload(repo, scenario):

    expected = dict(O.COMMITTED_PAYLOADS[scenario])

    if scenario == "sync_pull_dirty":

        expected["work.txt"] = "dirty local edit"

    for relative, marker in expected.items():

        path = repo / relative

        path.parent.mkdir(parents=True, exist_ok=True)

        if path.is_symlink() or path.exists():

            path.unlink()

        path.write_text(marker + "\n")

        os.chmod(path, 0o644)

READER = reader_python()  # captured before setUp isolates the environment

@unittest.skipUnless(GIT and OAK, "real Git and Oak required")

class OracleParityMatrixTest(unittest.TestCase):

    def setUp(self):

        patcher = mock.patch.dict(os.environ)

        patcher.start()

        self.addCleanup(patcher.stop)

        os.environ.pop("OAK_BENCH_OAK_READER_PYTHON", None)

        os.environ.pop("OAK_BENCH_GIT_INSTRUMENT", None)

        # The "oak-export" arm is the explicit subject-produced opt-in; the

        # "oak-reader" arm passes the reader interpreter (the default arm).

        os.environ["OAK_BENCH_OAK_COMMITTED_READER"] = "export"

        self.tmp = Path(tempfile.mkdtemp(prefix="oakbench-parity-")).resolve()

        self.addCleanup(shutil.rmtree, self.tmp, True)

        self.stubs = self.tmp / "stubs"

        self.stubs.mkdir()

    def stub(self, name, command, code):

        path = self.stubs / name

        path.write_text(f'#!/bin/sh\n[ "$1" = {command} ] && {{ echo "simulated oak {command} bug" >&2; exit {code}; }}\n'

                        f'exec {OAK} "$@"\n')

        path.chmod(0o755)

        return str(path)

    def arms(self):

        yield "git", GIT, None

        yield "oak-export", OAK, None

        if READER:

            yield "oak-reader", OAK, READER

    def evaluate(self, scenario, situation, *, oak_binary=None):

        results = {}

        for label, binary, reader in self.arms():

            kind = "git" if label == "git" else "oak"

            repo = Path(tempfile.mkdtemp(dir=self.tmp)).resolve()

            init(kind, repo)

            situation(kind, repo)

            evidence = O.task_evidence(scenario, repo, kind=kind,

                                       binary=(oak_binary or binary) if kind == "oak" else binary,

                                       oak_reader_python=reader)

            results[label] = (evidence["workflow_integrity_passed"],

                              evidence["workflow_integrity_failure_reason"] or evidence["workflow_integrity_unmeasured_reason"])

        return results

    def assert_all(self, results, passed):

        for label, (value, reason) in results.items():

            self.assertIs(value, passed, f"{label}: {reason} (all arms: {results})")

    def test_payload_situations_agree_across_arms(self):

        scenario = "sync_push_divergence"

        def correct(kind, repo):

            write_payload(repo, scenario)

            commit(kind, repo)

        def wrong_committed(kind, repo):

            write_payload(repo, scenario)

            (repo / "notes.txt").write_text("wrong\n")

            commit(kind, repo)

            write_payload(repo, scenario)

        def exec_committed(kind, repo):

            write_payload(repo, scenario)

            os.chmod(repo / "notes.txt", 0o755)

            commit(kind, repo)

            os.chmod(repo / "notes.txt", 0o644)

        def oversized_committed(kind, repo):  # A4: task limit (1 MiB) fails first, as in Git

            write_payload(repo, scenario)

            (repo / "notes.txt").write_bytes(b"upstream change\n" + b"a" * (2 * 1024 * 1024))

            commit(kind, repo)

            write_payload(repo, scenario)

        def never_committed(kind, repo):

            write_payload(repo, scenario)

        def metadata_destroyed(kind, repo):  # A1 case 1

            write_payload(repo, scenario)

            commit(kind, repo)

            shutil.rmtree(repo / (".git" if kind == "git" else ".oak"))

        self.assert_all(self.evaluate(scenario, correct), True)

        for situation in (wrong_committed, exec_committed, oversized_committed, never_committed, metadata_destroyed):

            with self.subTest(situation=situation.__name__):

                self.assert_all(self.evaluate(scenario, situation), False)

    def test_failing_subject_oak_is_a_measured_failure_in_every_payload_workflow(self):

        # A1 case 2: a released Oak whose export or status exits nonzero must

        # fail the Oak arm, never turn it into an unmeasured skip.

        failing = {"export": self.stub("oak-export-fails", "export", 1),

                   "status": self.stub("oak-status-fails", "status", 1)}

        for scenario in sorted(O.COMMITTED_PAYLOADS):

            def correct(kind, repo, scenario=scenario):

                write_payload(repo, scenario)

                commit(kind, repo)

            for command, binary in failing.items():

                with self.subTest(scenario=scenario, command=command):

                    repo = Path(tempfile.mkdtemp(dir=self.tmp)).resolve()

                    init("oak", repo)

                    correct("oak", repo)

                    evidence = O.task_evidence(scenario, repo, kind="oak", binary=binary)

                    self.assertIs(evidence["workflow_integrity_passed"], False, evidence)

                    self.assertIsNone(evidence["workflow_integrity_unmeasured_reason"])

    def test_oracle_cli_exit_is_failure_not_sentinel_for_failing_export(self):

        scenario = "sync_pull_upstream"

        repo = Path(tempfile.mkdtemp(dir=self.tmp)).resolve()

        init("oak", repo)

        write_payload(repo, scenario)

        commit("oak", repo)

        binary = self.stub("oak-export-fails", "export", 1)

        result = subprocess.run([sys.executable, "-I", str(ROOT / "scripts/oakbench/workflow_oracles.py"), scenario,

                                 str(repo), "--oak-head", "--binary", binary], capture_output=True, text=True,

                                env={k: v for k, v in os.environ.items() if k != "OAK_BENCH_OAK_READER_PYTHON"})

        self.assertEqual(result.returncode, 1, result.stdout + result.stderr)

        self.assertIn("oak_head_export_failed", result.stdout)

    def test_reader_arm_is_exercised_when_configured(self):

        if not READER:

            self.skipTest("optional reader dependencies not configured")

        self.assertIn("oak-reader", [label for label, _, _ in self.arms()])

    def test_missing_instrument_is_still_unmeasured(self):

        scenario = "sync_pull_upstream"

        repo = Path(tempfile.mkdtemp(dir=self.tmp)).resolve()

        init("oak", repo)

        write_payload(repo, scenario)

        commit("oak", repo)

        evidence = O.task_evidence(scenario, repo, kind="oak", binary=str(self.tmp / "no-such-oak"))

        self.assertIsNone(evidence["workflow_integrity_passed"])

        evidence = O.task_evidence(scenario, repo, kind="oak", oak_reader_python=str(self.tmp / "no-python"))

        self.assertIsNone(evidence["workflow_integrity_passed"])

if __name__ == "__main__":

    unittest.main()