Conflict

#!/usr/bin/env python3
"""conflict — refuse to commit a file that no longer parses.

A merge that goes wrong leaves `<<<<<<<`, `=======`, `>>>>>>>` markers wedged
into a file. Once committed, that file no longer parses — it isn't "a merge in
progress," it's broken source that landed. Worse, it hides: the failure surfaces
only when something tries to read the file, and if that read happens in a quiet
place, it can sit on your main branch for days while everything downstream of it
silently degrades.

`conflict` is the one-command gate that catches it. It scans a git tree for the
merge-marker triad and exits non-zero if any file carries an unresolved merge, so
you can wire it into a pre-commit hook or CI and a broken file simply cannot land.

WHY IT'S HONEST

- **The triad, never the lone middle.** `=======` on its own is a legal line — a
  Markdown horizontal rule, a Python separator comment, an ASCII banner. It is a
  conflict marker ONLY in the company of `<<<<<<<` and `>>>>>>>` in the same
  file. `conflict` fires on the *triad within one file*, never on a bare
  `=======`. A lint that cries wolf on a legal line gets disabled, and a disabled
  lint is worse than none.

- **Decidable from bytes.** A conflict marker on a branch needs no judgment, no
  review, no discussion — it is arithmetic. `conflict` reports the finding as an
  exit code, not a paragraph: exit codes don't decay, don't need finding, and
  cost nothing to re-run.

- **It is read-only.** It reads blobs out of git (or off disk) and compares. It
  writes nothing and fixes nothing.

THE HONEST EDGE

This is a check, not an immunity — it protects you only when it is RUN. Wire it
into a hook or CI so "run it" isn't something a human has to remember. And it
detects the standard git marker triad; a tool that uses different markers needs a
different pattern. Visibility, not immunity.

USAGE
    python3 conflict.py                 # scan the working tree (tracked files)
    python3 conflict.py --ref main      # scan any committed tree by ref
    python3 conflict.py --ignore _snapshots/ --ignore vendor/
                                        # skip path substrings (repeatable)

EXIT  0 = clean · 3 = unresolved merge found (the alarm) · 2 = error/usage

MIT licensed. Python standard library only. Read-only. Deterministic.
"""
from __future__ import annotations

import subprocess
import sys

# The three markers git leaves on a conflicted merge. `=======` is only a marker
# in the company of the other two — see the triad rule in main().
MARKERS = ("<<<<<<< ", "=======", ">>>>>>> ")


def git_files(ref: str | None) -> list[str]:
    """List tracked files at `ref` (committed tree) or in the working tree."""
    if ref:
        out = subprocess.run(
            ["git", "ls-tree", "-r", "--name-only", ref],
            capture_output=True, text=True,
        ).stdout
    else:
        out = subprocess.run(["git", "ls-files"], capture_output=True, text=True).stdout
    return out.split()


def read_blob(f: str, ref: str | None) -> bytes:
    """Read a file's bytes at `ref` or from the working tree."""
    if ref:
        return subprocess.run(["git", "show", f"{ref}:{f}"], capture_output=True).stdout
    with open(f, "rb") as fh:
        return fh.read()


def marker_lines(text: str) -> dict[str, list[int]]:
    """Map each marker to the 1-based line numbers where a line STARTS with it."""
    lines = text.split("\n")
    return {m: [i + 1 for i, ln in enumerate(lines) if ln.startswith(m)] for m in MARKERS}


def has_conflict(text: str) -> bool:
    """True iff `text` contains the full marker TRIAD (not a lone `=======`)."""
    found = marker_lines(text)
    return all(found[m] for m in MARKERS)


def is_ignored(path: str, ignore: tuple[str, ...]) -> bool:
    """True iff `path` contains any ignore substring (skip it)."""
    return any(ig in path for ig in ignore)


def scan(ref: str | None = None, ignore: tuple[str, ...] = ()) -> list[tuple[str, dict]]:
    """Return [(path, marker_lines)] for every tracked file carrying the triad."""
    hits: list[tuple[str, dict]] = []
    for f in git_files(ref):
        if is_ignored(f, ignore):
            continue
        try:
            raw = read_blob(f, ref)
        except (OSError, FileNotFoundError):
            continue
        try:
            text = raw.decode("utf-8")
        except UnicodeDecodeError:
            continue  # a binary blob has no markers to read
        if has_conflict(text):
            hits.append((f, marker_lines(text)))
    return hits


def _parse_args(argv: list[str]) -> tuple[str | None, tuple[str, ...]]:
    ref: str | None = None
    ignore: list[str] = []
    i = 0
    while i < len(argv):
        a = argv[i]
        if a == "--ref":
            if i + 1 >= len(argv):
                raise ValueError("--ref needs a git ref")
            ref = argv[i + 1]
            i += 2
        elif a == "--ignore":
            if i + 1 >= len(argv):
                raise ValueError("--ignore needs a path substring")
            ignore.append(argv[i + 1])
            i += 2
        elif a in ("-h", "--help"):
            raise ValueError("__help__")
        else:
            raise ValueError(f"unknown argument: {a}")
    return ref, tuple(ignore)


def main(argv: list[str]) -> int:
    try:
        ref, ignore = _parse_args(argv)
    except ValueError as exc:
        if str(exc) == "__help__":
            sys.stdout.write(__doc__)
            return 2
        sys.stderr.write(f"conflict: {exc}\n")
        return 2

    hits = scan(ref, ignore)
    where = f"`{ref}`" if ref else "the working tree"

    if not hits:
        sys.stdout.write(f"conflict: clean — no unresolved merge in {where}.\n")
        return 0

    sys.stderr.write(
        f"conflict: {len(hits)} file(s) in {where} carry an UNRESOLVED MERGE.\n"
    )
    for f, found in hits:
        ln = found["<<<<<<< "][0]
        sys.stderr.write(
            f"  {f}:{ln}  <<<<<<< / ======= / >>>>>>>  "
            f"({len(found['<<<<<<< '])} conflict block(s))\n"
        )
    sys.stderr.write(
        "\nA conflict marker on a branch is not a merge in progress — it is a "
        "FILE THAT NO LONGER PARSES, landed. Resolve it. Do not commit over it.\n"
    )
    return 3


if __name__ == "__main__":
    raise SystemExit(main(sys.argv[1:]))

#!/usr/bin/env python3
"""test_conflict.py — the certifying properties of the conflict gift.

Run:  python3 test_conflict.py    (exit 0 = all pass, 1 = a failure)

The load-bearing property is the TRIAD RULE: a lone `=======` (a legal line) must
NOT fire, while the full marker triad must. Each test is MUTATION-BITTEN — it is
here because a plausible mutation of conflict.py makes it fail loud. The core is
pure (marker_lines / has_conflict / is_ignored), so it's tested directly; the
git-facing shell is exercised in a throwaway repo.
"""
import os
import subprocess
import sys

import conflict as c

_FAILURES: list[str] = []
_PASSES = 0


def check(cond: bool, msg: str) -> None:
    global _PASSES
    if cond:
        _PASSES += 1
    else:
        _FAILURES.append(msg)


CONFLICTED = """\
line before
<<<<<<< HEAD
ours
=======
theirs
>>>>>>> origin/main
line after
"""

# a lone `=======` in legal, innocent company — MUST NOT fire
INNOCENT_RULE = """\
Section One
=======
Section Two

Some prose under a Markdown setext rule.
"""

INNOCENT_BANNER = """\
# ============================================================
# a decorative comment banner, not a conflict
# ============================================================
def f(): pass
"""


# --- 1. the triad fires; a lone middle does not (the core property) ----------
def test_triad_rule():
    check(c.has_conflict(CONFLICTED), "the full marker triad must be detected")
    check(not c.has_conflict(INNOCENT_RULE), "a lone ======= (Markdown rule) must NOT fire")
    check(not c.has_conflict(INNOCENT_BANNER), "a ==== banner must NOT fire")
    # a triad that is missing one leg must NOT fire
    missing_close = CONFLICTED.replace(">>>>>>> origin/main\n", "")
    check(not c.has_conflict(missing_close), "two-of-three markers must NOT fire (needs the triad)")


# --- 2. golden: marker_lines reports the exact line numbers ------------------
def test_marker_lines_golden():
    found = c.marker_lines(CONFLICTED)
    check(found["<<<<<<< "] == [2], f"open marker line wrong: {found['<<<<<<< ']}")
    check(found["======="] == [4], f"middle marker line wrong: {found['=======']}")
    check(found[">>>>>>> "] == [6], f"close marker line wrong: {found['>>>>>>> ']}")
    # the three exact line-number asserts above ARE the golden: they pin the
    # detected positions, not merely the count.


# --- 3. only a line that STARTS with a marker counts ------------------------
def test_startswith_only():
    # a single marker mid-line (e.g. inside a string) is not a line-start conflict
    embedded = 'x = "<<<<<<< not a real marker"\n'
    check(not c.has_conflict(embedded), "a marker mid-line must not count as a conflict")
    # the discriminating case: ALL THREE markers mid-line. Under startswith this
    # is clean (no line STARTS with a marker); a naive `m in ln` would flag it.
    triad_midline = "a <<<<<<< b\nc ======= d\ne >>>>>>> f\n"
    check(
        not c.has_conflict(triad_midline),
        "three markers mid-line must NOT fire (startswith, not substring)",
    )


# --- 4. is_ignored honors substrings ----------------------------------------
def test_is_ignored():
    check(c.is_ignored("a/_snapshots/x.txt", ("_snapshots/",)), "ignore substring should match")
    check(not c.is_ignored("a/src/x.txt", ("_snapshots/",)), "non-matching path must not be ignored")
    check(not c.is_ignored("anything", ()), "empty ignore never ignores")


# --- 5. arg parse: --ref, --ignore (repeatable), unknown ---------------------
def test_parse_args():
    ref, ig = c._parse_args(["--ref", "main", "--ignore", "a/", "--ignore", "b/"])
    check(ref == "main", f"--ref not parsed: {ref}")
    check(ig == ("a/", "b/"), f"--ignore not repeatable: {ig}")
    raised = False
    try:
        c._parse_args(["--bogus"])
    except ValueError:
        raised = True
    check(raised, "unknown arg must raise")


# --- 6. END-TO-END against a throwaway repo ---------------------------------
def _run_git(cwd, *args):
    subprocess.run(["git", *args], cwd=cwd, capture_output=True, text=True, check=True)


def test_end_to_end():
    import tempfile
    with tempfile.TemporaryDirectory() as repo:
        _run_git(repo, "init", "-q")
        _run_git(repo, "config", "user.email", "t@t")
        _run_git(repo, "config", "user.name", "t")
        # one clean file, one conflicted file
        with open(os.path.join(repo, "clean.py"), "w") as fh:
            fh.write("def ok(): pass\n")
        with open(os.path.join(repo, "broken.py"), "w") as fh:
            fh.write(CONFLICTED)
        _run_git(repo, "add", "-A")
        _run_git(repo, "commit", "-qm", "with a conflict")

        cwd0 = os.getcwd()
        os.chdir(repo)
        try:
            hits = c.scan(ref=None)
        finally:
            os.chdir(cwd0)
        paths = sorted(f for f, _ in hits)
        check(paths == ["broken.py"], f"end-to-end should flag only broken.py: {paths}")


# --- 7. ignore actually excludes a conflicted file --------------------------
def test_end_to_end_ignore():
    import tempfile
    with tempfile.TemporaryDirectory() as repo:
        _run_git(repo, "init", "-q")
        _run_git(repo, "config", "user.email", "t@t")
        _run_git(repo, "config", "user.name", "t")
        os.makedirs(os.path.join(repo, "_snapshots"))
        with open(os.path.join(repo, "_snapshots", "past.txt"), "w") as fh:
            fh.write(CONFLICTED)  # a deliberate record of a past conflict
        _run_git(repo, "add", "-A")
        _run_git(repo, "commit", "-qm", "snapshot")
        cwd0 = os.getcwd()
        os.chdir(repo)
        try:
            with_ignore = c.scan(ref=None, ignore=("_snapshots/",))
            without = c.scan(ref=None)
        finally:
            os.chdir(cwd0)
        check(with_ignore == [], f"ignored path should not be flagged: {with_ignore}")
        check(len(without) == 1, f"without ignore the file should flag: {without}")


def run() -> int:
    tests = [v for k, v in sorted(globals().items()) if k.startswith("test_") and callable(v)]
    for t in tests:
        try:
            t()
        except Exception as exc:
            _FAILURES.append(f"{t.__name__} raised: {exc!r}")
    total = _PASSES + len(_FAILURES)
    if _FAILURES:
        print(f"FAIL — {len(_FAILURES)} of {total} checks failed:")
        for f in _FAILURES:
            print(f"  ✗ {f}")
        return 1
    print(f"OK — {_PASSES}/{total} checks passed ({len(tests)} tests).")
    return 0


if __name__ == "__main__":
    sys.exit(run())