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

470 statements  

« prev     ^ index     » next       coverage.py v7.15.4, created at 2026-08-21 21:33 +0000

1#!/usr/bin/env python3 

2 

3# Standard libraries 

4from collections import namedtuple 

5import contextlib 

6from json import dump as json_dump, JSONDecodeError, load as json_load 

7from os import remove 

8from os.path import join 

9from pathlib import Path 

10from shutil import make_archive, rmtree, unpack_archive 

11from tempfile import TemporaryDirectory 

12from time import sleep 

13from typing import Any, cast, Dict, List, Optional, Tuple 

14 

15# Modules libraries 

16from gitlab import Gitlab 

17from gitlab.exceptions import ( 

18 GitlabDeleteError, 

19 GitlabGetError, 

20 GitlabHttpError, 

21 GitlabListError, 

22) 

23from gitlab.v4.objects import ( 

24 Group, 

25 Namespace, 

26 Project, 

27 ProjectBadgeManager, 

28 ProjectBranchManager, 

29 ProjectCommitManager, 

30 ProjectImport, 

31 ProjectIssueManager, 

32 ProjectLabelManager, 

33 ProjectMergeRequestManager, 

34 ProjectPackageManager, 

35 ProjectPipelineManager, 

36 ProjectReleaseManager, 

37 ProjectSnippetManager, 

38 ProjectTagManager, 

39 User, 

40) 

41 

42# Components 

43from ..prints.colors import Colors 

44from ..system.platform import Platform 

45from ..types.gitlab import MigrationEntities, RunnerHelpers 

46from ..types.namespaces import Namespaces 

47 

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

49class GitLabFeature: 

50 

51 # Constants 

52 DIFFERENCES_UNAVAILABLE: str = 'Unavailable' 

53 LIMITATIONS_ERROR: str = 'limitations-error' 

54 LIMITATIONS_WARNING: str = 'limitations-warning' 

55 TIMEOUT_CREATION: int = 60 

56 TIMEOUT_DELETION: int = 300 

57 TIMEOUT_EXPORT: int = 300 

58 TIMEOUT_IMPORT: int = 30 

59 

60 # Members 

61 __dry_run: bool = False 

62 __gitlab: Gitlab 

63 

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

65 def __init__( 

66 self, 

67 url: str, 

68 private_token: str, 

69 job_token: str, 

70 ssl_verify: bool | str = True, 

71 dry_run: bool = False, 

72 ) -> None: 

73 

74 # Initialize members 

75 self.__dry_run = dry_run 

76 

77 # Create GitLab client 

78 if private_token: 

79 self.__gitlab = Gitlab( 

80 url=url, 

81 private_token=private_token, 

82 ssl_verify=ssl_verify, 

83 ) 

84 elif job_token: 

85 self.__gitlab = Gitlab( 

86 url=url, 

87 job_token=job_token, 

88 ssl_verify=ssl_verify, 

89 ) 

90 else: 

91 self.__gitlab = Gitlab( 

92 url=url, 

93 ssl_verify=ssl_verify, 

94 ) 

95 

96 # Authenticate if available 

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

98 self.__gitlab.auth() 

99 

100 # Group 

101 def group( 

102 self, 

103 criteria: str, 

104 optional: bool = False, 

105 ) -> Group: 

106 

107 # Optional group for dry run 

108 if self.__dry_run and optional: 

109 try: 

110 return self.group(criteria, optional=False) 

111 

112 # Simulate group 

113 except GitlabGetError: 

114 GroupTuple = namedtuple('GroupTuple', [ 

115 'id', 

116 'name', 

117 'full_path', 

118 'projects', 

119 ]) 

120 

121 # pylint: disable=no-self-use,too-few-public-methods 

122 class GroupProjects: 

123 def list(self, **_kwargs: Optional[Any]) -> List[str]: 

124 return [] 

125 

126 # Create fake group 

127 _, path = Namespaces.split_namespace( 

128 criteria, 

129 relative=False, 

130 ) 

131 return cast( 

132 Group, 

133 GroupTuple( 

134 id='dry-run', 

135 name=path, 

136 full_path=criteria, 

137 projects=GroupProjects(), 

138 ), 

139 ) 

140 

141 # Get group 

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

143 

144 # Group create 

145 def group_create( 

146 self, 

147 path: str, 

148 name: str, 

149 ) -> None: 

150 

151 # Create group 

152 if not self.__dry_run: 

153 self.__gitlab.groups.create({ 

154 'parent_id': None, 

155 'path': path, 

156 'name': name, 

157 }) 

158 

159 # Wait for creation 

160 for _ in range(GitLabFeature.TIMEOUT_CREATION): 

161 sleep(1) 

162 try: 

163 self.group(path) 

164 break 

165 except GitlabGetError: 

166 continue 

167 

168 # Group delete 

169 def group_delete( 

170 self, 

171 criteria: str, 

172 ) -> None: 

173 

174 # Delete group 

175 if not self.__dry_run: 

176 group = self.group(criteria) 

177 group.delete() 

178 sleep(1) 

179 try: 

180 group = self.group(criteria) 

181 group.delete(query_data={ 

182 'full_path': group.full_path, 

183 'permanently_remove': 'true', 

184 }) 

185 except (GitlabDeleteError, GitlabGetError): 

186 pass 

187 

188 # Wait for deletion 

189 for _ in range(GitLabFeature.TIMEOUT_DELETION): 

190 sleep(1) 

191 try: 

192 group = self.group(criteria) 

193 if group.marked_for_deletion_on: 

194 break 

195 except AttributeError: 

196 pass 

197 except GitlabGetError: 

198 break 

199 

200 # Delay for deletion 

201 sleep(3) 

202 

203 # Group export 

204 def group_export( 

205 self, 

206 archive: str, 

207 criteria: str, 

208 reset_entities_string: str = '', 

209 ) -> None: 

210 

211 # Validate group access 

212 if self.__dry_run: 

213 return 

214 

215 # Create group export 

216 group = self.group(criteria) 

217 group_export = group.exports.create() 

218 sleep(5) 

219 

220 # Download group export 

221 for _ in range(GitLabFeature.TIMEOUT_EXPORT): 

222 sleep(1) 

223 try: 

224 with open(archive, 'wb') as file: 

225 group_export.download(streamed=True, action=file.write) 

226 break 

227 except GitlabGetError: 

228 continue 

229 

230 # Parse migration entities 

231 reset_entities: List[str] = GitLabFeature.migration_entities_parse( 

232 reset_entities_string) 

233 reset_entities_names: List[str] = [ 

234 MigrationEntities.get(entity).name for entity in reset_entities 

235 ] 

236 print(f'{Colors.BOLD} - Reset entities: ' 

237 f'{Colors.CYAN}{", ".join(reset_entities_names)}' 

238 f'{Colors.RESET}') 

239 

240 # Reset migration entities 

241 if reset_entities: 

242 with TemporaryDirectory() as temp_directory: 

243 stem = archive 

244 if stem.endswith('.tar.gz'): 

245 stem = stem[:-len('.tar.gz')] 

246 unpack_archive(archive, temp_directory, 'gztar') 

247 for reset_entity in reset_entities: 

248 entity = MigrationEntities.get(reset_entity) 

249 for group_glob in entity.group_globs: 

250 for path in Path(temp_directory).glob(group_glob): 

251 with contextlib.suppress(FileNotFoundError): 

252 path.unlink() 

253 remove(archive) 

254 make_archive(stem, 'gztar', temp_directory) 

255 

256 # Group export limitations 

257 def group_export_limitations( 

258 self, 

259 criteria: str, 

260 ) -> Dict[str, Tuple[str, List[str]]]: 

261 

262 # Variables 

263 result: Dict[str, Tuple[str, List[str]]] = {} 

264 

265 # Get group 

266 group = self.group(criteria) 

267 

268 # Limitations: CI variables 

269 try: 

270 if group.variables.list(get_all=False): 

271 result['Variables'] = ( 

272 GitLabFeature.LIMITATIONS_ERROR, 

273 [variable.key for variable in group.variables.list(get_all=True)], 

274 ) 

275 except GitlabListError: 

276 pass 

277 

278 # Limitations: Runners 

279 try: 

280 if [ 

281 runner.id # 

282 for runner in group.runners.list( 

283 get_all=False, 

284 type='group_type', 

285 ) # 

286 if RunnerHelpers.active(runner) 

287 ]: 

288 result['Runners'] = ( 

289 GitLabFeature.LIMITATIONS_ERROR, 

290 [ 

291 f'#{runner.id}: {runner.description}' 

292 for runner in group.runners.list( 

293 get_all=True, 

294 type='group_type', 

295 ) # 

296 if RunnerHelpers.active(runner) 

297 ], 

298 ) 

299 except GitlabListError: 

300 pass 

301 

302 # Result 

303 return result 

304 

305 # Group import, pylint: disable=too-many-arguments 

306 def group_import( 

307 self, 

308 archive: str, 

309 parent: str, 

310 path: str, 

311 name: str, 

312 ) -> None: 

313 

314 # Variables 

315 group: Group 

316 subgroups_count: int 

317 subgroups_last: int = 0 

318 

319 # Validate group access 

320 if self.__dry_run: 

321 return 

322 

323 # Upload group import 

324 with open(archive, 'rb') as file: 

325 self.__gitlab.groups.import_group( 

326 file, 

327 path=path, 

328 name=name, 

329 parent_id=self.group(parent).id if parent else None, 

330 ) 

331 

332 # Delay group import 

333 for _ in range(GitLabFeature.TIMEOUT_IMPORT): 

334 sleep(3) 

335 try: 

336 group = self.group(f'{parent}/{path}') 

337 subgroups_count = len( 

338 group.descendant_groups.list( # 

339 get_all=True, 

340 # include_subgroups=True, 

341 )) 

342 if subgroups_last == subgroups_count: 

343 break 

344 subgroups_last = subgroups_count 

345 continue 

346 except (GitlabGetError, GitlabListError): 

347 continue 

348 

349 # Group reset members 

350 def group_reset_members( 

351 self, 

352 criteria: str, 

353 ) -> None: 

354 

355 # Remove group members 

356 group = self.group(criteria) 

357 for member in group.members.list(get_all=True): 

358 if not self.__dry_run: 

359 group.members.delete(member.id) 

360 

361 # Save group 

362 if not self.__dry_run: 

363 group.save() 

364 

365 # Group set avatar 

366 def group_set_avatar( 

367 self, 

368 criteria: str, 

369 file: str, 

370 ) -> None: 

371 

372 # Set group avatar 

373 if not self.__dry_run: 

374 group = self.group(criteria) 

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

376 group.avatar = avatar 

377 

378 # Save group 

379 group.save() 

380 

381 # Group set description 

382 def group_set_description( 

383 self, 

384 criteria: str, 

385 description: str, 

386 ) -> bool: 

387 

388 # Variables 

389 changed: bool = False 

390 

391 # Set group description 

392 if not self.__dry_run: 

393 group = self.group(criteria) 

394 if group.description != description: 

395 group.description = description 

396 changed = True 

397 

398 # Save group 

399 group.save() 

400 

401 # Result 

402 return changed 

403 

404 # Migration entities parser 

405 @staticmethod 

406 def migration_entities_parse(input_string: str) -> List[str]: 

407 

408 # Handle empty input 

409 if not input_string: 

410 return [] 

411 

412 # Parse entities from input 

413 return [ 

414 key # Key 

415 for search in input_string.split(',') # Input entities 

416 for key in MigrationEntities.keys() # GitLab entities 

417 if MigrationEntities.get(key).name.lower().startswith(search.strip().lower()) 

418 ] 

419 

420 # Migration projects comparator, pylint: disable=too-many-statements 

421 @staticmethod 

422 def migration_projects_compare( 

423 input_project: Project, 

424 output_project: Project, 

425 reset_entities: str = '', 

426 ) -> List[Tuple[str, str]]: 

427 

428 # Variables 

429 result_input: str 

430 result_output: str 

431 differences: List[Tuple[str, str]] = [] 

432 

433 # Count project items 

434 def count_project_items( 

435 resource: ProjectBadgeManager | ProjectBranchManager 

436 | ProjectCommitManager | ProjectIssueManager | ProjectLabelManager 

437 | ProjectMergeRequestManager | ProjectPackageManager 

438 | ProjectPipelineManager | ProjectReleaseManager 

439 | ProjectSnippetManager | ProjectTagManager, 

440 flag_all: bool = False, 

441 ) -> str: 

442 if not resource: 

443 return '0' 

444 try: 

445 if flag_all: 

446 return str(len(resource.list(get_all=True, all=True))) 

447 return str(len(resource.list(get_all=True))) 

448 except GitlabListError: 

449 try: 

450 if flag_all: 

451 return str(len(resource.list(get_all=True, all=True))) 

452 return str(len(resource.list(get_all=True))) 

453 except GitlabListError: 

454 return GitLabFeature.DIFFERENCES_UNAVAILABLE 

455 

456 # Test missing items 

457 def test_missing_items(result_input: str, result_output: str) -> bool: 

458 if result_input == '0' and result_output == GitLabFeature.DIFFERENCES_UNAVAILABLE: 

459 return False 

460 if result_input in [result_output, GitLabFeature.DIFFERENCES_UNAVAILABLE]: 

461 return False 

462 return True 

463 

464 # Validate badges count 

465 result_input = count_project_items(input_project.badges) 

466 result_output = count_project_items(output_project.badges) 

467 if MigrationEntities.BADGES.name not in reset_entities \ 

468 and test_missing_items(result_input, result_output): 

469 differences += [( 

470 'Badges count', 

471 f'{result_input} badges in input project, ' 

472 f'{result_output} in output project', 

473 )] 

474 

475 # Validate branches count 

476 result_input = count_project_items(input_project.branches) 

477 result_output = count_project_items(output_project.branches) 

478 if test_missing_items(result_input, result_output): 

479 differences += [( 

480 'Branches count', 

481 f'{result_input} branches in input project, ' 

482 f'{result_output} in output project', 

483 )] 

484 

485 # Validate commits count 

486 result_input = count_project_items(input_project.commits, flag_all=True) 

487 result_output = count_project_items(output_project.commits, flag_all=True) 

488 if test_missing_items(result_input, result_output): 

489 differences += [( 

490 'Commits count', 

491 f'{result_input} commits in input project, ' 

492 f'{result_output} in output project', 

493 )] 

494 

495 # Validate issues count 

496 result_input = count_project_items(input_project.issues) 

497 result_output = count_project_items(output_project.issues) 

498 if MigrationEntities.ISSUES.name not in reset_entities \ 

499 and test_missing_items(result_input, result_output): 

500 differences += [( 

501 'Issues count', 

502 f'{result_input} issues in input project, ' 

503 f'{result_output} in output project', 

504 )] 

505 

506 # Validate tags count 

507 result_input = count_project_items(input_project.tags) 

508 result_output = count_project_items(output_project.tags) 

509 if test_missing_items(result_input, result_output): 

510 differences += [( 

511 'Tags count', 

512 f'{result_input} tags in input project, ' 

513 f'{result_output} in output project', 

514 )] 

515 

516 # Validate merge requests count 

517 result_input = count_project_items(input_project.mergerequests) 

518 result_output = count_project_items(output_project.mergerequests) 

519 if MigrationEntities.MERGE_REQUESTS.name not in reset_entities \ 

520 and test_missing_items(result_input, result_output): 

521 differences += [( 

522 'Merge requests count', 

523 f'{result_input} merge requests in input project, ' 

524 f'{result_output} in output project', 

525 )] 

526 

527 # Validate labels count 

528 result_input = count_project_items(input_project.labels) 

529 result_output = count_project_items(output_project.labels) 

530 if MigrationEntities.LABELS.name not in reset_entities \ 

531 and test_missing_items(result_input, result_output): 

532 differences += [( 

533 'Labels count', 

534 f'{result_input} labels in input project, ' 

535 f'{result_output} in output project', 

536 )] 

537 

538 # Validate pipelines count 

539 result_input = count_project_items(input_project.pipelines) 

540 result_output = count_project_items(output_project.pipelines) 

541 if MigrationEntities.PIPELINES.name not in reset_entities \ 

542 and test_missing_items(result_input, result_output): 

543 differences += [( 

544 'Pipelines count', 

545 f'{result_input} pipelines in input project, ' 

546 f'{result_output} in output project', 

547 )] 

548 

549 # Validate packages count 

550 result_input = count_project_items(input_project.packages) 

551 result_output = count_project_items(output_project.packages) 

552 if test_missing_items(result_input, result_output): 

553 differences += [( 

554 'Packages count', 

555 f'{result_input} packages in input project, ' 

556 f'{result_output} in output project', 

557 )] 

558 

559 # Validate releases count 

560 result_input = count_project_items(input_project.releases) 

561 result_output = count_project_items(output_project.releases) 

562 if MigrationEntities.RELEASES.name not in reset_entities \ 

563 and test_missing_items(result_input, result_output): 

564 differences += [( 

565 'Releases count', 

566 f'{result_input} releases in input project, ' 

567 f'{result_output} in output project', 

568 )] 

569 

570 # Validate snippets count 

571 result_input = count_project_items(input_project.snippets) 

572 result_output = count_project_items(output_project.snippets) 

573 if MigrationEntities.SNIPPETS.name not in reset_entities \ 

574 and test_missing_items(result_input, result_output): 

575 differences += [( 

576 'Snippets count', 

577 f'{result_input} snippets in input project, ' 

578 f'{result_output} in output project', 

579 )] 

580 

581 # Result 

582 return differences 

583 

584 # Namespace 

585 def namespace(self, criteria: str, optional: bool = False) -> Namespace: 

586 

587 # Optional namespace for dry run 

588 if self.__dry_run and optional: 

589 try: 

590 return self.namespace(criteria, optional=False) 

591 

592 # Simulate namespace 

593 except GitlabGetError: 

594 NamespaceTuple = namedtuple('NamespaceTuple', [ 

595 'id', 

596 'name', 

597 'full_path', 

598 'kind', 

599 ]) 

600 

601 # Create fake namespace 

602 _, path = Namespaces.split_namespace( 

603 criteria, 

604 relative=False, 

605 ) 

606 return cast( 

607 Namespace, 

608 NamespaceTuple( 

609 id='dry-run', 

610 name=path, 

611 full_path=criteria, 

612 kind='group', 

613 ), 

614 ) 

615 

616 # Get namespace 

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

618 

619 # Namespace projects 

620 def namespace_projects( 

621 self, 

622 criteria: str, 

623 include_subgroups: bool, 

624 ) -> Any: 

625 

626 # Acquire namespace 

627 namespace = self.namespace(criteria, optional=self.__dry_run) 

628 

629 # Get user projects 

630 if namespace.kind == 'user': 

631 user = self.user(criteria) 

632 return user.projects.list(get_all=True) 

633 

634 # Get group projects 

635 return self.group(criteria, optional=self.__dry_run).projects.list( 

636 get_all=True, 

637 include_subgroups=include_subgroups, 

638 ) 

639 

640 # Project 

641 def project( 

642 self, 

643 criteria: str, 

644 statistics: bool = False, 

645 ) -> Project: 

646 return self.__gitlab.projects.get( 

647 criteria, 

648 statistics=statistics, 

649 ) 

650 

651 # Project delete 

652 def project_delete( 

653 self, 

654 criteria: str, 

655 ) -> None: 

656 

657 # Delete project 

658 if not self.__dry_run: 

659 try: 

660 project = self.project(criteria) 

661 project.delete() 

662 sleep(1) 

663 try: 

664 project = self.project(criteria) 

665 project.delete( 

666 query_data={ 

667 'full_path': project.path_with_namespace, 

668 'permanently_remove': 'true', 

669 }) 

670 except (GitlabDeleteError, GitlabGetError): 

671 pass 

672 

673 # Ignore missing project 

674 except GitlabGetError: 

675 pass 

676 

677 # Wait for deletion 

678 for _ in range(GitLabFeature.TIMEOUT_DELETION): 

679 sleep(1) 

680 try: 

681 project = self.project(criteria) 

682 if project.marked_for_deletion_on: 

683 break 

684 except AttributeError: 

685 pass 

686 except GitlabGetError: 

687 break 

688 

689 # Delay for deletion 

690 sleep(3) 

691 

692 # Project export, pylint: disable=too-many-locals,too-many-statements 

693 def project_export( 

694 self, 

695 archive: str, 

696 criteria: str, 

697 reset_entities_string: str = '', 

698 ) -> None: 

699 

700 # Validate project access 

701 if self.__dry_run: 

702 return 

703 

704 # Create project export 

705 project = self.project(criteria) 

706 project_export = project.exports.create() 

707 

708 # Verify project export 

709 sleep(3) 

710 project_export.refresh() 

711 if project_export.export_status == 'none': 

712 raise RuntimeError(project_export) 

713 

714 # Wait project export 

715 print( 

716 f'{Colors.BOLD} - Started project export: ' 

717 f'{Colors.RESET}', 

718 end='', 

719 ) 

720 Platform.flush() 

721 while project_export.export_status in [ 

722 'none', 

723 'queued', 

724 'started', 

725 # 'finished', 

726 'regeneration_in_progress', 

727 # 'failed', 

728 ]: 

729 sleep(1) 

730 project_export.refresh() 

731 

732 # Failed project export 

733 if project_export.export_status in [ 

734 'none', 

735 'failed', 

736 ]: 

737 print(f'{Colors.RED}Failure' 

738 f'{Colors.RESET}') 

739 Platform.flush() 

740 raise RuntimeError(project_export) 

741 

742 # Successful project export 

743 print(f'{Colors.CYAN}Success' 

744 f'{Colors.RESET}') 

745 Platform.flush() 

746 

747 # Download project export 

748 print( 

749 f'{Colors.BOLD} - Downloading project export: ' 

750 f'{Colors.RESET}', 

751 end='', 

752 ) 

753 Platform.flush() 

754 with open(archive, 'wb') as file: 

755 project_export.download(streamed=True, action=file.write) 

756 print(f'{Colors.CYAN}Success' 

757 f'{Colors.RESET}') 

758 Platform.flush() 

759 

760 # Parse migration entities 

761 reset_entities: List[str] = GitLabFeature.migration_entities_parse( 

762 reset_entities_string) 

763 reset_entities_names: List[str] = [ 

764 MigrationEntities.get(entity).name for entity in reset_entities 

765 ] 

766 print(f'{Colors.BOLD} - Reset entities: ' 

767 f'{Colors.CYAN}{", ".join(reset_entities_names)}' 

768 f'{Colors.RESET}') 

769 

770 # Reset migration entities 

771 if reset_entities: 

772 with TemporaryDirectory() as temp_directory: 

773 stem = archive 

774 if stem.endswith('.tar.gz'): 

775 stem = stem[:-len('.tar.gz')] 

776 unpack_archive(archive, temp_directory, 'gztar') 

777 project_json_file = Path(temp_directory) / 'tree' / 'project.json' 

778 project_json_data: Dict[str, Any] 

779 if project_json_file.exists(): 

780 try: 

781 with project_json_file.open('r', encoding='utf-8') as f: 

782 project_json_data = json_load(f) 

783 if project_json_data.get('container_registry_enabled') is None: 

784 project_json_data['container_registry_enabled'] = True 

785 with project_json_file.open('w', encoding='utf-8') as f: 

786 json_dump(project_json_data, f) 

787 except JSONDecodeError: 

788 pass 

789 for reset_entity in reset_entities: 

790 entity = MigrationEntities.get(reset_entity) 

791 for project_directory in entity.project_directories: 

792 with contextlib.suppress(FileNotFoundError): 

793 rmtree(Path(join(temp_directory, project_directory)), 

794 ignore_errors=True) 

795 for project_file in entity.project_files: 

796 with contextlib.suppress(FileNotFoundError): 

797 Path(join(temp_directory, project_file)).unlink() 

798 remove(archive) 

799 make_archive(stem, 'gztar', temp_directory) 

800 

801 # Project export limitations, pylint: disable=too-many-branches 

802 def project_export_limitations( 

803 self, 

804 criteria: str, 

805 ) -> Dict[str, Tuple[str, List[str]]]: 

806 

807 # Variables 

808 result: Dict[str, Tuple[str, List[str]]] = {} 

809 

810 # Get project 

811 project = self.project(criteria) 

812 

813 # Limitations: Packages registry 

814 try: 

815 if project.packages.list(get_all=False): 

816 result['Packages registry'] = ( 

817 GitLabFeature.LIMITATIONS_WARNING, 

818 [ 

819 f'{package.name}:{package.version}' 

820 for package in project.packages.list(get_all=True) 

821 ], 

822 ) 

823 except GitlabListError: 

824 pass 

825 

826 # Limitations: Container registry 

827 try: 

828 if project.repositories.list(get_all=False): 

829 result['Container registry'] = ( 

830 GitLabFeature.LIMITATIONS_ERROR, 

831 [ 

832 f'{container.location}:{container.name if container.name else "latest"}' 

833 for container in project.repositories.list(get_all=True) 

834 ], 

835 ) 

836 except GitlabListError: 

837 pass 

838 

839 # Limitations: CI variables 

840 try: 

841 if project.variables.list(get_all=False): 

842 result['Variables'] = ( 

843 GitLabFeature.LIMITATIONS_ERROR, 

844 [variable.key for variable in project.variables.list(get_all=True)], 

845 ) 

846 except GitlabListError: 

847 pass 

848 

849 # Limitations: Pipeline triggers 

850 try: 

851 if project.triggers.list(get_all=False): 

852 result['Pipeline triggers'] = ( 

853 GitLabFeature.LIMITATIONS_ERROR, 

854 [ 

855 trigger.description 

856 for trigger in project.triggers.list(get_all=True) 

857 ], 

858 ) 

859 except GitlabListError: 

860 pass 

861 

862 # Limitations: Webhooks 

863 try: 

864 if self.__gitlab.http_get(f'/projects/{project.id}/hooks', get_all=True): 

865 result['Webhooks'] = ( 

866 GitLabFeature.LIMITATIONS_ERROR, 

867 [ 

868 hook.get('name', hook.get('url')) # type: ignore[union-attr] 

869 for hook in self.__gitlab.http_get( 

870 f'/projects/{project.id}/hooks', get_all=True) 

871 ], 

872 ) 

873 except GitlabHttpError: 

874 pass 

875 

876 # Limitations: Project Access Tokens 

877 try: 

878 if project.access_tokens.list(get_all=False): 

879 result['Project Access Tokens'] = ( 

880 GitLabFeature.LIMITATIONS_ERROR, 

881 [token.name for token in project.access_tokens.list(get_all=True)], 

882 ) 

883 except GitlabListError: 

884 pass 

885 

886 # Limitations: Runners 

887 try: 

888 if [ 

889 runner.id # 

890 for runner in project.runners.list( 

891 get_all=False, 

892 type='project_type', 

893 ) # 

894 if RunnerHelpers.active(runner) 

895 ]: 

896 result['Runners'] = ( 

897 GitLabFeature.LIMITATIONS_ERROR, 

898 [ 

899 f'#{runner.id}: {runner.description}' 

900 for runner in project.runners.list( 

901 get_all=True, 

902 type='project_type', 

903 ) # 

904 if RunnerHelpers.active(runner) 

905 ], 

906 ) 

907 except GitlabListError: 

908 pass 

909 

910 # Result 

911 return result 

912 

913 # Project get branch default 

914 def project_get_branch_default( 

915 self, 

916 criteria: str, 

917 ) -> str: 

918 

919 # Get default branch 

920 project = self.project(criteria) 

921 try: 

922 if project.default_branch: 

923 return str(project.default_branch) 

924 except AttributeError: 

925 pass 

926 

927 # Result 

928 return '' 

929 

930 # Project import, pylint: disable=too-many-arguments,too-many-positional-arguments 

931 def project_import( 

932 self, 

933 archive: str, 

934 group: str, 

935 path: str, 

936 name: str, 

937 overwrite: bool = False, 

938 sudo: Optional[str] = None, 

939 ) -> ProjectImport: 

940 

941 # Validate project access 

942 if self.__dry_run: 

943 ProjectTuple = namedtuple('ProjectTuple', [ 

944 'id', 

945 'name', 

946 'description', 

947 'path_with_namespace', 

948 ]) 

949 return cast( 

950 ProjectImport, 

951 ProjectTuple( 

952 id='dry-run', 

953 name=name, 

954 description=Namespaces.capitalize( 

955 name, 

956 words=True, 

957 ), 

958 path_with_namespace=path, 

959 ), 

960 ) 

961 

962 # Upload project import 

963 with open(archive, 'rb') as file: 

964 project_imported = self.__gitlab.projects.import_project( 

965 file, 

966 path=path, 

967 name=name, 

968 namespace=group, 

969 overwrite=overwrite, 

970 sudo=sudo, 

971 ) 

972 

973 # Wait project import 

974 project_import = self.__gitlab.projects.get( 

975 project_imported['id'], # type: ignore[index] 

976 lazy=True, 

977 ).imports.get() 

978 while project_import.import_status not in ['finished', 'failed']: 

979 sleep(1) 

980 project_import.refresh() 

981 

982 # Handle failed import 

983 if project_import.import_status == 'failed': 

984 if not project_import.failed_relations \ 

985 and 'key not found: nil' in project_import.import_error: 

986 print(f'{Colors.BOLD} - Legacy GitLab export/import error ignored: ' 

987 f'{Colors.RED}{project_import.import_error}' 

988 f'{Colors.RESET}') 

989 else: 

990 raise RuntimeError(project_import.import_error) 

991 

992 # Result 

993 return project_import 

994 

995 # Project reset members 

996 def project_reset_members( 

997 self, 

998 criteria: str, 

999 ) -> None: 

1000 

1001 # Remove project members 

1002 if not self.__dry_run: 

1003 project = self.project(criteria) 

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

1005 try: 

1006 project.members.delete(member.id) 

1007 except GitlabDeleteError: 

1008 pass 

1009 

1010 # Save project 

1011 project.save() 

1012 

1013 # Project set archive 

1014 def project_set_archive( 

1015 self, 

1016 criteria: str, 

1017 enabled: bool, 

1018 ) -> bool: 

1019 

1020 # Variables 

1021 changed: bool = False 

1022 

1023 # Archive project 

1024 if not self.__dry_run and enabled: 

1025 project = self.project(criteria) 

1026 if not project.archived: 

1027 project.archive() 

1028 changed = True 

1029 sleep(1) 

1030 

1031 # Unarchive project 

1032 elif not self.__dry_run: 

1033 project = self.project(criteria) 

1034 if project.archived: 

1035 project.unarchive() 

1036 changed = True 

1037 sleep(1) 

1038 

1039 # Result 

1040 return changed 

1041 

1042 # Project set avatar 

1043 def project_set_avatar( 

1044 self, 

1045 criteria: str, 

1046 file: str, 

1047 ) -> None: 

1048 

1049 # Set project avatar 

1050 if not self.__dry_run: 

1051 project = self.project(criteria) 

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

1053 project.avatar = avatar 

1054 

1055 # Save project 

1056 project.save() 

1057 

1058 # Project set branch default 

1059 def project_set_branch_default( 

1060 self, 

1061 criteria: str, 

1062 default_branch: str, 

1063 ) -> bool: 

1064 

1065 # Variables 

1066 changed: bool = False 

1067 

1068 # Set default branch 

1069 if not self.__dry_run: 

1070 project = self.project(criteria) 

1071 try: 

1072 if project.default_branch and project.default_branch != default_branch: 

1073 project.default_branch = default_branch 

1074 changed = True 

1075 except AttributeError: 

1076 return changed 

1077 

1078 # Save project 

1079 if changed: 

1080 project.save() 

1081 

1082 # Result 

1083 return changed 

1084 

1085 # Project set description 

1086 def project_set_description( 

1087 self, 

1088 criteria: str, 

1089 description: str, 

1090 ) -> bool: 

1091 

1092 # Variables 

1093 changed: bool = False 

1094 

1095 # Set project description 

1096 if not self.__dry_run: 

1097 project = self.project(criteria) 

1098 if project.description != description: 

1099 project.description = description 

1100 changed = True 

1101 

1102 # Save project 

1103 project.save() 

1104 

1105 # Result 

1106 return changed 

1107 

1108 # Subgroup create 

1109 def subgroup_create( 

1110 self, 

1111 parent: str, 

1112 path: str, 

1113 name: str, 

1114 ) -> None: 

1115 

1116 # Create subgroup 

1117 if not self.__dry_run: 

1118 group = self.group(parent) 

1119 self.__gitlab.groups.create({ 

1120 'parent_id': group.id, 

1121 'path': path, 

1122 'name': name, 

1123 }) 

1124 

1125 # Wait for creation 

1126 for _ in range(GitLabFeature.TIMEOUT_CREATION): 

1127 sleep(1) 

1128 try: 

1129 self.group(path) 

1130 break 

1131 except GitlabGetError: 

1132 continue 

1133 

1134 # URL 

1135 @property 

1136 def url(self) -> str: 

1137 return str(self.__gitlab.api_url) 

1138 

1139 # User 

1140 def user( 

1141 self, 

1142 criteria: str, 

1143 ) -> User: 

1144 users = self.__gitlab.users.list(all=True, iterator=True, username=criteria) 

1145 for user in users: 

1146 return user 

1147 raise RuntimeError(f'Could not find user {criteria}') 

1148 

1149 # User name 

1150 @property 

1151 def username(self) -> str: 

1152 if self.__gitlab.user: 

1153 return str(self.__gitlab.user.username) 

1154 return '/'