Coverage for src/features/gitlab.py: 12.17%

1#!/usr/bin/env python3 

2 

3# Standard libraries 

4from time import sleep 

5from typing import Dict, List, Optional 

6 

7# Modules libraries 

8from gitlab import Gitlab 

9from gitlab.const import AccessLevel as GitLabAccessLevel 

10from gitlab.exceptions import ( 

11 GitlabDeleteError, 

12 GitlabGetError, 

13 GitlabJobEraseError, 

14 GitlabListError, 

15) 

16from gitlab.v4.objects import ( 

17 Group, 

18 GroupLabel, 

19 Namespace, 

20 Project, 

21 ProjectBoard, 

22 ProjectBoardList, 

23 ProjectLabel, 

24 User, 

25) 

26 

27# Components 

28from ..types.gitlab import ( 

29 AccessLevels, 

30 ProjectFeatures, 

31 ProtectionLevels, 

32 RolesCreateProjects, 

33 RolesCreateSubgroups, 

34) 

35 

36# GitLabFeature class, pylint: disable=too-many-lines,too-many-public-methods 

37class GitLabFeature: 

38 

39 # Constants 

40 TIMEOUT_DELETION: int = 300 

41 

42 # Members 

43 __dry_run: bool = False 

44 __gitlab: Gitlab 

45 

46 # Constructor, pylint: disable=too-many-arguments,too-many-positional-arguments 

47 def __init__( 

48 self, 

49 url: str, 

50 private_token: str, 

51 job_token: str, 

52 ssl_verify: bool | str = True, 

53 dry_run: bool = False, 

54 ) -> None: 

55 

56 # Initialize members 

57 self.__dry_run = dry_run 

58 

59 # Create GitLab client 

60 if private_token: 

61 self.__gitlab = Gitlab( 

62 url=url, 

63 private_token=private_token, 

64 ssl_verify=ssl_verify, 

65 ) 

66 elif job_token: 

67 self.__gitlab = Gitlab( 

68 url=url, 

69 job_token=job_token, 

70 ssl_verify=ssl_verify, 

71 ) 

72 else: 

73 self.__gitlab = Gitlab( 

74 url=url, 

75 ssl_verify=ssl_verify, 

76 ) 

77 

78 # Authenticate if available 

79 if self.__gitlab.private_token or self.__gitlab.oauth_token: 

80 self.__gitlab.auth() 

81 

82 # Group 

83 def group( 

84 self, 

85 criteria: str, 

86 ) -> Group: 

87 return self.__gitlab.groups.get(criteria) 

88 

89 # Group delete 

90 def group_delete( 

91 self, 

92 criteria: str, 

93 ) -> None: 

94 

95 # Acquire group 

96 group = self.group(criteria) 

97 

98 # Unarchive projects 

99 if not self.__dry_run: 

100 projects = sorted( 

101 group.projects.list( 

102 get_all=True, 

103 with_shared=False, 

104 include_subgroups=True, 

105 order_by='path', 

106 sort='asc', 

107 ), 

108 key=lambda item: item.path_with_namespace, 

109 ) 

110 for project in projects: 

111 if project.archived: 

112 self.project(project.id).unarchive() 

113 

114 # Delete group 

115 if not self.__dry_run: 

116 group.delete() 

117 sleep(1) 

118 try: 

119 group = self.group(criteria) 

120 group.delete(query_data={ 

121 'full_path': group.full_path, 

122 'permanently_remove': 'true', 

123 }) 

124 except (GitlabDeleteError, GitlabGetError): 

125 pass 

126 

127 # Wait for deletion 

128 for _ in range(GitLabFeature.TIMEOUT_DELETION): 

129 sleep(1) 

130 try: 

131 group = self.group(criteria) 

132 if group.marked_for_deletion_on: 

133 break 

134 except AttributeError: 

135 pass 

136 except GitlabGetError: 

137 break 

138 

139 # Delay for deletion 

140 sleep(3) 

141 

142 # Group get labels 

143 def group_get_labels( 

144 self, 

145 criteria: str, 

146 ) -> List[GroupLabel]: 

147 

148 # Get group labels 

149 group = self.group(criteria) 

150 return group.labels.list(get_all=True) 

151 

152 # Group reset members 

153 def group_reset_members( 

154 self, 

155 criteria: str, 

156 ) -> None: 

157 

158 # Remove group members 

159 group = self.group(criteria) 

160 for member in sorted( 

161 group.members.list(get_all=True), 

162 key=lambda item: item.access_level, 

163 ): 

164 if not self.__dry_run: 

165 try: 

166 group.members.delete(member.id) 

167 except GitlabDeleteError as exception: 

168 if member.access_level == GitLabAccessLevel.OWNER \ 

169 and len(group.members.list(get_all=True)) == 1: 

170 continue 

171 raise exception 

172 

173 # Save group 

174 if not self.__dry_run: 

175 group.save() 

176 

177 # Group set avatar 

178 def group_set_avatar( 

179 self, 

180 criteria: str, 

181 file: str, 

182 ) -> None: 

183 

184 # Set group avatar 

185 group = self.group(criteria) 

186 if not self.__dry_run: 

187 with open(file, 'rb') as avatar: 

188 group.avatar = avatar 

189 

190 # Save group 

191 group.save() 

192 

193 # Group set description 

194 def group_set_description( 

195 self, 

196 criteria: str, 

197 description: str, 

198 ) -> bool: 

199 

200 # Variables 

201 changed: bool = False 

202 

203 # Set group description 

204 group = self.group(criteria) 

205 if not self.__dry_run and group.description != description: 

206 group.description = description 

207 changed = True 

208 

209 # Save group 

210 group.save() 

211 

212 # Result 

213 return changed 

214 

215 # Group set label, pylint: disable=too-many-arguments,too-many-positional-arguments 

216 def group_set_label( 

217 self, 

218 criteria: str, 

219 name: str, 

220 description: str, 

221 text_color: str, 

222 color: str, 

223 ) -> GroupLabel: 

224 

225 # Create group label 

226 group = self.group(criteria) 

227 label: GroupLabel 

228 try: 

229 label = group.labels.get(name, include_ancestor_groups=False) 

230 except GitlabGetError: 

231 group.labels.create({ 

232 'name': name, 

233 'description': description, 

234 'text_color': text_color, 

235 'color': color, 

236 }) 

237 label = group.labels.get(name, include_ancestor_groups=False) 

238 

239 # Update group label 

240 label.description = description 

241 label.text_color = text_color 

242 label.color = color 

243 label.save() 

244 

245 # Result 

246 return label 

247 

248 # Group set roles create projects 

249 def group_set_roles_create_projects( 

250 self, 

251 criteria: str, 

252 level: str, 

253 ) -> bool: 

254 

255 # Variables 

256 changed: bool = False 

257 

258 # Validate level 

259 if level not in RolesCreateProjects.names(): 

260 raise SyntaxError( 

261 f'Unknown role level: {level} ({",".join(RolesCreateProjects.names())})') 

262 

263 # Set roles allowed to create projects 

264 group = self.group(criteria) 

265 if not self.__dry_run and group.project_creation_level != level: 

266 group.project_creation_level = level 

267 changed = True 

268 

269 # Save group 

270 group.save() 

271 

272 # Result 

273 return changed 

274 

275 # Group set roles create subgroups 

276 def group_set_roles_create_subgroups( 

277 self, 

278 criteria: str, 

279 level: str, 

280 ) -> bool: 

281 

282 # Variables 

283 changed: bool = False 

284 

285 # Validate level 

286 if level not in RolesCreateSubgroups.names(): 

287 raise SyntaxError( 

288 f'Unknown role level: {level} ({",".join(RolesCreateSubgroups.names())})') 

289 

290 # Set roles allowed to create subgroups 

291 group = self.group(criteria) 

292 if not self.__dry_run and group.subgroup_creation_level != level: 

293 group.subgroup_creation_level = level 

294 changed = True 

295 

296 # Save group 

297 group.save() 

298 

299 # Result 

300 return changed 

301 

302 # Namespace 

303 def namespace( 

304 self, 

305 criteria: str, 

306 ) -> Namespace: 

307 return self.__gitlab.namespaces.get(criteria) 

308 

309 # Project 

310 def project( 

311 self, 

312 criteria: str, 

313 ) -> Project: 

314 return self.__gitlab.projects.get(criteria) 

315 

316 # Project delete 

317 def project_delete( 

318 self, 

319 criteria: str, 

320 ) -> None: 

321 

322 # Acquire project 

323 project = self.project(criteria) 

324 

325 # Delete snippets 

326 try: 

327 snippets = project.snippets.list(all=True) 

328 if snippets: 

329 if project.archived: 

330 project.unarchive() 

331 for snippet in snippets: 

332 snippet.delete() 

333 except GitlabListError: 

334 pass 

335 

336 # Delete project 

337 if not self.__dry_run: 

338 project.delete() 

339 sleep(1) 

340 try: 

341 project = self.project(criteria) 

342 project.delete(query_data={ 

343 'full_path': project.path_with_namespace, 

344 'permanently_remove': 'true', 

345 }) 

346 except (GitlabDeleteError, GitlabGetError): 

347 pass 

348 

349 # Wait for deletion 

350 for _ in range(GitLabFeature.TIMEOUT_DELETION): 

351 sleep(1) 

352 try: 

353 project = self.project(criteria) 

354 if project.marked_for_deletion_on: 

355 break 

356 except AttributeError: 

357 pass 

358 except GitlabGetError: 

359 break 

360 

361 # Delay for deletion 

362 sleep(3) 

363 

364 # Project erase jobs artifacts 

365 def project_erase_jobs_artifacts( 

366 self, 

367 criteria: str, 

368 ) -> None: 

369 

370 # Erase project jobs artifacts 

371 project = self.project(criteria) 

372 if not self.__dry_run: 

373 for job in project.jobs.list(): 

374 if job.artifacts_expire_at is not None: 

375 job.delete_artifacts() 

376 

377 # Project erase jobs artifacts and traces 

378 def project_erase_jobs_contents( 

379 self, 

380 criteria: str, 

381 ) -> None: 

382 

383 # Erase project jobs artifacts and traces 

384 project = self.project(criteria) 

385 if not self.__dry_run: 

386 for job in project.jobs.list(): 

387 if job.artifacts_expire_at is not None: 

388 try: 

389 job.erase() 

390 except GitlabJobEraseError: 

391 pass 

392 

393 # Add path to project CI/CD job token allowlist 

394 def project_job_allowlist_add( 

395 self, 

396 criteria: str, 

397 path: str, 

398 ) -> None: 

399 

400 # Add path to project CI/CD job token allowlist 

401 project = self.project(criteria) 

402 try: 

403 path_project = self.project(path) 

404 if not any(path_project.id == allowlist.get_id() 

405 for allowlist in project.job_token_scope.get().allowlist.list() # 

406 ) and not self.__dry_run: 

407 project.job_token_scope.get().allowlist.create({ 

408 'target_project_id': path_project.id, 

409 }) 

410 except GitlabGetError: 

411 path_group = self.group(path) 

412 if not any(path_group.id == allowlist.get_id() for allowlist in 

413 project.job_token_scope.get().groups_allowlist.list() # 

414 ) and not self.__dry_run: 

415 project.job_token_scope.get().groups_allowlist.create({ 

416 'target_project_id': path_group.id, 

417 }) 

418 

419 # Remove path from project CI/CD job token allowlist 

420 def project_job_allowlist_remove( 

421 self, 

422 criteria: str, 

423 path: str, 

424 ) -> None: 

425 

426 # Remove path from project CI/CD job token allowlist 

427 project = self.project(criteria) 

428 try: 

429 path_project = self.project(path) 

430 if any(path_project.id == allowlist.get_id() 

431 for allowlist in project.job_token_scope.get().allowlist.list() # 

432 ) and not self.__dry_run: 

433 project.job_token_scope.get().allowlist.delete(path_project.id) 

434 except GitlabGetError: 

435 path_group = self.group(path) 

436 if any(path_group.id == allowlist.get_id() for allowlist in 

437 project.job_token_scope.get().groups_allowlist.list() # 

438 ) and not self.__dry_run: 

439 project.job_token_scope.get().groups_allowlist.delete(path_group.id) 

440 

441 # Project protect branches 

442 def project_protect_branches( 

443 self, 

444 criteria: str, 

445 ) -> List[str]: 

446 

447 # Validate project feature 

448 result: List[str] = [] 

449 project = self.project(criteria) 

450 try: 

451 assert project.branches.list(get_all=True) 

452 except (AssertionError, GitlabListError): 

453 return result 

454 

455 # Acquire project, branches and protected branches 

456 branches = [branch.name for branch in project.branches.list(get_all=True)] 

457 protectedbranches = [ 

458 protectedbranch.name 

459 for protectedbranch in project.protectedbranches.list(get_all=True) 

460 ] 

461 

462 # Protect main/master 

463 for branch in ['main', 'master']: 

464 if branch in branches and branch not in protectedbranches: 

465 if not self.__dry_run: 

466 project.protectedbranches.create({ 

467 'name': branch, 

468 'merge_access_level': 40, 

469 'push_access_level': 40, 

470 'allow_force_push': False 

471 }) 

472 result += [branch] 

473 

474 # Protect develop 

475 for branch in ['develop']: 

476 if branch in branches and branch not in protectedbranches: 

477 if not self.__dry_run: 

478 project.protectedbranches.create({ 

479 'name': branch, 

480 'merge_access_level': 40, 

481 'push_access_level': 40, 

482 'allow_force_push': True 

483 }) 

484 result += [branch] 

485 

486 # Protect staging 

487 for branch in ['staging']: 

488 if branch in branches and branch not in protectedbranches: 

489 if not self.__dry_run: 

490 project.protectedbranches.create({ 

491 'name': branch, 

492 'merge_access_level': 30, 

493 'push_access_level': 30, 

494 'allow_force_push': True 

495 }) 

496 result += [branch] 

497 

498 # Save project 

499 if not self.__dry_run: 

500 project.save() 

501 

502 # Result 

503 return result 

504 

505 # Project parse features 

506 @staticmethod 

507 def project_features_parse(input_string: str) -> List[str]: 

508 

509 # Handle empty input 

510 if not input_string: 

511 return [] 

512 

513 # Parse features from input 

514 return [ 

515 key # Key 

516 for search in input_string.split(',') # Input features 

517 for key in ProjectFeatures.keys() # GitLab features 

518 if ProjectFeatures.get(key).name.lower().startswith(search.strip().lower()) 

519 ] 

520 

521 # Project disable features 

522 def project_features_disable( 

523 self, 

524 criteria: str, 

525 features: List[str], 

526 ) -> List[str]: 

527 

528 # Variables 

529 changed: bool 

530 result: List[str] = [] 

531 project = self.__gitlab.projects.get(criteria, statistics=True) 

532 

533 # Iterate through features 

534 for key in features: 

535 if key in ProjectFeatures.keys(): 

536 changed = False 

537 feature = ProjectFeatures.get(key) 

538 

539 # Disable 'access_level' feature 

540 for level in feature.access_level: 

541 if hasattr(project, level.key) \ 

542 and getattr(project, level.key) != AccessLevels.DISABLED: 

543 changed = True 

544 setattr( 

545 project, 

546 level.key, 

547 AccessLevels.DISABLED, 

548 ) 

549 

550 # Disable 'enabled' feature 

551 for subkey in feature.enabled: 

552 if hasattr(project, key) \ 

553 and getattr(project, subkey): 

554 changed = True 

555 setattr( 

556 project, 

557 subkey, 

558 False, 

559 ) 

560 

561 # Add changed feature 

562 if changed: 

563 result.append(feature.name) 

564 

565 # Save project 

566 if not self.__dry_run: 

567 project.save() 

568 

569 # Result 

570 return result 

571 

572 # Project enable features 

573 def project_features_enable( 

574 self, 

575 criteria: str, 

576 features: List[str], 

577 ) -> List[str]: 

578 

579 # Variables 

580 changed: bool 

581 result: List[str] = [] 

582 project = self.__gitlab.projects.get(criteria, statistics=True) 

583 

584 # Iterate through features 

585 for key in features: 

586 if key in ProjectFeatures.keys(): 

587 changed = False 

588 feature = ProjectFeatures.get(key) 

589 

590 # Enable 'access_level' feature 

591 for level in feature.access_level: 

592 if hasattr(project, level.key) \ 

593 and getattr(project, level.key) == AccessLevels.DISABLED: 

594 changed = True 

595 setattr( 

596 project, 

597 level.key, 

598 level.settings[project.visibility], 

599 ) 

600 

601 # Enable 'enabled' feature 

602 for subkey in feature.enabled: 

603 if hasattr(project, subkey) \ 

604 and not getattr(project, subkey): 

605 changed = True 

606 setattr( 

607 project, 

608 subkey, 

609 True, 

610 ) 

611 

612 # Add changed feature 

613 if changed: 

614 result.append(feature.name) 

615 

616 # Save project 

617 if not self.__dry_run: 

618 project.save() 

619 

620 # Result 

621 return result 

622 

623 # Project reset features 

624 def project_features_reset( 

625 self, 

626 criteria: str, 

627 keep_features: List[str], 

628 ) -> List[str]: 

629 

630 # Variables 

631 changed: bool 

632 result: List[str] = [] 

633 project = self.__gitlab.projects.get(criteria, statistics=True) 

634 

635 # Iterate through features 

636 for key in ProjectFeatures.keys(): 

637 if key not in keep_features: 

638 changed = False 

639 feature = ProjectFeatures.get(key) 

640 

641 # Disable 'access_level' feature 

642 for level in feature.access_level: 

643 if changed or (hasattr(project, level.key) \ 

644 and getattr(project, level.key) != AccessLevels.DISABLED \ 

645 and not ProjectFeatures.test(self.__gitlab, project, feature.tests)): 

646 changed = True 

647 setattr( 

648 project, 

649 level.key, 

650 AccessLevels.DISABLED, 

651 ) 

652 

653 # Disable 'enabled' feature 

654 for subkey in feature.enabled: 

655 if changed or (hasattr(project, subkey) \ 

656 and getattr(project, subkey) \ 

657 and not ProjectFeatures.test(self.__gitlab, project, feature.tests)): 

658 changed = True 

659 setattr( 

660 project, 

661 subkey, 

662 False, 

663 ) 

664 

665 # Add changed feature 

666 if changed: 

667 result.append(feature.name) 

668 

669 # Save project 

670 if not self.__dry_run: 

671 project.save() 

672 

673 # Result 

674 return result 

675 

676 # Project get issues boards 

677 def project_get_issues_boards( 

678 self, 

679 criteria: str, 

680 ) -> List[ProjectBoard]: 

681 

682 # Get project issues boards 

683 project = self.project(criteria) 

684 return project.boards.list(get_all=True) 

685 

686 # Project get labels 

687 def project_get_labels( 

688 self, 

689 criteria: str, 

690 ) -> List[ProjectLabel]: 

691 

692 # Get project labels 

693 project = self.project(criteria) 

694 return [ 

695 label for label in project.labels.list(get_all=True) if label.is_project_label 

696 ] 

697 

698 # Project protect tags, pylint: disable=too-many-branches 

699 def project_protect_tags( 

700 self, 

701 criteria: str, 

702 protect_level: str, 

703 ) -> List[str]: 

704 

705 # Validate project feature 

706 result: List[str] = [] 

707 project = self.project(criteria) 

708 try: 

709 assert project.tags.list(get_all=True) 

710 except (AssertionError, GitlabListError): 

711 return result 

712 

713 # Prepare access level 

714 access_level: int 

715 if protect_level == ProtectionLevels.NO_ONE: 

716 access_level = 0 

717 elif protect_level == ProtectionLevels.ADMINS: 

718 access_level = 60 

719 elif protect_level == ProtectionLevels.MAINTAINERS: 

720 access_level = 40 

721 elif protect_level == ProtectionLevels.DEVELOPERS: 

722 access_level = 30 

723 else: 

724 raise SyntaxError(f'Unknown protection level: {protect_level}') 

725 

726 # Acquire protected tags 

727 protectedtags = [ 

728 protectedtag.name for protectedtag in project.protectedtags.list(get_all=True) 

729 ] 

730 

731 # Update protected tags 

732 for protectedtag in project.protectedtags.list(get_all=True): 

733 protectedtag_level = protectedtag.create_access_levels[0]['access_level'] 

734 if protectedtag_level != 0 and (access_level == 0 

735 or protectedtag_level < access_level): 

736 name = protectedtag.name 

737 if not self.__dry_run: 

738 protectedtag.delete() 

739 project.protectedtags.create({ 

740 'name': name, 

741 'create_access_level': access_level 

742 }) 

743 result += [name] 

744 

745 # Protect unprotected tags 

746 for tag in project.tags.list(get_all=True): 

747 if tag.name not in protectedtags: 

748 if not self.__dry_run: 

749 project.protectedtags.create({ 

750 'name': tag.name, 

751 'create_access_level': access_level 

752 }) 

753 result += [tag.name] 

754 

755 # Save project 

756 if not self.__dry_run: 

757 project.save() 

758 

759 # Result 

760 result.sort() 

761 return result 

762 

763 # Project prune unreachable objects 

764 def project_prune_unreachable_objects( 

765 self, 

766 criteria: str, 

767 ) -> None: 

768 

769 # Prune project unreachable objects 

770 project = self.project(criteria) 

771 if not self.__dry_run: 

772 project.housekeeping(post_data={'task': 'prune'}) 

773 

774 # Project reset members 

775 def project_reset_members( 

776 self, 

777 criteria: str, 

778 ) -> None: 

779 

780 # Remove project members 

781 project = self.project(criteria) 

782 if not self.__dry_run: 

783 for member in project.members.list(get_all=True): 

784 try: 

785 project.members.delete(member.id) 

786 except GitlabDeleteError: 

787 pass 

788 

789 # Save project 

790 project.save() 

791 

792 # Project run housekeeping 

793 def project_run_housekeeping( 

794 self, 

795 criteria: str, 

796 ) -> None: 

797 

798 # Run project housekeeping 

799 project = self.project(criteria) 

800 if not self.__dry_run: 

801 project.housekeeping() 

802 

803 # Project set archive 

804 def project_set_archive( 

805 self, 

806 criteria: str, 

807 enabled: bool, 

808 ) -> bool: 

809 

810 # Variables 

811 changed: bool = False 

812 

813 # Archive project 

814 project = self.project(criteria) 

815 if not self.__dry_run and enabled: 

816 if not project.archived: 

817 project.archive() 

818 changed = True 

819 

820 # Unarchive project 

821 elif not self.__dry_run: 

822 if project.archived: 

823 project.unarchive() 

824 changed = True 

825 

826 # Result 

827 return changed 

828 

829 # Project set attribute 

830 def project_set_attribute( 

831 self, 

832 criteria: str, 

833 name: str, 

834 value: bool | str, 

835 ) -> Optional[bool]: 

836 

837 # Variables 

838 changed: bool = False 

839 

840 # Check project attribute 

841 project = self.project(criteria) 

842 if not self.__dry_run and not hasattr(project, name): 

843 return None 

844 

845 # Set project attribute 

846 if not self.__dry_run and getattr(project, name) != value: 

847 setattr(project, name, value) 

848 changed = True 

849 

850 # Save project 

851 project.save() 

852 

853 # Result 

854 return changed 

855 

856 # Project set avatar 

857 def project_set_avatar( 

858 self, 

859 criteria: str, 

860 file: str, 

861 ) -> None: 

862 

863 # Set project avatar 

864 project = self.project(criteria) 

865 if not self.__dry_run: 

866 with open(file, 'rb') as avatar: 

867 project.avatar = avatar 

868 

869 # Save project 

870 project.save() 

871 

872 # Project set description 

873 def project_set_description( 

874 self, 

875 criteria: str, 

876 description: str, 

877 ) -> bool: 

878 

879 # Variables 

880 changed: bool = False 

881 

882 # Set project description 

883 project = self.project(criteria) 

884 if not self.__dry_run and project.description != description: 

885 project.description = description 

886 changed = True 

887 

888 # Save project 

889 project.save() 

890 

891 # Result 

892 return changed 

893 

894 # Project set issues_board, pylint: disable=too-many-arguments,too-many-positional-arguments 

895 def project_set_issues_board( 

896 self, 

897 criteria: str, 

898 name: str, 

899 hide_backlog_list: bool, 

900 hide_closed_list: bool, 

901 lists: List[Dict[str, str]], 

902 ) -> ProjectBoard: 

903 

904 # Create project issues board