Coverage for src/gitlabracadabra/objects/object.py: 68%

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.

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

20import re

21from copy import deepcopy

22from http import HTTPStatus

23from typing import TYPE_CHECKING, ClassVar

25from gitlab.exceptions import GitlabCreateError, GitlabDeleteError, GitlabGetError, GitlabListError, GitlabUpdateError

26from jsonschema.validators import validator_for

28from gitlabracadabra.gitlab.connections import GitlabConnections

30if TYPE_CHECKING:

31 from typing import Any

33 from gitlab import Gitlab

34 from jsonschema.exceptions import ValidationError

36 from gitlabracadabra.gitlab.connection import GitlabConnection

38logger = logging.getLogger(__name__)

41class GitLabracadabraObject:

42 EXAMPLE_YAML_HEADER: ClassVar[str] = "myobject:\n"

43 DOC: ClassVar[list[str]] = []

44 SCHEMA: ClassVar[dict[str, Any]] = {

45 "$schema": "http://json-schema.org/draft-04/schema#",

46 "title": "Object",

47 "type": "object",

48 "properties": {},

49 "additionalProperties": False,

50 # 'required': [],

51 }

53 # If not None, use find(FIND_PARAM=...) instead of get(...)

54 FIND_PARAM: str | None = None

56 # If not None, set to id of the object on create

57 CREATE_KEY: str | None = None

59 CREATE_PARAMS: ClassVar[list[str]] = []

61 IGNORED_PARAMS: ClassVar[list[str]] = []

63 def __init__(self, action_file, name, content):

64 self._action_file = action_file

65 self._name = name

66 self._content = content

67 self._just_created = False

68 validator_class = validator_for(self.SCHEMA)

69 validator_class.check_schema(self.SCHEMA)

70 validator = validator_class(self.SCHEMA)

71 self._errors: list[ValidationError] = sorted(validator.iter_errors(content), key=lambda e: e.path)

72 self._gitlab_id = self._content.pop("gitlab_id", None)

73 self._create_object = self._content.pop("create_object", None)

74 self._delete_object = self._content.pop("delete_object", None)

76 """"web_url()

78 Returns the project's web URL.

79 (Allows to mock the web URL).

80 """

82 def web_url(self) -> str:

83 if isinstance(self._obj.web_url, str): 83 ↛ 85line 83 didn't jump to line 85 because the condition on line 83 was always true

84 return self._obj.web_url

85 raise ValueError

87 @property

88 def connection(self) -> GitlabConnection:

89 return GitlabConnections().get_connection(self._gitlab_id)

91 @property

92 def pygitlab(self) -> Gitlab:

93 return self.connection.pygitlab

95 def errors(self) -> list[ValidationError]:

96 return self._errors

98 """"type_name()

100 GitLabracadabraProject -> project.

101 """

102

103 @classmethod

104 def type_name(cls) -> str:

105 return cls.__name__[15:].lower()

106

107 """"type_name_plural()

108

109 GitLabracadabraProject -> projects.

110 """

111

112 @classmethod

113 def _type_name_plural(cls):

114 return cls.type_name() + "s"

115

116 """"_object_manager()

117

118 Return the python-gitlab Gilab object.

119 """

120

121 def _object_manager(self):

122 return getattr(self.pygitlab, self._type_name_plural())

123

124 """"_create()

125

126 Create the object.

127 """

128

129 def _create(self, *, dry_run: bool = False) -> Any:

130 obj_manager = self._object_manager() # type:ignore

131 namespace_manager = self.pygitlab.namespaces

132 namespaces = self._name.split("/")

133 object_path = namespaces.pop()

134 create_params = {

135 "path": object_path,

136 }

137 if self.CREATE_KEY: 137 ↛ 139line 137 didn't jump to line 139 because the condition on line 137 was always true

138 create_params[self.CREATE_KEY] = object_path

139 for param_name in self.CREATE_PARAMS:

140 if param_name in self._content:

141 create_params[param_name] = self._content[param_name]

142 if len(namespaces):

143 try:

144 parent_namespace = namespace_manager.get("/".join(namespaces))

145 except GitlabGetError as e:

146 error_message = e.error_message

147 if e.response_code == HTTPStatus.NOT_FOUND: 147 ↛ 149line 147 didn't jump to line 149 because the condition on line 147 was always true

148 error_message = "parent namespace not found"

149 logger.error("[%s] NOT Creating %s (%s)", self._name, self.type_name(), error_message)

150 return None

151 if self.type_name() == "group":

152 create_params["parent_id"] = parent_namespace.id

153 else:

154 create_params["namespace_id"] = parent_namespace.id

155 if dry_run: 155 ↛ 156line 155 didn't jump to line 156 because the condition on line 155 was never true

156 logger.info("[%s] NOT Creating %s (dry-run)", self._name, self.type_name())

157 return None

158 logger.info("[%s] Creating %s", self._name, self.type_name())

159 try:

160 return obj_manager.create(create_params)

161 except GitlabCreateError as e:

162 logger.error("[%s] NOT Creating %s (%s)", self._name, self.type_name(), e.error_message)

163 return None

164

165 """"_delete()

166

167 Delete the object.

168 """

169

170 def _delete(self, *, dry_run: bool = False) -> None:

171 if self._obj is None: 171 ↛ 172line 171 didn't jump to line 172 because the condition on line 171 was never true

172 logger.debug("[%s] NOT Deleting %s (not found)", self._name, self.type_name())

173 elif dry_run: 173 ↛ 174line 173 didn't jump to line 174 because the condition on line 173 was never true

174 logger.info("[%s] NOT Deleting %s (dry-run)", self._name, self.type_name())

175 else:

176 logger.info("[%s] Deleting %s", self._name, self.type_name())

177 try:

178 self._obj.delete()

179 except GitlabCreateError as e:

180 logger.error("[%s] Unable to delete %s (%s)", self._name, self.type_name(), e.error_message)

181

182 """"mangle_param()

183

184 Convert a param value from GitLabracadabra form to API form.

185 """

186

187 def mangle_param(

188 self,

189 param_name, # noqa: ARG002

190 param_value,

191 ):

192 return param_value

193

194 """"unmangle_param()

195

196 Convert a param value from API form to GitLabracadabra form.

197 """

198

199 def unmangle_param(self, param_name, param_value):

200 if param_value is None:

201 # Map None values to empty string

202 # (can be overridden with x-gitlabracadabra-none-value in schema)

203 return self.SCHEMA.get("properties", {}).get(param_name).get("x-gitlabracadabra-none-value", "")

204 if isinstance(param_value, str):

205 # GitLab normalize to CRLF

206 # YAML normalize to LF

207 return param_value.replace("\r\n", "\n")

208 return param_value

209

210 """"_canonalize_param()

211

212 Canonalize a param value.

213 """

214

215 def _canonalize_param(

216 self,

217 param_name, # noqa: ARG002

218 param_value,

219 ):

220 if isinstance(param_value, list):

221 return sorted(param_value)

222 return param_value

223

224 """"_get_param()

225

226 Get a param value.

227 """

228

229 def _get_param(self, param_name):

230 return getattr(self._obj, param_name)

231

232 """"_process_param()

233

234 Process one param.

235 """

236

237 def _process_param(self, param_name, param_value, *, dry_run=False, skip_save=False):

238 if param_name in self.IGNORED_PARAMS and not skip_save:

239 return

240 target_value = self._canonalize_param(param_name, param_value)

241 try:

242 current_value = self._canonalize_param(

243 param_name, self.unmangle_param(param_name, self._get_param(param_name))

244 )

245 except AttributeError:

246 if not skip_save:

247 # FIXME: Hidden attributes cannot be idempotent (i.e password)

248 logger.info(

249 "[%s] NOT Changing param %s: %s -> %s (current value is not available)",

250 self._name,

251 param_name,

252 None,

253 target_value,

254 )

255 return

256 current_value = None

257 if current_value != target_value or skip_save:

258 if "dependencies" in self.SCHEMA and param_name in self.SCHEMA["dependencies"]:

259 for dependency in self.SCHEMA["dependencies"][param_name]:

260 process_method = getattr(self, "_process_" + dependency, self._process_param)

261 process_method(dependency, self._content[dependency], dry_run=dry_run, skip_save=True)

262 if dry_run: 262 ↛ 263line 262 didn't jump to line 263 because the condition on line 262 was never true

263 logger.info(

264 "[%s] NOT Changing param %s: %s -> %s (dry-run)",

265 self._name,

266 param_name,

267 current_value,

268 target_value,

269 )

270 setattr(self._obj, param_name, self.mangle_param(param_name, target_value))

271 else:

272 logger.info("[%s] Changing param %s: %s -> %s", self._name, param_name, current_value, target_value)

273 setattr(self._obj, param_name, self.mangle_param(param_name, target_value))

274 if not skip_save:

275 try:

276 self._obj.save()

277 except GitlabUpdateError as e:

278 logger.error(

279 "[%s] Unable to change param %s (%s -> %s): %s",

280 self._name,

281 param_name,

282 current_value,

283 target_value,

284 e.error_message,

285 )

286

287 """"_get()

288

289 Get the _object attribute

290 """

291

292 def _get(self) -> Any:

293 obj_manager = self._object_manager() # type:ignore

294 if self.FIND_PARAM:

295 params = {self.FIND_PARAM: self._name}

296 try:

297 self._obj = obj_manager.list(**params)[0]

298 except IndexError:

299 self._obj = None

300 else:

301 try:

302 self._obj = obj_manager.get(self._name)

303 except GitlabGetError as err:

304 if err.response_code != HTTPStatus.NOT_FOUND: 304 ↛ 305line 304 didn't jump to line 305 because the condition on line 304 was never true

305 pass

306 self._obj = None

307

308 """"process()

309

310 Process the object.

311 """

312

313 def process(self, *, dry_run: bool = False) -> None:

314 content_copy = deepcopy(self._content)

315 self._get()

316 if self._delete_object:

317 self._delete(dry_run=dry_run)

318 return

319 if self._obj is None:

320 if self._create_object:

321 self._obj = self._create(dry_run=dry_run)

322 if self._obj is None:

323 return

324 self._just_created = True

325 else:

326 logger.info("[%s] NOT Creating %s (create_object is false)", self._name, self.type_name())

327 return

328 for param_name, param_value in sorted(self._content.items(), key=self._content_items_sort_key):

329 process_method = getattr(self, "_process_" + param_name, self._process_param)

330 try:

331 process_method(param_name, param_value, dry_run=dry_run)

332 except (GitlabCreateError, GitlabDeleteError, GitlabGetError, GitlabListError, GitlabUpdateError) as e:

333 logger.error("[%s] Error while processing param %s: %s", self._name, param_name, e.error_message)

334 if content_copy != self._content: 334 ↛ 335line 334 didn't jump to line 335 because the condition on line 334 was never true

335 msg = f"[{self._name}] Changed values during processing"

336 raise RuntimeError(msg)

337

338 """"_content_items_sort_key()

339

340 Get content.items() sort key from DOC.

341 """

342

343 def _content_items_sort_key(self, content_item: tuple[str, Any]) -> tuple[int, str]:

344 return (

345 self.SCHEMA.get("properties", {}).get(content_item[0]).get("x-gitlabracadabra-order", 0),

346 content_item[0],

347 )

348

349 """"_markdown_link()

350

351 Generate a Markdown link.

352 """

353

354 @classmethod

355 def _markdown_link(cls, header: str) -> str:

356 # Remove heading #

357 out = re.sub(r"^#+", "", header)

358 # Trim and lower

359 out = out.strip().lower()

360 # Remove non-word characters

361 out = re.sub(r"[^\w\- ]+", " ", out)

362 # Replace multiple spaces with dash

363 out = re.sub(r"\s+", "-", out)

364 # Remove trailing dashed

365 out = re.sub(r"-+$", "", out)

366 return "#" + out

367

368 """"doc_markdown()

369

370 Generate Markdown documentation.

371 """

372

373 @classmethod

374 def doc_markdown(cls) -> str:

375 output = ""

376 properties = cls.SCHEMA.get("properties", {})

377 first_undocumented = True

378 for p in sorted(properties):

379 if p not in cls.DOC:

380 if first_undocumented:

381 cls.DOC.append("# Undocumented")

382 first_undocumented = False

383 cls.DOC.append(p)

384 # Generate TOC

385 output += "# Table of Contents " + "\n\n"

386 for doc in cls.DOC:

387 matches = re.match(r"^(#+) ?(.*)$", doc)

388 if matches:

389 indent = " " * (len(matches.group(1)) - 1)

390 output += indent + "- [" + doc.replace("#", "").strip() + "](" + cls._markdown_link(doc) + ")\n"

391 output += "\n"

392 # Detail

393 for doc in cls.DOC:

394 if doc[0] == "#":

395 output += doc + "\n\n"

396 elif doc in properties:

397 example = cls.EXAMPLE_YAML_HEADER

398 if "_example" in properties[doc]:

399 if properties[doc]["_example"].startswith("\n"):

400 example += " " + doc + ":" + properties[doc]["_example"] + "\n"

401 else:

402 example += " " + doc + ": " + properties[doc]["_example"] + "\n"

403 elif "enum" in properties[doc]:

404 enum = properties[doc]["enum"]

405 example += (

406 " " + doc + ": " + str(enum[0]) + " # one of " + ", ".join([str(item) for item in enum]) + "\n"

407 )

408 elif properties[doc]["type"] == "boolean":

409 example += " " + doc + ": true # or false\n"

410 elif properties[doc]["type"] == "integer":

411 example += " " + doc + ": 42\n"

412 elif properties[doc]["type"] == "array":

413 example += " " + doc + ": [My " + doc.replace("_", " ") + "]\n"

414 else:

415 example += " " + doc + ": My " + doc.replace("_", " ") + "\n"

416 output += (

417 "`" + doc + "` " + properties[doc].get("description", "Undocumented").removesuffix(".") + ":\n"

418 )

419 output += "```yaml\n" + example + "```\n\n"

420 if "_doc_link" in properties[doc]:

421 output += "More information can be found [here]({})\n\n".format(properties[doc]["_doc_link"])

422 else:

423 output += "`" + doc + "` *Undocumented*.\n\n"

424 return output