Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Empty file added src/cmds/list/__init__.py
Empty file.
6 changes: 6 additions & 0 deletions src/cmds/list/args_list.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,6 @@
from dataclasses import dataclass


@dataclass()
class ListArgs:
current_working_dir: str
92 changes: 92 additions & 0 deletions src/cmds/list/cmd_list.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,92 @@
import logging

import pygit2 as pg
from result import Err, Ok, Result

from ...errors.not_bare_repo_err import NotBareRepoErr
from ...helpers.config_file import ensure_config_exists, read_config
from ...helpers.find_git import get_git_dir
from .args_list import ListArgs
from .result_list import ListWorktreesErr
from .worktree_info import WorktreeInfo


def list_worktrees(list_args: ListArgs) -> Result[list[WorktreeInfo], ListWorktreesErr]:
log = logging.getLogger(__name__)

git_dir = get_git_dir(list_args.current_working_dir)
if not git_dir:
log.warning("No bare git repository found; working_directory=%s", list_args.current_working_dir)
return Err(NotBareRepoErr())

log.info("Git repository found; git_dir=%s", git_dir)
bare_repo: pg.Repository = pg.Repository(git_dir, flags=pg.enums.RepositoryOpenFlag.BARE)

worktree_names: list[str] = bare_repo.list_worktrees()
log.debug("Found worktrees; count=%s, names=%s", len(worktree_names), worktree_names)

worktrees: list[WorktreeInfo] = []

for name in worktree_names:
try:
wt: pg.Worktree = bare_repo.lookup_worktree(name)
except (KeyError, pg.GitError) as e:
log.warning("Failed to look up worktree, skipping; name=%s, error=%s", name, e)
continue

try:
wt_repo: pg.Repository = pg.Repository(wt.path)
actual_branch: str = wt_repo.head.shorthand
except (KeyError, pg.GitError) as e:
log.warning("Could not resolve branch for worktree; name=%s, error=%s", wt.name, e)
actual_branch = wt.name # fallback to name

has_unmerged: bool = _has_unmerged_commits(bare_repo, git_dir, actual_branch)

worktrees.append(
WorktreeInfo(
name=wt.name,
path=wt.path,
is_prunable=wt.is_prunable,
has_unmerged_commits=has_unmerged,
)
)

return Ok(worktrees)


def _has_unmerged_commits(repo: pg.Repository, git_dir: str, branch_name: str) -> bool:
log = logging.getLogger(__name__)

ensure_res = ensure_config_exists()
match ensure_res:
case Err(e):
log.warning("Config not found, defaulting to main; error=%s", e)
default_branch = "main"
case Ok(config_path):
read_res = read_config(config_path)
match read_res:
case Err(e):
log.warning("Failed to read config, defaulting to main; error=%s", e)
default_branch = "main"
case Ok(config):
default_branch = config.get(git_dir, "default_branch_name", fallback="main")

log.debug(
"Checking unmerged commits; branch=%s, default_branch=%s",
branch_name,
default_branch,
)

try:
branch_commit: pg.Commit = repo.lookup_reference(f"refs/heads/{branch_name}").peel(pg.Commit)
default_commit: pg.Commit = repo.lookup_reference(f"refs/heads/{default_branch}").peel(pg.Commit)

merge_base: pg.Oid | None = repo.merge_base(branch_commit.id, default_commit.id)
if merge_base is None:
return False
return merge_base != branch_commit.id

except (KeyError, pg.GitError) as e:
log.warning("Could not resolve refs for merge check, skipping; error=%s", e)
return False
3 changes: 3 additions & 0 deletions src/cmds/list/result_list.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,3 @@
from ...errors.not_bare_repo_err import NotBareRepoErr

ListWorktreesErr = NotBareRepoErr
9 changes: 9 additions & 0 deletions src/cmds/list/worktree_info.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,9 @@
from dataclasses import dataclass


@dataclass()
class WorktreeInfo:
name: str
path: str
is_prunable: bool
has_unmerged_commits: bool
2 changes: 1 addition & 1 deletion src/helpers/find_git.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@ def get_git_dir(start: str) -> str | None:

# If git is a file, then we're in a branch of the worktree
if git.is_file():
return str(path.parent)
return str(path.parent.absolute())

# If git is a dir, then we're not in a bare repo
if git.is_dir():
Expand Down
44 changes: 44 additions & 0 deletions src/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,8 @@
from .cmds.config.cmd_config import configure
from .cmds.destroy.args_destroy import DestroyArgs
from .cmds.destroy.cmd_destroy import destroy_repo
from .cmds.list.args_list import ListArgs
from .cmds.list.cmd_list import list_worktrees
from .cmds.pull.args_pull import PullArgs
from .cmds.pull.cmd_pull import pull_worktree
from .cmds.rm.args_rm import RmArgs
Expand Down Expand Up @@ -457,6 +459,48 @@ def pull(branch_name: str):
exit(ExitCode.SUCCESS)


@cli.command(name="list")
@click.option("--verbose", "-v", is_flag=True, default=False, help="Show worktree paths alongside names.")
def list_cmd(verbose: bool):
"""List all worktrees in the current bare repository."""

log = logging.getLogger(__name__)

list_args = ListArgs(current_working_dir=os.getcwd())

list_res = list_worktrees(list_args)

log.debug("Worktree list result; result=%s", list_res)

match list_res:
case Err(NotBareRepoErr()):
log.error("Cannot find a BARE git repository in the current working directory.")
exit(ExitCode.ERR_NOT_BARE_REPO)

case Err(_):
log.fatal("Something has gone horribly wrong. Aborting immediately!")
exit(ExitCode.ERR_GENERAL)

case Ok([]):
click.echo("No worktrees found.")
exit(ExitCode.SUCCESS)

case Ok(worktrees):
col_width = max(len(wt.name) for wt in worktrees) + 2 if verbose else 0
for wt in worktrees:
tags: list[str] = []
if wt.is_prunable:
tags.append("[prunable]")
if wt.has_unmerged_commits:
tags.append("[unmerged]")
tag_str = " " + " ".join(tags) if tags else ""
if verbose:
click.echo(f"{wt.name:<{col_width}}{wt.path}{tag_str}")
else:
click.echo(f"{wt.name}{tag_str}")
exit(ExitCode.SUCCESS)


@cli.command()
def switch():
"""
Expand Down
139 changes: 139 additions & 0 deletions tests/test_cmd_list.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,139 @@
from pathlib import Path

import pygit2 as pg
from result import Err, Ok

from src.cmds.list.args_list import ListArgs
from src.cmds.list.cmd_list import list_worktrees
from src.errors.not_bare_repo_err import NotBareRepoErr


def _make_bare_repo(path: Path) -> pg.Repository:
repo = pg.init_repository(str(path), bare=True)
sig = pg.Signature("Test", "test@test.com")
tree_id = repo.TreeBuilder().write()
repo.create_commit("refs/heads/main", sig, sig, "Initial commit", tree_id, [])
repo.set_head("refs/heads/main")
return repo


def _add_branch(repo: pg.Repository, branch_name: str, wt_path: Path) -> pg.Worktree:
main_commit: pg.Commit = repo.lookup_reference("refs/heads/main").peel(pg.Commit)
branch_ref: pg.Reference = repo.create_reference(f"refs/heads/{branch_name}", main_commit.id, False)
return repo.add_worktree(branch_name, str(wt_path), branch_ref)


class TestListWorktreesNotBareRepo:
def test_non_bare_repo_returns_err(self, tmp_path: Path) -> None:
non_bare = tmp_path / "not-bare"
non_bare.mkdir()
pg.init_repository(str(non_bare), bare=False)

args = ListArgs(current_working_dir=str(non_bare), )
result = list_worktrees(args)

assert isinstance(result, Err)
assert isinstance(result.err(), NotBareRepoErr)

def test_plain_directory_returns_err(self, tmp_path: Path) -> None:
args = ListArgs(current_working_dir=str(tmp_path), )
result = list_worktrees(args)

assert isinstance(result, Err)
assert isinstance(result.err(), NotBareRepoErr)


class TestListWorktreesEmpty:
def test_bare_repo_with_no_worktrees_returns_empty_list(self, tmp_path: Path) -> None:
bare = tmp_path / "bare"
bare.mkdir()
_make_bare_repo(bare)

args = ListArgs(current_working_dir=str(bare))
result = list_worktrees(args)

assert isinstance(result, Ok)
assert result.ok() == []


class TestListWorktreesSingle:
def test_single_worktree_name_and_path(self, tmp_path: Path) -> None:
bare = tmp_path / "bare"
bare.mkdir()
repo = _make_bare_repo(bare)

wt_path = tmp_path / "feat-x"
_add_branch(repo, "feat-x", wt_path)

args = ListArgs(current_working_dir=str(bare))
result = list_worktrees(args)

assert isinstance(result, Ok)
worktrees = result.ok()
assert len(worktrees) == 1
assert worktrees[0].name == "feat-x"
assert worktrees[0].path.rstrip("/") == str(wt_path)

def test_worktree_not_prunable_when_path_exists(self, tmp_path: Path) -> None:
bare = tmp_path / "bare"
bare.mkdir()
repo = _make_bare_repo(bare)
_add_branch(repo, "feat-x", tmp_path / "feat-x")

args = ListArgs(current_working_dir=str(bare))
result = list_worktrees(args)

assert isinstance(result, Ok)
assert result.ok()[0].is_prunable is False


class TestListWorktreesMultiple:
def test_multiple_worktrees_all_listed(self, tmp_path: Path) -> None:
bare = tmp_path / "bare"
bare.mkdir()
repo = _make_bare_repo(bare)

_add_branch(repo, "feat-a", tmp_path / "feat-a")
_add_branch(repo, "feat-b", tmp_path / "feat-b")
_add_branch(repo, "feat-c", tmp_path / "feat-c")

args = ListArgs(current_working_dir=str(bare))
result = list_worktrees(args)

assert isinstance(result, Ok)
names = {wt.name for wt in result.ok()}
assert names == {"feat-a", "feat-b", "feat-c"}


class TestListWorktreesUnmerged:
def test_worktree_on_same_commit_as_main_has_no_unmerged(self, tmp_path: Path) -> None:
bare = tmp_path / "bare"
bare.mkdir()
repo = _make_bare_repo(bare)
_add_branch(repo, "my-branch", tmp_path / "my-branch")

args = ListArgs(current_working_dir=str(bare))
result = list_worktrees(args)

assert isinstance(result, Ok)
assert result.ok()[0].has_unmerged_commits is False

def test_worktree_with_extra_commit_has_unmerged(self, tmp_path: Path) -> None:
bare = tmp_path / "bare"
bare.mkdir()
repo = _make_bare_repo(bare)
_add_branch(repo, "feat-x", tmp_path / "feat-x")

sig = pg.Signature("Test", "test@test.com")
blob_id = repo.create_blob(b"extra")
tb = repo.TreeBuilder()
tb.insert("extra.txt", blob_id, pg.GIT_FILEMODE_BLOB)
tree_id = tb.write()
parent: pg.Commit = repo.lookup_reference("refs/heads/feat-x").peel(pg.Commit)
repo.create_commit("refs/heads/feat-x", sig, sig, "extra commit", tree_id, [parent.id])

args = ListArgs(current_working_dir=str(bare))
result = list_worktrees(args)

assert isinstance(result, Ok)
assert result.ok()[0].has_unmerged_commits is True
Loading