Migrate to the new SonarQube issue structure.

Migrate to the new SonarQube issue structure introduced in SonarQube 10.2. This means:
- Replace the "Severities" parameter with the "Impact severities" parameter. The possible values of the impact severity parameter are "low", "medium", and "high". Old severity parameter values are migrated to the new values according to the mapping documented by SonarSource: https://docs.sonarsource.com/sonarqube/latest/user-guide/issues/#severity-mapping.
- Replace the issue type parameter with the "software quality" parameter and use the impactSoftwareQualities parameter for the api/issues/search endpoint. Old issue type parameter values are migrated to software qualities (bug -> reliability, vulnerability -> security, code smell -> maintainability).
- Add a "clean code attribute" parameter and use the cleanCodeAttributeCategories parameter for the api/issues/search endpoint.
- Change the security_types parameter so it does not depend on the issue type.
- Document that the lowest supported SonarQube version is 10.2.

Closes #8354.

components/api_server/src/initialization/migrations.py

@@ -20,6 +20,7 @@ def perform_migrations(database: Database) -> None:
                 change_accessibility_violation_metrics_to_violations(report),
                 fix_branch_parameters_without_value(report),
                 change_ci_subject_types_to_development_environment(report),
+                change_sonarqube_parameters(report),
             ]
         ):
             change_description = " and to ".join([change for change in changes if change])
@@ -120,6 +121,76 @@ def add_source_parameter_hash_to_latest_measurement(database: Database, report)
     return count
 
 
+def change_sonarqube_parameters(report) -> str:
+    """Replace the SonarQube parameters to adapt to the new (SonarQube 10.2) issue structure.
+
+    Return a description of the change, if any.
+    """
+    # Added after Quality-time v5.13.0, see https://github.com/ICTU/quality-time/issues/8354
+    if change_sonarqube_severities(report) or change_sonarqube_types(report) or change_sonarqube_security_types(report):
+        return "change the SonarQube parameters"
+    return ""
+
+
+def change_sonarqube_severities(report) -> bool:
+    """Change the SonarQube severities parameter to the new values and rename it to impact_severities."""
+    # Severity mapping conform https://docs.sonarsource.com/sonarqube/latest/user-guide/issues/#severity-mapping:
+    value_mapping = {"blocker": "high", "critical": "high", "major": "medium", "minor": "low", "info": "low"}
+    metric_types = ("security_warnings", "suppressed_violations", "violations")
+    return change_sonarqube_parameter(report, metric_types, "severities", "impact_severities", value_mapping)
+
+
+def change_sonarqube_types(report) -> bool:
+    """Change the SonarQube types parameter to the new values and rename it to impacted_software_qualities."""
+    value_mapping = {"bug": "reliability", "vulnerability": "security", "code_smell": "maintainability"}
+    metric_types = ("suppressed_violations", "violations")
+    return change_sonarqube_parameter(report, metric_types, "types", "impacted_software_qualities", value_mapping)
+
+
+def change_sonarqube_security_types(report) -> bool:
+    """Change the SonarQube security types parameter to the new values."""
+    value_mapping = {"security_hotspot": "security hotspot", "vulnerability": "issue with security impact"}
+    metric_types = ("security_warnings",)
+    return change_sonarqube_parameter(report, metric_types, "security_types", "security_types", value_mapping)
+
+
+def change_sonarqube_parameter(
+    report, metric_types: Sequence[str], old_parameter_name: str, new_parameter_name: str, value_mapping: dict[str, str]
+) -> bool:
+    """Change the values of the specified parameter. Also rename the parameter if the old and new name differ."""
+    changed = False
+    for source in sources(report, metric_types=metric_types, source_type="sonarqube", parameter=old_parameter_name):
+        old_values = source["parameters"][old_parameter_name]
+        log_unknown_parameter_values(value_mapping, old_values, old_parameter_name, report)
+        if new_values := new_parameter_values(value_mapping, old_values):
+            source["parameters"][new_parameter_name] = new_values
+            changed = True
+        if new_parameter_name != old_parameter_name:
+            del source["parameters"][old_parameter_name]
+            changed = True
+    return changed
+
+
+def log_unknown_parameter_values(value_mapping: dict[str, str], old_values: list[str], value_type: str, report) -> None:
+    """Log old parameter values that do not exist in the mapping."""
+    if unknown_values := [old_value for old_value in old_values if old_value not in value_mapping]:
+        message = "Ignoring one or more unknown SonarQube parameter values of type '%s' in report %s: %s"
+        logging.warning(message, value_type, report["report_uuid"], ", ".join(unknown_values))
+
+
+def new_parameter_values(value_mapping: dict[str, str], old_values: list[str]) -> list[str]:
+    """Return the new values for each of the old values that exist in the mapping."""
+    return sorted({value_mapping[old_value] for old_value in old_values if old_value in value_mapping})
+
+
+def sources(report, metric_types: Sequence[str], source_type: str, parameter: str):
+    """Yield the sources in the report, filtered by metric type, source type, and parameter."""
+    for metric in metrics(report, metric_types):
+        for source in metric.get("sources", {}).values():
+            if source["type"] == source_type and parameter in source["parameters"]:
+                yield source
+
+
 def metrics(report, metric_types: Sequence[str] | None = None):
     """Yield the metrics in the report, optionally filtered by metric type."""
     for subject in subjects(report):

components/api_server/tests/initialization/test_migrations.py

@@ -2,7 +2,7 @@
 
 from initialization.migrations import perform_migrations
 
-from tests.base import DataModelTestCase
+from tests.base import DataModelTestCase, disable_logging
 from tests.fixtures import SourceId, REPORT_ID, SUBJECT_ID, METRIC_ID, METRIC_ID2, METRIC_ID3, SOURCE_ID, SOURCE_ID2
 
 
@@ -222,3 +222,94 @@ def test_ci_environment_with_title_and_subtitle(self):
         perform_migrations(self.database)
         inserted_report = self.inserted_report(subject_name="CI", subject_description="My CI")
         self.database.reports.replace_one.assert_called_once_with({"_id": "id"}, inserted_report)
+
+
+class SonarQubeParameterTest(MigrationTestCase):
+    """Unit tests for the SonarQube parameter database migration."""
+
+    def existing_report(
+        self,
+        metric_type: str = "violations",
+        sources: dict[SourceId, dict[str, str | dict[str, str | list[str]]]] | None = None,
+    ):
+        """Extend to add sources."""
+        report = super().existing_report(metric_type=metric_type)
+        report["subjects"][SUBJECT_ID]["metrics"][METRIC_ID]["sources"] = sources
+        return report
+
+    def sources(self, source_type: str = "sonarqube", **parameters):
+        """Create the sources fixture."""
+        return {SOURCE_ID: {"type": source_type, "parameters": {"branch": "main", **parameters}}}
+
+    def test_report_without_severity_or_types_parameter(self):
+        """Test that the migration succeeds when the SonarQube source has no severity or types parameter."""
+        self.database.reports.find.return_value = [self.existing_report(sources=self.sources())]
+        perform_migrations(self.database)
+        self.database.reports.replace_one.assert_not_called()
+
+    def test_report_with_violation_metric_but_no_sonarqube(self):
+        """Test that the migration succeeds when a violations metric has no SonarQube sources."""
+        self.database.reports.find.return_value = [self.existing_report(sources=self.sources("sarif"))]
+        perform_migrations(self.database)
+        self.database.reports.replace_one.assert_not_called()
+
+    def test_report_with_severity_parameter(self):
+        """Test that the migration succeeds when the SonarQube source has a severity parameter."""
+        self.database.reports.find.return_value = [self.existing_report(sources=self.sources(severities=["info"]))]
+        perform_migrations(self.database)
+        inserted_report = self.inserted_report(sources=self.sources(impact_severities=["low"]))
+        self.database.reports.replace_one.assert_called_once_with({"_id": "id"}, inserted_report)
+
+    def test_report_with_multiple_old_severity_values_that_map_to_the_same_new_value(self):
+        """Test a severity parameter with multiple old values that map to the same new value."""
+        reports = [self.existing_report(sources=self.sources(severities=["info", "minor"]))]
+        self.database.reports.find.return_value = reports
+        perform_migrations(self.database)
+        inserted_report = self.inserted_report(sources=self.sources(impact_severities=["low"]))
+        self.database.reports.replace_one.assert_called_once_with({"_id": "id"}, inserted_report)
+
+    @disable_logging
+    def test_report_with_unknown_old_severity_values(self):
+        """Test that unknown severity parameter values are ignored."""
+        sources = self.sources(severities=["info", ""])
+        sources[SOURCE_ID2] = {"type": "sonarqube", "parameters": {"branch": "main", "severities": ["foo"]}}
+        self.database.reports.find.return_value = [self.existing_report(sources=sources)]
+        perform_migrations(self.database)
+        inserted_sources = self.sources(impact_severities=["low"])
+        inserted_sources[SOURCE_ID2] = {"type": "sonarqube", "parameters": {"branch": "main"}}
+        inserted_report = self.inserted_report(sources=inserted_sources)
+        self.database.reports.replace_one.assert_called_once_with({"_id": "id"}, inserted_report)
+
+    def test_report_with_types_parameter(self):
+        """Test that the migration succeeds when the SonarQube source has a types parameter."""
+        self.database.reports.find.return_value = [self.existing_report(sources=self.sources(types=["bug"]))]
+        perform_migrations(self.database)
+        inserted_report = self.inserted_report(sources=self.sources(impacted_software_qualities=["reliability"]))
+        self.database.reports.replace_one.assert_called_once_with({"_id": "id"}, inserted_report)
+
+    def test_report_with_types_parameter_without_values(self):
+        """Test that the migration succeeds when the SonarQube source has a types parameter without values."""
+        self.database.reports.find.return_value = [self.existing_report(sources=self.sources(types=[]))]
+        perform_migrations(self.database)
+        inserted_report = self.inserted_report(sources=self.sources())
+        self.database.reports.replace_one.assert_called_once_with({"_id": "id"}, inserted_report)
+
+    def test_report_with_security_types_parameter(self):
+        """Test that the migration succeeds when the SonarQube source has a security types parameter."""
+        self.database.reports.find.return_value = [
+            self.existing_report(
+                metric_type="security_warnings",
+                sources=self.sources(security_types=["security_hotspot", "vulnerability"]),
+            ),
+        ]
+        perform_migrations(self.database)
+        inserted_sources = self.sources(security_types=["issue with security impact", "security hotspot"])
+        inserted_report = self.inserted_report(metric_type="security_warnings", sources=inserted_sources)
+        self.database.reports.replace_one.assert_called_once_with({"_id": "id"}, inserted_report)
+
+    def test_report_with_security_types_parameter_without_values(self):
+        """Test that the migration succeeds when the SonarQube source has a security types parameter without values."""
+        reports = [self.existing_report(metric_type="security_warnings", sources=self.sources(security_types=[]))]
+        self.database.reports.find.return_value = reports
+        perform_migrations(self.database)
+        self.database.reports.replace_one.assert_not_called()

components/collector/src/source_collectors/sonarqube/base.py

@@ -11,6 +11,8 @@
 class SonarQubeCollector(SourceCollector, ABC):
     """Base class for SonarQube collectors."""
 
+    PAGE_SIZE = 500  # 500 is the maximum. See e.g. https://next.sonarqube.com/sonarqube/web_api/api/issues/search
+
     async def _get_source_responses(self, *urls: URL) -> SourceResponses:
         """Extend to check the component exists before getting data about it."""
         # SonarQube sometimes gives results (e.g. zero violations) even if the component does not exist, so we

components/collector/src/source_collectors/sonarqube/security_warnings.py

@@ -9,59 +9,59 @@
 
 
 class SonarQubeSecurityWarnings(SonarQubeViolations):
-    """SonarQube security warnings. The security warnings are a sum of the vulnerabilities and security hotspots."""
-
-    types_parameter = "security_types"
+    """SonarQube security warnings, which are a combination of issues with security impact and security hotspots."""
 
     async def _landing_url(self, responses: SourceResponses) -> URL:
         """Extend to return the correct landing url depending on the selected security types."""
-        security_types = self._parameter(self.types_parameter)
         base_landing_url = await SourceCollector._landing_url(self, responses)  # noqa: SLF001
         component = self._parameter("component")
         branch = self._parameter("branch")
         common_url_parameters, extra_url_parameters = f"?id={component}&branch={branch}", ""
-        if "vulnerability" in security_types and "security_hotspot" in security_types:
+        if self.__issues_with_security_impact_selected() and self.__security_hotspots_selected():
             landing_path = "dashboard"
-        elif "vulnerability" in security_types:
+        elif self.__issues_with_security_impact_selected():
             landing_path = "project/issues"
-            # We don't use self._query_parameter() for the types parameter because when we get here,
-            # the value of the types parameter is fixed
-            extra_url_parameters = (
-                f"{self._query_parameter('severities', uppercase=True)}&resolved=false&types=VULNERABILITY"
-                f"{self._query_parameter('tags')}"
-            )
+            extra_url_parameters = f"&resolved=false{self._url_parameters()}"
         else:
             landing_path = "project/security_hotspots"
         return URL(f"{base_landing_url}/{landing_path}{common_url_parameters}{extra_url_parameters}")
 
     async def _get_source_responses(self, *urls: URL) -> SourceResponses:
         """Extend to add urls for the selected security types."""
         api_urls = []
-        security_types = self._parameter(self.types_parameter)
         component = self._parameter("component")
         branch = self._parameter("branch")
-        base_url = await SonarQubeCollector._api_url(self)  # noqa: SLF001
-        if "vulnerability" in security_types:
+        if self.__issues_with_security_impact_selected():
+            api_urls.append(await super()._api_url())
+        if self.__security_hotspots_selected():
+            base_url = await SonarQubeCollector._api_url(self)  # noqa: SLF001
             api_urls.append(
-                URL(
-                    f"{base_url}/api/issues/search?componentKeys={component}&resolved=false&ps=500"
-                    f"{self._query_parameter('severities', uppercase=True)}&branch={branch}&types=VULNERABILITY"
-                    f"{self._query_parameter('tags')}",
-                ),
+                URL(f"{base_url}/api/hotspots/search?projectKey={component}&branch={branch}&ps={self.PAGE_SIZE}")
             )
-        if "security_hotspot" in security_types:
-            api_urls.append(URL(f"{base_url}/api/hotspots/search?projectKey={component}&branch={branch}&ps=500"))
         return await super()._get_source_responses(*api_urls)
 
+    def _url_parameters(self) -> str:
+        """Override to return parameters needed for issues with security impact, common to API URL and landing URL."""
+        return (
+            self._query_parameter("impact_severities", uppercase=True)
+            + "&impactSoftwareQualities=SECURITY"
+            + self._query_parameter("tags")
+        )
+
+    async def _entity(self, issue) -> Entity:
+        """Extend to set the entity security type."""
+        entity = await super()._entity(issue)
+        entity["security_type"] = "issue with security impact"
+        return entity
+
     async def _parse_source_responses(self, responses: SourceResponses) -> SourceMeasurement:
         """Override to parse the selected security types."""
-        security_types = self._parameter(self.types_parameter)
         vulnerabilities = (
             await super()._parse_source_responses(SourceResponses(responses=[responses[0]]))
-            if "vulnerability" in security_types
+            if self.__issues_with_security_impact_selected()
             else SourceMeasurement()
         )
-        if "security_hotspot" in security_types:
+        if self.__security_hotspots_selected():
             json = await responses[-1].json()
             hotspots = [
                 await self.__hotspot_entity(hotspot)
@@ -89,7 +89,7 @@ async def __hotspot_entity(self, hotspot: dict[str, str]) -> Entity:
             key=hotspot["key"],
             component=hotspot["component"],
             message=hotspot["message"],
-            type="security_hotspot",
+            security_type="security hotspot",
             url=await self.__hotspot_landing_url(hotspot["key"]),
             review_priority=hotspot["vulnerabilityProbability"].lower(),
             creation_date=hotspot["creationDate"],
@@ -113,3 +113,11 @@ async def __hotspot_landing_url(self, hotspot_key: str) -> URL:
         component = self._parameter("component")
         branch = self._parameter("branch")
         return URL(f"{url}/security_hotspots?id={component}&branch={branch}&hotspots={hotspot_key}")
+
+    def __issues_with_security_impact_selected(self) -> bool:
+        """Return whether the user selected issues with security impact."""
+        return "issue with security impact" in self._parameter("security_types")
+
+    def __security_hotspots_selected(self) -> bool:
+        """Return whether the user selected security hotspots."""
+        return "security hotspot" in self._parameter("security_types")

components/collector/src/source_collectors/sonarqube/suppressed_violations.py

@@ -35,8 +35,7 @@ async def _get_source_responses(self, *urls: URL) -> SourceResponses:
         all_issues_api_url = URL(f"{url}/api/issues/search?componentKeys={component}&branch={branch}")
         resolved_issues_api_url = URL(
             f"{all_issues_api_url}&statuses=RESOLVED&resolutions=WONTFIX,FALSE-POSITIVE&additionalFields=comments"
-            f"{self._query_parameter('severities', uppercase=True)}"
-            f"{self._query_parameter(self.types_parameter, uppercase=True)}&ps=500",
+            f"{self._query_parameter('impact_severities', uppercase=True)}&ps={self.PAGE_SIZE}",
         )
         return await super()._get_source_responses(*[*urls, resolved_issues_api_url, all_issues_api_url])