SCANPY-102 New bootstrapping: Execute the Scan (#148)

Author: thomas-serre-sonarsource

src/pysonar_scanner/__main__.py

@@ -18,6 +18,22 @@
 # Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
 #
 
+from pysonar_scanner import cache
+from pysonar_scanner.configuration import Configuration, ConfigurationLoader
+from pysonar_scanner.api import get_base_urls, SonarQubeApi
+from pysonar_scanner.scannerengine import ScannerEngine
+from pysonar_scanner.cache import Cache
+from pysonar_scanner.jre import JREProvisioner, JREResolvedPath, JREResolver
+
 
 def scan():
-    pass
+    configuration = ConfigurationLoader().initialize_configuration()
+    cache_manager = cache.get_default()
+    api = __build_api(configuration)
+    scanner = ScannerEngine(api, cache_manager)
+    return scanner.run(configuration)
+
+
+def __build_api(configuration) -> SonarQubeApi:
+    base_urls = get_base_urls(configuration)
+    return SonarQubeApi(base_urls, configuration.sonar.token)

src/pysonar_scanner/scannerengine.py

@@ -17,32 +17,42 @@
 # along with this program; if not, write to the Free Software Foundation,
 # Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
 #
+import json
+import pathlib
+from typing import Optional
+
 import pysonar_scanner.api as api
+
 from pysonar_scanner.api import SonarQubeApi
 from pysonar_scanner.cache import Cache, CacheFile
 from pysonar_scanner.exceptions import ChecksumException, SQTooOldException
+from pysonar_scanner.configuration import Configuration
+from pysonar_scanner.jre import JREProvisioner, JREResolvedPath, JREResolver
+from subprocess import Popen, PIPE
 
 
 class ScannerEngineProvisioner:
     def __init__(self, api: SonarQubeApi, cache: Cache):
         self.api = api
         self.cache = cache
 
-    def provision(self) -> None:
-        if self.__download_and_verify():
-            return
+    def provision(self) -> pathlib.Path:
+        scanner_file = self.__download_and_verify()
+        if scanner_file is not None:
+            return scanner_file.filepath
         # Retry once in case the checksum failed due to the scanner engine being updated between getting the checksum and downloading the jar
-        if self.__download_and_verify():
-            return
+        scanner_file = self.__download_and_verify()
+        if scanner_file is not None:
+            return scanner_file.filepath
         else:
             raise ChecksumException("Failed to download and verify scanner engine")
 
-    def __download_and_verify(self) -> bool:
+    def __download_and_verify(self) -> Optional[CacheFile]:
         engine_info = self.api.get_analysis_engine()
         cache_file = self.cache.get_file(engine_info.filename, engine_info.sha256)
         if not cache_file.is_valid():
             self.__download_scanner_engine(cache_file)
-        return cache_file.is_valid()
+        return cache_file if cache_file.is_valid() else None
 
     def __download_scanner_engine(self, cache_file: CacheFile) -> None:
         with cache_file.open(mode="wb") as f:
@@ -54,6 +64,46 @@ def __init__(self, api: SonarQubeApi, cache: Cache):
         self.api = api
         self.cache = cache
 
+    def __fetch_scanner_engine(self) -> pathlib.Path:
+        return ScannerEngineProvisioner(self.api, self.cache).provision()
+
+    def run(self, configuration: Configuration):
+        self.__version_check()
+        jre_path = self.__resolve_jre(configuration)
+        scanner_engine_path = self.__fetch_scanner_engine()
+        cmd = self.__build_command(jre_path, scanner_engine_path)
+        return self.__execute_scanner_engine(configuration, cmd)
+
+    def __build_command(self, jre_path: JREResolvedPath, scanner_engine_path: pathlib.Path) -> list[str]:
+        cmd = []
+        cmd.append(jre_path.path)
+        cmd.append("-jar")
+        cmd.append(scanner_engine_path)
+        return cmd
+
+    def __execute_scanner_engine(self, configuration: Configuration, cmd: list[str]) -> int:
+        popen = Popen(cmd, stdout=PIPE, stderr=PIPE, stdin=PIPE)
+        outs, _ = popen.communicate(configuration.to_json().encode())
+        exitcode = popen.wait()  # 0 means success
+        if exitcode != 0:
+            errors = self.__extract_errors_from_log(outs)
+            raise RuntimeError(f"Scan failed with exit code {exitcode}", errors)
+        return exitcode
+
+    def __extract_errors_from_log(self, outs: str) -> list[str]:
+        try:
+            errors = []
+            for line in outs.decode("utf-8").split("\n"):
+                if line.strip() == "":
+                    continue
+                out_json = json.loads(line)
+                if out_json["level"] == "ERROR":
+                    errors.append(out_json["message"])
+            return errors
+        except Exception as e:
+            print(e)
+            return []
+
     def __version_check(self):
         if self.api.is_sonar_qube_cloud():
             return
@@ -63,5 +113,7 @@ def __version_check(self):
                 f"Only SonarQube versions >= {api.MIN_SUPPORTED_SQ_VERSION} are supported, but got {version}"
             )
 
-    def __fetch_scanner_engine(self):
-        ScannerEngineProvisioner(self.api, self.cache).provision()
+    def __resolve_jre(self, configuration: Configuration) -> JREResolvedPath:
+        jre_provisionner = JREProvisioner(self.api, self.cache)
+        jre_resolver = JREResolver(configuration, jre_provisionner)
+        return jre_resolver.resolve_jre()

src/pysonar_scanner/utils.py

@@ -20,7 +20,6 @@
 import hashlib
 import pathlib
 import platform
-import re
 import typing
 from enum import Enum
 

tests/test_main.py

@@ -0,0 +1,66 @@
+#
+# 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 unittest
+from unittest import mock
+from unittest.mock import patch
+
+from pysonar_scanner.jre import JRE, JREResolvedPath
+from pysonar_scanner.__main__ import scan
+from pysonar_scanner.scannerengine import ScannerEngine
+
+
+class TestMain(unittest.TestCase):
+
+    @patch("pysonar_scanner.scannerengine.Popen")
+    @patch.object(ScannerEngine, "_ScannerEngine__resolve_jre")
+    @patch.object(ScannerEngine, "_ScannerEngine__fetch_scanner_engine")
+    @patch(
+        "sys.argv",
+        ["pysonar-scanner", "-t", "myToken", "--sonar-project-key", "myProjectKey"],
+    )
+    def test_minimal_run_success(self, mock_fetch_scanner_engine, mock_resolve_jre, mock_popen):
+        mock_resolve_jre.return_value = JREResolvedPath("")
+        mock_fetch_scanner_engine.return_value = None
+        process_mock = mock.Mock()
+        attrs = {"communicate.return_value": ("output", "error"), "wait.return_value": 0}
+        process_mock.configure_mock(**attrs)
+        mock_popen.return_value = process_mock
+
+        exitcode = scan()
+        self.assertEqual(exitcode, 0)
+
+    @patch("pysonar_scanner.scannerengine.Popen")
+    @patch.object(ScannerEngine, "_ScannerEngine__resolve_jre")
+    @patch.object(ScannerEngine, "_ScannerEngine__fetch_scanner_engine")
+    @patch(
+        "sys.argv",
+        ["pysonar-scanner", "-t", "myToken", "--sonar-project-key", "myProjectKey"],
+    )
+    def test_minimal_run_failure(self, mock_fetch_scanner_engine, mock_resolve_jre, mock_popen):
+        mock_resolve_jre.return_value = JREResolvedPath("")
+        process_mock = mock.Mock()
+        mock_fetch_scanner_engine.return_value = None
+        attrs = {"communicate.return_value": ("output", "error"), "wait.return_value": 2}
+        process_mock.configure_mock(**attrs)
+        mock_popen.return_value = process_mock
+        with self.assertRaises(RuntimeError) as error:
+            scan()
+        print(str(error))
+        self.assertIn("Scan failed with exit code 2", str(error.exception))

tests/test_scannerengine.py

@@ -17,8 +17,8 @@
 # along with this program; if not, write to the Free Software Foundation,
 # Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
 #
-import pathlib
 import unittest
+import pathlib
 import pyfakefs.fake_filesystem_unittest as pyfakefs
 
 from pysonar_scanner import cache
@@ -30,7 +30,35 @@
 from tests import sq_api_utils
 
 
-class TestScannerEngine(pyfakefs.TestCase):
+class TestScannerEngine(unittest.TestCase):
+
+    def test_error_log_extraction(self):
+        log_failure_example = b"""{"level":"INFO","message":"CPD Executor CPD calculation finished (done) | time=15ms"}\n
+            {"level":"INFO","message":"SCM revision ID \'a53e6a3193a049d0f77fc2ff16cf52e7a66c7adb\'"}\n
+            {"level":"INFO","message":"Analysis report generated in 152ms, dir size=760.6 kB"}\n
+            {"level":"INFO","message":"Analysis report compressed in 83ms, zip size=321.7 kB"}\n
+            {"level":"ERROR","message":"You\'re not authorized to analyze this project or the project doesn\'t exist on SonarQube and you\'re not authorized to create it. Please contact an administrator."}\n"""
+
+        scanner = ScannerEngine(None, None)
+        expected = "You're not authorized to analyze this project or the project doesn't exist on SonarQube and you're not authorized to create it. Please contact an administrator."
+        errors = scanner._ScannerEngine__extract_errors_from_log(log_failure_example)
+        print(errors)
+        self.assertEqual(
+            errors,
+            [expected],
+        )
+
+    def test_exception_in_error_log_extraction(self):
+        unexpected_log_failure_example = b"""unexpected log format'"""
+        scanner = ScannerEngine(None, None)
+        errors = scanner._ScannerEngine__extract_errors_from_log(unexpected_log_failure_example)
+        self.assertEqual(
+            errors,
+            [],
+        )
+
+
+class TestScannerEngineWithFake(pyfakefs.TestCase):
     def setUp(self):
         self.setUpPyfakefs()