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

1"""Diff computation and parsing for the sync engine.""" 

2 

3import subprocess # nosec B404 

4from dataclasses import dataclass 

5from pathlib import Path 

6 

7from loguru import logger 

8 

9from rhiza.models._git._base import GitContextBase 

10 

11_SRC_PREFIX = "upstream-template-old/" 

12_DST_PREFIX = "upstream-template-new/" 

13 

14 

15def _path_after(line: str, marker: str, prefix: str) -> str | None: 

16 """Return the diff path on a ``---``/``+++`` header line, stripped of *prefix*.""" 

17 raw = line[len(marker) :].strip().strip('"').split("\t")[0] 

18 if raw != "/dev/null" and raw.startswith(prefix): 

19 return raw[len(prefix) :] 

20 return None 

21 

22 

23@dataclass 

24class _DiffFileState: 

25 """Accumulates the per-file flags/paths seen while scanning one ``diff --git`` block.""" 

26 

27 is_new: bool = False 

28 is_deleted: bool = False 

29 src_path: str | None = None 

30 dst_path: str | None = None 

31 started: bool = False 

32 

33 def reset(self) -> None: 

34 """Begin a new file block, clearing all accumulated state.""" 

35 self.is_new = False 

36 self.is_deleted = False 

37 self.src_path = None 

38 self.dst_path = None 

39 self.started = True 

40 

41 def update(self, line: str) -> None: 

42 """Update state from a single non-``diff --git`` header line.""" 

43 if line.startswith("new file mode"): 

44 self.is_new = True 

45 elif line.startswith("deleted file mode"): 

46 self.is_deleted = True 

47 elif line.startswith("--- "): 

48 self.src_path = _path_after(line, "--- ", _SRC_PREFIX) or self.src_path 

49 elif line.startswith("+++ "): 

50 self.dst_path = _path_after(line, "+++ ", _DST_PREFIX) or self.dst_path 

51 

52 def entry(self) -> tuple[str, bool, bool] | None: 

53 """Return the ``(rel_path, is_new, is_deleted)`` entry for this block, if a path was captured.""" 

54 rel = self.src_path if self.is_deleted else self.dst_path 

55 return (rel, self.is_new, self.is_deleted) if rel else None 

56 

57 

58class DiffMixin(GitContextBase): 

59 """Compute and parse diffs between template snapshot trees.""" 

60 

61 def get_diff(self, repo0: Path, repo1: Path) -> str: 

62 """Compute the raw diff between two directory trees using ``git diff --no-index``. 

63 

64 Args: 

65 repo0: Path to the base (old) directory tree. 

66 repo1: Path to the upstream (new) directory tree. 

67 """ 

68 repo0_str = repo0.resolve().as_posix() 

69 repo1_str = repo1.resolve().as_posix() 

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

71 [ 

72 self.executable, 

73 "-c", 

74 "diff.noprefix=", 

75 "diff", 

76 "--no-index", 

77 "--relative", 

78 "--binary", 

79 "--src-prefix=upstream-template-old/", 

80 "--dst-prefix=upstream-template-new/", 

81 "--no-ext-diff", 

82 "--no-color", 

83 repo0_str, 

84 repo1_str, 

85 ], 

86 cwd=repo0_str, 

87 capture_output=True, 

88 env=self.env, 

89 ) 

90 diff = result.stdout.decode() if isinstance(result.stdout, bytes) else (result.stdout or "") 

91 for repo in [repo0_str, repo1_str]: 

92 from re import sub 

93 

94 repo_nix = sub("/[a-z]:", "", repo) 

95 diff = diff.replace(f"upstream-template-old{repo_nix}", "upstream-template-old").replace( 

96 f"upstream-template-new{repo_nix}", "upstream-template-new" 

97 ) 

98 diff = diff.replace(repo0_str + "/", "").replace(repo1_str + "/", "") 

99 return diff 

100 

101 def sync_diff(self, target: Path, upstream_snapshot: Path) -> None: 

102 """Execute the diff (dry-run) strategy. 

103 

104 Shows what would change without modifying any files. 

105 

106 Args: 

107 target: Path to the target repository. 

108 upstream_snapshot: Path to the upstream snapshot directory. 

109 """ 

110 diff = self.get_diff(target, upstream_snapshot) 

111 if diff.strip(): 

112 logger.info(f"\n{diff}") 

113 changes = diff.count("diff --git") 

114 logger.info(f"{changes} file(s) would be changed") 

115 else: 

116 logger.success("No differences found") 

117 

118 def _parse_diff_filenames(self, diff: str) -> list[tuple[str, bool, bool]]: 

119 """Parse a unified diff produced by :func:`DiffMixin.get_diff` into file entries. 

120 

121 Each entry is ``(rel_path, is_new, is_deleted)`` where *rel_path* is the 

122 path relative to both snapshot directories. 

123 

124 Args: 

125 diff: Unified diff string from :func:`DiffMixin.get_diff`. 

126 

127 Returns: 

128 List of ``(rel_path, is_new, is_deleted)`` tuples, one per changed file. 

129 """ 

130 results: list[tuple[str, bool, bool]] = [] 

131 state = _DiffFileState() 

132 

133 def _flush() -> None: 

134 """Emit the current file entry into results if a path was captured.""" 

135 entry = state.entry() 

136 if entry: 

137 results.append(entry) 

138 

139 for line in diff.splitlines(): 

140 if line.startswith("diff --git "): 

141 if state.started: 

142 _flush() 

143 state.reset() 

144 else: 

145 state.update(line) 

146 

147 if state.started: 

148 _flush() 

149 

150 return results