Coverage for app/venv/lib/python3.14/site-packages/weblate/addons/git.py: 14%
155 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-07 07:15 +0000
1# Copyright © Michal Čihař <michal@weblate.org>
2#
3# SPDX-License-Identifier: GPL-3.0-or-later
4from __future__ import annotations
6from collections import defaultdict
7from itertools import chain
8from typing import TYPE_CHECKING, ClassVar, cast
10from django.utils.translation import gettext_lazy
12from weblate.addons.base import BaseAddon
13from weblate.addons.events import AddonEvent
14from weblate.addons.forms import GitSquashForm
15from weblate.utils.errors import report_error
16from weblate.vcs.base import RepositoryError
17from weblate.vcs.models import VCS_REGISTRY
19if TYPE_CHECKING: 19 ↛ 20line 19 didn't jump to line 20 because the condition on line 19 was never true
20 from weblate.addons.base import CompatDict
21 from weblate.trans.models import Component
22 from weblate.vcs.git import GitRepository
25class GitSquashAddon(BaseAddon):
26 name = "weblate.git.squash"
27 verbose = gettext_lazy("Squash Git commits")
28 description = gettext_lazy("Squash Git commits prior to pushing changes.")
29 settings_form = GitSquashForm
30 compat: ClassVar[CompatDict] = {
31 "vcs": VCS_REGISTRY.git_based,
32 }
33 events: ClassVar[set[AddonEvent]] = {
34 AddonEvent.EVENT_POST_COMMIT,
35 }
36 icon = "compress.svg"
37 repo_scope = True
39 def squash_repo(
40 self,
41 component: Component,
42 repository: GitRepository,
43 remote: str,
44 author: str | None = None,
45 ) -> None:
46 message = self.get_squash_commit_message(repository, "%B", remote)
47 repository.execute(["reset", "--mixed", remote])
48 # Can happen for added and removed translation
49 component.commit_files(
50 author=author, message=message, signals=False, skip_push=True
51 )
53 def squash_all(self, component: Component, repository: GitRepository) -> None:
54 self.squash_repo(component, repository, repository.get_remote_branch_name())
56 def get_filenames(self, component: Component) -> dict[str, list[str]]:
57 languages: dict[str, list[str]] = defaultdict(list)
58 for origin in [component, *list(component.linked_children)]:
59 for translation in origin.translation_set.prefetch_related("language"):
60 code = translation.language.code
61 if not translation.filename:
62 continue
63 languages[code].extend(translation.filenames)
64 return languages
66 def get_git_commit_messages(
67 self,
68 repository: GitRepository,
69 log_format: str,
70 remote: str,
71 filenames: list[str] | None,
72 ) -> str:
73 command = [
74 "log",
75 f"--format={log_format}",
76 f"{remote}..HEAD",
77 ]
78 if filenames:
79 command += ["--", *filenames]
81 return repository.execute(command)
83 def get_squash_commit_message(
84 self,
85 repository: GitRepository,
86 log_format: str,
87 remote: str,
88 filenames: list[str] | None = None,
89 ) -> str:
90 commit_message = self.instance.configuration.get("commit_message")
92 if self.instance.configuration.get("append_trailers", True):
93 command = [
94 "log",
95 "--format=%(trailers)%nCo-authored-by: %an <%ae>",
96 f"{remote}..HEAD",
97 ]
98 if filenames:
99 command += ["--", *filenames]
101 trailer_lines = set()
102 change_id_line = None
103 for trailer in repository.execute(command).split("\n"):
104 # Skip blank lines
105 if not trailer.strip():
106 continue
108 # Pick only last Change-Id, there suppose to be only one in the
109 # commit (used by Gerrit)
110 if trailer.startswith("Change-Id:"):
111 change_id_line = trailer
112 continue
114 trailer_lines.add(trailer)
116 if change_id_line is not None:
117 trailer_lines.add(change_id_line)
119 if commit_message:
120 # Predefined commit message
121 body = [commit_message]
122 else:
123 # Extract commit messages from the log
124 body = [
125 line
126 for line in self.get_git_commit_messages(
127 repository, log_format, remote, filenames
128 ).split("\n")
129 if line not in trailer_lines
130 ]
132 commit_message = "\n".join(
133 chain(
134 # Body
135 body,
136 # Blank line
137 [""],
138 # Trailers
139 sorted(trailer_lines),
140 )
141 ).strip("\n")
142 elif not commit_message:
143 commit_message = self.get_git_commit_messages(
144 repository, log_format, remote, filenames
145 )
147 return commit_message
149 def squash_language(self, component: Component, repository: GitRepository) -> None:
150 remote = repository.get_remote_branch_name()
151 languages = self.get_filenames(component)
153 messages = {}
154 for code, filenames in languages.items():
155 if not filenames:
156 continue
157 messages[code] = self.get_squash_commit_message(
158 repository, "%B", remote, filenames
159 )
161 repository.execute(["reset", "--mixed", remote])
163 for code, message in messages.items():
164 if not message:
165 continue
166 component.commit_files(
167 message=message, files=languages[code], signals=False, skip_push=True
168 )
170 def squash_file(self, component: Component, repository: GitRepository) -> None:
171 remote = repository.get_remote_branch_name()
172 languages = self.get_filenames(component)
174 messages = {}
175 for filenames in languages.values():
176 for filename in filenames:
177 messages[filename] = self.get_squash_commit_message(
178 repository, "%B", remote, [filename]
179 )
181 repository.execute(["reset", "--mixed", remote])
183 for filename, message in messages.items():
184 if not message:
185 continue
186 component.commit_files(
187 message=message, files=[filename], signals=False, skip_push=True
188 )
190 def squash_author(self, component: Component, repository: GitRepository) -> None:
191 remote = repository.get_remote_branch_name()
192 # Get list of pending commits with authors
193 commits: list[tuple[str, str]] = [
194 x.split(None, 1)
195 for x in reversed(
196 repository.execute(
197 ["log", "--no-merges", "--format=%H %aE", f"{remote}..HEAD"]
198 ).splitlines()
199 )
200 ]
201 gpg_sign = repository.get_gpg_sign_args()
203 tmp = "weblate-squash-tmp"
204 repository.delete_branch(tmp)
205 try:
206 # Create local branch for upstream
207 repository.execute(["branch", tmp, remote])
208 # Checkout upstream branch
209 repository.execute(["checkout", tmp])
210 while commits:
211 commit, author = commits.pop(0)
212 # Remember current revision for final squash
213 base = repository.get_last_revision()
214 # Cherry pick current commit (this should work
215 # unless something is messed up)
216 repository.execute(
217 ["cherry-pick", commit, *gpg_sign],
218 environment={"WEBLATE_MERGE_SKIP": "1"},
219 )
220 handled = []
221 # Pick other commits by same author
222 for i, other in enumerate(commits):
223 if other[1] != author:
224 continue
225 try:
226 repository.execute(
227 ["cherry-pick", other[0], *gpg_sign],
228 environment={"WEBLATE_MERGE_SKIP": "1"},
229 )
230 handled.append(i)
231 except RepositoryError:
232 # If fails, continue to another author, we will
233 # pick this commit later (it depends on some other)
234 repository.execute(["cherry-pick", "--abort"])
235 break
236 # Remove processed commits from list
237 for i in reversed(handled):
238 del commits[i]
239 # Squash all current commits from one author
240 self.squash_repo(component, repository, base, author)
242 # Update working copy with squashed commits
243 repository.execute(["checkout", repository.branch])
244 repository.execute(["reset", "--hard", tmp])
245 repository.delete_branch(tmp)
247 except Exception:
248 report_error("Failed squash", project=component.project)
249 # Revert to original branch without any changes
250 repository.execute(["reset", "--hard"])
251 repository.execute(["checkout", repository.branch])
252 repository.delete_branch(tmp)
254 def post_commit(
255 self, component: Component, store_hash: bool, activity_log_id: int | None = None
256 ) -> None:
257 # Operate on parent
258 if component.linked_component:
259 component = component.linked_component
261 repository = cast("GitRepository", component.repository)
262 branch_updated = False
263 with repository.lock:
264 # Ensure repository is rebased on current remote prior to squash, otherwise
265 # we might be squashing upstream changes as well due to reset.
266 if component.repo_needs_merge():
267 try:
268 branch_updated = component.update_branch(
269 method="rebase", skip_push=True
270 )
271 except RepositoryError:
272 return
273 if not repository.needs_push():
274 return
275 match self.instance.configuration["squash"]:
276 case "all":
277 self.squash_all(component, repository)
278 case "language":
279 self.squash_language(component, repository)
280 case "file":
281 self.squash_file(component, repository)
282 case "author":
283 self.squash_author(component, repository)
284 case _:
285 msg = f"Unsupported squash style: {self.instance.configuration['squash']}"
286 raise ValueError(msg)
287 # Commit any left files, those were most likely generated
288 # by addon and do not exactly match patterns above
289 component.commit_files(
290 template=component.addon_message,
291 extra_context={"addon_name": self.verbose},
292 signals=False,
293 skip_push=True,
294 )
295 # Parse translation files to process any updates fetched by update_branch
296 if branch_updated:
297 component.create_translations()