diff --git a/cc/conflict_catcher.py b/cc/conflict_catcher.py index b146494..7cb975d 100644 --- a/cc/conflict_catcher.py +++ b/cc/conflict_catcher.py @@ -3,6 +3,7 @@ from __future__ import annotations import argparse import json import os +import shlex import shutil import subprocess import sys @@ -13,6 +14,7 @@ from dataclasses import dataclass from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from typing import Any +from urllib.parse import urlparse APP_TITLE = "Conflict Catcher" @@ -95,6 +97,137 @@ def verify_commit(repo: Path, ref: str, label: str) -> str: return result.stdout +def git_lines(repo: Path, args: list[str]) -> list[str]: + result = run_git(repo, args) + if result.code != 0: + raise CheckError(result.combined or f"Git command failed: {' '.join(args)}") + return [line.strip() for line in result.stdout.splitlines() if line.strip()] + + +def list_branches(repo: Path) -> dict[str, Any]: + current = run_git(repo, ["branch", "--show-current"]).stdout.strip() + local = git_lines(repo, ["branch", "--format=%(refname:short)"]) + remote = [ + branch + for branch in git_lines(repo, ["branch", "--remotes", "--format=%(refname:short)"]) + if "/" in branch and not branch.endswith("/HEAD") + ] + return { + "repo": str(repo), + "current": current, + "local": local, + "remote": remote, + "all": sorted(dict.fromkeys([*local, *remote])), + } + + +def fetch_updates(repo: Path) -> dict[str, Any]: + result = run_git(repo, ["fetch", "--all", "--prune"], timeout=300) + if result.code != 0: + raise CheckError(f"Fetch failed.\n{result.combined}") + return { + "ok": True, + "summary": "Fetched all remotes and pruned stale remote-tracking branches.", + "gitOutput": result.combined, + } + + +def upstream_status(repo: Path, ref: str) -> dict[str, Any]: + upstream = run_git(repo, ["rev-parse", "--abbrev-ref", f"{ref}@{{upstream}}"]) + if upstream.code != 0 or not upstream.stdout.strip(): + return { + "ref": ref, + "upstream": "", + "ahead": None, + "behind": None, + "summary": f"{ref} has no configured upstream.", + "state": "unknown", + } + + upstream_ref = upstream.stdout.strip() + counts = run_git(repo, ["rev-list", "--left-right", "--count", f"{ref}...{upstream_ref}"]) + if counts.code != 0: + return { + "ref": ref, + "upstream": upstream_ref, + "ahead": None, + "behind": None, + "summary": f"Could not compare {ref} with {upstream_ref}.", + "state": "unknown", + } + + left, right = (counts.stdout.split() + ["0", "0"])[:2] + ahead = int(left) + behind = int(right) + if ahead == 0 and behind == 0: + summary = f"{ref} is up to date with {upstream_ref}." + state = "ok" + elif behind: + summary = f"{ref} is {behind} commit{'s' if behind != 1 else ''} behind {upstream_ref}." + state = "behind" + else: + summary = f"{ref} is {ahead} commit{'s' if ahead != 1 else ''} ahead of {upstream_ref}." + state = "ahead" + + return { + "ref": ref, + "upstream": upstream_ref, + "ahead": ahead, + "behind": behind, + "summary": summary, + "state": state, + } + + +def compare_freshness(repo: Path, target: str, source: str) -> list[dict[str, Any]]: + return [upstream_status(repo, target), upstream_status(repo, source)] + + +def resolve_repo_file(repo: Path, relative_path: str) -> Path: + if not relative_path or not str(relative_path).strip(): + raise CheckError("Choose a file path.") + candidate = (repo / relative_path).resolve() + try: + candidate.relative_to(repo) + except ValueError as exc: + raise CheckError("File path must stay inside the selected repository.") from exc + return candidate + + +def reveal_path(payload: dict[str, Any]) -> dict[str, Any]: + repo = repo_root(str(payload.get("repoPath", ""))) + target = resolve_repo_file(repo, str(payload.get("path", ""))) + reveal_target = target if target.exists() else target.parent + if sys.platform.startswith("win"): + subprocess.Popen(["explorer", "/select,", str(reveal_target)]) + else: + opener = "open" if sys.platform == "darwin" else "xdg-open" + subprocess.Popen([opener, str(reveal_target.parent if reveal_target.is_file() else reveal_target)]) + return {"ok": True, "path": str(target)} + + +def open_in_editor(payload: dict[str, Any]) -> dict[str, Any]: + repo = repo_root(str(payload.get("repoPath", ""))) + target = resolve_repo_file(repo, str(payload.get("path", ""))) + editor_command = str(payload.get("editorCommand", "")).strip() + + if editor_command: + parts = shlex.split(editor_command, posix=not sys.platform.startswith("win")) + if not parts: + raise CheckError("Editor command is empty.") + command = [part.replace("{path}", str(target)) for part in parts] + if not any("{path}" in part for part in parts): + command.append(str(target)) + subprocess.Popen(command, cwd=str(repo)) + elif sys.platform.startswith("win"): + os.startfile(str(target)) # type: ignore[attr-defined] + else: + opener = "open" if sys.platform == "darwin" else "xdg-open" + subprocess.Popen([opener, str(target)]) + + return {"ok": True, "path": str(target)} + + 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]] = [] @@ -207,8 +340,13 @@ def check_conflicts(payload: dict[str, Any]) -> dict[str, Any]: if operation not in {"merge", "rebase", "both"}: raise CheckError("Operation must be merge, rebase, or both.") + fetch_result = None + if bool(payload.get("fetchBeforeCheck")): + fetch_result = fetch_updates(repo) + target_sha = verify_commit(repo, target, "target") source_sha = verify_commit(repo, source, "source") + freshness = compare_freshness(repo, target, source) checks: list[dict[str, Any]] = [] if operation in {"merge", "both"}: @@ -222,10 +360,49 @@ def check_conflicts(payload: dict[str, Any]) -> dict[str, Any]: "source": source, "targetSha": target_sha[:12], "sourceSha": source_sha[:12], + "fetch": fetch_result, + "freshness": freshness, "checks": checks, } +def generate_report(data: dict[str, Any]) -> str: + checks = data.get("checks", []) + has_conflict = any(not check.get("ok") for check in checks) + lines = [ + "# Conflict Catcher Report", + "", + f"- Repository: `{data.get('repo', '')}`", + f"- Target/base: `{data.get('target', '')}` (`{data.get('targetSha', '')}`)", + f"- Source/incoming: `{data.get('source', '')}` (`{data.get('sourceSha', '')}`)", + f"- Overall: {'conflicts need review' if has_conflict else 'clean'}", + "", + "## Freshness", + "", + ] + for item in data.get("freshness", []): + lines.append(f"- {item.get('summary', '')}") + + lines.extend(["", "## Checks", ""]) + for check in checks: + lines.append(f"### {str(check.get('operation', '')).title()}") + lines.append("") + lines.append(f"- Status: {'clean' if check.get('ok') else 'conflicts'}") + lines.append(f"- Summary: {check.get('summary', '')}") + conflicts = check.get("conflicts", []) + if conflicts: + lines.extend(["", "| Status | Meaning | Path |", "| --- | --- | --- |"]) + for conflict in conflicts: + lines.append( + f"| {conflict.get('status', '')} | {conflict.get('meaning', '')} | `{conflict.get('path', '')}` |" + ) + else: + lines.append("- Conflict files: none") + lines.append("") + + return "\n".join(lines).strip() + "\n" + + INDEX_HTML = r""" @@ -235,17 +412,26 @@ INDEX_HTML = r"""
-
-
-

Conflict Catcher

-

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

+
+ Temporary worktrees only + + +
+
+

Check merge and rebase risk before your real branch gets involved.

+

Run clean-room Git dry runs, review exactly which files would conflict, then go back to your normal workflow with the answer in hand.

+
+
+
No checkoutYour active working tree stays untouched.
+
No commitDry runs abort and clean up after themselves.
+
Both pathsTest merge, rebase, or both in one pass.
+
+
-
- +
+ +
+
+

Preflight setup

+

Point at a repo and name the branch direction.

+
+ Ready +
- -

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

+
+ +
+ + + +
+ -
- - -
+ +

Use target<source. You can also fill the separate fields below if that is clearer.

- +
+ + +
- - + + +
+ Check type +
+ + + +
+
+ + +
+ + + +
@@ -528,11 +1343,24 @@ INDEX_HTML = r""" Idle
-

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

+
+ +
+

Ready when your branch pair is.

+

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

+
+
+
+ @@ -648,7 +1806,8 @@ class ConflictCatcherHandler(BaseHTTPRequestHandler): self.wfile.write(encoded) def do_GET(self) -> None: - if self.path not in {"/", "/index.html"}: + path = urlparse(self.path).path + if path not in {"/", "/index.html"}: self.send_error(404) return @@ -660,14 +1819,23 @@ class ConflictCatcherHandler(BaseHTTPRequestHandler): self.wfile.write(encoded) def do_POST(self) -> None: - if self.path != "/api/check": + path = urlparse(self.path).path + handlers = { + "/api/check": check_conflicts, + "/api/branches": lambda payload: list_branches(repo_root(str(payload.get("repoPath", "")))), + "/api/report": lambda payload: {"markdown": generate_report(payload)}, + "/api/reveal": reveal_path, + "/api/open-editor": open_in_editor, + } + handler = handlers.get(path) + if handler is None: 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)) + self.send_json(200, handler(payload)) except CheckError as exc: self.send_json(400, {"error": str(exc)}) except json.JSONDecodeError: diff --git a/cc/test_conflict_catcher.py b/cc/test_conflict_catcher.py index f3713d9..7d271df 100644 --- a/cc/test_conflict_catcher.py +++ b/cc/test_conflict_catcher.py @@ -2,7 +2,15 @@ import tempfile import unittest from pathlib import Path -from conflict_catcher import CheckError, check_conflicts, parse_branch_spec, run_git +from conflict_catcher import ( + CheckError, + check_conflicts, + compare_freshness, + generate_report, + list_branches, + parse_branch_spec, + run_git, +) def git(repo: Path, *args: str) -> None: @@ -59,6 +67,56 @@ class ConflictCatcherTests(unittest.TestCase): ) ) + def test_lists_branches_and_reports_freshness(self) -> None: + with tempfile.TemporaryDirectory() as temp: + origin = Path(temp) / "origin.git" + repo = Path(temp) / "repo" + + git(Path(temp), "init", "--bare", str(origin)) + git(Path(temp), "clone", str(origin), str(repo)) + git(repo, "checkout", "-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, "push", "-u", "origin", "main") + git(repo, "checkout", "-b", "develop") + + branches = list_branches(repo) + self.assertEqual(branches["current"], "develop") + self.assertIn("main", branches["local"]) + self.assertIn("develop", branches["local"]) + self.assertIn("origin/main", branches["remote"]) + + freshness = compare_freshness(repo, "main", "develop") + self.assertEqual(freshness[0]["state"], "ok") + self.assertEqual(freshness[1]["state"], "unknown") + + def test_generates_markdown_report(self) -> None: + report = generate_report( + { + "repo": "C:/repo", + "target": "main", + "source": "feature", + "targetSha": "abc123", + "sourceSha": "def456", + "freshness": [{"summary": "main is up to date with origin/main."}], + "checks": [ + { + "ok": False, + "operation": "merge", + "summary": "feature would conflict when merged into main.", + "conflicts": [{"status": "UU", "meaning": "both modified", "path": "note.txt"}], + } + ], + } + ) + + self.assertIn("# Conflict Catcher Report", report) + self.assertIn("Target/base: `main`", report) + self.assertIn("| UU | both modified | `note.txt` |", report) + if __name__ == "__main__": unittest.main()