"""Robustness oracles: fault-fired accounting, torn worktrees, actionability, disk fill, row validation."""

from __future__ import annotations

import json

import os

import platform

import shutil

import subprocess

import sys

import tempfile

import unittest

from pathlib import Path

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

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

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

import run_robustness  # noqa: E402

from robustness import FAULT_NOT_FIRED, HARNESS_ERROR, HELD, VIOLATED, crashpoints, known_bugs  # noqa: E402

from robustness.concurrency import reported_pushed_head  # noqa: E402

from robustness.crash import run_time_point, run_trial, time_fault  # noqa: E402

from robustness.engine import actionable, classify, torn_state, torn_verdict, torn_worktree  # noqa: E402

from robustness.fixtures import build_git  # noqa: E402

from robustness.procs import Result, run_killed_after  # noqa: E402

from robustness.readers import worktree_manifest  # noqa: E402

from oakbench.rows import SKIP_RETURNCODE, validate_rows  # noqa: E402

from oakbench.subjects import which_measured_git  # noqa: E402

GIT = which_measured_git()

def result(rc, stdout="", stderr=""):

    return Result(["x"], rc, stdout, stderr, 1.0)

def f(tag):

    return {"kind": "file", "size": len(tag), "sha256": tag, "executable": False}

class FaultFiredTests(unittest.TestCase):

    def test_not_fired_is_never_held(self):

        self.assertEqual(classify({"a": True}, fault_fired=False), FAULT_NOT_FIRED)

        self.assertEqual(classify({"a": False}, fault_fired=False), FAULT_NOT_FIRED)

        self.assertEqual(classify({"a": True}, fault_fired=True), HELD)

        self.assertEqual(classify({"a": True}, fault_fired=None), HELD)

        self.assertEqual(classify({"a": True}, harness_error="x", fault_fired=False), HARNESS_ERROR)

    def test_reader_exception_is_harness_error_not_violation(self):

        self.assertEqual(classify({"a": None}, reader_errors=["reader error: boom"]), HARNESS_ERROR)

        # an independent subject failure still wins

        self.assertEqual(classify({"repo_opens": False, "a": None}, reader_errors=["x"]), VIOLATED)

    def test_process_that_exits_before_the_kill_is_not_fired(self):

        with tempfile.TemporaryDirectory() as temp:

            res = run_killed_after(["/bin/sh", "-c", "exit 0"], Path(temp), dict(os.environ), 2.0)

            self.assertEqual(res.returncode, 0)

            class Stub:

                def op_argv(self):

                    return ["/bin/sh", "-c", "exit 0"]

                def op_cwd(self):

                    return Path(temp)

                class adapter:  # noqa: N801

                    @staticmethod

                    def env():

                        return dict(os.environ)

            _, fired, point = time_fault(1.0)(Stub())

            self.assertFalse(fired)

            self.assertEqual(point, "t=1000.0ms")

    def test_process_killed_mid_run_is_fired(self):

        with tempfile.TemporaryDirectory() as temp:

            res = run_killed_after(["/bin/sleep", "5"], Path(temp), dict(os.environ), 0.05)

            self.assertEqual(res.returncode, -9)

            self.assertTrue(res.killed_by_harness)

@unittest.skipUnless(GIT and shutil.which(GIT), "real git required")

class NotFiredEndToEndTests(unittest.TestCase):

    def test_git_commit_that_finishes_before_the_kill_is_fault_not_fired(self):

        with tempfile.TemporaryDirectory() as temp:

            temp = Path(temp)

            fx = build_git(GIT, "commit", temp / "fx")

            from robustness.crash import calibrate

            calib = calibrate(fx, GIT, temp / "trials", None)

            self.assertTrue(calib.ok, calib.detail)

            row = run_trial(fx, GIT, temp / "trials", "late", calib, time_fault(60.0),

                            lambda trial, c: shutil.rmtree(trial.dir, ignore_errors=True))

            self.assertIs(row["fault_fired"], False)

            self.assertEqual(row["classification"], FAULT_NOT_FIRED)

            self.assertIn("did not fire", row["skip_reason"])

            args = run_robustness.parse_args(["--subjects", "git", "--results", str(temp / "r"), "--workdir",

                                              str(temp / "w"), "--bench-id", "20260929T000000Z", "--git-bin", GIT])

            runner = run_robustness.Runner(args)

            runner.versions = {"git": "git test"}

            emitted = runner.emit("git", "crash_time_v1", "commit", "crash_time", row)

            self.assertEqual(emitted["returncode"], SKIP_RETURNCODE)

            self.assertTrue(emitted["skipped"])

            self.assertEqual(validate_rows([emitted], "robustness"), [])

    def test_time_point_re_aims_from_observed_duration(self):

        with tempfile.TemporaryDirectory() as temp:

            temp = Path(temp)

            fx = build_git(GIT, "commit", temp / "fx")

            from robustness.crash import calibrate

            calib = calibrate(fx, GIT, temp / "trials", None)

            calib.elapsed_ms = 120000.0  # a stale calibration: the first kill lands long after the op

            row = run_time_point(fx, GIT, temp / "trials", "reaim", calib, 0.5,

                                 lambda trial, c: shutil.rmtree(trial.dir, ignore_errors=True))

            attempts = row["observations"]["time_attempts"]

            self.assertIs(attempts[0]["fired"], False)

            self.assertGreater(len(attempts), 1)

            self.assertLess(attempts[1]["delay_ms"], attempts[0]["delay_ms"])

            if row["fault_fired"] is False:

                self.assertEqual(row["classification"], FAULT_NOT_FIRED)

            else:

                self.assertIn(row["classification"], (HELD, VIOLATED))

class TornWorktreeTests(unittest.TestCase):

    PRE = {"a": f("a0"), "b": f("b0"), "c": f("c0")}

    POST = {"a": f("a1"), "b": f("b1"), "c": f("c0"), "n": f("n1")}

    def test_split(self):

        wt = {"a": f("a1"), "b": f("b0"), "c": f("c0"), "a.tmp-1": f("x")}

        split = torn_worktree(wt, self.PRE, self.POST)

        self.assertEqual(split["op_output"], ["a"])

        self.assertEqual(split["pre_side"], ["b", "n"])

        self.assertEqual(split["stray"], ["a.tmp-1"])

    def test_states(self):

        pre_split = torn_worktree(dict(self.PRE), self.PRE, self.POST)

        post_split = torn_worktree(dict(self.POST), self.PRE, self.POST)

        self.assertFalse(torn_state(pre_split, "pre"))

        self.assertFalse(torn_state(post_split, "post"))

        # every file already on the target side while the ref still names the old side

        self.assertTrue(torn_state(post_split, "pre"))

        # ref moved but the worktree lags

        self.assertTrue(torn_state(pre_split, "post"))

        mixed = torn_worktree({"a": f("a1"), "b": f("b0"), "c": f("c0")}, self.PRE, self.POST)

        self.assertTrue(torn_state(mixed, "pre"))

        self.assertTrue(torn_state(mixed, None))

        # user edits to files the op does not touch are not the op's output

        user = torn_worktree({**self.PRE, "c": f("mine")}, self.PRE, self.POST)

        self.assertFalse(torn_state(user, "pre"))

    def _verdict(self, changes, in_progress, head="old"):

        class Adapter:

            def head(self, repo):

                return head

            def status_state(self, repo):

                return changes, in_progress, result(0 if changes is not None else 4)

        class T:

            adapter = Adapter()

        from unittest import mock

        with mock.patch("robustness.readers.worktree_manifest",

                        return_value={"a": f("a1"), "b": f("b0"), "c": f("c0")}):

            return torn_verdict(T(), Path("."), self.PRE, self.POST, "old")

    def test_partial_output_presented_as_user_changes_fails(self):

        ok, detail = self._verdict(True, False)

        self.assertIs(ok, False)

        self.assertEqual(detail["torn"], "PARTIAL_OP_OUTPUT_AS_USER_CHANGES")

        self.assertEqual(detail["ref_side"], "pre")

    def test_flagged_in_progress_or_refused_passes(self):

        self.assertIs(self._verdict(True, True)[0], True)

        self.assertIs(self._verdict(None, False)[0], True)

@unittest.skipUnless(GIT and shutil.which(GIT), "real git required")

class TornWorktreeRealGitTests(unittest.TestCase):

    def test_half_switched_git_worktree_is_torn_but_merge_head_is_not(self):

        with tempfile.TemporaryDirectory() as temp:

            temp = Path(temp)

            fx = build_git(GIT, "switch", temp / "fx")

            client = fx.pristine / "client"

            pre = worktree_manifest(client)

            from robustness.engine import Trial

            trial = Trial(fx, GIT, temp / "trials", "torn")

            trial.adapter.must([GIT, "switch", "-q", fx.meta["target"]], trial.client)

            post = worktree_manifest(trial.client)

            trial.adapter.must([GIT, "switch", "-q", fx.branch], trial.client)

            # simulate a crash mid-checkout: some files already hold the target side, HEAD is old

            for name in sorted(p for p in post if pre.get(p) != post.get(p))[:5]:

                src = subprocess.run([GIT, "show", f"{fx.meta['target']}:{name}"], cwd=trial.client,

                                     capture_output=True, check=True).stdout

                (trial.client / name).parent.mkdir(parents=True, exist_ok=True)

                (trial.client / name).write_bytes(src)

            ok, detail = torn_verdict(trial, trial.client, pre, post, fx.meta["pre_head"])

            self.assertIs(ok, False, detail)

            (trial.client / ".git" / "MERGE_HEAD").write_text(fx.meta["pre_head"] + "\n")

            ok, detail = torn_verdict(trial, trial.client, pre, post, fx.meta["pre_head"])

            self.assertIs(ok, True, detail)

            trial.cleanup()

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

class GitStaleLockTests(unittest.TestCase):

    """M1: git locks that make the next commit refuse count as 'in progress', like index.lock.

    Synthetic, interposer-free: a real repo with a lock file planted, and real

    `git commit` confirming which locks actually block."""

    def repo(self, temp: Path) -> Path:

        repo = temp / "r"

        env = {**os.environ, "GIT_CONFIG_GLOBAL": "/dev/null", "GIT_CONFIG_NOSYSTEM": "1"}

        for argv in (["init", "-q", "-b", "main", str(repo)],):

            subprocess.run([GIT, *argv], check=True, env=env)

        for argv in (["config", "user.email", "a@b"], ["config", "user.name", "a"]):

            subprocess.run([GIT, *argv], cwd=repo, check=True, env=env)

        (repo / "f").write_text("a\n")

        subprocess.run([GIT, "add", "f"], cwd=repo, check=True, env=env)

        subprocess.run([GIT, "commit", "-qm", "a"], cwd=repo, check=True, env=env)

        subprocess.run([GIT, "branch", "other"], cwd=repo, check=True, env=env)

        subprocess.run([GIT, "tag", "t1"], cwd=repo, check=True, env=env)

        self.env = env

        return repo

    def test_blocking_locks_match_what_git_commit_refuses(self):

        from robustness.adapters import GitAdapter

        blocking = ("index.lock", "HEAD.lock", "refs/heads/main.lock")

        harmless = ("config.lock", "refs/heads/other.lock", "refs/tags/t1.lock", "ORIG_HEAD.lock")

        with tempfile.TemporaryDirectory() as temp:

            repo = self.repo(Path(temp))

            adapter = GitAdapter(GIT, Path(temp) / "home")

            for n, lock in enumerate(blocking + harmless):

                path = repo / ".git" / lock

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

                path.write_text("")

                (repo / "f").write_text(f"change {n}\n")  # a pending change status lists

                changes, in_progress, _ = adapter.status_state(repo)

                commit = subprocess.run([GIT, "commit", "-qam", f"c{n}"], cwd=repo, capture_output=True,

                                        text=True, env=self.env)

                path.unlink()

                if lock in blocking:

                    self.assertTrue(in_progress, lock)

                    self.assertEqual(adapter.blocking_locks(repo), [], lock)

                    self.assertNotEqual(commit.returncode, 0, lock)

                    self.assertIn("remove the file manually", commit.stderr, lock)

                    self.assertTrue(actionable(result(commit.returncode, stderr=commit.stderr)), lock)

                else:

                    self.assertTrue(changes, lock)

                    self.assertFalse(in_progress, lock)

                    self.assertEqual(commit.returncode, 0, (lock, commit.stderr))

    def test_half_switched_worktree_with_head_lock_is_flagged_not_torn(self):

        # `git crash_syscall switch before rename#2`: worktree/index on the target, HEAD

        # still names the old branch, HEAD.lock left behind.

        with tempfile.TemporaryDirectory() as temp:

            temp = Path(temp)

            fx = build_git(GIT, "switch", temp / "fx")

            client = fx.pristine / "client"

            pre = worktree_manifest(client)

            from robustness.engine import Trial

            trial = Trial(fx, GIT, temp / "trials", "torn")

            trial.adapter.must([GIT, "switch", "-q", fx.meta["target"]], trial.client)

            post = worktree_manifest(trial.client)

            (trial.client / ".git" / "HEAD").write_text(f"ref: refs/heads/{fx.branch}\n")

            ok, detail = torn_verdict(trial, trial.client, pre, post, fx.meta["pre_head"])

            self.assertIs(ok, False, detail)  # without the lock: presented as user changes

            (trial.client / ".git" / "HEAD.lock").write_text(f"ref: refs/heads/{fx.meta['target']}\n")

            ok, detail = torn_verdict(trial, trial.client, pre, post, fx.meta["pre_head"])

            self.assertIs(ok, True, detail)

            self.assertEqual(detail["torn"], "flagged in progress")

            (trial.client / ".git" / "HEAD.lock").unlink()

            (trial.client / ".git" / "config.lock").write_text("")  # does not block a commit

            ok, detail = torn_verdict(trial, trial.client, pre, post, fx.meta["pre_head"])

            self.assertIs(ok, False, detail)

            trial.cleanup()

class ActionableTests(unittest.TestCase):

    def test_corruption_internal_and_panic_are_not_actionable(self):

        for text in ("error: failed to open oak database: file is not a database",

                     "error: internal error in oak core",

                     "error: push rejected: corrupted object",

                     "fatal: not a git repository (or any of the parent directories): .git",

                     "error: Database error: database disk image is malformed",

                     "thread 'main' panicked at src/lib.rs:9:5",

                     "error: internal error; please try again"):

            self.assertFalse(actionable(result(1, stderr=text)), text)

        internal = {"error": {"code": "internal", "message": "store malformed"}}

        self.assertFalse(actionable(result(1, stdout=json.dumps(internal))))

        generic = {"error": {"code": "error", "message": "something failed", "recommended_next_commands": []}}

        self.assertFalse(actionable(result(1, stdout=json.dumps(generic))))

    def test_recovery_commands_and_typed_codes_are_actionable(self):

        for text in ("error: working tree has uncommitted changes — run `oak commit` to keep them",

                     "Remove this directory and re-run `oak clone`.",

                     "The remote branch has commits this clone doesn't have.\nRun 'oak pull' to bring them in",

                     "remove the file manually to continue.",

                     "hint: Updates were rejected ... ! [rejected] main -> main (fetch first)",

                     "error: repository locked by another process; try again"):

            self.assertTrue(actionable(result(4, stderr=text)), text)

        rec = {"error": {"code": "error", "message": "x", "recommended_next_commands": ["oak status --json"]}}

        self.assertTrue(actionable(result(6, stdout=json.dumps(rec))))

        self.assertTrue(actionable(result(5, stdout=json.dumps({"error": {"code": "merge_in_progress"}}))))

        self.assertFalse(actionable(result(0, stderr="run `oak pull`")))

    CONFLICT = ("Auto-merging f.txt\nCONFLICT (content): Merge conflict in f.txt\n"

                "Automatic merge failed; fix conflicts and then commit the result.\n")

    FF = "fatal: Not possible to fast-forward, aborting.\n"

    PULL_FF = "From /tmp/srv/remote.git\n * branch            main       -> FETCH_HEAD\n" + FF

    def test_corruption_beats_any_recovery_mention(self):

        from robustness.envfaults import env_actionable

        cases = [

            (["git", "pull"], "fatal: loose object 0123abcd (stored in .git/objects/01/23abcd) is corrupt\n"

                              "hint: run git merge --abort"),

            (["git", "merge"], "fatal: index file corrupt\nhint: git merge --abort"),

            (["git", "merge"], "Auto-merging f.txt\nCONFLICT (content): Merge conflict in f.txt\n"

                               "Automatic merge failed; fix conflicts and then commit the result.\n"

                               "fatal: bad object HEAD\nrun git commit"),

            (["git", "pull"], "fatal: unable to read tree 0123abcd\nretry with git pull"),

            (["git", "pull"], "error: object file .git/objects/12/34 is empty\nrun git fsck"),

            (["git", "pull"], "fatal: early EOF\nfatal: fetch-pack: invalid index-pack output\ntry again"),

            (["oak", "pull"], "error: Database error: database disk image is malformed (oak pull)"),

            (["oak", "push"], "error: Server error: HTTP 500 Internal Server Error; run oak push again"),

            (["oak", "merge"], "error: internal error in merge engine; use --abort"),

            (["oak", "pull"], "error: Database error: unable to open database file\nrun `oak pull` again"),

            (["oak", "pull"], "error: Database error: unable to open database file"),

        ]

        for argv, text in cases:

            for rc in (1, 2, 128):

                res = Result(argv, rc, "", text, 1.0)

                self.assertFalse(actionable(res), text)

                self.assertFalse(env_actionable(res), text)

            self.assertFalse(actionable(Result(argv, 1, text, "", 1.0)), text)

        # the recovery rules still work on healthy output

        self.assertTrue(actionable(Result(["git", "merge"], 1, "", "error: Merging is not possible because you "

                                          "have unmerged files.\nhint: Fix them up and use git commit", 1.0)))

        self.assertTrue(actionable(Result(["oak", "pull"], 1, "", "error: working tree has uncommitted changes "

                                          "— run `oak commit` to keep them", 1.0)))

    def test_git_conflict_and_fast_forward_outcomes_are_actionable(self):

        def git(rc, **kw):

            return Result(["/usr/bin/git", "merge"], rc, kw.get("stdout", ""), kw.get("stderr", ""), 1.0)

        ff_hints = ("hint: Diverging branches can't be fast-forwarded, you need to either:\nhint:\n"

                    "hint: \tgit merge --no-ff\nhint:\nhint: or:\nhint:\nhint: \tgit rebase\nhint:\n"

                    "hint: Disable this message with \"git config set advice.diverging false\"\n" + self.FF)

        self.assertTrue(actionable(git(1, stdout=self.CONFLICT)))

        for text in (self.FF, self.PULL_FF, ff_hints):

            self.assertTrue(actionable(git(128, stderr=text)), text)

    def test_git_outcome_hints_are_git_only_and_allowlisted(self):

        def git(text, rc=1):

            return Result(["git", "pull"], rc, "", text, 1.0)

        # the same sentences from another subject are not git's hints

        for text in (self.CONFLICT, self.FF):

            self.assertFalse(actionable(Result(["oak", "pull"], 1, "", text, 1.0)), text)

            self.assertFalse(actionable(result(1, stderr=text)), text)

        broken_lines = ("fatal: not a git repository (or any of the parent directories): .git",

                        "error: object file .git/objects/ab/cd is empty",

                        "fatal: loose object abcd is corrupt",

                        "error: RPC failed; HTTP 500 curl 22 The requested URL returned error: 500",

                        "fatal: internal error in merge-ort",

                        "thread 'main' panicked at src/merge.rs:1:1",

                        "error: Database error: database disk image is malformed",

                        "fatal: bad object HEAD",

                        "error: unable to read sha1 file of f.txt (abcdef)",

                        "fatal: invalid object 1234567 for 'f.txt'",

                        "fatal: early EOF",

                        "Segmentation fault: 11",

                        "fatal: unable to read tree 0123456789abcdef",

                        "error: object file .git/objects/12/34 is empty",

                        "fatal: unable to write new index file",

                        "error: Database error: unable to open database file")

        for broken in broken_lines:

            for hint in (self.CONFLICT, self.FF, self.PULL_FF):

                for text in (broken + "\n" + hint, hint + broken + "\n"):

                    self.assertFalse(actionable(git(text)), text)

        # the phrase must be git's own sentence, not any mention of a failed merge

        self.assertFalse(actionable(git("error: automatic merge failed")))

        self.assertFalse(actionable(git("error: fast-forward not possible")))

@unittest.skipUnless(platform.system() == "Darwin" and shutil.which("hdiutil"), "macOS RAM disk required")

class DiskFillTests(unittest.TestCase):

    def test_fill_leaves_less_than_one_block_and_records_measured_free(self):

        from robustness.envfaults import RamDisk, fill, free_bytes

        with tempfile.TemporaryDirectory() as temp:

            mount = Path(temp) / "mnt"

            with RamDisk(mount, mib=8):

                _, free = fill(mount, 0)

                self.assertLess(free, 4096)

                self.assertEqual(free, free_bytes(mount))

                with self.assertRaises(OSError):

                    with (mount / "more.bin").open("wb") as fh:

                        fh.write(b"x" * 8192)

                        fh.flush()

                        os.fsync(fh.fileno())

                (mount / "more.bin").unlink(missing_ok=True)

                (mount / "p4-filler.bin").unlink()

                _, free = fill(mount, 64)

                self.assertGreaterEqual(free, 60 * 1024)

                self.assertLess(free, 72 * 1024)

class EnvFaultFiredTests(unittest.TestCase):

    def test_disk_full_judges_every_failure_and_skips_only_a_roomy_success(self):

        from robustness.envfaults import disk_full_fired

        # only a successful op that had room is not fired

        fired, evidence = disk_full_fired(1 << 20, 4096, result(0))

        self.assertIs(fired, False)

        self.assertIn("succeeded", evidence)

        for text in ("error: IO error: No space left on device (os error 28)",

                     "fatal: unable to write new index file: ENOSPC",

                     "Error: database or disk is full", "sqlite: SQLITE_FULL"):

            fired, evidence = disk_full_fired(1 << 20, 4096, result(1, stderr=text))

            self.assertTrue(fired, text)

            self.assertIn("ENOSPC", evidence)

        self.assertTrue(disk_full_fired(0, 4096, result(0))[0])  # measured short: fired even if it succeeded

        # a failure with room and no ENOSPC text is judged, never skipped

        fb548 = "error: Database error: unable to open database file"

        panic = "thread 'main' panicked at src/store.rs:9:5:\nnote: run with `RUST_BACKTRACE=1`"

        hang = Result(["oak", "pull"], None, "", "", 90000.0, killed_by_harness=True, timed_out=True)

        for res in (result(1, stderr=fb548), result(101, stderr=panic), result(1, stderr="error: something else"),

                    hang):

            for free in (0, 1 << 20):

                fired, evidence = disk_full_fired(free, 4096, res)

                self.assertIs(fired, True, (res, free))

    def test_opaque_failures_on_a_full_disk_are_violations(self):

        from robustness.envfaults import disk_full_fired, env_actionable

        fb548 = "error: Database error: unable to open database file"

        panic = "thread 'main' panicked at src/store.rs:9:5:\nnote: run with `RUST_BACKTRACE=1`"

        for res in (result(1, stderr=fb548), result(101, stderr=panic)):

            fired, _ = disk_full_fired(1 << 20, 4096, res)

            inv = {"op_no_hang": True, "op_failure_actionable": env_actionable(res)}

            self.assertEqual(classify(inv, fault_fired=fired), VIOLATED, res.stderr)

        hang = Result(["oak", "pull"], None, "", "", 90000.0, killed_by_harness=True, timed_out=True)

        fired, _ = disk_full_fired(1 << 20, 4096, hang)

        self.assertEqual(classify({"op_no_hang": False}, fault_fired=fired), VIOLATED)

        # the cause named plainly stays actionable

        named = result(1, stderr="error: IO error: No space left on device (os error 28)")

        self.assertTrue(env_actionable(named))

    def test_failed_calibration_is_a_harness_error_not_zero_need(self):

        from robustness.envfaults import disk_full_trial

        row = disk_full_trial(None, "oak", Path("/nonexistent"), "x", None, 0, None)

        self.assertEqual(row["classification"], HARNESS_ERROR)

        self.assertIn("unmeasured", row["harness_error"])

    def test_readonly_judges_failures_and_skips_only_unexercised_success(self):

        from robustness.envfaults import readonly_fired

        hang = Result(["oak", "pull"], None, "", "", 90000.0, killed_by_harness=True, timed_out=True)

        for res in (result(1, stderr="Permission denied"), result(1, stderr="error: unable to open database file"),

                    result(101, stderr="thread 'main' panicked at x"), hang):

            for writable, exercised in ((True, False), (False, False), (False, None), (False, True)):

                self.assertIs(readonly_fired(writable, exercised, res)[0], True, (res, writable, exercised))

        self.assertTrue(readonly_fired(False, False, result(0, stderr="warning: permission denied on d3"))[0])

        self.assertTrue(readonly_fired(False, True, result(0))[0])

        fired, evidence = readonly_fired(False, False, result(0))

        self.assertIs(fired, False)

        self.assertTrue(evidence.startswith("not exercised"))

        self.assertIs(readonly_fired(True, True, result(0))[0], False)  # root: fault never applied

        self.assertIsNone(readonly_fired(False, None, result(0))[0])

    def test_write_snapshot_sees_writes_under_the_locked_path(self):

        from robustness.envfaults import write_snapshot

        with tempfile.TemporaryDirectory() as temp:

            d = Path(temp) / "d3"

            (d / "sub").mkdir(parents=True)

            (d / "a").write_text("a")

            before = write_snapshot(d, "subdir")

            (d / "sub" / "deep").write_text("x")  # not blocked by a read-only d3 itself

            (d / "a").write_text("changed in place")  # in-place write: allowed

            self.assertEqual(write_snapshot(d, "subdir"), before)

            (d / "b").write_text("new entry")

            self.assertNotEqual(write_snapshot(d, "subdir"), before)

            store = write_snapshot(d, "store")

            os.utime(d / "sub" / "deep", ns=(1, 1))

            self.assertNotEqual(write_snapshot(d, "store"), store)

            self.assertIsNone(write_snapshot(d / "missing", "store"))

    @unittest.skipUnless(GIT and os.geteuid() != 0, "git and a non-root user required")

    def test_git_readonly_rows_record_fired_or_not_exercised(self):

        from robustness.crash import calibrate

        from robustness.envfaults import readonly_exercised, readonly_trial

        with tempfile.TemporaryDirectory() as temp:

            temp = Path(temp)

            fx = build_git(GIT, "commit", temp / "fx")

            calib = calibrate(fx, GIT, temp / "trials", None)

            exercised, detail = readonly_exercised(fx, GIT, temp / "trials", "probe", "store")

            self.assertTrue(exercised, detail)

            row = readonly_trial(fx, GIT, temp / "trials", "ro", calib, "store", exercised, detail)

            self.assertIs(row["fault_fired"], True, row)

            # a lock-down no write reaches is recorded as not exercised, never held

            row = readonly_trial(fx, GIT, temp / "trials", "ro2", calib, "store", False, {})

            if row["fault_fired"] is False:

                self.assertEqual(row["classification"], FAULT_NOT_FIRED)

                self.assertTrue(row["skip_reason"].startswith("not exercised"))

            else:  # git failed on the read-only store: judged, never skipped

                self.assertNotEqual(row["command"], [])

                self.assertIn(row["classification"], (HELD, VIOLATED))

                self.assertRegex(row["observations"]["fault_fired_evidence"], "judged|permission")

class RowValidationTests(unittest.TestCase):

    def _row(self, **kw):

        row = {"bench_id": "20260929T000000Z", "scenario": "crash_time_v1", "operation": "commit", "run": 0,

               "subject": "git", "subject_kind": "git", "elapsed_ms": 1.0, "returncode": 0, "classification": "held",

               "fault_family": "crash_time", "fault_point": "t=1ms", "fault_fired": True, "violations": []}

        row.update(kw)

        return row

    def test_rejects_unknown_and_inconsistent_classifications(self):

        self.assertEqual(validate_rows([self._row()], "robustness"), [])

        self.assertTrue(validate_rows([self._row(classification="bogus")], "robustness"))

        self.assertTrue(validate_rows([self._row(returncode=1)], "robustness"))

        self.assertTrue(validate_rows([self._row(classification="violated", returncode=0)], "robustness"))

        self.assertTrue(validate_rows([self._row(fault_fired=False)], "robustness"))

        self.assertTrue(validate_rows([self._row(violations=["x"])], "robustness"))

        not_fired = self._row(classification="fault_not_fired", returncode=77, fault_fired=False, skipped=True,

                              skip_reason="op exited before the kill")

        self.assertEqual(validate_rows([not_fired], "robustness"), [])

        self.assertTrue(validate_rows([{**not_fired, "returncode": 0, "skipped": False}], "robustness"))

class MiscTests(unittest.TestCase):

    def test_known_bug_tags(self):

        self.assertEqual(known_bugs("oak", "crash_time_v1", "pull_ff", ["worktree_not_torn"]), ["fb-535"])

        self.assertEqual(known_bugs("oak", "crash_syscall_v1", "switch", ["worktree_not_torn", "x"]), ["fb-535"])

        self.assertEqual(known_bugs("oak", "conc_same_branch_v1", "concurrent.push",

                                    ["no_lost_acknowledged_write", "linearizable"]), ["fb-534"])

        self.assertEqual(known_bugs("git", "crash_time_v1", "switch", ["worktree_not_torn"]), [])

        self.assertEqual(known_bugs("oak", "crash_time_v1", "commit", ["repo_opens"]), [])

    def test_no_boundary_reasons_are_subject_specific(self):

        self.assertNotIn("git", crashpoints.NO_BOUNDARY_REASON["oak"])

        self.assertIn("git", crashpoints.NO_BOUNDARY_REASON["git"])

        inst = crashpoints.CrashInstrument(Path("x"), None, Path("g"), None, {}, {"oak": "codesign failed"})

        self.assertEqual(crashpoints.unavailable_reason(inst, "", "oak"), "codesign failed")

        self.assertEqual(crashpoints.unavailable_reason(inst, "", "git"), "")

        self.assertEqual(crashpoints.unavailable_reason(None, "no clang", "git"), "no clang")

    def test_reported_pushed_head(self):

        self.assertEqual(reported_pushed_head(result(0, stdout=json.dumps({"pushed": True, "pushed_head": "abc"}))),

                         "abc")

        self.assertIsNone(reported_pushed_head(result(0, stdout="plain text")))

if __name__ == "__main__":

    unittest.main()