SCANPY-119 New bootstrapping: Version Check (#129)

Author: Seppli11

its/pyproject.toml

@@ -39,7 +39,7 @@ licenseheaders = ">=0.8.8"
 black = "25.1.0"
 pytest = ">=7.4.3"
 pytest-docker = ">=2.0.1"
-requests = "2.31.0"
+requests = "^2.32.3"
 pysonar-scanner = { "path" = "../" }
 
 [[tool.poetry.packages]]

poetry.lock

@@ -30,6 +30,8 @@ relative_files = true
 [tool.poetry.dependencies]
 python = '>=3.9'
 toml = '>=0.10.2'
+requests = "^2.32.3"
+responses = "^0.25.6"
 
 [tool.poetry.group]
 [tool.poetry.group.dev]

pyproject.toml

@@ -19,7 +19,46 @@
 #
 from dataclasses import dataclass
 from pysonar_scanner.configuration import Configuration
+from pysonar_scanner.exceptions import SonarQubeApiException
 from pysonar_scanner.utils import remove_trailing_slash
+import requests
+
+
+@dataclass(frozen=True)
+class SQVersion:
+    parts: list[str]
+
+    def __get_part(self, index: int) -> int:
+        if index >= len(self.parts):
+            return 0
+        part = self.parts[index]
+        if not part.isdigit():
+            return 0
+        return int(part)
+
+    def major(self) -> int:
+        return self.__get_part(0)
+
+    def minor(self) -> int:
+        return self.__get_part(1)
+
+    def does_support_bootstrapping(self) -> bool:
+        if len(self.parts) == 0:
+            return False
+
+        return self.major() > MIN_SUPPORTED_SQ_VERSION.major() or (
+            self.major() == MIN_SUPPORTED_SQ_VERSION.major() and self.minor() >= MIN_SUPPORTED_SQ_VERSION.minor()
+        )
+
+    def __str__(self) -> str:
+        return ".".join(self.parts)
+
+    @staticmethod
+    def from_str(version: str) -> "SQVersion":
+        return SQVersion(version.split("."))
+
+
+MIN_SUPPORTED_SQ_VERSION: SQVersion = SQVersion.from_str("10.6")
 
 
 @dataclass(frozen=True)
@@ -55,6 +94,30 @@ def region_with_dot(region: str) -> str:
         return BaseUrls(base_url=sonar_host_url, api_base_url=api_base_url, is_sonar_qube_cloud=False)
 
 
+class BearerAuth(requests.auth.AuthBase):
+    def __init__(self, token):
+        self.token = token
+
+    def __call__(self, r):
+        r.headers["Authorization"] = f"Bearer {self.token}"
+        return r
+
+
 class SonarQubeApi:
-    def __init__(self, base_urls: BaseUrls):
-        self.base_url = base_urls
+    def __init__(self, base_urls: BaseUrls, token: str):
+        self.base_urls = base_urls
+        self.auth = BearerAuth(token)
+
+    def is_sonar_qube_cloud(self) -> bool:
+        return self.base_urls.is_sonar_qube_cloud
+
+    def get_analysis_version(self) -> SQVersion:
+        try:
+            res = requests.get(f"{self.base_urls.api_base_url}/analysis/version", auth=self.auth)
+            if res.status_code != 200:
+                res = requests.get(f"{self.base_urls.base_url}/api/server/version", auth=self.auth)
+
+            res.raise_for_status()
+            return SQVersion.from_str(res.text)
+        except requests.RequestException as e:
+            raise SonarQubeApiException("Error while fetching the analysis version") from e

src/pysonar_scanner/api.py

@@ -83,6 +83,7 @@ class Sonar:
     token: str = ""
     host_url: str = ""
     region: str = ""
+    token: str = ""
 
 
 @dataclass(frozen=True)

src/pysonar_scanner/configuration.py

@@ -0,0 +1,27 @@
+#
+# Sonar Scanner Python
+# Copyright (C) 2011-2024 SonarSource SA.
+# mailto:info AT sonarsource DOT com
+#
+# This program is free software; you can redistribute it and/or
+# modify it under the terms of the GNU Lesser General Public
+# License as published by the Free Software Foundation; either
+# version 3 of the License, or (at your option) any later version.
+# This program is distributed in the hope that it will be useful,
+#
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+# Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public License
+# along with this program; if not, write to the Free Software Foundation,
+# Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
+#
+
+
+class SonarQubeApiException(Exception):
+    pass
+
+
+class SQTooOldException(Exception):
+    pass

src/pysonar_scanner/exceptions.py

@@ -17,5 +17,20 @@
 # along with this program; if not, write to the Free Software Foundation,
 # Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
 #
+from pysonar_scanner.api import SonarQubeApi
+import pysonar_scanner.api as api
+from pysonar_scanner.exceptions import SQTooOldException
+
+
 class ScannerEngine:
-    pass
+    def __init__(self, api: SonarQubeApi):
+        self.api = api
+
+    def __version_check(self):
+        if self.api.is_sonar_qube_cloud():
+            return
+        version = self.api.get_analysis_version()
+        if not version.does_support_bootstrapping():
+            raise SQTooOldException(
+                f"Only SonarQube versions >= {api.MIN_SUPPORTED_SQ_VERSION} are supported, but got {version}"
+            )

src/pysonar_scanner/scannerengine.py

@@ -17,5 +17,7 @@
 # along with this program; if not, write to the Free Software Foundation,
 # Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
 #
+
+
 def remove_trailing_slash(url: str) -> str:
     return url.rstrip("/ ").lstrip()

src/pysonar_scanner/utils.py

@@ -0,0 +1,61 @@
+#
+# Sonar Scanner Python
+# Copyright (C) 2011-2024 SonarSource SA.
+# mailto:info AT sonarsource DOT com
+#
+# This program is free software; you can redistribute it and/or
+# modify it under the terms of the GNU Lesser General Public
+# License as published by the Free Software Foundation; either
+# version 3 of the License, or (at your option) any later version.
+# This program is distributed in the hope that it will be useful,
+#
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+# Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public License
+# along with this program; if not, write to the Free Software Foundation,
+# Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
+#
+import contextlib
+from typing import Optional
+from typing_extensions import Self
+from pysonar_scanner.api import BaseUrls, SonarQubeApi
+import responses
+
+
+def get_sq_server() -> SonarQubeApi:
+    return SonarQubeApi(
+        base_urls=BaseUrls(base_url="http://sq.home", api_base_url="http://sq.home/api/v2", is_sonar_qube_cloud=False),
+        token="<fake_token>",
+    )
+
+
+def get_sq_cloud() -> SonarQubeApi:
+    return SonarQubeApi(
+        base_urls=BaseUrls(
+            base_url="http://sonarcloud.io", api_base_url="http://api.sonarcloud.io", is_sonar_qube_cloud=True
+        ),
+        token="<fake_token>",
+    )
+
+
+class SQApiMocker:
+    def __init__(self, base_url: str = "http://sq.home", rsps: Optional[responses.RequestsMock] = None):
+        self.base_url = base_url
+        self.api_url = f"{base_url}/api/v2"
+        self.rsps = rsps or responses
+
+    def mock_analysis_version(self, version: str = "", status: int = 200) -> Self:
+        self.rsps.get(url=f"{self.api_url}/analysis/version", body=version, status=status)
+        return self
+
+    def mock_server_version(self, version: str = "", status: int = 200) -> Self:
+        self.rsps.get(url=f"{self.base_url}/api/server/version", body=version, status=status)
+        return self
+
+
+@contextlib.contextmanager
+def sq_api_mocker(base_url: str = "http://sq.home"):
+    with responses.RequestsMock() as rsps:
+        yield SQApiMocker(base_url=base_url, rsps=rsps)

tests/sq_api_utils.py

@@ -20,11 +20,40 @@
 from typing import TypedDict
 import unittest
 
-from pysonar_scanner.api import BaseUrls, get_base_urls
+from pysonar_scanner.api import BaseUrls, SonarQubeApi, SonarQubeApiException, get_base_urls
 from pysonar_scanner.configuration import Configuration, Scanner, Sonar
 
+from pysonar_scanner.api import SQVersion
+from tests import sq_api_utils
+from tests.sq_api_utils import sq_api_mocker
 
-class ApiTest(unittest.TestCase):
+
+class TestSQVersion(unittest.TestCase):
+    def test_does_support_bootstrapping(self):
+        self.assertTrue(SQVersion.from_str("10.6").does_support_bootstrapping())
+        self.assertTrue(SQVersion.from_str("10.6.5").does_support_bootstrapping())
+        self.assertTrue(SQVersion.from_str("10.6.5a").does_support_bootstrapping())
+        self.assertTrue(SQVersion.from_str("10.7").does_support_bootstrapping())
+        self.assertTrue(SQVersion.from_str("11.1").does_support_bootstrapping())
+        self.assertTrue(SQVersion.from_str("11.1.1").does_support_bootstrapping())
+        self.assertTrue(SQVersion.from_str("11").does_support_bootstrapping())
+
+        self.assertFalse(SQVersion.from_str("9.9.9").does_support_bootstrapping())
+        self.assertFalse(SQVersion.from_str("9.9.9a").does_support_bootstrapping())
+        self.assertFalse(SQVersion.from_str("9.9").does_support_bootstrapping())
+        self.assertFalse(SQVersion.from_str("9").does_support_bootstrapping())
+        self.assertFalse(SQVersion.from_str("10.5").does_support_bootstrapping())
+        self.assertFalse(SQVersion.from_str("10").does_support_bootstrapping())
+
+        self.assertFalse(SQVersion.from_str("a").does_support_bootstrapping())
+        self.assertFalse(SQVersion.from_str("1.a").does_support_bootstrapping())
+
+    def test_str(self):
+        self.assertEqual(str(SQVersion.from_str("10.6")), "10.6")
+        self.assertEqual(str(SQVersion.from_str("9.9.9aa")), "9.9.9aa")
+
+
+class TestApi(unittest.TestCase):
     def test_BaseUrls_normalization(self):
         self.assertEqual(
             BaseUrls("test1/", "test2/", is_sonar_qube_cloud=True),
@@ -136,3 +165,44 @@ class TestCaseDict(TypedDict):
                 self.assertEqual(base_urls.api_base_url, expected.api_base_url)
                 self.assertEqual(base_urls.is_sonar_qube_cloud, expected.is_sonar_qube_cloud)
                 self.assertEqual(base_urls, expected)
+
+
+class TestSonarQubeApiWithUnreachableSQServer(unittest.TestCase):
+    def setUp(self):
+        self.sq = SonarQubeApi(
+            base_urls=BaseUrls("https://localhost:1000", "https://localhost:1000/api", is_sonar_qube_cloud=True),
+            token="<invalid_token>",
+        )
+
+    def test_get_analysis_version(self):
+        with self.assertRaises(SonarQubeApiException):
+            self.sq.get_analysis_version()
+
+
+class TestSonarQubeApi(unittest.TestCase):
+    def setUp(self):
+        self.sq = sq_api_utils.get_sq_server()
+
+    def test_get_analysis_version(self):
+        with self.subTest("/analysis/version returns 200"), sq_api_mocker() as mocker:
+            mocker.mock_analysis_version("10.7")
+            self.assertEqual(self.sq.get_analysis_version(), SQVersion.from_str("10.7"))
+
+        with self.subTest("/analysis/version returns error"), sq_api_mocker() as mocker:
+            mocker.mock_analysis_version(status=404).mock_server_version("10.8")
+            self.assertEqual(self.sq.get_analysis_version(), SQVersion.from_str("10.8"))
+
+        with (
+            self.subTest("both version endpoints return error"),
+            sq_api_mocker() as mocker,
+            self.assertRaises(SonarQubeApiException),
+        ):
+            mocker.mock_analysis_version(status=404).mock_server_version(status=404)
+            self.sq.get_analysis_version()
+
+        with (
+            self.subTest("request raises an exception"),
+            sq_api_mocker() as mocker,
+            self.assertRaises(SonarQubeApiException),
+        ):
+            self.sq.get_analysis_version()