diff --git a/.github/workflows/validation.yml b/.github/workflows/validation.yml index d81b408..e66e64c 100644 --- a/.github/workflows/validation.yml +++ b/.github/workflows/validation.yml @@ -30,3 +30,26 @@ jobs: - name: Check typing with mypy run: | mypy . + + test: + runs-on: ubuntu-latest + needs: lint + steps: + - uses: actions/checkout@v2 + - name: Set up Python 3.12 + uses: actions/setup-python@v2 + with: + python-version: "3.12" + - name: Install dependencies + run: | + pip install -r requirements-dev.txt + - name: Run pytest + run: | + pytest --maxfail=3 --disable-warnings --tb=short + continue-on-error: false + - name: Upload pytest results on failure + if: failure() + uses: actions/upload-artifact@v4 + with: + name: pytest-failures + path: .pytest_cache diff --git a/.gitignore b/.gitignore index c18dd8d..9c882b7 100644 --- a/.gitignore +++ b/.gitignore @@ -1 +1,2 @@ __pycache__/ +*.egg-info/ diff --git a/.python-version b/.python-version new file mode 100644 index 0000000..e4fba21 --- /dev/null +++ b/.python-version @@ -0,0 +1 @@ +3.12 diff --git a/README.md b/README.md index 29de660..af2f4ad 100644 --- a/README.md +++ b/README.md @@ -5,7 +5,7 @@ Utility script that synchronizes a local repository with all remotes: - fetches all remotes - fast-forwards all local branches that have a remote upstream - pushes upstream changes to the default push remote - - fast-forwards branches associated with merged PRs to the merge commit + - fast-forwards or deletes branches associated with merged PRs to the merge commit Not all git configuration is taken into account; please open an issue if this causes problems. diff --git a/git_sync/__init__.py b/git_sync/__init__.py index d7ab7ce..ba7598d 100644 --- a/git_sync/__init__.py +++ b/git_sync/__init__.py @@ -1,32 +1,104 @@ import sys +from argparse import ArgumentParser, Namespace from asyncio import create_task, run from os import environ from .git import ( GitError, - fast_forward_merged_prs, + Remote, fast_forward_to_downstream, fetch_and_fast_forward_to_upstream, get_branches_with_remote_upstreams, get_default_push_remote, get_remotes, + in_git_repo, + update_merged_prs, ) -from .github import fetch_pull_requests +from .github import fetch_pull_requests, repos_by_domain + + +def github_token_envvar(domain: str) -> str: + return domain.split(".")[0].upper() + "_TOKEN" def github_token(domain: str) -> str | None: - envvar = domain.split(".")[0].upper() + "_TOKEN" - return environ.get(envvar) + return environ.get(github_token_envvar(domain)) + + +def get_description( + is_in_git_repo: bool, + push_remote: bytes | None, + remotes: list[Remote], + domains: list[str], +) -> str: + description = "Synchronize with remote repositories." + if not is_in_git_repo: + description += "\nRun --help inside a git repository to see what will be done." + return description + + description += "\nPulls all remotes, including updating the current branch if safe." + if push_remote and any(remote.name != push_remote for remote in remotes): + description += f"\nPushes any upstream changes to {push_remote.decode()}." + if domains: + domains_with_tokens = sorted( + domain for domain in domains if github_token(domain) + ) + domains_without_tokens = sorted( + domain for domain in domains if not github_token(domain) + ) + if domains_with_tokens: + description += ( + "\nMerged PR branches will be fast-forwarded or, if safe, deleted." + ) + if domains_without_tokens: + description += "\n" + description += "Full github" if domains_with_tokens else "Github" + description += " integration can be enabled by setting " + env_vars = [ + "$" + github_token_envvar(domain) for domain in domains_without_tokens + ] + description += ", ".join(env_vars) + description += "." + description += ( + "\n(NOTE: These behaviours are based on the current repository and environment" + " variables. Run --help inside another repository to determine the behaviour" + " there.)" + ) + return description + + +def get_command_line_args(description: str) -> Namespace: + parser = ArgumentParser(description=description) + parser.add_argument( + "--no-delete", + action="store_false", + help="Never delete branches", + default=True, + dest="allow_delete", + ) + return parser.parse_args() async def git_sync() -> None: - push_remote = await get_default_push_remote() + is_in_git_repo = await in_git_repo() + + push_remote = (await get_default_push_remote()) if is_in_git_repo else None + remotes: list[Remote] = [] + remote_urls: list[str] = [] if push_remote: remotes = await get_remotes() - if remotes: - pull_request_task = create_task( - fetch_pull_requests(github_token, [remote.url for remote in remotes]) - ) + remote_urls = [remote.url for remote in remotes] + domains = sorted(repos_by_domain(remote_urls).keys()) + + description = get_description(is_in_git_repo, push_remote, remotes, domains) + args = get_command_line_args(description) + + if not is_in_git_repo: + print("Error: Not in a git repository", file=sys.stderr) + sys.exit(2) + + if remote_urls: + pull_request_task = create_task(fetch_pull_requests(github_token, remote_urls)) try: branches = await get_branches_with_remote_upstreams() @@ -39,12 +111,14 @@ async def git_sync() -> None: if push_remote: await fast_forward_to_downstream(push_remote, branches) - if remotes: + if remote_urls: pull_requests = await pull_request_task push_remote_url = next( remote.url for remote in remotes if remote.name == push_remote ) - await fast_forward_merged_prs(push_remote_url, pull_requests) + await update_merged_prs( + push_remote_url, pull_requests, allow_delete=args.allow_delete + ) def main() -> None: diff --git a/git_sync/git.py b/git_sync/git.py index d9b2fda..44e4dc4 100644 --- a/git_sync/git.py +++ b/git_sync/git.py @@ -1,4 +1,4 @@ -from asyncio.subprocess import PIPE, create_subprocess_exec +from asyncio.subprocess import DEVNULL, PIPE, create_subprocess_exec from collections.abc import Iterable from dataclasses import dataclass @@ -44,6 +44,13 @@ class Branch: is_current: bool +async def in_git_repo() -> bool: + proc = await create_subprocess_exec( + "git", "rev-parse", "--is-inside-work-tree", stdout=DEVNULL, stderr=DEVNULL + ) + return await proc.wait() == 0 + + async def get_current_branch() -> bytes | None: raw_bytes = await git_output("symbolic-ref", "-q", "HEAD", check_return=False) return raw_bytes or None @@ -100,15 +107,7 @@ async def get_remote_branches(remote: bytes) -> list[bytes]: async def is_ancestor(commit1: _ExecArg, commit2: _ExecArg) -> bool: - """Return true if commit1 is an ancestor of commit2 - - For instance, 2d406492de55 is merge #11 and 8ee9b133bb73 #12: - >>> from asyncio import run - >>> run(is_ancestor("2d406492de55", "8ee9b133bb73")) - True - >>> run(is_ancestor("8ee9b133bb73", "2d406492de55")) - False - """ + """Return true if commit1 is an ancestor of commit2.""" try: await git("merge-base", "--is-ancestor", commit1, commit2) return True @@ -143,11 +142,68 @@ async def fast_forward_to_downstream( await git("push", push_remote, short_branch_name) -async def fast_forward_merged_prs( - push_remote_url: str, prs: Iterable[PullRequest] +async def get_upstream_branch(branch_name: bytes) -> bytes | None: + upstream = await git_output( + "for-each-ref", "--format=%(upstream)", b"refs/heads/" + branch_name + ) + if upstream and upstream.startswith(b"refs/heads/"): + return upstream.removeprefix(b"refs/heads/") + return None + + +async def branch_is_an_upstream(branch_name: bytes) -> bool: + """Return true if the branch is upstream of any other branch.""" + upstreams = await git_output("for-each-ref", "--format=%(upstream)", "refs/heads") + upstream_heads = { + upstream.removeprefix(b"refs/heads/") + for upstream in upstreams.strip().splitlines() + } + return branch_name in upstream_heads + + +async def update_merged_pr_branch( + branch_name: bytes, + merged_hash: str, + *, + allow_delete: bool = True, ) -> None: + """Delete or fast-forward a merged PR branch. + + If there are any uncommitted changes on the branch, it will be skipped. + If the branch is not an upstream of any other branch, it will be deleted. + """ current_branch = await get_current_branch() current_branch = current_branch and current_branch[11:] + if current_branch == branch_name: + any_staged_changes = await git_output( + "diff", "--cached", "--exit-code", check_return=False + ) + if any_staged_changes: + print(f"Staged changes on {branch_name.decode()}, skipping fast-forward") + return + any_unstaged_changes_to_committed_files = await git_output( + "diff", "--exit-code", check_return=False + ) + if any_unstaged_changes_to_committed_files: + print(f"Unstaged changes on {branch_name.decode()}, skipping fast-forward") + return + if allow_delete and not await branch_is_an_upstream(branch_name): + upstream = (await get_upstream_branch(branch_name)) or b"main" + await git("checkout", upstream) + await git("branch", "-D", branch_name) + else: + await git("reset", "--hard", merged_hash) + else: # noqa: PLR5501 + if allow_delete and not await branch_is_an_upstream(branch_name): + await git("branch", "-D", branch_name) + else: + await git("branch", "--force", branch_name, merged_hash) + print(f"Fast-forward {branch_name.decode()} to {merged_hash}") + + +async def update_merged_prs( + push_remote_url: str, prs: Iterable[PullRequest], *, allow_delete: bool = True +) -> None: branch_hashes = await get_branch_hashes() for pr in prs: branch_name = pr.branch_name.encode("utf-8") @@ -159,13 +215,13 @@ async def fast_forward_merged_prs( and push_remote_url in pr.repo_urls ): try: - branch_is_ancestor = await is_ancestor(branch_name, pr.branch_name) + branch_is_ancestor = await is_ancestor(branch_name, pr.branch_hash) except GitError: pass # Probably no longer have the commit hash else: if branch_is_ancestor: - print(f"Fast-forward {pr.branch_name} to {pr.merged_hash}") - if branch_name == current_branch: - await git("reset", "--hard", merged_hash) - else: - await git("branch", "--force", branch_name, merged_hash) + await update_merged_pr_branch( + branch_name=branch_name, + merged_hash=merged_hash, + allow_delete=allow_delete, + ) diff --git a/pyproject.toml b/pyproject.toml index fb19002..a0a84f7 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "git-sync" -version = "0.3" +version = "0.4" description = "Synchronize local git repo with remotes" authors = [{ name = "Alice Purcell", email = "alicederyn@gmail.com" }] requires-python = ">= 3.12" @@ -15,6 +15,7 @@ strict = true [tool.pytest.ini_options] addopts = "--doctest-modules" +asyncio_mode = "auto" [tool.ruff] target-version = "py310" diff --git a/requirements-dev.txt b/requirements-dev.txt index f8c8bd5..92008e2 100644 --- a/requirements-dev.txt +++ b/requirements-dev.txt @@ -1,4 +1,5 @@ -r requirements.txt pytest >= 7.1.2 +pytest-asyncio >= 1.0.0 mypy >= 0.950 ruff >= 0.6.6 diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/integration/__init__.py b/tests/integration/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py new file mode 100644 index 0000000..9624429 --- /dev/null +++ b/tests/integration/conftest.py @@ -0,0 +1,25 @@ +import os +import subprocess +from collections.abc import Iterator +from pathlib import Path + +import pytest + + +@pytest.fixture(autouse=True) +def run_in_git_repo(tmp_path: Path) -> Iterator[None]: + repo = tmp_path / "repo" + repo.mkdir() + original_dir = os.getcwd() + try: + os.chdir(repo) + subprocess.run(["git", "init", "-b", "main"], check=True) + subprocess.run(["git", "config", "advice.detachedHead", "false"], check=True) + subprocess.run(["git", "config", "user.email", "ci@example.com"], check=True) + subprocess.run(["git", "config", "user.name", "CI"], check=True) + subprocess.run( + ["git", "commit", "--allow-empty", "-m", "Initial commit"], check=True + ) + yield + finally: + os.chdir(original_dir) diff --git a/tests/integration/gitutils.py b/tests/integration/gitutils.py new file mode 100644 index 0000000..9fb1e1b --- /dev/null +++ b/tests/integration/gitutils.py @@ -0,0 +1,95 @@ +import subprocess +from pathlib import Path + + +def run_split_stdout(cmd: list[str]) -> list[str]: + return subprocess.run( + cmd, + check=True, + stdout=subprocess.PIPE, + encoding="utf-8", + ).stdout.splitlines() + + +def get_current_branch() -> str: + return run_split_stdout(["git", "symbolic-ref", "-q", "HEAD"])[0].removeprefix( + "refs/heads/" + ) + + +def stage_changes(**files: str) -> None: + for filename, content in files.items(): + Path(filename + ".txt").write_text(content) + subprocess.run( + ["git", "add", *(filename + ".txt" for filename in files)], check=True + ) + + +def all_staged_changes() -> dict[str, str]: + return { + Path(filename).stem: Path(filename).read_text() + for filename in run_split_stdout(["git", "diff", "--name-only", "--cached"]) + } + + +def create_commit(base_commit: str, **files: str) -> str: + subprocess.run(["git", "checkout", base_commit], check=True) + for filename, content in files.items(): + Path(filename + ".txt").write_text(content) + subprocess.run(["git", "add", "*"], check=True) + subprocess.run(["git", "commit", "-m", "commit"], check=True) + return subprocess.check_output(["git", "rev-parse", "HEAD"]).strip().decode() + + +def squash_merge(base_commit: str, pr_commit: str) -> str: + subprocess.run(["git", "checkout", base_commit], check=True) + subprocess.run(["git", "merge", "--squash", pr_commit], check=True) + subprocess.run(["git", "commit", "-m", "Squash commit"], check=True) + return subprocess.check_output(["git", "rev-parse", "HEAD"]).strip().decode() + + +def setup_branches(*, active_branch: str | None = None, **branches: str) -> None: + # Put git into detached head state + subprocess.run(["git", "checkout", "--detach", "-q"], check=True) + # List all branches + all_branches = subprocess.run( + ["git", "for-each-ref", "--format=%(refname:short)"], + check=True, + stdout=subprocess.PIPE, + encoding="utf-8", + ).stdout.splitlines() + # Delete all branches + for branch in all_branches: + subprocess.run(["git", "branch", "-q", "-D", branch], check=True) + # Set up the branches as desired + for branch_name, commit_hash in branches.items(): + subprocess.run( + ["git", "checkout", "-q", commit_hash, "-b", branch_name], check=True + ) + # Set the active branch, if any + if active_branch: + subprocess.run(["git", "checkout", "-q", active_branch], check=True) + + +def setup_upstreams(**upstreams: str) -> None: + for branch_name, upstream in upstreams.items(): + subprocess.run( + ["git", "branch", "--set-upstream-to", upstream, branch_name], check=True + ) + + +def all_branches() -> dict[str, str]: + return { + branch_name: subprocess.run( + ["git", "rev-parse", branch_name], + check=True, + stdout=subprocess.PIPE, + encoding="utf-8", + ).stdout.strip() + for branch_name in subprocess.run( + ["git", "branch", "-a", "--format", "%(refname:short)"], + check=True, + stdout=subprocess.PIPE, + encoding="utf-8", + ).stdout.splitlines() + } diff --git a/tests/integration/test_fast_forward_merged_prs.py b/tests/integration/test_fast_forward_merged_prs.py new file mode 100644 index 0000000..69981f1 --- /dev/null +++ b/tests/integration/test_fast_forward_merged_prs.py @@ -0,0 +1,290 @@ +from pathlib import Path + +from git_sync.git import update_merged_prs +from git_sync.github import PullRequest + +from .gitutils import ( + all_branches, + all_staged_changes, + create_commit, + get_current_branch, + setup_branches, + setup_upstreams, + squash_merge, + stage_changes, +) + +REPO_URL = "https://github.com/example/example.git" # Dummy URL for test + + +async def test_delete_merged_inactive_pr_branch() -> None: + # Given a merged PR + commit_a = create_commit("main", file="A\n") + commit_b = create_commit(commit_a, file="B\n") + commit_c = squash_merge(commit_a, commit_b) + commit_d = create_commit(commit_b, file="C\n") + setup_branches( + main=commit_c, my_pr=commit_b, more_work=commit_d, active_branch="main" + ) + setup_upstreams(my_pr="main", more_work="main") + pr = PullRequest( + branch_name="my_pr", + repo_urls=frozenset([REPO_URL]), + branch_hash=commit_b, + merged_hash=commit_c, + ) + + # When we update merged PRs + await update_merged_prs(REPO_URL, [pr]) + + # Then the PR branch is deleted + assert all_branches() == { + "main": commit_c, + "more_work": commit_d, + } + + +async def test_force_inactive_upstream_branch_to_merged_commit() -> None: + # Given a merged PR + commit_a = create_commit("main", file="A\n") + commit_b = create_commit(commit_a, file="B\n") + commit_c = squash_merge(commit_a, commit_b) + commit_d = create_commit(commit_b, file="C\n") + setup_branches( + main=commit_c, my_pr=commit_b, more_work=commit_d, active_branch="main" + ) + setup_upstreams(my_pr="main", more_work="my_pr") + pr = PullRequest( + branch_name="my_pr", + repo_urls=frozenset([REPO_URL]), + branch_hash=commit_b, + merged_hash=commit_c, + ) + + # When we update merged PRs + await update_merged_prs(REPO_URL, [pr]) + + # Then the PR branch is fast-forwarded to the merged commit + assert all_branches() == { + "main": commit_c, + "my_pr": commit_c, + "more_work": commit_d, + } + + +async def test_merged_inactive_pr_branch_with_deletion_disabled() -> None: + # Given a merged PR + commit_a = create_commit("main", file="A\n") + commit_b = create_commit(commit_a, file="B\n") + commit_c = squash_merge(commit_a, commit_b) + commit_d = create_commit(commit_b, file="C\n") + setup_branches( + main=commit_c, my_pr=commit_b, more_work=commit_d, active_branch="main" + ) + setup_upstreams(my_pr="main", more_work="main") + pr = PullRequest( + branch_name="my_pr", + repo_urls=frozenset([REPO_URL]), + branch_hash=commit_b, + merged_hash=commit_c, + ) + + # When we update merged PRs + await update_merged_prs(REPO_URL, [pr], allow_delete=False) + + # Then the PR branch is fast-forwarded to the merged commit + assert all_branches() == { + "main": commit_c, + "my_pr": commit_c, + "more_work": commit_d, + } + + +async def test_delete_merged_active_pr_branch() -> None: + # Given a merged PR + commit_a = create_commit("main", file="A\n") + commit_b = create_commit(commit_a, file="B\n") + commit_c = squash_merge(commit_a, commit_b) + commit_d = create_commit(commit_b, file="C\n") + setup_branches( + main=commit_c, my_pr=commit_b, more_work=commit_d, active_branch="my_pr" + ) + setup_upstreams(my_pr="main", more_work="main") + pr = PullRequest( + branch_name="my_pr", + repo_urls=frozenset([REPO_URL]), + branch_hash=commit_b, + merged_hash=commit_c, + ) + + # When we update merged PRs + await update_merged_prs(REPO_URL, [pr]) + + # Then the PR branch is deleted + assert all_branches() == { + "main": commit_c, + "more_work": commit_d, + } + # And the active branch is set to its upstream + assert get_current_branch() == "main" + + +async def test_force_active_upstream_branch_to_merged_commit() -> None: + # Given a merged PR + commit_a = create_commit("main", file="A\n") + commit_b = create_commit(commit_a, file="B\n") + commit_c = squash_merge(commit_a, commit_b) + commit_d = create_commit(commit_b, file="C\n") + setup_branches( + main=commit_c, my_pr=commit_b, more_work=commit_d, active_branch="my_pr" + ) + setup_upstreams(my_pr="main", more_work="my_pr") + pr = PullRequest( + branch_name="my_pr", + repo_urls=frozenset([REPO_URL]), + branch_hash=commit_b, + merged_hash=commit_c, + ) + + # When we update merged PRs + await update_merged_prs(REPO_URL, [pr]) + + # Then the PR branch is fast-forwarded to the merged commit + assert all_branches() == { + "main": commit_c, + "my_pr": commit_c, + "more_work": commit_d, + } + + +async def test_merged_active_upstream_branch_with_deletion_disabled() -> None: + # Given a merged PR + commit_a = create_commit("main", file="A\n") + commit_b = create_commit(commit_a, file="B\n") + commit_c = squash_merge(commit_a, commit_b) + commit_d = create_commit(commit_b, file="C\n") + setup_branches( + main=commit_c, my_pr=commit_b, more_work=commit_d, active_branch="my_pr" + ) + setup_upstreams(my_pr="main", more_work="main") + pr = PullRequest( + branch_name="my_pr", + repo_urls=frozenset([REPO_URL]), + branch_hash=commit_b, + merged_hash=commit_c, + ) + + # When we update merged PRs + await update_merged_prs(REPO_URL, [pr], allow_delete=False) + + # Then the PR branch is fast-forwarded to the merged commit + assert all_branches() == { + "main": commit_c, + "my_pr": commit_c, + "more_work": commit_d, + } + + +async def test_staged_changes_not_lost() -> None: + # Given staged changes over a merged PR + commit_a = create_commit("main", file="A\n") + commit_b = create_commit(commit_a, file="B\n") + commit_c = squash_merge(commit_a, commit_b) + setup_branches(main=commit_c, my_pr=commit_b, active_branch="my_pr") + stage_changes(file="C\n") + pr = PullRequest( + branch_name="my_pr", + repo_urls=frozenset([REPO_URL]), + branch_hash=commit_b, + merged_hash=commit_c, + ) + + # When we update merged PRs + await update_merged_prs(REPO_URL, [pr]) + + # Then the active branch is unaffected + assert all_branches() == { + "main": commit_c, + "my_pr": commit_b, + } + # And the staged changes are preserved + assert all_staged_changes() == {"file": "C\n"} + + +async def test_unstaged_changes_to_committed_files_not_lost() -> None: + # Given unstaged changes to committed files over a merged PR + commit_a = create_commit("main", file="A\n") + commit_b = create_commit(commit_a, file="B\n") + commit_c = squash_merge(commit_a, commit_b) + setup_branches(main=commit_c, my_pr=commit_b, active_branch="my_pr") + Path("file.txt").write_text("C\n") + pr = PullRequest( + branch_name="my_pr", + repo_urls=frozenset([REPO_URL]), + branch_hash=commit_b, + merged_hash=commit_c, + ) + + # When we update merged PRs + await update_merged_prs(REPO_URL, [pr]) + + # Then the active branch is unaffected + assert all_branches() == { + "main": commit_c, + "my_pr": commit_b, + } + # And the unstaged changes are preserved + assert Path("file.txt").read_text() == "C\n" + + +async def test_fastforward_when_pr_had_additional_commits() -> None: + # Given a merged PR with additional commits + commit_a = create_commit("main", file="A\n") + commit_b = create_commit(commit_a, file="B\n") + commit_c = create_commit(commit_b, file="C\n") + commit_d = squash_merge(commit_a, commit_c) + commit_e = create_commit(commit_b, file="D\n") + setup_branches( + main=commit_d, my_pr=commit_b, more_work=commit_e, active_branch="my_pr" + ) + setup_upstreams(my_pr="main", more_work="my_pr") + pr = PullRequest( + branch_name="my_pr", + repo_urls=frozenset([REPO_URL]), + branch_hash=commit_c, + merged_hash=commit_d, + ) + + # When we update merged PRs + await update_merged_prs(REPO_URL, [pr]) + + # Then the PR branch is fast-forwarded to the merged commit + assert all_branches() == { + "main": commit_d, + "my_pr": commit_d, + "more_work": commit_e, + } + + +async def test_no_fastforward_when_branch_has_additional_commits() -> None: + # Given a branch with additional commits + commit_a = create_commit("main", file="A\n") + commit_b = create_commit(commit_a, file="B\n") + commit_c = create_commit(commit_b, file="C\n") + commit_d = squash_merge(commit_a, commit_c) + setup_branches(main=commit_d, my_pr=commit_c, active_branch="my_pr") + pr = PullRequest( + branch_name="my_pr", + repo_urls=frozenset([REPO_URL]), + branch_hash=commit_b, + merged_hash=commit_d, + ) + + # When we update merged PRs + await update_merged_prs(REPO_URL, [pr]) + + # Then the PR branch is unaffected + assert all_branches() == { + "main": commit_d, + "my_pr": commit_c, + } diff --git a/tests/integration/test_is_ancestor.py b/tests/integration/test_is_ancestor.py new file mode 100644 index 0000000..c8b68e9 --- /dev/null +++ b/tests/integration/test_is_ancestor.py @@ -0,0 +1,13 @@ +from git_sync.git import is_ancestor + +from .gitutils import create_commit + + +async def test_is_ancestor_simple() -> None: + # Given two commits + base = create_commit("HEAD", foo="a") + second = create_commit(base, foo="b") + # Then base is an ancestor of second + assert await is_ancestor(base, second) is True + # And second is not an ancestor of base + assert await is_ancestor(second, base) is False