diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index fa9835b..ed6c013 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -20,7 +20,7 @@ jobs: with: python-version: "3.14" - - uses: astral-sh/setup-uv@v7 + - uses: astral-sh/setup-uv@v8.1.0 - run: uvx --with tox-uv tox -e py,style - run: uv build - name: Publish to PyPI diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 2226f31..9deca78 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -21,7 +21,7 @@ jobs: with: python-version: ${{ matrix.python-version }} - - uses: astral-sh/setup-uv@v7 + - uses: astral-sh/setup-uv@v8.1.0 - run: uvx --with tox-uv tox -e py linters: @@ -34,5 +34,5 @@ jobs: with: python-version: "3.14" - - uses: astral-sh/setup-uv@v7 + - uses: astral-sh/setup-uv@v8.1.0 - run: uvx --with tox-uv tox -e style diff --git a/README.rst b/README.rst index 5f1768a..6805805 100644 --- a/README.rst +++ b/README.rst @@ -1,5 +1,7 @@ -Fetch collections of git projects -================================= +mgit +==== + +A small companion for humans who like tidy git checkouts. .. image:: https://img.shields.io/pypi/v/mgit.svg :target: https://pypi.org/project/mgit/ @@ -14,122 +16,168 @@ Fetch collections of git projects :alt: Python versions tested -Overview -======== +What it is +========== + +``mgit`` is a tiny CLI for people who like their git checkouts clean, current, +and easy to scan. It gives you a quick read on one repo, or on every repo +directly inside a workspace folder. -With ``mgit``, you can pull/fetch several projects at once, -and also auto-cleanup dangling branches (from past pull requests). +- What state did I leave this checkout in? +- Can I pull, or did I leave work pending here? +- Is a merged branch still hanging around, waiting to be cleaned? +- I just wrapped up a branch; can I safely get back to ``main`` and clear it away? -A colored output is provided if possible, ``mgit`` should come in handy in general for: +``mgit`` is not a git replacement. It is the compact, careful layer for those +ordinary moments where you would rather see the answer than reconstruct it +from a handful of commands. -- quickly getting an overview of what's up with N git projects -- fetch/pull N git objects at once -- clone missing projects (useful if you tend to clone projects from same remote in one common folder) +The rhythm +========== -Example usage -============= +Start with a glance:: -``mgit`` can show you what's the status of all your git projects in a folder, for example my repos:: + mgit # show status here, or across a workspace - ~/dev/github: mgit - ~/dev/github: 4 github/zsimic - mgit: [main] up to date - pickley: [main] 1 diff, up to date* last fetch 1w 4d ago - runez: [main] up to date* last fetch 1w 4d ago - setupmeta: [main] up to date* last fetch 3d 23h ago +Refresh or move forward only when you mean to:: + mgit f # fetch --all --prune, then show the refreshed status + mgit p # pull --rebase, but never over pending work -Here we can see that: +And for the wonderfully common ``PR merged`` moment:: -- There are 4 git repos in folder ``~/dev/github`` + mgit g # fetch, return to the default branch, pull, clean this branch -- All 4 come from ``github/zsimic`` +The short names are the intended interface: -- 3 of them haven't been fetched in a while +- ``status`` / ``s``: show whether anything here needs attention. This is the default. +- ``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. +- ``groom`` / ``g``: safely finish with the merged branch you are currently on. -We can fetch them all at once with ``--fetch`` (or ``-f``):: +The command shape is:: - ~/dev/github: mgit --fetch - ~/dev/github: 4 github/zsimic - mgit: [main] up to date - pickley: [main] 1 diff, up to date - runez: [main] behind 2 - setupmeta: [main] up to date + mgit [-v] [--color auto|always|never] [COMMAND] [FOLDER] +You can pass a folder to most commands:: -Now all projects have been refreshed, and we can see there's nothing new in 2 of them, -but one is 2 commits behind (ie: 2 commits are on the remote, not pulled yet). -The output also shows that one of the projects has uncommitted files. + mgit ~/github + mgit f ~/github + mgit g ~/github/mgit -Modified files are shown (by default) if only one project is in scope, for example:: +Workspace scans are shallow on purpose: ``mgit ~/github`` inspects direct +children like ``~/github/*/.git`` and does not crawl nested folders. - ~/dev/github: mgit pickley - pickley: [main] 1 diff, up to date - M tox.ini +What you see +============ + +A workspace stays pleasantly skimmable:: -Above, we can see that the modified file in question is ``tox.ini`` in that project. -We can get the same effect using the ``--verbose`` (or ``-v``) flag, -like for example with 2 projects with modified files:: + mgit: main ✅ + pickley: main ☑ī¸ [+1đŸĒĻ] ⌛4w 6d + runez: feature đŸĒĻ ✏ī¸1 + detached: HEAD đŸ‘ģ - ~/dev/github: mgit -v - ~/dev/github: 4 github/zsimic - mgit: [main] 1 diff, up to date - M README.rst - pickley: [main] 1 diff, up to date - M tox.ini - runez: [main] up to date - setupmeta: [main] up to date +In one glance you get: +- the current branch +- local diffs and untracked files +- ahead/behind/gone tracking state +- fetch freshness +- a compact hint that a stale local branch is still lurking -Synopsis:: +``✅`` 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 +``[+2đŸĒĻ+1]`` says that, besides the current and default branches, two local +branches have been proven cleanable and one remains. - ~/dev/github: mgit --help - Usage: mgit [OPTIONS] [TARGET] +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:: - Fetch collections of git projects + mgit ~/github/mgit - Options: - --version Show the version and exit. - --debug Show debugging information. - --color / --no-color Use colors (on by default on ttys) - --log PATH Override log file location. - --clean [show|local|remote|all|reset] - Auto-clean branches - -f, --fetch Fetch from all remotes - -p, --pull Pull from tracking remote - -s, --short / -v, --verbose Short/verbose output - -cs Handy shortcut for '--clean show' - -cl Handy shortcut for '--clean local' - -cr Handy shortcut for '--clean remote' - -ca Handy shortcut for '--clean all' - -h, --help Show this message and exit. +Color is automatic on terminals and can be controlled explicitly:: -Installation + mgit --color auto + mgit --color always + mgit --color never + + +Safety model ============ -Easiest way to get mgit is via pickley_ or pipx_:: +``mgit`` is read-only by default. Asking how things look does not change your repos. - pickley install mgit +Commands that act are explicit: + +- ``mgit f`` updates local remote refs with ``git fetch --all --prune``. +- ``mgit p`` pulls with rebase only when the checkout passes safety checks. +- ``mgit g`` fetches, verifies that the current branch can be cleaned before + switching branches, pulls safely, and deletes that local branch only. If it + still exists on ``origin``, it deletes that one remote branch only after + independently proving that the fetched remote ref is merged or + content-equivalent to the fetched default branch; an exact-ref lease prevents + deleting a branch that advanced meanwhile. No other branch is deleted. + When already on the default branch, it fetches and reports that fact without + pulling or cleaning branches. Use ``mgit m`` when you only want to switch to + the default branch. + + +Coming next +=========== + +A larger convenience waiting in the wings is ``mgit clone``: give it a full +repo URL and let ``mgit`` choose the local destination from simple config +rules. + +Planned config shape:: + + locations = [ + { match = "github.com/zsimic/*", dir = "~/github" }, + { match = "github.com/*", dir = "~/ext" }, + { match = "git.mycompany.com/*", dir = "~/dev" }, + ] +The goal is predictable placement without memorizing where each family of repos +belongs. Status, fetch, pull, main, and groom do not require any mgit config. + + +Install +======= + +Install with pickley_ or pipx_:: + + pickley install mgit or:: pipx install mgit - -You can also compile from source:: +Install from a checkout for development:: git clone https://github.com/zsimic/mgit.git cd mgit - uv venv - uv pip install -e . - + uv sync .venv/bin/mgit --help - source .venv/bin/activate - mgit --help + +Develop +======= + +Fast local checks:: + + .venv/bin/pytest -q + ruff check + +Full confidence check:: + + tox .. _pickley: https://pypi.org/project/pickley/ diff --git a/docs/contributing.rst b/docs/contributing.rst index a4a5d4f..7c1ca88 100644 --- a/docs/contributing.rst +++ b/docs/contributing.rst @@ -1,6 +1,7 @@ Contributions are welcome! -tox_ is used for building and testing, ``setup.py`` is kept simple thanks to setupmeta_. +tox_ is used for full test runs. Packaging metadata lives in +``pyproject.toml`` and versions come from ``setuptools-scm``. Development =========== @@ -10,10 +11,9 @@ To get going locally, simply do this:: git clone https://github.com/zsimic/mgit.git cd mgit - uv venv - uv pip install -r tests/requirements.txt -e . + uv sync - # You have a venv now in ./.venv, use it, open it with pycharm etc + # You have a venv now in ./.venv source .venv/bin/activate which python which mgit @@ -25,15 +25,24 @@ To get going locally, simply do this:: Running the tests ================= -To run the tests, simply run ``tox``. +Fast local checks:: -Run: + .venv/bin/pytest -q + ruff check -* ``tox -e py314`` (for example) to limit test run to only one python version. +Full confidence check:: -* ``tox -e style`` to run style checks only + tox -* etc +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 @@ -45,5 +54,3 @@ Run ``tox``, then open ``.tox/test-reports/htmlcov/index.html`` .. _pyenv: https://github.com/pyenv/pyenv .. _tox: https://github.com/tox-dev/tox - -.. _setupmeta: https://pypi.org/project/setupmeta/ diff --git a/pyproject.toml b/pyproject.toml index 1d1a24f..17a1c37 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -12,7 +12,7 @@ authors = [ ] description = "Fetch collections of git projects" readme = "README.rst" -requires-python = ">=3.10" +requires-python = ">=3.9" license = "MIT" license-files = ["LICENSE.txt"] classifiers = [ @@ -24,6 +24,7 @@ classifiers = [ "Operating System :: Unix", "Programming Language :: Python", "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.9", "Programming Language :: Python :: 3.10", "Programming Language :: Python :: 3.11", "Programming Language :: Python :: 3.12", @@ -32,7 +33,6 @@ classifiers = [ "Topic :: Utilities", ] dependencies = [ - "click<9", "runez<6", ] dynamic = ["version"] @@ -43,8 +43,16 @@ mgit = "mgit.cli:main" [project.urls] Source = "https://github.com/zsimic/mgit" +[dependency-groups] +test = [ + "pytest-cov", +] +dev = [ + { include-group = "test" }, +] + [tool.pyright] -pythonVersion = "3.10" +pythonVersion = "3.9" include = ["src"] [tool.ruff] diff --git a/src/mgit/__init__.py b/src/mgit/__init__.py index 2a18407..7d73efe 100644 --- a/src/mgit/__init__.py +++ b/src/mgit/__init__.py @@ -1,401 +1,40 @@ -import collections -import logging -import os - -import runez - -from mgit.git import GitDir, GitRunReport - -LOG = logging.getLogger(__name__) - - -def git_parent_path(path): - """ - :param str path: Path to search - :return str|None: First parent of 'path' that has a .git folder, if any - """ - if not path or len(path) <= 5: - return None - - if os.path.isdir(os.path.join(path, ".git")): - return path - - return git_parent_path(os.path.dirname(path)) - - -def find_actual_path(path): - """ - :param str|None path: Base path, if None or '.': look for first parent folder with a .git subfolder - :return str: Actual path to use as target - """ - if not path or path == ".": - path = git_parent_path(os.getcwd()) or "." - - return runez.resolved_path(path) - - -def get_target(path, **kwargs): - """ - :param str path: Path to target - :param kwargs: Optional preferences - """ - prefs = MgitPreferences(**kwargs) - actual_path = find_actual_path(path) - if not actual_path or not os.path.isdir(actual_path): - runez.abort("No folder '%s'" % runez.short(actual_path)) - - if os.path.isdir(os.path.join(actual_path, ".git")): - return GitCheckout(actual_path, prefs=prefs) - - return ProjectDir(actual_path, prefs=prefs) - - -def print_modified(items, color1, color2=None): - for item in items: - state = item[0:2] - if color2: - state = f"{color1(item[0])}{color2(item[1])}" - - elif color1: - state = color1(state) - - print(f" {state} {item[3:]}") - - -class MgitPreferences: - """Various prefs""" - - name_size = None # How many chars to align names when displaying list of checkouts - align = True # Whether to align names or not - verbose = False # Show verbose output - all = False # Show all entries, including missing/invalid checkout folders - fetch = False # Auto-fetch before showing status - pull = False # Auto-pull before showing status - inspect_remotes = False # Inspect remote branches to report cleanable (slower) - - def __init__(self, **kwargs): - self.update(**kwargs) - - def __repr__(self): - result = [self._value_representation(k) for k in sorted(self.__dict__)] - return " ".join(s for s in result if s is not None) - - def _value_representation(self, name): - value = getattr(self, name, None) - if value is None: - return None - - if value is True: - return name - - if value is False: - return "!%s" % name - - return f"{name}={value}" - - def set_short(self, value): - """ - :param bool|None value: Value from parsed click command line: - None (default): align, verbose for individual git checkouts, short one line otherwise - True (--short): everything compact - False (--verbose): everything verbose - """ - self.align = value is None - self.verbose = value is False - - def update(self, **kwargs): - for name, value in kwargs.items(): - if hasattr(self, name): - setattr(self, name, value) - continue - - func = getattr(self, "set_%s" % name, None) - if func: - func(value) - continue - - raise Exception("Internal error: add support for flag '%s'" % name) - - -class RemoteProject: - """ - Hashable object representing a remote repo - - 'type' will indicate whether it's a stash repo, or github or other - - 'name' will correspond to project for stash, or owner for github etc - """ - - def __init__(self, url): - """ - :param GitURL url: URL of remote repo - """ - self.url = url - self.name = url.repo or "unknown" - - def __repr__(self): - return f"{self.type}/{self.name}" - - def __hash__(self): - return hash(str(self)) - - def __eq__(self, other): - return str(self) == str(other) - - def __ne__(self, other): - return str(self) != str(other) - - @classmethod - def from_url(cls, url): - if url and url.hostname: - if "stash" in url.hostname: - return StashProject(url) - - if "github" in url.hostname: - return GithubProject(url) - - return UnknownProject(url) - - @property - def type(self): - return self.__class__.__name__.lower()[:-7] - - def projects(self): - """ - :return dict: Projects on server hashed by their canonical name, if this is a stash server - """ - LOG.warning("Ignoring --all option, no implementation for listing %s projects", self.type) - return {} - - -class StashProject(RemoteProject): - """Bitbucket stash repo""" +from __future__ import annotations +from typing import TYPE_CHECKING -class GithubProject(RemoteProject): - """Github repo""" - - -class UnknownProject(RemoteProject): - """Unknown repo""" - - -class GitCheckout: - """Represents a local git checkout""" - - def __init__(self, path, parent=None, prefs=None): - """ - :param str path: Full path to local checkout - :param ProjectDir|None parent: Parent project dir - :param MgitPreferences|None prefs: Optional prefs to use - """ - # Basename of local git folder (usually matches remote repo base name) - self.basename = os.path.basename(path) - self.directory_exists = os.path.isdir(path) - self.git = GitDir(path) - self.parent = parent - self._prefs = prefs if prefs or parent else MgitPreferences() - - def __repr__(self): - return self.basename - - @property - def prefs(self): - if self.parent: - return self.parent.prefs - - return self._prefs - - @runez.cached_property - def name(self): - """ - :return str: Basename of local git folder + remote basename if it differs - """ - if not self.git.config.repo_name or not self.git.is_git_checkout or self.basename == self.git.config.repo_name: - return self.basename - - return f"{self.basename} ({self.git.config.repo_name})" - - @runez.cached_property - def origin_project(self): - return RemoteProject.from_url(self.git.config.origin) - - @runez.cached_property - def aligned_name(self): - name = self.name - if self.parent and self.parent.prefs.name_size: - name = ("%%%ss" % self.parent.prefs.name_size) % name - - return name - - def header(self, report=None): - """ - :param GitRunReport|None report: Optional report to show (defaults to self.git.report) - :return str: Textual representation - """ - report = GitRunReport(report or self.git.report(inspect_remotes=self.prefs.inspect_remotes)) - - result = "%s:" % self.aligned_name - - if self.git.is_git_checkout: - branch = runez.bold(self.git.branches.shortened_current_branch) - n = len(self.git.branches.local) - if n > 1: - branch += " +%s" % (n - 1) - - result += " [%s]" % branch - - freshness = self.git.status.freshness - if freshness: - result += " %s" % freshness - - if ( - not report.has_problems - and self.parent - and self.prefs - and self.prefs.all - and self.parent.predominant - and self.origin_project != self.parent.predominant - ): - report.add(note="not part of %s" % self.parent.predominant) - - if report: - rep = report.representation() - if rep: - result += " %s" % rep - - return result - - def apply(self): - """Apply switches as specified by prefs""" - report = GitRunReport() - if self.prefs.pull: - if self.prefs.all and not self.git.folder_exists: - if self.git.remote_info and self.git.remote_info.clone_url: - report.add(self.git.clone(self.git.remote_info.clone_url)) - - else: - return report.cant_pull("couldn't determine clone url") - - else: - report.add(self.git.pull()) - - elif self.prefs.fetch: - report.add(self.git.fetch()) - - if not report.has_problems: - report.add(self.git.report(inspect_remotes=self.prefs.inspect_remotes)) - - return report +import runez - def print_status(self): - """Show checkout status""" - report = self.apply() - print(self.header(report)) - if self.prefs.verbose or (not self.parent and self.prefs.align): - if len(self.git.orphan_branches) > 1: - print(" Orphan branches: %s" % (", ".join(self.git.orphan_branches))) +from mgit.git import GitDir, Reporter - print_modified(self.git.status.modified, runez.teal, runez.red) - print_modified(self.git.status.untracked, runez.orange) +if TYPE_CHECKING: + from pathlib import Path class ProjectDir: - """Info shown for a given directory""" + """One requested workspace, represented as one or more git dirs.""" - def __init__(self, path, prefs=None): + def __init__(self, path: Path): """ - :param str path: Path to folder - :param MgitPreferences|None prefs: Display prefs + :param Path path: Path to folder """ - self.path = path # Path to folder to examine - self.prefs = prefs or MgitPreferences() # Preferences on how to output result - self.checkouts = [] # Actual git checkouts in 'path' - self.projects = collections.defaultdict(set) # Seen remotes - self.predominant = None # Predominant remote, if any - self.additional = None # Additional projects (sorted by checkouts, descending) - self.stash_projects = {} # Corresponding projects from stash, when applicable + self.path = path + self.git_dirs: list[GitDir] = [] + self.name_size: int | None = None self.scan() - def __repr__(self): - return os.path.basename(self.path) - def scan(self): - self.checkouts = [] - for fname in os.listdir(self.path): - if fname and fname.startswith("."): - continue - - source_path = os.path.join(self.path, fname) - if os.path.isdir(source_path): - r = GitCheckout(source_path, parent=self) - self.checkouts.append(r) - if r.git.is_git_checkout: - self.projects[r.origin_project].add(r) - - self.predominant = None - self.additional = None - counts = [(project, len(self.projects[project])) for project in sorted(self.projects, key=lambda x: -len(self.projects[x]))] - if counts: - self.additional = [t[0] for t in counts] - top, top_count = counts.pop(0) - threshold = top_count // 2 - if not counts or all(t[1] <= threshold for t in counts): - self.predominant = top - self.additional = self.additional[1:] - - if not self.prefs.all or not self.predominant: - self.stash_projects = {} - - else: - self.stash_projects = self.predominant.projects() - seen = {} - for checkout in self.checkouts: - if not checkout.git.is_git_checkout: - continue - - canonical_name = checkout.git.config.repo_name - seen[canonical_name] = True - checkout.git.remote_info = self.stash_projects.get(canonical_name) - - for name, project in self.stash_projects.items(): - if name in seen: - continue - - path = os.path.join(self.path, name) - if os.path.isdir(path): - path += ".1" - - r = GitCheckout(path, parent=self) - r.git.remote_info = project - self.checkouts.append(r) - - self.checkouts = sorted(self.checkouts, key=lambda x: x.basename) - if self.prefs.align and self.projects: - self.prefs.name_size = min(36, max(len(c.name) for c in self.checkouts)) - - else: - self.prefs.name_size = None - - @runez.cached_property - def header(self): - result = "%s:" % runez.purple(runez.short(self.path)) - - if not self.projects: - return "{} {}".format(result, runez.orange("no git folders")) - - if self.predominant: - result += runez.bold(f" {len(self.projects[self.predominant])} {self.predominant}") - - else: - result += runez.orange(" no predominant project") - - if self.additional: - result += " (%s)" % runez.purple(", ".join(f"+{len(self.projects[project])} {project}" for project in self.additional)) - - return result - - def print_status(self): - """Show checkout status""" - print(self.header) - for checkout in self.checkouts: - if self.prefs.all or checkout.git.is_git_checkout: - checkout.print_status() + self.git_dirs = [ + GitDir(source_path) + for source_path in self.path.iterdir() + if not source_path.name.startswith(".") and source_path.is_dir() and (source_path / ".git").is_dir() + ] + self.git_dirs = sorted(self.git_dirs, key=lambda x: x.basename) + Reporter.abort_if(not self.git_dirs, f"{runez.short(self.path)}: no git folders") + self.name_size = min(36, max(len(git.basename) for git in self.git_dirs)) if len(self.git_dirs) > 1 else None + + def prefixed_line(self, git: GitDir, line: str) -> str: + name = git.basename + if self.name_size: + name = f"{name:>{self.name_size}}" + + return f"{name}: {line}" diff --git a/src/mgit/cli.py b/src/mgit/cli.py index 0902de8..01ca1f5 100644 --- a/src/mgit/cli.py +++ b/src/mgit/cli.py @@ -1,228 +1,371 @@ -""" -\b -Advanced usage: - --clean show Show which local/remote branches can be cleaned - --clean local Clean local branches that were deleted from their corresponding remote - --clean remote Clean merged remote branches - --clean all Clean local and merged remote branches - --clean reset Do a git --reset --hard + clean -fdx (nuke all changes, get back to pristine state) -\b -""" +from __future__ import annotations +import argparse import logging +import re import sys +from abc import ABC, abstractmethod +from importlib.metadata import version +from pathlib import Path -import click import runez -from mgit import get_target, GitCheckout -from mgit.git import GitRunReport - -LOG = logging.getLogger(__name__) -VALID_CLEAN_ACTIONS = ("show", "local", "remote", "all", "reset") - - -@runez.click.command() -@runez.click.version() -@runez.click.debug() -@runez.click.color() -@runez.click.log() -@click.option("--clean", default=None, type=click.Choice(VALID_CLEAN_ACTIONS), help="Auto-clean branches") -@click.option("-f", "--fetch", is_flag=True, default=False, help="Fetch from all remotes") -@click.option("-p", "--pull", is_flag=True, default=False, help="Pull from tracking remote") -@click.option("-s/-v", "--short/--verbose", is_flag=True, default=None, help="Short/verbose output") -@click.option("-cs", is_flag=True, default=False, help="Handy shortcut for '--clean show'") -@click.option("-cl", is_flag=True, default=False, help="Handy shortcut for '--clean local'") -@click.option("-cr", is_flag=True, default=False, help="Handy shortcut for '--clean remote'") -@click.option("-ca", is_flag=True, default=False, help="Handy shortcut for '--clean all'") -@click.argument("target", required=False, default=None) -def main(debug, log, clean, target, **kwargs): - """ - Fetch collections of git projects - """ - runez.system.AbortException = SystemExit - runez.date.DEFAULT_DURATION_SPAN = -2 - runez.log.setup(debug=debug, console_format="%(levelname)s %(message)s", file_location=log, locations=None) - - hc = handy_clean(kwargs) - if not clean: - clean = hc - - target = get_target(target, **kwargs) - - if clean is not None: - handle_clean(target, clean) - sys.exit(0) - - target.print_status() - - -def handy_clean(kwargs): - """ - :param kwargs: Pop all handy shortcuts from kwargs - :return str|None: Equivalent full --clean option - """ - cs = kwargs.pop("cs") - cl = kwargs.pop("cl") - cr = kwargs.pop("cr") - ca = kwargs.pop("ca") - if cs: - return "show" - - if cl: - return "local" - - if cr: - return "remote" - - if ca: - return "all" - - return None - - -def run_git(target, fatal, *args): - """Run git command on target, abort if command exits with error code""" - error = target.git.run_raw_git_command(*args) - if error.has_problems: - if fatal: - runez.abort(error.representation()) - - print(error.representation()) - return 0 - - return 1 - - -def clean_reset(target): - """ - :param GitCheckout target: Target to reset - """ - fallback = target.git.fallback_branch() - if not fallback: - runez.abort("Can't determine a branch that can be used for reset") - - run_git(target, True, "reset", "--hard", "HEAD") - run_git(target, True, "clean", "-fdx") - if fallback != target.git.branches.current: - run_git(target, True, "checkout", fallback) - - run_git(target, True, "pull") - target.git.reset_cached_properties() - print(target.header()) - - -def clean_show(target): - """ - :param GitCheckout target: Target to show - """ - print(target.header()) - if not target.git.local_cleanable_branches: - print(" No local branches can be cleaned") +from mgit import ProjectDir, Reporter +from mgit.git import GitDir - else: - for branch in target.git.local_cleanable_branches: - print(" {} branch {} can be cleaned".format(runez.bold("local"), runez.bold(branch))) - if not target.git.remote_cleanable_branches: - print(" No remote branches can be cleaned") +def command_name_from_class(cls: type) -> str: + name = cls.__name__.removesuffix("Command") + return re.sub(r"(? str: + return command_name_from_class(cls) - if what == "reset": - return clean_reset(target) + @classmethod + def summary(cls) -> str: + """Every command class must have a summary (intentional crash otherwise)""" + assert cls.__doc__ # noqa: S101, internal error + return cls.__doc__.splitlines()[0] - if what == "show": - return clean_show(target) + @classmethod # noqa: B027 - optional hook + def add_arguments(cls, _parser: argparse.ArgumentParser): + """Add command-specific arguments.""" - total_cleaned = 0 - print(target.header()) + @classmethod + @abstractmethod + def from_namespace(cls, _namespace: argparse.Namespace) -> CliCommand: + """Create a command from parsed CLI arguments.""" - if what in "remote all": - if not target.git.remote_cleanable_branches: - print(" No remote branches can be cleaned") + @abstractmethod + def run(self): + """Run the command.""" - else: - total = len(target.git.remote_cleanable_branches) - cleaned = 0 - for branch in target.git.remote_cleanable_branches: - remote, _, name = branch.partition("/") - if not remote and name: - raise Exception("Unknown branch spec '%s'" % branch) - if run_git(target, False, "branch", "--delete", "--remotes", branch): - cleaned += run_git(target, False, "push", "--delete", remote, name) +class FolderTargetCommand(CliCommand): + folder: Path | None = None - total_cleaned += cleaned - if cleaned == total: - print("%s cleaned" % runez.plural(cleaned, "remote branch")) + def __init__(self, folder: Path | None): + self.folder = folder - else: - print(f"{cleaned}/{total} remote branches cleaned") + @classmethod + def add_arguments(cls, parser: argparse.ArgumentParser): + parser.add_argument("folder", nargs="?", type=Path) - target.git.reset_cached_properties() - if what == "all": - # Fetch to update remote branches (and correctly detect new dangling local) - target.git.fetch() + @classmethod + def from_namespace(cls, namespace: argparse.Namespace) -> FolderTargetCommand: + return cls(folder=namespace.folder) - if what in "local all": - if not target.git.local_cleanable_branches: - print(" No local branches can be cleaned") + @staticmethod + def current_checkout() -> Path | None: + current = Path.cwd() + for candidate in (current, *current.parents): + if (candidate / ".git").is_dir(): + return candidate - else: - total = len(target.git.local_cleanable_branches) - cleaned = 0 - for branch in target.git.local_cleanable_branches: - if branch == target.git.branches.current: - fallback = target.git.fallback_branch() - if not fallback: - print("Skipping branch '%s', can't determine fallback branch" % target.git.branches.current) - continue + def target(self) -> GitDir | ProjectDir: + if self.folder is None: + current = self.current_checkout() + if current: + return GitDir(current) + + folder = (self.folder or Path(".")).expanduser().absolute() + Reporter.abort_if(not folder.is_dir(), f"No folder '{runez.short(folder)}'") + if (folder / ".git").is_dir(): + return GitDir(folder) + + return ProjectDir(folder) + + def run(self): + """Run the command.""" + target = self.target() + if isinstance(target, GitDir): + return self.run_single(target) + + return self.run_multi(target) + + @abstractmethod + def run_single(self, git: GitDir): + """Run this command for one git dir.""" + + def run_multi(self, _project_dir: ProjectDir) -> None: + """Run this command for all git dirs in `project_dir`.""" + Reporter.abort(f"{self.command_name()} only supports one git checkout") + + +COMMANDS: list[type[CliCommand]] = [] +COMMAND_BY_TOKEN: dict[str, type[CliCommand]] = {} + + +def register_cli_command(name: str | None, command: type[CliCommand]): + assert name # noqa: S101, this would be an internal error, detected at test time + assert name not in COMMAND_BY_TOKEN # noqa: S101 + COMMAND_BY_TOKEN[name] = command + + +def cli_command(command: type[CliCommand]) -> type[CliCommand]: + COMMANDS.append(command) + register_cli_command(command.command_name(), command) + register_cli_command(command.short_name, command) + return command - run_git(target, True, "checkout", fallback) - run_git(target, True, "pull") - cleaned += run_git(target, False, "branch", "--delete", branch) +@cli_command +class StatusCommand(FolderTargetCommand): + """Show repo or workspace status.""" - total_cleaned += cleaned - if cleaned == total: - print(runez.bold("%s cleaned" % runez.plural(cleaned, "local branch"))) + short_name = "s" - else: - print(runez.orange(f"{cleaned}/{total} local branches cleaned")) + def run_single(self, git: GitDir): + print(git.status_line()) + details = git.status_details() + if details: + print(details) - target.git.reset_cached_properties() + def run_multi(self, project_dir: ProjectDir): + for git in project_dir.git_dirs: + print(project_dir.prefixed_line(git, git.status_line())) - if total_cleaned: - print(target.header()) +@cli_command +class FetchCommand(FolderTargetCommand): + """Fetch remotes, then show status.""" -def handle_clean(target, what): - if isinstance(target, GitCheckout): - handle_single_clean(target, what) - return + short_name = "f" - if what in "remote reset": - runez.abort("Only '--clean show' and '--clean local' supported for multiple git checkouts for now") + def run_single(self, git: GitDir): + report = git.fetch_now().require_success("fetch") + print(git.status_line(report)) + details = git.status_details() + if details: + print(details) - target.prefs.name_size = None - target.prefs.set_short(True) - for sub_target in target.checkouts: - handle_single_clean(sub_target, what) - print("----") + def run_multi(self, project_dir: ProjectDir): + for git in project_dir.git_dirs: + report = git.fetch_now() + print(project_dir.prefixed_line(git, git.status_line(report))) + + +@cli_command +class PullCommand(FolderTargetCommand): + """Pull with rebase when the worktree is safe.""" + + short_name = "p" + + def run_single(self, git: GitDir): + report = git.pull().require_success("pull") + print(git.status_line(report)) + details = git.status_details() + if details: + print(details) + + def run_multi(self, project_dir: ProjectDir): + for git in project_dir.git_dirs: + report = git.pull() + print(project_dir.prefixed_line(git, git.status_line(report))) + + +@cli_command +class MainCommand(FolderTargetCommand): + """Checkout the default branch.""" + + short_name = "m" + + def run_single(self, git: GitDir): + report = git.checkout_default_branch() + print(git.status_line(report)) + details = git.status_details() + if details: + print(details) + + +@cli_command +class BranchesCommand(FolderTargetCommand): + """Show local branches.""" + + short_name = "b" + + def run_single(self, git: GitDir): + details = git.branch_details() + if details: + print(details) + + def run_multi(self, project_dir: ProjectDir): + show_names = len(project_dir.git_dirs) > 1 + for git in project_dir.git_dirs: + if show_names: + print(f"{git.basename}:") + + indent = " " if show_names else "" + details = git.branch_details(indent=indent) + if details: + print(details) + + +@cli_command +class GroomCommand(FolderTargetCommand): + """Fetch, return to default branch, pull, and clean the groomed branch.""" + + 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 + current_branch = refs.current + default_branch = git.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 + 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) + 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") + + self._checkout_default_branch(git, default_branch) + report = git.pull().require_success("groom") + git.clear_cached_state() + if remote_cleanup: + self._delete_remote_branch(git, remote_cleanup) + + self._delete_local_branch(git, local_cleanup) + print(git.status_line(report)) + + +class CommandHelpFormatter(argparse.HelpFormatter): + """Show subcommands as a compact command list.""" + + def _format_action(self, action: argparse.Action) -> str: + if isinstance(action, argparse._SubParsersAction): + return "".join(self._format_action(choice_action) for choice_action in action._get_subactions()) + + return super()._format_action(action) + + +def add_command_parser(subparsers: argparse._SubParsersAction, command: type[CliCommand]): + aliases = [command.short_name] if command.short_name else [] + parser = subparsers.add_parser( + command.command_name(), + aliases=aliases, + description=command.summary(), + formatter_class=CommandHelpFormatter, + help=command.summary(), + prog=f"mgit {command.command_name()}", + ) + parser._action_groups[0].title = "Arguments" + command.add_arguments(parser) + parser.set_defaults(command_type=command) + + +def split_global_args(args: list[str]) -> tuple[list[str], list[str]]: + global_args = [] + i = 0 + while i < len(args): + arg = args[i] + if arg in {"--help", "-v", "--verbose", "--version"}: + global_args.append(arg) + i += 1 + continue + + if arg == "--color": + global_args.append(arg) + i += 1 + if i < len(args): + global_args.append(args[i]) + i += 1 + continue + + if arg.startswith("--color="): + global_args.append(arg) + i += 1 + continue + + if arg.startswith("-"): + global_args.append(arg) + i += 1 + + break + + return global_args, args[i:] + + +def normalized_cli_args(args: list[str]) -> list[str]: + global_args, command_args = split_global_args(args) + if global_args and global_args[-1] == "--color": + return global_args + + if command_args: + command = COMMAND_BY_TOKEN.get(command_args[0]) + if command: + command_args[0] = command.command_name() + + else: + command_args.insert(0, StatusCommand.command_name()) + + else: + command_args.append(StatusCommand.command_name()) + + return global_args + command_args + + +def main(): + parser = argparse.ArgumentParser( + prog="mgit", + usage="mgit [GLOBAL_OPTIONS] [COMMAND] [ARGS...]", + description="Inspect and update git checkouts.", + formatter_class=CommandHelpFormatter, + ) + parser.add_argument("-v", "--verbose", action="store_true", help="Enable verbose logging.") + parser.add_argument("--color", choices=("auto", "always", "never"), default="auto", help="Control ANSI color output.") + parser.add_argument("--version", action="version", version=f"mgit {version('mgit')}") + subparsers = parser.add_subparsers(dest="command", title="Commands", metavar="COMMAND") + subparsers.required = True + for command in COMMANDS: + add_command_parser(subparsers, command) + + namespace = parser.parse_args(normalized_cli_args(sys.argv[1:])) + command = namespace.command_type.from_namespace(namespace) + with runez.ActivateColors(None if namespace.color == "auto" else namespace.color == "always"): + runez.date.DEFAULT_DURATION_SPAN = -2 + runez.log.setup(debug=namespace.verbose, level=logging.INFO, console_format="%(levelname)s %(message)s", locations=None) + command.run() diff --git a/src/mgit/git.py b/src/mgit/git.py index d4fcb92..12c6af0 100644 --- a/src/mgit/git.py +++ b/src/mgit/git.py @@ -1,86 +1,115 @@ -import collections +from __future__ import annotations + import logging -import os -import re import subprocess +import sys import time -from urllib.parse import urlparse +from dataclasses import dataclass +from functools import cached_property +from typing import NoReturn, TYPE_CHECKING import runez -LOG = logging.getLogger(__name__) -FETCH_AGE_FILES = ("FETCH_HEAD", "HEAD") -FRESHNESS_THRESHOLD = 12 * runez.date.SECONDS_IN_ONE_HOUR -BRANCH_INVALID_CHARS = "~^: \t\\" -GIT_ERROR_PREFIXES = {"git", "error", "fatal"} +if TYPE_CHECKING: + from pathlib import Path -RE_GITHUB_SSH = re.compile(r"^git@([^:]+):(\w+)/([^/]+)$") -RE_BRANCH_STATUS = re.compile(r"^## (.+)\.\.\.(([^/]+)/)?([^ ]+)\s*(\[(.+)])?$") +class Reporter: + """Central user-facing reporting helpers.""" -def is_valid_branch_name(name): - """ - :param str|None name: Branch name to validate - :return bool: True if branch name appears valid, as per https://wincent.com/wiki/Legal_Git_branch_names - """ - if not name or name[0] == "." or ".." in name or name.endswith(("/", ".lock")): - return False + log = logging.getLogger("mgit") - return not any(ord(char) < 32 or char in BRANCH_INVALID_CHARS for char in name) + branch_default = runez.green + branch_orphaned = runez.orange + problem = runez.red + note = runez.purple + progress = runez.plain + index_change = runez.teal + worktree_change = runez.red + untracked_change = runez.orange + @staticmethod + def joined(*args, separator=" "): + """Join non-false-ish display fragments.""" + return runez.joined(*args, delimiter=separator, keep_empty=None) -def shortened_message(text, keep_lines=2, separator=" "): - """ - :param str text: Original git error message (those can be verbose, and include progress) - :param int keep_lines: Max lines to keep - :param str separator: Lines are split for shortening, separator to use to re-join lines - :return str: Shortened git error message - """ - lines = [] - prefixed = [] - for line in text.strip().split("\n"): - line = line.strip().strip(".") - if not line: - continue + @staticmethod + def joined_lines(*args, header="", indent="", separator="\n"): + """Join non-false-ish display lines with optional header and indentation.""" + lines = [f"{indent}{line}" for line in runez.flattened(args, keep_empty=None)] + return Reporter.joined(header, lines, separator=separator) - p = line.partition(":") - if p[2] and p[0] in GIT_ERROR_PREFIXES: - prefixed.append(p[2].strip()) + @staticmethod + def abort(message, exit_code: int = 1) -> NoReturn: + print(Reporter.problem(message), file=sys.stderr) + sys.exit(exit_code) + + @staticmethod + def abort_if(condition: object, message, exit_code: int = 1): + if condition: + Reporter.abort(message, exit_code=exit_code) + + @staticmethod + 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/" - else: - lines.append(line) - if prefixed: - lines = prefixed +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 keep_lines and len(lines) > keep_lines: - lines = lines[:keep_lines] + if "repository not found" in detail: + return "repository not found" - return separator.join(lines).replace(" ", " ").strip() + 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()) + + else: + lines.append(line) + + return " ".join((prefixed or lines)[:2]).replace(" ", " ").strip() class GitRunReport: """Convenient and easy to compose reporting class""" - def __init__(self, *args, **kwargs): + def __init__( + self, + other: GitRunReport | None = None, + *, + progress: str | None = None, + note: str | None = None, + problem: str | None = None, + ): self._progress = [] self._note = [] self._problem = [] - self.add(*args, **kwargs) + self.add(other, progress=progress, note=note, problem=problem) - def __repr__(self): - return f"{len(self._problem)} problems, {len(self._progress)} progress, {len(self._note)} notes" + def __str__(self): + return self.representation() - def __contains__(self, text): - """ - :param str text: Text to look up - :return bool: True if 'text' is mentioned in one of the messages in self._problem - """ - return bool(text) and any(text in problem for problem in self._problem) + def require_success(self, operation: str) -> GitRunReport: + if self.has_problems: + Reporter.abort(self.add(problem=f" max_chars: - result = "%s..." % (result[: max_chars - 3]) + result = f"{result[: max_chars - 3]}..." return result - def _add(self, target, items): - """ - :param list target: Where to add 'items' - :param items: items to add - """ - if not items: - return + def add( + self, + other: GitRunReport | None = None, + *, + progress: str | None = None, + note: str | None = None, + problem: str | None = None, + ) -> GitRunReport: + if other is not None: + _add_messages(self._progress, other._progress) + _add_messages(self._note, other._note) + _add_messages(self._problem, other._problem) + + _add_messages(self._progress, progress) + _add_messages(self._note, note) + _add_messages(self._problem, problem) + return self - if isinstance(items, (list, tuple)): - for item in items: - self._add(target, item) - elif items not in target: - target.append(items) +def git_error_report(proc: subprocess.CompletedProcess[str]) -> GitRunReport: + """GitRunReport describing a failed git command.""" + return GitRunReport(problem=git_error_message(proc)) - def cumulate(self, other): - """ - :param GitRunReport other: Cumulate 'other' with current report - :return GitRunReport: Returns self - """ - if isinstance(other, GitRunReport): - self._add(self._progress, other._progress) - self._add(self._note, other._note) - self._add(self._problem, other._problem) - return self +@dataclass +class BranchUpstream: + """Configured upstream for a local branch.""" - def add(self, *args, **kwargs): - """ - :param args: Optional, other reports to cumulate - :param kwargs: Optional, attributes to add - :return GitRunReport: Returns self - """ - for item in args: - self.cumulate(item) + remote: str + branch: str - for key, value in kwargs.items(): - attribute_name = "_%s" % key - target = getattr(self, attribute_name, None) - if target is None: - raise Exception("Internal error: invalid GitRunReport target '%s'" % key) - if isinstance(value, (list, tuple)): - for item in value: - self.add(**{key: item}) +@dataclass(frozen=True) +class CleanableLocalBranch: + """Local branch that is safe to delete.""" - elif isinstance(value, GitRunReport): - self._add(target, getattr(value, attribute_name)) + name: str + force_delete: bool = False - else: - self._add(target, value) - return self +@dataclass(frozen=True) +class CleanableRemoteBranch: + """Tracked remote branch that is safe to delete with an exact-ref lease.""" + remote: str + branch: str + expected_oid: str -class GitURL: - """Parse and extract meaningful info from a git repo url""" - def __init__(self): - self.url = None - self.protocol = None - self.hostname = None - self.relative_path = None - self.username = None - self.name = None - self.repo = None +class GitRefs: + """Repository ref and upstream snapshot.""" - def __repr__(self): - return self.url or "" + 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] - def _set_name(self, basename): - """ - :param str|None basename: Set 'self.name' from 'basename' of url - """ - if basename and basename.endswith(".git"): - basename = basename[:-4] + 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.default_branches = {} + self.upstreams = {} + lines = parent.checked_git_command_lines( + "for-each-ref", + "--format=%(refname)%09%(upstream:remotename)%09%(upstream:remoteref)%09%(symref)", + "refs/heads", + "refs/remotes", + ) + for line in lines: + self._add_line(line) - self.name = basename or "unknown" + def _add_line(self, line: str): + fields = (line + "\t" * 3).split("\t") + refname, upstream_remote, upstream_ref, symref = fields[:4] + if refname.startswith(LOCAL_REF_PREFIX): + local_branch_name = refname[len(LOCAL_REF_PREFIX) :] + self.local.add(local_branch_name) - def _set_repo(self, dirname): - """ - :param str|None dirname: Set 'self.repo' from 'dirname' of url - """ - if dirname and "/" in dirname: - dirname = os.path.basename(dirname) + 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 - self.repo = dirname or "unknown" + elif refname.startswith(REMOTE_REF_PREFIX): + remote, _, branch = refname[len(REMOTE_REF_PREFIX) :].partition("/") + if remote and branch: + if branch == "HEAD": + prefix = f"{REMOTE_REF_PREFIX}{remote}/" + if len(symref) > len(prefix) and symref.startswith(prefix): + self.default_branches[remote] = symref[len(prefix) :] - def set(self, url): - """ - :param str url: Set fields of this object, extracted from git repo 'url' - """ - self.url = url or "" - if not url: - self.protocol = "unknown" - self.hostname = "unknown" - self.relative_path = "" - self.username = None - self._set_name(None) - self._set_repo(None) - return - - if url.startswith("git@"): - m = RE_GITHUB_SSH.match(url) - if m: - self.protocol = "ssh" - self.hostname = m.group(1) or "unknown" - self.relative_path = f"{m.group(2)}/{m.group(3)}" - self.username = "git" - self._set_name(m.group(3)) - self._set_repo(m.group(2)) - return - url = "ssh://%s" % url - - p = urlparse(url) - self.protocol = p.scheme or "file" - self.hostname = p.hostname or "local" - self.relative_path = p.path.rstrip("/") - self.username = p.username - self._set_name(os.path.basename(self.relative_path)) - self._set_repo(os.path.dirname(self.relative_path)) + else: + self.by_remote.setdefault(remote, set()).add(branch) + + 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)) + + 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)) class GitDir: """Model a local git repo""" - def __init__(self, path): + def __init__(self, path: Path): """ - :param str path: Path to local repo + :param Path path: Path to local repo """ self.path = path - self.folder_exists = os.path.exists(path) - self.is_git_checkout = self.folder_exists and os.path.isdir(os.path.join(path, ".git")) - self.remote_info = None - - def __repr__(self): - if not self.is_git_checkout: - return "! %s" % self.path + self.basename = path.name + self.age = self._current_age() - return self.path - - def report(self, bare=False, inspect_remotes=False): - """ - :param bool bare: Bare report only - :param bool inspect_remotes: If True, report on which remote branches are cleanable - :return GitRunReport: General report on current checkout state - """ - if not self.is_git_checkout: - if self.remote_info: - return GitRunReport(problem="not cloned yet") + 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]"), + ) - return GitRunReport.not_git() + @staticmethod + def _detail_lines(items, state_style, worktree_style=None) -> list[str]: + lines = [] + for item in items: + state = item[0:2] + if worktree_style: + state = f"{state_style(item[0])}{worktree_style(item[1])}" - result = GitRunReport() + elif state_style: + state = state_style(state) - if not self.config.remotes: - result.add(problem="no remotes") + lines.append(f"{state} {item[3:]}") - if bare: - return result + return lines + def status_line(self, report: GitRunReport | None = None) -> str: + refs = self.refs + status = self.status + edits, deletes, new = status.pending_change_counts() + report = GitRunReport(report) if self.age is not None and self.age > FRESHNESS_THRESHOLD: - result.add(note="last fetch %s ago" % runez.represented_duration(self.age)) - - orphan_branches = self.orphan_branches - if self.branches.current in orphan_branches: - # Current is no more on its remote (should possibly checkout another branch and cleanup, or push) - orphan_branches = orphan_branches[:] - orphan_branches.remove(self.branches.current) - result.add(note="current branch '%s' is orphaned" % self.branches.current) - - if len(orphan_branches) == 1: - result.add(note="local branch '%s' can be pruned" % orphan_branches[0]) - - elif orphan_branches: - result.add(note="%s can be pruned" % runez.plural(orphan_branches, "local branch")) - - result.add(self.branches.report) + 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) + remaining = len(other_branches) - cleanable + branch_summary = Reporter.joined(cleanable and f"+{cleanable}đŸĒĻ", remaining and f"+{remaining}", separator="") + if branch_summary: + branch_summary = f"[{branch_summary}]" + + return Reporter.joined( + self.represented_current_branch(), + branch_summary, + status.upstream_delta(), + edits and f"✏ī¸{edits}", + deletes and f"🗑ī¸{deletes}", + new and f"🆕{new}", + report, + ) + + def represented_current_branch(self) -> str: + refs = self.refs + branch = refs.current + if refs.detached: + icon = " đŸ‘ģ" + + elif self.is_orphan_branch(branch): + icon = " đŸĒĻ" - if inspect_remotes and self.remote_cleanable_branches: - if len(self.remote_cleanable_branches) == 1: - cleanable = "'%s'" % next(iter(self.remote_cleanable_branches)) - - else: - cleanable = runez.plural(self.remote_cleanable_branches, "remote branch") - - result.add(note="%s can be cleaned" % cleanable) + else: + icon = " ✅" if self.age is not None and self.age <= FRESH_FETCH_THRESHOLD else " ☑ī¸" - return result + return self.represented_branch(branch) + icon - def _git_command(self, args): - """ - :param list|tuple args: Git command + args to use - :return list, str: Full git invocation + human friendly representation - """ - cmd = ["git"] - if args and args[0] == "clone": - args_represented = "git %s" % " ".join(args) + 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)}") + + status = self.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}" - else: - args_represented = "git -C {} {}".format(runez.short(self.path), " ".join(args)) - cmd.extend(["-C", self.path]) + result.append(line) - cmd.extend(args) - return cmd, args_represented + return Reporter.joined_lines(result, indent=indent) - def run_raw_git_command(self, *args): + def run_git_command(self, *args: str) -> subprocess.CompletedProcess[str]: """ - :param args: Execute git command with provided args, don't capture its output, but let it show through stdout/stderr - :return GitRunReport: Report + :param args: Execute git command with provided args + :return subprocess.CompletedProcess: Result from git """ - cmd, pretty_args = self._git_command(args) - pretty_args = "git %s" % " ".join(args) - print("Running: %s" % runez.bold(pretty_args)) - proc = subprocess.Popen(cmd) # noqa: S603 - proc.communicate() - if proc.returncode: - return GitRunReport(problem="git exited with code %s" % proc.returncode) + 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 - return GitRunReport() - - def run_git_command(self, *args): + def checked_git_command(self, *args: str) -> str: """ :param args: Execute git command with provided args - :return str, GitRunReport: Output from git command + report on eventual error + :return str: Output from git command, aborting if git fails """ - cmd, pretty_args = self._git_command(args) - LOG.debug("Running: %s", pretty_args) - proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) # noqa: S603 - output, error = proc.communicate() - if proc.returncode == 0: - return output, GitRunReport() + proc = self.run_git_command(*args) + Reporter.abort_if(proc.returncode, f"git {' '.join(args)} failed: {git_error_message(proc)}") + return proc.stdout.strip() - if not error: - return output, GitRunReport(problem="git exited with code %s" % proc.returncode) + 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()] - return output, GitRunReport(problem=shortened_message(error)) + 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 fallback_branch(self): - """ - :return str: Best branch to fallback to if we need to clean or reset - """ - if self.branches.current not in self.orphan_branches: - return self.branches.current + 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 - for branch in self.branches.local: - if branch not in self.orphan_branches: - return branch + 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 - return self.branches.default_branches.get("origin") + Reporter.abort(f"Could not inspect git HEAD: {git_error_message(proc)}") - def reset_cached_properties(self): - """Reset cached properties that may have changed after a fetch or pull""" - runez.cached_property.reset(self) + def fetch_now(self) -> GitRunReport: + proc = self.run_git_command("fetch", "--all", "--prune") + if proc.returncode == 0: + self.age = 0 - def fetch(self, age=30): + return git_error_report(proc) + + def checkout_default_branch(self) -> GitRunReport: """ - :param int|None age: Fetch if age is older than specified number of seconds, use None to fetch unconditionally - :return GitRunReport: + :return GitRunReport: Checkout report """ - if not self.is_git_checkout: - return GitRunReport.not_git() - - if age is not None: - current_age = self.age - if current_age is not None and current_age <= age: - return GitRunReport() + branch = self.default_branch + if self.refs.current == branch: + return GitRunReport() - _, error = self.run_git_command("fetch", "--all", "--prune") - self.reset_cached_properties() - return error + self.checked_git_command("checkout", branch) + self.clear_cached_state() + return GitRunReport(progress=f"checked out {self.represented_branch(branch)}") - def pull(self): + def pull(self) -> GitRunReport: """Pull from tracked remote""" - if not self.is_git_checkout: - return GitRunReport.not_git().cant_pull() - - report = self.report(bare=True) - if report.has_problems: - return report.cant_pull() - - if self.status.modified: - return GitRunReport().cant_pull("pending changes") - - if self.status.report.has_problems: - return GitRunReport(problem=self.status.report).cant_pull() - - if not self.branches.current: - return GitRunReport(problem="no remote branch") - - if self.branches.current == "HEAD" and self.branches.current in self.orphan_branches: - # Untracked HEAD - branch = self.fallback_branch() - if not branch: - return GitRunReport(problem="can't determine fallback branch") - - output, error = self.run_git_command("checkout", branch) - if error.has_problems: - self.reset_cached_properties() - return error - - output, error = self.run_git_command("pull", "--rebase") - self.reset_cached_properties() - - if error.has_problems: - if "following untracked" in error: - return GitRunReport().cant_pull("untracked files would be overwritten") - - if "Repository not found" in error: - return GitRunReport().cant_pull("repository not found") - - return error.cant_pull() - - if "up to date" in output or "up-to-date" in output: - return GitRunReport(progress="") - - if "Fast-forward" in output: - return GitRunReport(progress="pulled successfully") - - # Shouldn't be reached - LOG.debug("Check pull --rebase output: %s, error: %s", output, error) - lines = [] - if output: - lines.extend(s.strip() for s in output.strip().split("\n") if s.strip()) - - lines.append(error.representation(progress=False, note=False).strip()) - output = lines[0] if lines else "no output" - return GitRunReport(note="pull may have been unsuccessful (%s)" % output) - - def clone(self, url): - if self.folder_exists: - return GitRunReport(problem="folder already exists, can't clone") - - _, error = self.run_git_command("clone", url, self.path) - self.folder_exists = os.path.exists(self.path) - self.is_git_checkout = self.folder_exists and os.path.isdir(os.path.join(self.path, ".git")) - self.reset_cached_properties() + refs = self.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) + + note = status.upstream_delta() or "up-to-date" + proc = self.run_git_command("pull", "--rebase") + self.clear_cached_state() + report = GitRunReport(note=f"was {note}") + if proc.returncode: + report.cant_pull(git_error_message(proc)) - if error.has_problems: - return error.add(problem=" int | None: + """Elapsed time in seconds since last fetch""" + for name in ("FETCH_HEAD", "HEAD"): try: - last_fetch = os.path.getmtime(os.path.join(os.path.join(self.path, ".git"), name)) + last_fetch = (self.path / ".git" / name).stat().st_mtime return int(time.time() - last_fetch) except OSError: pass - @runez.cached_property - def status(self): - """ - :return GitStatus: Parsed info from 'git status --porcelain --branch' - """ + @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: + """Parsed info from 'git status --porcelain=v2 --branch'""" return GitStatus(self) - @runez.cached_property - def config(self): - """ - :return GitConfig: Parsed info from 'git config --list' - """ - return GitConfig(self) + @cached_property + def refs(self) -> GitRefs: + return GitRefs(self) - @runez.cached_property - def branches(self): - """ - :return GitConfig: Parsed info from 'git branch --list --all' - """ - return GitBranches(self) - - @runez.cached_property - def orphan_branches(self): - """ - :return list(str): Local branch names that were deleted on their corresponding remote - """ + @cached_property + def orphan_branches(self) -> list[str]: + """Local branch names that were deleted on their corresponding remote""" result = [] - for name in self.branches.local: - remote = self.config.tracking_remote.get(name) - if not remote or remote not in self.branches.by_remote or name not in self.branches.by_remote[remote]: + 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 - @runez.cached_property - def special_branches(self): - result = set(self.branches.default_branches.values()) - result.add("HEAD") - result.add("main") - result.add("master") - 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())) - @runez.cached_property - def local_cleanable_branches(self): - """ - :return set: Local branches that can be cleaned - """ - result = {name for name in self.orphan_branches if name not in self.special_branches} - for branch in self.remote_cleanable_branches: - remote, _, name = branch.partition("/") - tracking = self.config.tracking_remote.get(name) - if tracking == remote: - result.add(name) + 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 result + 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) - @runez.cached_property - def remote_cleanable_branches(self): - """ - :return set: Remote branches that can be cleaned - """ - result = set() - aspect = GitBranches(self, auto_load=False) - aspect._command = "branch --list --remote --merged" - aspect._remote_prefix = "" - default = self.branches.default_branches.get("origin") - if default: - aspect._command += " %s" % default - - aspect.reload() - for remote, branches in aspect.by_remote.items(): - url = self.config.remotes.get(remote) - if not url or url.protocol != "ssh": - continue - - result.update([f"{remote}/{branch}" for branch in branches if branch not in self.special_branches]) + if self.is_orphan_branch(name): + return runez.orange(result) 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}" -class GitAspect: - """Common ancestor for info gathered from git""" - - _command = None - - def __init__(self, parent, auto_load=True): - self._parent = parent - self._lines = None # Lines from output of last command run, for troubleshooting - if auto_load: - self.reload() - - def __repr__(self): - return self._command - - def reload(self): - for k in self.__class__.__dict__: - if k.startswith("_"): - continue - - v = getattr(self.__class__, k, None) - if v is None or isinstance(v, (property, runez.cached_property)) or callable(v): - continue - - v = collections.defaultdict(v.default_factory) if isinstance(v, collections.defaultdict) else v.__class__() - setattr(self, k, v) - - if not self._parent.is_git_checkout: - return - - output, error = self._parent.run_git_command(*self._command.split()) - if error.has_problems: - LOG.debug("Prev git command had error output: [%s]", error.representation()) - - self._lines = [line for line in output.split("\n") if line.strip()] - for line in self._lines: - self._process_line(line) - - def _process_line(self, line): - """Process `line`""" - - -class GitBranches(GitAspect): - """Branch info""" + return base_ref - _command = "branch --list --all" - _remote_prefix = "remotes/" - - def __init__(self, parent, auto_load=True): - self.current = "" # Current local branch - self.local = set() # Local branches - self.by_remote = collections.defaultdict(set) # Branches by remote (usually origin and optionally upstream) - self.default_branches = {} # Default branch per remote - self.report = GitRunReport() - super().__init__(parent, auto_load=auto_load) - - @property - def shortened_current_branch(self): - return str(self.current or "HEAD").replace("feature/", "f/").replace("bugfix/", "b/") - - def _process_line(self, line): - if not line or len(line) <= 3 or line[0] not in " *" or line[1] != " ": - LOG.warning("Internal error: malformed branch --list line: %s", line) - return - - name = line[2:] - if name.startswith(self._remote_prefix): - name = name[len(self._remote_prefix) :] - default = None - try: - i = name.index(" -> ") - first = name[:i] - if first.endswith("/HEAD"): - default = name = name[i + 4 :] - - except ValueError: - pass - - remote, _, name = name.partition("/") - self.by_remote[remote].add(name) - if default: - self.default_branches[remote] = name - - return - - if name.startswith("("): - name = name[1:] - if name.endswith(")"): - name = name[:-1] - - name, _, problem = name.partition(" ") - self.report.add(note=f"{name} {problem}") - - self.local.add(name) - if line[0] == "*": - self.current = name - - -class GitConfig(GitAspect): - """Remote info""" + 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 - _command = "config --list" + def merge_is_noop(self, ref: str, target_ref: str) -> 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}}") + proc = self.run_git_command("merge-tree", "--write-tree", "--no-messages", target_ref, ref) + if proc.returncode: + return False - def __init__(self, parent, auto_load=True): - self.origin = GitURL() # URL to remote called 'origin' - self.remotes = {} # GitURL by remote name map - self.tracking_remote = {} # Remotes that each local branch is tracking - self.content = {} - super().__init__(parent, auto_load=auto_load) + trees = [line.strip() for line in proc.stdout.splitlines() if line.strip()] + return len(trees) == 1 and bool(target_tree) and trees[0] == target_tree - @runez.cached_property - def repo_name(self): + def cleanable_local_branch(self, name: str, include_current=False) -> CleanableLocalBranch | None: """ - :return str: Most significant repository name + :param str name: Local branch name + :param bool include_current: If True, allow checking the current branch + :return CleanableLocalBranch|None: Cleanup details if local branch is safe to clean """ - if self.origin: - return self.origin.name + 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 - for r in self.remotes.values(): - return r.name + if self.is_ancestor(name, base_ref): + return CleanableLocalBranch(name) - return None + if self.merge_is_noop(name, base_ref): + return CleanableLocalBranch(name, force_delete=True) - def _process_line(self, line): - k, _, v = line.partition("=") - self.content[k] = v - if k.startswith("remote."): - if k.endswith(".url"): - k = k[7:-4] - url = GitURL() - url.set(v) - self.remotes[k] = url - if k == "origin": - self.origin = url + return None - elif k.startswith("branch.") and k.endswith(".remote"): - self.tracking_remote[k[7:-7]] = v + 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 + + 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)} -class GitStatus(GitAspect): +class GitStatus: """Currently modified files""" - _command = "status --porcelain --branch" - - def __init__(self, parent, auto_load=True): + def __init__(self, parent: GitDir): + self.ahead = 0 + self.behind = 0 self.modified = [] self.untracked = [] self.report = GitRunReport() - super().__init__(parent, auto_load=auto_load) - - @property - def freshness(self): - """Short freshness overview""" - result = [] - if self.report._problem: - result.append(runez.red(" ".join(self.report._problem))) - - if self.modified: - result.append(runez.red(runez.plural(self.modified, "diff"))) - - if self.untracked: - result.append(runez.orange("%s untracked" % len(self.untracked))) - - if self.report._note: - result.append(runez.purple(" ".join(self.report._note))) - - if not self.report._problem and not self.report._note and self._parent.age is not None: - message = "up to date" - if self._parent.age > FRESHNESS_THRESHOLD: - message += "*" - - result.append(runez.teal(message)) - - return ", ".join(result) - - def _process_line(self, line): - if line[0] == "#": - if "..." not in line: - return + lines = parent.checked_git_command_lines("status", "--porcelain=v2", "--branch") + for line in lines: + prefix = line[0] + info = line[2:] + if prefix == "#": + keyword, _, value = info.partition(" ") + if keyword == "branch.ab": + ab = value.partition(" ") + self.ahead = int(ab[0]) + self.behind = -int(ab[2]) + + elif prefix == "?": + self.untracked.append(f" ? {info}") - m = RE_BRANCH_STATUS.match(line) - if not m: - LOG.warning("Unrecognised git status line: '%s'", line) - return - - text = str(m.group(6) or "") # behind, ahead, or gone - if not text: - return - - for message in text.split(","): - message = message.strip() - if "gone" in message: - line = line.lower() - if "no commits yet" in line or "initial commit on" in line: - self.report.add(note="no commits yet") - - else: - self.report.add(problem="remote branch gone") + else: + fields = info.split(" ") + state = fields[0].replace(".", " ") + path = fields[-1].partition("\t")[0] + self.modified.append(f"{state} {path}") - elif "ahead" in message: - self.report.add(problem=message) + if parent.refs.upstream_gone(): + self.report.add(problem="remote branch gone") - else: - self.report.add(note=message) + @property + def dirty_note(self) -> str: + """Short overview of pending changes.""" + return Reporter.joined( + self.modified and Reporter.problem(runez.plural(self.modified, "diff")), + self.untracked and Reporter.untracked_change(f"{len(self.untracked)} untracked"), + ) - return + @property + def has_pending_changes(self) -> bool: + return bool(self.modified or self.untracked) + + def upstream_delta(self) -> str: + """Short report on divergence from the upstream branch.""" + return Reporter.joined( + self.ahead and Reporter.note(f"{self.ahead} ahead"), + self.behind and Reporter.note(f"{self.behind} behind"), + ) + + def pending_change_counts(self) -> tuple[int, int, int]: + edits = 0 + deletes = 0 + new = len(self.untracked) + for item in self.modified: + state = item[0:2] + if "D" in state: + deletes += 1 + + elif "A" in state: + new += 1 - if line[0] == "?": - self.untracked.append(line) - return + else: + edits += 1 - self.modified.append(line) + return edits, deletes, new + def require_clean(self, operation: str): + if self.has_pending_changes: + Reporter.abort(f"can't {operation}: pending changes: {self.dirty_note}") -def _report_sorter(enum): - """ - :param tuple(int, str) enum: Tuple from enumerate() - :return int: Value to use for sorting messages in this report - """ - _, message = enum - if message[0] == "<": - return -enum[0] # '<' makes message sort towards front, but keeping order with other such prefixed messages - if message[0] == ">": - return 1000000 + enum[0] # '>' makes message sort towards end +def _add_messages(target: list[str], messages: str | list[str] | None): + if messages: + if not isinstance(messages, list): + messages = [messages] - return enum[0] # Non-prefixed message stay where they were + for message in messages: + if message not in target: + target.append(message) -def _add_sorted(result, target, color, n, max_chars): +def _add_sorted(result, target, color, n, max_chars) -> int: """ :param list(str) result: Where to accumulate sorted report :param list(str) target: Target to sort, respecting '<' and '>' prefixing @@ -808,3 +695,18 @@ def _add_sorted(result, target, color, n, max_chars): result.extend(color(s) for s in items) return n + + +def _report_sorter(enum): + """ + :param tuple(int, str) enum: Tuple from enumerate() + :return int: Value to use for sorting messages in this report + """ + _, message = enum + if message[0] == "<": + return -enum[0] # '<' makes message sort towards front, but keeping order with other such prefixed messages + + if message[0] == ">": + return 1000000 + enum[0] # '>' makes message sort towards end + + return enum[0] # Non-prefixed message stay where they were diff --git a/tests/conftest.py b/tests/conftest.py index e5437de..f868dc3 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,5 +1,137 @@ +from __future__ import annotations + +import shutil +import subprocess +import tempfile +from pathlib import Path + +import pytest +import runez from runez.conftest import cli, ClickRunner # noqa: F401, fixtures from mgit.cli import main ClickRunner.default_main = main +GIT_PATH = shutil.which("git") + + +class TempGitRepo: + def __init__(self, relative_path: str, configure=True): + cwd = Path.cwd().resolve() + # Ensure this fixture is used together with `cli` fixture, which ensures we're in a temp folder + # We can reconsider this later... but git operations here assume we're working in a temp folder + assert cwd.is_relative_to(Path(tempfile.gettempdir()).resolve()) + self.cwd = cwd / relative_path + if configure: + self.run_git("config", "user.email", "tester@example.com") + self.run_git("config", "user.name", "Test User") + + @classmethod + def clone(cls, remote_url: str, relative_path: str) -> TempGitRepo: + cls.ensure_parent(relative_path) + cls.run_git_command("clone", remote_url, relative_path) + repo = cls(relative_path) + return repo + + @classmethod + def init(cls, relative_path: str, configure=True, initial_branch="main", include_readme=True) -> TempGitRepo: + cls.ensure_parent(relative_path) + args = [] + if initial_branch: + args.append(f"--initial-branch={initial_branch}") + + cls.run_git_command("init", *args, relative_path) + repo = cls(relative_path, configure=configure) + if include_readme: + repo.add_file("README.md", f"# README for {relative_path}") + repo.commit("Initial commit") + + return repo + + @classmethod + def init_bare(cls, relative_path: str, initial_branch="main") -> TempGitRepo: + cls.ensure_parent(relative_path) + cls.run_git_command("init", "--bare", f"--initial-branch={initial_branch}", relative_path) + return TempGitRepo(relative_path, configure=False) + + @staticmethod + def ensure_parent(relative_path: str): + parent = Path(relative_path).parent + if parent != Path("."): + parent.mkdir(parents=True, exist_ok=True) + + @staticmethod + def run_git_command(*args, check=True, cwd=None) -> str: + if cwd: + args = ["-C", str(cwd), *args] + + assert GIT_PATH + proc = subprocess.run([GIT_PATH, *args], check=check, capture_output=True, text=True) # noqa: S603 + return proc.stdout.strip() + + @classmethod + def seeded(cls, relative_path="work") -> TempGitRepo: + """Repo seeded with a remote""" + return cls.seeded_set(relative_path).work + + @classmethod + def seeded_set(cls, relative_path="work") -> SeededRepoSet: + """Set of 3 repos, one remote, one seed and one work repo""" + return SeededRepoSet(relative_path) + + def run_git(self, *args, check=True) -> str: + return self.run_git_command(*args, check=check, cwd=self.cwd) + + def add(self, relative_path: str) -> str: + return self.run_git("add", relative_path) + + def add_file(self, relative_path: str, contents: str): + self.write_file(relative_path, contents) + self.add(relative_path) + + def branch(self, *args) -> str: + return self.run_git("branch", *args) + + def commit_file(self, relative_path: str, contents: str, message: str): + self.add_file(relative_path, contents) + self.commit(message) + + def checkout(self, *args) -> str: + return self.run_git("checkout", *args) + + def commit(self, message="Initial commit") -> str: + return self.run_git("commit", "-m", message) + + def push(self, *args) -> str: + return self.run_git("push", *args) + + def remote(self, *args) -> str: + return self.run_git("remote", *args) + + @property + def current_branch(self) -> str: + return self.branch("--show-current") + + def has_branch(self, name: str) -> bool: + return bool(self.branch("--list", name)) + + def write_file(self, relative_path: str, contents: str): + runez.write(self.cwd / relative_path, contents + "\n", logger=None) + + +class SeededRepoSet: + def __init__(self, relative_path: str): + self.remote = TempGitRepo.init_bare(f"{relative_path}-bare.git") + self.seed = TempGitRepo.init(f"{relative_path}-seed") + self.seed.remote("add", "origin", self.remote_url) + self.seed.push("-u", "origin", "main") + self.work = TempGitRepo.clone(self.remote_url, relative_path) + + @property + def remote_url(self): + return str(self.remote.cwd) + + +@pytest.fixture +def git(): + return TempGitRepo diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 0000000..6a7c983 --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,275 @@ +from __future__ import annotations + +import os +import time +from pathlib import Path + +import runez + +from mgit.cli import normalized_cli_args + + +def test_command_help(cli): + cli.run("--help") + assert cli.succeeded + assert "Commands:" in cli.logged + assert "status (s)" in cli.logged + + cli.run("s --help") + assert cli.succeeded + assert "usage: mgit status" in cli.logged + assert "Show repo or workspace status." in cli.logged + assert "folder" in cli.logged + + +def test_cli_arg_normalization(): + assert normalized_cli_args([]) == ["status"] + 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"] + + +def test_abort_reporting(cli, git): + cli.run("--color") + assert cli.failed + assert "argument --color: expected one argument" in cli.logged.stderr + + cli.run("-z") + assert cli.failed + assert "error: unrecognized arguments: -z" in cli.logged.stderr + + cli.run("--color never missing") + assert cli.failed + assert "No folder" in cli.logged + + Path("folder").mkdir() + cli.run("folder") + assert cli.failed + assert "no git folders" in cli.logged + + git.init("folder/repo") + cli.run("main folder") + assert cli.failed + assert "main only supports one git checkout" in cli.logged + + +def test_single_status_shows_pending_paths_and_discovers_parent_checkout(cli, git): + repo = git.init("repo") + repo.write_file("README.md", "changed") + repo.write_file("new.txt", "new") + + cli.run("repo") + assert cli.succeeded + assert "main ✅ ✏ī¸1 🆕1" in cli.logged + assert "README.md" in cli.logged + assert "new.txt" in cli.logged + + (repo.cwd / "foo").mkdir() + os.chdir(repo.cwd / "foo") + cli.run() + assert cli.succeeded + assert "README.md" in cli.logged + assert "new.txt" in cli.logged + + cli.run(".") + assert cli.failed + assert "repo/foo: no git folders" in cli.logged + + +def test_status_line_summarizes_stale_and_cleanable_branches(cli, git): + repo = git.seeded("repo") + repo.checkout("-b", "stale") + repo.push("-u", "origin", "stale") + repo.checkout("main") + repo.push("origin", "--delete", "stale") + old_time = time.time() - (13 * 60 * 60 + 60) + for name in ("FETCH_HEAD", "HEAD"): + path = repo.cwd / ".git" / name + if path.exists(): + os.utime(path, (old_time, old_time)) + + cli.run("repo") + + assert cli.succeeded + assert "main ☑ī¸ [+1đŸĒĻ] ⌛13h" in cli.logged + + +def test_status_line_partitions_other_branches(cli, git): + repo = git.seeded("repo") + for branch in ("foo", "bar"): + repo.checkout("-b", branch, "main") + repo.push("-u", "origin", branch) + repo.checkout("main") + repo.push("origin", "--delete", branch) + + repo.checkout("-b", "baz", "main") + repo.commit_file("baz.txt", "pending", "baz work") + repo.checkout("main") + + cli.run("repo") + assert cli.succeeded + assert "main ✅ [+2đŸĒĻ+1]" in cli.logged + + repo.checkout("foo") + cli.run("repo") + assert cli.succeeded + assert "foo đŸĒĻ [+1đŸĒĻ+1]" in cli.logged + + +def test_dirty_orphan_status_keeps_tombstone(cli, git): + repo = git.seeded("repo") + repo.checkout("-b", "orphan") + repo.push("-u", "origin", "orphan") + repo.checkout("main") + repo.push("origin", "--delete", "orphan") + repo.checkout("orphan") + repo.write_file("scratch.txt", "pending") + + cli.run("repo") + + assert cli.succeeded + assert "orphan đŸĒĻ 🆕1" in cli.logged + + +def test_verbose_enables_debug_logging(cli, git): + git.init("repo") + + cli.run("repo") + assert cli.succeeded + assert "DEBUG Running:" not in cli.logged + + cli.run("-v repo") + assert cli.succeeded + assert "DEBUG Running:" 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") + + cli.run("branches workspace") + + assert cli.succeeded + assert "workspace:" not in cli.logged + assert "one:" in cli.logged + assert "two:" in cli.logged + assert "* topic" in cli.logged + assert "[default]" in cli.logged + assert "[orphaned]" in cli.logged + + +def test_branch_names_are_styled_consistently(cli, git): + repo = git.seeded("repo") + repo.checkout("-b", "topic") + repo.push("-u", "origin", "topic") + repo.checkout("-b", "orphan") + repo.push("-u", "origin", "orphan") + repo.checkout("topic") + repo.push("origin", "--delete", "orphan") + with runez.ActivateColors(True): + default_name = runez.green(runez.bold("main")) + topic_name = runez.bold("topic") + orphan_name = runez.orange(runez.bold("orphan")) + + 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 + + cli.run("--color always repo") + assert cli.succeeded + assert f"{topic_name} ✅" in cli.logged + assert "[+1đŸĒĻ]" in cli.logged + + cli.run("--color always pull repo") + assert cli.succeeded + assert f"{topic_name} ✅" in cli.logged + + cli.run("--color always main repo") + assert cli.succeeded + assert f"checked out {default_name}" in cli.logged + + repo.checkout("orphan") + cli.run("--color always repo") + assert cli.succeeded + assert f"{orphan_name} đŸĒĻ" in cli.logged + + cli.run("--color always fetch repo") + assert cli.succeeded + assert f"{orphan_name} đŸĒĻ" in cli.logged + + +def test_pull_workspace_recaps_each_checkout(cli, git): + foo_source = git.seeded_set("foo-source") + bar_source = git.seeded_set("bar-source") + foo = git.clone(foo_source.remote_url, "workspace/foo") + git.clone(bar_source.remote_url, "workspace/bar-baz") + detached = git.clone(foo_source.remote_url, "workspace/detached") + detached.checkout("--detach", "HEAD") + git.init("workspace/broken") + foo_source.seed.commit_file("next.txt", "next", "next") + foo_source.seed.push() + foo.run_git("fetch") + + cli.run("workspace") + assert cli.succeeded + assert "workspace:" not in cli.logged + assert "foo: main ✅ 1 behind" in cli.logged + assert "detached: HEAD đŸ‘ģ" in cli.logged + assert "HEAD detached" not in cli.logged + + cli.run("fetch workspace") + assert cli.succeeded + + cli.run("pull workspace") + assert cli.succeeded + assert "foo: main ✅ was 1 behind" in cli.logged + assert "bar-baz: main ✅ was up-to-date" in cli.logged + assert "broken: main ✅ can't pull; no remotes" in cli.logged + assert "detached: HEAD đŸ‘ģ can't pull; HEAD detached" in cli.logged + assert "HEAD detached; HEAD detached" not in cli.logged + assert foo.run_git("rev-parse", "HEAD") == foo.run_git("rev-parse", "origin/main") + assert detached.current_branch == "" + + # Single fails with exit code != 0 + cli.run("pull workspace/broken") + assert cli.failed + assert "can't pull; no remotes" in cli.logged + + cli.run("pull workspace/detached") + assert cli.failed + assert "can't pull; HEAD detached" in cli.logged + assert detached.current_branch == "" + + +def test_pull_refuses_untracked_changes(cli, git): + repo = git.seeded("repo") + repo.write_file("scratch.txt", "pending") + + cli.run("pull repo") + + assert cli.failed + assert "can't pull; pending changes" in cli.logged + + +def test_pull_failure_retains_previous_status_note(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 + + +def test_main_checks_out_default_branch(cli, git): + repo = git.init("repo") + repo.checkout("-b", "feature") + + cli.run("m repo") + + assert cli.succeeded + assert repo.current_branch == "main" diff --git a/tests/test_git.py b/tests/test_git.py deleted file mode 100644 index 965fe80..0000000 --- a/tests/test_git.py +++ /dev/null @@ -1,53 +0,0 @@ -from mgit.git import GitURL, is_valid_branch_name - - -def test_valid_branch_names(): - assert is_valid_branch_name("foo") - assert is_valid_branch_name("feature/foo") - assert is_valid_branch_name("feature/foo.bar") - - -def test_invalid_branch_names(): - assert not is_valid_branch_name(None) - assert not is_valid_branch_name("") - assert not is_valid_branch_name("~/foo") - assert not is_valid_branch_name("foo..bar") - assert not is_valid_branch_name("foo bar") - - -def check_url(url, protocol, hostname, username, relative_path, repo, name): - u = GitURL() - u.set(url) - assert str(u) == (url or "") - assert u.url == (url or "") - assert u.protocol == protocol - assert u.hostname == hostname - assert u.username == username - assert u.relative_path == relative_path - assert u.name == name - assert u.repo == repo - - -def test_git_urls(): - check_url(None, "unknown", "unknown", None, "", "unknown", "unknown") - check_url("", "unknown", "unknown", None, "", "unknown", "unknown") - check_url("~/foo/bar", "file", "local", None, "~/foo/bar", "foo", "bar") - check_url("/some/repo/foo", "file", "local", None, "/some/repo/foo", "repo", "foo") - - check_url( - "ssh://git@stash.corp.foo.com:7999/myproject/bin.git", "ssh", "stash.corp.foo.com", "git", "/myproject/bin.git", "myproject", "bin" - ) - check_url( - "https://user@stash.corp.foo.com/scm/myproject/bin.git", - "https", - "stash.corp.foo.com", - "user", - "/scm/myproject/bin.git", - "myproject", - "bin", - ) - - check_url("git@github.com:foo/vmaf.git", "ssh", "github.com", "git", "foo/vmaf.git", "foo", "vmaf") - check_url("https://github.com/foo/vmaf.git", "https", "github.com", None, "/foo/vmaf.git", "foo", "vmaf") - - check_url("git@example.com:80/foo/vmaf.git", "ssh", "example.com", "git", "/foo/vmaf.git", "foo", "vmaf") diff --git a/tests/test_groom.py b/tests/test_groom.py new file mode 100644 index 0000000..84247a9 --- /dev/null +++ b/tests/test_groom.py @@ -0,0 +1,174 @@ +import pytest + +from mgit.cli import GroomCommand + + +def make_stale_tracked_branch(work, branch="stale", merged=True): + work.checkout("-b", branch) + if not merged: + work.commit_file(f"{branch}.txt", "still in progress", f"{branch} work") + + work.push("-u", "origin", branch) + work.checkout("main") + work.push("origin", "--delete", branch) + work.checkout(branch) + + +def make_cleanable_remote_branch(work, branch="published", squash=False): + work.checkout("-b", branch) + work.commit_file(f"{branch}.txt", "completed", f"{branch} work") + work.push("-u", "origin", branch) + work.checkout("main") + if squash: + work.run_git("merge", "--squash", branch) + work.commit(f"squash {branch}") + else: + work.run_git("merge", "--no-ff", branch, "-m", f"merge {branch}") + + work.push("origin", "main") + work.checkout(branch) + + +def test_groom_deletes_stale_tracked_branch(cli, git): + work = git.seeded("work") + make_stale_tracked_branch(work) + + cli.run("g work") + assert cli.succeeded + assert "Checked out main branch" in cli.logged + assert "Deleted branch stale" in cli.logged + assert "main ✅ was up-to-date" in cli.logged + assert work.current_branch == "main" + assert not work.has_branch("stale") + + # 2nd groom: no-op + cli.run("g work") + assert cli.succeeded + assert "main ✅ already on default branch" in cli.logged + + +def test_groom_only_deletes_the_branch_being_groomed(cli, git): + work = git.seeded("work") + make_stale_tracked_branch(work, branch="leave-me") + work.checkout("main") + make_stale_tracked_branch(work, branch="clean-me") + + cli.run("g work") + + assert cli.succeeded + assert "Deleted branch clean-me" in cli.logged + assert "Deleted branch leave-me" not in cli.logged + assert "leave-me" not in cli.logged + assert not work.has_branch("clean-me") + assert work.has_branch("leave-me") + + cli.run("g work") + assert cli.succeeded + assert "main ✅ [+1đŸĒĻ] already on default branch" in cli.logged + assert "leave-me" not in cli.logged + assert work.has_branch("leave-me") + + +def test_groom_deletes_cleanable_local_only_current_branch(cli, git): + work = git.seeded("work") + work.checkout("-b", "local-only") + + cli.run("g work") + + assert cli.succeeded + assert "Deleted branch local-only" in cli.logged + assert not work.has_branch("local-only") + + +def test_groom_on_default_branch_only_fetches_and_reports(cli, git): + work = git.seeded("work") + work.write_file("scratch.txt", "pending") + + cli.run("g work") + + assert cli.succeeded + assert "already on default branch" in cli.logged + assert work.current_branch == "main" + + +@pytest.mark.parametrize("squash", [False, True]) +def test_groom_deletes_cleanable_current_remote_branch(cli, git, squash): + work = git.seeded(f"work-{squash}") + make_cleanable_remote_branch(work, squash=squash) + + cli.run(f"g {work.cwd.name}") + + assert cli.succeeded + assert "Deleted remote branch origin/published" in cli.logged + assert "Deleted branch published" in cli.logged + assert work.current_branch == "main" + assert not work.has_branch("published") + assert not work.run_git("ls-remote", "--heads", "origin", "refs/heads/published") + + +def test_groom_does_not_delete_remote_branch_that_advanced_after_fetch(cli, git, monkeypatch): + repos = git.seeded_set("work") + work = repos.work + make_cleanable_remote_branch(work) + racer = git.clone(repos.remote_url, "racer") + racer.checkout("-b", "published", "origin/published") + + checkout_default_branch = GroomCommand._checkout_default_branch + + def advance_remote_after_checkout(command, git_dir, branch): + checkout_default_branch(command, git_dir, branch) + racer.commit_file("late.txt", "not merged", "advance published") + racer.push("origin", "published") + + monkeypatch.setattr(GroomCommand, "_checkout_default_branch", advance_remote_after_checkout) + cli.run("g work") + + assert cli.failed + assert "git push --force-with-lease=refs/heads/published:" in cli.logged + assert "--delete origin refs/heads/published failed:" in cli.logged + assert work.has_branch("published") + assert work.run_git("ls-remote", "--heads", "origin", "refs/heads/published") + + +def test_groom_refuses_published_branch_with_unmerged_remote_work(cli, git): + repos = git.seeded_set("work") + work = repos.work + work.checkout("-b", "published") + work.push("-u", "origin", "published") + racer = git.clone(repos.remote_url, "racer") + racer.checkout("-b", "published", "origin/published") + racer.commit_file("remote-only.txt", "not merged", "advance published") + racer.push("origin", "published") + + cli.run("g work") + + assert cli.failed + assert "Remote branch origin/published can't be cleaned automatically" in cli.logged + assert work.current_branch == "published" + assert work.has_branch("published") + assert work.run_git("ls-remote", "--heads", "origin", "refs/heads/published") + + +def test_groom_refuses_pending_changes(cli, git): + work = git.seeded("work") + make_stale_tracked_branch(work) + work.write_file("scratch.txt", "pending") + + cli.run("groom work") + + assert cli.failed + assert "can't groom: pending changes" in cli.logged + assert work.current_branch == "stale" + assert work.has_branch("stale") + + +def test_groom_refuses_uncleanable_current_branch(cli, git): + work = git.seeded("work") + make_stale_tracked_branch(work, branch="unmerged", merged=False) + + cli.run("g work") + + assert cli.failed + assert "Branch unmerged can't be cleaned" in cli.logged + assert work.current_branch == "unmerged" + assert work.has_branch("unmerged") diff --git a/tests/test_mgit.py b/tests/test_mgit.py deleted file mode 100644 index 2a35dce..0000000 --- a/tests/test_mgit.py +++ /dev/null @@ -1,50 +0,0 @@ -import os - -import pytest -import runez - -import mgit - - -def test_edge_cases(): - assert mgit.git_parent_path("/") is None - assert mgit.git_parent_path(runez.DEV.tests_folder) == runez.DEV.project_folder - - prefs = mgit.MgitPreferences(all=True, fetch=False, pull=False, short=None) - assert str(prefs) == "align all !fetch !pull !verbose" - - prefs = mgit.MgitPreferences(name_size=5) - prefs.fetch = None - assert str(prefs) == "name_size=5" - - prefs = mgit.MgitPreferences() - assert not str(prefs) - - with pytest.raises(Exception, match="Internal error: add support for flag 'foo'"): - prefs.update(foo=1) - - -def test_usage(cli): - cli.expect_success("--help") - cli.expect_success("--version") - cli.expect_failure("--foo", "No such option") - - -def test_status(cli): - # Note: using explicit lists below, to support case where used directory path may have a space in it - # [wouldn't work if args passed as string, due to naive split in run()] - # Status on a non-existing folder should fail - cli.expect_failure("foo", "No folder 'foo'") - - # Status on this test folder should succeed and report no git folders found - cli.expect_success(cli.tests_folder, "no git folders") - - # Status on project folder should succeed (we're not calling fetch) - project = runez.DEV.project_folder - cli.expect_success(project, "mgit") - with runez.CurrentFolder(project): - cli.run() - assert cli.succeeded - assert "%s:" % os.path.basename(project) in cli.logged.stdout - - cli.expect_success("-cs") diff --git a/tests/test_reporting.py b/tests/test_reporting.py index e93d3b6..5dfaf44 100644 --- a/tests/test_reporting.py +++ b/tests/test_reporting.py @@ -1,11 +1,15 @@ -import pytest +from __future__ import annotations -from mgit.git import GitRunReport +from mgit.git import GitRunReport, Reporter -def check_sorting(given: str, expected, max_chars=160): +def check_sorting(given: GitRunReport | str, expected, max_chars=120): if not isinstance(given, GitRunReport): - given = GitRunReport(problem=given.split()) + report = GitRunReport() + for problem in given.split(): + report.add(problem=problem) + + given = report assert isinstance(given, GitRunReport) s = given.representation(max_chars=max_chars) @@ -13,8 +17,6 @@ def check_sorting(given: str, expected, max_chars=160): def test_reporting(): - assert GitRunReport.not_git().has_problems - # Sorting check_sorting("a b c", "a; b; c") # Messages stay in order they were provided check_sorting("a b c b a", "a; b; c") # No dupes @@ -24,35 +26,33 @@ def test_reporting(): check_sorting("a c e f", "d; b; a; f; c; e") # > ordered pushing # Typical issues - check_sorting(GitRunReport.not_git(), "not a git checkout") - check_sorting(GitRunReport.not_git().cant_pull(), "can't pull; not a git checkout") check_sorting(GitRunReport().cant_pull(), "can't pull") check_sorting(GitRunReport().cant_pull("foo"), "can't pull; foo") - # Adding nothing does nothing + # Empty messages are ignored check_sorting(GitRunReport().cant_pull().add(), "can't pull") - check_sorting(GitRunReport().cant_pull().add(None), "can't pull") - check_sorting(GitRunReport().cant_pull().add([]), "can't pull") - check_sorting(GitRunReport().cant_pull().add(""), "can't pull") - - # Adding bogus things is detected - with pytest.raises(Exception, match="Internal error: invalid GitRunReport"): - GitRunReport().cant_pull().add(a=1) + check_sorting(GitRunReport().cant_pull().add(problem=""), "can't pull") - # Cumulating + # Adding another report r1 = GitRunReport(problem="p1", note="n1") r2 = GitRunReport(problem="p2", note="n2") check_sorting(GitRunReport(r1).add(r2), "p1; p2; n1; n2") - # Filtering - check_sorting(GitRunReport(r1).add(problem=r2), "p1; p2; n1") - # Problems come ahead of notes check_sorting(GitRunReport().add(problem="p1").add(problem="p2").add(problem="p3"), "p1; p2; p3") check_sorting(GitRunReport().add(problem="p1", note="n1").add(problem="p2"), "p1; p2; n1") # Progress comes ahead of notes, but after problems - check_sorting(GitRunReport().add(note=("n1", "