refactor: migrate branches configuration feature to use python-gitlab…

… library

closes #625

gitlabform/processors/project/files_processor.py

@@ -171,6 +171,7 @@ def modify_file_dealing_with_branch_protection(
     ):
         # perhaps your user permissions are ok to just perform this operation regardless
         # of the branch protection...
+        project = self.gl.projects.get(project_and_group)
 
         try:
             self.just_modify_file(
@@ -189,7 +190,8 @@ def modify_file_dealing_with_branch_protection(
                     debug(
                         f"> Temporarily unprotecting the branch to {operation} a file in it..."
                     )
-                    self.branch_protector.unprotect_branch(project_and_group, branch)
+
+                    self.branch_protector.unprotect_branch(project, branch)
                 else:
                     fatal(
                         f"Operation {operation} on file {file} in branch {branch} not permitted,"
@@ -212,9 +214,10 @@ def modify_file_dealing_with_branch_protection(
                 finally:
                     # ...and protect the branch again after the operation
                     if configuration.get("branches|" + branch + "|protected"):
+                        branch_configuration = configuration["branches"][branch]
                         debug("> Protecting the branch again.")
                         self.branch_protector.protect_branch(
-                            project_and_group, configuration, branch
+                            project, branch_configuration, branch
                         )
 
             else:

gitlabform/processors/util/branch_protector.py

@@ -1,9 +1,10 @@
 from logging import debug
 from cli_ui import warning, fatal
+from gitlab import Gitlab, GitlabGetError, GitlabDeleteError
+from gitlab.v4.objects import Project, ProjectProtectedBranch
 
 from gitlabform.constants import EXIT_PROCESSING_ERROR, EXIT_INVALID_INPUT
-from gitlabform.gitlab import GitLab
-from gitlabform.gitlab.core import NotFoundException
+from gitlabform.gitlab import GitLab, GitlabWrapper
 
 
 class BranchProtector:
@@ -18,74 +19,56 @@ class BranchProtector:
     ]
 
     def __init__(self, gitlab: GitLab, strict: bool):
-        self.gitlab = gitlab
         self.strict = strict
+        if gitlab:
+            self.gl: Gitlab = GitlabWrapper(gitlab)._gitlab
 
     def apply_branch_protection_configuration(
         self, project_and_group, configuration, branch
     ):
-        try:
-            requested_configuration = configuration["branches"][branch]
-
-            if requested_configuration.get("protected"):
-                self.protect_branch(project_and_group, configuration, branch)
-            else:
-                self.unprotect_branch(project_and_group, branch)
+        requested_configuration = configuration["branches"][branch]
+        project = self.gl.projects.get(project_and_group)
 
-        except NotFoundException:
-            message = f"Branch '{branch}' not found when trying to set it as protected/unprotected!"
-            if self.strict:
-                fatal(
-                    message,
-                    exit_code=EXIT_PROCESSING_ERROR,
-                )
-            else:
-                warning(message)
+        if requested_configuration.get("protected"):
+            self.protect_branch(project, requested_configuration, branch)
+        else:
+            self.unprotect_branch(project, branch)
 
-    def protect_branch(self, project_and_group, configuration, branch):
-        try:
-            requested_configuration = configuration["branches"][branch]
+    def protect_branch(self, project: Project, branch_configuration, branch):
+        self.validate_branch_protection_config(project, branch_configuration, branch)
 
-            self.validate_branch_protection_config(
-                project_and_group, requested_configuration, branch
+        if any(
+            extra_key in branch_configuration for extra_key in self.extra_param_keys
+        ):
+            for extra_param_key in self.extra_param_keys:
+                # check if an extra_param is in config, and it contains the user parameter
+                if extra_param_key in branch_configuration and any(
+                    "user" in d for d in branch_configuration[extra_param_key]
+                ):
+                    for extra_config in branch_configuration[extra_param_key]:
+                        # loop over the array of extra param and get the user_id related to user
+                        if "user" in extra_config.keys():
+                            user_id = self._get_user_id_by_name(
+                                extra_config.pop("user")
+                            )
+                            extra_config["user_id"] = user_id
+
+        if self.configuration_update_needed(branch_configuration, project, branch):
+            self.do_protect_branch(branch_configuration, project, branch)
+        else:
+            debug(
+                "Skipping setting branch '%s' protection configuration because it's already as requested.",
+                branch,
             )
 
-            if any(
-                extra_key in requested_configuration
-                for extra_key in self.extra_param_keys
-            ):
-                for extra_param_key in self.extra_param_keys:
-                    # check if an extra_param is in config, and it contains the user parameter
-                    if extra_param_key in requested_configuration and any(
-                        "user" in d for d in requested_configuration[extra_param_key]
-                    ):
-                        for extra_config in requested_configuration[extra_param_key]:
-                            # loop over the array of extra param and get the user_id related to user
-                            if "user" in extra_config.keys():
-                                user_id = self.gitlab._get_user_id(
-                                    extra_config.pop("user")
-                                )
-                                extra_config["user_id"] = user_id
-
-            if self.configuration_update_needed(
-                requested_configuration, project_and_group, branch
-            ):
-                self.do_protect_branch(
-                    requested_configuration, project_and_group, branch
-                )
-            else:
-                debug(
-                    "Skipping setting branch '%s' protection configuration because it's already as requested.",
-                    branch,
-                )
-
-            if "code_owner_approval_required" in requested_configuration:
-                self.set_code_owner_approval_required(
-                    requested_configuration, project_and_group, branch
-                )
+        if "code_owner_approval_required" in branch_configuration:
+            self.set_code_owner_approval_required(branch_configuration, project, branch)
 
-        except NotFoundException:
-            message = f"Branch '{branch}' not found when trying to set it as protected/unprotected!"
+    def unprotect_branch(self, project: Project, branch):
+        try:
+            project.branches.get(branch)
+        except GitlabGetError:
+            message = f"Branch '{branch}' could not be found in project '{project.path_with_namespace}!"
             if self.strict:
                 fatal(
                     message,
@@ -94,23 +77,15 @@ def protect_branch(self, project_and_group, configuration, branch):
             else:
                 warning(message)
 
-    def unprotect_branch(self, project_and_group, branch):
         try:
             debug("Setting branch '%s' as unprotected", branch)
-            self.gitlab.unprotect_branch(project_and_group, branch)
-
-        except NotFoundException:
-            message = f"Branch '{branch}' not found when trying to set it as protected/unprotected!"
-            if self.strict:
-                fatal(
-                    message,
-                    exit_code=EXIT_PROCESSING_ERROR,
-                )
-            else:
-                warning(message)
+            project.protectedbranches.delete(branch)
+        except GitlabDeleteError:
+            message = f"Branch '{branch}' could not be set to unprotected!"
+            warning(message)
 
     def validate_branch_protection_config(
-        self, project_and_group, requested_configuration, branch
+        self, project: Project, requested_configuration, branch
     ):
         # for the new API any key needs to be defined...
         if any(
@@ -120,30 +95,33 @@ def validate_branch_protection_config(
             return
         else:
             fatal(
-                f"Invalid configuration for protecting branches in project '{project_and_group}',"
+                f"Invalid configuration for protecting branches in project '{project.path_with_namespace}',"
                 f" branch '{branch}' - missing keys. Required is any of these: "
                 f"{self.new_api_keys + self.extra_param_keys}",
                 exit_code=EXIT_INVALID_INPUT,
             )
 
-    def do_protect_branch(self, requested_configuration, project_and_group, branch):
+    def do_protect_branch(self, requested_configuration, project: Project, branch):
         debug("Setting branch '%s' access level", branch)
 
-        # Protected Branches API is one of those that do not support editing entities
-        # (PUT is not documented for it, at least). so you need to delete existing
-        # branch protection (DELETE) and recreate it (POST) to perform an update
-        # (otherwise you get HTTP 409 "Protected branch 'foo' already exists")
-        self.gitlab.unprotect_branch(project_and_group, branch)
+        try:
+            # Protected Branches API is one of those that do not support editing entities
+            # (PUT is not documented for it, at least). so you need to delete existing
+            # branch protection (DELETE) and recreate it (POST) to perform an update
+            # (otherwise you get HTTP 409 "Protected branch 'foo' already exists")
+            project.protectedbranches.delete(branch)
+        except GitlabDeleteError:
+            debug("Branch already unprotected")
 
         # replace in our config our custom "user" and "group" entries with supported by
         # the Protected Branches API "user_id" and "group_id"
         for extra_param_key in self.extra_param_keys:
             for element in requested_configuration.get(extra_param_key, []):
                 if "user" in element.keys():
-                    user_id = self.gitlab._get_user_id(element.pop("user"))
+                    user_id = self._get_user_id_by_name(element.pop("user"))
                     element["user_id"] = user_id
                 elif "group" in element.keys():
-                    group_id = self.gitlab._get_group_id(element.pop("group"))
+                    group_id = self._get_group_id_by_name(element.pop("group"))
                     element["group_id"] = group_id
 
         protect_rules = {
@@ -152,30 +130,45 @@ def do_protect_branch(self, requested_configuration, project_and_group, branch):
             if key != "protected"
         }
 
-        self.gitlab.protect_branch(
-            project_and_group,
-            branch,
-            protect_rules,
-        )
+        project.protectedbranches.create({"name": branch, **protect_rules})
 
     def set_code_owner_approval_required(
-        self, requested_configuration, project_and_group, branch
+        self, requested_configuration, project: Project, branch: str
     ):
         debug(
             "Setting branch '%s' \"code owner approval required\" option",
             branch,
         )
-        self.gitlab.set_branch_code_owner_approval_required(
-            project_and_group,
-            branch,
-            requested_configuration["code_owner_approval_required"],
-        )
+
+        code_owner_approval_required = requested_configuration[
+            "code_owner_approval_required"
+        ]
+
+        try:
+            protected_branch: ProjectProtectedBranch = project.protectedbranches.get(
+                branch
+            )
+
+            if protected_branch:
+                protected_branch.code_owner_approval_required = (
+                    code_owner_approval_required
+                )
+                protected_branch.save()
+        except GitlabGetError:
+            message = f"Protected Branch '{branch}' not found when trying to set code_owner_approval_required!"
+            if self.strict:
+                fatal(
+                    message,
+                    exit_code=EXIT_PROCESSING_ERROR,
+                )
+            else:
+                warning(message)
 
     def configuration_update_needed(
-        self, requested_configuration, project_and_group, branch
+        self, requested_configuration, project: Project, branch
     ):
         # get current configuration of branch access level
-        current = self.get_current_branch_configuration(project_and_group, branch)
+        current = self.get_current_branch_configuration(project, branch)
 
         # get requested configuration of branch access level
         requested = self.get_requested_branch_configuration(requested_configuration)
@@ -184,7 +177,7 @@ def configuration_update_needed(
         debug("Requested branch '%s' access levels: %s", branch, requested)
         return current != requested
 
-    def get_current_branch_configuration(self, project_and_group_name, branch):
+    def get_current_branch_configuration(self, project: Project, branch):
         # from a response for a request to Protected Branches API GET request
         # (https://docs.gitlab.com/ee/api/protected_branches.html#get-a-single-protected-branch-or-wildcard-protected-branch)
         # like:
@@ -212,10 +205,8 @@ def get_current_branch_configuration(self, project_and_group_name, branch):
         # the other extra settings that can be applied using Protected Branches API POST request
         #
         try:
-            protected_branches_response = self.gitlab.get_branch_access_levels(
-                project_and_group_name, branch
-            )
-
+            protected_branch = project.protectedbranches.get(branch)
+            protected_branches_response = protected_branch.attributes
             return (
                 *(
                     self.get_current_permissions(protected_branches_response, "push")
@@ -231,10 +222,11 @@ def get_current_branch_configuration(self, project_and_group_name, branch):
                 protected_branches_response.get("allow_force_push"),
                 # code_owner_approval_required has to use PATCH request, see set_code_owner_approval_required()
             )
-        except NotFoundException:
+        except GitlabGetError:
             return tuple([None] * 10)  # = 3 * 3 + 1
 
-    def get_current_permissions(self, protected_branches_response, action):
+    @staticmethod
+    def get_current_permissions(protected_branches_response: dict, action):
         # from a response for a request to Protected Branches API GET request
         # (https://docs.gitlab.com/ee/api/protected_branches.html#get-a-single-protected-branch-or-wildcard-protected-branch)
         # like:
@@ -344,10 +336,18 @@ def get_requested_permissions(self, requested_configuration, action):
                 elif "user_id" in config:
                     user_ids.add(config["user_id"])
                 elif "user" in config:
-                    user_ids.add(self.gitlab._get_user_id(config["user"]))
+                    user_id = self._get_user_id_by_name(config["user"])
+                    user_ids.add(user_id)
                 elif "group_id" in config:
                     group_ids.add(config["group_id"])
                 elif "group" in config:
-                    group_ids.add(self.gitlab._get_group_id(config["group"]))
+                    group_id = self._get_group_id_by_name(config["group"])
+                    group_ids.add(group_id)
 
         return sorted(levels), sorted(user_ids), sorted(group_ids)
+
+    def _get_group_id_by_name(self, group_name):
+        return self.gl.groups.list(search=group_name).id
+
+    def _get_user_id_by_name(self, username):
+        return self.gl.users.username(username=username)[0].id