"""P1 honest measurement: Git shim resolution, rusage CPU/IO, cache labels,

per-row provenance, and the Scalar-tuned Git subject — from real producer rows."""

from __future__ import annotations

import argparse

import os

import platform

import shutil

import subprocess

import sys

import tempfile

import unittest

from pathlib import Path

from unittest import mock

sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts"))

import bench  # noqa: E402

from oakbench import subjects as oakbench_subjects  # noqa: E402

from oakbench.execution import run_timed  # noqa: E402

from oakbench.rows import validate_rows  # noqa: E402

GIT = shutil.which("git")

class GitShimTest(unittest.TestCase):

    def test_shim_is_detected_only_at_the_darwin_stub_path(self):

        self.assertTrue(oakbench_subjects.is_xcrun_git_shim("/usr/bin/git", "Darwin"))

        self.assertFalse(oakbench_subjects.is_xcrun_git_shim("/usr/bin/git", "Linux"))

        self.assertFalse(oakbench_subjects.is_xcrun_git_shim("/opt/homebrew/bin/git", "Darwin"))

    def test_shim_is_resolved_and_recorded(self):

        real = Path("/real/xcode/git")

        path, requested, measured = oakbench_subjects.resolve_git_subject_bin(

            Path("/usr/bin/git"), allow_shim=False, system="Darwin", resolver=lambda: real)

        self.assertEqual((path, requested, measured), (real, Path("/usr/bin/git"), False))

    def test_unresolvable_shim_is_refused_unless_allowed(self):

        with self.assertRaisesRegex(SystemExit, "xcrun git shim"):

            oakbench_subjects.resolve_git_subject_bin(Path("/usr/bin/git"), allow_shim=False,

                                                      system="Darwin", resolver=lambda: None)

        path, requested, measured = oakbench_subjects.resolve_git_subject_bin(

            Path("/usr/bin/git"), allow_shim=True, system="Darwin", resolver=lambda: None)

        self.assertEqual((path, measured), (Path("/usr/bin/git"), True))

    @unittest.skipUnless(platform.system() == "Darwin" and Path("/usr/bin/git").exists(), "macOS xcrun shim required")

    def test_real_host_resolution_differs_from_shim_bytes(self):

        real = oakbench_subjects.resolve_real_git()

        if real is None:

            self.skipTest("no developer tools selected")

        self.assertNotEqual(str(real), "/usr/bin/git")

        self.assertNotEqual(oakbench_subjects.binary_sha256(real), oakbench_subjects.binary_sha256(Path("/usr/bin/git")))

        version = subprocess.run([str(real), "--version"], capture_output=True, text=True, check=True).stdout

        self.assertIn("git version", version)

    @unittest.skipUnless(platform.system() == "Darwin" and Path("/usr/bin/git").exists(), "macOS xcrun shim required")

    def test_load_subjects_never_times_the_shim_by_default(self):

        with tempfile.TemporaryDirectory() as tmp:

            config = Path(tmp) / "subjects.toml"

            config.write_text('[subjects.git]\nenabled = true\nkind = "git"\nlabel = "Git"\nbin = "/usr/bin/git"\n')

            args = argparse.Namespace(config=config, subjects="git", git_bin=None, oak_installed_bin=None,

                                      oak_local_bin=None, allow_git_shim=False)

            with mock.patch.dict(os.environ, {oakbench_subjects.ALLOW_GIT_SHIM_ENV: ""}):

                if oakbench_subjects.resolve_real_git() is None:

                    self.skipTest("no developer tools selected")

                subject = oakbench_subjects.load_subjects(args)[0]

            self.assertNotEqual(str(subject.bin), "/usr/bin/git")

            self.assertEqual(subject.requested_bin, Path("/usr/bin/git"))

            details = oakbench_subjects.subject_details([subject])["git"]

            self.assertTrue(details["git_shim_resolved"])

            self.assertFalse(details["git_shim_measured"])

            self.assertEqual(details["git_config_profile"], "stock")

class RusageTest(unittest.TestCase):

    def test_cpu_and_io_are_measured_for_a_real_child(self):

        capture = run_timed([sys.executable, "-c", "sum(i*i for i in range(300000))"], Path.cwd(), dict(os.environ), 100)

        self.assertIsNotNone(capture.cpu_user_ms)

        self.assertGreater(capture.cpu_user_ms + capture.cpu_sys_ms, 0.0)

        self.assertIsInstance(capture.io_read_blocks, int)

        self.assertIsInstance(capture.io_write_blocks, int)

@unittest.skipUnless(GIT, "Git required")

class CoreRowContractTest(unittest.TestCase):

    """Real bench.py producer rows carry the P1 fields and still validate."""

    def run_scenario(self, subject, metadata=None, skip_determinism_probe=False, name="p1_contract"):

        root = Path(tempfile.mkdtemp(prefix="oakbench-p1-"))

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

        scenario = bench.Scenario(name, 3, 64, 2, False, 1)

        fixture = bench.ensure_fixture(root / "fixtures", scenario)

        meta = {"bench_id": "p1-test", "profile": "micro", "subject_details": oakbench_subjects.subject_details([subject]), "cache_mode_requested": "warm",

                "timing_contract_version": bench.TIMING_CONTRACT_VERSION, **bench.fixture_fingerprint(fixture),

                **(metadata or {})}

        return bench.run_subject_scenario(subject, scenario, fixture, 0, root / "runs", meta, False, 10000,

                                          "core-equivalent", skip_determinism_probe=skip_determinism_probe,

                                          skip_remote=True)

    def test_timed_rows_carry_cache_rusage_and_provenance(self):

        subject = bench.Subject("git", "git", "Git", Path(GIT))

        rows = self.run_scenario(subject)

        self.assertEqual(validate_rows(rows, "core"), [])

        timed = [row for row in rows if row["operation"] in bench.MAIN_TIMED_OPS]

        self.assertTrue(timed)

        for row in timed:

            self.assertEqual(row["page_cache_state"], "warm", row["operation"])

            self.assertEqual(row["cache_state"], "warm")

            self.assertIsNotNone(row["cpu_user_ms"])

            self.assertIsInstance(row["io_read_blocks"], int)

            self.assertEqual(row["rusage_source"], "os_wait4_child_rusage")

            self.assertEqual(row["subject_bin"], GIT)

            self.assertRegex(row["binary_sha256"], "^[0-9a-f]{64}$")

            self.assertIn("git version", row["subject_version"])

            self.assertRegex(row["fixture_sha256"], "^[0-9a-f]{64}$")

            self.assertEqual(row["fixture_stats"]["files"], 3)

            self.assertEqual(row["git_config_profile"], "stock")

        snapshot = next(row for row in rows if row["operation"] == "snapshot.dirty.commit")

        self.assertEqual(snapshot["harness_reads_before_op"], ["repository_state_fingerprint", "payload_sample_hashes"])

        replay = next(row for row in rows if row["operation"] == "diff.dirty")

        self.assertEqual(replay["cache_state"], "hash_warmed_isolated_dirty_replay")  # caller label preserved

        self.assertEqual(replay["page_cache_state"], "warm")

    def test_cold_mode_purges_after_harness_reads_or_labels_unknown(self):

        subject = bench.Subject("git", "git", "Git", Path(GIT))

        events = []

        def fake_purge():

            events.append("purge")

            return {"attempted": True, "ok": True, "method": "test_purge", "error": None}

        original_fingerprint = bench.oakbench_integrity.repository_state_fingerprint

        def fingerprint(repo):

            events.append("fingerprint")

            return original_fingerprint(repo)

        original_run_timed = bench.run_timed

        def timed(*args, **kwargs):

            events.append("timed")

            return original_run_timed(*args, **kwargs)

        with mock.patch.object(bench.oakbench_cachectl, "purge_fs_caches", side_effect=fake_purge), \

             mock.patch.object(bench.oakbench_integrity, "repository_state_fingerprint", side_effect=fingerprint), \

             mock.patch.object(bench, "run_timed", side_effect=timed):

            # Diagnostic probes are untimed (null elapsed) and exempt; skip them here.

            rows = self.run_scenario(subject, {"cache_mode_requested": "cold"}, skip_determinism_probe=True)

        commit = next(row for row in rows if row["operation"] == "snapshot.dirty.commit")

        self.assertEqual((commit["page_cache_state"], commit["cache_state"]), ("purged", "cold"))

        # Every timed command is IMMEDIATELY preceded by a purge, so no harness

        # read (fingerprint/payload hash) sits between the purge and the clock.

        self.assertIn("fingerprint", events)

        for index, event in enumerate(events):

            if event == "timed":

                self.assertEqual(events[index - 1], "purge", events[max(0, index - 3):index + 1])

        with mock.patch.object(bench.oakbench_cachectl, "purge_fs_caches",

                               return_value={"attempted": True, "ok": False, "method": "x", "error": "EPERM"}):

            rows = self.run_scenario(subject, {"cache_mode_requested": "cold"})

        status = next(row for row in rows if row["operation"] == "status.clean")

        self.assertEqual((status["page_cache_state"], status["cache_state"]), ("unknown", "unknown"))

        self.assertIn("purge_failed", status["cache_state_reason"])

    @unittest.skipUnless(GIT and subprocess.run([GIT, "fsmonitor--daemon", "--help"], capture_output=True).returncode in (0, 129),

                         "Git fsmonitor required")

    def test_scalar_tuned_subject_is_labelled_and_configured(self):

        stock = bench.Subject("git", "git", "Git", Path(GIT))

        tuned = bench.expand_git_modes([stock], "scalar")[1]

        self.assertEqual((tuned.name, tuned.git_mode), ("git_scalar", "scalar"))

        # Scalar targets large trees; on tiny fixtures it is not a stronger baseline.

        applies, _ = bench.git_mode_applies_to_scenario("scalar", bench.Scenario("tiny_text", 1, 1, 1, False, 1))

        self.assertFalse(applies)

        applies, _ = bench.git_mode_applies_to_scenario("scalar", bench.Scenario("many_small_files", 1, 1, 1, False, 1))

        self.assertTrue(applies)

        rows = self.run_scenario(tuned, name="many_small_files")

        # Daemon start and cache warm-up are recorded setup, before any timed op.

        ops = [row["operation"] for row in rows]

        self.assertLess(ops.index("mode.setup.scalar_warm_status"), ops.index("snapshot.initial"))

        setup = [row["operation"] for row in rows if row["operation"].startswith("mode.setup.scalar_")]

        self.assertEqual(len(setup), len(bench.WIRED_GIT_MODES["scalar"]))

        self.assertTrue(all(row["returncode"] == 0 for row in rows if row["operation"].startswith("mode.setup.")))

        status = next(row for row in rows if row["operation"] == "status.clean")

        self.assertEqual(status["git_config_profile"], "scalar_recommended")

        self.assertEqual(validate_rows(rows, "core"), [])

    def test_subject_exit_77_is_a_recorded_failure_not_a_harness_skip(self):

        # QA P1 A7: a subject returning the harness sentinel used to crash the

        # row contract; it is a subject failure with its real code preserved.

        root = Path(tempfile.mkdtemp(prefix="oakbench-p1-77-"))

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

        stub = root / "git77"

        stub.write_text(f'#!/bin/sh\n[ "$1" = status ] && exit 77\nexec {GIT} "$@"\n')

        stub.chmod(0o755)

        rows = self.run_scenario(bench.Subject("git", "git", "Git", stub))

        status = next(row for row in rows if row["operation"] == "status.clean")

        self.assertEqual((status["returncode"], status["process_returncode"]), (1, 77))

        self.assertFalse(status.get("skipped", False))

        self.assertTrue(all(row.get("skipped") for row in rows[rows.index(status) + 1:]))

        self.assertEqual(validate_rows(rows, "core"), [])

    def test_summary_compares_oak_with_the_strongest_git(self):

        subjects = [bench.Subject("git", "git", "Git", Path("/g")),

                    bench.Subject("git_scalar", "git", "Git scalar", Path("/g"), git_mode="scalar"),

                    bench.Subject("oak_installed", "oak", "Oak", Path("/o"))]

        by_key = {("git", "s/status.dirty"): 5.0, ("git_scalar", "s/status.dirty"): 11.0,

                  ("oak_installed", "s/status.dirty"): 7.0,

                  ("git", "s/snapshot.dirty"): 20.0, ("git_scalar", "s/snapshot.dirty"): 12.0,

                  ("oak_installed", "s/snapshot.dirty"): 14.0}

        best = bench.strongest_git_baseline(by_key, subjects)

        self.assertEqual(best["s/status.dirty"], ("git", 5.0))       # tuned-but-slower never flatters Oak

        self.assertEqual(best["s/snapshot.dirty"], ("git_scalar", 12.0))

        self.assertIn("strongest Git", "\n".join(bench.strongest_git_lines(by_key, subjects)))

if __name__ == "__main__":

    unittest.main()