Remove --interactive and refactor

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"
 

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__":

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

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