diff --git a/CHANGELOG.md b/CHANGELOG.md
index 1f4de8d..0b6ec69 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -16,6 +16,17 @@ project intends to follow [Semantic Versioning](https://semver.org/) once it rea
row carries `tree_source`; the receipt and the round summary carry the totals. Unset,
nothing changes. See `docs/TREE_PROVIDER.md`.
+### Fixed
+
+- `[repair] union_merge` paths are now merged by git's own union merge, byte for byte what
+ `git merge` gives with `merge=union` declared for them in `.gitattributes`: inside each
+ conflicting hunk the pull request's lines and then the base branch's, in place, with
+ repeated lines kept, a line one side deleted left deleted, and line endings and bytes
+ untouched. Previously the whole base-side file came first, the pull request's new lines
+ were appended at the end, repeated lines were collapsed and CRLF line endings were lost,
+ so a union-merged step did not match a tree provider or a `git merge` computing git's
+ union.
+
## [0.2.0] - 2026-09-25
### Changed
diff --git a/README.md b/README.md
index 6383128..adf59c8 100644
--- a/README.md
+++ b/README.md
@@ -241,7 +241,7 @@ Unknown keys are refused.
| `train.interval_seconds` | `300` | sleep between rounds in `watch` |
| `labels.hold` | `"train:hold"` | label that keeps a pull request out |
| `repair.enabled` | `true` | push mechanical fixes (union merges, stale base) to pull-request branches |
-| `repair.union_merge` | `[]` | globs of files merged by line union instead of conflicting |
+| `repair.union_merge` | `[]` | globs of files merged by git's union merge (`merge=union`) instead of conflicting |
| `history.order` | `"off"` | `"tests-first"` refuses code committed before its tests |
| `history.reland` | `true` | re-land refused histories in order instead of holding them |
| `history.tests`, `history.docs` | common globs | how paths are classified for the history check |
diff --git a/proofs/BraidedTrain/Union.lean b/proofs/BraidedTrain/Union.lean
index 47200ca..895ea11 100644
--- a/proofs/BraidedTrain/Union.lean
+++ b/proofs/BraidedTrain/Union.lean
@@ -3,11 +3,13 @@ import BraidedTrain.Braid
/-!
# BraidedTrain.Union — the line-level union merge as a model step
-`svrf.rules.union_lines(theirs, ours)` keeps the base side's lines in order and appends each
-line of the pull request's side that is not already there, once. `unionLines` is that
-function on lists of lines. `git.RealGit.union_step` applies it only on configured union
-paths, and only when every path git's merge reports as conflicted is a union path; any
-other conflicted path keeps the two heads apart. Finite lists, no Mathlib.
+`unionLines theirs ours` keeps the base side's lines in order and appends each line of the
+pull request's side that is not already there, once: the union of two line lists, as a
+model. `git.RealGit.union_step` applies a union only on configured union paths, and only
+when every path git's merge reports as conflicted is a union path; any other conflicted
+path keeps the two heads apart. The code's union is git's own (`merge=union`), declared
+for those paths through `svrf.rules.union_attributes`; see the modelling boundary below.
+Finite lists, no Mathlib.
1. `count_unionLines`: the count of a line after a union is its count on the base side,
plus one exactly when the base side lacks it and the pull request's side has it. Every
@@ -30,6 +32,12 @@ other conflicted path keeps the two heads apart. Finite lists, no Mathlib.
Modelling boundary: git merges a non-union file edited on both sides hunk by hunk; this
model treats such a path as a conflict (the coarser, path-level reading used by
`apply_comm_of_disjoint`), and it treats every edit of a union path as a union of lines.
+git's union merge is also hunk by hunk: inside each conflicting hunk it keeps the pull
+request's lines and then the base side's, outside them it is git's ordinary merge, and it
+neither drops repeated lines nor restores lines one side deleted. `unionLines` is the
+line-set reading of that merge, not its exact byte order; the train never relies on that
+order, because it fixes the fold order once, replays it at landing, and compares the
+landed tree with the gated tree byte for byte.
-/
set_option linter.unusedSimpArgs false
@@ -43,7 +51,7 @@ variable {Line : Type} [DecidableEq Line]
def addLine (acc : List Line) (l : Line) : List Line :=
if l ∈ acc then acc else acc ++ [l]
-/-- `union_lines(theirs, ours)`: the base side's lines, then each new line of ours, once. -/
+/-- The line-set union: the base side's lines, then each new line of ours, once. -/
def unionLines (theirs ours : List Line) : List Line :=
ours.foldl addLine theirs
diff --git a/proofs/README.md b/proofs/README.md
index a6ee436..fc6d9b1 100644
--- a/proofs/README.md
+++ b/proofs/README.md
@@ -64,7 +64,7 @@ states it and the function or method that enforces it in the running train.
| `interleaved-owners` | `interleaved_landing` | `choose_families` |
| `repair-mechanical` | `retry_iff` | `repair_class` |
| `ordered-reland` | `reland_tree` | `reland_class` |
-| `union-merge` | `change_comm` | `union_lines` |
+| `union-merge` | `change_comm` | `union_attributes` |
| `speculative-stacking` | `stack_lands_gated` | `Train.round` |
| `speculation-void` | `stackStatus_void_iff` | `Train.round` |
| `bisect-holds-exactly-red` | `settle_outcome` | `Train.settle_red` |
@@ -88,7 +88,7 @@ or rename a rule, update the table above and the code together; the test fails o
| `BraidedTrain/Braid.lean` | The ungated-window invariant (`AllGated`, `mains`); retry as a function of a read key; commutation of disjoint writes (`strands_comm`). |
| `BraidedTrain/Interleaving.lean` | Two owners landing path-disjoint families interleaved still land one tree (`interleaved_landing`); when a gate's own read paths miss the other owner's writes, no joint gate is needed (`two_owners_end_gated`). |
| `BraidedTrain/Reland.lean` | Re-landing a history as tests-then-code-then-docs commits lands the identical tree (`reland_tree`), so any tree-reading gate's verdict is unchanged (`reland_gate`). |
-| `BraidedTrain/Union.lean` | The line-level union merge (`unionLines`, the model of `union_lines`): exact associativity (`unionLines_assoc`), commutation up to line order (`unionLines_comm`, sharp by `union_order_visible`), and at the path level `change_comm` / `family_perm` under `UnionOnlyOverlap`, the union repair's hypothesis. |
+| `BraidedTrain/Union.lean` | The line-level union merge (`unionLines`, the line-set model of git's `merge=union`, which `union_attributes` declares for the union paths): exact associativity (`unionLines_assoc`), commutation up to line order (`unionLines_comm`, sharp by `union_order_visible`), and at the path level `change_comm` / `family_perm` under `UnionOnlyOverlap`, the union repair's hypothesis. |
| `BraidedTrain/Stacking.lean` | Speculative stacking: families gated on the fold of every family before them land gated trees at every family boundary (`stack_lands_gated`); a stacked verdict does not carry past a red family (`stack_fold_through`, `stacked_verdict_does_not_carry`); the round's bookkeeping voids exactly the families above the first red one (`stackStatus_void_iff`, `stackStatus_landed_iff`, `stackStatus_bisected_iff`). |
| `BraidedTrain/Bisection.lean` | Bisection of a red family (`settle`, the model of `Train.settle_red`) terminates (well-founded on family length), holds exactly the bad pull requests and lands the rest under a monotone gate (`settle_outcome`), and costs at most `2·r·⌈log₂ n⌉ + 1` gates including the family's own (`settle_gates_le`, `bisection_gates_le`). |
| `BraidedTrain/Families.lean` | The Bron–Kerbosch recursion of `families`, with arbitrary pivot and iteration order: every family it reports is a maximal compatible set (`bk_maximal`), so the family `choose_families` keeps holds no conflicting pair and every pull request left out conflicts with a kept one (`chosen_family_maximal`). With a pivot drawn from the candidates or excluded nodes (the code's rule, `codePivot_mem`) and an order visiting every node, it reports every maximal compatible set (`bk_complete`), so the first family of the size-sorted list is a maximum compatible set (`chosen_family_maximum`). |
diff --git a/src/svrf/git.py b/src/svrf/git.py
index dc0902b..36d219a 100644
--- a/src/svrf/git.py
+++ b/src/svrf/git.py
@@ -18,7 +18,7 @@
from .errors import ReadFailed
from .redact import redact
-from .rules import union_lines
+from .rules import union_attributes
@dataclass
@@ -63,6 +63,10 @@ def _run(self, *args, input: str | None = None, env: dict | None = None) -> subp
return subprocess.run(["git", "-C", str(self.root), *args], capture_output=True, text=True, input=input,
env=env or self.env)
+ def _raw(self, *args, input: bytes | None = None) -> subprocess.CompletedProcess:
+ """`_run` with bytes in and out: file contents and paths exactly as git holds them."""
+ return subprocess.run(["git", "-C", str(self.root), *args], capture_output=True, input=input, env=self.env)
+
def _out(self, *args, input: str | None = None, env: dict | None = None) -> str:
done = self._run(*args, input=input, env=env)
if done.returncode != 0:
@@ -152,9 +156,11 @@ def pairs(self, rows: list[dict], base: str) -> dict:
def union_step(self, acc: str, head: str, message: str, union: Callable[[str], bool] | None = None) -> Step:
"""Merge `acc` (the base branch, or the fold so far) into the pull request `head`:
- git's merge with the pull request as ours and the base as theirs, configured
- union-merge paths resolved by `union_lines`, committed with parents (head, acc)
- exactly as `git merge ` on the branch would."""
+ git's merge with the pull request as ours and the base as theirs, committed with
+ parents (head, acc) exactly as `git merge ` on the branch would. When every
+ path that merge leaves conflicted is a configured union-merge path, the merge is
+ made again by git with `merge=union` declared for those paths, so each is git's own
+ union merge, byte for byte what `.gitattributes` declaring it would give."""
union = union or self.union
try:
if self.is_ancestor(acc, head):
@@ -163,34 +169,61 @@ def union_step(self, acc: str, head: str, message: str, union: Callable[[str], b
return Step("CLEAN", commit=acc, tree=self.tree(acc))
# Same attribute-source fix as `merge_preview`: read .gitattributes from `acc`
# (the base/theirs side), never from this clone's checkout.
- done = self._run("-c", f"attr.tree={acc}", "merge-tree", "--write-tree", head, acc)
- if done.returncode not in (0, 1):
- return Step("FAILED", reason=f"MERGE_TREE_FAILED:{done.returncode}")
- lines = done.stdout.split("\n")
- tree = lines[0].strip()
- if done.returncode == 1:
- stages: dict[str, dict[int, tuple[str, str]]] = {}
- for line in lines[1:]:
- if not line.strip():
- break
- meta, _, path = line.partition("\t")
- mode, sha, stage = meta.split()
- stages.setdefault(path, {})[int(stage)] = (mode, sha)
+ code, tree, stages = self._merge_tree(acc, head, acc)
+ if code == 1:
other = sorted(p for p in stages if not union(p))
if other:
return Step("CONFLICT", conflicts=other)
- for path, by_stage in stages.items():
+ for path, by_stage in sorted(stages.items()):
if 2 not in by_stage or 3 not in by_stage:
return Step("CONFLICT", conflicts=[path])
- ours = self._out("cat-file", "blob", by_stage[2][1])
- theirs = self._out("cat-file", "blob", by_stage[3][1])
- blob = self._out("hash-object", "-w", "--stdin", input=union_lines(theirs=theirs, ours=ours))
- tree = self._overlay_tree(tree, {path: (by_stage[2][0], blob)})
+ code, tree, stages = self._merge_tree(self._union_attribute_tree(acc, sorted(stages)), head, acc)
+ if code == 1:
+ return Step("CONFLICT", conflicts=sorted(stages))
commit = self._out("commit-tree", tree, "-p", head, "-p", acc, "-m", message)
return Step("CLEAN", commit=commit, tree=tree)
except ReadFailed as failure:
return Step("FAILED", reason=failure.reason)
+ def _merge_tree(self, attributes: str, ours: str, theirs: str) -> tuple[int, str, dict[str, dict[int, str]]]:
+ """git's merge of `ours` and `theirs` with .gitattributes read from `attributes`:
+ (0 clean or 1 conflicted, the tree, each conflicted path's stages)."""
+ done = self._raw("-c", f"attr.tree={attributes}", "merge-tree", "--write-tree", "-z", "--no-messages",
+ ours, theirs)
+ if done.returncode not in (0, 1):
+ raise ReadFailed(f"MERGE_TREE_FAILED:{done.returncode}")
+ fields = done.stdout.split(b"\0")
+ tree = fields[0].decode().strip()
+ stages: dict[str, dict[int, str]] = {}
+ if done.returncode == 1:
+ for field in fields[1:]:
+ if not field:
+ break
+ meta, _, path = field.partition(b"\t")
+ _mode, sha, stage = meta.decode().split()
+ stages.setdefault(path.decode("utf-8", "surrogateescape"), {})[int(stage)] = sha
+ return done.returncode, tree, stages
+
+ def _union_attribute_tree(self, commit: str, paths: list[str]) -> str:
+ """`commit`'s tree with `merge=union` declared for `paths`, each in the
+ `.gitattributes` of its own directory after whatever that file already declares:
+ an attribute source for git's merge, never committed."""
+ by_dir: dict[str, list[str]] = {}
+ for path in paths:
+ folder, _, name = path.rpartition("/")
+ by_dir.setdefault(folder, []).append(name)
+ sets: dict[str, tuple[str, str] | None] = {}
+ for folder, names in by_dir.items():
+ attributes = f"{folder}/.gitattributes" if folder else ".gitattributes"
+ read = self._raw("cat-file", "blob", f"{commit}:{attributes}")
+ text = read.stdout.decode("utf-8", "surrogateescape") if read.returncode == 0 else ""
+ body = union_attributes(text, names).encode("utf-8", "surrogateescape")
+ written = self._raw("hash-object", "-w", "--stdin", input=body)
+ if written.returncode != 0:
+ raise ReadFailed(f"GIT_FAILED:hash-object:{written.returncode}")
+ sets[attributes] = ("100644", written.stdout.decode().strip())
+ return self._overlay_tree(self.tree(commit), sets)
+
repair_step = union_step
def _overlay_tree(self, base_tree: str, sets: dict[str, tuple[str, str] | None]) -> str:
diff --git a/src/svrf/rules.py b/src/svrf/rules.py
index 5589f3a..8331e08 100644
--- a/src/svrf/rules.py
+++ b/src/svrf/rules.py
@@ -34,7 +34,7 @@
"interleaved-owners": {"lean": "interleaved_landing", "check": "choose_families"},
"repair-mechanical": {"lean": "retry_iff", "check": "repair_class"},
"ordered-reland": {"lean": "reland_tree", "check": "reland_class"},
- "union-merge": {"lean": "change_comm", "check": "union_lines"},
+ "union-merge": {"lean": "change_comm", "check": "union_attributes"},
"speculative-stacking": {"lean": "stack_lands_gated", "check": "Train.round"},
"speculation-void": {"lean": "stackStatus_void_iff", "check": "Train.round"},
"bisect-holds-exactly-red": {"lean": "settle_outcome", "check": "Train.settle_red"},
@@ -49,17 +49,21 @@
# --------------------------------------------------------------------------- merging
-def union_lines(*, theirs: str, ours: str) -> str:
- """A union-merged file: the base branch's lines in order, then the lines the pull
- request adds, each once. Suited to files whose lines are independent entries: import
- indexes, requirement lists, changelog bullets."""
- out = theirs.splitlines()
- seen = set(out)
- for line in ours.splitlines():
- if line not in seen:
- out.append(line)
- seen.add(line)
- return "\n".join(out) + "\n"
+def union_attributes(text: str, names) -> str:
+ """A directory's `.gitattributes` body (`text`, possibly empty) with git's union merge
+ (`merge=union`) declared for each file name in `names`, after everything `text` already
+ declares, so for those files it is the line git reads last. Each name is anchored to the
+ directory and its pattern characters escaped; a name with whitespace, a quote or a
+ control character is written as a quoted pattern."""
+ out = text if not text or text.endswith("\n") else text + "\n"
+ for name in sorted(set(names)):
+ pattern = "/" + re.sub(r"([\\*?\[])", r"\\\1", name)
+ if any(c.isspace() or c == '"' or ord(c) < 32 or ord(c) == 127 for c in pattern):
+ quoted = pattern.replace("\\", "\\\\").replace('"', '\\"')
+ quoted = re.sub(r"[\x00-\x1f\x7f]", lambda m: "\\%03o" % ord(m.group()), quoted)
+ pattern = f'"{quoted}"'
+ out += f"{pattern} merge=union\n"
+ return out
def chunk(items: list, size: int) -> list[list]:
diff --git a/tests/test_history.py b/tests/test_history.py
index b3a9466..08b6c24 100644
--- a/tests/test_history.py
+++ b/tests/test_history.py
@@ -175,7 +175,7 @@ def test_the_repair_pushes_a_union_merge_while_the_branch_is_checked_out_elsewhe
self.assertEqual(pushed, step.commit)
self.assertEqual(g.parents(step.commit), [head, main])
text = git(self.work, "cat-file", "-p", f"{step.commit}:requirements.txt")
- self.assertEqual(text.splitlines(), ["a", "main", "lane"])
+ self.assertEqual(text.splitlines(), ["a", "lane", "main"])
rows = history.dispositions(g, main, step.commit, KIND)
self.assertEqual([r["disposition"] for r in rows], ["TESTS", "CODE", "MERGE"])
diff --git a/tests/test_train.py b/tests/test_train.py
index 3c38e5c..77d8881 100644
--- a/tests/test_train.py
+++ b/tests/test_train.py
@@ -30,11 +30,6 @@ def train(repo, gh, gate, tmp, clock=None, **kw):
class PureReads(unittest.TestCase):
- def test_union_lines_keep_base_order_then_the_additions_once(self):
- base = "a\nb\nm\n"
- head = "a\nb\nl\n"
- self.assertEqual(rules.union_lines(theirs=base, ours=head), "a\nb\nm\nl\n")
-
def test_a_rate_limit_is_a_read_failure_never_a_conflict(self):
with self.assertRaises(RateLimited) as caught:
classify_gh_failure(1, "GraphQL: API rate limit already exceeded for user ID 7.")
diff --git a/tests/test_union_merge.py b/tests/test_union_merge.py
new file mode 100644
index 0000000..843d1af
--- /dev/null
+++ b/tests/test_union_merge.py
@@ -0,0 +1,267 @@
+"""The union merge of a configured path is git's own union merge, byte for byte.
+
+Each case builds a base, a pull request side ("ours") and a base-branch side ("theirs")
+of one file in a real repository and runs the union step on it. The result is compared
+with git itself in two ways: `git merge-file --union` on the three versions, and a git
+merge of the same three commits in a repository whose `.gitattributes` declares
+`merge=union` for the file (what `git merge` does with that attribute). Neither
+reference repository is the one the union step runs in.
+"""
+
+from __future__ import annotations
+
+import os
+import random
+import shutil
+import subprocess
+import tempfile
+import unittest
+from pathlib import Path
+
+from fakes import IDENT, git
+
+from svrf import rules
+from svrf.git import RealGit
+from svrf.globs import PathSet
+
+
+def _raw(cwd: Path, *args: str, input: bytes | None = None, ok=(0,)) -> bytes:
+ done = subprocess.run(["git", "-C", str(cwd), *args], capture_output=True, input=input,
+ env={**os.environ, **IDENT})
+ if done.returncode not in ok:
+ raise AssertionError(f"git {' '.join(args)}: {done.returncode}: {done.stderr!r}")
+ return done.stdout
+
+
+def _histogram() -> list[str]:
+ """`git merge` runs its content merges with the histogram diff. `git merge-file` takes
+ that option where this git offers it; the fixed cases below merge the same either way."""
+ usage = subprocess.run(["git", "merge-file", "-h"], capture_output=True, text=True)
+ return ["--diff-algorithm=histogram"] if "diff-algorithm" in usage.stdout + usage.stderr else []
+
+
+class Repository:
+ """A scratch repository whose commits are built by plumbing from {path: bytes}."""
+
+ def __init__(self, root: Path):
+ self.root = root
+ root.mkdir(parents=True)
+ git(root, "init", "-q", "-b", "main")
+
+ def commit(self, files: dict[str, bytes], parent: str | None = None) -> str:
+ index = self.root / ".git" / "scratch-index"
+ if index.exists():
+ index.unlink()
+ env = {"GIT_INDEX_FILE": str(index)}
+ for path, body in files.items():
+ blob = _raw(self.root, "hash-object", "-w", "--stdin", input=body).decode().strip()
+ git(self.root, "update-index", "--add", "--cacheinfo", f"100644,{blob},{path}", env=env)
+ tree = git(self.root, "write-tree", env=env)
+ index.unlink(missing_ok=True)
+ return git(self.root, "commit-tree", tree, "-m", "c", *(["-p", parent] if parent else []))
+
+ def blob(self, commit_or_tree: str, path: str) -> bytes:
+ return _raw(self.root, "cat-file", "blob", f"{commit_or_tree}:{path}")
+
+
+class UnionIsGitsUnion(unittest.TestCase):
+ PATH = "index.txt"
+
+ def setUp(self):
+ self.tmp = Path(tempfile.mkdtemp())
+ self.addCleanup(shutil.rmtree, self.tmp, ignore_errors=True)
+ self.count = 0
+
+ # ---- the three readings of one case
+
+ def _sides(self, repo: Repository, base, ours, theirs, path, extra=None):
+ common = dict(extra or {})
+ if base is not None:
+ common[path] = base
+ root = repo.commit(common)
+ return (repo.commit({**common, path: ours}, root), repo.commit({**common, path: theirs}, root))
+
+ def svrf_union(self, base, ours, theirs, path=None, extra=None) -> bytes:
+ path = path or self.PATH
+ self.count += 1
+ repo = Repository(self.tmp / f"svrf{self.count}")
+ head, acc = self._sides(repo, base, ours, theirs, path, extra)
+ g = RealGit(repo.root, union=PathSet([path]))
+ step = g.union_step(acc, head, "Merge main into lane")
+ self.assertEqual(step.status, "CLEAN", step)
+ self.assertEqual(g.parents(step.commit), [head, acc])
+ return repo.blob(step.tree, path)
+
+ def merge_file_union(self, base, ours, theirs) -> bytes:
+ scratch = Path(tempfile.mkdtemp(dir=self.tmp))
+ names = []
+ for role, body in (("ours", ours), ("base", base or b""), ("theirs", theirs)):
+ (scratch / role).write_bytes(body)
+ names.append(str(scratch / role))
+ done = subprocess.run(["git", "merge-file", "-p", "--union", *_histogram(), *names], capture_output=True)
+ self.assertEqual(done.returncode, 0, done.stderr)
+ return done.stdout
+
+ def attribute_union(self, base, ours, theirs, path=None) -> bytes:
+ """git's merge of the same three versions with `merge=union` declared for the path."""
+ path = path or self.PATH
+ self.count += 1
+ repo = Repository(self.tmp / f"attr{self.count}")
+ attrs = {".gitattributes": f"/{path} merge=union\n".encode()}
+ head, acc = self._sides(repo, base, ours, theirs, path, attrs)
+ done = subprocess.run(["git", "-C", str(repo.root), "-c", f"attr.tree={acc}", "merge-tree", "--write-tree",
+ head, acc], capture_output=True, text=True)
+ self.assertEqual(done.returncode, 0, done.stdout + done.stderr)
+ return repo.blob(done.stdout.split("\n", 1)[0].strip(), path)
+
+ def assertGitUnion(self, base, ours, theirs, *, expected: bytes | None = None):
+ got = self.svrf_union(base, ours, theirs)
+ self.assertEqual(got, self.merge_file_union(base, ours, theirs))
+ self.assertEqual(got, self.attribute_union(base, ours, theirs))
+ if expected is not None:
+ self.assertEqual(got, expected)
+
+ # ---- fixed cases
+
+ def test_insertions_at_the_same_spot_keep_ours_then_theirs_in_place(self):
+ self.assertGitUnion(b"a\nb\nc\n", b"a\nb\nO1\nO2\nc\n", b"a\nb\nT1\nc\n",
+ expected=b"a\nb\nO1\nO2\nT1\nc\n")
+
+ def test_the_pull_request_side_comes_first_within_a_hunk_not_after_the_whole_base(self):
+ self.assertGitUnion(b"a\nm\nz\n", b"a\nm\nO\nz\n", b"a\nm\nT\nz\nT2\n",
+ expected=b"a\nm\nO\nT\nz\nT2\n")
+
+ def test_both_sides_append_at_the_end(self):
+ self.assertGitUnion(b"a\n", b"a\nO\n", b"a\nT\n", expected=b"a\nO\nT\n")
+
+ def test_interleaved_hunks(self):
+ base = b"".join(b"%d\n" % i for i in range(1, 13))
+ ours = base.replace(b"2\n", b"2o\n").replace(b"5\n", b"5o\n").replace(b"9\n", b"9o\n9p\n")
+ theirs = base.replace(b"2\n", b"2t\n").replace(b"6\n", b"6t\n").replace(b"9\n", b"9t\n")
+ self.assertGitUnion(base, ours, theirs)
+
+ def test_clean_hunks_and_a_union_hunk_in_one_file(self):
+ base = b"head\none\ntwo\nthree\nfour\nfive\ntail\n"
+ ours = b"head\none\nTWO\nthree\nfour\nfive-o\ntail\n"
+ theirs = b"head\nthree\nfour\nfive-t\ntail\nmore\n"
+ self.assertGitUnion(base, ours, theirs)
+
+ def test_a_deletion_on_one_side_is_kept(self):
+ self.assertGitUnion(b"a\ngone\nb\nc\n", b"a\nb\nc\nO\n", b"a\ngone\nb\nc\nT\n",
+ expected=b"a\nb\nc\nO\nT\n")
+
+ def test_repeated_lines_are_not_collapsed(self):
+ self.assertGitUnion(b"x\n", b"x\nO\nO\n", b"x\nT\nO\n")
+
+ def test_a_line_both_sides_add_appears_as_git_puts_it(self):
+ self.assertGitUnion(b"a\nc\n", b"a\nb\nO\nc\n", b"a\nb\nT\nc\n")
+
+ def test_one_side_empties_the_file(self):
+ self.assertGitUnion(b"a\nb\n", b"", b"a\nb\nT\n")
+
+ def test_no_trailing_newline_on_ours(self):
+ self.assertGitUnion(b"a\nb\n", b"a\nb\nO", b"a\nb\nT\n")
+
+ def test_no_trailing_newline_on_theirs(self):
+ self.assertGitUnion(b"a\nb\n", b"a\nb\nO\n", b"a\nb\nT")
+
+ def test_no_trailing_newline_anywhere(self):
+ self.assertGitUnion(b"a\nb", b"a\nb\nO", b"a\nb\nT")
+
+ def test_a_trailing_newline_added_on_one_side_only(self):
+ self.assertGitUnion(b"a\nb", b"a\nb\n", b"a\nb\nT")
+
+ def test_crlf_lines_stay_crlf(self):
+ self.assertGitUnion(b"a\r\nb\r\n", b"a\r\nO\r\nb\r\n", b"a\r\nT\r\nb\r\n",
+ expected=b"a\r\nO\r\nT\r\nb\r\n")
+
+ def test_mixed_line_endings(self):
+ self.assertGitUnion(b"a\r\nb\nc\r\n", b"a\r\nb\nO\nc\r\n", b"a\r\nb\nT\r\nc\r\n")
+
+ def test_bytes_that_are_not_utf8(self):
+ self.assertGitUnion(b"a\n\xff\xfe\n", b"a\n\xff\xfe\nO \xe9\n", b"a\n\xff\xfe\nT \x80\n")
+
+ def test_a_file_added_on_both_sides(self):
+ self.assertGitUnion(None, b"x\ny\n", b"x\nz\n", expected=b"x\ny\nz\n")
+
+ def test_a_path_with_characters_special_to_attribute_patterns(self):
+ path = "sub dir/a b*[1]?.txt"
+ got = self.svrf_union(b"a\n", b"a\nO\n", b"a\nT\n", path=path)
+ self.assertEqual(got, b"a\nO\nT\n")
+
+ def test_a_directory_attributes_file_that_declares_another_merge(self):
+ extra = {"sub/.gitattributes": b"*.txt merge=binary\n"}
+ got = self.svrf_union(b"a\n", b"a\nO\n", b"a\nT\n", path="sub/list.txt", extra=extra)
+ self.assertEqual(got, b"a\nO\nT\n")
+
+ def test_seeded_random_cases_match_gits_merge_with_the_union_attribute(self):
+ rng = random.Random(20261004)
+ words = ["a", "b", "c", "d", "{", "}", "x", "y", ""]
+ for _ in range(40):
+ base = [rng.choice(words) for _ in range(rng.randint(0, 10))]
+
+ def edit(lines):
+ lines = list(lines)
+ for _ in range(rng.randint(1, 4)):
+ at = rng.randint(0, len(lines))
+ roll = rng.random()
+ if roll < 0.5 or not lines:
+ lines.insert(at, rng.choice(words + [f"n{rng.randint(0, 3)}"]))
+ elif roll < 0.8:
+ del lines[min(at, len(lines) - 1)]
+ else:
+ lines[min(at, len(lines) - 1)] = rng.choice(words)
+ return lines
+
+ def body(lines):
+ end = rng.choice(["\n", "\n", "\n", "", "\r\n"])
+ return (end if end else "\n").join(lines).encode() + (end.encode() if lines else b"")
+
+ b, o, t = body(base), body(edit(base)), body(edit(base))
+ with self.subTest(base=b, ours=o, theirs=t):
+ self.assertEqual(self.svrf_union(b, o, t), self.attribute_union(b, o, t))
+
+ # ---- what still conflicts
+
+ def test_a_union_path_deleted_on_one_side_still_conflicts(self):
+ repo = Repository(self.tmp / "delete")
+ root = repo.commit({self.PATH: b"a\n", "keep.txt": b"k\n"})
+ head = repo.commit({self.PATH: b"a\nO\n", "keep.txt": b"k\n"}, root)
+ acc = repo.commit({"keep.txt": b"k\n"}, root)
+ step = RealGit(repo.root, union=PathSet([self.PATH])).union_step(acc, head, "m")
+ self.assertEqual((step.status, step.conflicts), ("CONFLICT", [self.PATH]))
+
+ def test_a_conflict_on_another_path_still_conflicts(self):
+ repo = Repository(self.tmp / "other")
+ root = repo.commit({self.PATH: b"a\n", "code.py": b"x = 0\n"})
+ head = repo.commit({self.PATH: b"a\nO\n", "code.py": b"x = 1\n"}, root)
+ acc = repo.commit({self.PATH: b"a\nT\n", "code.py": b"x = 2\n"}, root)
+ step = RealGit(repo.root, union=PathSet([self.PATH])).union_step(acc, head, "m")
+ self.assertEqual((step.status, step.conflicts), ("CONFLICT", ["code.py"]))
+
+
+class UnionAttributes(unittest.TestCase):
+ """The pure part: declaring git's union merge for given files in one directory's
+ `.gitattributes`, after everything that file already declares."""
+
+ def test_appends_after_existing_lines(self):
+ self.assertEqual(rules.union_attributes("*.md text\n", ["list.txt"]),
+ "*.md text\n/list.txt merge=union\n")
+
+ def test_terminates_an_unterminated_last_line(self):
+ self.assertEqual(rules.union_attributes("*.md text", ["list.txt"]),
+ "*.md text\n/list.txt merge=union\n")
+
+ def test_escapes_pattern_characters_and_quotes_whitespace(self):
+ self.assertEqual(rules.union_attributes("", ["a*b?[c]\\d.txt"]),
+ "/a\\*b\\?\\[c]\\\\d.txt merge=union\n")
+ self.assertEqual(rules.union_attributes("", ['a b"c.txt']),
+ '"/a b\\"c.txt" merge=union\n')
+
+ def test_each_name_once_in_order(self):
+ self.assertEqual(rules.union_attributes("", ["b", "a", "b"]),
+ "/a merge=union\n/b merge=union\n")
+
+
+if __name__ == "__main__":
+ unittest.main()