9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157 | class GitProject:
def __init__(self, dir: Path):
self.dir = Path(dir)
@classmethod
def from_existing(cls, dir: Path) -> Self | None:
if not is_a_git_worktree(dir):
return None
return cls(dir)
@classmethod
@contextmanager
def clone_if_necessary(cls, repo_uri: str) -> Generator[Self, None, None]:
"""Construct an instance, cloning given repo first if necessary.
If ``repo_uri`` is remote, it will be cloned to a temporary directory
that this removed on exit of the context.
If ``repo_uri`` is a local path, it is not deleted on exit.
"""
if Path(repo_uri).exists():
yield cls(Path(repo_uri))
else:
with TemporaryDirectory() as dir:
dir_path = Path(dir)
clone_result = clone_to(repo_uri, dir_path)
clone_result.check_returncode()
yield cls(dir_path)
def _run_cmd(self, *args: Any, **kwargs: Any) -> subprocess.CompletedProcess[str]:
return run_text(*args, **kwargs, cwd=self.dir)
def has_merge_conflicts(self) -> bool:
result = self._run_cmd(
[
"git",
"-c",
"core.whitespace=-trailing-space,-space-before-tab,-indent-with-non-tab,-tab-in-indent,-cr-at-eol",
"diff",
"--check",
]
)
return result.returncode != 0
def is_clean(self) -> bool:
result = self._run_cmd(["git", "status", "--porcelain"])
return result.stdout.strip() == ""
def is_git(self) -> bool:
return is_a_git_worktree(self.dir)
def init(self) -> None:
self._run_cmd(["git", "init", "--initial-branch=main"])
def checkout(self, *args: str) -> subprocess.CompletedProcess[str]:
return self._run_cmd(["git", "checkout", *list(args)])
def add(self, *args: str) -> subprocess.CompletedProcess[str]:
return self._run_cmd(["git", "add", *list(args)])
def commit(self, msg: str) -> subprocess.CompletedProcess[str]:
return self._run_cmd(["git", "commit", "-m", msg])
def commit_all(self, msg: str) -> subprocess.CompletedProcess[str]:
result = self.add(".")
if result.returncode != 0:
return result
return self.commit(msg)
def log(self, *args: str) -> subprocess.CompletedProcess[str]:
return self._run_cmd(["git", "log", *list(args)])
def reset(self, *args: str) -> subprocess.CompletedProcess[str]:
return self._run_cmd(["git", "reset", *list(args)])
def stash(self) -> None:
self._run_cmd(["git", "stash"])
def pop(self) -> None:
self._run_cmd(["git", "stash", "pop"])
def tag(self, tag: str) -> None:
self._run_cmd(["git", "tag", tag])
def rename_branch(self, new_branch_name: str) -> None:
self._run_cmd(["git", "branch", "-m", new_branch_name])
def get_commit_hash_for_head(self) -> str | None:
result = self._run_cmd(["git", "rev-parse", "HEAD"])
# say you run this against an empty repo, you'll get a return code of
# 128 and message "fatal: ambiguous argument 'HEAD': unknown revision or
# path not in the working tree."
if result.returncode != 0:
return None
return result.stdout.strip()
def is_path_ignored(self, path: str) -> bool:
result = self._run_cmd(["git", "check-ignore", "-q", path])
if result.returncode not in (0, 1):
result.check_returncode()
return result.returncode == 0
def get_tracked_files(self) -> list[Path]:
tracked_files = [
Path(file) for file in self._run_cmd(["git", "ls-files"]).stdout.splitlines()
]
return tracked_files
def get_untracked_files(self) -> list[str]:
result = self._run_cmd(["git", "ls-files", "--exclude-standard", "--others"])
return result.stdout.splitlines()
def get_tags(self, *args: str) -> list[str]:
result = self._run_cmd(["git", "tag", *list(args)])
return result.stdout.splitlines()
def get_closest_tag(self, commit_hash: str) -> str | None:
result = self._run_cmd(
["git", "describe", "--exclude", commit_hash, "--contains", commit_hash]
)
if result.returncode != 0:
return None
first_tag = result.stdout.partition("~")[0]
return first_tag
def get_commit_description(self, commit_ish: str = "HEAD") -> str | None:
result = self._run_cmd(["git", "describe", "--tags", "--always", commit_ish])
if result.returncode != 0:
return None
return result.stdout.strip()
def get_commit_count(self, ref: str = "HEAD") -> int | None:
result = self._run_cmd(["git", "rev-list", "--count", ref])
if result.returncode != 0:
return None
return int(result.stdout.strip())
|