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
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-21 21:33 +0000
1#!/usr/bin/env python3
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
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)
42# Components
43from ..prints.colors import Colors
44from ..system.platform import Platform
45from ..types.gitlab import MigrationEntities, RunnerHelpers
46from ..types.namespaces import Namespaces
48# GitLabFeature class, pylint: disable=too-many-lines,too-many-public-methods
49class GitLabFeature:
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
60 # Members
61 __dry_run: bool = False
62 __gitlab: Gitlab
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:
74 # Initialize members
75 self.__dry_run = dry_run
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 )
96 # Authenticate if available
97 if self.__gitlab.private_token or self.__gitlab.oauth_token:
98 self.__gitlab.auth()
100 # Group
101 def group(
102 self,
103 criteria: str,
104 optional: bool = False,
105 ) -> Group:
107 # Optional group for dry run
108 if self.__dry_run and optional:
109 try:
110 return self.group(criteria, optional=False)
112 # Simulate group
113 except GitlabGetError:
114 GroupTuple = namedtuple('GroupTuple', [
115 'id',
116 'name',
117 'full_path',
118 'projects',
119 ])
121 # pylint: disable=no-self-use,too-few-public-methods
122 class GroupProjects:
123 def list(self, **_kwargs: Optional[Any]) -> List[str]:
124 return []
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 )
141 # Get group
142 return self.__gitlab.groups.get(criteria)
144 # Group create
145 def group_create(
146 self,
147 path: str,
148 name: str,
149 ) -> None:
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 })
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
168 # Group delete
169 def group_delete(
170 self,
171 criteria: str,
172 ) -> None:
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
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
200 # Delay for deletion
201 sleep(3)
203 # Group export
204 def group_export(
205 self,
206 archive: str,
207 criteria: str,
208 reset_entities_string: str = '',
209 ) -> None:
211 # Validate group access
212 if self.__dry_run:
213 return
215 # Create group export
216 group = self.group(criteria)
217 group_export = group.exports.create()
218 sleep(5)
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
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}')
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)
256 # Group export limitations
257 def group_export_limitations(
258 self,
259 criteria: str,
260 ) -> Dict[str, Tuple[str, List[str]]]:
262 # Variables
263 result: Dict[str, Tuple[str, List[str]]] = {}
265 # Get group
266 group = self.group(criteria)
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
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
302 # Result
303 return result
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:
314 # Variables
315 group: Group
316 subgroups_count: int
317 subgroups_last: int = 0
319 # Validate group access
320 if self.__dry_run:
321 return
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 )
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
349 # Group reset members
350 def group_reset_members(
351 self,
352 criteria: str,
353 ) -> None:
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)
361 # Save group
362 if not self.__dry_run:
363 group.save()
365 # Group set avatar
366 def group_set_avatar(
367 self,
368 criteria: str,
369 file: str,
370 ) -> None:
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
378 # Save group
379 group.save()
381 # Group set description
382 def group_set_description(
383 self,
384 criteria: str,
385 description: str,
386 ) -> bool:
388 # Variables
389 changed: bool = False
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
398 # Save group
399 group.save()
401 # Result
402 return changed
404 # Migration entities parser
405 @staticmethod
406 def migration_entities_parse(input_string: str) -> List[str]:
408 # Handle empty input
409 if not input_string:
410 return []
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 ]
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]]:
428 # Variables
429 result_input: str
430 result_output: str
431 differences: List[Tuple[str, str]] = []
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
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
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 )]
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 )]
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 )]
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 )]
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 )]
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 )]
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 )]
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 )]
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 )]
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 )]
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 )]
581 # Result
582 return differences
584 # Namespace
585 def namespace(self, criteria: str, optional: bool = False) -> Namespace:
587 # Optional namespace for dry run
588 if self.__dry_run and optional:
589 try:
590 return self.namespace(criteria, optional=False)
592 # Simulate namespace
593 except GitlabGetError:
594 NamespaceTuple = namedtuple('NamespaceTuple', [
595 'id',
596 'name',
597 'full_path',
598 'kind',
599 ])
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 )
616 # Get namespace
617 return self.__gitlab.namespaces.get(criteria)
619 # Namespace projects
620 def namespace_projects(
621 self,
622 criteria: str,
623 include_subgroups: bool,
624 ) -> Any:
626 # Acquire namespace
627 namespace = self.namespace(criteria, optional=self.__dry_run)
629 # Get user projects
630 if namespace.kind == 'user':
631 user = self.user(criteria)
632 return user.projects.list(get_all=True)
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 )
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 )
651 # Project delete
652 def project_delete(
653 self,
654 criteria: str,
655 ) -> None:
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
673 # Ignore missing project
674 except GitlabGetError:
675 pass
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
689 # Delay for deletion
690 sleep(3)
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:
700 # Validate project access
701 if self.__dry_run:
702 return
704 # Create project export
705 project = self.project(criteria)
706 project_export = project.exports.create()
708 # Verify project export
709 sleep(3)
710 project_export.refresh()
711 if project_export.export_status == 'none':
712 raise RuntimeError(project_export)
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()
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)
742 # Successful project export
743 print(f'{Colors.CYAN}Success'
744 f'{Colors.RESET}')
745 Platform.flush()
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()
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}')
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)
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]]]:
807 # Variables
808 result: Dict[str, Tuple[str, List[str]]] = {}
810 # Get project
811 project = self.project(criteria)
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
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
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
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
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
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
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
910 # Result
911 return result
913 # Project get branch default
914 def project_get_branch_default(
915 self,
916 criteria: str,
917 ) -> str:
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
927 # Result
928 return ''
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:
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 )
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 )
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()
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)
992 # Result
993 return project_import
995 # Project reset members
996 def project_reset_members(
997 self,
998 criteria: str,
999 ) -> None:
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
1010 # Save project
1011 project.save()
1013 # Project set archive
1014 def project_set_archive(
1015 self,
1016 criteria: str,
1017 enabled: bool,
1018 ) -> bool:
1020 # Variables
1021 changed: bool = False
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)
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)
1039 # Result
1040 return changed
1042 # Project set avatar
1043 def project_set_avatar(
1044 self,
1045 criteria: str,
1046 file: str,
1047 ) -> None:
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
1055 # Save project
1056 project.save()
1058 # Project set branch default
1059 def project_set_branch_default(
1060 self,
1061 criteria: str,
1062 default_branch: str,
1063 ) -> bool:
1065 # Variables
1066 changed: bool = False
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
1078 # Save project
1079 if changed:
1080 project.save()
1082 # Result
1083 return changed
1085 # Project set description
1086 def project_set_description(
1087 self,
1088 criteria: str,
1089 description: str,
1090 ) -> bool:
1092 # Variables
1093 changed: bool = False
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
1102 # Save project
1103 project.save()
1105 # Result
1106 return changed
1108 # Subgroup create
1109 def subgroup_create(
1110 self,
1111 parent: str,
1112 path: str,
1113 name: str,
1114 ) -> None:
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 })
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
1134 # URL
1135 @property
1136 def url(self) -> str:
1137 return str(self.__gitlab.api_url)
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}')
1149 # User name
1150 @property
1151 def username(self) -> str:
1152 if self.__gitlab.user:
1153 return str(self.__gitlab.user.username)
1154 return '/'