Generalize class APIs (#1535)

* SEARCH API * CREATE API * DELETE API * Class API pattern * Class API pattern * Class API SEARCH * Class CREATE API * Class UPDATE API * Remove unused symbols * Formatting * Remove useless imports * Class API * Formatting * Class API * Class API * Class API * Order as CRUDL * Class API * Class API * Class API * Class API * Class API * Use SET/GET_TAGS * Use SET/GET_TAGS * Remove unused constant * Class API * Class API * Class API * Class API * Remove SEARCH_API symbol * Remove SEARCH_API * Formatting * Fixes * Review docstrings * Quality pass * Fix * Quality pass
okorach · Dec 22, 2024 · 3dd90f8 · 3dd90f8
1 parent 224a0e4
commit 3dd90f8
Show file tree

Hide file tree

Showing 22 changed files with 192 additions and 209 deletions.
diff --git a/sonar/app_branches.py b/sonar/app_branches.py
@@ -50,8 +50,8 @@ class ApplicationBranch(Component):
     API = {
         c.CREATE: "applications/create_branch",
         c.GET: "applications/show",
-        c.DELETE: "applications/delete_branch",
         c.UPDATE: "applications/update_branch",
+        c.DELETE: "applications/delete_branch",
     }
 
     def __init__(self, app: object, name: str, project_branches: list[Branch], is_main: bool = False) -> None:

diff --git a/sonar/applications.py b/sonar/applications.py
@@ -54,7 +54,6 @@ class Application(aggr.Aggregation):
 
     CACHE = cache.Cache()
 
-    SEARCH_API = "components/search_projects"
     SEARCH_KEY_FIELD = "key"
     SEARCH_RETURN_FIELD = "components"
     API = {

diff --git a/sonar/branches.py b/sonar/branches.py
@@ -29,7 +29,7 @@
 import requests.utils
 
 from sonar import platform
-from sonar.util import types, cache
+from sonar.util import types, cache, constants as c
 import sonar.logging as log
 import sonar.sqobject as sq
 from sonar import components, settings, exceptions, tasks
@@ -40,14 +40,6 @@
 from sonar.audit.rules import get_rule, RuleId
 
 
-#: APIs used for branch management
-APIS = {
-    "list": "project_branches/list",
-    "rename": "project_branches/rename",
-    "get_new_code": "new_code_periods/list",
-    "delete": "project_branches/delete",
-}
-
 _UNSUPPORTED_IN_CE = "Branches not available in Community Edition"
 
 
@@ -57,6 +49,12 @@ class Branch(components.Component):
     """
 
     CACHE = cache.Cache()
+    API = {
+        c.LIST: "project_branches/list",
+        c.DELETE: "project_branches/delete",
+        "rename": "project_branches/rename",
+        "get_new_code": "new_code_periods/list",
+    }
 
     def __init__(self, project: projects.Project, name: str) -> None:
         """Don't use this, use class methods to create Branch objects
@@ -92,7 +90,7 @@ def get_object(cls, concerned_object: projects.Project, branch_name: str) -> Bra
         if o:
             return o
         try:
-            data = json.loads(concerned_object.get(APIS["list"], params={"project": concerned_object.key}).text)
+            data = json.loads(concerned_object.get(Branch.API[c.LIST], params={"project": concerned_object.key}).text)
         except (ConnectionError, RequestException) as e:
             util.handle_error(e, f"searching {str(concerned_object)} for branch '{branch_name}'", catch_http_statuses=(HTTPStatus.NOT_FOUND,))
             raise exceptions.ObjectNotFound(concerned_object.key, f"{str(concerned_object)} not found")
@@ -140,7 +138,7 @@ def refresh(self) -> Branch:
         :rtype: Branch
         """
         try:
-            data = json.loads(self.get(APIS["list"], params={"project": self.concerned_object.key}).text)
+            data = json.loads(self.get(Branch.API[c.LIST], params={"project": self.concerned_object.key}).text)
         except (ConnectionError, RequestException) as e:
             util.handle_error(e, f"refreshing {str(self)}", catch_http_statuses=(HTTPStatus.NOT_FOUND,))
             Branch.CACHE.pop(self)
@@ -189,7 +187,7 @@ def delete(self) -> bool:
         :rtype: bool
         """
         try:
-            return sq.delete_object(self, APIS["delete"], {"branch": self.name, "project": self.concerned_object.key}, Branch.CACHE)
+            return sq.delete_object(self, Branch.API[c.DELETE], {"branch": self.name, "project": self.concerned_object.key}, Branch.CACHE)
         except (ConnectionError, RequestException) as e:
             util.handle_error(e, f"deleting {str(self)}", catch_all=True)
             if isinstance(e, HTTPError) and e.response.status_code == HTTPStatus.BAD_REQUEST:
@@ -205,7 +203,7 @@ def new_code(self) -> str:
             self._new_code = settings.new_code_to_string({"inherited": True})
         elif self._new_code is None:
             try:
-                data = json.loads(self.get(api=APIS["get_new_code"], params={"project": self.concerned_object.key}).text)
+                data = json.loads(self.get(api=Branch.API["get_new_code"], params={"project": self.concerned_object.key}).text)
             except (ConnectionError, RequestException) as e:
                 util.handle_error(e, f"getting new code period of {str(self)}", catch_http_statuses=(HTTPStatus.NOT_FOUND,))
                 Branch.CACHE.pop(self)
@@ -267,7 +265,7 @@ def rename(self, new_name: str) -> bool:
             return False
         log.info("Renaming main branch of %s from '%s' to '%s'", str(self.concerned_object), self.name, new_name)
         try:
-            self.post(APIS["rename"], params={"project": self.concerned_object.key, "name": new_name})
+            self.post(Branch.API["rename"], params={"project": self.concerned_object.key, "name": new_name})
         except (ConnectionError, RequestException) as e:
             util.handle_error(e, f"Renaming {str(self)}", catch_http_statuses=(HTTPStatus.NOT_FOUND, HTTPStatus.BAD_REQUEST))
             if isinstance(e, HTTPError):
@@ -405,7 +403,7 @@ def get_list(project: projects.Project) -> dict[str, Branch]:
         raise exceptions.UnsupportedOperation(_UNSUPPORTED_IN_CE)
 
     log.debug("Reading all branches of %s", str(project))
-    data = json.loads(project.endpoint.get(APIS["list"], params={"project": project.key}).text)
+    data = json.loads(project.endpoint.get(Branch.API[c.LIST], params={"project": project.key}).text)
     return {branch["name"]: Branch.load(project, branch["name"], data=branch) for branch in data.get("branches", {})}
 
 

diff --git a/sonar/components.py b/sonar/components.py
@@ -43,15 +43,15 @@
 KEY_SEPARATOR = " "
 
 _ALT_COMPONENTS = ("project", "application", "portfolio")
-SEARCH_API = "components/search"
-_DETAILS_API = "components/show"
 
 
 class Component(sq.SqObject):
     """
     Abstraction of the Sonar component concept
     """
 
+    API = {c.SEARCH: "components/search"}
+
     def __init__(self, endpoint: pf.Platform, key: str, data: types.ApiPayload = None) -> None:
         """Constructor"""
         super().__init__(endpoint=endpoint, key=key)

diff --git a/sonar/dce/app_nodes.py b/sonar/dce/app_nodes.py
@@ -49,7 +49,7 @@ def plugins(self) -> Optional[dict[str, str]]:
         return self.json.get("Plugins", None)
 
     def health(self) -> str:
-        """Returns app node health, RED by default if heLTH NOT AVAILABLE"""
+        """Returns app node health, RED by default if health not available"""
         return self.json.get("Health", dce_nodes.HEALTH_RED)
 
     def node_type(self) -> str:
@@ -68,6 +68,7 @@ def version(self) -> Union[tuple[int, ...], None]:
             return None
 
     def edition(self) -> str:
+        """Returns the node edition"""
         return self.sif.edition()
 
     def name(self) -> str:
@@ -109,11 +110,10 @@ def __audit_official(self) -> list[Problem]:
 def audit(sub_sif: dict[str, str], sif_object: object, audit_settings: types.ConfigSettings) -> list[Problem]:
     """Audits application nodes of a DCE instance
 
-    :param dict sub_sif: The JSON subsection of the SIF pertaining to the App Nodes
+    :param sub_sif: The JSON subsection of the SIF pertaining to the App Nodes
     :param Sif sif_object: The Sif object
-    :param ConfigSettings audit_settings: Config settings for audit
+    :param audit_settings: Config settings for audit
     :return: List of Problems
-    :rtype: list[Problem]
     """
     nodes = []
     problems = []

diff --git a/sonar/dce/search_nodes.py b/sonar/dce/search_nodes.py
@@ -23,7 +23,7 @@
 
 """
 
-from typing import Union
+from typing import Optional
 
 import sonar.logging as log
 from sonar.util import types
@@ -61,7 +61,8 @@ def audit(self, audit_settings: types.ConfigSettings) -> list[Problem]:
         log.info("%s: Auditing...", str(self))
         return self.__audit_store_size() + self.__audit_available_disk()
 
-    def max_heap(self) -> Union[int, None]:
+    def max_heap(self) -> Optional[int]:
+        """Returns the node max heap or None if not found"""
         if self.sif.edition() != "datacenter" and self.sif.version() < (9, 0, 0):
             return util.jvm_heap(self.sif.search_jvm_cmdline())
         try:
@@ -72,7 +73,7 @@ def max_heap(self) -> Union[int, None]:
         return int(float(sz.split(" ")[0]) * 1024)
 
     def __audit_store_size(self) -> list[Problem]:
-        """Auditing the search node store size vs heap allocated to ES"""
+        """Audits the search node store size vs heap allocated to ES"""
         log.info("%s: Auditing store size", str(self))
         es_heap = self.max_heap()
         if es_heap is None:
@@ -97,19 +98,15 @@ def __audit_store_size(self) -> list[Problem]:
         return es_pb
 
     def __audit_available_disk(self) -> list[Problem]:
+        """Audits whether the node has enough free disk space"""
         log.info("%s: Auditing available disk space", str(self))
         try:
             space_avail = util.int_memory(self.json[_ES_STATE]["Disk Available"])
         except ValueError:
             log.warning("%s: disk space available not found in SIF, skipping this check", str(self))
             return []
         store_size = self.store_size()
-        log.info(
-            "%s: Search server available disk size of %d MB and store size is %d MB",
-            str(self),
-            space_avail,
-            store_size,
-        )
+        log.info("%s: Search server available disk size of %d MB and store size is %d MB", str(self), space_avail, store_size)
         if space_avail < 10000:
             return [Problem(get_rule(RuleId.LOW_FREE_DISK_SPACE_2), self, str(self), space_avail // 1024)]
         elif store_size * 2 > space_avail:
@@ -119,7 +116,7 @@ def __audit_available_disk(self) -> list[Problem]:
 
 
 def __audit_index_balance(searchnodes: list[SearchNode]) -> list[Problem]:
-    """Audits whether ES index is decently balanced acros search nodes"""
+    """Audits whether ES index is decently balanced across search nodes"""
     log.info("Auditing search nodes store size balance")
     nbr_search_nodes = len(searchnodes)
     for i in range(nbr_search_nodes):

diff --git a/sonar/devops.py b/sonar/devops.py
@@ -27,7 +27,7 @@
 from requests import RequestException
 
 import sonar.logging as log
-from sonar.util import types, cache
+from sonar.util import types, cache, constants as c
 from sonar import platform
 import sonar.sqobject as sq
 from sonar import exceptions
@@ -41,7 +41,6 @@
 _CREATE_API_AZURE = "alm_settings/create_azure"
 _CREATE_API_BITBUCKET = "alm_settings/create_bitbucket"
 _CREATE_API_BBCLOUD = "alm_settings/create_bitbucketcloud"
-APIS = {"list": "alm_settings/list_definitions"}
 
 _TO_BE_SET = "TO_BE_SET"
 _IMPORTABLE_PROPERTIES = ("key", "type", "url", "workspace", "clientId", "appId")
@@ -53,6 +52,7 @@ class DevopsPlatform(sq.SqObject):
     """
 
     CACHE = cache.Cache()
+    API = {c.LIST: "alm_settings/list_definitions"}
 
     def __init__(self, endpoint: platform.Platform, key: str, platform_type: str) -> None:
         """Constructor"""
@@ -69,7 +69,7 @@ def read(cls, endpoint: platform.Platform, key: str) -> DevopsPlatform:
         o = DevopsPlatform.CACHE.get(key, endpoint.url)
         if o:
             return o
-        data = json.loads(endpoint.get(APIS["list"]).text)
+        data = json.loads(endpoint.get(DevopsPlatform.API[c.LIST]).text)
         for plt_type, platforms in data.items():
             for p in platforms:
                 if p["key"] == key:
@@ -137,7 +137,7 @@ def refresh(self) -> bool:
         :return: Whether the operation succeeded
         :rtype: bool
         """
-        data = json.loads(self.get(APIS["list"]).text)
+        data = json.loads(self.get(DevopsPlatform.API[c.LIST]).text)
         for alm_data in data.get(self.type, {}):
             if alm_data["key"] != self.key:
                 self.sq_json = alm_data
@@ -209,7 +209,7 @@ def get_list(endpoint: platform.Platform) -> dict[str, DevopsPlatform]:
     """
     if endpoint.is_sonarcloud():
         raise exceptions.UnsupportedOperation("Can't get list of DevOps platforms on SonarCloud")
-    data = json.loads(endpoint.get(APIS["list"]).text)
+    data = json.loads(endpoint.get(DevopsPlatform.API[c.LIST]).text)
     for alm_type in DEVOPS_PLATFORM_TYPES:
         for alm_data in data.get(alm_type, {}):
             DevopsPlatform.load(endpoint, alm_type, alm_data)

diff --git a/sonar/groups.py b/sonar/groups.py
@@ -35,16 +35,10 @@
 
 from sonar.audit import rules
 from sonar.audit.problem import Problem
-from sonar.util import types, cache
+from sonar.util import types, cache, constants as c
 
 SONAR_USERS = "sonar-users"
 
-_CREATE_API = "user_groups/create"
-_UPDATE_API = "user_groups/update"
-ADD_USER_API = "user_groups/add_user"
-REMOVE_USER_API = "user_groups/remove_user"
-_UPDATE_API_V2 = "v2/authorizations/groups"
-
 
 class Group(sq.SqObject):
     """
@@ -53,8 +47,15 @@ class Group(sq.SqObject):
     """
 
     CACHE = cache.Cache()
-    SEARCH_API = "user_groups/search"
-    SEARCH_API_V2 = "v2/authorizations/groups"
+    SEARCH_API_V1 = "user_groups/search"
+    UPDATE_API_V1 = "user_groups/update"
+    API = {
+        c.CREATE: "user_groups/create",
+        c.UPDATE: "v2/authorizations/groups",
+        c.SEARCH: "v2/authorizations/groups",
+        "ADD_USER": "user_groups/add_user",
+        "REMOVE_USER": "user_groups/remove_user",
+    }
     SEARCH_KEY_FIELD = "name"
     SEARCH_RETURN_FIELD = "groups"
 
@@ -82,7 +83,7 @@ def read(cls, endpoint: pf.Platform, name: str) -> Group:
         o = Group.CACHE.get(name, endpoint.url)
         if o:
             return o
-        data = util.search_by_name(endpoint, name, Group.SEARCH_API, "groups")
+        data = util.search_by_name(endpoint, name, Group.get_search_api(endpoint), "groups")
         if data is None:
             raise exceptions.ObjectNotFound(name, f"Group '{name}' not found.")
         # SonarQube 10 compatibility: "id" field is dropped, use "name" instead
@@ -103,7 +104,7 @@ def create(cls, endpoint: pf.Platform, name: str, description: str = None) -> Gr
         :rtype: Group or None
         """
         log.debug("Creating group '%s'", name)
-        endpoint.post(_CREATE_API, params={"name": name, "description": description})
+        endpoint.post(Group.API[c.SEARCH], params={"name": name, "description": description})
         return cls.read(endpoint=endpoint, name=name)
 
     @classmethod
@@ -119,9 +120,9 @@ def load(cls, endpoint: pf.Platform, data: types.ApiPayload) -> Group:
 
     @classmethod
     def get_search_api(cls, endpoint: object) -> Optional[str]:
-        api = cls.SEARCH_API
-        if endpoint.version() >= (10, 4, 0):
-            api = cls.SEARCH_API_V2
+        api = cls.API[c.SEARCH]
+        if endpoint.version() < (10, 4, 0):
+            api = cls.SEARCH_API_V1
         return api
 
     def __str__(self) -> str:
@@ -161,7 +162,7 @@ def add_user(self, user_login: str) -> bool:
         :rtype: bool
         """
         try:
-            r = self.post(ADD_USER_API, params={"login": user_login, "name": self.name})
+            r = self.post(Group.API["ADD_USER"], params={"login": user_login, "name": self.name})
         except (ConnectionError, RequestException) as e:
             util.handle_error(e, "adding user to group")
             if isinstance(e, HTTPError):
@@ -179,7 +180,7 @@ def remove_user(self, user_login: str) -> bool:
         :return: Whether the operation succeeded
         :rtype: bool
         """
-        return self.post(REMOVE_USER_API, params={"login": user_login, "name": self.name}).ok
+        return self.post(Group.API["REMOVE_USER"], params={"login": user_login, "name": self.name}).ok
 
     def audit(self, audit_settings: types.ConfigSettings = None) -> list[Problem]:
         """Audits a group and return list of problems found
@@ -226,9 +227,9 @@ def set_description(self, description: str) -> bool:
         log.debug("Updating %s with description = %s", str(self), description)
         if self.endpoint.version() >= (10, 4, 0):
             data = json.dumps({"description": description})
-            r = self.patch(f"{_UPDATE_API_V2}/{self._id}", data=data, headers={"content-type": "application/merge-patch+json"})
+            r = self.patch(f"{Group.API[c.UPDATE]}/{self._id}", data=data, headers={"content-type": "application/merge-patch+json"})
         else:
-            r = self.post(_UPDATE_API, params={"currentName": self.key, "description": description})
+            r = self.post(Group.UPDATE_API_V1, params={"currentName": self.key, "description": description})
         if r.ok:
             self.description = description
         return r.ok
@@ -245,9 +246,9 @@ def set_name(self, name: str) -> bool:
             return True
         log.debug("Updating %s with name = %s", str(self), name)
         if self.endpoint.version() >= (10, 4, 0):
-            r = self.patch(f"{_UPDATE_API_V2}/{self.key}", params={"name": name})
+            r = self.patch(f"{Group.API[c.UPDATE]}/{self.key}", params={"name": name})
         else:
-            r = self.post(_UPDATE_API, params={"currentName": self.key, "name": name})
+            r = self.post(Group.UPDATE_API_V1, params={"currentName": self.key, "name": name})
         if r.ok:
             Group.CACHE.pop(self)
             self.name = name