"""Conformance matrix oracles: every behaviour passes a correct VCS and fails injected faults.

``CopyVCS`` is a directory-copy "VCS" with no bugs; each fault flag injects

one class of defect at one point of the roundtrip (commit, source worktree,

clone, pull, status). Every behaviour must pass the faultless model and fail

each applicable fault at the named leg, so no behaviour has a vacuous oracle.

"""

import os

from pathlib import Path

import re

import shutil

import sys

import tempfile

import unittest

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

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

from oakbench import conformance_gate  # noqa: E402

from oakbench import conformance_matrix as matrix  # noqa: E402

from oakbench.conformance import SCENARIOS  # noqa: E402

GIT_TEST_IDS = {"t0010", "t0020", "t0021", "t0027", "t0050", "t2003", "t2007", "t2021", "t3700", "t3902", "t3910",

                "t4011", "t4012", "t4102", "t4114", "t4129", "t7001"}

def copy_tree(source: Path, destination: Path, skip=(".fake",)):

    destination.mkdir(parents=True, exist_ok=True)

    for entry in sorted(os.listdir(source)):

        if entry in skip:

            continue

        src, dst = source / entry, destination / entry

        if src.is_symlink():

            os.symlink(os.readlink(src), dst)

        elif src.is_dir():

            copy_tree(src, dst, skip=())

        else:

            shutil.copy2(src, dst)

def clear(directory: Path):

    for entry in os.listdir(directory):

        path = directory / entry

        if entry == ".fake":

            continue

        if path.is_symlink() or path.is_file():

            path.unlink()

        else:

            shutil.rmtree(path)

class CopyVCS:

    """A faultless copy-based model of commit/push/clone/pull, with injectable faults."""

    kind = "fake"

    def __init__(self, base: Path, faults=()):

        self.base, self.faults = base, set(faults)

        self.snapshots = {}

        self.counter = 0

    def _snapshot(self, source: Path) -> Path:

        self.counter += 1

        target = self.base / f"snap{self.counter}"

        copy_tree(source, target)

        return target

    def init(self, repo):

        (repo / ".fake").mkdir()

    def commit(self, repo, message):

        snap = self._snapshot(repo)

        if "commit_inject" in self.faults:

            (snap / "zz-injected-by-commit").write_bytes(b"x")

        if "mutate_source" in self.faults:

            (repo / "zz-written-by-commit").write_bytes(b"x")

        self.snapshots[repo] = snap

    def push(self, repo):

        self.remote = self.snapshots[repo]

    def clone(self, repo, destination):

        copy_tree(self.remote, destination)

        (destination / ".fake").mkdir()

        self.snapshots[destination] = self.remote

        if "clone_inject" in self.faults:

            (destination / "zz-injected-by-clone").write_bytes(b"x")

    def pull(self, repo):

        if "pull_stale" in self.faults:

            (repo / "zz-stale-after-pull").write_bytes(b"x")

            return

        clear(repo)

        copy_tree(self.remote, repo)

        self.snapshots[repo] = self.remote

    def status_clean(self, repo):

        if "dirty" in self.faults:

            raise matrix.LegFailure("status reports changes")

    def committed(self, repo):

        return matrix.fs_manifest(self.snapshots[repo]), {"head": str(self.snapshots[repo])}

class BehaviourOracleTests(unittest.TestCase):

    @classmethod

    def setUpClass(cls):

        cls.caps = matrix.probe_capabilities(Path(tempfile.gettempdir()) / "oakbench-caps-probe")

    def runnable(self):

        for behaviour in matrix.BEHAVIOURS:

            if behaviour.gated or any(not self.caps.get(cap) for cap in behaviour.requires):

                continue

            yield behaviour

    def run_fake(self, behaviour, faults=()):

        with tempfile.TemporaryDirectory() as tmp:

            base = Path(tmp)

            (base / "work").mkdir()

            vcs = CopyVCS(base, faults)

            legs, committed = matrix.run_behaviour(vcs, behaviour, base / "work", {})

            return legs

    def test_every_behaviour_passes_a_correct_vcs(self):

        ran = 0

        for behaviour in self.runnable():

            with self.subTest(behaviour=behaviour.id):

                legs = self.run_fake(behaviour)

                self.assertTrue(legs)

                self.assertEqual({name: leg for name, leg in legs.items() if leg["status"] != "pass"}, {})

                if len(behaviour.steps) > 1:

                    self.assertEqual(legs["pull_fs"]["status"], "pass")

                ran += 1

        self.assertGreater(ran, 40)

    def test_every_behaviour_detects_each_fault_at_its_leg(self):

        for behaviour in self.runnable():

            if behaviour.id in ("large_file_chunked_v1", "many_small_files_v1"):

                continue  # same oracle path as the small behaviours; their copies are slow (covered by the pass test)

            multi = len(behaviour.steps) > 1

            cases = {"commit_inject": "commit_read@1" if multi else "commit_read",

                     "mutate_source": "source_preserved@1" if multi else "source_preserved", "clone_inject": "clone_fs",

                     "dirty": "clone_status_clean"}

            if multi:

                cases["pull_stale"] = "pull_fs"

            for fault, leg in cases.items():

                with self.subTest(behaviour=behaviour.id, fault=fault):

                    legs = self.run_fake(behaviour, {fault})

                    self.assertEqual(legs[leg]["status"], "fail", legs)

    def test_clone_fault_is_not_mistaken_for_committed_state(self):

        legs = self.run_fake(matrix.BY_ID["eol_crlf_v1"], {"clone_inject"})

        self.assertEqual(legs["clone_fs"]["status"], "fail")

        self.assertEqual(legs["clone_read"]["status"], "pass")

        self.assertEqual(legs["commit_read"]["status"], "pass")

class ManifestOracleTests(unittest.TestCase):

    def test_each_loss_class_is_detected(self):

        with tempfile.TemporaryDirectory() as tmp:

            root = Path(tmp)

            (root / "f").write_bytes(b"payload")

            (root / "l").symlink_to("f")

            expected = matrix.fs_manifest(root)

            mutations = [

                lambda: (root / "f").write_bytes(b"payloae"),

                lambda: (root / "f").chmod(0o755),

                lambda: ((root / "l").unlink(), (root / "l").symlink_to("g")),

                lambda: ((root / "l").unlink(), (root / "l").write_bytes(b"f")),

                lambda: (root / "extra").write_bytes(b""),

                lambda: (root / "f").rename(root / "F2"),

            ]

            for mutate in mutations:

                mutate()

                self.assertIsNotNone(matrix.diff_manifests(expected, matrix.fs_manifest(root)))

                shutil.rmtree(root)

                root.mkdir()

                (root / "f").write_bytes(b"payload")

                (root / "f").chmod(0o644)

                (root / "l").symlink_to("f")

                self.assertIsNone(matrix.diff_manifests(expected, matrix.fs_manifest(root)))

    def test_empty_directories_are_observations_not_content(self):

        with tempfile.TemporaryDirectory() as tmp:

            root = Path(tmp)

            (root / "void" / "deeper").mkdir(parents=True)

            dirs = []

            self.assertEqual(matrix.fs_manifest(root, dirs), {})

            self.assertEqual(dirs, ["void", "void/deeper"])

    def test_inplace_edit_keeps_size_and_mtime(self):

        with tempfile.TemporaryDirectory() as tmp:

            root = Path(tmp)

            matrix.apply_op(root, ("write", "r", b"aaaa", False))

            before = (root / "r").stat()

            matrix.apply_op(root, ("inplace", "r", b"bbbb"))

            after = (root / "r").stat()

            self.assertEqual((before.st_size, before.st_mtime_ns), (after.st_size, after.st_mtime_ns))

            self.assertEqual((root / "r").read_bytes(), b"bbbb")

    def test_sparse_op_writes_declared_size_and_tail(self):

        with tempfile.TemporaryDirectory() as tmp:

            root = Path(tmp)

            matrix.apply_op(root, ("sparse", "s", 1 << 20, b"tail\n"))

            data = (root / "s").read_bytes()

            self.assertEqual(len(data), 1 << 20)

            self.assertTrue(data.endswith(b"tail\n"))

class LoopbackServerTests(unittest.TestCase):

    @unittest.skipUnless(os.name == "posix", "process groups required")

    def test_stop_reaches_grandchild_behind_a_wrapper_script(self):

        with tempfile.TemporaryDirectory() as tmp:

            root = Path(tmp)

            pidfile = root / "grandchild.pid"

            listener = root / "listen.py"

            listener.write_text(

                "import socket, sys, time, os\n"

                "port = int(sys.argv[sys.argv.index('--port') + 1])\n"

                f"open({str(pidfile)!r}, 'w').write(str(os.getpid()))\n"

                "s = socket.socket(); s.bind(('127.0.0.1', port)); s.listen()\n"

                "time.sleep(300)\n")

            wrapper = root / "fake-oak"

            # No exec: the listener is a grandchild of the process Popen started.

            wrapper.write_text(f"#!/bin/sh\n{sys.executable} {listener} \"$@\" &\nwait\n")

            wrapper.chmod(0o755)

            with matrix.LoopbackOakServer(str(wrapper), root / "data", dict(os.environ)):

                grandchild = int(pidfile.read_text())

                os.kill(grandchild, 0)

            for _ in range(50):

                try:

                    os.kill(grandchild, 0)

                except ProcessLookupError:

                    break

                import time

                time.sleep(0.1)

            else:

                self.fail("grandchild server survived LoopbackOakServer exit")

class CatalogTests(unittest.TestCase):

    def test_ids_are_versioned_unique_and_disjoint_from_legacy(self):

        ids = [b.id for b in matrix.BEHAVIOURS]

        self.assertEqual(len(ids), len(set(ids)))

        self.assertTrue(all(re.fullmatch(r"[a-z0-9_]+_v\d+", i) for i in ids))

        self.assertFalse(set(ids) & set(SCENARIOS))

        self.assertGreaterEqual(len(ids), 50)

    def test_required_coverage_groups(self):

        groups = {b.group for b in matrix.BEHAVIOURS}

        self.assertEqual(groups, {"paths", "modes", "symlinks", "file_dir", "empty", "content", "history"})

        for needed in ("path_tab_v1", "path_unicode_nfd_v1", "path_case_only_rename_v1", "mode_exec_remove_v1",

                       "symlink_absolute_v1", "file_to_symlink_v1", "dir_to_file_v1", "empty_dir_semantics_v1",

                       "huge_file_sparse_v1", "binary_detection_v1", "eol_mixed_v1", "delete_readd_v1",

                       "many_small_files_v1"):

            self.assertIn(needed, matrix.BY_ID)

        self.assertTrue(matrix.BY_ID["huge_file_sparse_v1"].gated)

    def test_prior_art_cites_known_git_test_ids_only(self):

        for behaviour in matrix.BEHAVIOURS:

            for citation in behaviour.prior_art:

                match = re.match(r"git (t\d{4})-", citation)

                if match:

                    self.assertIn(match.group(1), GIT_TEST_IDS, citation)

    def test_expected_failures_name_real_behaviours_legs_and_tickets(self):

        expected = conformance_gate.load_expected()

        known = set(matrix.BY_ID) | set(SCENARIOS)

        legs = set(matrix.LEGS) | {"commit_push", "behaviour"}

        for behaviour, entry in expected["entries"].items():

            self.assertIn(behaviour, known)

            for leg in entry["failed_legs"]:

                self.assertIn(leg.split("@")[0], legs)

        tickets = {b: e["ticket"] for b, e in expected["entries"].items()}

        self.assertEqual(tickets["path_tab_v1"], "fb-531")

        self.assertEqual(tickets["path_roundtrip_v1"], "fb-531")

        self.assertEqual(tickets["symlink_to_dir_real_v1"], "fb-542")

        self.assertEqual(expected["entries"]["path_tab_v1"]["outcome"], "refused")

        for skip in expected["coverage"]["darwin-ci-ni"]["expected_skips"]:

            self.assertIn(skip, matrix.BY_ID)

@unittest.skipUnless(shutil.which("git"), "Git required")

class RealGitMatrixTests(unittest.TestCase):

    def test_git_passes_a_representative_subset(self):

        subset = ["path_tab_v1", "mode_exec_add_v1", "file_to_dir_v1", "eol_crlf_v1", "delete_all_v1"]

        with tempfile.TemporaryDirectory() as tmp:

            rows = matrix.run_matrix(Path(tmp), {"git": "git"}, subset)

        for row in rows:

            with self.subTest(behaviour=row["scenario"]):

                if row["returncode"] == 77:

                    continue

                self.assertEqual(row["returncode"], 0, row.get("failure"))

                multi = len(matrix.BY_ID[row["scenario"]].steps) > 1

                self.assertEqual(row["legs"]["commit_read@1" if multi else "commit_read"]["status"], "pass")

@unittest.skipUnless(shutil.which("oak") and shutil.which("git"), "Oak and Git required")

class RealOakMatrixTests(unittest.TestCase):

    def test_oak_rows_carry_independent_reader_and_git_differential(self):

        with tempfile.TemporaryDirectory() as tmp:

            rows = matrix.run_matrix(Path(tmp), {"git": "git", "oak": "oak"}, ["eol_mixed_v1", "path_tab_v1"])

        oak = {row["scenario"]: row for row in rows if row["subject"] == "oak"}

        mixed = oak["eol_mixed_v1"]

        self.assertEqual(mixed["returncode"], 0, mixed.get("failure"))

        self.assertEqual(mixed["observations"]["committed_reader"]["reader"], "independent_oak_sqlite_reader")

        self.assertEqual(mixed["git_differential"], "identical")

        # fb-531: rejected at commit; if Oak fixes it this flips and the gate reports FIXED.

        self.assertIn(oak["path_tab_v1"]["returncode"], (0, 1))

        if oak["path_tab_v1"]["returncode"] == 1:

            self.assertEqual(oak["path_tab_v1"]["failed_legs"], ["commit_push"])

if __name__ == "__main__":

    unittest.main()