"""Conformance gate semantics and its wiring into devloop and publish_gate."""

import contextlib

import io

import json

from pathlib import Path

import sys

import tempfile

import unittest

from unittest import mock

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

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

import devloop  # noqa: E402

import publish_gate  # noqa: E402

from oakbench import conformance_gate as gate  # noqa: E402

from oakbench.conformance import SCENARIOS  # noqa: E402

from oakbench.conformance_matrix import BEHAVIOURS  # noqa: E402

ALL = list(SCENARIOS) + [b.id for b in BEHAVIOURS]

PROFILE = "darwin-ci-ni"

SHA = "b" * 64

OBSERVED = json.loads((ROOT / "tests/data/conformance_oak_0105_failures.json").read_text())["details"]

def conformance_row(behaviour, code=0, failed=None, subject="oak", sha=SHA, profile=PROFILE, **extra):

    """failed: {leg: detail}; legacy scenarios use {'behaviour': failure}."""

    row = {"evidence_class": "diagnostic_conformance", "subject": subject, "scenario": behaviour,

           "returncode": code, "binary_sha256": sha, "fs_profile": profile, **extra}

    if code == 1:

        failed = failed or {"clone_fs": "injected"}

        if behaviour in SCENARIOS:

            row["failure"] = failed["behaviour"]

        else:

            row["legs"] = {leg: {"status": "fail", "detail": detail} for leg, detail in failed.items()}

            row["failed_legs"] = sorted(failed)

            first = sorted(failed)[0]

            row["failure"] = f"{first}: {failed[first]}"

    if code == 77:

        row.setdefault("skip_reason", "filesystem capability unavailable: case_sensitive_fs")

    return row

def healthy_rows(expected, subject="oak", sha=SHA, profile=PROFILE):

    """Oak 0.105.0 as observed: expected failures with their real details, recorded skips."""

    skips = set(expected["coverage"].get(profile, {}).get("expected_skips", []))

    rows = []

    for behaviour in ALL:

        if behaviour in expected["entries"] and gate.entry_applies(expected["entries"][behaviour], profile):

            rows.append(conformance_row(behaviour, 1, OBSERVED[behaviour], subject, sha, profile))

        elif behaviour in skips:

            rows.append(conformance_row(behaviour, 77, None, subject, sha, profile))

        else:

            rows.append(conformance_row(behaviour, 0, None, subject, sha, profile))

    return rows

def replace(rows, row):

    return [r for r in rows if r["scenario"] != row["scenario"] or r["subject"] != row["subject"]] + [row]

class EvaluateTests(unittest.TestCase):

    def setUp(self):

        self.expected = gate.load_expected()

    def verdict(self, rows):

        return gate.evaluate(rows, self.expected, ALL)

    def test_observed_oak_0105_details_match_every_entry(self):

        self.assertEqual(set(OBSERVED), set(self.expected["entries"]))

        result = self.verdict(healthy_rows(self.expected))

        self.assertEqual(result["verdict"], "PASS", result)

        self.assertEqual(len(result["expected_failures"]) + len(result["refused"]) + len(result["intermittent"]),

                         len(self.expected["entries"]))

        self.assertEqual(result["intermittent"], ["large_file_chunked_v1: fb-547"])

        self.assertFalse(any("unfiled tickets" in note for note in result["notes"]))

        self.assertEqual(sorted(x.split(":")[0] for x in result["refused"]),

                         ["path_control_chars_v1", "path_newline_v1", "path_roundtrip_v1", "path_tab_v1"])

    def test_new_failure_is_a_regression(self):

        rows = replace(healthy_rows(self.expected), conformance_row("eol_crlf_v1", 1, {"clone_fs": "changed=['crlf.txt']"}))

        result = self.verdict(rows)

        self.assertEqual(result["verdict"], "REGRESSED")

        self.assertIn("eol_crlf_v1", result["regressions"][0])

    def test_same_legs_with_a_different_error_is_a_regression(self):

        cases = {

            # EEXIST turning into a path escape at the same leg.

            "file_to_dir_v1": {"pull_fs": "extra=['../../etc/evil']"},

            # A policy refusal turning into a crash at the same leg.

            "path_tab_v1": {"commit_push": "oak commit --no-verify exited 101: thread 'main' panicked at core/src/tree.rs"},

            # The legacy single-leg scenario cannot hide a different failure.

            "path_roundtrip_v1": {"behaviour": "state mismatch: missing=[], extra=['tab'], changed=['nested/leaf']"},

            # A pattern must match the whole detail, not a substring.

            "file_to_deep_dir_v1": {"pull_fs": "oak pull exited 1: error: IO error: Not a directory (os error 20); "

                                               "extra=['../outside']"},

        }

        for behaviour, failed in cases.items():

            with self.subTest(behaviour=behaviour):

                result = self.verdict(replace(healthy_rows(self.expected), conformance_row(behaviour, 1, failed)))

                self.assertEqual(result["verdict"], "REGRESSED", result)

                self.assertIn("changed", result["regressions"][0])

    def test_known_failure_at_different_legs_is_a_regression(self):

        failed = dict(OBSERVED["file_to_dir_v1"], clone_fs="missing=['node/child.txt']")

        result = self.verdict(replace(healthy_rows(self.expected), conformance_row("file_to_dir_v1", 1, failed)))

        self.assertEqual(result["verdict"], "REGRESSED")

        self.assertIn("fb-546", result["regressions"][0])

    def test_fix_flips_to_fixed_without_regression(self):

        result = self.verdict(replace(healthy_rows(self.expected), conformance_row("path_tab_v1", 0)))

        self.assertEqual(result["verdict"], "PASS")

        self.assertIn("fb-531", result["fixed"][0])

    def test_intermittent_entry_pass_is_a_pass_and_other_failures_regress(self):

        rows = replace(healthy_rows(self.expected), conformance_row("large_file_chunked_v1", 0))

        result = self.verdict(rows)

        self.assertEqual(result["verdict"], "PASS")

        self.assertEqual(result["fixed"], [])

        self.assertIn("large_file_chunked_v1", result["passed"])

        rows = replace(rows, conformance_row("large_file_chunked_v1", 1, {"clone_fs": "changed=['large.bin']"}))

        self.assertEqual(self.verdict(rows)["verdict"], "REGRESSED")

    def test_profile_scoped_entry_is_not_fixed_or_expected_elsewhere(self):

        linux = healthy_rows(self.expected, profile="linux-cs-ns")

        # Case-only rename is scoped to darwin-ci-ni: a pass on Linux is a plain pass...

        rows = replace(linux, conformance_row("path_case_only_rename_v1", 0, profile="linux-cs-ns"))

        result = self.verdict(rows)

        self.assertEqual(result["fixed"], [])

        self.assertIn("path_case_only_rename_v1", result["passed"])

        self.assertTrue(any("no recorded coverage" in n for n in result["notes"]))

        # ...and the darwin failure shape on Linux is a regression, not an xfail.

        rows = replace(linux, conformance_row("path_case_only_rename_v1", 1, OBSERVED["path_case_only_rename_v1"],

                                              profile="linux-cs-ns"))

        self.assertEqual(self.verdict(rows)["verdict"], "REGRESSED")

    def test_coverage_loss_pass_or_xfail_to_skip_is_unmeasured(self):

        for behaviour in ("eol_lf_v1", "file_to_dir_v1"):

            with self.subTest(behaviour=behaviour):

                rows = replace(healthy_rows(self.expected), conformance_row(behaviour, 77))

                result = self.verdict(rows)

                self.assertEqual(result["verdict"], "UNMEASURED")

                self.assertIn("coverage loss", result["unmeasured"][0])

    # QA round 2: skips must never add up to a PASS.

    def all_skipped(self, profile=PROFILE, drop_profile=False):

        rows = [conformance_row(b, 77, None, profile=profile) for b in ALL]

        if drop_profile:

            for row in rows:

                row.pop("fs_profile")

        return rows

    def test_all_skip_on_unrecorded_profile_is_unmeasured(self):

        result = self.verdict(self.all_skipped(profile="linux-cs-ns"))

        self.assertEqual(result["verdict"], "UNMEASURED")

        self.assertTrue(any("zero measured" in item for item in result["unmeasured"]))

        self.assertTrue(any("unrecorded profile" in item for item in result["unmeasured"]))

    def test_rows_without_fs_profile_are_unmeasured(self):

        result = self.verdict(self.all_skipped(drop_profile=True))

        self.assertEqual(result["verdict"], "UNMEASURED")

        self.assertTrue(any("without fs_profile" in item for item in result["unmeasured"]))

        # Even one otherwise-healthy row missing the field is unmeasured.

        rows = healthy_rows(self.expected)

        rows[0].pop("fs_profile")

        self.assertEqual(self.verdict(rows)["verdict"], "UNMEASURED")

    def test_zero_measured_is_unmeasured_even_on_recorded_profile(self):

        result = self.verdict(self.all_skipped())

        self.assertEqual(result["verdict"], "UNMEASURED")

        self.assertEqual(result["measured"], 0)

    def test_unrecorded_profile_allows_only_known_filesystem_dependent_skips(self):

        linux = "linux-cs-ns"

        rows = healthy_rows(self.expected, profile=linux)

        # Legitimate on Linux: the case-insensitive-only behaviour skips.

        ok = replace(rows, conformance_row("path_case_only_rename_v1", 77, profile=linux,

                                           skip_reason="filesystem capability unavailable: case_insensitive_fs"))

        ok = replace(ok, conformance_row("huge_file_sparse_v1", 77, profile=linux, skip_reason="gated: off"))

        self.assertEqual(self.verdict(ok)["verdict"], "PASS")

        # A broken probe skipping symlink behaviours is not accepted.

        bad = replace(ok, conformance_row("symlink_relative_v1", 77, profile=linux,

                                          skip_reason="filesystem capability unavailable: symlink"))

        result = self.verdict(bad)

        self.assertEqual(result["verdict"], "UNMEASURED")

        self.assertIn("symlink_relative_v1", result["unmeasured"][0])

    def test_mixed_profiles_are_unmeasured(self):

        rows = replace(healthy_rows(self.expected), conformance_row("eol_lf_v1", 0, profile="linux-cs-ns"))

        self.assertEqual(self.verdict(rows)["verdict"], "UNMEASURED")

    def test_missing_rows_harness_errors_and_reader_skips_are_unmeasured(self):

        rows = [r for r in healthy_rows(self.expected) if r["scenario"] != "eol_lf_v1"]

        self.assertEqual(self.verdict(rows)["verdict"], "UNMEASURED")

        self.assertEqual(self.verdict(rows + [conformance_row("eol_lf_v1", 2, failure="harness")])["verdict"], "UNMEASURED")

        reader = conformance_row("eol_lf_v1", 77, skip_reason="leg unmeasured: reader bound")

        self.assertEqual(self.verdict(rows + [reader])["verdict"], "UNMEASURED")

        self.assertEqual(gate.evaluate([], self.expected)["verdict"], "UNMEASURED")

    def test_git_rows_never_gate(self):

        rows = healthy_rows(self.expected) + [conformance_row("eol_lf_v1", 1, {"clone_fs": "x"}, subject="git")]

        result = self.verdict(rows)

        self.assertEqual(result["verdict"], "PASS")

        self.assertEqual(len(result["git_reference_failures"]), 1)

    def test_list_validation(self):

        good = {"behaviour": "a", "ticket": "fb-1", "failed_legs": {"clone_fs": "x"}}

        bad = [dict(good, ticket="proposed:x"), dict(good, ticket="TODO"), dict(good, ticket="pending:Bad Slug"), dict(good, failed_legs=["clone_fs"]),

               dict(good, failed_legs={"clone_fs": ""}), dict(good, outcome="flaky"), dict(good, fs_profiles=[])]

        with tempfile.TemporaryDirectory() as tmp:

            path = Path(tmp) / "x.json"

            for entry in bad:

                path.write_text(json.dumps({"schema_version": 2, "entries": [entry]}))

                with self.assertRaises(ValueError):

                    gate.load_expected(path)

            path.write_text(json.dumps({"schema_version": 1, "entries": [good]}))

            with self.assertRaises(ValueError):

                gate.load_expected(path)

    def test_catch_all_patterns_are_rejected(self):

        broad = [".*", ".+", "(.*)", "(?:.*)(?:.*)", "[\\s\\S]*", ".*?", "\\S+.*", "x|.*", ".*error.*",

                 ".*panicked.*", ".*corrupt.*", "[a-z ]+", "(panicked|corrupt|.*)", "zzz-sentinel",

                 "[0-9a-f]{64}", "\\w{3,}\\d", "[^\\n]{3,}\\.\\s*", "a?b?c?.*", "(error)?.+",

                 # Oak/git-shaped catch-alls (leg detail shape "oak <cmd> exited N: error: ...")

                 ".*oak.*", "(?:oak|git) .*", "oak .* exited \\d+: .*", ".*error: .*", "oak \\w+ .*",

                 "oak (commit|push|pull|clone) .*exited \\d+: error: .*", "command .* failed \\(\\d+\\): .*",

                 "oak push --repo exited \\d+: .*", "(?:oak|git) \\S+( \\S+)* exited [0-9]+: (?:error|fatal): .+"]

        for pattern in broad:

            self.assertIsNotNone(gate.pattern_too_broad(pattern), pattern)

        with tempfile.TemporaryDirectory() as tmp:

            path = Path(tmp) / "x.json"

            for pattern in broad:

                entry = {"behaviour": "a", "ticket": "fb-1", "failed_legs": {"clone_fs": pattern}}

                path.write_text(json.dumps({"schema_version": 2, "entries": [entry]}))

                with self.assertRaisesRegex(ValueError, "catch-all", msg=pattern):

                    gate.load_expected(path)

            entry = {"behaviour": "a", "ticket": "fb-1", "failed_legs": {"clone_fs": "(unclosed"}}

            path.write_text(json.dumps({"schema_version": 2, "entries": [entry]}))

            with self.assertRaisesRegex(ValueError, "invalid"):

                gate.load_expected(path)

    def test_specific_patterns_are_accepted(self):

        specific = ["oak pull exited 1: error: IO error: File exists \\(os error 17\\)",

                    "missing=\\['file\\.txt'\\]", "exit(ed)? \\d+", "(foo|bar) failed at \\d+"]

        for pattern in specific:

            self.assertIsNone(gate.pattern_too_broad(pattern), pattern)

        # every shipped entry is specific

        for entry in gate.load_expected()["entries"].values():

            for pattern in entry["patterns"].values():

                self.assertIsNone(gate.pattern_too_broad(pattern.pattern), pattern.pattern)

class WiringTests(unittest.TestCase):

    def setUp(self):

        self.expected = gate.load_expected()

    def test_devloop_lane_is_default_and_maps_verdicts(self):

        self.assertIn("conformance", devloop.LANES)

        regressions, notes = devloop.conformance_checks(healthy_rows(self.expected))

        self.assertEqual(regressions, [])

        self.assertFalse(any(n.startswith("UNMEASURED") for n in notes))

        rows = replace(healthy_rows(self.expected), conformance_row("eol_lf_v1", 1, {"pull_fs": "x"}))

        regressions, notes = devloop.conformance_checks(rows)

        self.assertTrue(regressions and regressions[0].startswith("[conformance] CONFORMANCE eol_lf_v1"))

        rows = [r for r in healthy_rows(self.expected) if r["scenario"] != "eol_lf_v1"]

        regressions, notes = devloop.conformance_checks(rows)

        self.assertTrue(any(n.startswith("UNMEASURED conformance") for n in notes))

    def gate_output(self, rows):

        with tempfile.TemporaryDirectory() as tmp:

            path = Path(tmp) / "rows.jsonl"

            path.write_text("".join(json.dumps(r) + "\n" for r in rows))

            with contextlib.redirect_stdout(io.StringIO()) as out:

                code = publish_gate.main([str(path)])

            return code, out.getvalue()

    def oak_row(self, sha=SHA, **extra):

        return {"subject": "oak_installed", "binary_sha256": sha, "returncode": 0,

                "source": {"oak_hash": "abc", "oak_status": ""}, **extra}

    def test_requires_conformance_for_oak_rows(self):

        code, out = self.gate_output([self.oak_row()])

        self.assertIn("FAIL   conformance: Oak rows published without", out)

        code, out = self.gate_output([self.oak_row(), *healthy_rows(self.expected)])

        self.assertIn("PASS   [bbbbbbbbbbbb] conformance (oak, darwin-ci-ni): PASS", out)

        self.assertNotIn("FAIL   conformance", out)

    def test_oak_rows_without_binary_identity_fail(self):

        anonymous = self.oak_row()

        anonymous.pop("binary_sha256")

        code, out = self.gate_output([anonymous, *healthy_rows(self.expected)])

        self.assertIn("Oak rows lack binary identity", out)

        via_details = self.oak_row(subject_details={"oak_installed": {"binary_sha256": SHA}})

        via_details.pop("binary_sha256")

        code, out = self.gate_output([via_details, *healthy_rows(self.expected)])

        self.assertNotIn("lack binary identity", out)

    def test_rows_of_another_binary_never_satisfy_or_mix(self):

        other = "c" * 64

        code, out = self.gate_output([self.oak_row(sha=other), *healthy_rows(self.expected)])

        self.assertIn(f"conformance[{other[:12]}]: UNMEASURED", out)

        # A good binary's rows cannot mask a regressed binary's row, and vice versa.

        mixed = healthy_rows(self.expected) + replace(healthy_rows(self.expected, sha=other),

                                                      conformance_row("eol_lf_v1", 1, {"pull_fs": "x"}, sha=other))

        code, out = self.gate_output([self.oak_row(), self.oak_row(sha=other), *mixed])

        self.assertIn(f"conformance[{other[:12]}]: REGRESSED", out)

        self.assertNotIn(f"conformance[{SHA[:12]}]: REGRESSED", out)

    def test_publish_requires_a_recorded_profile(self):

        linux = [dict(row, fs_profile="linux-cs-ns") for row in healthy_rows(self.expected, profile="linux-cs-ns")]

        code, out = self.gate_output([self.oak_row(), *linux])

        self.assertEqual(code, 1)

        self.assertIn("has no recorded coverage", out)

        skipped = [conformance_row(b, 77, None, profile="linux-cs-ns") for b in ALL]

        code, out = self.gate_output([self.oak_row(), *skipped])

        self.assertIn("UNMEASURED", out)

        self.assertNotIn("PASS   [", out)

        missing = healthy_rows(self.expected)

        for row in missing:

            row.pop("fs_profile")

        code, out = self.gate_output([self.oak_row(), *missing])

        self.assertIn("FAIL   conformance[", out)

        self.assertNotIn("PASS   [", out)

    def test_pending_tickets_refused_by_publish_but_only_noted_by_devloop(self):

        base = gate.load_expected()

        for outcome in ("intermittent", "bug", "refused"):

            entries = dict(base["entries"])

            entries["large_file_chunked_v1"] = dict(entries["large_file_chunked_v1"], ticket="pending:db-locked",

                                                    outcome=outcome)

            expected = dict(base, entries=entries)

            self.assertEqual(gate.pending_tickets(expected), [("large_file_chunked_v1", "pending:db-locked", outcome)])

            with mock.patch.object(gate, "load_expected", return_value=expected):

                code, out = self.gate_output([self.oak_row(), *healthy_rows(base)])

                self.assertEqual(code, 1, outcome)

                self.assertIn("unfiled pending: tickets", out)

                self.assertIn("large_file_chunked_v1 (pending:db-locked", out)

                regressions, notes = devloop.conformance_checks(healthy_rows(base))

            # devloop still runs and only notes the placeholder

            self.assertEqual(regressions, [])

        self.assertEqual(gate.pending_tickets(base), [])

    def test_stale_list_fails(self):

        rows = replace(healthy_rows(self.expected), conformance_row("path_tab_v1", 0))

        code, out = self.gate_output([self.oak_row(), *rows])

        self.assertIn("expected-failures list is stale", out)

if __name__ == "__main__":

    unittest.main()