Coverage for src/gitlabracadabra/packages/gitlab.py: 93%

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/>. 

16 

17from __future__ import annotations 

18 

19from logging import getLogger 

20from typing import TYPE_CHECKING 

21from urllib.parse import quote 

22 

23from gitlab.v4.objects.packages import ProjectPackage, ProjectPackageFile 

24 

25from gitlabracadabra.packages.destination import Destination 

26 

27if TYPE_CHECKING: 

28 from gitlabracadabra.gitlab.connection import GitlabConnection 

29 from gitlabracadabra.packages.package_file import PackageFile 

30 

31 

32HELM = "helm" 

33PYPI = "pypi" 

34 

35logger = getLogger(__name__) 

36 

37 

38class Gitlab(Destination): 

39 """Gitlab repository.""" 

40 

41 def __init__( 

42 self, 

43 *, 

44 connection: GitlabConnection, 

45 full_path: str, 

46 project_id: int, 

47 ) -> None: 

48 """Initialize Gitlab repository. 

49 

50 Args: 

51 connection: A Gitlab connection. 

52 full_path: Project full path. 

53 project_id: Project ID. 

54 """ 

55 super().__init__(log_prefix=f"[{full_path}] ") 

56 self._connection = connection 

57 self._full_path = full_path 

58 self._project_id = project_id 

59 self._connection.session_callback(self.session) 

60 # dict[ 

61 # tuple[package_name: str, package_version: str], 

62 # list[ProjectPackageFile] 

63 # ] 

64 self._project_package_package_files_cache: dict[tuple[str, str], list[ProjectPackageFile]] = {} 

65 

66 def upload_method(self, package_file: PackageFile) -> str: 

67 """Get upload HTTP method. 

68 

69 Args: 

70 package_file: Source package file. 

71 

72 Returns: 

73 The upload method. 

74 """ 

75 if package_file.package_type in {HELM, PYPI}: 

76 return "POST" 

77 

78 return super().upload_method(package_file) 

79 

80 def get_url(self, package_file: PackageFile) -> str: 

81 """Get URL to test existence of destination package file with a HEAD request. 

82 

83 Args: 

84 package_file: Source package file. 

85 

86 Returns: 

87 An URL. 

88 

89 Raises: 

90 NotImplementedError: For unsupported package types. 

91 """ 

92 if package_file.package_type == "generic": 

93 return "{}/projects/{}/packages/generic/{}/{}/{}".format( 

94 self._connection.api_url, 

95 quote(self._full_path, safe=""), 

96 quote(package_file.package_name, safe=""), # [A-Za-z0-9\.\_\-\+]+ 

97 quote(package_file.package_version, safe=""), # (\.?[\w\+-]+\.?)+ 

98 quote(package_file.file_name, safe=""), # [A-Za-z0-9\.\_\-\+]+ 

99 ) 

100 if package_file.package_type == HELM: 

101 channel = package_file.metadata.get("channel") or "stable" 

102 file_name = f"{package_file.package_name}-{package_file.package_version}.tgz" 

103 return "{}/projects/{}/packages/helm/{}/charts/{}".format( 

104 self._connection.api_url, 

105 quote(self._full_path, safe=""), 

106 quote(channel, safe=""), 

107 quote(file_name, safe=""), 

108 ) 

109 if package_file.package_type == PYPI: 109 ↛ 117line 109 didn't jump to line 117 because the condition on line 109 was always true

110 return "{}/projects/{}/packages/pypi/files/{}/{}".format( 

111 self._connection.api_url, 

112 self._project_id, 

113 quote(package_file.metadata.get("sha256", ""), safe=""), 

114 quote(package_file.file_name, safe=""), 

115 ) 

116 

117 raise NotImplementedError 

118 

119 def upload_url(self, package_file: PackageFile) -> str: 

120 """Get URL to upload to. 

121 

122 Args: 

123 package_file: Source package file. 

124 

125 Returns: 

126 The upload URL. 

127 """ 

128 if package_file.package_type == HELM: 

129 channel = package_file.metadata.get("channel") or "stable" 

130 return "{}/projects/{}/packages/helm/api/{}/charts".format( 

131 self._connection.api_url, 

132 quote(self._full_path, safe=""), 

133 quote(channel, safe=""), 

134 ) 

135 if package_file.package_type == PYPI: 

136 return ( 

137 "{}/projects/{}/packages/pypi?requires_python={}&name={}&version={}&md5_digest={}&sha256_digest={}" 

138 ).format( 

139 self._connection.api_url, 

140 self._project_id, 

141 quote(package_file.metadata.get("requires-python", ""), safe=""), 

142 quote(package_file.package_name, safe=""), 

143 quote(package_file.package_version, safe=""), 

144 quote(package_file.metadata.get("md5", ""), safe=""), 

145 quote(package_file.metadata.get("sha256", ""), safe=""), 

146 ) 

147 

148 return super().upload_url(package_file) 

149 

150 def cache_project_package_package_files(self, package_type: str, package_name: str, package_version: str) -> None: 

151 if (package_name, package_version) not in self._project_package_package_files_cache: 

152 self._project_package_package_files_cache[(package_name, package_version)] = ( 

153 self._project_package_package_files(package_type, package_name, package_version) 

154 ) 

155 

156 def delete_package_file(self, package_file: PackageFile) -> None: 

157 # No DELETE endpoint for generic packages 

158 # https://gitlab.com/gitlab-org/gitlab/-/issues/536839 

159 self.cache_project_package_package_files( 

160 package_file.package_type, package_file.package_name, package_file.package_version 

161 ) 

162 for project_package_file in self._project_package_package_files_cache[ 

163 (package_file.package_name, package_file.package_version) 

164 ]: 

165 if project_package_file.file_name == package_file.file_name: 

166 project_package_file.delete() 

167 

168 def files_key(self, package_file: PackageFile) -> str | None: 

169 """Get files key, to upload to. If None, uploaded as body. 

170 

171 Args: 

172 package_file: Source package file. 

173 

174 Returns: 

175 The files key, or None. 

176 """ 

177 if package_file.package_type == HELM: 

178 return "chart" 

179 if package_file.package_type == PYPI: 

180 return "content" 

181 

182 return super().files_key(package_file) 

183 

184 def _project_package_package_files( 

185 self, package_type: str, package_name: str, package_version: str 

186 ) -> list[ProjectPackageFile]: 

187 project = self._connection.pygitlab.projects.get(self._full_path, lazy=True) 

188 for project_package in project.packages.list( 188 ↛ 202line 188 didn't jump to line 202 because the loop on line 188 didn't complete

189 package_type=package_type, 

190 package_name=package_name, 

191 package_version=package_version, 

192 iterator=True, 

193 ): 

194 if not isinstance(project_package, ProjectPackage): 194 ↛ 195line 194 didn't jump to line 195 because the condition on line 194 was never true

195 raise TypeError 

196 return [ 

197 project_package_file 

198 for project_package_file in project_package.package_files.list(iterator=True) 

199 if isinstance(project_package_file, ProjectPackageFile) 

200 ] 

201 

202 return []