Coverage for src / mafw / devtools / gitlab / docs_registry.py: 96%
94 statements
« prev ^ index » next coverage.py v7.14.0, created at 2026-06-28 13:34 +0000
1# Copyright 2025–2026 European Union
2# Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
3# SPDX-License-Identifier: EUPL-1.2
4"""
5Documentation-specific GitLab Generic Package Registry operations.
7This module wraps the generic GitLab registry helpers with
8MAFw-docs-specific defaults (package name, filename conventions).
9"""
11from __future__ import annotations
13from pathlib import Path
14from typing import Any, List
16import requests
18from mafw.devtools.gitlab.api import (
19 GitlabAPIConfiguration,
20 build_gitlab_auth_headers,
21)
24def list_mafw_docs_generic_packages(
25 api_config: GitlabAPIConfiguration,
26 package_name: str = 'mafw-docs',
27) -> List[dict[str, Any]]:
28 """List generic packages for mafw-docs in the GitLab Package Registry.
30 Uses the Packages API:
31 ``GET /projects/:id/packages``
32 and filters for ``package_type=generic`` and exact ``name == package_name``.
34 :param api_config: GitLab API configuration
35 :type api_config: GitlabAPIConfiguration
36 :param package_name: Package name, defaults to ``mafw-docs``
37 :type package_name: str
38 :return: List of package dictionaries from the API
39 :rtype: List[dict[str, Any]]
40 """
41 base_url = api_config.api_url.rstrip('/')
42 url = f'{base_url}/projects/{api_config.project_id}/packages'
43 headers = build_gitlab_auth_headers(api_config)
45 per_page = 100
46 page = 1
47 out: List[dict[str, Any]] = []
48 while True:
49 params: dict[str, str | int] = {
50 'package_type': 'generic',
51 'package_name': package_name,
52 'per_page': per_page,
53 'page': page,
54 'order_by': 'version',
55 'sort': 'asc',
56 }
57 resp = requests.get(url, headers=headers, params=params, timeout=60.0)
58 if not (200 <= resp.status_code < 300):
59 body_preview = (resp.text or '')[:500].replace('\n', ' ')
60 raise RuntimeError(f'GitLab list packages failed ({resp.status_code}): {body_preview}')
61 data = resp.json()
62 if not data:
63 break
64 for pkg in data:
65 if pkg.get('package_type') != 'generic':
66 continue
67 if pkg.get('name') != package_name:
68 continue
69 out.append(pkg)
70 if len(data) < per_page:
71 break
72 page += 1
73 return out
76def resolve_mafw_docs_package_ids_by_version(
77 api_config: GitlabAPIConfiguration, package_name: str = 'mafw-docs'
78) -> dict[str, int]:
79 """Resolve package IDs for mafw-docs generic packages by version.
81 :param api_config: GitLab API configuration
82 :type api_config: GitlabAPIConfiguration
83 :param package_name: Package name, defaults to ``mafw-docs``
84 :type package_name: str
85 :return: Mapping from version string to package id
86 :rtype: dict[str, int]
87 """
88 pkgs = list_mafw_docs_generic_packages(api_config, package_name=package_name)
89 mapping: dict[str, int] = {}
90 for pkg in pkgs:
91 version = pkg.get('version')
92 pkg_id = pkg.get('id')
93 if isinstance(version, str) and isinstance(pkg_id, int): 93 ↛ 90line 93 didn't jump to line 90 because the condition on line 93 was always true
94 mapping[version] = pkg_id
95 return mapping
98def upload_docs_zip_to_gitlab_generic_registry(
99 api_config: GitlabAPIConfiguration,
100 package_version: str,
101 zip_path: Path,
102 package_name: str = 'mafw-docs',
103 timeout_s: float = 60.0,
104) -> bool:
105 """Upload a documentation zip archive to the GitLab Generic Package Registry.
107 The equivalent GitLab API endpoint is:
109 ``PUT /projects/:id/packages/generic/:package_name/:package_version/:file_name``
111 Authentication headers:
112 - ``JOB-TOKEN`` when running on CI (``api_config.on_ci`` is True)
113 - ``PRIVATE-TOKEN`` for local execution
115 :param api_config: GitLab API configuration
116 :type api_config: GitlabAPIConfiguration
117 :param package_version: Package version (typically the git tag, e.g. ``v2.1.0``)
118 :type package_version: str
119 :param zip_path: Path to the zip file to upload
120 :type zip_path: Path
121 :param package_name: Generic package name, defaults to ``mafw-docs``
122 :type package_name: str
123 :param timeout_s: Request timeout in seconds, defaults to 60.0
124 :type timeout_s: float
125 :return: True if the file was uploaded, False if the upload was skipped because the target already exists
126 :rtype: bool
127 :raises FileNotFoundError: If ``zip_path`` does not exist
128 :raises RuntimeError: If the upload fails (non-2xx response)
129 """
130 zip_path = Path(zip_path).resolve()
131 if not zip_path.exists():
132 raise FileNotFoundError(f'Zip file not found: {zip_path}')
134 base_url = api_config.api_url.rstrip('/')
135 file_name = zip_path.name
136 url = f'{base_url}/projects/{api_config.project_id}/packages/generic/{package_name}/{package_version}/{file_name}'
138 headers = build_gitlab_auth_headers(api_config)
140 head_resp = requests.head(url, headers=headers, timeout=timeout_s)
141 if head_resp.status_code == 200:
142 print(f'ℹ️ Zip already present on GitLab, skipping upload: {url}')
143 return False
144 if head_resp.status_code != 404:
145 body_preview = (head_resp.text or '')[:500].replace('\n', ' ')
146 raise RuntimeError(f'GitLab existence check failed ({head_resp.status_code}): {body_preview}')
148 with open(zip_path, 'rb') as f:
149 put_headers = dict(headers)
150 put_headers['Content-Type'] = 'application/zip'
151 resp = requests.put(url, headers=put_headers, data=f, timeout=timeout_s)
153 if not (200 <= resp.status_code < 300):
154 body_preview = (resp.text or '')[:500].replace('\n', ' ')
155 raise RuntimeError(f'GitLab upload failed ({resp.status_code}): {body_preview}')
156 return True
159def download_docs_zip_from_gitlab_generic_registry(
160 api_config: GitlabAPIConfiguration,
161 package_version: str,
162 download_dir: Path,
163 package_name: str = 'mafw-docs',
164 file_name: str | None = None,
165 timeout_s: float = 60.0,
166) -> Path | None:
167 """Download a documentation zip archive from the GitLab Generic Package Registry.
169 The equivalent GitLab API endpoint is:
171 ``GET /projects/:id/packages/generic/:package_name/:package_version/:file_name``
173 The function first issues a HEAD request to determine if the file exists:
174 - 404: the package file does not exist (cache miss) and ``None`` is returned.
175 - 200: the file exists and is downloaded.
176 - otherwise: a warning is printed and ``None`` is returned.
178 Authentication headers:
179 - ``JOB-TOKEN`` when running on CI (``api_config.on_ci`` is True)
180 - ``PRIVATE-TOKEN`` for local execution
182 :param api_config: GitLab API configuration
183 :type api_config: GitlabAPIConfiguration
184 :param package_version: Package version (typically the git tag, e.g. ``v2.1.0``)
185 :type package_version: str
186 :param download_dir: Directory where the downloaded zip is stored
187 :type download_dir: Path
188 :param package_name: Generic package name, defaults to ``mafw-docs``
189 :type package_name: str
190 :param file_name: File name to retrieve, defaults to ``mafw-docs-<version>.zip``
191 :type file_name: str | None
192 :param timeout_s: Request timeout in seconds, defaults to 60.0
193 :type timeout_s: float
194 :return: Path to the downloaded zip archive, or None if it does not exist or cannot be retrieved
195 :rtype: Path | None
196 """
197 download_dir = Path(download_dir).resolve()
198 download_dir.mkdir(parents=True, exist_ok=True)
200 base_url = api_config.api_url.rstrip('/')
201 target_file_name = file_name or f'mafw-docs-{package_version}.zip'
202 url = (
203 f'{base_url}/projects/{api_config.project_id}/packages/generic/'
204 f'{package_name}/{package_version}/{target_file_name}'
205 )
207 headers = build_gitlab_auth_headers(api_config)
209 try:
210 head_resp = requests.head(url, headers=headers, timeout=timeout_s)
211 except Exception as e: # pragma: no cover
212 print(f'⚠️ Warning: cache HEAD request failed for {package_version}: {e}')
213 return None
215 if head_resp.status_code == 404:
216 return None
217 if head_resp.status_code != 200:
218 body_preview = (head_resp.text or '')[:200].replace('\n', ' ')
219 print(
220 f'⚠️ Warning: cache existence check returned {head_resp.status_code} for {package_version}: {body_preview}'
221 )
222 return None
224 dest = download_dir / target_file_name
225 try:
226 resp = requests.get(url, headers=headers, stream=True, timeout=timeout_s)
227 except Exception as e: # pragma: no cover
228 print(f'⚠️ Warning: cache download request failed for {package_version}: {e}')
229 return None
231 if not (200 <= resp.status_code < 300):
232 body_preview = (resp.text or '')[:200].replace('\n', ' ')
233 print(f'⚠️ Warning: cache download failed for {package_version} ({resp.status_code}): {body_preview}')
234 return None
236 try:
237 with open(dest, 'wb') as f:
238 for chunk in resp.iter_content(chunk_size=1024 * 1024):
239 if chunk: 239 ↛ 238line 239 didn't jump to line 238 because the condition on line 239 was always true
240 f.write(chunk)
241 except OSError as e:
242 print(f'⚠️ Warning: could not write downloaded zip for {package_version} to {dest}: {e}')
243 return None
245 return dest