diff --git a/pyproject.toml b/pyproject.toml index 54754f2..8e18903 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -2,11 +2,11 @@ name = "git-rebase-branches" description = "Rebase multiple branches at once" readme = "README.md" +license = "LGPL-2.0-or-later" requires-python = ">= 3.9" classifiers = [ "Development Status :: 5 - Production/Stable", "Intended Audience :: End Users/Desktop", - "License :: OSI Approved :: GNU Lesser General Public License v2 or later (LGPLv2+)", ] dynamic = ["version"] dependencies = [ @@ -16,9 +16,6 @@ dependencies = [ name = "David Tucker" email = "david@tucker.name" -[project.license] -file = "LICENSE.txt" - [project.scripts] git-rebase-branches = "git_rebase_branches:main" diff --git a/src/git_rebase_branches.py b/src/git_rebase_branches.py index 8bdad56..e7d42c0 100755 --- a/src/git_rebase_branches.py +++ b/src/git_rebase_branches.py @@ -9,8 +9,9 @@ import shlex import subprocess import sys +from contextlib import contextmanager from importlib.metadata import version -from typing import Any, Dict, List, Optional +from typing import Any, Iterator, Optional FAILURE_STATUS = "failed" @@ -40,12 +41,6 @@ def cli(parser: Optional[argparse.ArgumentParser] = None) -> argparse.ArgumentPa nargs="+", help="the branches to rebase", ) - parser.add_argument( - "-i", - "--interactive", - action="store_true", - help="exit instead of aborting a failed rebase", - ) parser.add_argument( "--version", action="version", @@ -55,26 +50,20 @@ def cli(parser: Optional[argparse.ArgumentParser] = None) -> argparse.ArgumentPa return parser -def main(argv: Optional[List[str]] = None) -> None: - """Execute CLI commands.""" - if argv is None: - argv = sys.argv[1:] - args = cli().parse_args(argv) +def branches_that_do_not_contain(ref: str, /) -> list[str]: + """Get a list of branches that do not contain a given ref.""" + result = run( + ["git", "branch", "--no-contains", ref, "--format=%(refname:short)"], + capture_output=True, + check=True, + encoding="utf-8", + ) + print(result.stdout, end="", flush=True) + return result.stdout.splitlines() - # Stash any changes. - if ( - args.interactive - and run(["git", "diff-index", "--exit-code", "HEAD", "--"]).returncode - ): - try: - input( - "\n" - "There are local changes that need to be stashed or committed.\n" - "Press ^C to stop here or Enter to stash them and continue.\n" - ) - except KeyboardInterrupt: - print() - sys.exit(1) + +def stash_changes() -> bool: + """Stash any local changes, and return whether changes were stashed.""" result = subprocess.run( ["git", "stash", "list"], capture_output=True, @@ -89,95 +78,96 @@ def main(argv: Optional[List[str]] = None) -> None: check=True, encoding="utf-8", ) - stashed_changes = len(result.stdout.splitlines()) - stashed_changes + return bool(len(result.stdout.splitlines()) - stashed_changes) + - # Note where we are so we can come back. +@contextmanager +def changes_stashed() -> Iterator[bool]: + """Stash any local changes, then un-stash them.""" + stashed_changes = stash_changes() + try: + yield stashed_changes + finally: + if stashed_changes: + run(["git", "stash", "pop"], check=True) + + +def ref_commit(ref: str, /) -> str: + """Get the commit for a given ref.""" result = subprocess.run( - ["git", "branch", "--show-current"], + ["git", "rev-parse", ref], capture_output=True, check=True, encoding="utf-8", ) - start = result.stdout.strip() - if not start: - result = subprocess.run( - ["git", "rev-parse", "HEAD"], - capture_output=True, - check=True, - encoding="utf-8", - ) - start = result.stdout.strip() + return result.stdout.strip() + + +def current_ref() -> str: + """Get the current Git branch, commit, etc.""" run(["git", "log", "-n1"], check=True) + result = subprocess.run( + ["git", "branch", "--show-current"], + capture_output=True, + check=True, + encoding="utf-8", + ) + return result.stdout.strip() or ref_commit("HEAD") + - # Get all the branches that need rebasing. +@contextmanager +def original_ref_preserved() -> Iterator[str]: + """Note the current ref, then restore it.""" + og_ref = current_ref() + try: + yield og_ref + finally: + run(["git", "-c", "advice.detachedHead=false", "checkout", og_ref], check=True) + + +@contextmanager +def original_state_preserved() -> Iterator[tuple[bool, str]]: + """Stash any local changes and note the current ref, then restore both.""" + with changes_stashed() as stashed_changes: + with original_ref_preserved() as og_ref: + yield stashed_changes, og_ref + + +def main(argv: Optional[list[str]] = None) -> None: + """Execute CLI commands.""" + if argv is None: + argv = sys.argv[1:] + args = cli().parse_args(argv) if args.branches is None: - # Check out to the ref to ensure it won't show up in the next command. - run( - ["git", "-c", "advice.detachedHead=false", "checkout", args.base_ref], - check=True, - ) - result = run( - ["git", "branch", "--no-contains", "HEAD", "--format=%(refname:short)"], - capture_output=True, - check=True, - encoding="utf-8", - ) - print(result.stdout, end="", flush=True) - args.branches = result.stdout.splitlines() + args.branches = branches_that_do_not_contain(args.base_ref) + + statuses: dict[str, str] = {} # Rebase each branch. - statuses: Dict[str, str] = {} - - def print_report() -> int: - failures = 0 - for i, branch in enumerate(sorted(args.branches)): - if i == 0: - print() - print("=" * 36, "SUMMARY", "=" * 36) - status = statuses.get(branch, "not attempted") - if status == FAILURE_STATUS: - failures += 1 - print("-", branch, f"({status})") + with original_state_preserved(): + for branch in args.branches: + try: + run(["git", "rebase", args.base_ref, branch], check=True) + except subprocess.CalledProcessError: + statuses[branch] = FAILURE_STATUS + run(["git", "rebase", "--abort"], check=True) + else: + statuses[branch] = "succeeded" - if stashed_changes: + # Report what happened. + for i, branch in enumerate(args.branches): + if i == 0: print() - run(["git", "stash", "list", f"-n{stashed_changes}"], check=True) - - return failures - - for branch in args.branches: - run(["git", "switch", branch], check=True) - try: - run(["git", "rebase", args.base_ref], check=True) - except subprocess.CalledProcessError: - statuses[branch] = FAILURE_STATUS - if args.interactive: - try: - input("\nPress ^C to stop here or Enter to continue.\n") - except KeyboardInterrupt: - print() - failures = print_report() - run(["git", "status"], check=True) - sys.exit(failures) - run(["git", "rebase", "--abort"], check=True) - else: - statuses[branch] = "succeeded" - - # Return to the original state. - run(["git", "-c", "advice.detachedHead=false", "checkout", start], check=True) - if stashed_changes: - run(["git", "stash", "pop"], check=True) - stashed_changes -= 1 + print("=" * 36, "SUMMARY", "=" * 36) + print("-", branch, f"({statuses[branch]})") - # Report what happened. - failures = print_report() - sys.exit(failures) + sys.exit(FAILURE_STATUS in statuses.values()) -def run(command: List[str], **kwargs: Any) -> "subprocess.CompletedProcess[str]": +def run(command: list[str], **kwargs: Any) -> "subprocess.CompletedProcess[str]": """Print a command before running it.""" print("$", shlex.join(str(token) for token in command), flush=True) - return subprocess.run(command, **kwargs) + return subprocess.run(command, check=kwargs.pop("check", True), **kwargs) if __name__ == "__main__": diff --git a/tests/test_git_rebase_branches.py b/tests/test_git_rebase_branches.py index 9b05bca..734801d 100644 --- a/tests/test_git_rebase_branches.py +++ b/tests/test_git_rebase_branches.py @@ -1,5 +1,7 @@ """Test git_rebase_branches.""" +import subprocess + import pytest import git_rebase_branches @@ -10,3 +12,58 @@ def test_main_version(): with pytest.raises(SystemExit) as excinfo: git_rebase_branches.main(["--version"]) assert excinfo.value.code == 0 + + +def branch_commit(): + """Get a mapping of branches to commits.""" + result = subprocess.run( + [ + "git", + "for-each-ref", + "--format=%(refname:short) %(objectname)", + "refs/heads", + ], + capture_output=True, + check=True, + encoding="utf-8", + ) + return { + branch: commit + for line in result.stdout.splitlines() + for branch, _, commit in [line.partition(" ")] + } + + +def test_git_rebase_branches(tmp_path, monkeypatch): + """Test a basic example.""" + base_ref = "main" + branch_no_contains = "init" + branch_contains = "latest" + + monkeypatch.chdir(tmp_path) + commit_opts = ["--allow-empty"] + for git_command in [ + ["init", "-b", base_ref], + ["config", "commit.gpgsign", "false"], + ["config", "user.name", "Coder Joe"], + ["config", "user.email", "Coder@Joe.com"], + ["commit"] + commit_opts + ["-m", "Initial Commit"], + ["branch", branch_no_contains], + ["commit"] + commit_opts + ["-m", "New Commit on main"], + ["branch", branch_contains], + ]: + subprocess.run(["git"] + git_command, check=True) + + head = git_rebase_branches.ref_commit("HEAD") + target_state = { + base_ref: head, + branch_no_contains: head, + branch_contains: head, + } + assert branch_commit() != target_state + + with pytest.raises(SystemExit) as excinfo: + git_rebase_branches.main([base_ref]) + assert excinfo.value.code == 0 + + assert branch_commit() == target_state diff --git a/tox.ini b/tox.ini index 211c0d8..8d7c913 100644 --- a/tox.ini +++ b/tox.ini @@ -25,7 +25,7 @@ deps = pytest-randomly ~= 3.16.0 pytest-xdist ~= 3.8.0 commands = - pytest --cov git_rebase_branches --cov-branch --cov-fail-under 20 --cov-report term-missing {posargs:-n auto} + pytest --cov git_rebase_branches --cov-branch --cov-fail-under 80 --cov-report term-missing {posargs:-n auto} [pytest] testpaths = tests