Bugfix local commits on remote deleted branch

Author: Joschi3

scripts/hooks/update/50.workspace.py

@@ -1,36 +1,40 @@
 #!/usr/bin/env python3
 """
-This script updates all git repositories under the *src* directory of the workspace.
-- it fetches all remotes
-- it pulls the current branch (if it has an upstream)
-- it detects branches that are deleted on the remote and have no new commits
-    - it offers to delete those branches
-- it runs in parallel for all repositories 
+Update every git repository under the *src* directory of a ROS 2 workspace.
+
+* fetches & prunes all remotes
+* pulls the current branch (if it has an upstream)
+* detects local branches whose upstream was deleted **and** have no commits
+  unknown to any remote – offers to delete them
+* performs the heavy work in parallel, prints results sequentially
 """
 from __future__ import annotations
 
 import os
 import signal
 import subprocess
-from concurrent.futures import ThreadPoolExecutor, as_completed
-from pathlib import Path
-from typing import List
 import time
 import shutil
-from tuda_workspace_scripts.print import *
+from concurrent.futures import ThreadPoolExecutor, as_completed
+from pathlib import Path
+from typing import List, Tuple
+
+from tuda_workspace_scripts.print import *  # TU DA helper
 from tuda_workspace_scripts.workspace import get_workspace_root
 
 try:
     import git
 except ImportError:
     print(
-        "GitPython is required! Install using 'pip3 install --user gitpython' or 'apt install python3-git'"
+        "GitPython is required! Install with 'pip3 install --user gitpython' or "
+        "'apt install python3-git'"
     )
     raise
 
 
+# ───────────────────────── helpers ──────────────────────────────────────────────
 def launch_subprocess(cmd: list[str] | tuple[str, ...], cwd: str | Path):
-    """Run *cmd* in *cwd* forwarding *SIGINT* to the child process group."""
+    """Run *cmd* in *cwd*, forwarding Ctrl-C to the child process group."""
     try:
         return subprocess.run(
             cmd,
@@ -42,45 +46,51 @@ def launch_subprocess(cmd: list[str] | tuple[str, ...], cwd: str | Path):
             preexec_fn=os.setpgrp,
         )
     except KeyboardInterrupt:
-        # propagate Ctrl‑C to the subprocess group so *git* exits cleanly
         os.killpg(0, signal.SIGINT)
         raise
 
 
-def _is_deleted_branch(repo: "git.Repo", branch: "git.Head") -> bool:
-    """Return *True* if *branch* was deleted upstream and has no unique commits."""
+def _has_unpushed_commits(repo: git.Repo, branch_name: str) -> bool:
+    """True iff *branch_name* contains commits unknown to **any** remote."""
     try:
-        tracking_branch = branch.tracking_branch()
-        if tracking_branch is None:
-            return False  # not tracking → ignore
-
-        for remote in repo.remotes:
-            if remote.name == tracking_branch.remote_name:
-                if (
-                    tracking_branch in remote.refs
-                    and tracking_branch not in remote.stale_refs
-                ):
-                    # upstream still exists & not stale
-                    return False
-                break
-    except (git.exc.GitCommandError, Exception) as exc:
-        print_error(
-            f"{os.path.basename(repo.working_tree_dir)} has error on branch {branch.name}: {exc}"
+        cnt = int(
+            repo.git.rev_list("--count", branch_name, "--not", "--remotes").strip()
+            or "0"
         )
+        return cnt > 0
+    except git.exc.GitCommandError:
         return False
 
-    # Upstream appears gone – do we have local commits not pushed anywhere?
+
+def _is_deleted_branch(repo: git.Repo, branch: git.Head) -> Tuple[bool, str | None]:
+    """
+    Returns (deletable, warning)
+
+    * deletable → upstream vanished **and** branch is fully merged everywhere.
+    * warning   → explanatory message when *not* deletable (None if none).
+    """
+    tracking = branch.tracking_branch()
+    if tracking is None:
+        return False, None
+
     try:
-        if any(True for _ in repo.iter_commits("{0}@{{u}}..{0}".format(branch.name))):
-            print_warn(
-                f"Branch {branch.name} seems to be deleted on remote but still has commits that were not pushed."
-            )
-            return False
-    except git.exc.GitCommandError:
-        pass  # Ignore error if branch is not tracking anything
-    return True
+        remote = repo.remotes[tracking.remote_name]
+        if tracking in remote.refs:  # upstream still exists
+            return False, None
+    except (IndexError, ValueError):  # remote itself lost
+        return False, None
+
+    if _has_unpushed_commits(repo, branch.name):
+        warn = (
+            f"Branch {branch.name} was deleted on the remote but still has "
+            "commits that are not present on any remote."
+        )
+        return False, warn
 
+    return True, None
 
+
+# ───────────────────────── result container ─────────────────────────────────────
 class RepoResult:
     __slots__ = (
         "path",
@@ -89,6 +99,7 @@ class RepoResult:
         "pull_attempted",
         "pull_ok",
         "deletable",
+        "warnings",
         "stdout",
         "stderr",
         "error",
@@ -102,44 +113,48 @@ def __init__(
         pull_attempted: bool,
         pull_ok: bool,
         deletable: List[str],
+        warnings: List[str],
         stdout: str,
         stderr: str,
         error: str | None = None,
-    ) -> None:
+    ):
         self.path = path
         self.branch = branch
         self.fetch_ok = fetch_ok
         self.pull_attempted = pull_attempted
         self.pull_ok = pull_ok
         self.deletable = deletable
+        self.warnings = warnings
         self.stdout = stdout
         self.stderr = stderr
         self.error = error
 
 
+# ───────────────────────── worker (parallel) ────────────────────────────────────
 def process_repo(repo_path: Path) -> RepoResult:
-    """Fetch + optional pull + stale‑branch detection (runs in a worker)."""
+    """Fetch, optional pull, stale-branch detection – runs in a thread."""
     try:
         repo = git.Repo(repo_path)
 
         if not repo.head.is_valid():
-            return RepoResult(repo_path, "no-HEAD", True, False, True, [], "", "", None)
+            return RepoResult(
+                repo_path, "no-HEAD", True, False, True, [], [], "", "", None
+            )
 
         branch_name = (
             f"detached@{repo.head.commit.hexsha[:7]}"
             if repo.head.is_detached
             else repo.head.ref.name
         )
 
-        # -- always fetch/prune first so other branches are updated --------- #
+        # fetch & prune
         fetch = launch_subprocess(["git", "fetch", "--all", "--prune"], cwd=repo_path)
         fetch_ok = fetch.returncode == 0
 
-        # ------------------------------------------------------------------ #
+        # pull current branch (fast-forward only)
         pull_attempted = False
-        pull_ok = True  # will stay True when pull is skipped
-        pull_out, pull_err = "", ""
-
+        pull_ok = True
+        pull_out = pull_err = ""
         if not repo.head.is_detached:
             upstream = repo.head.ref.tracking_branch()
             if upstream is not None and upstream in repo.refs:
@@ -149,15 +164,16 @@ def process_repo(repo_path: Path) -> RepoResult:
                 pull_out = pull.stdout or ""
                 pull_err = pull.stderr or ""
 
-        # deletable branches – only when fetch succeeded so refs are current
+        # stale-branch detection
         deletable: List[str] = []
+        warnings: List[str] = []
         if fetch_ok:
-            deletable = [
-                br.name for br in repo.branches if _is_deleted_branch(repo, br)
-            ]
-
-        stdout_combined = (fetch.stdout or "") + pull_out
-        stderr_combined = (fetch.stderr or "") + pull_err
+            for br in repo.branches:
+                can_del, warn = _is_deleted_branch(repo, br)
+                if can_del:
+                    deletable.append(br.name)
+                if warn:
+                    warnings.append(warn)
 
         return RepoResult(
             path=repo_path,
@@ -166,25 +182,28 @@ def process_repo(repo_path: Path) -> RepoResult:
             pull_attempted=pull_attempted,
             pull_ok=pull_ok,
             deletable=deletable,
-            stdout=stdout_combined,
-            stderr=stderr_combined,
+            warnings=warnings,
+            stdout=(fetch.stdout or "") + pull_out,
+            stderr=(fetch.stderr or "") + pull_err,
         )
 
-    except Exception as exc:  # broad catch for isolation between threads
-        return RepoResult(repo_path, "?", False, False, False, [], "", "", str(exc))
+    except Exception as exc:  # keep other repos going
+        return RepoResult(repo_path, "?", False, False, False, [], [], "", "", str(exc))
 
 
+# ───────────────────────── discovery ────────────────────────────────────────────
 def collect_repos(ws_src: Path) -> List[Path]:
-    """Return absolute paths of *top‑level* git work‑trees under *ws_src*."""
+    """Return absolute paths of *top-level* git repos under ws_src."""
     repos: List[Path] = []
     for root, dirs, _ in os.walk(ws_src):
         root_p = Path(root)
         if (root_p / ".git").is_dir():
             repos.append(root_p)
-            dirs[:] = []  # prune recursion into this repo
+            dirs[:] = []  # don’t recurse into repo
     return repos
 
 
+# ───────────────────────── main ────────────────────────────────────────────────
 def update(**_) -> bool:
     ws_root = get_workspace_root()
     if ws_root is None:
@@ -199,15 +218,13 @@ def update(**_) -> bool:
         print_info("No git repositories found.")
         return True
 
-    # ----------------------- parallel phase ------------------------------- #
+    # ───── parallel phase ────────────────────────────────────────────────────
     total = len(repos)
-
     _BAR_START = time.monotonic()
 
     def _progress(idx: int):
-        """Draw a simple progress bar that lives on one terminal line."""
         cols = shutil.get_terminal_size((80, 20)).columns
-        bar_len = max(10, min(50, cols - 30))  # leave space for counters & percent
+        bar_len = max(10, min(50, cols - 30))
         filled = int(bar_len * idx / total)
         bar = (
             ("=" * filled + ">" + " " * (bar_len - filled - 1))
@@ -240,7 +257,7 @@ def _progress(idx: int):
     finally:
         print()  # newline after progress bar
 
-    # ----------------------- sequential phase ----------------------------- #
+    # ───── sequential phase ─────────────────────────────────────────────────
     overall_ok = True
     for res in sorted(results, key=lambda r: r.path):
         rel = res.path.relative_to(ws_src)
@@ -251,15 +268,15 @@ def _progress(idx: int):
             overall_ok = False
             continue
 
-        # -- fetch status --------------------------------------------------- #
+        # fetch status
         if not res.fetch_ok:
             print_error("git fetch failed – repository might be out of date:")
             if res.stderr.strip():
                 print(res.stderr.rstrip())
             overall_ok = False
-            continue  # pull & deletion checks rely on fresh refs
+            continue
 
-        # -- pull status ---------------------------------------------------- #
+        # pull status
         if res.pull_attempted:
             if not res.pull_ok:
                 print_error("git pull failed:")
@@ -272,10 +289,15 @@ def _progress(idx: int):
         else:
             print_info("skipped pull – current branch has no upstream")
 
-        # -- stale branches ------------------------------------------------- #
+        # branch-specific warnings
+        for msg in res.warnings:
+            print_warn(msg)
+
+        # candidate branches for deletion
         if res.deletable:
             msg = (
-                "The following local branches are deleted on the remote and have no extra commits:\n"
+                "The following local branches are deleted on the remote and "
+                "have no extra commits:\n"
                 + "\n".join(f"  {b}" for b in res.deletable)
                 + "\nDelete them now?"
             )