Coverage for src / mafw / devtools / cli / dependencies / registry.py: 100%
42 statements
« prev ^ index » next coverage.py v7.14.0, created at 2026-06-28 13:34 +0000
1# Copyright 2026 European Union
2# Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
3# SPDX-License-Identifier: EUPL-1.2
4"""GitLab registry operations for dependency reference files."""
6from __future__ import annotations
8import re
9from pathlib import Path
10from typing import Any, Final
12import click
13from rich.prompt import Confirm
15from mafw.__about__ import __version__ as MAFW_VERSION
16from mafw.devtools.gitlab import (
17 GitlabAPIConfiguration,
18 build_gitlab_api_configuration,
19 delete_generic_package_file,
20 delete_generic_package_version,
21 download_generic_file,
22 iter_local_pylock_reference_files,
23 list_generic_package_files,
24 normalize_dependency_registry_item,
25 normalize_mafw_version,
26 parse_pylock_reference_filename,
27 resolve_package_ids_by_version,
28 upload_generic_file,
29)
30from mafw.tools.click_extensions import AbbreviateGroup
31from mafw.tools.shell_tools import CONSOLE
33DEPS_REGISTRY_PACKAGE_NAME: Final[str] = 'mafw-deps'
34"""Generic package name used for dependency reference uploads."""
36PYLOCK_REFERENCE_PATTERN = re.compile(r'^pylock\.py(?P<python_version>3\.\d+)_ref\.toml$')
37"""Regular expression used to validate dependency reference lock filenames."""
40def _registry_api_config(ctx: click.Context) -> GitlabAPIConfiguration:
41 obj = ctx.obj or {}
42 try:
43 return build_gitlab_api_configuration(
44 obj.get('gitlab_api_url'),
45 obj.get('gitlab_project_id'),
46 obj.get('gitlab_token'),
47 )
48 except ValueError as exc:
49 raise click.ClickException(str(exc)) from exc
52def _normalize_pylock_input(item: str) -> tuple[str, str]:
53 """Normalize a dependency registry item into its Python version and file name."""
54 try:
55 return normalize_dependency_registry_item(item)
56 except ValueError as exc:
57 raise click.ClickException(str(exc)) from exc
60def _normalize_mafw_version_option(value: str | None) -> str:
61 """Return the registry label version, defaulting to the current MAFw version."""
62 if value is None:
63 return normalize_mafw_version(MAFW_VERSION)
64 try:
65 return normalize_mafw_version(value)
66 except ValueError as exc:
67 raise click.ClickException(str(exc)) from exc
70@click.group(
71 context_settings={'help_option_names': ['-h', '--help']},
72 help='GitLab registry operations for dependency reference files.',
73 cls=AbbreviateGroup,
74)
75@click.option('--gitlab-api-url', default=None, envvar='CI_API_V4_URL', help='GitLab API v4 base URL.')
76@click.option(
77 '--gitlab-project-id',
78 default=None,
79 type=int,
80 envvar='CI_PROJECT_ID',
81 help='GitLab project numeric id (env: CI_PROJECT_ID).',
82)
83@click.option('--gitlab-token', default=None, envvar='CI_JOB_TOKEN', help='GitLab API token.')
84@click.pass_context
85def registry(
86 ctx: click.Context,
87 gitlab_api_url: str | None,
88 gitlab_project_id: int | None,
89 gitlab_token: str | None,
90) -> None:
91 """GitLab registry command group for dependency reference files."""
92 ctx.obj = ctx.obj or {}
93 ctx.obj['gitlab_api_url'] = gitlab_api_url
94 ctx.obj['gitlab_project_id'] = gitlab_project_id
95 ctx.obj['gitlab_token'] = gitlab_token
98@registry.command()
99@click.argument('items', nargs=-1, required=False)
100@click.option('-a', '--all', 'all_entries', is_flag=True, default=False, help='Upload all matching reference files.')
101@click.option(
102 '-f',
103 '--force',
104 is_flag=True,
105 default=False,
106 help='Force upload by removing existing remote files before uploading. '
107 'Without this flag, files already present in the registry are skipped.',
108)
109@click.option(
110 '--mafw-version',
111 default=None,
112 help='MAFw package version label to use in the registry. Accepts X.Y.Z or vX.Y.Z.',
113)
114@click.pass_context
115def upload( # pragma: no cover — thin CLI delegation to tested gitlab.registry
116 ctx: click.Context, items: tuple[str, ...], all_entries: bool, force: bool, mafw_version: str | None
117) -> None:
118 """Upload dependency reference files to the GitLab generic registry."""
119 api_config = _registry_api_config(ctx)
120 package_version = _normalize_mafw_version_option(mafw_version)
121 candidates: list[tuple[str, Path]] = []
123 if all_entries:
124 if items:
125 raise click.ClickException('Do not pass filenames when using --all.')
126 candidates = iter_local_pylock_reference_files(Path.cwd())
127 else:
128 if not items:
129 raise click.ClickException('Provide one or more filenames or Python versions, or use --all.')
130 for item in items:
131 python_version, normalized_name = _normalize_pylock_input(item)
132 fp = Path(item)
133 if not fp.is_absolute():
134 cwd_fp = Path.cwd() / fp
135 if cwd_fp.exists():
136 fp = cwd_fp
137 elif item == python_version:
138 fp = Path.cwd() / normalized_name
139 if not fp.exists():
140 raise click.ClickException(f'File not found: {fp}')
141 if not PYLOCK_REFERENCE_PATTERN.fullmatch(fp.name):
142 raise click.ClickException(f'File does not match pylock.py3.xx_ref.toml: {fp.name}')
143 candidates.append((python_version, fp))
145 if not candidates:
146 CONSOLE.print('No matching reference files found.')
147 return
149 for python_version, fp in candidates:
150 uploaded = upload_generic_file(
151 api_config,
152 DEPS_REGISTRY_PACKAGE_NAME,
153 package_version,
154 fp,
155 replace_existing=force,
156 )
157 if uploaded:
158 CONSOLE.print(
159 f'Uploaded {fp.name} for Python {python_version} as {DEPS_REGISTRY_PACKAGE_NAME}/{package_version}'
160 )
161 else:
162 CONSOLE.print(
163 f'Skipped {fp.name} (already exists in '
164 f'{DEPS_REGISTRY_PACKAGE_NAME}/{package_version}). Use --force to replace.'
165 )
168@registry.command()
169@click.argument('items', nargs=-1, required=False)
170@click.option('--output-dir', default='.', type=click.Path(file_okay=False, dir_okay=True, path_type=Path))
171@click.option('-a', '--all', 'all_entries', is_flag=True, default=False, help='Download all registry files.')
172@click.option(
173 '--mafw-version',
174 default=None,
175 help='MAFw package version label to use in the registry. Accepts X.Y.Z or vX.Y.Z.',
176)
177@click.pass_context
178def download( # pragma: no cover — thin CLI delegation to tested gitlab.registry
179 ctx: click.Context,
180 items: tuple[str, ...],
181 output_dir: Path,
182 all_entries: bool,
183 mafw_version: str | None,
184) -> None:
185 """Download dependency reference files from the GitLab generic registry."""
186 api_config = _registry_api_config(ctx)
187 package_version = _normalize_mafw_version_option(mafw_version)
188 output_dir = Path(output_dir).resolve()
189 output_dir.mkdir(parents=True, exist_ok=True)
191 if all_entries and items:
192 raise click.ClickException('Do not pass filenames when using --all.')
194 mapping = resolve_package_ids_by_version(api_config, DEPS_REGISTRY_PACKAGE_NAME)
195 pkg_id = mapping.get(package_version)
196 if pkg_id is None:
197 CONSOLE.print(f'Not found in registry: {DEPS_REGISTRY_PACKAGE_NAME}/{package_version}')
198 return
200 targets: list[tuple[str, str]] = []
201 if all_entries:
202 package_files = list_generic_package_files(api_config, pkg_id)
203 for pf in package_files:
204 fname = pf.get('file_name')
205 if isinstance(fname, str):
206 parsed = parse_pylock_reference_filename(fname)
207 if parsed is not None:
208 python_version, normalized_name = parsed
209 targets.append((python_version, normalized_name))
210 targets.sort(key=lambda item: tuple(int(part) for part in item[0].split('.')))
211 else:
212 if not items:
213 raise click.ClickException('Provide one or more filenames or Python versions, or use --all.')
214 for item in items:
215 python_version, file_name = _normalize_pylock_input(item)
216 targets.append((python_version, file_name))
218 if not targets:
219 CONSOLE.print(f'No reference files found in {DEPS_REGISTRY_PACKAGE_NAME}/{package_version}.')
220 return
222 for python_version, file_name in targets:
223 dest = download_generic_file(
224 api_config,
225 DEPS_REGISTRY_PACKAGE_NAME,
226 package_version,
227 file_name,
228 output_dir,
229 )
230 if dest is not None:
231 CONSOLE.print(f'Downloaded {python_version}: {dest}')
232 else:
233 CONSOLE.print(f'Not found in registry: {DEPS_REGISTRY_PACKAGE_NAME}/{package_version}/{file_name}')
236@registry.command()
237@click.argument('items', nargs=-1, required=False)
238@click.option('-a', '--all', 'all_entries', is_flag=True, default=False, help='Delete all registry files.')
239@click.option(
240 '-y',
241 '--yes',
242 is_flag=True,
243 default=False,
244 help='Skip confirmation prompt and proceed with deletion. Useful for CI/CD automation.',
245)
246@click.option(
247 '--mafw-version',
248 default=None,
249 help='MAFw package version label to use in the registry. Accepts X.Y.Z or vX.Y.Z.',
250)
251@click.pass_context
252def delete(
253 ctx: click.Context, items: tuple[str, ...], all_entries: bool, yes: bool, mafw_version: str | None
254) -> None: # pragma: no cover — thin CLI delegation to tested gitlab.registry
255 """Delete dependency reference files from the GitLab generic registry."""
256 api_config = _registry_api_config(ctx)
257 package_version = _normalize_mafw_version_option(mafw_version)
259 mapping = resolve_package_ids_by_version(api_config, DEPS_REGISTRY_PACKAGE_NAME)
260 pkg_id = mapping.get(package_version)
261 if pkg_id is None:
262 CONSOLE.print(f'Not found in registry: {DEPS_REGISTRY_PACKAGE_NAME}/{package_version}')
263 return
265 if all_entries and items:
266 raise click.ClickException('Do not pass filenames when using --all.')
268 if all_entries:
269 if not yes:
270 confirmed = Confirm.ask(
271 f'Delete all reference files from {DEPS_REGISTRY_PACKAGE_NAME}/{package_version}?',
272 default=False,
273 console=CONSOLE,
274 )
275 if not confirmed:
276 CONSOLE.print('Aborted.')
277 return
279 deleted = delete_generic_package_version(api_config, pkg_id, DEPS_REGISTRY_PACKAGE_NAME, package_version)
280 if deleted:
281 CONSOLE.print(f'Deleted all reference files from {DEPS_REGISTRY_PACKAGE_NAME}/{package_version}')
282 else:
283 CONSOLE.print(f'Not found in registry: {DEPS_REGISTRY_PACKAGE_NAME}/{package_version}')
284 return
286 if not items:
287 raise click.ClickException('Provide one or more filenames or Python versions, or use --all.')
289 targets: list[tuple[str, str]] = []
290 for item in items:
291 python_version, file_name = _normalize_pylock_input(item)
292 targets.append((python_version, file_name))
294 if not yes:
295 file_list = ', '.join(file_name for _, file_name in targets)
296 confirmed = Confirm.ask(
297 f'Delete {file_list} from {DEPS_REGISTRY_PACKAGE_NAME}/{package_version}?',
298 default=False,
299 console=CONSOLE,
300 )
301 if not confirmed:
302 CONSOLE.print('Aborted.')
303 return
305 package_files = list_generic_package_files(api_config, pkg_id)
306 file_name_to_id: dict[str, int] = {}
307 for pf in package_files:
308 fname = pf.get('file_name')
309 fid = pf.get('id')
310 if isinstance(fname, str) and isinstance(fid, int):
311 file_name_to_id[fname] = fid
313 for _python_version, file_name in targets:
314 file_id = file_name_to_id.get(file_name)
315 if file_id is None:
316 CONSOLE.print(f'Not found in registry: {DEPS_REGISTRY_PACKAGE_NAME}/{package_version}/{file_name}')
317 continue
318 deleted = delete_generic_package_file(api_config, pkg_id, file_id)
319 if deleted:
320 CONSOLE.print(f'Deleted {DEPS_REGISTRY_PACKAGE_NAME}/{package_version}/{file_name}')
321 else:
322 CONSOLE.print(f'Not found in registry: {DEPS_REGISTRY_PACKAGE_NAME}/{package_version}/{file_name}')
325@registry.command()
326@click.option(
327 '--keep-newest/--keep-oldest',
328 default=True,
329 show_default=True,
330 help='When duplicates are found, keep the newest (default) or the oldest file.',
331)
332@click.option(
333 '-f',
334 '--force',
335 is_flag=True,
336 default=False,
337 help='Skip confirmation prompt and proceed with pruning. Useful for CI/CD automation.',
338)
339@click.option(
340 '--mafw-version',
341 default=None,
342 help='MAFw package version label to use in the registry. Accepts X.Y.Z or vX.Y.Z.',
343)
344@click.pass_context
345def prune(
346 ctx: click.Context, keep_newest: bool, force: bool, mafw_version: str | None
347) -> None: # pragma: no cover — thin CLI delegation to tested gitlab.registry
348 """Remove duplicated reference files from a registry package version."""
349 api_config = _registry_api_config(ctx)
350 package_version = _normalize_mafw_version_option(mafw_version)
352 mapping = resolve_package_ids_by_version(api_config, DEPS_REGISTRY_PACKAGE_NAME)
353 pkg_id = mapping.get(package_version)
354 if pkg_id is None:
355 CONSOLE.print(f'Not found in registry: {DEPS_REGISTRY_PACKAGE_NAME}/{package_version}')
356 return
358 package_files = list_generic_package_files(api_config, pkg_id)
360 files_by_name: dict[str, list[dict[str, Any]]] = {}
361 for pf in package_files:
362 fname = pf.get('file_name')
363 if isinstance(fname, str):
364 files_by_name.setdefault(fname, []).append(pf)
366 to_delete: list[tuple[str, int, str]] = []
367 for fname, entries in files_by_name.items():
368 if len(entries) <= 1:
369 continue
370 sorted_entries = sorted(entries, key=lambda e: e.get('created_at', ''))
371 if keep_newest:
372 duplicates = sorted_entries[:-1]
373 else:
374 duplicates = sorted_entries[1:]
375 for entry in duplicates:
376 fid = entry.get('id')
377 created_at = entry.get('created_at', 'unknown')
378 if isinstance(fid, int):
379 to_delete.append((fname, fid, str(created_at)))
381 if not to_delete:
382 CONSOLE.print(f'No duplicates found in {DEPS_REGISTRY_PACKAGE_NAME}/{package_version}.')
383 return
385 keep_label = 'newest' if keep_newest else 'oldest'
386 CONSOLE.print(
387 f'Found {len(to_delete)} duplicate(s) to remove '
388 f'(keeping {keep_label}) in {DEPS_REGISTRY_PACKAGE_NAME}/{package_version}:'
389 )
390 for fname, _fid, created_at in to_delete:
391 CONSOLE.print(f' - {fname} (created: {created_at})')
393 if not force:
394 confirmed = Confirm.ask(
395 'Proceed with deletion?',
396 default=False,
397 console=CONSOLE,
398 )
399 if not confirmed:
400 CONSOLE.print('Aborted.')
401 return
403 deleted_count = 0
404 for fname, fid, created_at in to_delete:
405 deleted = delete_generic_package_file(api_config, pkg_id, fid)
406 if deleted:
407 CONSOLE.print(f'Deleted {fname} (created: {created_at})')
408 deleted_count += 1
409 else:
410 CONSOLE.print(f'Failed to delete {fname} (id={fid})')
412 CONSOLE.print(f'Pruning complete: {deleted_count} duplicate(s) removed.')