Enhance get_status method and bump version to 1.2.2

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "uv_build"
 
 [project]
 name = "gitdirector"
-version = "1.2.0"
+version = "1.2.2"
 description = "A terminal based control plane for developers working across multiple repositories. Launch multiple AI coding agents, multiple tmux sessions and track changes across all your repos in one place."
 readme = "README.md"
 license = { text = "MIT" }

src/gitdirector/manager.py

@@ -131,11 +131,11 @@ def _discover_and_remove(self, root: Path) -> Tuple[bool, str, List[Path]]:
         except Exception as e:
             return False, f"Error removing repositories: {str(e)}", []
 
-    def get_repository_status(self, path: Path) -> RepositoryInfo:
+    def get_repository_status(self, path: Path, *, fetch: bool = False) -> RepositoryInfo:
         if path.exists() and (path / ".git").is_dir():
             try:
                 repo = Repository(path)
-                return repo.get_status()
+                return repo.get_status(fetch=fetch)
             except Exception as e:
                 return RepositoryInfo(path, path.name, RepoStatus.UNKNOWN, None, str(e))
         return RepositoryInfo(

src/gitdirector/repo.py

@@ -125,54 +125,79 @@ def get_tracked_size(self) -> Optional[int]:
                     pass
         return total
 
-    def get_status(self) -> RepositoryInfo:
-        branch = self.get_current_branch()
-
-        code, out, err = self._run_git("fetch")
-        if code != 0:
-            return RepositoryInfo(self.path, self.name, RepoStatus.UNKNOWN, branch, err)
-
-        code, ahead_behind, _ = self._run_git("rev-list", "--left-right", "--count", "@{u}...HEAD")
-
+    def get_status(self, *, fetch: bool = False) -> RepositoryInfo:
+        if fetch:
+            code, _, err = self._run_git("fetch")
+            if code != 0:
+                branch = self.get_current_branch()
+                return RepositoryInfo(self.path, self.name, RepoStatus.UNKNOWN, branch, err)
+
+        code, out, _ = self._run_git("status", "--porcelain=v2", "--branch", _strip=False)
         if code != 0:
             return RepositoryInfo(
-                self.path, self.name, RepoStatus.UNKNOWN, branch, "No tracking branch"
+                self.path, self.name, RepoStatus.UNKNOWN, None, "git status failed"
             )
 
-        try:
-            behind, ahead = map(int, ahead_behind.split())
-            if ahead > 0 and behind > 0:
-                status = RepoStatus.DIVERGED
-                msg = f"ahead {ahead}, behind {behind}"
-            elif ahead > 0:
-                status = RepoStatus.AHEAD
-                msg = f"ahead {ahead}"
-            elif behind > 0:
-                status = RepoStatus.BEHIND
-                msg = f"behind {behind}"
-            else:
-                status = RepoStatus.UP_TO_DATE
-                msg = ""
-        except ValueError:
-            status = RepoStatus.UNKNOWN
-            msg = "Could not parse git status"
-
-        code, porcelain, _ = self._run_git("status", "--porcelain", _strip=False)
+        branch = None
+        has_upstream = False
+        ahead = 0
+        behind = 0
         staged = False
         unstaged = False
         staged_files: list[str] = []
         unstaged_files: list[str] = []
-        if code == 0 and porcelain:
-            for line in porcelain.splitlines():
-                if len(line) >= 2:
-                    x, y = line[0], line[1]
-                    filename = line[3:].strip()
-                    if x not in (" ", "?"):
-                        staged = True
-                        staged_files.append(filename)
-                    if y not in (" ", "?"):
-                        unstaged = True
-                        unstaged_files.append(filename)
+
+        for line in out.splitlines():
+            if line.startswith("# branch.head "):
+                branch = line[14:]
+                if branch == "(detached)":
+                    branch = None
+            elif line.startswith("# branch.ab "):
+                has_upstream = True
+                parts = line.split()
+                try:
+                    ahead = int(parts[2].lstrip("+"))
+                    behind = abs(int(parts[3]))
+                except (IndexError, ValueError):
+                    pass
+            elif line.startswith("1 ") or line.startswith("2 "):
+                xy = line[2:4]
+                x, y = xy[0], xy[1]
+                if line.startswith("1 "):
+                    parts = line.split(" ", 8)
+                    filename = parts[8] if len(parts) > 8 else ""
+                else:
+                    parts = line.split(" ", 9)
+                    filename = parts[9].split("\t")[0] if len(parts) > 9 else ""
+                if x not in (".", "?"):
+                    staged = True
+                    staged_files.append(filename)
+                if y not in (".", "?"):
+                    unstaged = True
+                    unstaged_files.append(filename)
+            elif line.startswith("u "):
+                parts = line.split(" ", 10)
+                filename = parts[10] if len(parts) > 10 else ""
+                staged = True
+                unstaged = True
+                staged_files.append(filename)
+                unstaged_files.append(filename)
+
+        if not has_upstream:
+            status = RepoStatus.UNKNOWN
+            msg = "No tracking branch"
+        elif ahead > 0 and behind > 0:
+            status = RepoStatus.DIVERGED
+            msg = f"ahead {ahead}, behind {behind}"
+        elif ahead > 0:
+            status = RepoStatus.AHEAD
+            msg = f"ahead {ahead}"
+        elif behind > 0:
+            status = RepoStatus.BEHIND
+            msg = f"behind {behind}"
+        else:
+            status = RepoStatus.UP_TO_DATE
+            msg = ""
 
         last_updated, last_commit_ts = self.get_last_commit_info()
         size = self.get_tracked_size()

tests/test_extras.py

@@ -272,39 +272,37 @@ class TestRepoGetStatusParsingErrors:
     """Test error paths when parsing git status output."""
 
     def test_get_status_invalid_ahead_behind_format(self, fake_git_repo, mocker):
-        """When rev-list output can't be parsed as two ints, returns UNKNOWN."""
+        """When branch.ab line has invalid format, ahead/behind default to 0."""
 
-        # Mock all subprocess.run calls
         def mock_run(*args, **kwargs):
-            # Check which git command this is
             git_cmd = args[0]
-            if "fetch" in git_cmd:
-                return MagicMock(returncode=0, stdout="", stderr="")
-            elif "rev-list" in git_cmd:
-                return MagicMock(returncode=0, stdout="invalid format\n", stderr="")
+            if "status" in git_cmd:
+                v2 = "# branch.oid abc\n# branch.head main\n"
+                v2 += "# branch.upstream origin/main\n"
+                v2 += "# branch.ab invalid format\n"
+                return MagicMock(returncode=0, stdout=v2, stderr="")
             return MagicMock(returncode=0, stdout="", stderr="")
 
         mocker.patch("subprocess.run", side_effect=mock_run)
         repo = Repository(fake_git_repo)
         status = repo.get_status()
-        assert status.status == RepoStatus.UNKNOWN
+        assert status.status == RepoStatus.UP_TO_DATE
 
     def test_get_status_no_tracking_branch(self, fake_git_repo, mocker):
         """When no tracking branch exists, returns UNKNOWN."""
 
         def mock_run(*args, **kwargs):
             git_cmd = args[0]
-            if "fetch" in git_cmd:
-                return MagicMock(returncode=0, stdout="", stderr="")
-            elif "rev-list" in git_cmd:
-                return MagicMock(returncode=128, stdout="", stderr="fatal: no upstream branch\n")
+            if "status" in git_cmd:
+                v2 = "# branch.oid abc\n# branch.head main\n"
+                return MagicMock(returncode=0, stdout=v2, stderr="")
             return MagicMock(returncode=0, stdout="", stderr="")
 
         mocker.patch("subprocess.run", side_effect=mock_run)
         repo = Repository(fake_git_repo)
         status = repo.get_status()
         assert status.status == RepoStatus.UNKNOWN
-        assert "upstream" in status.message.lower() or "tracking" in status.message.lower()
+        assert "tracking" in status.message.lower()
 
 
 # ---------------------------------------------------------------------------

tests/test_repo.py

@@ -178,15 +178,29 @@ def side_effect(cmd, **kwargs):
         git_args = args[1:]  # strip the repo path
         calls.append(git_args)
 
-        if "rev-parse" in git_args:
-            return _make_run_result(0, f"{branch}\n", "")
         if "fetch" in git_args:
             return _make_run_result(0 if fetch_ok else 1, "", "" if fetch_ok else "fetch error")
-        if "rev-list" in git_args:
-            code = 0 if ahead_behind else 1
-            return _make_run_result(code, ahead_behind + "\n" if ahead_behind else "", "")
         if "status" in git_args:
-            return _make_run_result(0, porcelain, "")
+            v2 = f"# branch.oid abc123\n# branch.head {branch}\n"
+            if ahead_behind:
+                try:
+                    behind_val, ahead_val = ahead_behind.split("\t")
+                    v2 += f"# branch.upstream origin/{branch}\n"
+                    v2 += f"# branch.ab +{ahead_val} -{behind_val}\n"
+                except ValueError:
+                    pass
+            if porcelain:
+                for line in porcelain.splitlines():
+                    if len(line) >= 2:
+                        x, y = line[0], line[1]
+                        filename = line[3:].strip() if len(line) > 3 else ""
+                        if x == "?" and y == "?":
+                            v2 += f"? {filename}\n"
+                        else:
+                            v2_x = x if x != " " else "."
+                            v2_y = y if y != " " else "."
+                            v2 += f"1 {v2_x}{v2_y} N... 100644 100644 100644 abc def {filename}\n"
+            return _make_run_result(0, v2, "")
         if "log" in git_args:
             return _make_run_result(0, "5 minutes ago\n", "")
         if "ls-files" in git_args:
@@ -224,7 +238,7 @@ def test_diverged(self, fake_git_repo, mocker):
 
     def test_fetch_failure(self, fake_git_repo, mocker):
         _setup_status_mocks(mocker, fetch_ok=False)
-        info = Repository(fake_git_repo).get_status()
+        info = Repository(fake_git_repo).get_status(fetch=True)
         assert info.status == RepoStatus.UNKNOWN
 
     def test_no_tracking_branch(self, fake_git_repo, mocker):