Add python_git: a leak-safe GitPython alternative
Every git invocation runs through a single blocking subprocess.run call in python_git/runner.py that always waits for the process to exit and closes its pipes before returning, so there is no long-lived Popen handle for a caller to forget to close (the root cause of GitPython's known fd/process leaks). Repo exposes status/log/diff/add/commit/push/pull/fetch, branch and remote management, and clone, backed by dataclasses in models.py. Covered by tests/test_repo.py, including an fd-count regression test. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
commit
ebe7ec6f4b
15 files changed
+527
No files matched your search
@@ -0,0 +1,119 @@
|
||||
import os
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from python_git import GitCommandError, InvalidGitRepositoryError, Repo
|
||||
|
||||
|
||||
def _init_repo(path: Path) -> Repo:
|
||||
subprocess.run(["git", "init", "-q", str(path)], check=True)
|
||||
subprocess.run(["git", "-C", str(path), "config", "user.email", "test@example.com"], check=True)
|
||||
subprocess.run(["git", "-C", str(path), "config", "user.name", "Test User"], check=True)
|
||||
return Repo(path)
|
||||
|
||||
|
||||
def test_invalid_repository_raises(tmp_path):
|
||||
with pytest.raises(InvalidGitRepositoryError):
|
||||
Repo(tmp_path)
|
||||
|
||||
|
||||
def test_missing_path_raises(tmp_path):
|
||||
with pytest.raises(InvalidGitRepositoryError):
|
||||
Repo(tmp_path / "does-not-exist")
|
||||
|
||||
|
||||
def test_status_reports_untracked_and_spaced_filenames(tmp_path):
|
||||
repo = _init_repo(tmp_path)
|
||||
(tmp_path / "a file.txt").write_text("hello\n")
|
||||
|
||||
statuses = repo.status()
|
||||
|
||||
assert len(statuses) == 1
|
||||
assert statuses[0].path == "a file.txt"
|
||||
assert statuses[0].index_status == "?"
|
||||
|
||||
|
||||
def test_add_commit_log_roundtrip(tmp_path):
|
||||
repo = _init_repo(tmp_path)
|
||||
(tmp_path / "file.txt").write_text("hello\n")
|
||||
repo.add("file.txt")
|
||||
repo.commit("initial commit")
|
||||
|
||||
commits = repo.log()
|
||||
|
||||
assert len(commits) == 1
|
||||
assert commits[0].message == "initial commit"
|
||||
assert commits[0].author_email == "test@example.com"
|
||||
assert not repo.is_dirty()
|
||||
|
||||
|
||||
def test_log_on_empty_repo_returns_empty_list(tmp_path):
|
||||
repo = _init_repo(tmp_path)
|
||||
assert repo.log() == []
|
||||
|
||||
|
||||
def test_diff_shows_modification(tmp_path):
|
||||
repo = _init_repo(tmp_path)
|
||||
(tmp_path / "file.txt").write_text("hello\n")
|
||||
repo.add("file.txt")
|
||||
repo.commit("initial")
|
||||
(tmp_path / "file.txt").write_text("goodbye\n")
|
||||
|
||||
assert "goodbye" in repo.diff()
|
||||
|
||||
|
||||
def test_rename_is_tracked(tmp_path):
|
||||
repo = _init_repo(tmp_path)
|
||||
content = "hello world this is long enough to be detected as a rename\n"
|
||||
(tmp_path / "old.txt").write_text(content)
|
||||
repo.add("old.txt")
|
||||
repo.commit("add old")
|
||||
(tmp_path / "old.txt").rename(tmp_path / "new.txt")
|
||||
repo.add("new.txt", "old.txt")
|
||||
|
||||
renamed = [s for s in repo.status() if s.orig_path]
|
||||
|
||||
assert len(renamed) == 1
|
||||
assert renamed[0].path == "new.txt"
|
||||
assert renamed[0].orig_path == "old.txt"
|
||||
|
||||
|
||||
def test_branch_create_and_checkout(tmp_path):
|
||||
repo = _init_repo(tmp_path)
|
||||
(tmp_path / "file.txt").write_text("hello\n")
|
||||
repo.add("file.txt")
|
||||
repo.commit("initial")
|
||||
|
||||
repo.create_branch("feature")
|
||||
repo.checkout("feature")
|
||||
|
||||
assert repo.current_branch() == "feature"
|
||||
branches = {b.name: b.is_current for b in repo.list_branches()}
|
||||
assert branches["feature"] is True
|
||||
|
||||
|
||||
def test_git_command_error_on_nonexistent_branch(tmp_path):
|
||||
repo = _init_repo(tmp_path)
|
||||
with pytest.raises(GitCommandError):
|
||||
repo.checkout("does-not-exist")
|
||||
|
||||
|
||||
def test_no_fd_leak_across_many_calls(tmp_path):
|
||||
repo = _init_repo(tmp_path)
|
||||
(tmp_path / "file.txt").write_text("hello\n")
|
||||
repo.add("file.txt")
|
||||
repo.commit("initial")
|
||||
|
||||
fd_dir = Path("/proc/self/fd")
|
||||
if not fd_dir.exists():
|
||||
pytest.skip("requires /proc/self/fd (Linux)")
|
||||
|
||||
before = len(os.listdir(fd_dir))
|
||||
for _ in range(200):
|
||||
repo.status()
|
||||
repo.log()
|
||||
after = len(os.listdir(fd_dir))
|
||||
|
||||
assert after <= before
|
||||
Reference in new issue
Block a user