434 lines
15 KiB
Python
434 lines
15 KiB
Python
#!/usr/bin/env python3
|
|
"""Small, conservative Git helper for the Labyricorn management launcher."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, replace
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import subprocess
|
|
from urllib.parse import urlsplit, urlunsplit
|
|
|
|
|
|
SYNCED = "synchronized"
|
|
REMOTE_AHEAD = "remote-ahead"
|
|
LOCAL_AHEAD = "local-ahead"
|
|
DIVERGED = "diverged"
|
|
UNAVAILABLE = "unavailable"
|
|
|
|
|
|
class GitRepositoryError(Exception):
|
|
"""An expected Git problem that can be displayed directly to a user."""
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class RepositoryState:
|
|
root: Path
|
|
repository_name: str
|
|
branch: str
|
|
origin_url: str | None
|
|
upstream: str | None
|
|
upstream_branch: str | None
|
|
dirty_count: int
|
|
ahead: int | None
|
|
behind: int | None
|
|
sync_state: str
|
|
problem: str | None = None
|
|
origin_push_url: str | None = None
|
|
|
|
@property
|
|
def status_text(self) -> str:
|
|
if self.sync_state == SYNCED:
|
|
return "Up to date"
|
|
if self.sync_state == REMOTE_AHEAD:
|
|
noun = "commit" if self.behind == 1 else "commits"
|
|
return f"Remote has {self.behind} newer {noun}"
|
|
if self.sync_state == LOCAL_AHEAD:
|
|
noun = "commit" if self.ahead == 1 else "commits"
|
|
return f"Local has {self.ahead} unpushed {noun}"
|
|
if self.sync_state == DIVERGED:
|
|
return "Local and remote have diverged"
|
|
return "Unavailable"
|
|
|
|
@property
|
|
def working_tree_text(self) -> str:
|
|
if not self.dirty_count:
|
|
return "Clean"
|
|
noun = "path" if self.dirty_count == 1 else "paths"
|
|
return f"{self.dirty_count} changed {noun}"
|
|
|
|
@property
|
|
def can_pull(self) -> bool:
|
|
return self.sync_state == REMOTE_AHEAD and self.dirty_count == 0
|
|
|
|
@property
|
|
def can_push(self) -> bool:
|
|
return self.sync_state == LOCAL_AHEAD
|
|
|
|
|
|
class GitRepository:
|
|
"""Narrow wrapper around the installed Git CLI for one working tree."""
|
|
|
|
def __init__(self, root: Path):
|
|
self.root = root.resolve()
|
|
|
|
@classmethod
|
|
def locate(cls, start: Path) -> "GitRepository":
|
|
start = start.resolve()
|
|
directory = start if start.is_dir() else start.parent
|
|
result = _run_git(directory, "rev-parse", "--show-toplevel", check=False)
|
|
if result.returncode != 0:
|
|
detail = _error_detail(result)
|
|
raise GitRepositoryError(
|
|
f"The launcher is not inside a Git repository. {detail}"
|
|
)
|
|
root = Path(result.stdout.strip()).resolve()
|
|
try:
|
|
start.relative_to(root)
|
|
except ValueError as exc:
|
|
raise GitRepositoryError(
|
|
"Git returned a repository that does not contain the launcher."
|
|
) from exc
|
|
return cls(root)
|
|
|
|
def refresh(self) -> RepositoryState:
|
|
"""Read local state, safely fetch origin metadata, and compare commits."""
|
|
before_fetch = self._read_state(compare_remote=False)
|
|
if before_fetch.problem:
|
|
return before_fetch
|
|
result = self._git(
|
|
"fetch", "--no-tags", "--prune", "origin", check=False, timeout=60
|
|
)
|
|
if result.returncode != 0:
|
|
return replace(
|
|
before_fetch,
|
|
sync_state=UNAVAILABLE,
|
|
ahead=None,
|
|
behind=None,
|
|
problem=f"Fetch failed: {self._safe_detail(result)}",
|
|
)
|
|
return self._read_state(compare_remote=True)
|
|
|
|
def pull(self, expected: RepositoryState) -> RepositoryState:
|
|
current = self.refresh()
|
|
self._verify_action_state(expected, current)
|
|
if current.dirty_count:
|
|
raise GitRepositoryError(
|
|
"Pull is disabled because the working tree has uncommitted changes. "
|
|
"Review or commit them before pulling."
|
|
)
|
|
if current.sync_state != REMOTE_AHEAD:
|
|
raise GitRepositoryError(self._not_actionable_message(current, "pull"))
|
|
assert current.upstream_branch is not None
|
|
result = self._git(
|
|
"pull",
|
|
"--ff-only",
|
|
"origin",
|
|
current.upstream_branch,
|
|
check=False,
|
|
timeout=60,
|
|
)
|
|
if result.returncode != 0:
|
|
raise GitRepositoryError(f"Pull failed: {self._safe_detail(result)}")
|
|
return self.refresh()
|
|
|
|
def push(self, expected: RepositoryState) -> RepositoryState:
|
|
current = self.refresh()
|
|
self._verify_action_state(expected, current)
|
|
if current.sync_state != LOCAL_AHEAD:
|
|
raise GitRepositoryError(self._not_actionable_message(current, "push"))
|
|
assert current.upstream_branch is not None
|
|
result = self._git(
|
|
"push",
|
|
"origin",
|
|
f"HEAD:refs/heads/{current.upstream_branch}",
|
|
check=False,
|
|
timeout=60,
|
|
)
|
|
if result.returncode != 0:
|
|
refreshed = self.refresh()
|
|
if refreshed.sync_state == DIVERGED:
|
|
raise GitRepositoryError(
|
|
"Push failed because the remote changed and the histories now "
|
|
"diverge. Resolve this manually."
|
|
)
|
|
raise GitRepositoryError(f"Push failed: {self._safe_detail(result)}")
|
|
return self.refresh()
|
|
|
|
def _read_state(self, *, compare_remote: bool) -> RepositoryState:
|
|
self._verify_root()
|
|
branch_result = self._git(
|
|
"symbolic-ref", "--quiet", "--short", "HEAD", check=False
|
|
)
|
|
branch = branch_result.stdout.strip()
|
|
detached = branch_result.returncode != 0 or not branch
|
|
if detached:
|
|
branch = "(detached HEAD)"
|
|
|
|
origin_result = self._git("remote", "get-url", "origin", check=False)
|
|
raw_origin = origin_result.stdout.strip() if origin_result.returncode == 0 else ""
|
|
origin_url = _display_url(raw_origin) if raw_origin else None
|
|
push_result = self._git(
|
|
"remote", "get-url", "--push", "origin", check=False
|
|
)
|
|
raw_push_url = push_result.stdout.strip() if push_result.returncode == 0 else ""
|
|
origin_push_url = _display_url(raw_push_url) if raw_push_url else origin_url
|
|
repository_name = _repository_name(raw_origin) if raw_origin else self.root.name
|
|
|
|
status_result = self._git(
|
|
"status", "--porcelain=v1", "--untracked-files=normal", check=True
|
|
)
|
|
dirty_count = len(status_result.stdout.splitlines())
|
|
|
|
if detached:
|
|
return RepositoryState(
|
|
self.root,
|
|
repository_name,
|
|
branch,
|
|
origin_url,
|
|
None,
|
|
None,
|
|
dirty_count,
|
|
None,
|
|
None,
|
|
UNAVAILABLE,
|
|
"The checkout has a detached HEAD. Switch branches manually before syncing.",
|
|
origin_push_url,
|
|
)
|
|
if not raw_origin:
|
|
return RepositoryState(
|
|
self.root,
|
|
repository_name,
|
|
branch,
|
|
None,
|
|
None,
|
|
None,
|
|
dirty_count,
|
|
None,
|
|
None,
|
|
UNAVAILABLE,
|
|
"No usable origin remote is configured. Configure it manually before syncing.",
|
|
origin_push_url,
|
|
)
|
|
|
|
upstream_result = self._git(
|
|
"rev-parse",
|
|
"--abbrev-ref",
|
|
"--symbolic-full-name",
|
|
"@{upstream}",
|
|
check=False,
|
|
)
|
|
upstream = upstream_result.stdout.strip()
|
|
remote_result = self._git(
|
|
"config", "--get", f"branch.{branch}.remote", check=False
|
|
)
|
|
merge_result = self._git(
|
|
"config", "--get", f"branch.{branch}.merge", check=False
|
|
)
|
|
upstream_remote = remote_result.stdout.strip()
|
|
merge_ref = merge_result.stdout.strip()
|
|
if (
|
|
not upstream
|
|
or upstream_remote != "origin"
|
|
or not merge_ref.startswith("refs/heads/")
|
|
):
|
|
return RepositoryState(
|
|
self.root,
|
|
repository_name,
|
|
branch,
|
|
origin_url,
|
|
upstream or None,
|
|
None,
|
|
dirty_count,
|
|
None,
|
|
None,
|
|
UNAVAILABLE,
|
|
"The current branch has no usable upstream on origin. Configure its "
|
|
"tracking branch manually before syncing.",
|
|
origin_push_url,
|
|
)
|
|
upstream_branch = merge_ref.removeprefix("refs/heads/")
|
|
expected_upstream = f"origin/{upstream_branch}"
|
|
if upstream != expected_upstream:
|
|
return RepositoryState(
|
|
self.root,
|
|
repository_name,
|
|
branch,
|
|
origin_url,
|
|
upstream,
|
|
upstream_branch,
|
|
dirty_count,
|
|
None,
|
|
None,
|
|
UNAVAILABLE,
|
|
f"The upstream relationship is ambiguous ({upstream}). Resolve it manually.",
|
|
origin_push_url,
|
|
)
|
|
if not compare_remote:
|
|
return RepositoryState(
|
|
self.root,
|
|
repository_name,
|
|
branch,
|
|
origin_url,
|
|
upstream,
|
|
upstream_branch,
|
|
dirty_count,
|
|
None,
|
|
None,
|
|
UNAVAILABLE,
|
|
origin_push_url=origin_push_url,
|
|
)
|
|
|
|
counts_result = self._git(
|
|
"rev-list", "--left-right", "--count", f"HEAD...{upstream}", check=False
|
|
)
|
|
if counts_result.returncode != 0:
|
|
return RepositoryState(
|
|
self.root,
|
|
repository_name,
|
|
branch,
|
|
origin_url,
|
|
upstream,
|
|
upstream_branch,
|
|
dirty_count,
|
|
None,
|
|
None,
|
|
UNAVAILABLE,
|
|
f"Cannot compare local and remote commits: {self._safe_detail(counts_result)}",
|
|
origin_push_url,
|
|
)
|
|
try:
|
|
ahead, behind = (int(value) for value in counts_result.stdout.split())
|
|
except (TypeError, ValueError) as exc:
|
|
raise GitRepositoryError("Git returned an invalid ahead/behind result.") from exc
|
|
|
|
if ahead and behind:
|
|
sync_state = DIVERGED
|
|
elif ahead:
|
|
sync_state = LOCAL_AHEAD
|
|
elif behind:
|
|
sync_state = REMOTE_AHEAD
|
|
else:
|
|
sync_state = SYNCED
|
|
return RepositoryState(
|
|
self.root,
|
|
repository_name,
|
|
branch,
|
|
origin_url,
|
|
upstream,
|
|
upstream_branch,
|
|
dirty_count,
|
|
ahead,
|
|
behind,
|
|
sync_state,
|
|
origin_push_url=origin_push_url,
|
|
)
|
|
|
|
def _verify_root(self) -> None:
|
|
result = self._git("rev-parse", "--show-toplevel", check=False)
|
|
if result.returncode != 0 or Path(result.stdout.strip()).resolve() != self.root:
|
|
raise GitRepositoryError("The repository containing the launcher has changed.")
|
|
|
|
def _verify_action_state(
|
|
self, expected: RepositoryState, current: RepositoryState
|
|
) -> None:
|
|
if current.root != expected.root:
|
|
raise GitRepositoryError("The repository changed; refresh before syncing.")
|
|
if current.branch != expected.branch:
|
|
raise GitRepositoryError(
|
|
f"The current branch changed from {expected.branch} to {current.branch}. "
|
|
"Review the refreshed status before syncing."
|
|
)
|
|
if (
|
|
current.origin_url != expected.origin_url
|
|
or current.origin_push_url != expected.origin_push_url
|
|
or current.upstream != expected.upstream
|
|
):
|
|
raise GitRepositoryError(
|
|
"The remote or upstream configuration changed. Review the refreshed "
|
|
"status before syncing."
|
|
)
|
|
if current.problem:
|
|
raise GitRepositoryError(current.problem)
|
|
|
|
@staticmethod
|
|
def _not_actionable_message(state: RepositoryState, action: str) -> str:
|
|
if state.sync_state == DIVERGED:
|
|
return (
|
|
"Local and remote histories have diverged. The launcher will not choose "
|
|
f"a {action} strategy; resolve the histories manually."
|
|
)
|
|
if state.sync_state == SYNCED:
|
|
return "The repository is already up to date."
|
|
return f"A safe {action} is not available for the current repository state."
|
|
|
|
def _git(
|
|
self,
|
|
*args: str,
|
|
check: bool,
|
|
timeout: int = 15,
|
|
) -> subprocess.CompletedProcess[str]:
|
|
result = _run_git(self.root, *args, check=False, timeout=timeout)
|
|
if check and result.returncode != 0:
|
|
raise GitRepositoryError(f"Git failed: {self._safe_detail(result)}")
|
|
return result
|
|
|
|
def _safe_detail(self, result: subprocess.CompletedProcess[str]) -> str:
|
|
return re.sub(r"(https?://)[^/\s@]+@", r"\1", _error_detail(result))
|
|
|
|
|
|
def _run_git(
|
|
cwd: Path,
|
|
*args: str,
|
|
check: bool,
|
|
timeout: int = 15,
|
|
) -> subprocess.CompletedProcess[str]:
|
|
command = ["git", *args]
|
|
options: dict[str, object] = {
|
|
"cwd": cwd,
|
|
"capture_output": True,
|
|
"text": True,
|
|
"encoding": "utf-8",
|
|
"errors": "replace",
|
|
"check": False,
|
|
"timeout": timeout,
|
|
"env": {**os.environ, "GIT_TERMINAL_PROMPT": "0"},
|
|
}
|
|
if os.name == "nt":
|
|
options["creationflags"] = subprocess.CREATE_NO_WINDOW
|
|
try:
|
|
result = subprocess.run(command, **options)
|
|
except FileNotFoundError as exc:
|
|
raise GitRepositoryError(
|
|
"Git is not installed or is not available on PATH."
|
|
) from exc
|
|
except subprocess.TimeoutExpired as exc:
|
|
raise GitRepositoryError("Git did not finish within the safety timeout.") from exc
|
|
if check and result.returncode != 0:
|
|
raise GitRepositoryError(f"Git failed: {_error_detail(result)}")
|
|
return result
|
|
|
|
|
|
def _error_detail(result: subprocess.CompletedProcess[str]) -> str:
|
|
return (result.stderr.strip() or result.stdout.strip() or "unknown Git error").replace(
|
|
"\n", " "
|
|
)
|
|
|
|
|
|
def _display_url(value: str) -> str:
|
|
"""Remove HTTP credentials before a remote URL reaches the GUI."""
|
|
parsed = urlsplit(value)
|
|
if parsed.scheme in {"http", "https"} and parsed.hostname:
|
|
host = parsed.hostname
|
|
if parsed.port:
|
|
host = f"{host}:{parsed.port}"
|
|
return urlunsplit((parsed.scheme, host, parsed.path, parsed.query, parsed.fragment))
|
|
return value
|
|
|
|
|
|
def _repository_name(value: str) -> str:
|
|
path = urlsplit(value).path if "://" in value else re.split(r"[\\/:]", value)[-1]
|
|
name = path.rstrip("/").rsplit("/", 1)[-1]
|
|
return name.removesuffix(".git") or "repository"
|