"""Robustness runner: row contract, classification, readers, and a real Git case."""

from __future__ import annotations

import json

import shutil

import sqlite3

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, UNSUPPORTED, VIOLATED, known_bugs  # noqa: E402

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

from robustness.hosted import configured_url  # noqa: E402

from robustness.procs import Result  # noqa: E402

from robustness.readers import OakGraph  # noqa: E402

from oakbench.lane_contract import LANE_TABLE  # noqa: E402

from oakbench.rows import 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)

class ClassificationTests(unittest.TestCase):

    def test_classify(self):

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

        self.assertEqual(classify({"a": True, "b": False}), VIOLATED)

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

    def test_actionable_errors(self):

        self.assertTrue(actionable(result(4, stderr="error: run `oak pull --abort` to recover")))

        self.assertTrue(actionable(result(6, stdout=json.dumps({"error": {"code": "remote_diverged", "message": "x"}}))))

        self.assertTrue(actionable(result(5, stderr="error: Conflict detected: remote has diverged. Pull changes first.")))

        self.assertFalse(actionable(result(4, stderr="error: Working directory has uncommitted changes")))

        self.assertFalse(actionable(result(101, stderr="thread 'main' panicked at src/x.rs:1")))

        self.assertFalse(actionable(result(-9)))

        self.assertFalse(actionable(result(128, stderr="fatal: Unable to write index.")))

    def test_hosted_refuses_non_loopback(self):

        import os

        from unittest import mock

        with mock.patch.dict(os.environ, {"OAK_BENCH_ROBUSTNESS_OAKSPACE_URL": "https://oak.space"}):

            url, reason = configured_url()

        self.assertIsNone(url)

        self.assertIn("refusing non-loopback", reason)

class OakGraphReaderTests(unittest.TestCase):

    def test_reachability_from_refs_on_a_crashed_wal_store(self):

        with tempfile.TemporaryDirectory() as temp:

            db = Path(temp) / "oak.db"

            conn = sqlite3.connect(db)

            conn.execute("PRAGMA journal_mode=WAL")

            conn.execute("CREATE TABLE metadata(key TEXT PRIMARY KEY, value TEXT)")

            conn.execute("CREATE TABLE branch_heads(branch_name TEXT PRIMARY KEY, head_hash TEXT)")

            conn.execute("CREATE TABLE commits(hash TEXT PRIMARY KEY, parent_hash TEXT, merge_parent_hash TEXT)")

            conn.executemany("INSERT INTO commits VALUES (?,?,?)",

                             [("c1", None, None), ("c2", "c1", None), ("orphan", "c1", None)])

            conn.execute("INSERT INTO branch_heads VALUES ('b', 'c2')")

            conn.execute("INSERT INTO metadata VALUES ('head', 'c2'), ('current_branch', 'b')")

            conn.commit()  # left in the WAL (connection still open) like a killed writer

            with OakGraph(db) as graph:

                self.assertEqual(graph.integrity(), "ok")

                self.assertEqual(graph.reachable(), {"c1", "c2"})

                self.assertNotIn("orphan", graph.reachable())

                self.assertTrue(graph.is_ancestor("c1", "c2"))

            conn.close()

class RowContractTests(unittest.TestCase):

    def _runner(self, temp: Path, extra: list[str] | None = None):

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

                                          "--workdir", str(temp / "work"), "--bench-id", "20260928T000000Z",

                                          "--git-bin", GIT or "git", *(extra or [])])

        runner = run_robustness.Runner(args)

        runner.versions = {"git": "git version test", "oak": "oak test"}

        return runner

    def test_emitted_rows_satisfy_the_robustness_lane_contract(self):

        with tempfile.TemporaryDirectory() as temp:

            runner = self._runner(Path(temp))

            runner.emit("git", "crash_time_v1", "commit", "crash_time",

                        {"classification": HELD, "invariants": {"repo_opens": True}, "fault_point": "t=1ms",

                         "elapsed_ms": 3.0, "remote_transport": "local_file", "remote_server": "local",

                         "remote_repo": "../srv/remote.git"})

            runner.emit("oak", "crash_time_v1", "merge_hosted", "crash_time", {"unsupported": "404 on serve"})

            runner.emit("git", "crash_time_v1", "commit", "crash_time",

                        {"classification": VIOLATED, "invariants": {"repo_opens": False}, "fault_point": "t=2ms"})

            rows = runner.rows

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

            self.assertEqual([r["returncode"] for r in rows], [0, 77, 1])

            self.assertEqual(rows[1]["classification"], UNSUPPORTED)

            self.assertTrue(rows[1]["skipped"])

            self.assertEqual(rows[2]["violations"], ["repo_opens"])

            self.assertEqual([r["run"] for r in rows], [0, 0, 1])

            contract = LANE_TABLE["robustness"]

            from oakbench.lane_contract import (catalog_operations, check_catalog_membership,

                                                check_measurement_source, check_transport_purity)

            self.assertEqual(check_measurement_source(rows, contract.allowed_sources), [])

            self.assertEqual(check_transport_purity(rows), [])

            self.assertEqual(check_catalog_membership(rows, catalog_operations(contract.catalog_path)), [])

            partial = Path(temp) / "results" / "20260928T000000Z.robustness.partial.jsonl"

            self.assertEqual(len(partial.read_text().splitlines()), 3)

    def test_summary_table_counts_classes(self):

        rows = [{"subject_kind": "oak", "scenario": "s", "operation": "push", "classification": c,

                 "violations": ["x"] if c == VIOLATED else [], "fault_point": "p"}

                for c in (HELD, HELD, VIOLATED, UNSUPPORTED)]

        text = run_robustness.summarize(rows, "id", 60)

        self.assertIn("| oak | s | push | 2 | 1 | 0 | 1 | 0 |", text)

        self.assertIn("## Violations", text)

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

class GitEndToEndTests(unittest.TestCase):

    def test_time_crash_on_git_commit_is_judged_and_written(self):

        with tempfile.TemporaryDirectory() as temp:

            args = run_robustness.parse_args(["--subjects", "git", "--families", "crash_time,hosted", "--ops",

                                              "commit,merge_hosted", "--quick", "--results", str(Path(temp) / "r"),

                                              "--workdir", str(Path(temp) / "w"), "--bench-id", "20260928T000001Z",

                                              "--git-bin", GIT])

            code = run_robustness.Runner(args).run()

            self.assertEqual(code, 0)

            rows = [json.loads(line) for line in

                    (Path(temp) / "r" / "20260928T000001Z.robustness.jsonl").read_text().splitlines()]

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

            commit_rows = [r for r in rows if r["operation"] == "commit"]

            self.assertEqual(len(commit_rows), 3)

            for row in commit_rows:

                # a commit that beats every re-aimed kill (host load) is fault_not_fired, never held

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

                self.assertEqual(row["classification"] == FAULT_NOT_FIRED, row["fault_fired"] is False)

                self.assertIn("acked_local_reachable", row["invariants"])

                self.assertEqual(row["remote_transport"], "local_file")

            self.assertTrue(all(r["classification"] == UNSUPPORTED for r in rows if r["operation"] == "merge_hosted"))

if __name__ == "__main__":

    unittest.main()