diff --git a/AGENTS.md b/AGENTS.md new file mode 100644 index 0000000..64c3aa6 --- /dev/null +++ b/AGENTS.md @@ -0,0 +1,58 @@ +# 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: + +- `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` 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. + +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/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/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..ef69725 --- /dev/null +++ b/docs/design.md @@ -0,0 +1,66 @@ +# 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.lazy_status`: A `GitStatus` snapshot produced from + `git status --porcelain=v2 --branch`. It includes pending changes and + 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.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. + +- `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 + +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 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 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 + +`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 01ca1f5..e385ff4 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: @@ -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.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: @@ -207,6 +226,22 @@ def run_multi(self, project_dir: ProjectDir): 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): @@ -214,61 +249,38 @@ 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") - remote_cleanup = None + 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) - if upstream and refs.has_remote_branch(upstream.remote, upstream.branch): + current_info = refs.local_branch(current_branch) + upstream = current_info.upstream if current_info else None + remote_cleanup = None + 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}/{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 12c6af0..2b2e1cc 100644 --- a/src/mgit/git.py +++ b/src/mgit/git.py @@ -14,13 +14,19 @@ 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.""" 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 @@ -53,37 +59,50 @@ 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", + ) -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" +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 - 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()) + 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() - else: - lines.append(line) - - return " ".join((prefixed or lines)[:2]).replace(" ", " ").strip() + return lines[0] class GitRunReport: @@ -164,11 +183,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.""" @@ -194,44 +208,174 @@ class CleanableRemoteBranch: expected_oid: str +@dataclass +class BranchInfo: + """Raw refs and evaluated cleanup state for one related branch pair.""" + + parent: GitRefs + name: str + local_ref: str | None = None + local_oid: str | None = None + local_tree: str | None = None + upstream: BranchUpstream | 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): - self.current, self.detached = parent._current_branch() - 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("/") @@ -242,15 +386,89 @@ 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) - 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)) + 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}" + + 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: + 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 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.""" + if name == self.default_branch: + name = Reporter.branch_default(name) + + elif cleanable: + name = Reporter.branch_cleanable(name) + + return runez.bold(name) class GitDir: @@ -263,12 +481,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]: @@ -286,17 +500,17 @@ 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)}") + report.add(note=f"{Reporter.i_stale}{runez.represented_duration(self.age)}") - other_branches = refs.local - {refs.current, self.default_branch} - cleanable = len(other_branches & self.local_cleanable_branches) + 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}🪦", 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}]" @@ -304,131 +518,119 @@ 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, ) def represented_current_branch(self) -> str: - refs = self.refs + refs = self.lazy_refs branch = refs.current if refs.detached: - icon = " 👻" + return f"{refs.represented_branch(branch)} {Reporter.i_detached}" + + info = refs.local_branch(branch) + cleanable = bool(info and info.cleanable) + if cleanable: + icon = Reporter.i_cleanable - elif self.is_orphan_branch(branch): - icon = " 🪦" + 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 self.represented_branch(branch) + icon + return f"{refs.represented_branch(branch, cleanable=cleanable)} {icon}" 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)] - if orphan_branches: - result.append(f"Orphan branches: {', '.join(self.represented_branch(branch) for branch in orphan_branches)}") + refs = self.lazy_refs + 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}") - 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 - 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") - 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 @@ -445,89 +647,84 @@ 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) + if self._status is None: + self._status = GitStatus(self) - @cached_property - 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 self._status - 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) + @property + def lazy_refs(self) -> GitRefs: + if self._refs is None: + self._refs = GitRefs(self) + + return self._refs + + 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 "", + ) + ) - if self.is_orphan_branch(name): - return runez.orange(result) + 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]")) - return result + if info.name == refs.current: + result.append("[current]") - @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}" + if info.cleanable: + result.append(Reporter.branch_cleanable("[cleanable]")) - return base_ref + return Reporter.joined(result) - 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) - Reporter.abort_if(proc.returncode > 1, f"Could not inspect git ancestry: {git_error_message(proc)}") - return proc.returncode == 0 + def branch_details(self, indent="") -> str: + 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) + + 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 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 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}}") + 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 @@ -541,47 +738,21 @@ 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 + 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 + 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.refs - upstream = refs.upstreams.get(refs.current) - default_branch = self.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 + refs = self.lazy_refs + info = refs.local_branch(refs.current) + if info: + return info.remote_cleanup - 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) - - @cached_property - def local_cleanable_branches(self) -> set[str]: - """ - :return set: Local branches that can be cleaned - """ - return {name for name in self.refs.local if self.cleanable_local_branch(name)} + return None class GitStatus: @@ -592,7 +763,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] @@ -613,9 +783,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..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): @@ -14,6 +15,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 +29,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): @@ -64,6 +67,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() @@ -116,19 +122,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): @@ -143,10 +149,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") @@ -155,28 +176,166 @@ 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_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") + 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"] + + 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") + 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.green(runez.bold("main")) + default_name = runez.bold(runez.green("main")) topic_name = runez.bold("topic") - orphan_name = runez.orange(runez.bold("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 @@ -191,14 +350,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): @@ -254,15 +413,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 +460,37 @@ def test_main_checks_out_default_branch(cli, git): assert cli.succeeded 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") + 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