Merge branch 'branch-3.16.x' into fix-2039

Author: okorach-sonar

.github/workflows/build.yml

@@ -28,8 +28,12 @@ jobs:
       - name: Install dependencies
         run: |
           python -m pip install --upgrade pip
-          if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
-          if [ -f requirements-to-build.txt ]; then pip install -r requirements-to-build.txt; fi
+          python -m pip install poetry
+          poetry install
+      - name: Build package
+        run: |
+          poetry build
+        continue-on-error: false
         # Linting is done in the run_linters.sh script
 
       - name: Prep tests
@@ -49,9 +53,10 @@ jobs:
       - name: Run linters
         working-directory: .
         run: |
+          python -m pip install ruff pylint flake8
           chmod +x conf/run_linters.sh
           conf/run_linters.sh
-      #- name: Cache SonarQube packages
+      # - name: Cache SonarQube packages
       #  uses: actions/cache@v4
       #  with:
       #    path: ./.sonar

.vscode/settings.json

@@ -6,5 +6,10 @@
     "sonarlint.focusOnNewCode": true,
     "pylint.args": [
         "[\"--rcfile=conf/pylintrc\"]"
-    ]
+    ],
+    "python.testing.pytestArgs": [
+        "test"
+    ],
+    "python.testing.unittestEnabled": false,
+    "python.testing.pytestEnabled": true
 }

README.md

@@ -18,15 +18,14 @@
 # along with this program; if not, write to the Free Software Foundation,
 # Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
 #
-"""
+"""Audits a SonarQube platform"""
 
-    Audits a SonarQube platform
+from __future__ import annotations
 
-"""
-import sys
 import json
 import csv
-from typing import TextIO
+import re
+from typing import TextIO, Optional
 from threading import Thread
 from queue import Queue
 from requests import RequestException
@@ -52,12 +51,14 @@
     options.WHAT_PORTFOLIOS: portfolios.audit,
 }
 
+PROBLEM_KEYS = "problems"
+
 
 def _audit_sif(sysinfo: str, audit_settings: types.ConfigSettings) -> tuple[str, list[problem.Problem]]:
     """Audits a SIF and return found problems"""
     log.info("Auditing SIF file '%s'", sysinfo)
     try:
-        with open(sysinfo, "r", encoding="utf-8") as f:
+        with open(sysinfo, encoding="utf-8") as f:
             sysinfo = json.loads(f.read())
     except json.decoder.JSONDecodeError:
         log.critical("File %s does not seem to be a legit JSON file", sysinfo)
@@ -72,34 +73,48 @@ def _audit_sif(sysinfo: str, audit_settings: types.ConfigSettings) -> tuple[str,
     return sif_obj.server_id(), sif_obj.audit(audit_settings)
 
 
+def __filter_problems(problems: list[problem.Problem], settings: types.ConfigSettings) -> list[problem.Problem]:
+    """Filters audit problems by severity and/or type and/or problem key"""
+    if settings.get(options.SEVERITIES, None):
+        log.debug("Filtering audit problems with severities: %s", settings[options.SEVERITIES])
+        problems = [p for p in problems if str(p.severity) in settings[options.SEVERITIES]]
+    if settings.get(options.TYPES, None):
+        log.debug("Filtering audit problems with types: %s", settings[options.TYPES])
+        problems = [p for p in problems if str(p.type) in settings[options.TYPES]]
+    if settings.get(PROBLEM_KEYS, None):
+        log.debug("Filtering audit problems with keys: %s", settings[PROBLEM_KEYS])
+        problems = [p for p in problems if re.match(rf"^{settings[PROBLEM_KEYS]}$", str(p.rule_id))]
+    return problems
+
+
 def write_csv(queue: Queue[list[problem.Problem]], fd: TextIO, settings: types.ConfigSettings) -> None:
-    """Writes the CSV file of audit problems"""
+    """Thread callback to write audit problems in a CSV file"""
     server_id = settings.get("SERVER_ID", None)
     with_url = settings.get("WITH_URL", False)
     csvwriter = csv.writer(fd, delimiter=settings.get("CSV_DELIMITER", ","))
     header = ["Server Id"] if server_id else []
-    header += ["Audit Check", "Category", "Severity", "Message"]
+    header += ["Problem", "Type", "Severity", "Message"]
     header += ["URL"] if with_url else []
     csvwriter.writerow(header)
     while (problems := queue.get()) is not util.WRITE_END:
+        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 += list(json_data.values())
+            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()
 
 
 def write_json(queue: Queue[list[problem.Problem]], fd: TextIO, settings: types.ConfigSettings) -> None:
-    """
-    Thread to write problems in a JSON file
-    """
+    """Thread callback to write problems in a JSON file"""
     server_id = settings.get("SERVER_ID", None)
     with_url = settings.get("WITH_URL", False)
     comma = ""
     print("[", file=fd)
     while (problems := queue.get()) is not util.WRITE_END:
+        problems = __filter_problems(problems, settings)
         for p in problems:
             json_data = p.to_json(with_url)
             if server_id:
@@ -113,7 +128,7 @@ def write_json(queue: Queue[list[problem.Problem]], fd: TextIO, settings: types.
 
 
 def _audit_sq(
-    sq: platform.Platform, settings: types.ConfigSettings, what_to_audit: list[str] = None, key_list: types.KeyList = None
+    sq: platform.Platform, settings: types.ConfigSettings, what_to_audit: Optional[list[str]] = None, key_list: Optional[types.KeyList] = None
 ) -> list[problem.Problem]:
     """Audits a SonarQube/Cloud platform"""
     everything = what_to_audit is None
@@ -170,6 +185,24 @@ def __parser_args(desc: str) -> object:
         nargs="*",
         help="Pass audit configuration settings on command line (-D<setting>=<value>)",
     )
+    parser.add_argument(
+        f"--{options.SEVERITIES}",
+        required=False,
+        default=None,
+        help="Report only audit problems with the given severities (comma separate values LOW, MEDIUM, HIGH, CRITICAL)",
+    )
+    parser.add_argument(
+        f"--{options.TYPES}",
+        required=False,
+        default=None,
+        help="Report only audit problems of the given comma separated problem types",
+    )
+    parser.add_argument(
+        f"--{PROBLEM_KEYS}",
+        required=False,
+        default=None,
+        help="Report only audit problems whose type key matches the given regexp",
+    )
     args = options.parse_and_check(parser=parser, logger_name=TOOL_NAME, verify_token=False)
     if args.sif is None and args.config is None:
         util.check_token(args.token)
@@ -178,9 +211,8 @@ def __parser_args(desc: str) -> object:
 
 def __check_keys_exist(key_regexp: list[str], sq: platform.Platform, what: list[str]) -> None:
     """Checks if project keys exist"""
-    if key_regexp and "projects" in what:
-        if len(component_helper.get_components(sq, "projects", key_regexp)) == 0:
-            raise options.ArgumentsError(f"No projects found with key matching regexp '{key_regexp}'")
+    if key_regexp and "projects" in what and len(component_helper.get_components(sq, "projects", key_regexp)) == 0:
+        raise options.ArgumentsError(f"No projects found with key matching regexp '{key_regexp}'")
 
 
 def main() -> None:
@@ -194,14 +226,14 @@ def main() -> None:
             key, value = val[0].split("=", maxsplit=1)
             cli_settings[key] = value
         settings = audit_conf.load(TOOL_NAME, cli_settings)
+        settings |= kwargs
         file = ofile = kwargs.pop(options.REPORT_FILE)
         fmt = util.deduct_format(kwargs[options.FORMAT], ofile)
         settings.update(
             {
                 "FILE": file,
                 "CSV_DELIMITER": kwargs[options.CSV_SEPARATOR],
                 "WITH_URL": kwargs[options.WITH_URL],
-                "threads": kwargs[options.NBR_THREADS],
                 "format": fmt,
             }
         )
@@ -213,8 +245,8 @@ def main() -> None:
             file = kwargs["sif"]
             errcode = errcodes.SIF_AUDIT_ERROR
             (settings["SERVER_ID"], problems) = _audit_sif(file, settings)
-            problem.dump_report(problems, file=ofile, server_id=settings["SERVER_ID"], format=fmt)
-
+            problems = __filter_problems(problems, settings)
+            problem.dump_report(problems, file=ofile, server_id=settings["SERVER_ID"], fmt=fmt)
         else:
             sq = platform.Platform(**kwargs)
             sq.verify_connection()

cli/audit.py

@@ -19,9 +19,10 @@
 # Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
 #
 """
-    Exports SonarQube platform configuration as JSON
+Exports SonarQube platform configuration as JSON
 """
-from typing import TextIO, Optional
+
+from typing import TextIO, Any
 from threading import Thread
 from queue import Queue
 
@@ -121,16 +122,20 @@ def __parse_args(desc: str) -> object:
     return options.parse_and_check(parser=parser, logger_name=TOOL_NAME)
 
 
-def __normalize_json(json_data: dict[str, any], remove_empty: bool = True, remove_none: bool = True) -> dict[str, any]:
+def __normalize_json(json_data: dict[str, Any], remove_empty: bool = True, remove_none: bool = True) -> dict[str, any]:
     """Sorts a JSON file and optionally remove empty and none values"""
     log.info("Normalizing JSON - remove empty = %s, remove nones = %s", str(remove_empty), str(remove_none))
-    if remove_empty:
-        json_data = utilities.remove_empties(json_data)
-    if remove_none:
-        json_data = utilities.remove_nones(json_data)
+    json_data = utilities.clean_data(json_data, remove_empty, remove_none)
     json_data = utilities.order_keys(json_data, *_SECTIONS_ORDER)
     for key in [k for k in _SECTIONS_TO_SORT if k in json_data]:
-        json_data[key] = {k: json_data[key][k] for k in sorted(json_data[key])}
+        if isinstance(json_data[key], (list, tuple, set)):
+            if len(json_data[key]) > 0:
+                sort_field = next((k for k in ("key", "name", "login") if k in json_data[key][0]), None)
+                if sort_field:
+                    tmp_d = {v[sort_field]: v for v in json_data[key]}
+                    json_data[key] = list(dict(sorted(tmp_d.items())).values())
+        else:
+            json_data[key] = {k: json_data[key][k] for k in sorted(json_data[key])}
     return json_data
 
 
@@ -170,28 +175,37 @@ def write_objects(queue: Queue[types.ObjectJsonRepr], fd: TextIO, object_type: s
     """
     done = False
     prefix = ""
+    objects_exported_as_lists = ("projects", "applications", "users", "portfolios")
+    objects_exported_as_whole = ("qualityGates", "groups")
     log.info("Waiting %s to write...", object_type)
-    print(f'"{object_type}": ' + "{", file=fd)
+    if object_type in objects_exported_as_lists:
+        start, stop = ("[", "]")
+    elif object_type in objects_exported_as_whole:
+        start, stop = ("", "")
+    else:
+        start, stop = ("{", "}")
+    print(f'"{object_type}": ' + start, file=fd)
     while not done:
         obj_json = queue.get()
         if not (done := obj_json is utilities.WRITE_END):
             if object_type == "groups":
                 obj_json = __prep_json_for_write(obj_json, {**export_settings, EXPORT_EMPTY: True})
             else:
                 obj_json = __prep_json_for_write(obj_json, export_settings)
-            if object_type in ("projects", "applications", "portfolios", "users"):
-                if object_type == "users":
-                    key = obj_json.pop("login", None)
-                else:
-                    key = obj_json.pop("key", None)
-                log.debug("Writing %s key '%s'", object_type[:-1], key)
+            key = "" if isinstance(obj_json, list) else obj_json.get("key", obj_json.get("login", obj_json.get("name", "unknown")))
+            log.debug("Writing %s key '%s'", object_type[:-1], key)
+            if object_type in objects_exported_as_lists:
+                print(f"{prefix}{utilities.json_dump(obj_json)}", end="", file=fd)
+            elif object_type in objects_exported_as_whole:
+                print(f"{prefix}{utilities.json_dump(obj_json)}", end="", file=fd)
+            elif object_type in ("applications", "portfolios", "users"):
                 print(f'{prefix}"{key}": {utilities.json_dump(obj_json)}', end="", file=fd)
             else:
                 log.debug("Writing %s", object_type)
                 print(f"{prefix}{utilities.json_dump(obj_json)[2:-1]}", end="", file=fd)
             prefix = ",\n"
         queue.task_done()
-    print("\n}", file=fd, end="")
+    print("\n" + stop, file=fd, end="")
     log.info("Writing %s complete", object_type)
 
 
@@ -256,10 +270,7 @@ def __prep_json_for_write(json_data: types.ObjectJsonRepr, export_settings: type
     if export_settings.get("MODE", "CONFIG") == "MIGRATION":
         return json_data
     if not export_settings.get("FULL_EXPORT", False):
-        json_data = utilities.remove_nones(json_data)
-        if not export_settings.get(EXPORT_EMPTY, False):
-            log.debug("Removing empties")
-            json_data = utilities.remove_empties(json_data)
+        json_data = utilities.clean_data(json_data, remove_empty=not export_settings.get(EXPORT_EMPTY, False), remove_none=True)
     if export_settings.get("INLINE_LISTS", True):
         json_data = utilities.inline_lists(json_data, exceptions=("conditions",))
     return json_data

cli/config.py

@@ -19,10 +19,10 @@
 # Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
 #
 """
-    This script manipulates custom measures. You may:
+This script manipulates custom measures. You may:
 
-    Update a custom measure value:
-        Usage: cust_measures.py -t <SQ_TOKEN> -u <SQ_URL> -k <projectKey> -m <metricKey> --updateValue <value>
+Update a custom measure value:
+    Usage: cust_measures.py -t <SQ_TOKEN> -u <SQ_URL> -k <projectKey> -m <metricKey> --updateValue <value>
 """
 
 import sys

cli/cust_measures.py

@@ -19,9 +19,9 @@
 # Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
 #
 """
-    This script exports findings as CSV, JSON, or SARIF
+This script exports findings as CSV, JSON, or SARIF
 
-    Usage: sonar-findings-export.py -t <SQ_TOKEN> -u <SQ_URL> [<filters>]
+Usage: sonar-findings-export.py -t <SQ_TOKEN> -u <SQ_URL> [<filters>]
 
 """
 
@@ -125,7 +125,7 @@ def parse_args(desc: str) -> Namespace:
     parser.add_argument(
         f"--{options.SEVERITIES}",
         required=False,
-        help="Comma separated severities among" + util.list_to_csv(idefs.STD_SEVERITIES + hotspots.SEVERITIES),
+        help="Comma separated severities among " + util.list_to_csv(idefs.STD_SEVERITIES + hotspots.SEVERITIES),
     )
     parser.add_argument(
         f"--{options.TYPES}",
@@ -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/findings_export.py

@@ -19,13 +19,13 @@
 # Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
 #
 """
-    This script propagates the manual issue changes (FP, WF, Change
-    of severity, of issue type, comments) from:
-    - One project to another (normally on different platforms but not necessarily).
-      The 2 platform don't need to be identical in version, edition or plugins
-    - One branch of a project to another branch of the same project (normally LLBs)
+This script propagates the manual issue changes (FP, WF, Change
+of severity, of issue type, comments) from:
+- One project to another (normally on different platforms but not necessarily).
+  The 2 platform don't need to be identical in version, edition or plugins
+- One branch of a project to another branch of the same project (normally LLBs)
 
-    Only issues with a 100% match are synchronized. When there's a doubt, nothing is done
+Only issues with a 100% match are synchronized. When there's a doubt, nothing is done
 """
 
 import datetime