Refactor-error-handling-part-2

cli/projects_cli.py

@@ -24,7 +24,6 @@
 
 """
 
-import sys
 import json
 
 from requests import RequestException

conf/prep_all_tests.sh

@@ -38,6 +38,8 @@ function create_fresh_project {
     fi
     curl -X POST -u "${usertoken}:" "${url}/api/projects/delete?project=${key}"
     conf/run_scanner.sh "${opts[@]}" -Dsonar.projectKey="${key}" -Dsonar.projectName="${key}" -Dsonar.host.url="${url}" "${opt_token}" "${opt_org}" 
+    conf/run_scanner.sh "${opts[@]}" -Dsonar.projectKey="${key}" -Dsonar.projectName="${key}" -Dsonar.host.url="${url}" "${opt_token}" "${opt_org}" -Dsonar.branch.name=develop
+    conf/run_scanner.sh "${opts[@]}" -Dsonar.projectKey="${key}" -Dsonar.projectName="${key}" -Dsonar.host.url="${url}" "${opt_token}" "${opt_org}" -Dsonar.branch.name=release-3.x
     return 0
 }
 

sonar/app_branches.py

@@ -24,8 +24,6 @@
 from typing import Optional
 
 import json
-from http import HTTPStatus
-from requests import RequestException
 from requests.utils import quote
 
 import sonar.logging as log
@@ -112,11 +110,7 @@ def create(cls, app: object, name: str, project_branches: list[Branch]) -> Appli
             else:  # Default main branch of project
                 params["project"].append(obj.key)
                 params["projectBranch"].append("")
-        try:
-            app.endpoint.post(ApplicationBranch.API[c.CREATE], params=params)
-        except (ConnectionError, RequestException) as e:
-            utilities.handle_error(e, f"creating branch {name} of {str(app)}", catch_http_statuses=(HTTPStatus.BAD_REQUEST,))
-            raise exceptions.ObjectAlreadyExists(f"{str(app)} branch '{name}", e.response.text)
+        app.endpoint.post(ApplicationBranch.API[c.CREATE], params=params)
         return ApplicationBranch(app=app, name=name, project_branches=project_branches)
 
     @classmethod
@@ -201,10 +195,9 @@ def update(self, name: str, project_branches: list[Branch]) -> bool:
             params["projectBranch"].append(br_name)
         try:
             ok = self.post(ApplicationBranch.API[c.UPDATE], params=params).ok
-        except (ConnectionError, RequestException) as e:
-            utilities.handle_error(e, f"updating {str(self)}", catch_http_statuses=(HTTPStatus.NOT_FOUND,))
+        except exceptions.ObjectNotFound:
             ApplicationBranch.CACHE.pop(self)
-            raise exceptions.ObjectNotFound(str(self), e.response.text)
+            raise
 
         self.name = name
         self._project_branches = project_branches

sonar/applications.py

@@ -93,11 +93,7 @@ def get_object(cls, endpoint: pf.Platform, key: str) -> Application:
         o = Application.CACHE.get(key, endpoint.local_url)
         if o:
             return o
-        try:
-            data = json.loads(endpoint.get(Application.API[c.GET], params={"application": key}).text)["application"]
-        except (ConnectionError, RequestException) as e:
-            util.handle_error(e, f"searching application {key}", catch_http_statuses=(HTTPStatus.NOT_FOUND,))
-            raise exceptions.ObjectNotFound(key, f"Application key '{key}' not found")
+        data = json.loads(endpoint.get(Application.API[c.GET], params={"application": key}).text)["application"]
         return cls.load(endpoint, data)
 
     @classmethod
@@ -132,11 +128,7 @@ def create(cls, endpoint: pf.Platform, key: str, name: str) -> Application:
         :rtype: Application
         """
         check_supported(endpoint)
-        try:
-            endpoint.post(Application.API["CREATE"], params={"key": key, "name": name})
-        except (ConnectionError, RequestException) as e:
-            util.handle_error(e, f"creating application {key}", catch_http_statuses=(HTTPStatus.BAD_REQUEST,))
-            raise exceptions.ObjectAlreadyExists(key, e.response.text)
+        endpoint.post(Application.API["CREATE"], params={"key": key, "name": name})
         log.info("Creating object")
         return Application(endpoint=endpoint, key=key, name=name)
 
@@ -151,10 +143,9 @@ def refresh(self) -> None:
             self.reload(json.loads(self.get("navigation/component", params={"component": self.key}).text))
             self.reload(json.loads(self.get(Application.API[c.GET], params=self.api_params(c.GET)).text)["application"])
             self.projects()
-        except (ConnectionError, RequestException) as e:
-            util.handle_error(e, f"refreshing {str(self)}", catch_http_statuses=(HTTPStatus.NOT_FOUND,))
+        except exceptions.ObjectNotFound:
             Application.CACHE.pop(self)
-            raise exceptions.ObjectNotFound(self.key, f"{str(self)} not found")
+            raise
 
     def __str__(self) -> str:
         """String name of object"""

sonar/branches.py

@@ -24,8 +24,8 @@
 from http import HTTPStatus
 from typing import Optional
 import json
+import re
 from urllib.parse import unquote
-from requests import HTTPError, RequestException
 import requests.utils
 
 from sonar import platform
@@ -89,16 +89,11 @@ def get_object(cls, concerned_object: projects.Project, branch_name: str) -> Bra
         o = Branch.CACHE.get(concerned_object.key, branch_name, concerned_object.base_url())
         if o:
             return o
-        try:
-            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")
-
-        for br in data.get("branches", []):
-            if br["name"] == branch_name:
-                return cls.load(concerned_object, branch_name, br)
-        raise exceptions.ObjectNotFound(branch_name, f"Branch '{branch_name}' of {str(concerned_object)} not found")
+        data = json.loads(concerned_object.get(Branch.API[c.LIST], params={"project": concerned_object.key}).text)
+        br = next((b for b in data.get("branches", []) if b["name"] == branch_name), None)
+        if not br:
+            raise exceptions.ObjectNotFound(branch_name, f"Branch '{branch_name}' of {str(concerned_object)} not found")
+        return cls.load(concerned_object, branch_name, br)
 
     @classmethod
     def load(cls, concerned_object: projects.Project, branch_name: str, data: types.ApiPayload) -> Branch:
@@ -112,9 +107,11 @@ def load(cls, concerned_object: projects.Project, branch_name: str, data: types.
         """
         branch_name = unquote(branch_name)
         o = Branch.CACHE.get(concerned_object.key, branch_name, concerned_object.base_url())
+        br_data = next((br for br in data.get("branches", []) if br["name"] == branch_name), None)
         if not o:
             o = cls(concerned_object, branch_name)
-        o._load(data)
+        if br_data:
+            o._load(br_data)
         return o
 
     def __str__(self) -> str:
@@ -135,44 +132,33 @@ def refresh(self) -> Branch:
         :return: itself
         :rtype: Branch
         """
-        try:
-            data = json.loads(self.get(Branch.API[c.LIST], params=self.api_params(c.LIST)).text)
-        except (ConnectionError, RequestException) as e:
-            util.handle_error(e, f"refreshing {str(self)}", catch_http_statuses=(HTTPStatus.NOT_FOUND,))
-            Branch.CACHE.pop(self)
-            raise exceptions.ObjectNotFound(self.key, f"{str(self)} not found in SonarQube")
-        for br in data.get("branches", []):
-            if br["name"] == self.name:
-                self._load(br)
-            else:
-                # While we're there let's load other branches with up to date branch data
-                Branch.load(self.concerned_object, br["name"], data)
+        data = json.loads(self.get(Branch.API[c.LIST], params=self.api_params(c.LIST)).text)
+        br_data = next((br for br in data.get("branches", []) if br["name"] == self.name), None)
+        if not br_data:
+            Branch.CACHE.clear()
+            raise exceptions.ObjectNotFound(self.name, f"{str(self)} not found")
+        self._load(br_data)
+        # While we're there let's load other branches with up to date branch data
+        for br in [b for b in data.get("branches", []) if b["name"] != self.name]:
+            Branch.load(self.concerned_object, br["name"], data)
         return self
 
     def _load(self, data: types.ApiPayload) -> None:
-        if self.sq_json is None:
-            self.sq_json = data
-        else:
-            self.sq_json.update(data)
+        log.debug("Loading %s with data %s", self, data)
+        self.sq_json = (self.sq_json or {}) | data
         self._is_main = self.sq_json["isMain"]
         self._last_analysis = util.string_to_date(self.sq_json.get("analysisDate", None))
         self._keep_when_inactive = self.sq_json.get("excludedFromPurge", False)
         self._is_main = self.sq_json.get("isMain", False)
 
     def is_kept_when_inactive(self) -> bool:
-        """
-        :return: Whether the branch is kept when inactive
-        :rtype: bool
-        """
+        """Returns whether the branch is kept when inactive"""
         if self._keep_when_inactive is None or self.sq_json is None:
             self.refresh()
         return self._keep_when_inactive
 
     def is_main(self) -> bool:
-        """
-        :return: Whether the branch is the project main branch
-        :rtype: bool
-        """
+        """Returns whether the branch is the project main branch"""
         if self._is_main is None or self.sq_json is None:
             self.refresh()
         return self._is_main
@@ -186,11 +172,32 @@ def delete(self) -> bool:
         """
         try:
             return super().delete()
-        except (ConnectionError, RequestException) as e:
-            if isinstance(e, HTTPError) and e.response.status_code == HTTPStatus.BAD_REQUEST:
-                log.warning("Can't delete %s, it's the main branch", str(self))
+        except exceptions.SonarException as e:
+            log.warning(e.message)
             return False
 
+    def get(
+        self, api: str, params: types.ApiParams = None, data: Optional[str] = None, mute: tuple[HTTPStatus] = (), **kwargs: str
+    ) -> requests.Response:
+        """Performs an HTTP GET request for the object"""
+        try:
+            return super().get(api=api, params=params, data=data, mute=mute, **kwargs)
+        except exceptions.ObjectNotFound as e:
+            if re.match(r"Project .+ not found", e.message):
+                log.warning("Clearing project cache")
+                projects.Project.CACHE.clear()
+            raise
+
+    def post(self, api: str, params: types.ApiParams = None, mute: tuple[HTTPStatus] = (), **kwargs: str) -> requests.Response:
+        """Performs an HTTP POST request for the object"""
+        try:
+            return super().post(api=api, params=params, mute=mute, **kwargs)
+        except exceptions.ObjectNotFound as e:
+            if re.match(r"Project .+ not found", e.message):
+                log.warning("Clearing project cache")
+                projects.Project.CACHE.clear()
+            raise
+
     def new_code(self) -> str:
         """
         :return: The branch new code period definition
@@ -199,13 +206,7 @@ def new_code(self) -> str:
         if self._new_code is None and self.endpoint.is_sonarcloud():
             self._new_code = settings.new_code_to_string({"inherited": True})
         elif self._new_code is None:
-            try:
-                data = json.loads(self.get(api=Branch.API["get_new_code"], params=self.api_params(c.LIST)).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)
-                raise exceptions.ObjectNotFound(self.concerned_object.key, f"{str(self.concerned_object)} not found")
-
+            data = json.loads(self.get(api=Branch.API["get_new_code"], params=self.api_params(c.LIST)).text)
             for b in data["newCodePeriods"]:
                 new_code = settings.new_code_to_string(b)
                 if b["branchKey"] == self.name:
@@ -245,24 +246,17 @@ def set_keep_when_inactive(self, keep: bool) -> bool:
         :return: Whether the operation was successful
         """
         log.info("Setting %s keep when inactive to %s", self, keep)
-        try:
-            self.post("project_branches/set_automatic_deletion_protection", params=self.api_params() | {"value": str(keep).lower()})
+        ok = self.post("project_branches/set_automatic_deletion_protection", params=self.api_params() | {"value": str(keep).lower()}).ok
+        if ok:
             self._keep_when_inactive = keep
-        except (ConnectionError, RequestException) as e:
-            util.handle_error(e, f"setting {str(self)} keep when inactive to {keep}", catch_all=True)
-            return False
         return True
 
     def set_as_main(self) -> bool:
         """Sets the branch as the main branch of the project
 
         :return: Whether the operation was successful
         """
-        try:
-            self.post("api/project_branches/set_main", params=self.api_params())
-        except (ConnectionError, RequestException) as e:
-            util.handle_error(e, f"setting {str(self)} as main branch", catch_all=True)
-            return False
+        self.post("api/project_branches/set_main", params=self.api_params())
         for b in self.concerned_object.branches().values():
             b._is_main = b.name == self.name
         return True
@@ -317,30 +311,21 @@ def rename(self, new_name: str) -> bool:
             log.debug("Skipping rename %s with same new name", str(self))
             return False
         log.info("Renaming main branch of %s from '%s' to '%s'", str(self.concerned_object), self.name, new_name)
-        try:
-            self.post(Branch.API[c.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):
-                if e.response.status_code == HTTPStatus.NOT_FOUND:
-                    Branch.CACHE.pop(self)
-                    raise exceptions.ObjectNotFound(self.concerned_object.key, f"str{self.concerned_object} not found")
-                if e.response.status_code == HTTPStatus.BAD_REQUEST:
-                    return False
+        self.post(Branch.API[c.RENAME], params={"project": self.concerned_object.key, "name": new_name})
         Branch.CACHE.pop(self)
         self.name = new_name
         Branch.CACHE.put(self)
         return True
 
-    def get_findings(self) -> dict[str, object]:
+    def get_findings(self, filters: Optional[types.ApiParams] = None) -> dict[str, object]:
         """Returns a branch list of findings
 
         :return: dict of Findings, with finding key as key
         :rtype: dict{key: Finding}
         """
-        findings = self.get_issues()
-        findings.update(self.get_hotspots())
-        return findings
+        if not filters:
+            return self.concerned_object.get_findings(branch=self.name)
+        return self.get_issues(filters) | self.get_hotspots(filters)
 
     def component_data(self) -> dict[str, str]:
         """Returns key data"""

sonar/devops.py

@@ -22,11 +22,8 @@
 
 from __future__ import annotations
 from typing import Optional, Union
-from http import HTTPStatus
 import json
 
-from requests import RequestException
-
 import sonar.logging as log
 from sonar.util import types, cache
 from sonar import platform
@@ -109,11 +106,10 @@ def create(cls, endpoint: platform.Platform, key: str, plt_type: str, url_or_wor
             elif plt_type == "bitbucketcloud":
                 params.update({"clientSecret": _TO_BE_SET, "clientId": _TO_BE_SET, "workspace": url_or_workspace})
                 endpoint.post(_CREATE_API_BBCLOUD, params=params)
-        except (ConnectionError, RequestException) as e:
-            util.handle_error(e, f"creating devops platform {key}/{plt_type}/{url_or_workspace}", catch_http_statuses=(HTTPStatus.BAD_REQUEST,))
+        except exceptions.SonarException as e:
             if endpoint.edition() in (c.CE, c.DE):
                 log.warning("Can't set DevOps platform '%s', don't you have more that 1 of that type?", key)
-            raise exceptions.UnsupportedOperation(f"Can't set DevOps platform '{key}', don't you have more that 1 of that type?")
+            raise exceptions.UnsupportedOperation(e.message) from e
         o = DevopsPlatform(endpoint=endpoint, key=key, platform_type=plt_type)
         o.refresh()
         return o
@@ -185,8 +181,7 @@ def update(self, **kwargs) -> bool:
             ok = self.post(f"alm_settings/update_{alm_type}", params=params).ok
             self.url = kwargs["url"]
             self._specific = {k: v for k, v in params.items() if k not in ("key", "url")}
-        except (ConnectionError, RequestException) as e:
-            util.handle_error(e, f"updating devops platform {self.key}/{alm_type}", catch_http_statuses=(HTTPStatus.BAD_REQUEST,))
+        except exceptions.SonarException:
             ok = False
         return ok
 

sonar/findings.py

@@ -24,8 +24,6 @@
 from datetime import datetime
 from typing import Optional
 import re
-from http import HTTPStatus
-from requests import RequestException
 import Levenshtein
 
 import sonar.logging as log
@@ -184,7 +182,7 @@ def assign(self, assignee: Optional[str] = None) -> str:
 
     def language(self) -> str:
         """Returns the finding language"""
-        return rules.get_object(endpoint=self.endpoint, key=self.rule).language
+        return rules.Rule.get_object(endpoint=self.endpoint, key=self.rule).language
 
     def to_csv(self, without_time: bool = False) -> list[str]:
         """
@@ -460,13 +458,10 @@ def search_siblings(
     def do_transition(self, transition: str) -> bool:
         try:
             return self.post("issues/do_transition", {"issue": self.key, "transition": transition}).ok
-        except (ConnectionError, RequestException) as e:
-            util.handle_error(e, f"applying transition {transition}", catch_http_statuses=(HTTPStatus.BAD_REQUEST, HTTPStatus.NOT_FOUND))
         except exceptions.SonarException as e:
             if re.match(r"Transition from state [A-Za-z]+ does not exist", e.message):
                 raise exceptions.UnsupportedOperation(e.message) from e
             raise
-        return False
 
     def get_branch_and_pr(self, data: types.ApiPayload) -> tuple[Optional[str], Optional[str]]:
         """