Kill-td (#2022)

* Remove useless ruff rule

* Reduce complexity

* Align CSV export

* Remove useless import

* Remove built-in shadow

* Remove built-in shadow

* Import for type checking only

* Add docstrings

* Kill TD

* Disable 2 ruff rules

* Quality pass

* Add tests on import projects

* Add test on import projects

* Fix regression

* New reference config

* Fix test import projects

* Revert format

* Fix regression

Author: okorach

cli/audit.py

@@ -100,8 +100,8 @@ def write_csv(queue: Queue[list[problem.Problem]], fd: TextIO, settings: types.C
         problems = __filter_problems(problems, settings)
         for p in problems:
             json_data = p.to_json(with_url)
-            data = [] if not server_id else [server_id]
-            data += [json_data[k] for k in ("problem", "type", "severity", "message") if k in json_data]
+            data = [server_id] if server_id else []
+            data += [json_data[k] for k in ("problem", "type", "severity", "message", "url") if k in json_data]
             csvwriter.writerow(data)
         queue.task_done()
     queue.task_done()
@@ -246,7 +246,7 @@ def main() -> None:
             errcode = errcodes.SIF_AUDIT_ERROR
             (settings["SERVER_ID"], problems) = _audit_sif(file, settings)
             problems = __filter_problems(problems, settings)
-            problem.dump_report(problems, file=ofile, server_id=settings["SERVER_ID"], format=fmt)
+            problem.dump_report(problems, file=ofile, server_id=settings["SERVER_ID"], fmt=fmt)
         else:
             sq = platform.Platform(**kwargs)
             sq.verify_connection()

cli/findings_export.py

@@ -373,6 +373,12 @@ def main() -> None:
             key_regexp=params.get(options.KEY_REGEXP, None),
             branch_regexp=branch_regexp,
         )
+        if params[options.COMPONENT_TYPE] == "portfolios":
+            components = []
+            for comp in components_list:
+                components += comp.components()
+            components_list = components
+
         if len(components_list) == 0:
             br = f"and branch matching regexp '{params[options.BRANCH_REGEXP]}'" if options.BRANCH_REGEXP in params else ""
             raise exceptions.SonarException(

cli/housekeeper.py

@@ -32,7 +32,6 @@
 from cli import options
 import sonar.logging as log
 from sonar import platform, tokens, users, projects, branches, pull_requests, version, errcodes
-from sonar.util import types
 import sonar.util.constants as c
 import sonar.utilities as util
 import sonar.exceptions as ex
@@ -145,41 +144,58 @@ def _parse_arguments() -> object:
     return options.parse_and_check(parser=parser, logger_name=TOOL_NAME)
 
 
-def _delete_objects(problems: problem.Problem, mode: str, settings: types.ConfigSettings) -> tuple[int, int, int, int, int]:
-    """Deletes objects (that should be housekept)"""
+def _revoke_tokens(problems: problem.Problem, mode: str) -> int:
+    """Revokes user tokens (that should be housekept)"""
     revoked_token_count = 0
-    deleted_projects = {}
-    deleted_branch_count = 0
-    deleted_pr_count = 0
-    deleted_loc = 0
-    for p in problems:
+    for p in [p for p in problems if isinstance(p.concerned_object, tokens.UserToken)]:
         obj = p.concerned_object
-        if obj is None:
-            continue  # BUG
         try:
-            if isinstance(obj, projects.Project):
-                loc = int(obj.get_measure("ncloc", fallback="0"))
-                if mode == "delete":
-                    log.info("Deleting %s, %d LoC", str(obj), loc)
-                else:
-                    log.info("%s, %d LoC should be deleted", str(obj), loc)
-                if mode != "delete" or obj.delete():
-                    deleted_projects[obj.key] = obj
-                    deleted_loc += loc
-            if isinstance(obj, (tokens.UserToken, users.User)) and (mode != "delete" or obj.revoke()):
+            if mode != "delete" or obj.revoke():
                 revoked_token_count += 1
-            elif settings[PROJ_MAX_AGE] > 0 and obj.project().key in deleted_projects:
-                log.info("%s deleted, so no need to delete %s", str(obj.project()), str(obj))
-            elif mode != "delete" or obj.delete():
-                log.info("%s to delete", str(obj))
-                if isinstance(obj, branches.Branch):
-                    deleted_branch_count += 1
-                elif isinstance(obj, pull_requests.PullRequest):
-                    deleted_pr_count += 1
+        except ex.ObjectNotFound:
+            log.warning("Token %s does not exist, revocation skipped...", obj)
+    return revoked_token_count
+
+
+def _delete_projects(problems: problem.Problem, mode: str) -> tuple[list[str], int]:
+    """Deletes projects (that should be housekept)"""
+    deleted_projects = []
+    loc_total = 0
+    for obj in [p.concerned_object for p in problems if isinstance(p.concerned_object, projects.Project)]:
+        try:
+            loc = int(obj.get_measure("ncloc", fallback="0"))
+            log.info("Deleting %s, %d LoC" if mode == "delete" else "%s, %d LoC should be deleted", str(obj), loc)
+            if mode != "delete" or obj.delete():
+                deleted_projects.append(obj.key)
+                loc_total += loc
+        except ex.ObjectNotFound:
+            log.warning("%s does not exist, deletion skipped...", str(obj))
+    return deleted_projects, loc_total
 
+
+def _delete_class(problems: problem.Problem, mode: str, proj_list: list[str], object_class: object) -> int:
+    """Deletes branches or PRs (that should be housekept)"""
+    counter = 0
+    for obj in [p.concerned_object for p in problems if isinstance(p.concerned_object, object_class)]:
+        try:
+            if obj.project().key in proj_list:
+                log.info("%s deleted, so no need to delete %s", str(obj.project()), str(obj))
+                continue
+            log.info("%s to delete", str(obj))
+            if mode != "delete" or obj.delete():
+                counter += 1
         except ex.ObjectNotFound:
             log.warning("%s does not exist, deletion skipped...", str(obj))
+    return counter
+
 
+def _delete_objects(problems: problem.Problem, mode: str) -> tuple[int, int, int, int, int]:
+    """Deletes objects (that should be housekept)"""
+    deleted_projects = {}
+    revoked_token_count = _revoke_tokens(problems, mode)
+    deleted_projects, deleted_loc = _delete_projects(problems, mode)
+    deleted_branch_count = _delete_class(problems, mode, deleted_projects, branches.Branch)
+    deleted_pr_count = _delete_class(problems, mode, deleted_projects, pull_requests.PullRequest)
     return (
         len(deleted_projects),
         deleted_loc,
@@ -224,13 +240,11 @@ def main() -> None:
         if token_age:
             problems += get_user_problems(settings, sq)
 
-        problem.dump_report(problems, file=kwargs[options.REPORT_FILE], format="csv")
+        problem.dump_report(problems, file=kwargs[options.REPORT_FILE], fmt="csv")
 
-        op = "to delete"
-        if mode == "delete":
-            op = "deleted"
-        (deleted_proj, deleted_loc, deleted_branches, deleted_prs, revoked_tokens) = _delete_objects(problems, mode, settings=settings)
+        (deleted_proj, deleted_loc, deleted_branches, deleted_prs, revoked_tokens) = _delete_objects(problems, mode)
 
+        op = "deleted" if mode == "delete" else "to delete"
         if proj_age > 0:
             log.info("%d projects older than %d days (%d LoCs) %s", deleted_proj, proj_age, deleted_loc, op)
         if branch_age > 0:

cli/support.py

@@ -166,7 +166,7 @@ def main():
             if problems:
                 found_problems = True
                 log.warning("%d issues found during audit", len(problems))
-                problem.dump_report(problems, file=None, format="csv")
+                problem.dump_report(problems, file=None, fmt="csv")
                 comment += build_jira_comments(problems)
             else:
                 log.info("%d issues found during audit", len(problems))

pyproject.toml

@@ -172,6 +172,9 @@ extend-ignore = [
     "I001",
     "TRY003",
     "EM102",
+    "C901", # Complexity rule, better covered by SonarQube
+    "ANN401", # Disallow Any type annotation
+    "TD001", # Disallow TODO/FIXME comments
 ]
 
 exclude = [

sonar/app_branches.py

@@ -21,21 +21,22 @@
 """Abstraction of Sonar Application Branch"""
 
 from __future__ import annotations
-from typing import Optional
-from datetime import datetime
-
+from typing import Optional, TYPE_CHECKING
 import json
 from requests.utils import quote
 
 import sonar.logging as log
-from sonar.util import types, cache
+from sonar.util import cache
 
 from sonar.components import Component
 
 from sonar.branches import Branch
 from sonar import exceptions, projects, utilities
 import sonar.util.constants as c
 
+if TYPE_CHECKING:
+    from sonar.util import types
+    from datetime import datetime
 
 _NOT_SUPPORTED = "Applications not supported in community edition"
 

sonar/applications.py

@@ -260,6 +260,7 @@ def delete(self) -> bool:
         return super().delete()
 
     def get_hotspots(self, filters: Optional[dict[str, str]] = None) -> dict[str, object]:
+        """Returns the security hotspots of the application (ie of its projects or branches)"""
         new_filters = filters.copy() if filters else {}
         pattern = new_filters.pop("branch", None) if new_filters else None
         if not pattern:
@@ -271,6 +272,7 @@ def get_hotspots(self, filters: Optional[dict[str, str]] = None) -> dict[str, ob
         return findings_list
 
     def get_issues(self, filters: Optional[dict[str, str]] = None) -> dict[str, object]:
+        """Returns the issues of the application (ie of its projects or branches)"""
         new_filters = filters.copy() if filters else {}
         pattern = new_filters.pop("branch", None) if new_filters else None
         if not pattern:

sonar/audit/problem.py

@@ -63,7 +63,7 @@ def to_json(self, with_url=False):
 
 
 def dump_report(
-    problems: list[Problem], file: str, server_id: Optional[str] = None, format: str = "csv", with_url: bool = False, separator: str = ","
+    problems: list[Problem], file: str, server_id: Optional[str] = None, fmt: str = "csv", with_url: bool = False, separator: str = ","
 ) -> None:
     """Dumps to file a report about a list of problems
 
@@ -74,7 +74,7 @@ def dump_report(
     :rtype: None
     """
     log.info("Writing report to %s", f"file '{file}'" if file else "stdout")
-    if format == "json":
+    if fmt == "json":
         __dump_json(problems=problems, file=file, server_id=server_id, with_url=with_url)
     else:
         __dump_csv(problems=problems, file=file, server_id=server_id, with_url=with_url, separator=separator)
@@ -91,14 +91,13 @@ def __dump_csv(problems: list[Problem], file: str, server_id: Optional[str] = No
     with utilities.open_file(file, "w") as fd:
         csvwriter = csv.writer(fd, delimiter=separator)
         header = ["Server Id"] if server_id else []
-        header += ["Audit Check", "Category", "Severity", "Message"]
+        header += ["Problem", "Category", "Severity", "Message"]
         header += ["URL"] if with_url else []
         csvwriter.writerow(header)
         for p in problems:
-            data = []
-            if server_id is not None:
-                data = [server_id]
-            data += list(p.to_json(with_url).values())
+            json_data = p.to_json(with_url)
+            data = [server_id] if server_id else []
+            data += [json_data[k] for k in ("problem", "type", "severity", "message", "url") if k in json_data]
             csvwriter.writerow(data)
 
 

sonar/branches.py

@@ -22,24 +22,27 @@
 
 from __future__ import annotations
 from http import HTTPStatus
-from typing import Optional
-from datetime import datetime
+from typing import Optional, TYPE_CHECKING
+
 import json
 import re
 from urllib.parse import unquote
 import requests.utils
 
 from sonar import platform
-from sonar.util import types, cache
+from sonar.util import cache
 import sonar.logging as log
 from sonar import components, settings, exceptions, tasks
-from sonar import projects
+from sonar import projects as proj
 import sonar.utilities as util
 
 from sonar.audit.problem import Problem
 from sonar.audit.rules import get_rule, RuleId
 import sonar.util.constants as c
 
+if TYPE_CHECKING:
+    from sonar.util import types
+    from datetime import datetime
 
 _UNSUPPORTED_IN_CE = "Branches not available in Community Edition"
 
@@ -57,7 +60,7 @@ class Branch(components.Component):
         "get_new_code": "new_code_periods/list",
     }
 
-    def __init__(self, project: projects.Project, name: str) -> None:
+    def __init__(self, project: proj.Project, name: str) -> None:
         """Don't use this, use class methods to create Branch objects
 
         :raises UnsupportedOperation: When attempting to branches on Community Edition
@@ -76,10 +79,10 @@ def __init__(self, project: projects.Project, name: str) -> None:
         log.debug("Created object %s", str(self))
 
     @classmethod
-    def get_object(cls, concerned_object: projects.Project, branch_name: str) -> Branch:
+    def get_object(cls, concerned_object: proj.Project, branch_name: str) -> Branch:
         """Gets a SonarQube Branch object
 
-        :param projects.Project concerned_object: projects.Project concerned by the branch
+        :param Project concerned_object: Project concerned by the branch
         :param str branch_name: The branch name
         :raises UnsupportedOperation: If trying to manipulate branches on a community edition
         :raises ObjectNotFound: If project key or branch name not found in SonarQube
@@ -97,10 +100,10 @@ def get_object(cls, concerned_object: projects.Project, branch_name: str) -> Bra
         return cls.load(concerned_object, branch_name, br)
 
     @classmethod
-    def load(cls, concerned_object: projects.Project, branch_name: str, data: types.ApiPayload) -> Branch:
+    def load(cls, concerned_object: proj.Project, branch_name: str, data: types.ApiPayload) -> Branch:
         """Gets a Branch object from JSON data gotten from a list API call
 
-        :param projects.Project concerned_object: the projects.Project the branch belonsg to
+        :param Project concerned_object: the Project the branch belonsg to
         :param str branch_name: Name of the branch
         :param dict data: Data received from API call
         :return: The Branch object
@@ -122,7 +125,7 @@ def __hash__(self) -> int:
         """Computes a uuid for the branch that can serve as index"""
         return hash((self.concerned_object.key, self.name, self.base_url()))
 
-    def project(self) -> projects.Project:
+    def project(self) -> proj.Project:
         """Returns the project key"""
         return self.concerned_object
 
@@ -186,7 +189,7 @@ def get(
         except exceptions.ObjectNotFound as e:
             if re.match(r"Project .+ not found", e.message):
                 log.warning("Clearing project cache")
-                projects.Project.CACHE.clear()
+                proj.Project.CACHE.clear()
             raise
 
     def post(self, api: str, params: types.ApiParams = None, mute: tuple[HTTPStatus] = (), **kwargs: str) -> requests.Response:
@@ -196,7 +199,7 @@ def post(self, api: str, params: types.ApiParams = None, mute: tuple[HTTPStatus]
         except exceptions.ObjectNotFound as e:
             if re.match(r"Project .+ not found", e.message):
                 log.warning("Clearing project cache")
-                projects.Project.CACHE.clear()
+                proj.Project.CACHE.clear()
             raise
 
     def new_code(self) -> str:
@@ -396,8 +399,7 @@ def audit(self, audit_settings: types.ConfigSettings) -> list[Problem]:
         try:
             if audit_settings.get(c.AUDIT_MODE_PARAM, "") == "housekeeper":
                 return self.__audit_last_analysis(audit_settings)
-            else:
-                return self.__audit_last_analysis(audit_settings) + self.__audit_never_analyzed() + self._audit_component(audit_settings)
+            return self.__audit_last_analysis(audit_settings) + self.__audit_never_analyzed() + self._audit_component(audit_settings)
         except Exception as e:
             log.error("%s while auditing %s, audit skipped", util.error_msg(e), str(self))
         return []
@@ -414,10 +416,10 @@ def last_task(self) -> Optional[tasks.Task]:
         return task
 
 
-def get_list(project: projects.Project) -> dict[str, Branch]:
+def get_list(project: proj.Project) -> dict[str, Branch]:
     """Retrieves the list of branches of a project
 
-    :param projects.Project project: projects.Project the branch belongs to
+    :param Project project: Project the branch belongs to
     :raises UnsupportedOperation: Branches not supported in Community Edition
     :return: List of project branches
     :rtype: dict{branch_name: Branch}
@@ -436,13 +438,13 @@ def exists(endpoint: platform.Platform, branch_name: str, project_key: str) -> b
 
     :param Platform endpoint: Reference to the SonarQube platform
     :param str branch_name: Branch name
-    :param str project_key: projects.Project key
+    :param str project_key: Project key
     :raises UnsupportedOperation: Branches not supported in Community Edition
     :return: Whether the branch exists in SonarQube
     :rtype: bool
     """
     try:
-        project = projects.Project.get_object(endpoint, project_key)
+        project = proj.Project.get_object(endpoint, project_key)
     except exceptions.ObjectNotFound:
         return False
     return branch_name in get_list(project)