Coverage for src / mafw / devtools / git.py: 99%

1# Copyright 2026 European Union 

2# Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu) 

3# SPDX-License-Identifier: EUPL-1.2 

4""" 

5Git operations for MAFw development tools. 

6 

7This module centralizes all functions that shell out to ``git`` commands, 

8used by the release workflow, documentation builder, and dependency management. 

9""" 

10 

11from __future__ import annotations 

12 

13import re 

14import subprocess 

15from pathlib import Path 

16from typing import Any, List, Tuple 

17 

18from mafw.devtools import DevtoolsError, ensure_devtools_available 

19 

20ensure_devtools_available() 

21 

22from packaging.version import InvalidVersion, Version # noqa: E402 

23 

24from mafw.tools.shell_tools import CONSOLE, run_stdout # noqa: E402 

25from mafw.tools.shell_tools import run as cmd # noqa: E402 

26 

27PYPROJECT_FILE = Path('pyproject.toml') 

28"""Path to the TOML file containing the project dependencies.""" 

29 

30STABLE_TAG_PATTERN = re.compile(r'^v(?P<major>\d+)\.(?P<minor>\d+)\.(?P<micro>\d+)$') 

31"""Regular expression used to identify stable git tags in the form ``vX.Y.Z``.""" 

32 

33 

34def check_main_branch() -> None: 

35 """Ensure the release process is executed from the ``main`` branch. 

36 

37 :raises DevtoolsError: If the current branch is not ``main``. 

38 """ 

39 branch = run_stdout('git rev-parse --abbrev-ref HEAD') 

40 if branch != 'main': 

41 raise DevtoolsError('Must be on main branch.') 

42 

43 

44def ensure_clean_git() -> None: 

45 """Ensure the git working tree is clean (excluding untracked files). 

46 

47 :raises DevtoolsError: If tracked changes are present. 

48 """ 

49 status = run_stdout('git status --porcelain -uno') 

50 if status: 

51 raise DevtoolsError('Git working tree is not clean. Commit or stash changes first.') 

52 

53 

54def get_last_stable_tag() -> str | None: 

55 """Return the latest stable tag matching ``vX.Y.Z``. 

56 

57 :return: Latest stable tag or ``None`` when no stable tags are present. 

58 :rtype: str | None 

59 """ 

60 versions = get_git_tags() 

61 # Filter to only strict vX.Y.Z tags (no pre/post/dev — already handled by get_git_tags) 

62 stable: list[tuple[Version, str]] = [] 

63 for v, tag in versions: 

64 if STABLE_TAG_PATTERN.fullmatch(tag): 64 ↛ 63line 64 didn't jump to line 63 because the condition on line 64 was always true

65 stable.append((v, tag)) 

66 if not stable: 

67 return None 

68 return stable[-1][1] 

69 

70 

71def prevent_duplicate_tag(version: str) -> None: 

72 """Ensure the target release tag does not already exist. 

73 

74 :param version: Target release version. 

75 :type version: str 

76 :raises DevtoolsError: If ``v<version>`` already exists. 

77 """ 

78 existing_tags = run_stdout('git tag').splitlines() 

79 if f'v{version}' in existing_tags: 

80 raise DevtoolsError(f'Tag v{version} already exists.') 

81 

82 

83def commit_changes(version: str, dry_run: bool, *, include_changelog: bool = True) -> None: 

84 """Commit tracked release artifacts for the target version. 

85 

86 :param version: Target release version. 

87 :type version: str 

88 :param dry_run: Whether command execution is disabled. 

89 :type dry_run: bool 

90 :param include_changelog: Whether ``CHANGELOG.md`` should be staged and 

91 committed as part of the release artifacts. 

92 :type include_changelog: bool 

93 """ 

94 from mafw.devtools.documentation.requirements import REQUIREMENTS_GROUPS 

95 from mafw.devtools.release.changelog import CHANGELOG_FILE 

96 from mafw.devtools.release.versioning import ABOUT_FILE, NOTICE_FILE 

97 

98 CONSOLE.print('Committing tracked release artifacts...') 

99 tracked_files: list[str] = [str(ABOUT_FILE), str(NOTICE_FILE), str(PYPROJECT_FILE), 'README.rst'] 

100 for group in REQUIREMENTS_GROUPS: 

101 tracked_files.append(f'docs/source/requirements/{group}_requirements.rst') 

102 

103 if include_changelog: 

104 tracked_files.append(str(CHANGELOG_FILE)) 

105 cmd(['git', 'add', *tracked_files], dry_run=dry_run) 

106 cmd(['git', 'commit', '-m', f'chore(release): v{version}'], dry_run=dry_run) 

107 

108 

109def create_tag(version: str, dry_run: bool) -> str: 

110 """Create the local git tag for the target version. 

111 

112 :param version: Target release version. 

113 :type version: str 

114 :param dry_run: Whether command execution is disabled. 

115 :type dry_run: bool 

116 :return: Created tag name. 

117 :rtype: str 

118 """ 

119 tag = f'v{version}' 

120 CONSOLE.print(f'Creating local tag {tag}...') 

121 cmd(['git', 'tag', tag], dry_run=dry_run) 

122 return tag 

123 

124 

125def push_changes(dry_run: bool) -> None: 

126 """Push commits and tags to the remote repository. 

127 

128 :param dry_run: Whether command execution is disabled. 

129 :type dry_run: bool 

130 """ 

131 CONSOLE.print('Pushing commits and tags...') 

132 cmd(['git', 'push', 'origin-ssh', 'main'], dry_run=dry_run) 

133 cmd(['git', 'push', '--tags'], dry_run=dry_run) 

134 

135 

136def commit_dependency_unfreeze(dry_run: bool) -> None: 

137 """Commit the unfreezing of dependency upper bounds to ``main``. 

138 

139 :param dry_run: Whether command execution is disabled. 

140 :type dry_run: bool 

141 """ 

142 from mafw.devtools.documentation.requirements import REQUIREMENTS_GROUPS 

143 

144 CONSOLE.print('Committing dependency unfreeze...') 

145 tracked_files: list[str] = [str(PYPROJECT_FILE), 'README.rst'] 

146 for group in REQUIREMENTS_GROUPS: 

147 tracked_files.append(f'docs/source/requirements/{group}_requirements.rst') 

148 

149 cmd(['git', 'add', *tracked_files], dry_run=dry_run) 

150 cmd(['git', 'commit', '-m', 'chore(dependencies): unfreeze upper bounds'], dry_run=dry_run) 

151 

152 

153def get_git_tags(min_version: str | None = None) -> List[Tuple[Version, Any]]: 

154 """Return list of (Version, tag) tuples sorted ascending. 

155 

156 :param min_version: Minimum version to consider, defaults to None 

157 :type min_version: str | None 

158 :return: List of (Version, tag) tuples 

159 :rtype: List[Tuple[Version, Any]] 

160 """ 

161 proc = cmd(['git', 'tag'], stdout=subprocess.PIPE, stderr=subprocess.STDOUT, check=False) 

162 tags = proc.stdout.split() 

163 versions = [] 

164 for t in tags: 

165 try: 

166 v = Version(t) 

167 if v.is_prerelease or v.is_devrelease: 

168 continue 

169 if min_version and v < Version(min_version): 

170 continue 

171 versions.append((v, t)) 

172 except InvalidVersion: 

173 continue 

174 versions.sort() 

175 return versions 

176 

177 

178def get_current_branch() -> str: 

179 """Get the name of the currently checked out branch. 

180 

181 :return: Name of the current branch 

182 :rtype: str 

183 """ 

184 proc = cmd( 

185 ['git', 'rev-parse', '--abbrev-ref', 'HEAD'], stdout=subprocess.PIPE, stderr=subprocess.STDOUT, check=False 

186 ) 

187 return str(proc.stdout.strip()) 

188 

189 

190def sort_tags_semver(tags: List[str]) -> List[str]: 

191 """Sort tags using semantic versioning comparison. 

192 

193 :param tags: List of tag strings to sort 

194 :type tags: List[str] 

195 :return: Sorted list of tag strings 

196 :rtype: List[str] 

197 """ 

198 from mafw.devtools.documentation.builder import parse_version_tuple 

199 

200 return sorted(tags, key=lambda t: parse_version_tuple(t)) 

201 

202 

203def git_rev_of(ref: str) -> str: 

204 """Get the git revision hash for a given reference. 

205 

206 :param ref: Git reference (tag, branch, commit hash) 

207 :type ref: str 

208 :return: Git revision hash 

209 :rtype: str 

210 :raises RuntimeError: If git rev-list fails 

211 """ 

212 proc = cmd(['git', 'rev-list', '-n', '1', ref], stdout=subprocess.PIPE, stderr=subprocess.STDOUT, check=False) 

213 if proc.returncode != 0: 

214 raise RuntimeError(f'git rev-list failed for {ref}:\n{proc.stdout}') 

215 return str(proc.stdout.strip()) 

216 

217 

218def is_ancestor(a: str, b: str) -> bool: 

219 """Return True if commit a is ancestor of commit b (git merge-base --is-ancestor). 

220 

221 :param a: First commit reference 

222 :type a: str 

223 :param b: Second commit reference 

224 :type b: str 

225 :return: True if a is ancestor of b 

226 :rtype: bool 

227 """ 

228 proc = cmd( 

229 ['git', 'merge-base', '--is-ancestor', a, b], stdout=subprocess.PIPE, stderr=subprocess.STDOUT, check=False 

230 ) 

231 return proc.returncode == 0