oak/benchmarks

from __future__ import annotations

import base64

import hashlib

import json

import shutil

import subprocess

import sys

import tempfile

import unittest

from pathlib import Path

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

GENERATOR = ROOT / "scripts" / "make_conflict_corpus.py"

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

from oakbench.fixture_registry import load_fixture_registry  # noqa: E402

# A fast, representative subset: one text size, the clean-merge/broken-build

# case, an unmergeable deletion, and the most common real agent conflict.

TEST_KINDS = ("text-overlap-small", "adjacent-lines", "delete-edit", "lockfile")

def run_generator(out: Path, seed: str) -> subprocess.CompletedProcess:

    return subprocess.run(

        [sys.executable, str(GENERATOR), "--out", str(out), "--seed", seed,

         "--quick", "--kinds", ",".join(TEST_KINDS)],

        capture_output=True, text=True,

    )

def git(repo: Path, *args: str, check: bool = True) -> subprocess.CompletedProcess:

    return subprocess.run(["git", "-C", str(repo), *args], check=check,

                          capture_output=True, text=True)

def resolution_bytes(resolution: dict) -> bytes:

    if "content" in resolution:

        return resolution["content"].encode()

    return base64.b64decode(resolution["content_b64"])

class MakeConflictCorpusTest(unittest.TestCase):

    """Generate once, then merge each kind in a scratch clone and check the oracle."""

    @classmethod

    def setUpClass(cls) -> None:

        if shutil.which("git") is None:

            raise unittest.SkipTest("git not available")

        cls.tmp = tempfile.TemporaryDirectory()

        cls.addClassCleanup(cls.tmp.cleanup)

        base = Path(cls.tmp.name)

        cls.corpus_a = base / "a"

        cls.corpus_b = base / "b"

        for out in (cls.corpus_a, cls.corpus_b):

            proc = run_generator(out, "test")

            if proc.returncode != 0:

                raise AssertionError(

                    f"generator failed for {out} (exit {proc.returncode}):\n"

                    f"{proc.stdout}\n{proc.stderr}")

        cls.merges = {kind: cls._merge_in_clone(kind) for kind in TEST_KINDS}

    @classmethod

    def _merge_in_clone(cls, kind: str) -> dict:

        clone = Path(cls.tmp.name) / "merges" / kind

        clone.parent.mkdir(parents=True, exist_ok=True)

        subprocess.run(["git", "clone", "--quiet", str(cls.corpus_a / kind), str(clone)],

                       check=True, capture_output=True)

        git(clone, "branch", "theirs", "origin/theirs")

        merge = git(clone, "-c", "user.name=Test", "-c", "[email protected]",

                    "merge", "--no-edit", "theirs", check=False)

        unmerged = git(clone, "diff", "--name-only", "--diff-filter=U")

        return {

            "clone": clone,

            "returncode": merge.returncode,

            "unmerged": sorted(p for p in unmerged.stdout.splitlines() if p),

        }

    def oracle(self, kind: str, corpus: Path | None = None) -> dict:

        corpus = corpus or self.corpus_a

        return json.loads((corpus / f"{kind}.RESOLUTION.json").read_text())

    def test_oracle_schema(self) -> None:

        for kind in TEST_KINDS:

            oracle = self.oracle(kind)

            self.assertEqual(oracle["schema_version"], 1)

            self.assertEqual(oracle["kind"], kind)

            self.assertEqual(oracle["seed"], "test")

            self.assertIn("expected_conflict", oracle)

            self.assertIn("conflicted_paths", oracle)

            self.assertIn("path", oracle["resolution"])

            self.assertIn("content_sha256", oracle["resolution"])

    def test_merge_outcome_matches_oracle(self) -> None:

        for kind in TEST_KINDS:

            oracle = self.oracle(kind)

            merge = self.merges[kind]

            conflicted = merge["returncode"] != 0

            self.assertEqual(conflicted, oracle["expected_conflict"],

                             f"{kind}: merge outcome disagrees with oracle")

            self.assertEqual(merge["unmerged"], oracle["conflicted_paths"],

                             f"{kind}: unmerged paths disagree with oracle")

    def test_adjacent_lines_clean_merge_breaks_build(self) -> None:

        oracle = self.oracle("adjacent-lines")

        merge = self.merges["adjacent-lines"]

        self.assertEqual(merge["returncode"], 0, "adjacent-lines must merge clean")

        check = oracle["post_merge_check"]

        self.assertIsNotNone(check)

        proc = subprocess.run(check["command"], cwd=merge["clone"],

                              capture_output=True, text=True)

        self.assertNotEqual(proc.returncode, 0, "build must break after the clean merge")

        self.assertEqual(proc.returncode, check["expected_returncode_clean_merge"])

    def test_adjacent_lines_check_passes_on_each_branch(self) -> None:

        oracle = self.oracle("adjacent-lines")

        repo = self.corpus_a / "adjacent-lines"

        for branch in ("theirs", "main", "ours"):

            git(repo, "checkout", "-q", branch)

            proc = subprocess.run(oracle["post_merge_check"]["command"], cwd=repo,

                                  capture_output=True, text=True)

            self.assertEqual(proc.returncode, 0, f"check must pass on branch {branch}")

    def test_text_overlap_resolution_applies(self) -> None:

        oracle = self.oracle("text-overlap-small")

        merge = self.merges["text-overlap-small"]

        self.assertNotEqual(merge["returncode"], 0)

        path = oracle["resolution"]["path"]

        self.assertIn(path, oracle["conflicted_paths"])

        content = resolution_bytes(oracle["resolution"])

        self.assertNotIn(b"<<<<<<<", content)

        # Apply the oracle: write the resolved content over the conflicted file.

        target = merge["clone"] / path

        target.write_bytes(content)

        self.assertEqual(hashlib.sha256(target.read_bytes()).hexdigest(),

                         oracle["resolution"]["content_sha256"])

        # The resolved file completes the merge cleanly.

        git(merge["clone"], "add", path)

        git(merge["clone"], "-c", "user.name=Test", "-c", "[email protected]",

            "commit", "-q", "-m", "resolve")

        status = git(merge["clone"], "status", "--porcelain")

        self.assertEqual(status.stdout.strip(), "")

    def test_resolution_sha_matches_content_all_kinds(self) -> None:

        for kind in TEST_KINDS:

            resolution = self.oracle(kind)["resolution"]

            self.assertEqual(hashlib.sha256(resolution_bytes(resolution)).hexdigest(),

                             resolution["content_sha256"], kind)

    def test_same_seed_yields_identical_oracles(self) -> None:

        for kind in TEST_KINDS:

            a = (self.corpus_a / f"{kind}.RESOLUTION.json").read_bytes()

            b = (self.corpus_b / f"{kind}.RESOLUTION.json").read_bytes()

            self.assertEqual(a, b, f"{kind}: oracle differs between identical-seed runs")

    def test_same_seed_yields_identical_history(self) -> None:

        for kind in TEST_KINDS:

            for ref in ("main", "ours", "theirs"):

                sha_a = git(self.corpus_a / kind, "rev-parse", ref).stdout

                sha_b = git(self.corpus_b / kind, "rev-parse", ref).stdout

                self.assertEqual(sha_a, sha_b, f"{kind}/{ref}: history differs")

class ConflictCorpusRegistryTest(unittest.TestCase):

    def test_registry_entry(self) -> None:

        registry = load_fixture_registry()

        self.assertIn("bench-conflict-corpus-v1", registry)

        spec = registry["bench-conflict-corpus-v1"]

        self.assertEqual(spec.generator, "scripts/make_conflict_corpus.py")

        self.assertEqual(spec.seed, "oakbench-conflicts-v1")

        self.assertEqual(spec.license, "synthetic")

        self.assertIsNone(spec.manifest)

        self.assertEqual(spec.hosting, ("local",))

        for kind in ("text-overlap-small", "text-overlap-medium", "text-overlap-large",

                     "adjacent-lines", "rename-rename", "rename-edit", "delete-edit",

                     "binary-binary", "lockfile"):

            self.assertIn(kind, spec.description)

if __name__ == "__main__":

    unittest.main()