Coverage for src / rhiza / models / _git / context.py: 100%

1"""The public :class:`GitContext` facade composing the git engine mixins.""" 

2 

3import os 

4import subprocess # nosec B404 

5from dataclasses import dataclass, field 

6from pathlib import Path 

7 

8from loguru import logger 

9 

10from rhiza.models._git.helpers import _log_git_stderr_errors, get_git_executable 

11from rhiza.models._git.merge import MergeMixin 

12 

13 

14@dataclass 

15class GitContext(MergeMixin): 

16 """Bundles the git executable path and environment for subprocess calls. 

17 

18 All git-invoking functions in the sync helpers accept a 

19 :class:`GitContext` instead of resolving the executable on their own, 

20 making them easily testable via binary injection. 

21 

22 The git operations are organised into focused mixins 

23 (:class:`~rhiza.models._git.remote.RemoteOpsMixin`, 

24 :class:`~rhiza.models._git.diff.DiffMixin`, 

25 :class:`~rhiza.models._git.merge.MergeMixin`); this class composes them and 

26 adds the working-tree/branch operations. See ADR-0005 for the rationale. 

27 

28 Attributes: 

29 executable: Absolute path to the git binary. 

30 env: Environment variables passed to every git subprocess. 

31 """ 

32 

33 executable: str 

34 env: dict[str, str] = field(default_factory=dict) 

35 

36 @classmethod 

37 def default(cls) -> "GitContext": 

38 """Create a GitContext using the system git and process environment. 

39 

40 Returns: 

41 A :class:`GitContext` populated with the real git executable path 

42 and a copy of the current process environment with 

43 ``GIT_TERMINAL_PROMPT`` set to ``"0"``. 

44 """ 

45 env = os.environ.copy() 

46 env["GIT_TERMINAL_PROMPT"] = "0" 

47 return cls(executable=get_git_executable(), env=env) 

48 

49 def assert_status_clean(self, target: Path) -> None: 

50 """Raise RuntimeError if the target repository has uncommitted changes. 

51 

52 Runs ``git status --porcelain`` and raises if the output is non-empty, 

53 preventing a sync from running on a dirty working tree. 

54 

55 Args: 

56 target: Path to the target repository. 

57 

58 Raises: 

59 RuntimeError: If the working tree has uncommitted changes. 

60 """ 

61 result = subprocess.run( # nosec B603 # noqa: S603 

62 [self.executable, "status", "--porcelain"], 

63 cwd=target, 

64 capture_output=True, 

65 text=True, 

66 env=self.env, 

67 ) 

68 if result.stdout.strip(): 

69 logger.error("Working tree is not clean. Please commit or stash your changes before syncing.") 

70 logger.error("Uncommitted changes:") 

71 for line in result.stdout.strip().splitlines(): 

72 logger.error(f" {line}") 

73 raise RuntimeError("Working tree is not clean. Please commit or stash your changes before syncing.") # noqa: TRY003 

74 

75 def handle_target_branch(self, target: Path, target_branch: str | None) -> None: 

76 """Handle target branch creation or checkout if specified. 

77 

78 Args: 

79 target: Path to the target repository. 

80 target_branch: Optional branch name to create/checkout. 

81 """ 

82 if not target_branch: 

83 return 

84 

85 logger.info(f"Creating/checking out target branch: {target_branch}") 

86 try: 

87 result = subprocess.run( # nosec B603 # noqa: S603 

88 [self.executable, "rev-parse", "--verify", target_branch], 

89 cwd=target, 

90 capture_output=True, 

91 text=True, 

92 env=self.env, 

93 ) 

94 

95 if result.returncode == 0: 

96 logger.info(f"Branch '{target_branch}' exists, checking out...") 

97 subprocess.run( # nosec B603 # noqa: S603 

98 [self.executable, "checkout", target_branch], 

99 cwd=target, 

100 check=True, 

101 capture_output=True, 

102 text=True, 

103 env=self.env, 

104 ) 

105 else: 

106 logger.info(f"Creating new branch '{target_branch}'...") 

107 subprocess.run( # nosec B603 # noqa: S603 

108 [self.executable, "checkout", "-b", target_branch], 

109 cwd=target, 

110 check=True, 

111 capture_output=True, 

112 text=True, 

113 env=self.env, 

114 ) 

115 except subprocess.CalledProcessError as e: 

116 logger.error(f"Failed to create/checkout branch '{target_branch}'") 

117 _log_git_stderr_errors(e.stderr) 

118 logger.error("Please ensure you have no uncommitted changes or conflicts") 

119 raise