Coverage for src/gitlabracadabra/mixins/mirrors.py: 79%
133 statements
« prev ^ index » next coverage.py v7.15.0, created at 2026-07-07 22:55 +0200
1#
2# Copyright (C) 2019-2025 Mathieu Parent <math.parent@gmail.com>
3#
4# This program is free software: you can redistribute it and/or modify
5# it under the terms of the GNU Lesser General Public License as published by
6# the Free Software Foundation, either version 3 of the License, or
7# (at your option) any later version.
8#
9# This program is distributed in the hope that it will be useful,
10# but WITHOUT ANY WARRANTY; without even the implied warranty of
11# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12# GNU Lesser General Public License for more details.
13#
14# You should have received a copy of the GNU Lesser General Public License
15# along with this program. If not, see <http://www.gnu.org/licenses/>.
17from __future__ import annotations
19import logging
20from contextlib import suppress
21from os.path import isdir
22from typing import TYPE_CHECKING, Any
23from urllib.parse import quote
25from pygit2 import GIT_FETCH_PRUNE, Commit, GitError, RemoteCallbacks, Repository, init_repository
27from gitlabracadabra.disk_cache import cache_dir
28from gitlabracadabra.gitlab.connections import GitlabConnections
29from gitlabracadabra.matchers import Matcher
30from gitlabracadabra.objects.object import GitLabracadabraObject
32if TYPE_CHECKING:
33 from typing import NotRequired, TypedDict
35 from pygit2 import Reference
37 class RemotePushArgs(TypedDict):
38 specs: list[str]
39 callbacks: NotRequired[RemoteCallbacks | None]
40 proxy: NotRequired[None | bool | str]
41 push_options: NotRequired[None | list[str]]
44GITLAB_REMOTE_NAME = "gitlab"
45MIRROR_PARAM_URL = "url"
46MIRROR_DIRECTION_PULL = "pull"
47MIRROR_REMOTE_NAME_PULL = "pull"
49logger = logging.getLogger(__name__)
52class MirrorsMixin(GitLabracadabraObject):
53 """Object with mirrors."""
55 def _process_mirrors(
56 self,
57 param_name: str,
58 param_value: Any,
59 *,
60 dry_run: bool = False,
61 skip_save: bool = False,
62 ) -> None:
63 """Process the mirrors param.
65 Args:
66 param_name: "mirrors".
67 param_value: List of mirror dicts.
68 dry_run: Dry run.
69 skip_save: False.
70 """
71 assert param_name == "mirrors" # noqa: S101
72 assert not skip_save # noqa: S101
74 mirrors = [mirror for mirror in param_value if mirror.get("enabled", True)]
76 if not mirrors: 76 ↛ 78line 76 didn't jump to line 78 because the condition on line 76 was never true
77 # do not fetch repo when there is no mirror
78 return
80 pull_mirror_count = 0
81 self._init_repo()
82 self._fetch_remote(
83 GITLAB_REMOTE_NAME,
84 self.connection.pygit2_remote_callbacks,
85 )
86 for mirror in mirrors:
87 direction = mirror.get("direction", MIRROR_DIRECTION_PULL)
88 push_options = mirror.get("push_options", [])
89 if "skip_ci" in mirror: 89 ↛ 90line 89 didn't jump to line 90 because the condition on line 89 was never true
90 push_options.append("ci.skip")
91 if direction == MIRROR_DIRECTION_PULL: 91 ↛ 102line 91 didn't jump to line 102 because the condition on line 91 was always true
92 if pull_mirror_count > 0: 92 ↛ 93line 92 didn't jump to line 93 because the condition on line 92 was never true
93 logger.warning(
94 "[%s] NOT Pulling mirror: %s (Only first pull mirror is processed)",
95 self._name,
96 mirror[MIRROR_PARAM_URL],
97 )
98 continue
99 self._pull_mirror(mirror, push_options, dry_run=dry_run)
100 pull_mirror_count += 1
101 else:
102 logger.warning(
103 "[%s] NOT Pushing mirror: %s (Not supported yet)",
104 self._name,
105 mirror[MIRROR_PARAM_URL],
106 )
108 def _init_repo(self) -> None:
109 """Init the cache repository."""
110 web_url_slug = quote(self.web_url(), safe="")
111 repo_dir = str(cache_dir("") / web_url_slug)
112 if isdir(repo_dir): 112 ↛ 113line 112 didn't jump to line 113 because the condition on line 112 was never true
113 self._repo = Repository(repo_dir)
114 else:
115 logger.debug(
116 "[%s] Creating cache repository in %s",
117 self._name,
118 repo_dir,
119 )
120 self._repo = init_repository(repo_dir, bare=True)
121 try:
122 self._repo.remotes[GITLAB_REMOTE_NAME]
123 except KeyError:
124 self._repo.remotes.create(
125 GITLAB_REMOTE_NAME,
126 self.web_url(),
127 "+refs/heads/*:refs/remotes/gitlab/heads/*",
128 )
129 self._repo.remotes.add_fetch(
130 GITLAB_REMOTE_NAME,
131 "+refs/tags/*:refs/remotes/gitlab/tags/*",
132 )
133 self._repo.remotes.add_push(GITLAB_REMOTE_NAME, "+refs/heads/*:refs/heads/*")
134 self._repo.remotes.add_push(GITLAB_REMOTE_NAME, "+refs/tags/*:refs/tags/*")
135 self._repo.config["remote.gitlab.mirror"] = True
137 def _fetch_remote(
138 self,
139 name: str,
140 remote_callbacks: RemoteCallbacks | None = None,
141 ) -> None:
142 """Fetch the repo with the given name.
144 Args:
145 name: Remote name.
146 remote_callbacks: Credentials and certificate check as pygit2.RemoteCallbacks.
147 """
148 remote = self._repo.remotes[name]
149 try:
150 # https://gitlab.com/gitlabracadabra/gitlabracadabra/-/issues/25
151 remote.fetch(
152 refspecs=remote.fetch_refspecs,
153 callbacks=remote_callbacks,
154 prune=GIT_FETCH_PRUNE,
155 proxy=True,
156 )
157 except TypeError:
158 # proxy arg in pygit2 1.6.0
159 logger.warning(
160 "[%s] Ignoring proxy for remote=%s refs=%s: requires pygit2>=1.6.0",
161 self._name,
162 name,
163 ",".join(remote.fetch_refspecs),
164 )
165 remote.fetch(
166 refspecs=remote.fetch_refspecs,
167 callbacks=remote_callbacks,
168 prune=GIT_FETCH_PRUNE,
169 )
171 def _pull_mirror(self, mirror: dict, push_options: list[str], *, dry_run: bool) -> None:
172 """Pull from the given mirror and push.
174 Args:
175 mirror: Current mirror dict.
176 push_options: push options.
177 dry_run: Dry run.
178 """
179 try:
180 self._repo.remotes[MIRROR_REMOTE_NAME_PULL]
181 except KeyError:
182 self._repo.remotes.create(
183 MIRROR_REMOTE_NAME_PULL,
184 mirror[MIRROR_PARAM_URL],
185 "+refs/heads/*:refs/heads/*",
186 )
187 self._repo.remotes.add_fetch(
188 MIRROR_REMOTE_NAME_PULL,
189 "+refs/tags/*:refs/tags/*",
190 )
191 self._repo.config["remote.pull.mirror"] = True
192 remote_callbacks = None
193 pull_auth_id = mirror.get("auth_id")
194 if pull_auth_id: 194 ↛ 195line 194 didn't jump to line 195 because the condition on line 194 was never true
195 remote_callbacks = GitlabConnections().get_connection(pull_auth_id).pygit2_remote_callbacks
196 self._fetch_remote(MIRROR_REMOTE_NAME_PULL, remote_callbacks)
197 for ref in self._repo.references.objects:
198 self._sync_ref(mirror, ref, push_options, dry_run=dry_run)
200 def _sync_ref(
201 self,
202 mirror: dict,
203 ref: Reference,
204 push_options: list[str],
205 *,
206 dry_run: bool,
207 ) -> None:
208 """Synchronize the given branch or tag.
210 Args:
211 mirror: Current mirror dict.
212 ref: reference objects.
213 push_options: push options.
214 dry_run: Dry run.
215 """
216 if ref.name.startswith("refs/heads/"):
217 ref_type = "head"
218 ref_type_human = "branch"
219 ref_type_human_plural = "branches"
220 elif ref.name.startswith("refs/tags/"): 220 ↛ 225line 220 didn't jump to line 225 because the condition on line 220 was always true
221 ref_type = "tag"
222 ref_type_human = "tag"
223 ref_type_human_plural = "tags"
224 else:
225 return
226 shorthand = ref.name.split("/", 2)[2]
228 # Ref mapping
229 dest_shortand: str | None = shorthand
230 if ref_type_human_plural in mirror:
231 dest_shortand = None
232 mappings: list[dict[str, str | list[str]]] = mirror.get(ref_type_human_plural) # type: ignore
233 for mapping in mappings:
234 matcher = Matcher(
235 mapping.get("from", ""),
236 None,
237 log_prefix=f"[{self._name}] {mirror[MIRROR_PARAM_URL]} {ref_type_human_plural}",
238 )
239 matches = matcher.match([shorthand])
240 if matches:
241 to_param = mapping.get("to", shorthand)
242 dest_shortand = matches[0].expand(to_param)
243 push_options = mapping.get("push_options", push_options) # type: ignore
244 break
246 if dest_shortand is None:
247 return
249 pull_commit = ref.peel(Commit).id
250 gitlab_ref = self._repo.references.get(
251 f"refs/remotes/gitlab/{ref_type}s/{dest_shortand}",
252 )
253 gitlab_commit = None
254 if gitlab_ref: 254 ↛ 255line 254 didn't jump to line 255 because the condition on line 254 was never true
255 with suppress(AttributeError):
256 gitlab_commit = gitlab_ref.peel(Commit).id
257 if pull_commit == gitlab_commit: 257 ↛ 258line 257 didn't jump to line 258 because the condition on line 257 was never true
258 return
259 if dry_run: 259 ↛ 260line 259 didn't jump to line 260 because the condition on line 259 was never true
260 logger.info(
261 "[%s] %s NOT Pushing %s %s to %s: %s -> %s (dry-run)",
262 self._name,
263 mirror[MIRROR_PARAM_URL],
264 ref_type_human,
265 shorthand,
266 dest_shortand,
267 gitlab_commit,
268 str(pull_commit),
269 )
270 return
271 logger.info(
272 "[%s] %s Pushing %s %s to %s: %s -> %s",
273 self._name,
274 mirror[MIRROR_PARAM_URL],
275 ref_type_human,
276 shorthand,
277 dest_shortand,
278 gitlab_commit,
279 str(pull_commit),
280 )
281 refspec = f"{ref.name}:refs/{ref_type}s/{dest_shortand}"
282 try:
283 self._push_remote(
284 GITLAB_REMOTE_NAME,
285 [refspec],
286 push_options,
287 self.connection.pygit2_remote_callbacks,
288 )
289 except GitError as err:
290 logger.error(
291 "[%s] Unable to push remote=%s refs=%s: %s",
292 self._name,
293 GITLAB_REMOTE_NAME,
294 refspec,
295 err,
296 )
298 def _push_remote(
299 self,
300 name: str,
301 refs: list[str],
302 push_options: list[str],
303 remote_callbacks: RemoteCallbacks | None,
304 ) -> None:
305 """Push to the repo with the given name.
307 Args:
308 name: Remote name.
309 refs: refs list.
310 push_options: push options.
311 remote_callbacks: Credentials and certificate check as pygit2.RemoteCallbacks.
312 """
313 remote = self._repo.remotes[name]
314 kwargs: RemotePushArgs = {
315 "specs": refs,
316 "callbacks": remote_callbacks,
317 "proxy": True,
318 }
319 if push_options:
320 kwargs["push_options"] = push_options
321 try:
322 remote.push(**kwargs)
323 except TypeError:
324 # push_options arg in pygit2 1.16.0
325 logger.warning(
326 "[%s] Ignoring push options %s for remote=%s refs=%s: requires pygit2>=1.16.0",
327 self._name,
328 ",".join(push_options),
329 name,
330 ",".join(refs),
331 )
332 kwargs.pop("push_options")
333 else:
334 return
335 try:
336 remote.push(**kwargs)
337 except TypeError:
338 # proxy arg in pygit2 1.6.0
339 logger.warning(
340 "[%s] Ignoring proxy for remote=%s refs=%s: requires pygit2>=1.6.0",
341 self._name,
342 name,
343 ",".join(refs),
344 )
345 kwargs.pop("proxy")
346 remote.push(**kwargs)