From a44b96358f653a94bcc5b706320e23a3f8547a70 Mon Sep 17 00:00:00 2001 From: Zoran Simic Date: Mon, 25 May 2026 13:37:32 -0700 Subject: [PATCH 1/4] Added docs --- AGENTS.md | 54 +++++++++++++++++++++++++++++++++++++ docs/contributing.md | 54 +++++++++++++++++++++++++++++++++++++ docs/contributing.rst | 56 -------------------------------------- docs/design.md | 58 ++++++++++++++++++++++++++++++++++++++++ src/mgit/cli.py | 2 +- src/mgit/git.py | 62 ++++++++++++++++++++----------------------- 6 files changed, 196 insertions(+), 90 deletions(-) create mode 100644 AGENTS.md create mode 100644 docs/contributing.md delete mode 100644 docs/contributing.rst create mode 100644 docs/design.md diff --git a/AGENTS.md b/AGENTS.md new file mode 100644 index 0000000..fc5d552 --- /dev/null +++ b/AGENTS.md @@ -0,0 +1,54 @@ +# AGENTS.md + +Guidance for future Codex sessions working in this repository. + +## Project Shape + +`mgit` is a small Python CLI for inspecting and operating on one git checkout +or a workspace containing many git checkouts. The CLI uses argparse subcommands, +short aliases, separation between command dispatch, git state modeling, repo discovery, +and output rendering. + +Important files: + +- `src/mgit/cli.py`: current Click entry point, option parsing, and clean flows. +- `src/mgit/git.py`: git command runner, URL parsing, branch/status/config + parsers, reports, fetch/pull/clone helpers, and cleanable-branch detection. +- `tests/`: current pytest coverage, prefer exercising CLI in general rather than unit tests. + +## Development Commands + +Use these from the repo root: + +- `.venv/bin/pytest -q` for fast iteration. +- `.venv/bin/pytest -vv` when you want the full test names. +- `.venv/bin/pytest -vv --cov=src tests` for local coverage checks. +- `ruff check` and `ruff format --diff` directly; `ruff` is on `PATH`. +- `tox -e style` for the packaged style environment. +- `tox -e py39` for the minimum supported Python version. +- `tox -e py314` for the newest supported Python version. +- `tox` as the final confidence run; it exercises multiple Python versions, + coverage, and linters. + +`uv run pytest ...` also works when the synced environment is desired, but the +local `.venv/bin/pytest` path is usually faster while iterating. + +The repo uses `ruff`, `pyright`, `tox`, and `setuptools-scm`. + +## Working Rules + +- Prefer `rg` and `rg --files` for navigation. +- Preserve user changes. +- Use `apply_patch` for manual edits. +- Keep changes scoped. This repo is intentionally small. +- Do not add new runtime dependencies without consulting with user. + +## Safety Notes + +Plain status inspection is safe. Some current v1 code paths are mutating: + +- `fetch` runs `git fetch --all --prune`. +- `pull` runs `git pull --rebase`, guarded by pending-change checks. +- `groom` can delete one local and/or remote branch. + +Do not run destructive clean/reset paths casually while planning or testing. diff --git a/docs/contributing.md b/docs/contributing.md new file mode 100644 index 0000000..9d58410 --- /dev/null +++ b/docs/contributing.md @@ -0,0 +1,54 @@ +# Contributing + +Contributions are welcome! + +[tox](https://github.com/tox-dev/tox) is used for full test runs. Packaging +metadata lives in `pyproject.toml` and versions come from `setuptools-scm`. + +## Development + +To get going locally, simply do this: + +```console +git clone https://github.com/zsimic/mgit.git +cd mgit + +uv sync + +# You have a venv now in ./.venv +source .venv/bin/activate +which python +which mgit +mgit + +deactivate +``` + +## Running The Tests + +Fast local checks: + +```console +.venv/bin/pytest -q +ruff check +``` + +Full confidence check: + +```console +tox +``` + +Useful focused tox runs: + +- `tox -e py39` for the minimum supported Python version. + +- `tox -e py314` for the newest supported Python version. + +- `tox -e style` for packaged lint/type checks. + +- `tox -e docs` for a strict `README.rst` parse check. + +## Test Coverage + +Run `tox`, then open `.tox/test-reports/htmlcov/index.html`. diff --git a/docs/contributing.rst b/docs/contributing.rst deleted file mode 100644 index 7c1ca88..0000000 --- a/docs/contributing.rst +++ /dev/null @@ -1,56 +0,0 @@ -Contributions are welcome! - -tox_ is used for full test runs. Packaging metadata lives in -``pyproject.toml`` and versions come from ``setuptools-scm``. - -Development -=========== - -To get going locally, simply do this:: - - git clone https://github.com/zsimic/mgit.git - cd mgit - - uv sync - - # You have a venv now in ./.venv - source .venv/bin/activate - which python - which mgit - mgit - - deactivate - - -Running the tests -================= - -Fast local checks:: - - .venv/bin/pytest -q - ruff check - -Full confidence check:: - - tox - -Useful focused tox runs: - -* ``tox -e py39`` for the minimum supported Python version. - -* ``tox -e py314`` for the newest supported Python version. - -* ``tox -e style`` for packaged lint/type checks. - -* ``tox -e docs`` for a strict ``README.rst`` parse check. - - -Test coverage -============= - -Run ``tox``, then open ``.tox/test-reports/htmlcov/index.html`` - - -.. _pyenv: https://github.com/pyenv/pyenv - -.. _tox: https://github.com/tox-dev/tox diff --git a/docs/design.md b/docs/design.md new file mode 100644 index 0000000..04b665b --- /dev/null +++ b/docs/design.md @@ -0,0 +1,58 @@ +# Design Notes + +This document records implementation decisions and constraints that are useful +when changing `mgit`. It is intentionally broader than a user-facing +architecture overview: add notes here when they explain how a part of the CLI +should continue to evolve. + +## Cached Repository State + +`GitDir` is used for one command run and lazily captures git state that can +otherwise require repeated subprocess calls. + +- `GitDir.default_branch`: The default branch selected for the command run. + Once selected, it should remain stable while a command proceeds, including + while `groom` checks out, pulls, and deletes branches. A newly fetched + `origin/HEAD` could influence the initial selection if this property has not + yet been read. + +- `GitDir.status`: A `GitStatus` snapshot produced from + `git status --porcelain=v2 --branch`. It includes pending changes and + upstream ahead/behind information. + +- `GitDir.refs`: A `GitRefs` snapshot of the current branch, local branches, + configured remotes, remote-tracking branches, remote default branches, and + upstream configuration. + +- `GitRefs.orphan_branches`: Derived cached data owned by a particular + `GitRefs` snapshot. It is discarded naturally when `GitDir.refs` is + replaced, so it must not be invalidated separately. + +## Invalidation Requirements + +The operations below may run during one lifetime of a `GitDir` instance: + +| Operation | `status` | `refs` | Reason | +| --- | --- | --- | --- | +| `fetch_now()` | stale | stale | Ahead/behind and remote-tracking branches can change; pruning can alter orphan detection. | +| checkout | stale | stale | The checked-out branch and displayed status can change. | +| `pull()` | stale | stale | Worktree/status and remote-tracking refs may change, including when the pull fails after doing some work. | +| delete local branch | unchanged in the `groom` flow | stale | The local branch collection changes; deletion occurs after checkout. | +| delete remote branch | unchanged in the `groom` flow | stale | The remote-tracking branch collection changes. | + +`default_branch` is deliberately not listed as transient state to clear after +these operations: it is the command run's chosen target branch. Fetch should +occur before its first access in any flow that wants freshly observed remote +default-branch information. + +## Current Implementation Note + +`GitDir.clear_cached_state()` currently discards every cached property on +`GitDir`. This is broader than the invalidation requirements above and clears +`default_branch` unnecessarily. Also, `fetch_now()` currently updates `age` +without clearing an already materialized `status` or `refs` snapshot; existing +command flows rely on fetching before reading them. + +A follow-up refactor should replace the generic reset with explicit +invalidation of `status` and `refs` at the mutation points above, including +`fetch_now()`. diff --git a/src/mgit/cli.py b/src/mgit/cli.py index 01ca1f5..d2fa9c7 100644 --- a/src/mgit/cli.py +++ b/src/mgit/cli.py @@ -251,12 +251,12 @@ def run_single(self, git: GitDir): return git.status.require_clean("groom") - remote_cleanup = None local_cleanup = git.cleanable_local_branch(current_branch, include_current=True) if not local_cleanup: Reporter.abort(f"Branch {git.represented_branch(current_branch)} can't be cleaned") upstream = refs.upstreams.get(current_branch) + remote_cleanup = None if upstream and refs.has_remote_branch(upstream.remote, upstream.branch): remote_cleanup = git.cleanable_current_remote_branch() if not remote_cleanup: diff --git a/src/mgit/git.py b/src/mgit/git.py index 12c6af0..142de5d 100644 --- a/src/mgit/git.py +++ b/src/mgit/git.py @@ -252,6 +252,17 @@ def upstream_gone(self, branch=None) -> bool: upstream = self.upstreams.get(branch or self.current) return bool(upstream and not self.has_remote_branch(upstream.remote, upstream.branch)) + @cached_property + def orphan_branches(self) -> list[str]: + """Local branch names that were deleted on their corresponding remote.""" + result = [] + for name in sorted(self.local): + upstream = self.upstreams.get(name) + if not upstream or not self.has_remote_branch(upstream.remote, upstream.branch): + result.append(name) + + return result + class GitDir: """Model a local git repo""" @@ -294,7 +305,7 @@ def status_line(self, report: GitRunReport | None = None) -> str: report.add(note=f"⌛{runez.represented_duration(self.age)}") other_branches = refs.local - {refs.current, self.default_branch} - cleanable = len(other_branches & self.local_cleanable_branches) + cleanable = len(other_branches & self._local_cleanable_branches()) remaining = len(other_branches) - cleanable branch_summary = Reporter.joined(cleanable and f"+{cleanable}🪦", remaining and f"+{remaining}", separator="") if branch_summary: @@ -326,7 +337,8 @@ def represented_current_branch(self) -> str: def status_details(self, indent=" ") -> str: result = [] - orphan_branches = [branch for branch in self.orphan_branches if branch != self.refs.current and self.is_orphan_branch(branch)] + refs = self.refs + orphan_branches = [branch for branch in refs.orphan_branches if branch != refs.current and self.is_orphan_branch(branch)] if orphan_branches: result.append(f"Orphan branches: {', '.join(self.represented_branch(branch) for branch in orphan_branches)}") @@ -470,25 +482,13 @@ def status(self) -> GitStatus: def refs(self) -> GitRefs: return GitRefs(self) - @cached_property - def orphan_branches(self) -> list[str]: - """Local branch names that were deleted on their corresponding remote""" - result = [] - refs = self.refs - for name in sorted(refs.local): - upstream = refs.upstreams.get(name) - if not upstream or not refs.has_remote_branch(upstream.remote, upstream.branch): - result.append(name) - - return result - def is_protected_branch(self, name: str) -> bool: """True if branch should not be cleaned or reported as orphaned""" return bool(name and (name == self.default_branch or name in self.refs.default_branches.values())) def is_orphan_branch(self, name: str) -> bool: """True if branch is an unprotected orphan.""" - return bool(name and name in self.orphan_branches and not self.is_protected_branch(name)) + return bool(name and name in self.refs.orphan_branches and not self.is_protected_branch(name)) def represented_branch(self, name: str) -> str: """Styled representation of a branch name.""" @@ -501,16 +501,6 @@ def represented_branch(self, name: str) -> str: return result - @cached_property - def cleanable_base_ref(self) -> str: - """Ref that cleanup candidates must already be merged into""" - base_ref = self.default_branch - remote_branches = self.refs.by_remote.get("origin") - if remote_branches and base_ref in remote_branches: - base_ref = f"origin/{base_ref}" - - return base_ref - def is_ancestor(self, ref: str, target_ref: str) -> bool: """ :param str ref: Candidate ref @@ -523,9 +513,9 @@ def is_ancestor(self, ref: str, target_ref: str) -> bool: def merge_is_noop(self, ref: str, target_ref: str) -> bool: """ - :param str ref: Candidate ref + :param str ref: Candidate `ref` :param str target_ref: Ref that should already contain the candidate content - :return bool: True if merging 'ref' into 'target_ref' would leave 'target_ref' unchanged + :return bool: True if merging 'ref' into `target_ref` would leave `target_ref` unchanged """ target_tree = self.checked_git_command("rev-parse", f"{target_ref}^{{tree}}") proc = self.run_git_command("merge-tree", "--write-tree", "--no-messages", target_ref, ref) @@ -541,7 +531,7 @@ def cleanable_local_branch(self, name: str, include_current=False) -> CleanableL :param bool include_current: If True, allow checking the current branch :return CleanableLocalBranch|None: Cleanup details if local branch is safe to clean """ - base_ref = self.cleanable_base_ref + base_ref = self._cleanable_base_ref() refs = self.refs if not base_ref or not name or self.is_protected_branch(name) or (not include_current and name == refs.current): return None @@ -576,11 +566,17 @@ def cleanable_current_remote_branch(self) -> CleanableRemoteBranch | None: expected_oid = self.checked_git_command("rev-parse", branch_ref) return CleanableRemoteBranch(upstream.remote, upstream.branch, expected_oid) - @cached_property - def local_cleanable_branches(self) -> set[str]: - """ - :return set: Local branches that can be cleaned - """ + def _cleanable_base_ref(self) -> str: + """Ref that cleanup candidates must already be merged into""" + base_ref = self.default_branch + remote_branches = self.refs.by_remote.get("origin") + if remote_branches and base_ref in remote_branches: + base_ref = f"origin/{base_ref}" + + return base_ref + + def _local_cleanable_branches(self) -> set[str]: + """Local branches that can be cleaned""" return {name for name in self.refs.local if self.cleanable_local_branch(name)} From f015bd397b004f392db48edee142fab6544566ef Mon Sep 17 00:00:00 2001 From: Zoran Simic Date: Mon, 25 May 2026 13:52:56 -0700 Subject: [PATCH 2/4] Fine-tuned cached properties --- docs/design.md | 41 +++--- src/mgit/cli.py | 58 +++----- src/mgit/git.py | 332 ++++++++++++++++++++++---------------------- tests/test_cli.py | 54 ++++++- tests/test_groom.py | 11 +- tox.ini | 10 +- 6 files changed, 261 insertions(+), 245 deletions(-) diff --git a/docs/design.md b/docs/design.md index 04b665b..4816f04 100644 --- a/docs/design.md +++ b/docs/design.md @@ -10,22 +10,22 @@ should continue to evolve. `GitDir` is used for one command run and lazily captures git state that can otherwise require repeated subprocess calls. -- `GitDir.default_branch`: The default branch selected for the command run. - Once selected, it should remain stable while a command proceeds, including - while `groom` checks out, pulls, and deletes branches. A newly fetched - `origin/HEAD` could influence the initial selection if this property has not - yet been read. - -- `GitDir.status`: A `GitStatus` snapshot produced from +- `GitDir.lazy_status`: A `GitStatus` snapshot produced from `git status --porcelain=v2 --branch`. It includes pending changes and - upstream ahead/behind information. + upstream ahead/behind information. It does not classify ref-derived + conditions such as a gone upstream; command flows combine that information + with `GitRefs` where needed. -- `GitDir.refs`: A `GitRefs` snapshot of the current branch, local branches, +- `GitDir.lazy_refs`: A `GitRefs` snapshot of the current branch, local branches, configured remotes, remote-tracking branches, remote default branches, and upstream configuration. +- `GitRefs.default_branch`: The default branch derived for a particular refs + snapshot, preferring `origin/HEAD` and falling back to a visible `main` or + `master` branch. + - `GitRefs.orphan_branches`: Derived cached data owned by a particular - `GitRefs` snapshot. It is discarded naturally when `GitDir.refs` is + `GitRefs` snapshot. It is discarded naturally when `GitDir.lazy_refs` is replaced, so it must not be invalidated separately. ## Invalidation Requirements @@ -40,19 +40,14 @@ The operations below may run during one lifetime of a `GitDir` instance: | delete local branch | unchanged in the `groom` flow | stale | The local branch collection changes; deletion occurs after checkout. | | delete remote branch | unchanged in the `groom` flow | stale | The remote-tracking branch collection changes. | -`default_branch` is deliberately not listed as transient state to clear after -these operations: it is the command run's chosen target branch. Fetch should -occur before its first access in any flow that wants freshly observed remote -default-branch information. +`default_branch` and `orphan_branches` share the lifetime of their owning +`GitRefs` snapshot. Replacing stale `refs` naturally recomputes both from the +fresh snapshot. ## Current Implementation Note -`GitDir.clear_cached_state()` currently discards every cached property on -`GitDir`. This is broader than the invalidation requirements above and clears -`default_branch` unnecessarily. Also, `fetch_now()` currently updates `age` -without clearing an already materialized `status` or `refs` snapshot; existing -command flows rely on fetching before reading them. - -A follow-up refactor should replace the generic reset with explicit -invalidation of `status` and `refs` at the mutation points above, including -`fetch_now()`. +`GitDir.lazy_status` and `GitDir.lazy_refs` are ordinary lazy properties backed by +`_status` and `_refs`. State-changing `GitDir` methods explicitly set the +affected backing field to `None` according to the table above. This keeps +invalidation local to the operation that makes a snapshot stale and avoids a +generic cache reset. diff --git a/src/mgit/cli.py b/src/mgit/cli.py index d2fa9c7..44d1146 100644 --- a/src/mgit/cli.py +++ b/src/mgit/cli.py @@ -140,7 +140,7 @@ class FetchCommand(FolderTargetCommand): short_name = "f" def run_single(self, git: GitDir): - report = git.fetch_now().require_success("fetch") + report = git.fetch_now(abort_on_failure=True) print(git.status_line(report)) details = git.status_details() if details: @@ -159,7 +159,7 @@ class PullCommand(FolderTargetCommand): short_name = "p" def run_single(self, git: GitDir): - report = git.pull().require_success("pull") + report = git.pull(abort_on_failure=True).require_success("pull") print(git.status_line(report)) details = git.status_details() if details: @@ -192,7 +192,7 @@ class BranchesCommand(FolderTargetCommand): short_name = "b" def run_single(self, git: GitDir): - details = git.branch_details() + details = git.lazy_refs.branch_details() if details: print(details) @@ -203,7 +203,7 @@ def run_multi(self, project_dir: ProjectDir): print(f"{git.basename}:") indent = " " if show_names else "" - details = git.branch_details(indent=indent) + details = git.lazy_refs.branch_details(indent=indent) if details: print(details) @@ -214,61 +214,37 @@ class GroomCommand(FolderTargetCommand): short_name = "g" - def _checkout_default_branch(self, git: GitDir, branch: str): - git.checked_git_command("checkout", branch) - git.clear_cached_state() - print(f"Checked out {git.represented_branch(branch)} branch") - - def _delete_local_branch(self, git: GitDir, cleanup): - args = ["branch", "--delete", cleanup.name] - if cleanup.force_delete: - args.insert(2, "--force") - - git.checked_git_command(*args) - print(f"Deleted branch {git.represented_branch(cleanup.name)}") - git.clear_cached_state() - - def _delete_remote_branch(self, git: GitDir, cleanup): - branch_ref = f"refs/heads/{cleanup.branch}" - git.checked_git_command( - "push", - f"--force-with-lease={branch_ref}:{cleanup.expected_oid}", - "--delete", - cleanup.remote, - branch_ref, - ) - print(f"Deleted remote branch {cleanup.remote}/{git.represented_branch(cleanup.branch)}") - git.clear_cached_state() - def run_single(self, git: GitDir): - report = git.fetch_now().require_success("groom") - refs = git.refs + report = git.fetch_now(abort_on_failure=True) + refs = git.lazy_refs current_branch = refs.current - default_branch = git.default_branch + default_branch = refs.default_branch if current_branch == default_branch: report.add(note="already on default branch") print(git.status_line(report)) return - git.status.require_clean("groom") + git.lazy_status.require_clean("groom") local_cleanup = git.cleanable_local_branch(current_branch, include_current=True) if not local_cleanup: - Reporter.abort(f"Branch {git.represented_branch(current_branch)} can't be cleaned") + Reporter.abort(f"Branch {refs.represented_branch(current_branch)} can't be cleaned") upstream = refs.upstreams.get(current_branch) remote_cleanup = None if upstream and refs.has_remote_branch(upstream.remote, upstream.branch): remote_cleanup = git.cleanable_current_remote_branch() if not remote_cleanup: - Reporter.abort(f"Remote branch {upstream.remote}/{git.represented_branch(upstream.branch)} can't be cleaned automatically") + Reporter.abort(f"Remote branch {upstream.remote}/{refs.represented_branch(upstream.branch)} can't be cleaned automatically") - self._checkout_default_branch(git, default_branch) - report = git.pull().require_success("groom") - git.clear_cached_state() + git.checkout_default_branch() + print(f"Checked out {git.lazy_refs.represented_branch(default_branch)} branch") + report = git.pull(abort_on_failure=True).require_success("groom") if remote_cleanup: - self._delete_remote_branch(git, remote_cleanup) + git.delete_remote_branch(remote_cleanup) + print(f"Deleted remote branch {remote_cleanup.remote}/{git.lazy_refs.represented_branch(remote_cleanup.branch)}") - self._delete_local_branch(git, local_cleanup) + git.delete_local_branch(local_cleanup) + print(f"Deleted branch {git.lazy_refs.represented_branch(local_cleanup.name)}") print(git.status_line(report)) diff --git a/src/mgit/git.py b/src/mgit/git.py index 142de5d..086d1f9 100644 --- a/src/mgit/git.py +++ b/src/mgit/git.py @@ -14,6 +14,12 @@ from pathlib import Path +FRESH_FETCH_THRESHOLD = 30 +FRESHNESS_THRESHOLD = 12 * runez.date.SECONDS_IN_ONE_HOUR +LOCAL_REF_PREFIX = "refs/heads/" +REMOTE_REF_PREFIX = "refs/remotes/" + + class Reporter: """Central user-facing reporting helpers.""" @@ -54,36 +60,19 @@ def debug(message: str, *args: object): Reporter.log.debug(message, *args) -FRESH_FETCH_THRESHOLD = 30 -FRESHNESS_THRESHOLD = 12 * runez.date.SECONDS_IN_ONE_HOUR -LOCAL_REF_PREFIX = "refs/heads/" -REMOTE_REF_PREFIX = "refs/remotes/" - - -def git_error_message(proc: subprocess.CompletedProcess[str]) -> str | None: - """Short user-facing description of a failed git command.""" - if proc.returncode: - detail = (proc.stderr or proc.stdout).strip() or f"git exited with code {proc.returncode}" - detail = detail.strip().lower() - if "following untracked" in detail: - return "untracked files would be overwritten" - - if "repository not found" in detail: - return "repository not found" - - lines = [] - prefixed = [] - for line in detail.splitlines(): - line = line.strip().strip(".") - if line: - p = line.partition(":") - if p[2] and p[0] in ("git", "error", "fatal"): - prefixed.append(p[2].strip()) +def compact_git_error(proc: subprocess.CompletedProcess[str]) -> str | None: + """Short detail from a failed git command, suitable for a status line.""" + if not proc.returncode: + return None - else: - lines.append(line) + detail = (proc.stderr or proc.stdout).strip() or f"git exited with code {proc.returncode}" + lines = [line.strip().rstrip(".") for line in detail.splitlines() if line.strip()] + for line in lines: + prefix, separator, message = line.partition(":") + if separator and prefix in ("git", "error", "fatal"): + return message.strip() - return " ".join((prefixed or lines)[:2]).replace(" ", " ").strip() + return lines[0] class GitRunReport: @@ -164,11 +153,6 @@ def add( return self -def git_error_report(proc: subprocess.CompletedProcess[str]) -> GitRunReport: - """GitRunReport describing a failed git command.""" - return GitRunReport(problem=git_error_message(proc)) - - @dataclass class BranchUpstream: """Configured upstream for a local branch.""" @@ -206,7 +190,9 @@ class GitRefs: upstreams: dict[str, BranchUpstream] def __init__(self, parent: GitDir): - self.current, self.detached = parent._current_branch() + r = parent.run_git_command("symbolic-ref", "--quiet", "--short", "HEAD", exit_codes=(0, 1)) + self.detached = r.returncode != 0 + self.current = r.stdout.strip() if r.returncode == 0 else "HEAD" self.local = set() self.remotes = parent.checked_git_command_lines("remote") self.by_remote = {} @@ -252,6 +238,24 @@ def upstream_gone(self, branch=None) -> bool: upstream = self.upstreams.get(branch or self.current) return bool(upstream and not self.has_remote_branch(upstream.remote, upstream.branch)) + @cached_property + def default_branch(self) -> str: + """Default branch name.""" + branch = self.default_branches.get("origin") + if not branch: + all_branches = self.local + origin_branches = self.by_remote.get("origin") + if origin_branches: + all_branches = all_branches | origin_branches + + branch = next((name for name in ("main", "master") if name in all_branches), "main") + + return branch + + def is_protected_branch(self, name: str) -> bool: + """True if branch should not be cleaned or reported as orphaned.""" + return bool(name and (name == self.default_branch or name in self.default_branches.values())) + @cached_property def orphan_branches(self) -> list[str]: """Local branch names that were deleted on their corresponding remote.""" @@ -263,6 +267,45 @@ def orphan_branches(self) -> list[str]: return result + def is_orphan_branch(self, name: str) -> bool: + """True if branch is an unprotected orphan.""" + return bool(name and name in self.orphan_branches and not self.is_protected_branch(name)) + + def annotated_branch(self, width: int, name: str) -> str: + padding = " " * (width - len(name)) + result = ["*" if name == self.current else " ", f"{self.represented_branch(name)}{padding}"] + if name == self.default_branch: + result.append(Reporter.branch_default("[default]")) + + if self.is_orphan_branch(name): + result.append(Reporter.branch_orphaned("[orphaned]")) + + return Reporter.joined(result) + + def branch_details(self, indent="") -> str: + branches = sorted(self.local) or [self.current] + width = max(len(name) for name in branches) + return Reporter.joined_lines((self.annotated_branch(width, name) for name in branches), indent=indent) + + def cleanable_base_ref(self) -> str: + """Ref that cleanup candidates must already be merged into.""" + base_ref = self.default_branch + remote_branches = self.by_remote.get("origin") + if remote_branches and base_ref in remote_branches: + base_ref = f"origin/{base_ref}" + + return base_ref + + def represented_branch(self, name: str) -> str: + """Styled representation of a branch name.""" + if name == self.default_branch: + name = Reporter.branch_default(name) + + elif self.is_orphan_branch(name): + name = Reporter.branch_orphaned(name) + + return runez.bold(name) + class GitDir: """Model a local git repo""" @@ -274,12 +317,8 @@ def __init__(self, path: Path): self.path = path self.basename = path.name self.age = self._current_age() - - def _branch_annotations(self, name): - return Reporter.joined( - name == self.default_branch and Reporter.branch_default("[default]"), - self.is_orphan_branch(name) and Reporter.branch_orphaned("[orphaned]"), - ) + self._status: GitStatus | None = None + self._refs: GitRefs | None = None @staticmethod def _detail_lines(items, state_style, worktree_style=None) -> list[str]: @@ -297,14 +336,14 @@ def _detail_lines(items, state_style, worktree_style=None) -> list[str]: return lines def status_line(self, report: GitRunReport | None = None) -> str: - refs = self.refs - status = self.status + refs = self.lazy_refs + status = self.lazy_status edits, deletes, new = status.pending_change_counts() report = GitRunReport(report) if self.age is not None and self.age > FRESHNESS_THRESHOLD: report.add(note=f"⌛{runez.represented_duration(self.age)}") - other_branches = refs.local - {refs.current, self.default_branch} + other_branches = refs.local - {refs.current, refs.default_branch} cleanable = len(other_branches & self._local_cleanable_branches()) remaining = len(other_branches) - cleanable branch_summary = Reporter.joined(cleanable and f"+{cleanable}🪦", remaining and f"+{remaining}", separator="") @@ -322,125 +361,106 @@ def status_line(self, report: GitRunReport | None = None) -> str: ) def represented_current_branch(self) -> str: - refs = self.refs + refs = self.lazy_refs branch = refs.current if refs.detached: icon = " 👻" - elif self.is_orphan_branch(branch): + elif refs.is_orphan_branch(branch): icon = " 🪦" else: icon = " ✅" if self.age is not None and self.age <= FRESH_FETCH_THRESHOLD else " ☑️" - return self.represented_branch(branch) + icon + return refs.represented_branch(branch) + icon def status_details(self, indent=" ") -> str: result = [] - refs = self.refs - orphan_branches = [branch for branch in refs.orphan_branches if branch != refs.current and self.is_orphan_branch(branch)] + refs = self.lazy_refs + orphan_branches = [branch for branch in refs.orphan_branches if branch != refs.current and refs.is_orphan_branch(branch)] if orphan_branches: - result.append(f"Orphan branches: {', '.join(self.represented_branch(branch) for branch in orphan_branches)}") + result.append(f"Orphan branches: {', '.join(refs.represented_branch(branch) for branch in orphan_branches)}") - status = self.status + status = self.lazy_status result.extend(self._detail_lines(status.modified, Reporter.index_change, Reporter.worktree_change)) result.extend(self._detail_lines(status.untracked, Reporter.untracked_change)) return Reporter.joined_lines(result, indent=indent) - def branch_details(self, indent="") -> str: - refs = self.refs - branches = sorted(refs.local) or [refs.current] - width = max(len(name) for name in branches) - result = [] - for name in branches: - padding = " " * (width - len(name)) - line = f"{'*' if name == refs.current else ' '} {self.represented_branch(name)}{padding}" - annotations = self._branch_annotations(name) - if annotations: - line += f" {annotations}" - - result.append(line) - - return Reporter.joined_lines(result, indent=indent) - - def run_git_command(self, *args: str) -> subprocess.CompletedProcess[str]: + def run_git_command(self, *args: str, exit_codes: tuple[int, ...] | None = None) -> subprocess.CompletedProcess[str]: """ :param args: Execute git command with provided args + :param exit_codes: Acceptable return codes, or None to return failures for reporting :return subprocess.CompletedProcess: Result from git """ cmd = ["git", "-C", str(self.path), *args] Reporter.debug("Running: git -C %s %s", runez.short(self.path), " ".join(args)) - return subprocess.run(cmd, check=False, capture_output=True, text=True) # noqa: S603 + proc = subprocess.run(cmd, check=False, capture_output=True, text=True) # noqa: S603 + if exit_codes is not None and proc.returncode not in exit_codes: + detail = (proc.stderr or proc.stdout).rstrip() or f"git exited with code {proc.returncode}" + Reporter.abort(f"git {' '.join(args)} failed:\n{detail}") + + return proc def checked_git_command(self, *args: str) -> str: """ :param args: Execute git command with provided args :return str: Output from git command, aborting if git fails """ - proc = self.run_git_command(*args) - Reporter.abort_if(proc.returncode, f"git {' '.join(args)} failed: {git_error_message(proc)}") + proc = self.run_git_command(*args, exit_codes=(0,)) return proc.stdout.strip() def checked_git_command_lines(self, *args: str) -> list[str]: return [line.strip() for line in self.checked_git_command(*args).splitlines() if line.strip()] - def clear_cached_state(self): - """Discard cached git state after a command may have changed refs or worktree state.""" - for k, v in vars(self.__class__).items(): - if isinstance(v, cached_property) and k in self.__dict__: - self.__dict__.pop(k, None) - - def _current_branch(self) -> tuple[str, bool]: - proc = self.run_git_command("symbolic-ref", "--quiet", "--short", "HEAD") - if proc.returncode == 0: - return proc.stdout.strip(), False - - Reporter.abort_if(proc.returncode != 1, f"Could not inspect git branch: {git_error_message(proc)}") - proc = self.run_git_command("rev-parse", "--verify", "--quiet", "HEAD") - if proc.returncode == 0 and proc.stdout.strip(): - return "HEAD", True - - Reporter.abort(f"Could not inspect git HEAD: {git_error_message(proc)}") - - def fetch_now(self) -> GitRunReport: - proc = self.run_git_command("fetch", "--all", "--prune") + def fetch_now(self, *, abort_on_failure=False) -> GitRunReport: + exit_codes = (0,) if abort_on_failure else None + proc = self.run_git_command("fetch", "--all", "--prune", exit_codes=exit_codes) + self._status = None + self._refs = None if proc.returncode == 0: self.age = 0 - return git_error_report(proc) + return GitRunReport(problem=compact_git_error(proc)) def checkout_default_branch(self) -> GitRunReport: """ :return GitRunReport: Checkout report """ - branch = self.default_branch - if self.refs.current == branch: - return GitRunReport() + refs = self.lazy_refs + branch = refs.default_branch + report = GitRunReport() + if refs.current != branch: + self.checked_git_command("checkout", branch) + report.add(progress=f"checked out {refs.represented_branch(branch)}") + self._status = None + self._refs = None - self.checked_git_command("checkout", branch) - self.clear_cached_state() - return GitRunReport(progress=f"checked out {self.represented_branch(branch)}") + return report - def pull(self) -> GitRunReport: + def pull(self, *, abort_on_failure=False) -> GitRunReport: """Pull from tracked remote""" - refs = self.refs + refs = self.lazy_refs if not refs.remotes: return GitRunReport().cant_pull("no remotes") if refs.detached: return GitRunReport().cant_pull("HEAD detached") - status = self.status - if status.has_pending_changes or status.report.has_problems: - reason = "pending changes" if status.has_pending_changes else None - return GitRunReport(status.report).cant_pull(reason) + if refs.upstream_gone(): + return GitRunReport().cant_pull("remote branch gone") + + status = self.lazy_status + if status.has_pending_changes: + return GitRunReport().cant_pull("pending changes") note = status.upstream_delta() or "up-to-date" - proc = self.run_git_command("pull", "--rebase") - self.clear_cached_state() + exit_codes = (0,) if abort_on_failure else None + proc = self.run_git_command("pull", "--rebase", exit_codes=exit_codes) + self._status = None + self._refs = None report = GitRunReport(note=f"was {note}") if proc.returncode: - report.cant_pull(git_error_message(proc)) + report.cant_pull(compact_git_error(proc)) else: self.age = 0 @@ -457,58 +477,47 @@ def _current_age(self) -> int | None: except OSError: pass - @cached_property - def default_branch(self) -> str: - """Default branch name""" - refs = self.refs - branch = refs.default_branches.get("origin") - if branch: - return branch - - origin_branches = refs.by_remote.get("origin") - if origin_branches: - for candidate in ("main", "master"): - if candidate in refs.local or candidate in origin_branches: - return candidate - - return "main" - - @cached_property - def status(self) -> GitStatus: + @property + def lazy_status(self) -> GitStatus: """Parsed info from 'git status --porcelain=v2 --branch'""" - return GitStatus(self) - - @cached_property - def refs(self) -> GitRefs: - return GitRefs(self) - - def is_protected_branch(self, name: str) -> bool: - """True if branch should not be cleaned or reported as orphaned""" - return bool(name and (name == self.default_branch or name in self.refs.default_branches.values())) - - def is_orphan_branch(self, name: str) -> bool: - """True if branch is an unprotected orphan.""" - return bool(name and name in self.refs.orphan_branches and not self.is_protected_branch(name)) - - def represented_branch(self, name: str) -> str: - """Styled representation of a branch name.""" - result = runez.bold(name) - if name == self.default_branch: - return runez.green(result) + if self._status is None: + self._status = GitStatus(self) - if self.is_orphan_branch(name): - return runez.orange(result) + return self._status - return result + @property + def lazy_refs(self) -> GitRefs: + if self._refs is None: + self._refs = GitRefs(self) + + return self._refs + + def delete_local_branch(self, cleanup: CleanableLocalBranch): + args = ["branch", "--delete", cleanup.name] + if cleanup.force_delete: + args.insert(2, "--force") + + self.checked_git_command(*args) + self._refs = None + + def delete_remote_branch(self, cleanup: CleanableRemoteBranch): + branch_ref = f"refs/heads/{cleanup.branch}" + self.checked_git_command( + "push", + f"--force-with-lease={branch_ref}:{cleanup.expected_oid}", + "--delete", + cleanup.remote, + branch_ref, + ) + self._refs = None def is_ancestor(self, ref: str, target_ref: str) -> bool: """ - :param str ref: Candidate ref + :param str ref: Candidate `ref` :param str target_ref: Ref that should contain the candidate :return bool: True if 'ref' is an ancestor of 'target_ref' """ - proc = self.run_git_command("merge-base", "--is-ancestor", ref, target_ref) - Reporter.abort_if(proc.returncode > 1, f"Could not inspect git ancestry: {git_error_message(proc)}") + proc = self.run_git_command("merge-base", "--is-ancestor", ref, target_ref, exit_codes=(0, 1)) return proc.returncode == 0 def merge_is_noop(self, ref: str, target_ref: str) -> bool: @@ -531,9 +540,9 @@ def cleanable_local_branch(self, name: str, include_current=False) -> CleanableL :param bool include_current: If True, allow checking the current branch :return CleanableLocalBranch|None: Cleanup details if local branch is safe to clean """ - base_ref = self._cleanable_base_ref() - refs = self.refs - if not base_ref or not name or self.is_protected_branch(name) or (not include_current and name == refs.current): + refs = self.lazy_refs + base_ref = refs.cleanable_base_ref() + if not base_ref or not name or refs.is_protected_branch(name) or (not include_current and name == refs.current): return None if self.is_ancestor(name, base_ref): @@ -546,9 +555,9 @@ def cleanable_local_branch(self, name: str, include_current=False) -> CleanableL def cleanable_current_remote_branch(self) -> CleanableRemoteBranch | None: """Return cleanup details for the current branch's safely deletable origin ref.""" - refs = self.refs + refs = self.lazy_refs upstream = refs.upstreams.get(refs.current) - default_branch = self.default_branch + default_branch = refs.default_branch if ( not upstream or upstream.remote != "origin" @@ -566,18 +575,9 @@ def cleanable_current_remote_branch(self) -> CleanableRemoteBranch | None: expected_oid = self.checked_git_command("rev-parse", branch_ref) return CleanableRemoteBranch(upstream.remote, upstream.branch, expected_oid) - def _cleanable_base_ref(self) -> str: - """Ref that cleanup candidates must already be merged into""" - base_ref = self.default_branch - remote_branches = self.refs.by_remote.get("origin") - if remote_branches and base_ref in remote_branches: - base_ref = f"origin/{base_ref}" - - return base_ref - def _local_cleanable_branches(self) -> set[str]: """Local branches that can be cleaned""" - return {name for name in self.refs.local if self.cleanable_local_branch(name)} + return {name for name in self.lazy_refs.local if self.cleanable_local_branch(name)} class GitStatus: @@ -588,7 +588,6 @@ def __init__(self, parent: GitDir): self.behind = 0 self.modified = [] self.untracked = [] - self.report = GitRunReport() lines = parent.checked_git_command_lines("status", "--porcelain=v2", "--branch") for line in lines: prefix = line[0] @@ -609,9 +608,6 @@ def __init__(self, parent: GitDir): path = fields[-1].partition("\t")[0] self.modified.append(f"{state} {path}") - if parent.refs.upstream_gone(): - self.report.add(problem="remote branch gone") - @property def dirty_note(self) -> str: """Short overview of pending changes.""" diff --git a/tests/test_cli.py b/tests/test_cli.py index 6a7c983..2fc632a 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -64,6 +64,9 @@ def test_single_status_shows_pending_paths_and_discovers_parent_checkout(cli, gi assert "README.md" in cli.logged assert "new.txt" in cli.logged + cli.run("fetch repo") + assert cli.succeeded + (repo.cwd / "foo").mkdir() os.chdir(repo.cwd / "foo") cli.run() @@ -168,9 +171,9 @@ def test_branch_names_are_styled_consistently(cli, git): repo.checkout("topic") repo.push("origin", "--delete", "orphan") with runez.ActivateColors(True): - default_name = runez.green(runez.bold("main")) + default_name = runez.bold(runez.green("main")) topic_name = runez.bold("topic") - orphan_name = runez.orange(runez.bold("orphan")) + orphan_name = runez.bold(runez.orange("orphan")) cli.run("--color always branches repo") assert cli.succeeded @@ -254,15 +257,43 @@ def test_pull_refuses_untracked_changes(cli, git): assert "can't pull; pending changes" in cli.logged -def test_pull_failure_retains_previous_status_note(cli, git): +def test_pull_refuses_branch_with_gone_upstream(cli, git): + repo = git.seeded("repo") + repo.checkout("-b", "gone") + repo.push("-u", "origin", "gone") + repo.checkout("main") + repo.push("origin", "--delete", "gone") + repo.checkout("gone") + + cli.run("pull repo") + + assert cli.failed + assert "can't pull; remote branch gone" in cli.logged + + +def test_single_pull_failure_shows_full_git_error(cli, git): repo = git.seeded("repo") repo.remote("set-url", "origin", "missing.git") cli.run("pull repo") assert cli.failed - assert "can't pull" in cli.logged - assert "was up-to-date" in cli.logged + assert "git pull --rebase failed:" in cli.logged + assert "fatal: 'missing.git' does not appear to be a git repository" in cli.logged + assert "Please make sure you have the correct access rights" in cli.logged + + +def test_workspace_pull_failure_is_compact(cli, git): + source = git.seeded_set("source") + repo = git.clone(source.remote_url, "workspace/repo") + repo.remote("set-url", "origin", "missing.git") + + cli.run("pull workspace") + + assert cli.succeeded + assert "repo: main ✅ can't pull; 'missing.git' does not appear to be a git repository; was up-to-date" in cli.logged + assert "fatal:" not in cli.logged + assert "Please make sure" not in cli.logged def test_main_checks_out_default_branch(cli, git): @@ -273,3 +304,16 @@ def test_main_checks_out_default_branch(cli, git): assert cli.succeeded assert repo.current_branch == "main" + + +def test_main_prefers_main_over_master_without_remote_head(cli, git): + repo = git.seeded("repo") + repo.branch("master") + repo.push("origin", "master") + repo.run_git("remote", "set-head", "origin", "--delete") + repo.checkout("-b", "feature") + + cli.run("m repo") + + assert cli.succeeded + assert repo.current_branch == "main" diff --git a/tests/test_groom.py b/tests/test_groom.py index 84247a9..50c71ce 100644 --- a/tests/test_groom.py +++ b/tests/test_groom.py @@ -1,6 +1,6 @@ import pytest -from mgit.cli import GroomCommand +from mgit.git import GitDir def make_stale_tracked_branch(work, branch="stale", merged=True): @@ -113,14 +113,15 @@ def test_groom_does_not_delete_remote_branch_that_advanced_after_fetch(cli, git, racer = git.clone(repos.remote_url, "racer") racer.checkout("-b", "published", "origin/published") - checkout_default_branch = GroomCommand._checkout_default_branch + checkout_default_branch = GitDir.checkout_default_branch - def advance_remote_after_checkout(command, git_dir, branch): - checkout_default_branch(command, git_dir, branch) + def advance_remote_after_checkout(git_dir): + report = checkout_default_branch(git_dir) racer.commit_file("late.txt", "not merged", "advance published") racer.push("origin", "published") + return report - monkeypatch.setattr(GroomCommand, "_checkout_default_branch", advance_remote_after_checkout) + monkeypatch.setattr(GitDir, "checkout_default_branch", advance_remote_after_checkout) cli.run("g work") assert cli.failed diff --git a/tox.ini b/tox.ini index dbe8b57..7e7a989 100644 --- a/tox.ini +++ b/tox.ini @@ -1,5 +1,5 @@ [tox] -envlist = py{39,310,314}, coverage, style, docs, typecheck +envlist = py{39,314}, coverage, style, docs, typecheck [testenv] setenv = COVERAGE_FILE={toxworkdir}/.coverage.{envname} @@ -39,7 +39,11 @@ commands = ruff check --fix ruff format -[coverage:xml] -output = .tox/test-reports/coverage.xml +[coverage:run] +data_file = .tox/.coverage [coverage:html] directory = .tox/test-reports/htmlcov +[coverage:xml] +output = .tox/test-reports/coverage.xml +[pytest] +cache_dir = .tox/pytest_cache From d3fb55ebb55436292d59eec8cf73d45f75aa48fd Mon Sep 17 00:00:00 2001 From: Zoran Simic Date: Mon, 25 May 2026 22:22:13 -0700 Subject: [PATCH 3/4] Added legend --- README.rst | 23 +++++- docs/design.md | 15 ++-- src/mgit/cli.py | 41 +++++++++- src/mgit/git.py | 196 ++++++++++++++++++++++++++++++++-------------- tests/test_cli.py | 109 ++++++++++++++++++++++---- 5 files changed, 300 insertions(+), 84 deletions(-) diff --git a/README.rst b/README.rst index 6805805..4648073 100644 --- a/README.rst +++ b/README.rst @@ -55,7 +55,8 @@ The short names are the intended interface: - ``fetch`` / ``f``: refresh remote refs, then show what changed. - ``pull`` / ``p``: pull with rebase only when pending work will not be disturbed. - ``main`` / ``m``: checkout the default branch, even if it is ``master``. -- ``branches`` / ``b``: list local branches with useful small annotations. +- ``branches`` / ``b``: list local branches with presence and cleanup annotations. +- ``legend`` / ``l``: explain status and branch symbols. - ``groom`` / ``g``: safely finish with the merged branch you are currently on. The command shape is:: @@ -92,10 +93,28 @@ In one glance you get: ``✅`` means recently refreshed, ``☑️`` means the local picture may be older, ``⌛`` calls out notably stale fetch information, a branch-level ``🪦`` means -its upstream is gone, and ``👻`` means detached ``HEAD``. A bracket such as +the branch has been proven cleanable, and ``👻`` means detached ``HEAD``. A bracket such as ``[+2🪦+1]`` says that, besides the current and default branches, two local branches have been proven cleanable and one remains. +The branch list spells that state out when you need to decide what is done:: + + mgit b + commands-instead-of-flags💾☁️🪦 [cleanable] + main [default] + * more-cleanup💾 [current] + v2gpt💾☁️ + +Run ``mgit legend`` for the symbol key, or ``mgit b --legend`` to append it +below the branch listing. + +``💾`` means the branch is local and ``☁️`` means its configured upstream is +still present in the fetched remote refs. ``🪦 [cleanable]`` appears only +when the local branch, and any shown upstream ref, is already merged into the +default branch or is content-equivalent after a squash merge. An unpublished +branch carrying new work therefore appears as ``💾`` rather than looking +abandoned. + For a single checkout, status also prints the pending paths. Workspace output keeps to one line per repo, so you can keep checking a directory full of projects without turning the terminal into a log dump:: diff --git a/docs/design.md b/docs/design.md index 4816f04..fcf4d39 100644 --- a/docs/design.md +++ b/docs/design.md @@ -24,9 +24,10 @@ otherwise require repeated subprocess calls. snapshot, preferring `origin/HEAD` and falling back to a visible `main` or `master` branch. -- `GitRefs.orphan_branches`: Derived cached data owned by a particular - `GitRefs` snapshot. It is discarded naturally when `GitDir.lazy_refs` is - replaced, so it must not be invalidated separately. +- `GitDir.branch_infos`: Derived branch display data combining a local ref, + its configured upstream presence, and local/remote cleanup proofs. It is + invalidated together with `GitDir.lazy_refs`, because either ref movement or + fetch/prune results can change whether a branch is cleanable. ## Invalidation Requirements @@ -34,15 +35,15 @@ The operations below may run during one lifetime of a `GitDir` instance: | Operation | `status` | `refs` | Reason | | --- | --- | --- | --- | -| `fetch_now()` | stale | stale | Ahead/behind and remote-tracking branches can change; pruning can alter orphan detection. | +| `fetch_now()` | stale | stale | Ahead/behind and remote-tracking branches can change; pruning can alter branch presence and cleanup proof. | | checkout | stale | stale | The checked-out branch and displayed status can change. | | `pull()` | stale | stale | Worktree/status and remote-tracking refs may change, including when the pull fails after doing some work. | | delete local branch | unchanged in the `groom` flow | stale | The local branch collection changes; deletion occurs after checkout. | | delete remote branch | unchanged in the `groom` flow | stale | The remote-tracking branch collection changes. | -`default_branch` and `orphan_branches` share the lifetime of their owning -`GitRefs` snapshot. Replacing stale `refs` naturally recomputes both from the -fresh snapshot. +`default_branch` shares the lifetime of its owning `GitRefs` snapshot. +Replacing stale `refs` also discards derived `branch_infos` so cleanup state +is recomputed from the fresh snapshot. ## Current Implementation Note diff --git a/src/mgit/cli.py b/src/mgit/cli.py index 44d1146..ad8abc4 100644 --- a/src/mgit/cli.py +++ b/src/mgit/cli.py @@ -187,15 +187,34 @@ def run_single(self, git: GitDir): @cli_command class BranchesCommand(FolderTargetCommand): - """Show local branches.""" + """Show local branches and their cleanup state.""" short_name = "b" + def __init__(self, folder: Path | None, with_legend=False): + super().__init__(folder) + self.with_legend = with_legend + + @classmethod + def add_arguments(cls, parser: argparse.ArgumentParser): + parser.add_argument("-l", "--legend", action="store_true", help="Show symbol legend below the branch list.") + super().add_arguments(parser) + + @classmethod + def from_namespace(cls, namespace: argparse.Namespace) -> FolderTargetCommand: + return cls(folder=namespace.folder, with_legend=namespace.legend) + + def print_legend(self): + if self.with_legend: + print(f"\n{Reporter.legend()}") + def run_single(self, git: GitDir): - details = git.lazy_refs.branch_details() + details = git.branch_details() if details: print(details) + self.print_legend() + def run_multi(self, project_dir: ProjectDir): show_names = len(project_dir.git_dirs) > 1 for git in project_dir.git_dirs: @@ -203,10 +222,26 @@ def run_multi(self, project_dir: ProjectDir): print(f"{git.basename}:") indent = " " if show_names else "" - details = git.lazy_refs.branch_details(indent=indent) + details = git.branch_details(indent=indent) if details: print(details) + self.print_legend() + + +@cli_command +class LegendCommand(CliCommand): + """Explain status and branch symbols.""" + + short_name = "l" + + @classmethod + def from_namespace(cls, _namespace: argparse.Namespace) -> LegendCommand: + return cls() + + def run(self): + print(Reporter.legend()) + @cli_command class GroomCommand(FolderTargetCommand): diff --git a/src/mgit/git.py b/src/mgit/git.py index 086d1f9..1c945d9 100644 --- a/src/mgit/git.py +++ b/src/mgit/git.py @@ -26,7 +26,7 @@ class Reporter: log = logging.getLogger("mgit") branch_default = runez.green - branch_orphaned = runez.orange + branch_cleanable = runez.orange problem = runez.red note = runez.purple progress = runez.plain @@ -59,6 +59,36 @@ def abort_if(condition: object, message, exit_code: int = 1): def debug(message: str, *args: object): Reporter.log.debug(message, *args) + i_fresh = "✅" + i_not_fresh = "☑️" + i_stale = "⌛" + i_local = "💾" + i_remote = "☁️" + i_cleanable = "🪦" + i_detached = "👻" + i_modified = "✏️" + i_deleted = "🗑️" + i_untracked = "🆕" + + @classmethod + def legend(cls) -> str: + """Explanation of status and branch symbols.""" + fresh = runez.represented_duration(FRESH_FETCH_THRESHOLD, span=1) + stale = runez.represented_duration(FRESHNESS_THRESHOLD, span=1) + return Reporter.joined_lines( + f"{cls.i_fresh} repo freshly fetched (within {fresh})", + f"{cls.i_not_fresh} repo not freshly fetched (more than {fresh} ago)", + f"{cls.i_stale} repo fetch notably stale (more than {stale} ago)", + f"{cls.i_local} branch present locally", + f"{cls.i_remote} configured upstream present in fetched remote refs", + f"{cls.i_cleanable} cleanable branch", + f"{cls.i_detached} detached HEAD", + f"{cls.i_modified} modified files", + f"{cls.i_deleted} deleted files", + f"{cls.i_untracked} new (untracked) files", + f"[+N{cls.i_cleanable}+N] cleanable and remaining local branches", + ) + def compact_git_error(proc: subprocess.CompletedProcess[str]) -> str | None: """Short detail from a failed git command, suitable for a status line.""" @@ -178,6 +208,23 @@ class CleanableRemoteBranch: expected_oid: str +@dataclass(frozen=True) +class BranchInfo: + """Presence and cleanup proof for one displayed local branch.""" + + name: str + has_local: bool + upstream: BranchUpstream | None = None + has_remote: bool = False + local_cleanup: CleanableLocalBranch | None = None + remote_cleanup: CleanableRemoteBranch | None = None + + @property + def cleanable(self) -> bool: + """True if each present ref represented by this branch can be removed.""" + return bool(self.local_cleanup and (not self.has_remote or self.remote_cleanup)) + + class GitRefs: """Repository ref and upstream snapshot.""" @@ -253,40 +300,9 @@ def default_branch(self) -> str: return branch def is_protected_branch(self, name: str) -> bool: - """True if branch should not be cleaned or reported as orphaned.""" + """True if branch should not be cleaned.""" return bool(name and (name == self.default_branch or name in self.default_branches.values())) - @cached_property - def orphan_branches(self) -> list[str]: - """Local branch names that were deleted on their corresponding remote.""" - result = [] - for name in sorted(self.local): - upstream = self.upstreams.get(name) - if not upstream or not self.has_remote_branch(upstream.remote, upstream.branch): - result.append(name) - - return result - - def is_orphan_branch(self, name: str) -> bool: - """True if branch is an unprotected orphan.""" - return bool(name and name in self.orphan_branches and not self.is_protected_branch(name)) - - def annotated_branch(self, width: int, name: str) -> str: - padding = " " * (width - len(name)) - result = ["*" if name == self.current else " ", f"{self.represented_branch(name)}{padding}"] - if name == self.default_branch: - result.append(Reporter.branch_default("[default]")) - - if self.is_orphan_branch(name): - result.append(Reporter.branch_orphaned("[orphaned]")) - - return Reporter.joined(result) - - def branch_details(self, indent="") -> str: - branches = sorted(self.local) or [self.current] - width = max(len(name) for name in branches) - return Reporter.joined_lines((self.annotated_branch(width, name) for name in branches), indent=indent) - def cleanable_base_ref(self) -> str: """Ref that cleanup candidates must already be merged into.""" base_ref = self.default_branch @@ -296,13 +312,13 @@ def cleanable_base_ref(self) -> str: return base_ref - def represented_branch(self, name: str) -> str: + def represented_branch(self, name: str, *, cleanable=False) -> str: """Styled representation of a branch name.""" if name == self.default_branch: name = Reporter.branch_default(name) - elif self.is_orphan_branch(name): - name = Reporter.branch_orphaned(name) + elif cleanable: + name = Reporter.branch_cleanable(name) return runez.bold(name) @@ -319,6 +335,7 @@ def __init__(self, path: Path): self.age = self._current_age() self._status: GitStatus | None = None self._refs: GitRefs | None = None + self._branch_infos: list[BranchInfo] | None = None @staticmethod def _detail_lines(items, state_style, worktree_style=None) -> list[str]: @@ -341,12 +358,12 @@ def status_line(self, report: GitRunReport | None = None) -> str: edits, deletes, new = status.pending_change_counts() report = GitRunReport(report) if self.age is not None and self.age > FRESHNESS_THRESHOLD: - report.add(note=f"⌛{runez.represented_duration(self.age)}") + report.add(note=f"{Reporter.i_stale}{runez.represented_duration(self.age)}") - other_branches = refs.local - {refs.current, refs.default_branch} - cleanable = len(other_branches & self._local_cleanable_branches()) + other_branches = [info for info in self.branch_infos if info.name not in (refs.current, refs.default_branch)] + cleanable = sum(info.cleanable for info in other_branches) remaining = len(other_branches) - cleanable - branch_summary = Reporter.joined(cleanable and f"+{cleanable}🪦", remaining and f"+{remaining}", separator="") + branch_summary = Reporter.joined(cleanable and f"+{cleanable}{Reporter.i_cleanable}", remaining and f"+{remaining}", separator="") if branch_summary: branch_summary = f"[{branch_summary}]" @@ -354,9 +371,9 @@ def status_line(self, report: GitRunReport | None = None) -> str: self.represented_current_branch(), branch_summary, status.upstream_delta(), - edits and f"✏️{edits}", - deletes and f"🗑️{deletes}", - new and f"🆕{new}", + edits and f"{Reporter.i_modified}{edits}", + deletes and f"{Reporter.i_deleted}{deletes}", + new and f"{Reporter.i_untracked}{new}", report, ) @@ -364,22 +381,28 @@ def represented_current_branch(self) -> str: refs = self.lazy_refs branch = refs.current if refs.detached: - icon = " 👻" + return f"{refs.represented_branch(branch)} {Reporter.i_detached}" - elif refs.is_orphan_branch(branch): - icon = " 🪦" + info = next((candidate for candidate in self.branch_infos if candidate.name == branch), None) + cleanable = bool(info and info.cleanable) + if cleanable: + icon = Reporter.i_cleanable + + elif self.age is not None and self.age <= FRESH_FETCH_THRESHOLD: + icon = Reporter.i_fresh else: - icon = " ✅" if self.age is not None and self.age <= FRESH_FETCH_THRESHOLD else " ☑️" + icon = Reporter.i_not_fresh - return refs.represented_branch(branch) + icon + return f"{refs.represented_branch(branch, cleanable=cleanable)} {icon}" def status_details(self, indent=" ") -> str: result = [] refs = self.lazy_refs - orphan_branches = [branch for branch in refs.orphan_branches if branch != refs.current and refs.is_orphan_branch(branch)] - if orphan_branches: - result.append(f"Orphan branches: {', '.join(refs.represented_branch(branch) for branch in orphan_branches)}") + cleanable_branches = [info for info in self.branch_infos if info.name != refs.current and info.cleanable] + if cleanable_branches: + names = ", ".join(refs.represented_branch(info.name, cleanable=True) for info in cleanable_branches) + result.append(f"Cleanable branches: {names}") status = self.lazy_status result.extend(self._detail_lines(status.modified, Reporter.index_change, Reporter.worktree_change)) @@ -417,6 +440,7 @@ def fetch_now(self, *, abort_on_failure=False) -> GitRunReport: proc = self.run_git_command("fetch", "--all", "--prune", exit_codes=exit_codes) self._status = None self._refs = None + self._branch_infos = None if proc.returncode == 0: self.age = 0 @@ -434,6 +458,7 @@ def checkout_default_branch(self) -> GitRunReport: report.add(progress=f"checked out {refs.represented_branch(branch)}") self._status = None self._refs = None + self._branch_infos = None return report @@ -458,6 +483,7 @@ def pull(self, *, abort_on_failure=False) -> GitRunReport: proc = self.run_git_command("pull", "--rebase", exit_codes=exit_codes) self._status = None self._refs = None + self._branch_infos = None report = GitRunReport(note=f"was {note}") if proc.returncode: report.cant_pull(compact_git_error(proc)) @@ -492,6 +518,60 @@ def lazy_refs(self) -> GitRefs: return self._refs + def branch_info(self, name: str) -> BranchInfo: + """Summarize the local branch and its configured upstream.""" + refs = self.lazy_refs + upstream = refs.upstreams.get(name) + has_local = name in refs.local + has_remote = bool(upstream and refs.has_remote_branch(upstream.remote, upstream.branch)) + local_cleanup = self.cleanable_local_branch(name, include_current=True) if has_local else None + remote_cleanup = self.cleanable_remote_branch(upstream) if has_remote else None + return BranchInfo(name, has_local, upstream, has_remote, local_cleanup, remote_cleanup) + + @property + def branch_infos(self) -> list[BranchInfo]: + """Local branches together with tracked remote presence and cleanup proof.""" + if self._branch_infos is None: + self._branch_infos = [self.branch_info(name) for name in sorted(self.lazy_refs.local)] + + return self._branch_infos + + def branch_symbols(self, info: BranchInfo) -> str: + """Compact branch presence and cleanup symbols for display.""" + refs = self.lazy_refs + if info.name == refs.default_branch: + return "" + + return "".join( + ( + Reporter.i_local if info.has_local else "", + Reporter.i_remote if info.has_remote else "", + Reporter.i_cleanable if info.cleanable else "", + ) + ) + + def annotated_branch(self, width: int, info: BranchInfo) -> str: + refs = self.lazy_refs + symbols = self.branch_symbols(info) + padding = " " * (width - len(info.name) - len(symbols)) + name = f"{refs.represented_branch(info.name, cleanable=info.cleanable)}{symbols}{padding}" + result = ["*" if info.name == refs.current else " ", name] + if info.name == refs.default_branch: + result.append(Reporter.branch_default("[default]")) + + if info.name == refs.current: + result.append("[current]") + + if info.cleanable: + result.append(Reporter.branch_cleanable("[cleanable]")) + + return Reporter.joined(result) + + def branch_details(self, indent="") -> str: + infos = self.branch_infos or [self.branch_info(self.lazy_refs.current)] + width = max(len(info.name) + len(self.branch_symbols(info)) for info in infos) + return Reporter.joined_lines((self.annotated_branch(width, info) for info in infos), indent=indent) + def delete_local_branch(self, cleanup: CleanableLocalBranch): args = ["branch", "--delete", cleanup.name] if cleanup.force_delete: @@ -499,6 +579,7 @@ def delete_local_branch(self, cleanup: CleanableLocalBranch): self.checked_git_command(*args) self._refs = None + self._branch_infos = None def delete_remote_branch(self, cleanup: CleanableRemoteBranch): branch_ref = f"refs/heads/{cleanup.branch}" @@ -510,6 +591,7 @@ def delete_remote_branch(self, cleanup: CleanableRemoteBranch): branch_ref, ) self._refs = None + self._branch_infos = None def is_ancestor(self, ref: str, target_ref: str) -> bool: """ @@ -553,10 +635,9 @@ def cleanable_local_branch(self, name: str, include_current=False) -> CleanableL return None - def cleanable_current_remote_branch(self) -> CleanableRemoteBranch | None: - """Return cleanup details for the current branch's safely deletable origin ref.""" + def cleanable_remote_branch(self, upstream: BranchUpstream | None) -> CleanableRemoteBranch | None: + """Return cleanup details for a safely deletable tracked origin ref.""" refs = self.lazy_refs - upstream = refs.upstreams.get(refs.current) default_branch = refs.default_branch if ( not upstream @@ -575,9 +656,10 @@ def cleanable_current_remote_branch(self) -> CleanableRemoteBranch | None: expected_oid = self.checked_git_command("rev-parse", branch_ref) return CleanableRemoteBranch(upstream.remote, upstream.branch, expected_oid) - def _local_cleanable_branches(self) -> set[str]: - """Local branches that can be cleaned""" - return {name for name in self.lazy_refs.local if self.cleanable_local_branch(name)} + def cleanable_current_remote_branch(self) -> CleanableRemoteBranch | None: + """Return cleanup details for the current branch's safely deletable origin ref.""" + refs = self.lazy_refs + return self.cleanable_remote_branch(refs.upstreams.get(refs.current)) class GitStatus: diff --git a/tests/test_cli.py b/tests/test_cli.py index 2fc632a..9eec592 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -14,6 +14,7 @@ def test_command_help(cli): assert cli.succeeded assert "Commands:" in cli.logged assert "status (s)" in cli.logged + assert "legend (l)" in cli.logged cli.run("s --help") assert cli.succeeded @@ -27,6 +28,7 @@ def test_cli_arg_normalization(): assert normalized_cli_args(["workspace"]) == ["status", "workspace"] assert normalized_cli_args(["f", "workspace"]) == ["fetch", "workspace"] assert normalized_cli_args(["--color=always", "g", "checkout"]) == ["--color=always", "groom", "checkout"] + assert normalized_cli_args(["l"]) == ["legend"] def test_abort_reporting(cli, git): @@ -119,19 +121,19 @@ def test_status_line_partitions_other_branches(cli, git): assert "foo 🪦 [+1🪦+1]" in cli.logged -def test_dirty_orphan_status_keeps_tombstone(cli, git): +def test_dirty_cleanable_status_keeps_tombstone(cli, git): repo = git.seeded("repo") - repo.checkout("-b", "orphan") - repo.push("-u", "origin", "orphan") + repo.checkout("-b", "cleanable") + repo.push("-u", "origin", "cleanable") repo.checkout("main") - repo.push("origin", "--delete", "orphan") - repo.checkout("orphan") + repo.push("origin", "--delete", "cleanable") + repo.checkout("cleanable") repo.write_file("scratch.txt", "pending") cli.run("repo") assert cli.succeeded - assert "orphan 🪦 🆕1" in cli.logged + assert "cleanable 🪦 🆕1" in cli.logged def test_verbose_enables_debug_logging(cli, git): @@ -146,10 +148,25 @@ def test_verbose_enables_debug_logging(cli, git): assert "DEBUG Running:" in cli.logged +def test_legend_describes_display_symbols(cli): + cli.run("legend") + + assert cli.succeeded + assert "✅ repo freshly fetched (within 30 seconds)" in cli.logged + assert "☑️ repo not freshly fetched (more than 30 seconds ago)" in cli.logged + assert "⌛ repo fetch notably stale (more than 12 hours ago)" in cli.logged + assert "💾 branch present locally" in cli.logged + assert "☁️ configured upstream present in fetched remote refs" in cli.logged + assert "🪦 cleanable branch" in cli.logged + assert "👻 detached HEAD" in cli.logged + assert "[+N🪦+N] cleanable and remaining local branches" in cli.logged + + def test_branches_workspace_lists_local_branches(cli, git): git.init("workspace/one") two = git.init("workspace/two") two.checkout("-b", "topic") + two.commit_file("topic.txt", "in progress", "topic work") cli.run("branches workspace") @@ -158,28 +175,90 @@ def test_branches_workspace_lists_local_branches(cli, git): assert "one:" in cli.logged assert "two:" in cli.logged assert "* topic" in cli.logged + assert "topic💾" in cli.logged + assert "[current]" in cli.logged assert "[default]" in cli.logged - assert "[orphaned]" in cli.logged + assert "[cleanable]" not in cli.logged + + +def test_branches_can_append_legend(cli, git): + git.init("repo") + + cli.run("branches --legend repo") + assert cli.succeeded + assert "* main [default] [current]\n\n✅ repo freshly fetched" in str(cli.logged.stdout) + + cli.run("b -l repo") + assert cli.succeeded + assert "💾 branch present locally" in cli.logged + + +def test_branches_distinguish_cleanable_and_in_progress_refs(cli, git): + repo = git.seeded("repo") + repo.checkout("-b", "done") + repo.commit_file("done.txt", "complete", "done work") + repo.push("-u", "origin", "done") + repo.checkout("main") + repo.run_git("merge", "--no-ff", "done", "-m", "merge done") + repo.push("origin", "main") + repo.checkout("-b", "remote-work", "main") + repo.commit_file("remote.txt", "in progress", "remote work") + repo.push("-u", "origin", "remote-work") + repo.checkout("-b", "local-work", "main") + repo.commit_file("local.txt", "not pushed", "local work") + + cli.run("branches repo") + + assert cli.succeeded + lines = { + name: next(line for line in str(cli.logged.stdout).splitlines() if name in line) for name in ("done", "local-work", "remote-work") + } + assert "done💾☁️🪦" in lines["done"] + assert "[cleanable]" in lines["done"] + assert "* local-work💾" in lines["local-work"] + assert "[current]" in lines["local-work"] + assert "☁️" not in lines["local-work"] + assert "🪦" not in lines["local-work"] + assert "remote-work💾☁️" in lines["remote-work"] + assert "🪦" not in lines["remote-work"] + + +def test_unmerged_branch_with_gone_upstream_is_not_cleanable(cli, git): + repo = git.seeded("repo") + repo.checkout("-b", "not-done") + repo.commit_file("work.txt", "in progress", "unmerged work") + repo.push("-u", "origin", "not-done") + repo.checkout("main") + repo.push("origin", "--delete", "not-done") + repo.checkout("not-done") + + cli.run("repo") + + assert cli.succeeded + assert "not-done ✅" in cli.logged + assert "🪦" not in cli.logged def test_branch_names_are_styled_consistently(cli, git): repo = git.seeded("repo") repo.checkout("-b", "topic") + repo.commit_file("topic.txt", "in progress", "topic work") repo.push("-u", "origin", "topic") - repo.checkout("-b", "orphan") - repo.push("-u", "origin", "orphan") + repo.checkout("main") + repo.checkout("-b", "cleanable") + repo.push("-u", "origin", "cleanable") repo.checkout("topic") - repo.push("origin", "--delete", "orphan") + repo.push("origin", "--delete", "cleanable") with runez.ActivateColors(True): default_name = runez.bold(runez.green("main")) topic_name = runez.bold("topic") - orphan_name = runez.bold(runez.orange("orphan")) + cleanable_name = runez.bold(runez.orange("cleanable")) cli.run("--color always branches repo") assert cli.succeeded assert default_name in cli.logged assert topic_name in cli.logged - assert orphan_name in cli.logged + assert cleanable_name in cli.logged cli.run("--color always repo") assert cli.succeeded @@ -194,14 +273,14 @@ def test_branch_names_are_styled_consistently(cli, git): assert cli.succeeded assert f"checked out {default_name}" in cli.logged - repo.checkout("orphan") + repo.checkout("cleanable") cli.run("--color always repo") assert cli.succeeded - assert f"{orphan_name} 🪦" in cli.logged + assert f"{cleanable_name} 🪦" in cli.logged cli.run("--color always fetch repo") assert cli.succeeded - assert f"{orphan_name} 🪦" in cli.logged + assert f"{cleanable_name} 🪦" in cli.logged def test_pull_workspace_recaps_each_checkout(cli, git): From a82d274c846f9446f420f246bfb6441b16441d8b Mon Sep 17 00:00:00 2001 From: Zoran Simic Date: Mon, 25 May 2026 22:51:53 -0700 Subject: [PATCH 4/4] Use BranchInfo extensively --- AGENTS.md | 10 +- docs/design.md | 32 +++-- src/mgit/cli.py | 5 +- src/mgit/git.py | 333 +++++++++++++++++++++++++++++----------------- tests/test_cli.py | 98 ++++++++++++++ 5 files changed, 343 insertions(+), 135 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index fc5d552..64c3aa6 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -20,15 +20,19 @@ Important files: Use these from the repo root: -- `.venv/bin/pytest -q` for fast iteration. +- `ruff check`, `ruff format --diff`, and `.venv/bin/pytest` are the normal + validation loop for changes. Run these before handing work back. +- `.venv/bin/pytest -q` is also useful for compact fast iteration. - `.venv/bin/pytest -vv` when you want the full test names. - `.venv/bin/pytest -vv --cov=src tests` for local coverage checks. - `ruff check` and `ruff format --diff` directly; `ruff` is on `PATH`. + +The following do similar work as above, just tox wrapped (no need to use these usually) - `tox -e style` for the packaged style environment. - `tox -e py39` for the minimum supported Python version. - `tox -e py314` for the newest supported Python version. -- `tox` as the final confidence run; it exercises multiple Python versions, - coverage, and linters. +- `tox` for occasional deeper confidence runs; the user and CI run it + periodically, so do not run it for every ordinary iteration unless asked. `uv run pytest ...` also works when the synced environment is desired, but the local `.venv/bin/pytest` path is usually faster while iterating. diff --git a/docs/design.md b/docs/design.md index fcf4d39..ef69725 100644 --- a/docs/design.md +++ b/docs/design.md @@ -16,18 +16,30 @@ otherwise require repeated subprocess calls. conditions such as a gone upstream; command flows combine that information with `GitRefs` where needed. -- `GitDir.lazy_refs`: A `GitRefs` snapshot of the current branch, local branches, - configured remotes, remote-tracking branches, remote default branches, and - upstream configuration. +- `GitDir.lazy_refs`: A `GitRefs` snapshot centered on + `GitRefs.all_branches: dict[str, BranchInfo]`. Local branches are keyed by + their local name; unpaired remote-tracking branches use a qualified key such + as `origin/topic`. Each `BranchInfo` carries raw local/tracked-remote refs, + OIDs, trees, upstream configuration, protection, and evaluated cleanup + proofs, plus its owning `GitRefs` snapshot for lazily derived ref metadata + such as protection. The snapshot in turn owns its `GitDir`, which branches + use for exceptional proof commands. A local branch and its tracked remote + share one `BranchInfo`; a remote branch with no matching local upstream gets + a qualified entry instead of a parallel remote-ref index. Plain inspection + does not run `git remote`; the pull flow queries configured remotes only + when it needs to distinguish a missing remote. - `GitRefs.default_branch`: The default branch derived for a particular refs snapshot, preferring `origin/HEAD` and falling back to a visible `main` or `master` branch. -- `GitDir.branch_infos`: Derived branch display data combining a local ref, - its configured upstream presence, and local/remote cleanup proofs. It is - invalidated together with `GitDir.lazy_refs`, because either ref movement or - fetch/prune results can change whether a branch is cleanable. +- `BranchInfo.cleanable`, `local_cleanup`, and `remote_cleanup`: Lazily ask + their owning `GitRefs` snapshot to fill merge and cleanup proof fields. + Normal ancestry-merged branches are discovered in one + `git for-each-ref --merged=` query. Candidates not found there require + individual `git merge-tree` checks to retain squash-merge/content + equivalence support. This remains lazy because flows such as checkout and + pull may inspect and then immediately invalidate a refs snapshot. ## Invalidation Requirements @@ -41,9 +53,9 @@ The operations below may run during one lifetime of a `GitDir` instance: | delete local branch | unchanged in the `groom` flow | stale | The local branch collection changes; deletion occurs after checkout. | | delete remote branch | unchanged in the `groom` flow | stale | The remote-tracking branch collection changes. | -`default_branch` shares the lifetime of its owning `GitRefs` snapshot. -Replacing stale `refs` also discards derived `branch_infos` so cleanup state -is recomputed from the fresh snapshot. +`default_branch` and evaluated `BranchInfo` cleanup state share the lifetime +of their owning `GitRefs` snapshot. Replacing stale `refs` discards the branch +map so raw refs and cleanup state are recomputed from the fresh snapshot. ## Current Implementation Note diff --git a/src/mgit/cli.py b/src/mgit/cli.py index ad8abc4..e385ff4 100644 --- a/src/mgit/cli.py +++ b/src/mgit/cli.py @@ -264,9 +264,10 @@ def run_single(self, git: GitDir): if not local_cleanup: Reporter.abort(f"Branch {refs.represented_branch(current_branch)} can't be cleaned") - upstream = refs.upstreams.get(current_branch) + current_info = refs.local_branch(current_branch) + upstream = current_info.upstream if current_info else None remote_cleanup = None - if upstream and refs.has_remote_branch(upstream.remote, upstream.branch): + if upstream and current_info and current_info.has_remote: remote_cleanup = git.cleanable_current_remote_branch() if not remote_cleanup: Reporter.abort(f"Remote branch {upstream.remote}/{refs.represented_branch(upstream.branch)} can't be cleaned automatically") diff --git a/src/mgit/git.py b/src/mgit/git.py index 1c945d9..2b2e1cc 100644 --- a/src/mgit/git.py +++ b/src/mgit/git.py @@ -208,63 +208,174 @@ class CleanableRemoteBranch: expected_oid: str -@dataclass(frozen=True) +@dataclass class BranchInfo: - """Presence and cleanup proof for one displayed local branch.""" + """Raw refs and evaluated cleanup state for one related branch pair.""" + parent: GitRefs name: str - has_local: bool + local_ref: str | None = None + local_oid: str | None = None + local_tree: str | None = None upstream: BranchUpstream | None = None - has_remote: bool = False - local_cleanup: CleanableLocalBranch | None = None - remote_cleanup: CleanableRemoteBranch | None = None + remote: str | None = None + remote_branch: str | None = None + remote_ref: str | None = None + remote_oid: str | None = None + remote_tree: str | None = None + local_merged: bool | None = None + remote_merged: bool | None = None + local_equivalent: bool | None = None + remote_equivalent: bool | None = None + _local_cleanup: CleanableLocalBranch | None = None + _remote_cleanup: CleanableRemoteBranch | None = None + + @property + def has_local(self) -> bool: + return bool(self.local_ref) + + @property + def has_remote(self) -> bool: + return bool(self.remote_ref) + + @property + def local_cleanup(self) -> CleanableLocalBranch | None: + """Cleanup proof for the local ref, evaluated only when needed.""" + if self.has_local and not self.protected: + self.parent.evaluate_cleanability() + + return self._local_cleanup + + @property + def remote_cleanup(self) -> CleanableRemoteBranch | None: + """Cleanup proof for the tracked remote ref, evaluated only when needed.""" + if self.has_remote and not self.protected: + self.parent.evaluate_cleanability() + + return self._remote_cleanup @property def cleanable(self) -> bool: """True if each present ref represented by this branch can be removed.""" + if not self.has_local or self.protected: + return False + return bool(self.local_cleanup and (not self.has_remote or self.remote_cleanup)) + @property + def preferred_ref(self) -> str | None: + """Prefer the fetched remote ref when this branch has one.""" + return self.remote_ref or self.local_ref + + @property + def preferred_tree(self) -> str | None: + """Tree corresponding to `preferred_ref`.""" + return self.remote_tree if self.remote_ref else self.local_tree + + @cached_property + def protected(self) -> bool: + """True if branch is protected by the owning refs snapshot.""" + return bool(self.name and (self.name == self.parent.default_branch or self.name in self.parent.default_branches.values())) + + def tracks(self, remote: str, branch: str) -> bool: + return bool(self.upstream and self.upstream.remote == remote and self.upstream.branch == branch) + + def attach_remote(self, remote: str, branch: str, refname: str, oid: str, tree: str | None): + self.remote = remote + self.remote_branch = branch + self.remote_ref = refname + self.remote_oid = oid + self.remote_tree = tree + + def evaluate_cleanability(self, base: BranchInfo, merged_refs: set[str]): + """Populate merge and cleanup proof state for this branch.""" + base_ref = base.preferred_ref + if not base_ref: + return + + base_tree = base.preferred_tree + self.local_merged = bool(self.local_ref and self.local_ref in merged_refs) if self.has_local else None + self.remote_merged = bool(self.remote_ref and self.remote_ref in merged_refs) if self.has_remote else None + if self.has_local and not self.protected: + if self.local_merged: + self._local_cleanup = CleanableLocalBranch(self.name) + + elif self.local_ref: + self.local_equivalent = self.parent.parent.merge_is_noop(self.local_ref, base_ref, target_tree=base_tree) + if self.local_equivalent: + self._local_cleanup = CleanableLocalBranch(self.name, force_delete=True) + + if not self._eligible_remote_cleanup(base): + return + + if not self.remote or not self.remote_branch or not self.remote_oid: + return + + if self.remote_merged: + self._remote_cleanup = CleanableRemoteBranch(self.remote, self.remote_branch, self.remote_oid) + + elif self.has_local and self.remote_ref: + if self.local_oid == self.remote_oid: + self.remote_equivalent = self.local_equivalent + + else: + self.remote_equivalent = self.parent.parent.merge_is_noop(self.remote_ref, base_ref, target_tree=base_tree) + + if self.remote_equivalent: + self._remote_cleanup = CleanableRemoteBranch(self.remote, self.remote_branch, self.remote_oid) + + def _eligible_remote_cleanup(self, base: BranchInfo) -> bool: + is_origin_feature = self.remote == "origin" and self.remote_branch != self.parent.default_branch + return bool(self.remote_ref and is_origin_feature and base.remote_ref and self.remote_oid) + class GitRefs: - """Repository ref and upstream snapshot.""" + """Repository branch/ref snapshot with lazily evaluated cleanup proofs.""" + parent: GitDir current: str detached: bool - local: set[str] - remotes: list[str] - by_remote: dict[str, set[str]] default_branches: dict[str, str] - upstreams: dict[str, BranchUpstream] + all_branches: dict[str, BranchInfo] def __init__(self, parent: GitDir): - r = parent.run_git_command("symbolic-ref", "--quiet", "--short", "HEAD", exit_codes=(0, 1)) - self.detached = r.returncode != 0 - self.current = r.stdout.strip() if r.returncode == 0 else "HEAD" - self.local = set() - self.remotes = parent.checked_git_command_lines("remote") - self.by_remote = {} + self.parent = parent + self.detached = True + self.current = "HEAD" self.default_branches = {} - self.upstreams = {} + self.all_branches = {} + self._cleanup_evaluated = False lines = parent.checked_git_command_lines( "for-each-ref", - "--format=%(refname)%09%(upstream:remotename)%09%(upstream:remoteref)%09%(symref)", + "--sort=refname", + "--format=%(refname)%09%(objectname)%09%(tree)%09%(HEAD)%09%(upstream:remotename)%09%(upstream:remoteref)%09%(symref)", "refs/heads", "refs/remotes", ) + # Sorted local refs precede remote refs, so tracked remotes attach in place. for line in lines: self._add_line(line) + if self.detached and not self.local_branch_names: + r = parent.run_git_command("symbolic-ref", "--quiet", "--short", "HEAD", exit_codes=(0, 1)) + if r.returncode == 0: + self.detached = False + self.current = r.stdout.strip() + def _add_line(self, line: str): - fields = (line + "\t" * 3).split("\t") - refname, upstream_remote, upstream_ref, symref = fields[:4] + fields = (line + "\t" * 6).split("\t") + refname, oid, tree, head, upstream_remote, upstream_ref, symref = fields[:7] if refname.startswith(LOCAL_REF_PREFIX): local_branch_name = refname[len(LOCAL_REF_PREFIX) :] - self.local.add(local_branch_name) - + info = BranchInfo(self, local_branch_name, local_ref=refname, local_oid=oid, local_tree=tree or None) if upstream_remote and upstream_ref: upstream_branch = upstream_ref[len(LOCAL_REF_PREFIX) :] if upstream_ref.startswith(LOCAL_REF_PREFIX) else upstream_ref - upstream = BranchUpstream(remote=upstream_remote, branch=upstream_branch) - self.upstreams[local_branch_name] = upstream + info.upstream = BranchUpstream(remote=upstream_remote, branch=upstream_branch) + + self.all_branches[local_branch_name] = info + if head == "*": + self.detached = False + self.current = local_branch_name elif refname.startswith(REMOTE_REF_PREFIX): remote, _, branch = refname[len(REMOTE_REF_PREFIX) :].partition("/") @@ -275,42 +386,79 @@ def _add_line(self, line: str): self.default_branches[remote] = symref[len(prefix) :] else: - self.by_remote.setdefault(remote, set()).add(branch) + matching = [info for info in self.local_branches if info.tracks(remote, branch)] + if matching: + for info in matching: + info.attach_remote(remote, branch, refname, oid, tree or None) + + else: + info = BranchInfo(self, branch) + info.attach_remote(remote, branch, refname, oid, tree or None) + key = f"{remote}/{branch}" + while key in self.all_branches: + key = f"remote:{key}" - def has_remote_branch(self, remote: str, branch: str) -> bool: - remote_branches = self.by_remote.get(remote) - return bool(remote_branches and (branch in remote_branches)) + self.all_branches[key] = info + + @property + def local_branch_names(self) -> list[str]: + return sorted(info.name for info in self.all_branches.values() if info.has_local) + + @property + def local_branches(self) -> list[BranchInfo]: + return [self.all_branches[name] for name in self.local_branch_names] + + def local_branch(self, name: str | None) -> BranchInfo | None: + info = self.all_branches.get(name or "") + return info if info and info.has_local else None + + def remote_branch(self, remote: str, branch: str) -> BranchInfo | None: + return next( + (info for info in self.all_branches.values() if info.remote == remote and info.remote_branch == branch and info.has_remote), + None, + ) def upstream_gone(self, branch=None) -> bool: - upstream = self.upstreams.get(branch or self.current) - return bool(upstream and not self.has_remote_branch(upstream.remote, upstream.branch)) + info = self.local_branch(branch or self.current) + return bool(info and info.upstream and not info.has_remote) @cached_property def default_branch(self) -> str: """Default branch name.""" branch = self.default_branches.get("origin") if not branch: - all_branches = self.local - origin_branches = self.by_remote.get("origin") - if origin_branches: - all_branches = all_branches | origin_branches - + origin_names = {info.remote_branch for info in self.all_branches.values() if info.remote == "origin" and info.remote_branch} + all_branches = set(self.local_branch_names) | origin_names branch = next((name for name in ("main", "master") if name in all_branches), "main") return branch - def is_protected_branch(self, name: str) -> bool: - """True if branch should not be cleaned.""" - return bool(name and (name == self.default_branch or name in self.default_branches.values())) - - def cleanable_base_ref(self) -> str: - """Ref that cleanup candidates must already be merged into.""" - base_ref = self.default_branch - remote_branches = self.by_remote.get("origin") - if remote_branches and base_ref in remote_branches: - base_ref = f"origin/{base_ref}" - - return base_ref + def cleanable_base(self) -> BranchInfo | None: + """Default branch ref used as the cleanup target, preferring origin.""" + return self.remote_branch("origin", self.default_branch) or self.local_branch(self.default_branch) + + def evaluate_cleanability(self): + """Populate branch merge and cleanup proof state once for this snapshot.""" + if self._cleanup_evaluated: + return + + self._cleanup_evaluated = True + base = self.cleanable_base() + if not base or not base.preferred_ref: + return + + base_ref = base.preferred_ref + merged_refs = set( + self.parent.checked_git_command_lines( + "for-each-ref", + f"--merged={base_ref}", + "--format=%(refname)", + "refs/heads", + "refs/remotes", + ) + ) + for info in self.all_branches.values(): + info.evaluate_cleanability(base, merged_refs) def represented_branch(self, name: str, *, cleanable=False) -> str: """Styled representation of a branch name.""" @@ -335,7 +483,6 @@ def __init__(self, path: Path): self.age = self._current_age() self._status: GitStatus | None = None self._refs: GitRefs | None = None - self._branch_infos: list[BranchInfo] | None = None @staticmethod def _detail_lines(items, state_style, worktree_style=None) -> list[str]: @@ -360,7 +507,7 @@ def status_line(self, report: GitRunReport | None = None) -> str: if self.age is not None and self.age > FRESHNESS_THRESHOLD: report.add(note=f"{Reporter.i_stale}{runez.represented_duration(self.age)}") - other_branches = [info for info in self.branch_infos if info.name not in (refs.current, refs.default_branch)] + other_branches = [info for info in refs.local_branches if info.name not in (refs.current, refs.default_branch)] cleanable = sum(info.cleanable for info in other_branches) remaining = len(other_branches) - cleanable branch_summary = Reporter.joined(cleanable and f"+{cleanable}{Reporter.i_cleanable}", remaining and f"+{remaining}", separator="") @@ -383,7 +530,7 @@ def represented_current_branch(self) -> str: if refs.detached: return f"{refs.represented_branch(branch)} {Reporter.i_detached}" - info = next((candidate for candidate in self.branch_infos if candidate.name == branch), None) + info = refs.local_branch(branch) cleanable = bool(info and info.cleanable) if cleanable: icon = Reporter.i_cleanable @@ -399,7 +546,7 @@ def represented_current_branch(self) -> str: def status_details(self, indent=" ") -> str: result = [] refs = self.lazy_refs - cleanable_branches = [info for info in self.branch_infos if info.name != refs.current and info.cleanable] + cleanable_branches = [info for info in refs.local_branches if info.name != refs.current and info.cleanable] if cleanable_branches: names = ", ".join(refs.represented_branch(info.name, cleanable=True) for info in cleanable_branches) result.append(f"Cleanable branches: {names}") @@ -440,7 +587,6 @@ def fetch_now(self, *, abort_on_failure=False) -> GitRunReport: proc = self.run_git_command("fetch", "--all", "--prune", exit_codes=exit_codes) self._status = None self._refs = None - self._branch_infos = None if proc.returncode == 0: self.age = 0 @@ -458,16 +604,15 @@ def checkout_default_branch(self) -> GitRunReport: report.add(progress=f"checked out {refs.represented_branch(branch)}") self._status = None self._refs = None - self._branch_infos = None return report def pull(self, *, abort_on_failure=False) -> GitRunReport: """Pull from tracked remote""" - refs = self.lazy_refs - if not refs.remotes: + if not self.checked_git_command_lines("remote"): return GitRunReport().cant_pull("no remotes") + refs = self.lazy_refs if refs.detached: return GitRunReport().cant_pull("HEAD detached") @@ -483,7 +628,6 @@ def pull(self, *, abort_on_failure=False) -> GitRunReport: proc = self.run_git_command("pull", "--rebase", exit_codes=exit_codes) self._status = None self._refs = None - self._branch_infos = None report = GitRunReport(note=f"was {note}") if proc.returncode: report.cant_pull(compact_git_error(proc)) @@ -518,24 +662,6 @@ def lazy_refs(self) -> GitRefs: return self._refs - def branch_info(self, name: str) -> BranchInfo: - """Summarize the local branch and its configured upstream.""" - refs = self.lazy_refs - upstream = refs.upstreams.get(name) - has_local = name in refs.local - has_remote = bool(upstream and refs.has_remote_branch(upstream.remote, upstream.branch)) - local_cleanup = self.cleanable_local_branch(name, include_current=True) if has_local else None - remote_cleanup = self.cleanable_remote_branch(upstream) if has_remote else None - return BranchInfo(name, has_local, upstream, has_remote, local_cleanup, remote_cleanup) - - @property - def branch_infos(self) -> list[BranchInfo]: - """Local branches together with tracked remote presence and cleanup proof.""" - if self._branch_infos is None: - self._branch_infos = [self.branch_info(name) for name in sorted(self.lazy_refs.local)] - - return self._branch_infos - def branch_symbols(self, info: BranchInfo) -> str: """Compact branch presence and cleanup symbols for display.""" refs = self.lazy_refs @@ -568,7 +694,8 @@ def annotated_branch(self, width: int, info: BranchInfo) -> str: return Reporter.joined(result) def branch_details(self, indent="") -> str: - infos = self.branch_infos or [self.branch_info(self.lazy_refs.current)] + refs = self.lazy_refs + infos = refs.local_branches or [BranchInfo(refs, refs.current)] width = max(len(info.name) + len(self.branch_symbols(info)) for info in infos) return Reporter.joined_lines((self.annotated_branch(width, info) for info in infos), indent=indent) @@ -579,7 +706,6 @@ def delete_local_branch(self, cleanup: CleanableLocalBranch): self.checked_git_command(*args) self._refs = None - self._branch_infos = None def delete_remote_branch(self, cleanup: CleanableRemoteBranch): branch_ref = f"refs/heads/{cleanup.branch}" @@ -591,24 +717,14 @@ def delete_remote_branch(self, cleanup: CleanableRemoteBranch): branch_ref, ) self._refs = None - self._branch_infos = None - def is_ancestor(self, ref: str, target_ref: str) -> bool: - """ - :param str ref: Candidate `ref` - :param str target_ref: Ref that should contain the candidate - :return bool: True if 'ref' is an ancestor of 'target_ref' - """ - proc = self.run_git_command("merge-base", "--is-ancestor", ref, target_ref, exit_codes=(0, 1)) - return proc.returncode == 0 - - def merge_is_noop(self, ref: str, target_ref: str) -> bool: + def merge_is_noop(self, ref: str, target_ref: str, *, target_tree: str | None = None) -> bool: """ :param str ref: Candidate `ref` :param str target_ref: Ref that should already contain the candidate content :return bool: True if merging 'ref' into `target_ref` would leave `target_ref` unchanged """ - target_tree = self.checked_git_command("rev-parse", f"{target_ref}^{{tree}}") + target_tree = target_tree or self.checked_git_command("rev-parse", f"{target_ref}^{{tree}}") proc = self.run_git_command("merge-tree", "--write-tree", "--no-messages", target_ref, ref) if proc.returncode: return False @@ -623,43 +739,20 @@ def cleanable_local_branch(self, name: str, include_current=False) -> CleanableL :return CleanableLocalBranch|None: Cleanup details if local branch is safe to clean """ refs = self.lazy_refs - base_ref = refs.cleanable_base_ref() - if not base_ref or not name or refs.is_protected_branch(name) or (not include_current and name == refs.current): + info = refs.local_branch(name) + if not info or info.protected or (not include_current and name == refs.current): return None - if self.is_ancestor(name, base_ref): - return CleanableLocalBranch(name) - - if self.merge_is_noop(name, base_ref): - return CleanableLocalBranch(name, force_delete=True) - - return None - - def cleanable_remote_branch(self, upstream: BranchUpstream | None) -> CleanableRemoteBranch | None: - """Return cleanup details for a safely deletable tracked origin ref.""" - refs = self.lazy_refs - default_branch = refs.default_branch - if ( - not upstream - or upstream.remote != "origin" - or upstream.branch == default_branch - or not refs.has_remote_branch(upstream.remote, upstream.branch) - or not refs.has_remote_branch(upstream.remote, default_branch) - ): - return None - - branch_ref = f"{REMOTE_REF_PREFIX}{upstream.remote}/{upstream.branch}" - base_ref = f"{REMOTE_REF_PREFIX}{upstream.remote}/{default_branch}" - if not self.is_ancestor(branch_ref, base_ref) and not self.merge_is_noop(branch_ref, base_ref): - return None - - expected_oid = self.checked_git_command("rev-parse", branch_ref) - return CleanableRemoteBranch(upstream.remote, upstream.branch, expected_oid) + return info.local_cleanup def cleanable_current_remote_branch(self) -> CleanableRemoteBranch | None: """Return cleanup details for the current branch's safely deletable origin ref.""" refs = self.lazy_refs - return self.cleanable_remote_branch(refs.upstreams.get(refs.current)) + info = refs.local_branch(refs.current) + if info: + return info.remote_cleanup + + return None class GitStatus: diff --git a/tests/test_cli.py b/tests/test_cli.py index 9eec592..8976e0a 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -7,6 +7,7 @@ import runez from mgit.cli import normalized_cli_args +from mgit.git import GitDir, GitRefs def test_command_help(cli): @@ -193,6 +194,16 @@ def test_branches_can_append_legend(cli, git): assert "💾 branch present locally" in cli.logged +def test_branches_fall_back_to_symbolic_head_for_unborn_repo(cli, git): + git.init("repo", include_readme=False) + + cli.run("-v branches repo") + + assert cli.succeeded + assert "* main [default] [current]" in cli.logged + assert "symbolic-ref --quiet --short HEAD" in cli.logged + + def test_branches_distinguish_cleanable_and_in_progress_refs(cli, git): repo = git.seeded("repo") repo.checkout("-b", "done") @@ -222,6 +233,72 @@ def test_branches_distinguish_cleanable_and_in_progress_refs(cli, git): assert "remote-work💾☁️" in lines["remote-work"] assert "🪦" not in lines["remote-work"] + git_dir = GitDir(repo.cwd) + refs = git_dir.lazy_refs + main = refs.all_branches["main"] + assert "protected" not in main.__dict__ + assert main.protected + assert main.__dict__["protected"] + done = refs.all_branches["done"] + assert done.local_merged is None + assert done.cleanable + assert refs.local_branch_names == ["done", "local-work", "main", "remote-work"] + assert done.parent is refs + assert done.parent.parent is git_dir + assert done.local_ref == "refs/heads/done" + assert done.remote_ref == "refs/remotes/origin/done" + assert done.local_oid + assert done.remote_oid == done.local_oid + assert done.local_tree + assert done.remote_tree == done.local_tree + assert done.local_merged + assert done.remote_merged + assert main.parent is refs + assert main.remote_ref == "refs/remotes/origin/main" + assert refs.cleanable_base() is main + assert main.protected + + +def test_refs_include_remote_only_branches(cli, git): + repo = git.seeded("repo") + repo.checkout("-b", "remote-only") + repo.commit_file("remote.txt", "published", "remote work") + repo.push("-u", "origin", "remote-only") + repo.checkout("main") + repo.branch("-D", "remote-only") + + refs = GitDir(repo.cwd).lazy_refs + remote_only = refs.all_branches["origin/remote-only"] + + assert not remote_only.has_local + assert remote_only.has_remote + assert remote_only.remote == "origin" + assert remote_only.remote_ref == "refs/remotes/origin/remote-only" + + cli.run("branches repo") + assert cli.succeeded + assert "remote-only" not in cli.logged + + +def test_branches_batch_normal_merge_proofs(cli, git): + repo = git.seeded("repo") + for branch in ("done-a", "done-b"): + repo.checkout("-b", branch, "main") + + repo.checkout("main") + + cli.run("-v branches repo") + + assert cli.succeeded + output = str(cli.logged) + merged_commands = {line for line in output.splitlines() if "for-each-ref --merged=" in line} + assert len(merged_commands) == 1 + assert " git -C repo remote" not in output + assert "symbolic-ref" not in output + assert "merge-base --is-ancestor" not in output + assert "merge-tree" not in output + assert "rev-parse" not in output + def test_unmerged_branch_with_gone_upstream_is_not_cleanable(cli, git): repo = git.seeded("repo") @@ -385,6 +462,27 @@ def test_main_checks_out_default_branch(cli, git): assert repo.current_branch == "main" +def test_main_only_evaluates_cleanability_for_reported_snapshot(cli, git, monkeypatch): + repo = git.seeded("repo") + repo.checkout("-b", "feature") + evaluated = [] + evaluate_cleanability = GitRefs.evaluate_cleanability + + def track_evaluation(refs): + if not refs._cleanup_evaluated: + evaluated.append(refs.current) + + return evaluate_cleanability(refs) + + monkeypatch.setattr(GitRefs, "evaluate_cleanability", track_evaluation) + + cli.run("-v main repo") + + assert cli.succeeded + assert repo.current_branch == "main" + assert evaluated == ["main"] + + def test_main_prefers_main_over_master_without_remote_head(cli, git): repo = git.seeded("repo") repo.branch("master")