"""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()

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`")))

@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 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()