git_shim.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/registry/round/git_shim.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/test_round_*.py"
"""
Mockable git shell-out seam for round-close orchestration.
 
Purpose
-------
Wrap the handful of ``git`` invocations the round-close executors need
behind one injectable seam so tests never shell out to a real repo.
 
Responsibilities
----------------
- Run git subcommands in a fixed repository directory.
- Surface conflict state from ``git merge`` without raising.
- Expose ancestry and fast-forward helpers used by ff_main / consolidate.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - lifecycle events (branch create, merge, ff)
    L3 - command and exit-code details
    L4 - deep tracing
 
Contracts
---------
- No method mutates the registry database; git is the only side effect.
- The injected ``runner`` matches the ``subprocess.run`` signature.
- ``merge`` never raises on conflict; callers inspect ``MergeResult``.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- subprocess (subprocess.run - the default git runner)
- pathlib (the repository directory)
"""
from __future__ import annotations
 
from dataclasses import dataclass, field
from pathlib import Path
import subprocess
from typing import Callable, Sequence
 
from oc_diagnostics import _dbg
 
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
Runner = Callable[..., "subprocess.CompletedProcess[str]"]
 
 
@dataclass(frozen=True)
class GitResult:
    """Outcome of a single git invocation."""
 
    returncode: int
    stdout: str
    stderr: str
 
    @property
    def ok(self) -> bool:
        """True when the git command exited zero."""
 
        return self.returncode == 0
 
 
@dataclass(frozen=True)
class MergeResult:
    """Result of a ``git merge`` attempt."""
 
    ok: bool
    head_sha: str
    conflicted: bool = False
    conflicted_paths: tuple[str, ...] = field(default_factory=tuple)
    already_up_to_date: bool = False
 
 
class GitError(RuntimeError):
    """Raised when a git command that must succeed exits non-zero."""
 
    def __init__(self, args: Sequence[str], result: GitResult) -> None:
        self.args_run = tuple(args)
        self.result = result
        super().__init__(
            f"git {' '.join(args)} failed ({result.returncode}): "
            f"{result.stderr.strip() or result.stdout.strip()}"
        )
 
 
class GitShim:
    """Thin, injectable wrapper over the git CLI for round-close."""
 
    def __init__(self, repo_dir: Path, *, runner: Runner | None = None) -> None:
        self.repo_dir = Path(repo_dir)
        self._runner: Runner = runner if runner is not None else subprocess.run
 
    def run(self, *args: str) -> GitResult:
        """Run ``git <args>`` in the repo dir and capture the result."""
 
        func = "run"
        completed = self._runner(
            ["git", *args],
            cwd=str(self.repo_dir),
            capture_output=True,
            text=True,
            check=False,
        )
        result = GitResult(
            returncode=int(getattr(completed, "returncode", 0)),
            stdout=str(getattr(completed, "stdout", "") or ""),
            stderr=str(getattr(completed, "stderr", "") or ""),
        )
        _dbg("OC_CHAT.MAIN", func, f"git {' '.join(args)} -> {result.returncode}", 3)
        return result
 
    def _checked(self, *args: str) -> GitResult:
        func = "_checked"
        result = self.run(*args)
        if not result.ok:
            _dbg("OC_CHAT.MAIN", func, f"git {' '.join(args)} non-zero", 2)
            raise GitError(args, result)
        return result
 
    def rev_parse(self, ref: str = "HEAD") -> str:
        """Return the full SHA for ``ref``."""
 
        func = "rev_parse"
        result = self._checked("rev-parse", ref)
        _dbg("OC_CHAT.MAIN", func, f"resolved {ref}", 4)
        return result.stdout.strip()
 
    def current_branch(self) -> str:
        """Return the current branch name (or empty for detached HEAD)."""
 
        func = "current_branch"
        result = self.run("rev-parse", "--abbrev-ref", "HEAD")
        _dbg("OC_CHAT.MAIN", func, "queried branch", 4)
        return result.stdout.strip()
 
    def branch_exists(self, name: str) -> bool:
        """True when a local branch ``name`` exists."""
 
        func = "branch_exists"
        result = self.run("rev-parse", "--verify", "--quiet", f"refs/heads/{name}")
        _dbg("OC_CHAT.MAIN", func, f"exists={result.ok} {name}", 4)
        return result.ok
 
    def create_branch(self, name: str, start_point: str) -> str:
        """Create ``name`` at ``start_point`` and check it out; return HEAD."""
 
        func = "create_branch"
        if self.branch_exists(name):
            self._checked("checkout", name)
        else:
            self._checked("checkout", "-b", name, start_point)
        _dbg("OC_CHAT.MAIN", func, f"on {name}", 2)
        return self.rev_parse("HEAD")
 
    def checkout(self, ref: str) -> None:
        """Check out ``ref``."""
 
        func = "checkout"
        self._checked("checkout", ref)
        _dbg("OC_CHAT.MAIN", func, f"checked out {ref}", 3)
 
    def merge(
        self,
        branch: str,
        *,
        no_ff: bool = True,
        message: str | None = None,
    ) -> MergeResult:
        """Merge ``branch`` into the current branch; report conflicts."""
 
        func = "merge"
        args = ["merge"]
        if no_ff:
            args.append("--no-ff")
        if message is not None:
            args.extend(["-m", message])
        args.append(branch)
        result = self.run(*args)
        if result.ok:
            up_to_date = "Already up to date" in result.stdout
            _dbg("OC_CHAT.MAIN", func, f"merged {branch} up_to_date={up_to_date}", 2)
            return MergeResult(
                ok=True,
                head_sha=self.rev_parse("HEAD"),
                already_up_to_date=up_to_date,
            )
        conflicted_paths = self._conflicted_paths()
        _dbg(
            "OC_CHAT.MAIN",
            func,
            f"merge {branch} failed conflicts={len(conflicted_paths)}",
            2,
        )
        return MergeResult(
            ok=False,
            head_sha=self.rev_parse("HEAD"),
            conflicted=bool(conflicted_paths),
            conflicted_paths=conflicted_paths,
        )
 
    def _conflicted_paths(self) -> tuple[str, ...]:
        func = "_conflicted_paths"
        result = self.run("diff", "--name-only", "--diff-filter=U")
        paths = tuple(p for p in result.stdout.splitlines() if p.strip())
        _dbg("OC_CHAT.MAIN", func, f"{len(paths)} conflicted", 4)
        return paths
 
    def merge_abort(self) -> None:
        """Abort an in-progress merge."""
 
        func = "merge_abort"
        self.run("merge", "--abort")
        _dbg("OC_CHAT.MAIN", func, "merge aborted", 2)
 
    def checkout_path(self, side: str, path: str) -> None:
        """Resolve a single conflicted ``path`` by taking ``side``."""
 
        func = "checkout_path"
        flag = "--theirs" if side == "current" else "--ours"
        self._checked("checkout", flag, "--", path)
        self._checked("add", "--", path)
        _dbg("OC_CHAT.MAIN", func, f"took {side} for {path}", 3)
 
    def is_ancestor(self, maybe_ancestor: str, descendant: str) -> bool:
        """True when ``maybe_ancestor`` is an ancestor of ``descendant``."""
 
        func = "is_ancestor"
        result = self.run(
            "merge-base", "--is-ancestor", maybe_ancestor, descendant
        )
        _dbg("OC_CHAT.MAIN", func, f"ancestor={result.ok}", 4)
        return result.ok
 
    def merge_base(self, a: str, b: str) -> str:
        """Return the merge base SHA of two refs."""
 
        func = "merge_base"
        result = self._checked("merge-base", a, b)
        _dbg("OC_CHAT.MAIN", func, "computed merge-base", 4)
        return result.stdout.strip()
 
    def fast_forward(self, branch: str, to_sha: str) -> str:
        """Fast-forward ``branch`` to ``to_sha``; return the new HEAD."""
 
        func = "fast_forward"
        self._checked("checkout", branch)
        self._checked("merge", "--ff-only", to_sha)
        head = self.rev_parse("HEAD")
        _dbg("OC_CHAT.MAIN", func, f"{branch} -> {head}", 2)
        return head
 
    def branch_delete(self, name: str, *, force: bool = False) -> bool:
        """Delete a local branch; return True on success."""
 
        func = "branch_delete"
        flag = "-D" if force else "-d"
        result = self.run("branch", flag, name)
        _dbg("OC_CHAT.MAIN", func, f"delete {name} ok={result.ok}", 2)
        return result.ok
 
    def worktree_remove(self, path: str, *, force: bool = False) -> bool:
        """Remove a git worktree; return True on success."""
 
        func = "worktree_remove"
        args = ["worktree", "remove"]
        if force:
            args.append("--force")
        args.append(path)
        result = self.run(*args)
        _dbg("OC_CHAT.MAIN", func, f"remove worktree {path} ok={result.ok}", 2)
        return result.ok