diff --git a/cc/README.md b/cc/README.md new file mode 100644 index 0000000..20aeaab --- /dev/null +++ b/cc/README.md @@ -0,0 +1,31 @@ +# Conflict Catcher + +Conflict Catcher is a tiny local web app for checking whether two Git branches are likely to conflict before you run a real merge or rebase. + +## Run + +```powershell +python conflict_catcher.py +``` + +Then open the printed local URL. By default it is: + +```text +http://127.0.0.1:8765 +``` + +## Branch Spec + +Use this format: + +```text +main str: + return "\n".join(part for part in (self.stdout, self.stderr) if part).strip() + + +def run_git(repo: Path, args: list[str], *, cwd: Path | None = None, timeout: int = 120) -> GitResult: + command = ["git", *args] + try: + completed = subprocess.run( + command, + cwd=str(cwd or repo), + text=True, + capture_output=True, + timeout=timeout, + check=False, + ) + except FileNotFoundError as exc: + raise CheckError("Git was not found on PATH. Install Git and try again.") from exc + except subprocess.TimeoutExpired as exc: + raise CheckError(f"Git command timed out: {' '.join(command)}") from exc + + return GitResult(completed.returncode, completed.stdout.strip(), completed.stderr.strip()) + + +def repo_root(path: str) -> Path: + if not path or not path.strip(): + raise CheckError("Choose a Git repository path.") + + candidate = Path(path).expanduser().resolve() + if not candidate.exists(): + raise CheckError(f"Path does not exist: {candidate}") + + result = run_git(candidate, ["rev-parse", "--show-toplevel"]) + if result.code != 0: + raise CheckError(f"Not a Git repository: {candidate}") + + return Path(result.stdout).resolve() + + +def parse_branch_spec(spec: str, target: str, source: str) -> tuple[str, str]: + spec = (spec or "").strip() + target = (target or "").strip() + source = (source or "").strip() + + if spec: + if "<" not in spec: + raise CheckError("Branch spec must look like main str: + result = run_git(repo, ["rev-parse", "--verify", f"{ref}^{{commit}}"]) + if result.code != 0: + raise CheckError(f"Could not find {label} branch/ref: {ref}") + return result.stdout + + +def list_conflicts(worktree: Path) -> list[dict[str, str]]: + result = run_git(worktree, ["diff", "--name-status", "--diff-filter=U"], cwd=worktree) + conflicts: list[dict[str, str]] = [] + if result.code != 0 or not result.stdout: + return conflicts + + status_names = { + "AA": "both added", + "AU": "added by us", + "UA": "added by them", + "DD": "both deleted", + "DU": "deleted by us", + "UD": "deleted by them", + "UU": "both modified", + "U": "unmerged", + } + for line in result.stdout.splitlines(): + parts = line.split("\t", 1) + if len(parts) == 2: + conflicts.append( + { + "status": parts[0], + "meaning": status_names.get(parts[0], "unmerged"), + "path": parts[1], + } + ) + return conflicts + + +def make_worktree(repo: Path, ref: str) -> Path: + temp_parent = Path(tempfile.mkdtemp(prefix="conflict-catcher-")) + worktree = temp_parent / "worktree" + result = run_git(repo, ["worktree", "add", "--detach", str(worktree), ref], timeout=180) + if result.code != 0: + shutil.rmtree(temp_parent, ignore_errors=True) + raise CheckError(f"Could not create a temporary worktree for {ref}.\n{result.combined}") + return worktree + + +def cleanup_worktree(repo: Path, worktree: Path) -> None: + run_git(repo, ["worktree", "remove", "--force", str(worktree)], timeout=180) + shutil.rmtree(worktree.parent, ignore_errors=True) + run_git(repo, ["worktree", "prune"], timeout=180) + + +def dry_run_merge(repo: Path, target: str, source: str) -> dict[str, Any]: + worktree = make_worktree(repo, target) + try: + result = run_git( + worktree, + ["merge", "--no-commit", "--no-ff", source], + cwd=worktree, + timeout=300, + ) + conflicts = list_conflicts(worktree) + ok = result.code == 0 and not conflicts + summary = ( + f"{source} can be merged into {target} without conflicts." + if ok + else f"{source} would conflict when merged into {target}." + ) + return { + "ok": ok, + "operation": "merge", + "summary": summary, + "conflicts": conflicts, + "gitOutput": result.combined, + } + finally: + run_git(worktree, ["merge", "--abort"], cwd=worktree) + cleanup_worktree(repo, worktree) + + +def dry_run_rebase(repo: Path, target: str, source: str) -> dict[str, Any]: + worktree = make_worktree(repo, source) + try: + result = run_git( + worktree, + ["rebase", "--no-verify", target], + cwd=worktree, + timeout=300, + ) + conflicts = list_conflicts(worktree) + ok = result.code == 0 and not conflicts + summary = ( + f"{source} can be rebased onto {target} without conflicts." + if ok + else f"{source} would conflict when rebased onto {target}." + ) + return { + "ok": ok, + "operation": "rebase", + "summary": summary, + "conflicts": conflicts, + "gitOutput": result.combined, + } + finally: + run_git(worktree, ["rebase", "--abort"], cwd=worktree) + cleanup_worktree(repo, worktree) + + +def check_conflicts(payload: dict[str, Any]) -> dict[str, Any]: + repo = repo_root(str(payload.get("repoPath", ""))) + target, source = parse_branch_spec( + str(payload.get("branchSpec", "")), + str(payload.get("targetBranch", "")), + str(payload.get("sourceBranch", "")), + ) + operation = str(payload.get("operation", "both")).lower() + if operation not in {"merge", "rebase", "both"}: + raise CheckError("Operation must be merge, rebase, or both.") + + target_sha = verify_commit(repo, target, "target") + source_sha = verify_commit(repo, source, "source") + + checks: list[dict[str, Any]] = [] + if operation in {"merge", "both"}: + checks.append(dry_run_merge(repo, target, source)) + if operation in {"rebase", "both"}: + checks.append(dry_run_rebase(repo, target, source)) + + return { + "repo": str(repo), + "target": target, + "source": source, + "targetSha": target_sha[:12], + "sourceSha": source_sha[:12], + "checks": checks, + } + + +INDEX_HTML = r""" + + + + + Conflict Catcher + + + +
+
+
+

Conflict Catcher

+

Dry-run branch merge and rebase checks before you touch the real working tree.

+
+ Git preflight +
+ +
+
+ + + +

Left side is the base/target. Right side is the incoming/source.

+ +
+ + +
+ + + + +
+ +
+
+
+

Results

+

Waiting for a repository and branch pair.

+
+ Idle +
+
+

Enter a repo path and a branch pair such as main<develop.

+
+
+
+
+ + + + +""" + + +class ConflictCatcherHandler(BaseHTTPRequestHandler): + server_version = "ConflictCatcher/1.0" + + def log_message(self, format: str, *args: Any) -> None: + sys.stderr.write("%s - %s\n" % (self.address_string(), format % args)) + + def send_json(self, status: int, payload: dict[str, Any]) -> None: + encoded = json.dumps(payload).encode("utf-8") + self.send_response(status) + self.send_header("Content-Type", "application/json; charset=utf-8") + self.send_header("Content-Length", str(len(encoded))) + self.end_headers() + self.wfile.write(encoded) + + def do_GET(self) -> None: + if self.path not in {"/", "/index.html"}: + self.send_error(404) + return + + encoded = INDEX_HTML.encode("utf-8") + self.send_response(200) + self.send_header("Content-Type", "text/html; charset=utf-8") + self.send_header("Content-Length", str(len(encoded))) + self.end_headers() + self.wfile.write(encoded) + + def do_POST(self) -> None: + if self.path != "/api/check": + self.send_error(404) + return + + try: + length = int(self.headers.get("Content-Length", "0")) + payload = json.loads(self.rfile.read(length).decode("utf-8")) + self.send_json(200, check_conflicts(payload)) + except CheckError as exc: + self.send_json(400, {"error": str(exc)}) + except json.JSONDecodeError: + self.send_json(400, {"error": "Request body must be valid JSON."}) + except Exception as exc: + self.send_json(500, {"error": f"Unexpected error: {exc}"}) + + +def run_server(host: str, port: int, open_browser: bool) -> None: + server = ThreadingHTTPServer((host, port), ConflictCatcherHandler) + url = f"http://{host}:{server.server_port}" + if open_browser: + threading.Timer(0.4, lambda: webbrowser.open(url)).start() + print(f"{APP_TITLE} running at {url}") + print("Press Ctrl+C to stop.") + server.serve_forever() + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description="Check Git branch conflicts before merging or rebasing.") + parser.add_argument("--host", default=os.environ.get("CONFLICT_CATCHER_HOST", DEFAULT_HOST)) + parser.add_argument("--port", type=int, default=int(os.environ.get("CONFLICT_CATCHER_PORT", DEFAULT_PORT))) + parser.add_argument("--no-browser", action="store_true", help="Start the server without opening a browser.") + return parser + + +if __name__ == "__main__": + args = build_parser().parse_args() + try: + run_server(args.host, args.port, not args.no_browser) + except KeyboardInterrupt: + print("\nStopped.") diff --git a/cc/test_conflict_catcher.py b/cc/test_conflict_catcher.py new file mode 100644 index 0000000..f3713d9 --- /dev/null +++ b/cc/test_conflict_catcher.py @@ -0,0 +1,64 @@ +import tempfile +import unittest +from pathlib import Path + +from conflict_catcher import CheckError, check_conflicts, parse_branch_spec, run_git + + +def git(repo: Path, *args: str) -> None: + result = run_git(repo, list(args)) + if result.code != 0: + raise AssertionError(result.combined) + + +def write(path: Path, text: str) -> None: + path.write_text(text, encoding="utf-8") + + +class ConflictCatcherTests(unittest.TestCase): + def test_parse_branch_spec(self) -> None: + self.assertEqual(parse_branch_spec("main None: + with tempfile.TemporaryDirectory() as temp: + repo = Path(temp) + git(repo, "init", "-b", "main") + git(repo, "config", "user.email", "test@example.com") + git(repo, "config", "user.name", "Test User") + write(repo / "note.txt", "base\n") + git(repo, "add", "note.txt") + git(repo, "commit", "-m", "base") + + git(repo, "checkout", "-b", "develop") + write(repo / "note.txt", "develop\n") + git(repo, "commit", "-am", "develop edit") + + git(repo, "checkout", "main") + write(repo / "note.txt", "main\n") + git(repo, "commit", "-am", "main edit") + + result = check_conflicts( + { + "repoPath": str(repo), + "branchSpec": "main