"""Instrument tests for suites/merge (classification, laws, oracle parsing, vendoring).

Hermetic: no Oak binary and no network. Git-backed tests skip when git is absent.

"""

import hashlib

import json

from pathlib import Path

import shutil

import sys

import tempfile

import unittest

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

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

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

from mergesuite import classify as cl  # noqa: E402

from mergesuite.gitops import (Git, git_conflicts, normalize_git_conflict_type,  # noqa: E402

                               parse_merge_tree_z, prepare_git_home, worktree_manifest)

from mergesuite.oak_adapter import parse_conflict_show, summarize_prediction  # noqa: E402

VENDOR = ROOT / "suites" / "merge" / "vendor" / "gitoxide"

def v(tag):

    return ("file", hashlib.sha256(tag.encode()).hexdigest())

def oak(tree, conflicts=(), clean=None):

    return {"tree": tree, "conflicts": {p: [] for p in conflicts},

            "clean": (not conflicts) if clean is None else clean, "error": None, "error_kind": None}

def gitres(tree, conflicts=()):

    return {"tree": tree, "conflicts": {p: ["content"] for p in conflicts}, "clean": not conflicts}

class ClassifyTests(unittest.TestCase):

    base = {"a": v("a0"), "b": v("b0")}

    ours = {"a": v("a1"), "b": v("b0")}

    theirs = {"a": v("a0"), "b": v("b1")}

    merged = {"a": v("a1"), "b": v("b1")}

    def test_equal(self):

        r = cl.classify(gitres(self.merged), oak(self.merged), self.base, self.ours, self.theirs)

        self.assertEqual(r["class"], "equal-to-git")

    def test_lost_change_is_oak_wrong_even_if_git_agrees(self):

        # Oak drops theirs' change to b.

        r = cl.classify(gitres(self.merged), oak(self.ours), self.base, self.ours, self.theirs)

        self.assertEqual(r["class"], "oak-wrong")

        self.assertIn("lost_change: b", r["reasons"][0])

    def test_lost_deletion(self):

        theirs = {"a": v("a0")}  # theirs deletes b

        r = cl.classify(gitres({"a": v("a1")}), oak({"a": v("a1"), "b": v("b0")}),

                        self.base, self.ours, theirs)

        self.assertEqual(r["class"], "oak-wrong")

    def test_fabricated_content(self):

        r = cl.classify(gitres(self.merged), oak({"a": v("a1"), "b": v("zzz")}),

                        self.base, self.ours, self.theirs)

        self.assertEqual(r["class"], "oak-wrong")

        self.assertIn("fabricated", r["reasons"][0])

    def test_false_conflict_is_differs_but_valid(self):

        r = cl.classify(gitres(self.merged), oak(self.merged, conflicts=["a"]),

                        self.base, self.ours, self.theirs)

        self.assertEqual(r["class"], "differs-but-valid")

        self.assertTrue(any("oak_conflicts_where_git_clean" in x for x in r["reasons"]))

    def test_rename_not_detected_keeps_both_sides(self):

        # ours renames a->c; theirs edits b. git moves; Oak path-level result matches git here,

        # but when git conflicts and Oak keeps everything, it is differs-but-valid.

        base = {"a": v("a0")}

        ours = {"b": v("a0")}

        theirs = {"c": v("a0")}

        r = cl.classify(gitres({"b": v("a0"), "c": v("a0")}, conflicts=["a", "b", "c"]),

                        oak({"b": v("a0"), "c": v("a0")}), base, ours, theirs)

        self.assertEqual(r["class"], "differs-but-valid")

        self.assertTrue(any("oak_clean_where_git_conflicts" in x for x in r["reasons"]))

    def test_expectation_violation_wins(self):

        r = cl.classify(gitres(self.merged), oak(self.merged), self.base, self.ours, self.theirs,

                        expect={"paths": {"a": v("a0")[1]}})

        self.assertEqual(r["class"], "oak-wrong")

        self.assertIn("expectation_violated", r["reasons"][0])

    def test_expectation_violation_still_reports_law_reasons(self):

        # Oak drops theirs' change to b AND violates the stated expectation: both are listed.

        r = cl.classify(gitres(self.merged), oak(self.ours), self.base, self.ours, self.theirs,

                        expect={"clean": False, "label": "lbl"})

        self.assertEqual(r["class"], "oak-wrong")

        self.assertEqual(r["reasons"], ["expectation_violated[lbl]: clean=True expected False", "lost_change: b"])

    def test_t6416_shape_known_violation_maps_and_new_loss_is_listed(self):

        from mergesuite.runner import load_known_bugs, match_known_bug

        bugs = load_known_bugs()

        # criss-cross: no unique base. Main keeps `file`, the branch deleted it; Oak picks main's side.

        ours = {"keep": v("k"), "ours.txt": v("o")}

        theirs = {"file": v("B"), "keep": v("k")}

        expect = {"clean": False, "label": "criss_cross_divergent_resolution"}

        git = gitres({"keep": v("k"), "ours.txt": v("o")}, conflicts=["file"])

        as_landed = {"keep": v("k"), "ours.txt": v("o"), "file": v("B")}

        r = cl.classify(git, oak(as_landed), None, ours, theirs, expect=expect)

        self.assertEqual(r["reasons"], ["expectation_violated[criss_cross_divergent_resolution]: clean=True expected False"])

        self.assertIsNone(r["evidence"]["law_no_lost_change"])

        row = {"class": r["class"], "reasons": r["reasons"], "scenario": "git-t6416-criss-cross-modify-delete"}

        self.assertEqual(match_known_bug(row, bugs), "fb-541")

        # Oak additionally loses `keep` (both sides agree on it): listed, and the row no longer maps.

        lossy = {"ours.txt": v("o"), "file": v("B")}

        r = cl.classify(git, oak(lossy), None, ours, theirs, expect=expect)

        self.assertEqual(r["reasons"][1:], ["lost_change: keep"])

        self.assertIsNone(match_known_bug(dict(row, reasons=r["reasons"]), bugs))

        # committed markers on a both-sides-changed file are also listed without a base

        ours2, theirs2 = dict(ours, t=v("t-o")), dict(theirs, t=v("t-t"))

        texts = {"t": {"base": b"", "ours": b"a\n", "theirs": b"b\n", "merged": b"<<<<<<< x\na\n=======\nb\n>>>>>>> y\n"}}

        r = cl.classify(git, oak(dict(as_landed, t=v("m"))), None, ours2, theirs2, expect=expect, texts=texts)

        self.assertIn("committed_conflict_markers: t", r["reasons"])

    def test_base_free_checks(self):

        ours = {"same": v("s"), "o_only": v("o"), "both": v("bo")}

        theirs = {"same": v("s"), "both": v("bt")}

        law = cl.base_free_checks(ours, theirs, {"both": v("bo"), "o_only": v("o")}, set())

        self.assertEqual(law["lost"], ["same"])

        law = cl.base_free_checks(ours, theirs, {"same": v("zz"), "both": v("bo")}, set())

        self.assertEqual((law["fabricated"], law["lost"]), (["same"], []))

        law = cl.base_free_checks(ours, theirs, {"same": ("exec", v("s")[1])}, {"both"})

        self.assertEqual(law["mode_lost"], ["same"])

        self.assertFalse(any(cl.base_free_checks(ours, theirs, {"same": v("s")}, {"both"}).values()))

    def test_expect_absent_and_clean(self):

        r = cl.classify(gitres(self.merged), oak(self.merged), self.base, self.ours, self.theirs,

                        expect={"clean": False})

        self.assertEqual(r["class"], "oak-wrong")

        r = cl.classify(gitres(self.merged), oak(self.merged), self.base, self.ours, self.theirs,

                        expect={"paths": {"gone": None}, "clean": True})

        self.assertEqual(r["class"], "equal-to-git")

    def test_unsupported_and_harness_and_oak_error(self):

        self.assertEqual(cl.classify(None, None, {}, {}, {}, unsupported="gitlink")["class"], "oak-unsupported")

        self.assertEqual(cl.classify(None, None, {}, {}, {}, harness_error="x")["class"], "harness-error")

        self.assertEqual(cl.classify(gitres({}), {"error_kind": "harness", "error": "shim"}, {}, {}, {})["class"],

                         "harness-error")

        r = cl.classify(gitres({}), {"error_kind": "oak", "error": "boom", "clean": None}, {}, {}, {})

        self.assertEqual(r["class"], "oak-wrong")

        self.assertEqual(cl.classify(gitres({}), None, {}, {}, {})["class"], "harness-error")

    def test_no_unique_base_leaves_law_unmeasured(self):

        r = cl.classify(gitres(self.merged), oak(self.merged), None, self.ours, self.theirs)

        self.assertEqual(r["class"], "equal-to-git")

        self.assertIsNone(r["evidence"]["law_no_lost_change"])

    def test_text_merge_law(self):

        base = b"1\n2\n3\n4\n5\n6\n"

        o = b"1 ours\n2\n3\n4\n5\n6\n"

        t = b"1\n2\n3\n4\n5\n6 theirs\n"

        good = b"1 ours\n2\n3\n4\n5\n6 theirs\n"

        self.assertTrue(cl.text_preserves_sides(base, o, t, good))

        self.assertFalse(cl.text_preserves_sides(base, o, t, o))

        self.assertFalse(cl.text_preserves_sides(base, o, t, b"<<<<<<< a\n" + good))

        trees = ({"f": v("b")}, {"f": v("o")}, {"f": v("t")})

        texts = {"f": {"base": base, "ours": o, "theirs": t, "merged": good}}

        law = cl.no_lost_change(*trees, {"f": v("m")}, set(), texts)

        self.assertEqual(law["text_merged"], ["f"])

        texts["f"]["merged"] = o

        self.assertEqual(cl.no_lost_change(*trees, {"f": v("m")}, set(), texts)["lost"], ["f"])

    def test_one_side_taken_wholesale_is_lost(self):

        # base exec+empty; ours: mode-only change; theirs: content change. Oak keeping theirs'

        # entry (content AND exec bit) drops ours' mode change.

        base = {"w": ("exec", "empty")}

        ours = {"w": ("file", "empty")}

        theirs = {"w": ("exec", "123")}

        law = cl.no_lost_change(base, ours, theirs, {"w": ("exec", "123")}, set())

        self.assertEqual(law["mode_lost"], ["w"])

        # a content change dropped wholesale stays a plain lost change

        law = cl.no_lost_change({"f": v("0")}, {"f": v("1")}, {"f": v("2")}, {"f": v("1")}, set())

        self.assertEqual(law["lost"], ["f"])

        law = cl.no_lost_change(base, ours, theirs, {"w": ("file", "123")}, set(),

                                {"w": {"base": b"", "ours": b"", "theirs": b"1\n2\n3\n", "merged": b"1\n2\n3\n"}})

        self.assertEqual(law["text_merged"], ["w"])

    def test_conflicted_paths_are_exempt_from_law(self):

        law = cl.no_lost_change(self.base, self.ours, self.theirs, {}, {"a", "b"})

        self.assertEqual(law, {"lost": [], "mode_lost": [], "markers": [], "markers_no_final_newline": [], "markers_symlink": [], "fabricated": [], "text_merged": []})

    def test_text_merged_but_mode_dropped_is_oak_wrong(self):

        # QA shape generated/s20260928-022: ours edits lines, theirs edits other lines AND chmods.

        base, ours, theirs = {"f": ("file", "b")}, {"f": ("file", "o")}, {"f": ("exec", "t")}

        merged = b"1 ours\n2\n3\n4\n5\n6 theirs\n"

        texts = {"f": {"base": b"1\n2\n3\n4\n5\n6\n", "ours": b"1 ours\n2\n3\n4\n5\n6\n",

                       "theirs": b"1\n2\n3\n4\n5\n6 theirs\n", "merged": merged}}

        bad = cl.classify(gitres({"f": ("exec", "m")}), oak({"f": ("file", "m")}), base, ours, theirs, texts=texts)

        self.assertEqual(bad["class"], "oak-wrong")

        self.assertTrue(bad["reasons"][0].startswith("lost_mode_change"))

        good = cl.classify(gitres({"f": ("exec", "m")}), oak({"f": ("exec", "m")}), base, ours, theirs, texts=texts)

        self.assertEqual(good["class"], "equal-to-git")

    def test_same_content_different_modes_taken_wholesale_is_mode_loss(self):

        # both sides rename to the same path with different modes; Oak keeps one side silently

        law = cl.no_lost_change({}, {"t": ("file", "x")}, {"t": ("exec", "x")}, {"t": ("exec", "x")}, set())

        self.assertEqual(law["mode_lost"], ["t"])

    def test_symlink_markers_are_distinguished(self):

        ev = {"base": b"", "ours": b"a", "theirs": b"b", "merged": b"<<<<<<< x\na=======\nb>>>>>>> main\n"}

        law = cl.no_lost_change({}, {"l": ("symlink", "a")}, {"l": ("symlink", "b")}, {"l": ("symlink", "m")},

                                set(), {"l": ev})

        self.assertEqual(law["markers_symlink"], ["l"])

    def test_committed_markers_have_their_own_reason(self):

        base, ours, theirs = {"f": v("b")}, {"f": v("o")}, {"f": v("t")}

        texts = {"f": {"base": b"a\nx", "ours": b"a\ny", "theirs": b"a\nz",

                       "merged": b"a\n<<<<<<< br\ny||||||| original\nx=======\nz>>>>>>> main\n"}}

        r = cl.classify(gitres({"f": v("g")}, conflicts=["f"]), oak({"f": v("m")}), base, ours, theirs, texts=texts)

        self.assertEqual(r["class"], "oak-wrong")

        self.assertTrue(r["reasons"][0].startswith("committed_conflict_markers_no_final_newline: f"))

    def test_failed_pull_dirty_worktree_reason(self):

        res = {"error_kind": "oak", "error": "IO error", "clean": None,

               "post_failure": {"hazard_worktree_dirty_head_unmoved": True, "worktree_changes_vs_head": 2,

                                "worktree_changes_sample": ["a", "b"]}}

        r = cl.classify(gitres({}), res, {}, {}, {})

        self.assertTrue(r["reasons"][1].startswith("failed_pull_left_worktree_dirty: "))

    def test_failed_pull_hazard_reason(self):

        res = {"error_kind": "oak", "error": "IO error", "clean": None,

               "post_failure": {"hazard_head_moved_worktree_stale": True, "head_before": "a" * 64,

                                "head_after": "b" * 64, "worktree_changes_vs_head": 2,

                                "repull_parent_sync": "up_to_date"}}

        r = cl.classify(gitres({}), res, {}, {}, {})

        self.assertEqual(r["class"], "oak-wrong")

        self.assertTrue(r["reasons"][1].startswith("failed_pull_moved_head_worktree_stale"))

    def test_unknown_class_is_rejected(self):

        self.assertEqual(cl.validate_class("equal-to-git"), "equal-to-git")

        with self.assertRaises(ValueError):

            cl.validate_class("pass")

    def test_symmetry(self):

        f = oak(self.merged, conflicts=["a"])

        r = oak({"a": v("other-side-markers"), "b": v("b1")}, conflicts=["a"])

        self.assertTrue(cl.symmetry(f, r)["holds"])

        self.assertFalse(cl.symmetry(f, oak(self.merged))["holds"])

        self.assertIsNone(cl.symmetry(f, {"clean": None}))

    def test_idempotence(self):

        self.assertIsNone(cl.idempotence_holds({}))

        good = {"idempotence": {"returncode": 0, "head_unchanged": True, "tree_unchanged": True}}

        self.assertTrue(cl.idempotence_holds(good))

        good["idempotence"]["tree_unchanged"] = False

        self.assertFalse(cl.idempotence_holds(good))

    def test_oak_conflict_type(self):

        base = {"a": ("file", "1"), "m": ("file", "1")}

        self.assertEqual(cl.oak_conflict_type("a", base, {"a": ("file", "2")}, {}), "modify/delete")

        self.assertEqual(cl.oak_conflict_type("n", base, {"n": ("file", "2")}, {"n": ("file", "3")}), "add/add")

        self.assertEqual(cl.oak_conflict_type("m", base, {"m": ("exec", "1")}, {"m": ("symlink", "1")}), "mode")

class RunnerAccountingTests(unittest.TestCase):

    def test_failed_pull_counts_as_prediction_disagreement(self):

        from mergesuite.runner import Suite

        oak_res = {"error_kind": "oak", "error": "x", "clean": None, "tree": {},

                   "prediction": {"available": True, "clean": True, "conflict_files": []},

                   "post_failure": {"hazard_head_moved_worktree_stale": True}}

        verdict = cl.classify(gitres({}), oak_res, {}, {}, {})

        out = Suite._result_fields(None, gitres({}), oak_res, verdict)

        self.assertIs(out["laws"]["prediction_agrees"], False)

        self.assertTrue(out["post_failure_hazard"])

    LAWS = {k: None for k in ("no_lost_change", "idempotent", "symmetric", "false_conflict",

                              "missed_conflict", "prediction_agrees")}

    def row(self, scenario, reasons, cls="oak-wrong", direction="A-B"):

        from mergesuite.runner import SCHEMA_VERSION

        return {"class": cls, "scenario": scenario, "direction": direction, "origin": "o",

                "reasons": reasons, "laws": dict(self.LAWS), "merge_suite_schema_version": SCHEMA_VERSION}

    def test_known_bug_mapping_positive(self):

        from mergesuite.runner import load_known_bugs, match_known_bug

        bugs = load_known_bugs()

        self.assertEqual({b["id"] for b in bugs}, {"fb-537", "fb-538", "fb-539", "fb-540", "fb-541"})

        cases = [

            (["committed_conflict_markers_symlink: link"], "fb-538"),

            (["committed_conflict_markers_no_final_newline: file"], "fb-537"),

            (["lost_mode_change: w"], "fb-539"),

            (["oak_error: IO error: File exists (os error 17)"], "fb-540"),

            (["oak_error: IO error: Not a directory (os error 20)",

              "failed_pull_left_worktree_dirty: HEAD unchanged, 2 worktree path(s) differ from HEAD (e.g. a)"], "fb-540"),

            (['oak_error: cannot build tree with both file "e" and child path "e/e"'], "fb-540"),

            (["oak_error: refusing to replace non-empty directory '/x/a' with a tracked file",

              "failed_pull_moved_head_worktree_stale: head a -> b"], "fb-540"),

            (["expectation_violated[criss_cross_divergent_resolution]: clean=True expected False"], "fb-541"),

        ]

        for reasons, bug in cases:

            self.assertEqual(match_known_bug(self.row("gix-x", reasons), bugs), bug, reasons)

    def test_known_bug_escape_routes_stay_unmapped(self):

        from mergesuite.runner import load_known_bugs, match_known_bug

        bugs = load_known_bugs()

        escapes = [

            # any other Oak error (e.g. a panic or server 500) is not fb-540

            ("gix-tree/a", ["oak_error: Server error: HTTP 500"]),

            ("gix-tree/a", ["oak_error: thread 'main' panicked at src/x.rs"]),

            # t6416 rows with another reason are not fb-541

            ("git-t6416-criss-cross-modify-delete", ["lost_change: file"]),

            ("git-t6416-criss-cross-modify-delete", ["expectation_violated: clean=True expected False"]),

            # one explained reason plus one unexplained reason stays unmapped

            ("gix-text/x", ["committed_conflict_markers_no_final_newline: f", "lost_change: g"]),

            ("gix-tree/x", ["oak_error: IO error: File exists (os error 17)", "fabricated_content: q"]),

            # markers in files that DO end with a newline are a different root cause

            ("gix-text/x", ["committed_conflict_markers: f"]),

            ("gix-text/x", []),

        ]

        for scenario, reasons in escapes:

            self.assertIsNone(match_known_bug(self.row(scenario, reasons), bugs), (scenario, reasons))

        # only oak-wrong rows are ever mapped

        self.assertIsNone(match_known_bug(self.row("gix-x", ["lost_mode_change: w"], cls="differs-but-valid"), bugs))

    def test_setup_failure_is_not_an_oak_verdict(self):

        res = {"error_kind": "oak-setup", "error": "oak clone exited 101: panicked", "clean": None}

        r = cl.classify(gitres({}), res, {}, {}, {})

        self.assertEqual(r["class"], "harness-error")

        self.assertTrue(r["reasons"][0].startswith("setup_failed: "))

    def test_markers_split_by_final_newline(self):

        base, ours, theirs = {"f": v("b")}, {"f": v("o")}, {"f": v("t")}

        glued = {"base": b"a\nx", "ours": b"a\ny", "theirs": b"a\nz",

                 "merged": b"a\n<<<<<<< br\ny||||||| original\nx=======\nz>>>>>>> main\n"}

        law = cl.no_lost_change(base, ours, theirs, {"f": v("m")}, set(), {"f": glued})

        self.assertEqual(law["markers_no_final_newline"], ["f"])

        clean_eol = {"base": b"a\nx\n", "ours": b"a\ny\n", "theirs": b"a\nz\n",

                     "merged": b"a\n<<<<<<< br\ny\n=======\nz\n>>>>>>> main\n"}

        law = cl.no_lost_change(base, ours, theirs, {"f": v("m")}, set(), {"f": clean_eol})

        self.assertEqual((law["markers"], law["markers_no_final_newline"]), (["f"], []))

    def test_pinned_rows_report_new_and_vanished(self):

        from mergesuite.runner import summarize

        bugs = [{"id": "fb-x", "family": "F", "reason_patterns": ["^lost_mode_change: "],

                 "expected_rows": ["gix-a|A-B", "gix-b|A-B"]},

                {"id": "fb-y", "family": "G", "reason_patterns": ["^nothing"], "expected_rows": ["z|A-B"]}]

        rows = [self.row("gix-a", ["lost_mode_change: w"]), self.row("gix-c", ["lost_mode_change: w"]),

                self.row("gix-d", ["lost_change: q"]), self.row("ok", [], cls="equal-to-git"),

                self.row("gix-b", [], cls="equal-to-git"), self.row("z", [], cls="equal-to-git")]

        summary = summarize(rows, bugs)

        fx = summary["known_bugs"]["fb-x"]

        self.assertEqual((fx["new_rows"], fx["vanished_rows"], fx["status"]),

                         (["gix-c|A-B"], ["gix-b|A-B"], "changed"))

        self.assertEqual(summary["known_bugs"]["fb-y"]["status"], "NOT reproduced (fixed?)")

        self.assertEqual(summary["oak_wrong_unmapped_rows"], ["gix-d|A-B"])

        # a subset run (--groups/--only) that never executed a pinned row reports nothing vanished

        # a pinned row that became a harness error is indeterminate, never "fixed?"

        broken = [self.row("gix-a", ["setup_failed: x"], cls="harness-error"), self.row("gix-b", [], cls="harness-error")]

        fx = summarize(broken, bugs)["known_bugs"]["fb-x"]

        self.assertEqual((fx["status"], fx["vanished_rows"], fx["indeterminate_rows"]),

                         ("harness-error (indeterminate)", [], ["gix-a|A-B", "gix-b|A-B"]))

        subset = summarize(rows[:4], bugs)

        self.assertEqual(subset["known_bugs"]["fb-x"]["vanished_rows"], [])

        self.assertEqual(subset["known_bugs"]["fb-x"]["expected_rows_total"], 2)

        self.assertTrue(subset["known_bugs"]["fb-y"]["status"].startswith("not in this run"))

    def test_stale_schema_rows_are_indeterminate_not_fixed(self):

        from mergesuite.runner import stale_schema_rows, summarize

        bugs = [{"id": "fb-y", "family": "G", "reason_patterns": ["^nothing"], "expected_rows": ["z|A-B"]}]

        rows = [self.row("gix-a", ["lost_mode_change: w"]), self.row("z", [], cls="equal-to-git")]

        self.assertEqual(stale_schema_rows(rows), {})

        self.assertEqual(summarize(rows, bugs)["known_bugs"]["fb-y"]["status"], "NOT reproduced (fixed?)")

        rows[0].pop("merge_suite_schema_version")

        rows[1]["merge_suite_schema_version"] = 1

        self.assertEqual(stale_schema_rows(rows), {"missing": 1, "1": 1})

        summary = summarize(rows, bugs)

        self.assertTrue(summary["known_bugs"]["fb-y"]["status"].startswith("indeterminate"))

        self.assertEqual(summary["stale_schema_rows"], {"missing": 1, "1": 1})

    def test_resummarize_refuses_pre_schema_runs(self):

        import contextlib

        import io

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

        import run_merge_suite

        with tempfile.TemporaryDirectory() as tmp:

            run = Path(tmp) / "run.jsonl"

            fb541 = self.row("git-t6416-criss-cross-modify-delete",

                             ["expectation_violated[criss_cross_divergent_resolution]: clean=True expected False"],

                             direction="branch-into-main-sync")

            fb541["repro"] = "x"

            old = dict(fb541, merge_suite_schema_version=1)

            run.write_text(json.dumps(old) + "\n")

            err = io.StringIO()

            with contextlib.redirect_stderr(err):

                self.assertEqual(run_merge_suite.main(["--workdir", tmp, "--resummarize", str(run)]), 2)

            self.assertIn("predate merge-suite row/reason schema", err.getvalue())

            # a newer-schema row is not called "predates"

            run.write_text(json.dumps(dict(fb541, merge_suite_schema_version=99)) + "\n")

            err = io.StringIO()

            with contextlib.redirect_stderr(err):

                self.assertEqual(run_merge_suite.main(["--workdir", tmp, "--resummarize", str(run)]), 2)

            self.assertIn("NEWER merge suite", err.getvalue())

            self.assertNotIn("predate", err.getvalue())

            # an empty run is an error, not an all-clear

            run.write_text("\n")

            with contextlib.redirect_stderr(io.StringIO()):

                self.assertNotEqual(run_merge_suite.main(["--workdir", tmp, "--resummarize", str(run)]), 0)

            run.write_text(json.dumps(old) + "\n")

            summary_path = Path(tmp) / "s.json"

            with contextlib.redirect_stderr(io.StringIO()):

                run_merge_suite.main(["--workdir", tmp, "--resummarize", str(run), "--allow-stale-schema",

                                      "--summary", str(summary_path)])

            summary = json.loads(summary_path.read_text())

            self.assertTrue(all(b["status"].startswith("indeterminate") for b in summary["known_bugs"].values()))

            self.assertNotIn("fixed?", json.dumps(summary["known_bugs"]))

            run.write_text(json.dumps(fb541) + "\n")

            with contextlib.redirect_stderr(io.StringIO()):

                run_merge_suite.main(["--workdir", tmp, "--resummarize", str(run), "--summary", str(summary_path)])

            summary = json.loads(summary_path.read_text())

            self.assertEqual(summary["known_bugs"]["fb-541"]["rows"], 1)

            self.assertEqual(summary["stale_schema_rows"], {})

    def test_committed_known_bugs_are_pinned(self):

        from mergesuite.runner import load_known_bugs

        for bug in load_known_bugs():

            self.assertTrue(bug["expected_rows"], bug["id"])

            self.assertEqual(len(bug["expected_rows"]), len(set(bug["expected_rows"])), bug["id"])

class OracleParsingTests(unittest.TestCase):

    def test_parse_merge_tree_z(self):

        out = (b"tree123\0"

               b"100644 aaa 1\ta\x00100644 bbb 2\ta\x00100644 ccc 3\ta\x00"

               b"\0"

               b"1\0a\0Auto-merging\0Auto-merging a\n\0"

               b"1\0a\0CONFLICT (contents)\0CONFLICT (content): Merge conflict in a\n\0")

        parsed = parse_merge_tree_z(out)

        self.assertEqual(parsed["tree_oid"], "tree123")

        self.assertEqual(parsed["conflict_stages"], {"a": [1, 2, 3]})

        self.assertEqual(git_conflicts(parsed), {"a": ["content"]})

    def test_parse_clean(self):

        parsed = parse_merge_tree_z(b"t\0\0")

        self.assertEqual(parsed["conflict_stages"], {})

        self.assertEqual(parsed["messages"], [])

    def test_type_vocabulary(self):

        self.assertIsNone(normalize_git_conflict_type("Auto-merging"))

        self.assertEqual(normalize_git_conflict_type("CONFLICT (modify/delete)"), "modify/delete")

        self.assertEqual(normalize_git_conflict_type("CONFLICT (rename/delete)"), "rename/delete")

        self.assertEqual(normalize_git_conflict_type("CONFLICT (directory rename split)"), "directory-rename")

    def test_conflict_show_and_prediction_parsing(self):

        doc = {"conflicts": [{"path": "f", "has_conflict_markers": False, "exists": True}]}

        self.assertEqual(parse_conflict_show(doc, {}), {"f": ["no-markers"]})

        self.assertEqual(parse_conflict_show(None, {"conflict_paths": ["x"]}), {"x": []})

        self.assertEqual(summarize_prediction(None, 6), {"available": False, "returncode": 6})

        pred = summarize_prediction({"prediction_available": True, "clean": False,

                                     "conflict_files": [{"path": "b"}, "a"]}, 0)

        self.assertEqual(pred["conflict_files"], ["a", "b"])

@unittest.skipUnless(shutil.which("git"), "git not installed")

class GitBackedTests(unittest.TestCase):

    def setUp(self):

        self.tmp = Path(tempfile.mkdtemp(prefix="mergesuite-test-"))

        self.home = self.tmp / "home"

        from oakbench.subjects import which_measured_git

        self.git_bin = which_measured_git() or shutil.which("git")

        prepare_git_home(self.home, self.git_bin)

    def tearDown(self):

        shutil.rmtree(self.tmp, ignore_errors=True)

    def test_builder_oracle_and_materialize_roundtrip(self):

        from mergesuite.builder import RepoBuilder, lines

        b = RepoBuilder(self.tmp / "r", self.git_bin, self.home)

        base = b.commit({"f": lines("1", "2", "3"), "x": ("exec", b"#!/bin/sh\n"), "l": ("link", "f")})

        a = b.commit({"f": lines("1 a", "2", "3"), "x": ("exec", b"#!/bin/sh\n"), "l": ("link", "f")}, [base])

        c = b.commit({"f": lines("1", "2", "3 c"), "l": ("link", "f")}, [base], message="unique")

        self.assertEqual(b.merge_bases(a, c), [base])

        merge = b.merge_tree(a, c)

        self.assertTrue(merge["clean"])

        self.assertEqual(sorted(merge["tree"]), ["f", "l"])

        self.assertEqual(merge["tree"]["l"][0], "symlink")

        out = self.tmp / "mat"

        out.mkdir()

        b.materialize(base, out)

        self.assertEqual(worktree_manifest(out), b.tree(base))

        # first parent order is preserved

        m = b.commit({"f": lines("z")}, [a, c])

        self.assertEqual(b.out("rev-parse", m + "^1"), a)

    def test_oracle_reports_conflict_types(self):

        from mergesuite.builder import RepoBuilder, lines

        b = RepoBuilder(self.tmp / "r2", self.git_bin, self.home)

        base = b.commit({"f": lines("1", "2"), "g": lines("g")})

        a = b.commit({"f": lines("1", "A")}, [base])

        c = b.commit({"f": lines("1", "C"), "g": lines("g2")}, [base])

        merge = b.merge_tree(a, c)

        self.assertFalse(merge["clean"])

        self.assertEqual(git_conflicts(merge), {"f": ["content"], "g": ["modify/delete"]})

    def test_sources_build_without_oak(self):

        from mergesuite import sources

        scen = sources.git_t64xx(self.tmp / "t64", self.git_bin, self.home)

        ids = [s["id"] for s in scen]

        self.assertIn("git-t6423-1a", ids)

        self.assertTrue(all(len(s["directions"]) == 2 and s["unsupported"] is None for s in scen))

        native = sources.native_incidents(self.tmp / "nat", self.git_bin, self.home)

        by_id = {s["id"]: s for s in native}

        cc = by_id["incident/criss-cross-tie-break-side-newer"]["directions"][0]

        self.assertEqual(len(cc["merge_bases"]), 2)  # two best common ancestors by construction

        self.assertIsNone(cc["trees"]["base"])

        skew = by_id["incident/clock-skew-merge-base-skewed"]["directions"][0]

        self.assertEqual(len(skew["merge_bases"]), 1)

        # self-merges produce a single direction (no duplicate rows sharing a case dir)

        rev = by_id["incident/revert-after-client-sync"]

        tip = rev["directions"][0]["oracle"]["theirs"]

        one = sources.symmetric("self", "o", {}, Git(rev["repo"], self.git_bin, self.home), tip, tip)

        self.assertEqual(len(one["directions"]), 1)

        gen = sources.generated(self.tmp / "gen", self.git_bin, self.home, 3, 7)

        gen2 = sources.generated(self.tmp / "gen2", self.git_bin, self.home, 3, 7)

        self.assertEqual([s["notes"] for s in gen], [s["notes"] for s in gen2])  # seeded

class VendoringTests(unittest.TestCase):

    def test_vendored_scripts_are_byte_identical_to_pinned_upstream(self):

        meta = json.loads((VENDOR / "UPSTREAM.json").read_text())

        self.assertEqual(meta["license"], "MIT OR Apache-2.0")

        for name, info in meta["files"].items():

            path = VENDOR / name

            self.assertTrue(path.is_file(), name)

            if "sha256" in info:

                self.assertEqual(hashlib.sha256(path.read_bytes()).hexdigest(), info["sha256"], name)

        notice = (VENDOR / "NOTICE").read_text()

        self.assertIn(meta["commit"], notice)

        self.assertIn("Permission is hereby granted", (VENDOR / "LICENSE-MIT").read_text())

        self.assertIn("Apache License", (VENDOR / "LICENSE-APACHE").read_text())

    def test_no_git_test_suite_text_is_vendored(self):

        merge_root = ROOT / "suites" / "merge"

        for path in merge_root.rglob("*"):

            if path.is_file() and path.suffix in (".py", ".sh", ".md", ".json"):

                text = path.read_text(errors="replace")

                self.assertNotIn("test_expect_success", text, path)

                self.assertNotIn(". ./test-lib.sh", text, path)

if __name__ == "__main__":

    unittest.main()