diff --git a/scripts/aggregate_merge.md b/scripts/aggregate_merge.md new file mode 100644 index 0000000000..e800bde15f --- /dev/null +++ b/scripts/aggregate_merge.md @@ -0,0 +1,63 @@ +# Report an aggregate-only merge conflict + +`aggregate_merge.py` checks the narrow generated `Complexity.lean` case from +[#2274](https://github.com/cslib-community/complexitylib/issues/2274). Run it against immutable, +locally available commits: + +```sh +python3 -B scripts/aggregate_merge.py --repo /path/to/checkout \ + --prefix FULL_ACCEPTED_PREFIX_SHA --source FULL_SOURCE_HEAD_SHA \ + --reviewed-base FULL_ORIGINAL_PARENT_SHA --reviewed-head FULL_REVIEWED_HEAD_SHA \ + > aggregate-report.json +``` + +It prints JSON and exits zero only for `status: verified`. `rejected` means an input or merge +falls outside the supported case; `limitation` means canonical reconstruction passed but the +ordinary tracked-attribute Git merge did not produce that exact tree. Neither status authorizes +integration. The report includes immutable inputs, trees, the complete candidate entry map, +reviewed payload entries, feature/aggregate digests and exact inverse recovery of the prefix. + +The verifier requires the complete neutral conflict set to be exactly `Complexity.lean`. +Each input aggregate must have the exact module-style preamble and complete sorted public +imports generated by `scripts/MkAll.lean`. Module names follow the conservative path syntax in +`quick_check.py`; dynamic Lake configuration, moved library roots, symlink modules and other +aggregate styles are rejected. This is an exact generated-format check, not a general Lean +parser that might overlook declarations or options after the imports. + +Canonical imports come from the merged library inventory, so additions and deletions are both +accounted for. Every other entry and mode must equal the neutral merge. The complete original +nonaggregate feature, including docs, tests and removals, must replay onto the accepted prefix; +reviewed payload modes/blobs must remain exact. Reversing that feature and restoring the prefix +aggregate must recover the exact prefix tree. Additional unreviewed source payloads are rejected. + +Finally, Git's ordinary merge with the prefix's tracked attributes must yield the identical +canonical tree. For example, an adjacent insertion may produce sorted imports with `merge=union`, +while reversing the insertion order may produce unsorted imports. Adjacent independent deletions +can also reconstruct correctly while union retains deleted imports. Both latter cases report a +limitation: the source owner must refresh the source branch through the usual reviewed process. +The tool never proposes an alternative queue API or enables a merge driver. + +Git operations use a temporary bare repository/index/object directory populated by an ordinary +local `file://` fetch. It has its own common directory, no alternates or shallow boundary, and +no object inode shared with the source. These properties are checked. Fetching complete history +can cost disk space and time; shallow source repositories are rejected. Pack threads are capped +at two. All inherited `GIT_*` environment variables are cleared before the first Git command, +including object, index, common-directory and worktree redirects. +No source/index/ref/config/object write, merge commit, +queue operation, workflow or cache action occurs in the inspected repository. User/system +attributes, local `info/attributes`, hooks and configured merge drivers are excluded; only the +prefix's tracked attributes participate in the ordinary comparison. Python 3.11+ and Git 2.43+ +are required. Unavailable history is an error, not a request to repair or reconstruct objects. + +This local standard-Git comparison does not certify GitHub's settings, current source ownership, +review approval, dependency landings or CI. Coordinators must still guard the live source/full +prefix, verify the actual hosted candidate tree, and require all six queue checks, including +`mk_all --check`. The report does not change admission policy. + +Focused regression tests use independent temporary Git repositories and run without Lean: + +```sh +python3 -B -m unittest discover -s scripts -p test_aggregate_merge.py -v +``` + +No Lean statement, definition, selected abstract, catalogue entry or rendered PDF changes. diff --git a/scripts/aggregate_merge.py b/scripts/aggregate_merge.py new file mode 100644 index 0000000000..21dcc87e75 --- /dev/null +++ b/scripts/aggregate_merge.py @@ -0,0 +1,261 @@ +#!/usr/bin/env python3 +"""Report whether an aggregate-only conflict has an exact canonical Git merge. + +All four inputs must be full commit IDs available locally. Git writes only to a temporary +repository; the inspected repository is read-only. See aggregate_merge.md for the contract. +""" + +from __future__ import annotations + +import argparse +from contextlib import contextmanager +import hashlib +import json +import os +from pathlib import Path +import re +import subprocess +import sys +import tempfile +import tomllib + + +AGGREGATE = "Complexity.lean" +# scripts/MkAll.lean, module-style branch; intentionally not a general Lean header parser. +MODULE_HEADER = b"module -- shake: keep-all --deprecated_module: ignore\n\n" +COMPONENT = re.compile(r"[A-Za-z_][A-Za-z_0-9]*\Z") # quick_check.source_modules convention +OID = re.compile(r"[0-9a-f]{40}\Z") + + +class Rejected(Exception): + """An input or merge is outside this verifier's bounded contract.""" + + +def require(condition: bool, message: str) -> None: + if not condition: + raise Rejected(message) + + +def environment() -> dict[str, str]: + # Do not inherit replacement objects, index/object directories or user merge drivers. + env = {key: value for key, value in os.environ.items() if not key.startswith("GIT_")} + env.update(GIT_CONFIG_NOSYSTEM="1", GIT_CONFIG_GLOBAL=os.devnull, + GIT_ATTR_NOSYSTEM="1", GIT_NO_REPLACE_OBJECTS="1", GIT_OPTIONAL_LOCKS="0") + return env + + +class Git: + def __init__(self, root: Path, env: dict[str, str]): + self.root, self.env = root, env + + def run(self, *args: str, data: bytes | None = None, allowed=(0,)): + result = subprocess.run( + ["git", "-c", "gc.auto=0", "-c", "maintenance.auto=false", + "-c", "pack.threads=2", "-c", "fetch.parallel=1", + "-c", f"core.attributesFile={os.devnull}", "-c", f"core.hooksPath={os.devnull}", + *args], cwd=self.root, env=self.env, input=data, capture_output=True) + require(result.returncode in allowed, + f"git {args[0]} failed: {result.stderr.decode(errors='replace').strip()}") + return result + + def text(self, *args: str) -> str: + return self.run(*args).stdout.decode().strip() + + def entries(self, tree: str) -> dict[str, dict[str, str]]: + entries = {} + for row in self.run("ls-tree", "-rz", tree).stdout.split(b"\0"): + if row: + metadata, path = row.split(b"\t", 1) + mode, kind, oid = metadata.decode().split() + entries[path.decode()] = {"mode": mode, "type": kind, "oid": oid} + return entries + + def blob(self, entry: dict[str, str]) -> bytes: + require(entry["type"] == "blob", "expected a blob") + return self.run("cat-file", "blob", entry["oid"]).stdout + + def set_aggregate(self, entry: dict[str, str]) -> str: + self.run("update-index", "--add", "--cacheinfo", entry["mode"], entry["oid"], AGGREGATE) + return self.text("write-tree") + + +@contextmanager +def scratch_git(repo: Path, commits: tuple[str, ...]): + require(all(OID.fullmatch(oid) for oid in commits), "use full immutable commit IDs") + source = Git(repo, environment()) + require(source.text("rev-parse", "--show-object-format") == "sha1", "only SHA-1 is supported") + require(source.text("rev-parse", "--is-shallow-repository") == "false", + "shallow source repositories are unsupported") + objects = Path(source.text("rev-parse", "--path-format=absolute", "--git-path", "objects")) + common = Path(source.text("rev-parse", "--path-format=absolute", "--git-common-dir")) + with tempfile.TemporaryDirectory(prefix="aggregate-merge-") as directory: + root = Path(directory) + source.run("init", "--bare", "--quiet", "--template=", "--object-format=sha1", str(root)) + git = Git(root, environment()) + require(Path(git.text("rev-parse", "--absolute-git-dir")).resolve() == root.resolve() and + Path(git.text("rev-parse", "--path-format=absolute", "--git-common-dir")).resolve() + == root.resolve() and root.resolve() != common.resolve(), + "temporary repository does not have an independent common directory") + own_objects = Path(git.text("rev-parse", "--path-format=absolute", "--git-path", "objects")) + require(own_objects.resolve() == (root / "objects").resolve() and + own_objects.resolve() != objects.resolve() and not (root / ".git").exists(), + "temporary object directory is not independent") + # Check identity before fetching or writing merge/index objects. A normal file:// + # fetch uses upload-pack, unlike clone's local hard-link shortcut. + git.run("fetch", "--no-tags", "--no-write-fetch-head", "--no-recurse-submodules", + "--no-auto-maintenance", repo.as_uri(), *dict.fromkeys(commits)) + require(not (own_objects / "info/alternates").exists() and + not (own_objects / "info/http-alternates").exists() and not (root / "shallow").exists(), + "temporary repository contains alternates or a shallow boundary") + source_inodes = {(p.stat().st_dev, p.stat().st_ino) for p in objects.rglob("*") if p.is_file()} + copied = [p for p in own_objects.rglob("*") if p.is_file()] + require(all(not p.is_symlink() and (p.stat().st_dev, p.stat().st_ino) not in source_inodes + for p in copied), "temporary objects share source storage") + git.isolation = {"independent_common_directory": True, "independent_object_directory": True, + "no_alternates_or_shallow_boundary": True, "no_shared_object_inodes": True, + "fetched_object_files": len(copied)} + yield git + + +def canonical(entries: dict[str, dict[str, str]]) -> bytes: + """Mirror mk_all's Complexity root: all .lean modules, sorted by module name.""" + modules = [] + for path, entry in entries.items(): + if path.startswith("Complexity/") and path.endswith(".lean"): + parts = path[:-5].split("/") + require(all(COMPONENT.fullmatch(part) for part in parts), + f"unsupported module path: {path}") + require(entry["type"] == "blob" and entry["mode"] in {"100644", "100755"}, + f"nonregular Lean module: {path}") + modules.append(".".join(parts)) + require(bool(modules), "empty Complexity library") + return MODULE_HEADER + b"".join(f"public import {name}\n".encode() for name in sorted(modules)) + + +def check_aggregate(git: Git, entries: dict, label: str) -> None: + require("lakefile.lean" not in entries and "lakefile.toml" in entries, + f"{label}: unsupported Lake configuration") + config = tomllib.loads(git.blob(entries["lakefile.toml"]).decode()) + libraries = config.get("lean_lib", []) + require(isinstance(libraries, list), f"{label}: invalid Lake libraries") + roots = [lib for lib in libraries if isinstance(lib, dict) and lib.get("name") == "Complexity"] + require(config.get("srcDir", ".") == "." and len(roots) == 1 and + roots[0].get("srcDir", ".") == "." and + roots[0].get("roots", ["Complexity"]) == ["Complexity"] and "globs" not in roots[0], + f"{label}: unsupported Complexity library layout") + require(AGGREGATE in entries and entries[AGGREGATE]["mode"] == "100644", + f"{label}: missing or nonregular aggregate") + content = git.blob(entries[AGGREGATE]) + require(content.startswith(MODULE_HEADER), f"{label}: changed known aggregate preamble") + # Exact generation rejects options/declarations, comments, malformed, repeated, missing, + # extra and unsorted imports. General Lean parsing would accept too much here. + require(content == canonical(entries), f"{label}: aggregate is not canonical for its inventory") + + +def merge(git: Git, prefix: str, source: str, attributes: str): + result = git.run("-c", f"attr.tree={attributes}", "merge-tree", "--write-tree", + "--name-only", "-z", prefix, source, allowed=(0, 1)) + fields = result.stdout.split(b"\0") + tree = fields.pop(0).decode() + require(bool(OID.fullmatch(tree)), "unexpected merge-tree output (Git >= 2.43 required)") + conflicts = [] + for field in fields: + if not field: + break + conflicts.append(field.decode()) + return result.returncode, tree, conflicts + + +def verify(git: Git, prefix: str, source: str, reviewed_base: str, reviewed_head: str, + report: dict) -> None: + refs = {"prefix": prefix, "source": source, + "reviewed_base": reviewed_base, "reviewed_head": reviewed_head} + for label, oid in refs.items(): + require(bool(OID.fullmatch(oid)), f"{label}: use a full immutable commit ID") + require(git.text("rev-parse", "--verify", f"{oid}^{{commit}}") == oid, + f"{label}: expected a commit") + require(git.run("merge-base", "--is-ancestor", reviewed_base, reviewed_head, + allowed=(0, 1)).returncode == 0, "reviewed base is not an ancestor of reviewed head") + require(git.run("merge-base", "--is-ancestor", reviewed_head, source, + allowed=(0, 1)).returncode == 0, "source does not retain reviewed history") + require(len(git.text("merge-base", "--all", prefix, source).splitlines()) == 1, + "merge needs exactly one base") + inventories = {label: git.entries(oid) for label, oid in refs.items()} + for label, entries in inventories.items(): + check_aggregate(git, entries, label) + report.update(inputs=refs, isolation=git.isolation, + input_trees={label: git.text("rev-parse", f"{oid}^{{tree}}") + for label, oid in refs.items()}) + empty = git.run("hash-object", "-t", "tree", "--stdin", "-w", data=b"").stdout.decode().strip() + # attr.tree must actually disable the tracked union attribute on this Git version. + attrs = git.run("-c", f"attr.tree={empty}", "check-attr", "-z", "merge", "--", AGGREGATE).stdout + require(attrs == b"Complexity.lean\0merge\0unspecified\0", "cannot disable tracked attributes") + status, neutral, conflicts = merge(git, prefix, source, empty) + report.update(neutral_tree=neutral, conflicts=conflicts) + require(status == 1 and conflicts == [AGGREGATE], + f"complete neutral conflict set must be exactly [{AGGREGATE!r}], got {conflicts!r}") + merged = git.entries(neutral) + content = canonical(merged) + entry = {"mode": "100644", "type": "blob", + "oid": git.run("hash-object", "-w", "--stdin", data=content).stdout.decode().strip()} + git.run("read-tree", neutral) + candidate = git.set_aggregate(entry) + candidate_entries = git.entries(candidate) + require({p: e for p, e in merged.items() if p != AGGREGATE} == + {p: e for p, e in candidate_entries.items() if p != AGGREGATE}, + "canonicalization changed a nonaggregate entry") + feature = git.run("diff", "--no-ext-diff", "--no-textconv", "--no-renames", "--binary", + "--full-index", reviewed_base, reviewed_head, "--", ".", + ":(exclude)Complexity.lean").stdout + require(bool(feature), "reviewed feature has no nonaggregate payload") + own_paths = sorted(p for p in inventories["reviewed_base"].keys() | inventories["reviewed_head"].keys() + if p != AGGREGATE and inventories["reviewed_base"].get(p) != + inventories["reviewed_head"].get(p)) + for path in own_paths: + require(candidate_entries.get(path) == inventories["reviewed_head"].get(path), + f"reviewed payload mode/blob changed: {path}") + # Whole-feature replay is stronger than checking only the named Lean payloads. It also + # guards docs, removals, modes and every unrelated accepted-prefix entry. + git.run("read-tree", prefix) + git.run("apply", "--cached", "--binary", data=feature) + require(git.set_aggregate(entry) == candidate, "feature forward replay does not recover candidate") + git.run("read-tree", candidate) + git.run("apply", "--cached", "--binary", "--reverse", data=feature) + recovered = git.set_aggregate(inventories["prefix"][AGGREGATE]) + require(recovered == report["input_trees"]["prefix"], "feature inverse does not recover exact prefix") + report.update(candidate_tree=candidate, candidate_entries=candidate_entries, + reviewed_paths=own_paths, reviewed_entries={p: candidate_entries.get(p) for p in own_paths}, + feature_patch_sha256=hashlib.sha256(feature).hexdigest(), recovered_prefix_tree=recovered, + canonical_aggregate_sha256=hashlib.sha256(content).hexdigest(), + canonical_import_count=len(content.splitlines()) - 2) + # A private canonical tree is not an admission mechanism. The ordinary tracked-attribute + # merge, in the target-prefix direction, must produce exactly that tree itself. + status, default, conflicts = merge(git, prefix, source, prefix) + report.update(default_merge_tree=default, default_conflicts=conflicts) + if status != 0 or conflicts or default != candidate: + report["status"] = "limitation" + raise Rejected("ordinary tracked-attribute merge cannot represent the exact canonical candidate; " + "refresh the source branch through its owner") + report["status"] = "verified" + + +def main(argv=None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--repo", type=Path, default=Path.cwd()) + for name in ("prefix", "source", "reviewed-base", "reviewed-head"): + parser.add_argument(f"--{name}", required=True) + args = parser.parse_args(argv) + report = {"status": "rejected", "report_only": True, + "limits": "No source review, CI, hosted candidate or queue authority is certified."} + try: + with scratch_git(args.repo.resolve(), + (args.prefix, args.source, args.reviewed_base, args.reviewed_head)) as git: + verify(git, args.prefix, args.source, args.reviewed_base, args.reviewed_head, report) + except (Rejected, OSError, ValueError) as error: + report["reason"] = str(error) + print(json.dumps(report, indent=2, sort_keys=True)) + return 0 if report["status"] == "verified" else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/test_aggregate_merge.py b/scripts/test_aggregate_merge.py new file mode 100644 index 0000000000..220d97abf7 --- /dev/null +++ b/scripts/test_aggregate_merge.py @@ -0,0 +1,289 @@ +#!/usr/bin/env python3 +"""Real Git fixtures for the report-only aggregate verifier; no Lean build is needed.""" + +import contextlib +import hashlib +import io +import json +import os +from pathlib import Path +import subprocess +import tempfile +import unittest +from unittest.mock import patch + +import aggregate_merge as verifier + + +class AggregateMergeTests(unittest.TestCase): + def setUp(self): + temp = tempfile.TemporaryDirectory(prefix="test-aggregate-merge-") + self.addCleanup(temp.cleanup) + self.repo = Path(temp.name) + self.env = verifier.environment() + self.env.update(GIT_AUTHOR_NAME="Fixture", GIT_AUTHOR_EMAIL="fixture@example.invalid", + GIT_COMMITTER_NAME="Fixture", GIT_COMMITTER_EMAIL="fixture@example.invalid") + self.git("init", "--quiet", "--initial-branch=main") + self.write("lakefile.toml", 'name = "fixture"\n[[lean_lib]]\nname = "Complexity"\n') + self.write(".gitattributes", "Complexity.lean merge=union\n") + for name in ("Init", "Zed"): + self.write(f"Complexity/{name}.lean", f"-- fixture {name}\n") + self.aggregate() + self.base = self.commit("base") + + def git(self, *args): + return subprocess.run(["git", "-c", "gc.auto=0", "-c", "maintenance.auto=false", *args], + cwd=self.repo, env=self.env, check=True, + capture_output=True).stdout.decode().strip() + + def write(self, path, content): + target = self.repo / path + target.parent.mkdir(parents=True, exist_ok=True) + target.write_text(content) + + def aggregate(self): + modules = sorted(".".join(p.relative_to(self.repo).with_suffix("").parts) + for p in (self.repo / "Complexity").rglob("*.lean")) + self.write("Complexity.lean", verifier.MODULE_HEADER.decode() + + "".join(f"public import {name}\n" for name in modules)) + + def commit(self, message): + self.git("add", ".") + self.git("commit", "--quiet", "-m", message) + return self.git("rev-parse", "HEAD") + + def additions(self, prefix_name="Able", source_name="Beta"): + self.write(f"Complexity/{prefix_name}.lean", "-- accepted prefix payload\n") + self.write("prefix-doc.md", "accepted prefix documentation\n") + self.aggregate() + prefix = self.commit("prefix") + self.git("switch", "--quiet", "--detach", self.base) + self.write(f"Complexity/{source_name}.lean", "-- reviewed feature payload\n") + self.write("review-doc.md", "reviewed feature documentation\n") + self.write("ComplexityTests/Feature.lean", "-- reviewed fixture\n") + self.aggregate() + return prefix, self.commit("reviewed feature") + + def report(self, prefix, source, reviewed=None, base=None): + output = io.StringIO() + with contextlib.redirect_stdout(output): + status = verifier.main(["--repo", str(self.repo), "--prefix", prefix, + "--source", source, "--reviewed-base", base or self.base, + "--reviewed-head", reviewed or source]) + return status, json.loads(output.getvalue()) + + def snapshot(self): + # Include objects and the fixture's index/refs/config/working files, not just status. + return {str(p.relative_to(self.repo)): hashlib.sha256(p.read_bytes()).hexdigest() + for p in self.repo.rglob("*") if p.is_file()} + + def test_adjacent_additions_exact_default_and_whole_feature_recovery(self): + prefix, source = self.additions() + before = self.snapshot() + status, report = self.report(prefix, source) + self.assertEqual(status, 0, report) + self.assertEqual(report["status"], "verified") + self.assertEqual(report["conflicts"], ["Complexity.lean"]) + self.assertEqual(report["candidate_tree"], report["default_merge_tree"]) + self.assertEqual(report["recovered_prefix_tree"], self.git("rev-parse", prefix + "^{tree}")) + self.assertEqual(report["reviewed_paths"], + ["Complexity/Beta.lean", "ComplexityTests/Feature.lean", "review-doc.md"]) + self.assertEqual(report["canonical_import_count"], 4) + self.assertTrue(report["isolation"]["no_shared_object_inodes"]) + self.assertEqual(before, self.snapshot(), "verifier mutated inspected repository") + + def test_actual_temporary_git_storage_is_independent(self): + prefix, source = self.additions() + with verifier.scratch_git(self.repo, (prefix, source)) as scratch: + self.assertNotEqual(scratch.root, self.repo / ".git") + self.assertFalse((scratch.root / ".git").exists()) + self.assertFalse((scratch.root / "objects/info/alternates").exists()) + self.assertFalse((scratch.root / "objects/info/http-alternates").exists()) + self.assertFalse((scratch.root / "shallow").exists()) + source_inodes = {(p.stat().st_dev, p.stat().st_ino) + for p in (self.repo / ".git/objects").rglob("*") if p.is_file()} + temporary_objects = [p for p in (scratch.root / "objects").rglob("*") if p.is_file()] + self.assertTrue(temporary_objects) + self.assertTrue(all((p.stat().st_dev, p.stat().st_ino) not in source_inodes + for p in temporary_objects)) + # Read a real reviewed payload through the independent fetched objects. + self.assertIn("Complexity/Beta.lean", scratch.entries(source)) + + def test_adjacent_deletions_reconstruct_but_default_union_is_a_limitation(self): + for name in ("Able", "Beta"): + self.write(f"Complexity/{name}.lean", "-- removal fixture\n") + self.aggregate() + base = self.commit("deletion base") + (self.repo / "Complexity/Able.lean").unlink() + self.aggregate() + prefix = self.commit("remove Able") + self.git("switch", "--quiet", "--detach", base) + (self.repo / "Complexity/Beta.lean").unlink() + self.aggregate() + source = self.commit("remove Beta") + status, report = self.report(prefix, source, base=base) + self.assertEqual(status, 1) + self.assertEqual(report["status"], "limitation", report) + self.assertEqual(report["canonical_import_count"], 2) + self.assertEqual(report["reviewed_entries"], {"Complexity/Beta.lean": None}) + self.assertEqual(report["recovered_prefix_tree"], self.git("rev-parse", prefix + "^{tree}")) + self.assertNotEqual(report["candidate_tree"], report["default_merge_tree"]) + + def test_noncanonical_default_merge_fails_closed(self): + prefix, source = self.additions("Beta", "Able") + status, report = self.report(prefix, source) + self.assertEqual(status, 1) + self.assertEqual(report["status"], "limitation", report) + self.assertIn("cannot represent", report["reason"]) + + def test_reviewed_feature_after_ordinary_main_merge(self): + self.write("Complexity/Mid.lean", "-- already landed dependency\n") + self.aggregate() + old_main = self.commit("older main") + self.write("Complexity/Able.lean", "-- accepted prefix addition\n") + self.aggregate() + prefix = self.commit("prefix") + self.git("switch", "--quiet", "--detach", self.base) + self.write("Complexity/Beta.lean", "-- reviewed feature\n") + self.aggregate() + reviewed = self.commit("reviewed feature") + self.git("merge", "--quiet", "--no-edit", old_main) + source = self.git("rev-parse", "HEAD") + status, report = self.report(prefix, source, reviewed=reviewed) + self.assertEqual(status, 0, report) + self.assertEqual(report["reviewed_paths"], ["Complexity/Beta.lean"]) + self.assertIn("Complexity/Mid.lean", report["candidate_entries"]) + + def test_without_tracked_union_default_conflict_is_a_limitation(self): + self.write(".gitattributes", "# no aggregate merge driver\n") + self.base = self.commit("no union base") + prefix, source = self.additions() + _, report = self.report(prefix, source) + self.assertEqual(report["status"], "limitation", report) + self.assertEqual(report["default_conflicts"], ["Complexity.lean"]) + + def test_another_conflicted_file_is_rejected_even_with_union_for_it(self): + self.write(".gitattributes", "Complexity.lean merge=union\n*.md merge=union\n") + self.write("shared.md", "base\n") + self.base = self.commit("shared base") + prefix, source = self.additions() + self.write("shared.md", "source\n") + source = self.commit("source conflict") + self.git("switch", "--quiet", "--detach", prefix) + self.write("shared.md", "prefix\n") + prefix = self.commit("prefix conflict") + _, report = self.report(prefix, source) + self.assertEqual(report["status"], "rejected") + self.assertEqual(report["conflicts"], ["Complexity.lean", "shared.md"]) + + def test_changed_preamble_and_hidden_commands_and_invalid_inventories(self): + prefix, source = self.additions() + valid = (self.repo / "Complexity.lean").read_text() + mutations = { + "preamble": valid.replace("module -- shake: keep-all", "module -- shake: keep-all"), + "declaration": valid + "def hidden := 1\n", + "option": valid + "set_option maxHeartbeats 0\n", + "comment spoof": valid + "-- public import Complexity.Hidden\n", + "malformed": valid.replace("public import Complexity.Beta", "public import Complexity..Beta"), + "duplicate": valid + "public import Complexity.Beta\n", + "missing": valid.replace("public import Complexity.Beta\n", ""), + "nonexistent": valid + "public import Complexity.Absent\n", + "unsorted": valid.replace("public import Complexity.Beta\n", "") + + "public import Complexity.Beta\n", + } + for label, text in mutations.items(): + with self.subTest(label=label): + self.git("switch", "--quiet", "--detach", source) + self.write("Complexity.lean", text) + bad = self.commit(label) + status, report = self.report(prefix, bad) + self.assertEqual(status, 1) + self.assertEqual(report["status"], "rejected", report) + self.assertIn("aggregate", report["reason"]) + + def test_unreviewed_extra_payload_cannot_hide_behind_aggregate_conflict(self): + prefix, reviewed = self.additions() + self.write("unreviewed.md", "not in reviewed feature\n") + source = self.commit("extra unreviewed payload") + _, report = self.report(prefix, source, reviewed=reviewed) + self.assertEqual(report["status"], "rejected", report) + self.assertIn("forward replay", report["reason"]) + + def test_reviewed_payload_byte_and_mode_drift_are_rejected(self): + prefix, reviewed = self.additions() + for mode_only in (False, True): + with self.subTest(mode_only=mode_only): + self.git("switch", "--quiet", "--detach", reviewed) + path = self.repo / "Complexity/Beta.lean" + if mode_only: + path.chmod(0o755) + else: + path.write_text("-- changed payload\n") + source = self.commit("drift") + _, report = self.report(prefix, source, reviewed=reviewed) + self.assertEqual(report["status"], "rejected", report) + self.assertIn("mode/blob changed", report["reason"]) + + def test_nonimmutable_reference_and_missing_review_history_are_rejected(self): + prefix, source = self.additions() + _, report = self.report("HEAD", source) + self.assertIn("immutable", report["reason"]) + _, report = self.report(prefix, prefix, reviewed=source) + self.assertIn("reviewed history", report["reason"]) + + def test_no_aggregate_conflict_is_out_of_scope(self): + _, source = self.additions() + _, report = self.report(self.base, source) + self.assertEqual(report["status"], "rejected", report) + self.assertIn("complete neutral conflict set", report["reason"]) + + def test_changed_library_layout_and_symlink_modules_are_rejected(self): + prefix, source = self.additions() + self.write("lakefile.toml", 'name = "fixture"\n[[lean_lib]]\n' + 'name = "Complexity"\nsrcDir = "elsewhere"\n') + bad = self.commit("changed source directory") + _, report = self.report(prefix, bad) + self.assertIn("library layout", report["reason"]) + self.git("switch", "--quiet", "--detach", source) + (self.repo / "Complexity/Beta.lean").unlink() + (self.repo / "Complexity/Beta.lean").symlink_to("Init.lean") + bad = self.commit("symlink module") + _, report = self.report(prefix, bad) + self.assertIn("nonregular Lean module", report["reason"]) + + def test_local_info_attributes_and_merge_drivers_do_not_change_result(self): + prefix, source = self.additions("Beta", "Able") + self.write(".git/info/attributes", "Complexity.lean merge=ours\n") + self.git("config", "merge.ours.driver", "touch SHOULD_NOT_RUN") + before = self.snapshot() + _, report = self.report(prefix, source) + self.assertEqual(report["status"], "limitation", report) + self.assertEqual(before, self.snapshot()) + self.assertFalse((self.repo / "SHOULD_NOT_RUN").exists()) + + def test_hostile_inherited_git_paths_cannot_redirect_temporary_writes(self): + prefix, source = self.additions() + control = self.repo / ".git" + hostile = {"GIT_DIR": str(control), "GIT_COMMON_DIR": str(control), + "GIT_OBJECT_DIRECTORY": str(control / "objects"), + "GIT_ALTERNATE_OBJECT_DIRECTORIES": str(control / "objects"), + "GIT_INDEX_FILE": str(control / "index"), "GIT_WORK_TREE": str(self.repo)} + before = self.snapshot() + with patch.dict(os.environ, hostile): + status, report = self.report(prefix, source) + self.assertEqual(status, 0, report) + self.assertEqual(before, self.snapshot()) + + def test_noncanonical_accepted_prefix_is_rejected(self): + prefix, source = self.additions() + self.git("switch", "--quiet", "--detach", prefix) + with (self.repo / "Complexity.lean").open("a") as output: + output.write("def hidden := 1\n") + prefix = self.commit("noncanonical accepted prefix") + _, report = self.report(prefix, source) + self.assertEqual(report["status"], "rejected", report) + self.assertIn("prefix: aggregate", report["reason"]) + + +if __name__ == "__main__": + unittest.main()