PYSCAN-24: Add test for scanner (#2)

Author: joke1196

pyproject.toml

@@ -65,6 +65,14 @@ dependencies = [
 license = 'python -m licenseheaders -t license_header.tmpl -o "SonarSource SA" -y 2011-2023 -n "Sonar Scanner Python" -E .py' 
 format = 'python -m black src/ tests/ --check'
 
+[tool.pytest.ini_options]
+pythonpath = [
+  "src",
+]
+addopts = [
+  "--import-mode=importlib",
+]
+
 [tool.black]
 line-length = 89
 target-version = ["py38", "py39", "py310", "py311", "py312"]

src/py_sonar_scanner/scanner.py

@@ -17,8 +17,9 @@
 # along with this program; if not, write to the Free Software Foundation,
 # Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
 #
-import subprocess
 import threading
+from threading import Thread
+from subprocess import Popen, PIPE
 
 from py_sonar_scanner.configuration import Configuration
 
@@ -30,22 +31,30 @@ def __init__(self, cfg: Configuration):
         self.cfg = cfg
 
     def scan(self):
-        cmd = [self.cfg.sonar_scanner_executable_path] + self.cfg.scan_arguments
-        process = subprocess.Popen(cmd, stdout=subprocess.PIPE,
-                                   stderr=subprocess.PIPE)
+        process = self.execute_command()
+        output_thread = threading.Thread(target=self._print_output, args=(process.stdout,))
+        error_thread = threading.Thread(target=self._print_output, args=(process.stderr,))
+        return self.process_output(output_thread, error_thread, process)
 
-        def print_output(stream):
-            for line in stream:
-                decoded_line = line.decode('utf-8')
-                print(decoded_line, end='', flush=True)
-
-        output_thread = threading.Thread(target=print_output, args=(process.stdout,))
-        error_thread = threading.Thread(target=print_output, args=(process.stderr,))
+    def process_output(self, output_thread: Thread, error_thread: Thread, process: Popen) -> int:
         output_thread.start()
         error_thread.start()
-
         process.wait()
         output_thread.join()
         error_thread.join()
 
         return process.returncode
+
+    def compute_command(self) -> list[str]:
+        if not self.cfg.sonar_scanner_executable_path:
+            raise ValueError("No executable path provided")
+        return [self.cfg.sonar_scanner_executable_path] + self.cfg.scan_arguments
+
+    def execute_command(self) -> Popen:
+        cmd = self.compute_command()
+        return Popen(cmd, stdout=PIPE, stderr=PIPE)
+
+    def _print_output(self, stream: list[str]):
+        for line in stream:
+            decoded_line = line.decode('utf-8')
+            print(decoded_line, end='', flush=True)

tests/test_placeholder.py

@@ -0,0 +1,59 @@
+import unittest
+from unittest.mock import Mock
+import threading
+from py_sonar_scanner.scanner import Scanner
+from py_sonar_scanner.configuration import Configuration
+
+
+class TestScanner(unittest.TestCase):
+
+    def test_scanner_compute_command(self):
+        cfg = Configuration()
+        cfg.sonar_scanner_executable_path = "test"
+        cfg.scan_arguments = ["a", "b"]
+        scanner = Scanner(cfg)
+
+        assert scanner.compute_command() == ["test", "a", "b"]
+
+        cfg.sonar_scanner_executable_path = "test"
+        cfg.scan_arguments = []
+        scanner = Scanner(cfg)
+
+        assert scanner.compute_command() == ["test"]
+
+        cfg.sonar_scanner_executable_path = ""
+        cfg.scan_arguments = []
+        scanner = Scanner(cfg)
+
+        self.assertRaises(ValueError, scanner.compute_command)
+
+    def test_process_output(self):
+        scanner = Scanner(Configuration())
+        output_thread = threading.Thread()
+        error_thread = threading.Thread()
+
+        process = Mock()
+        output_thread.start = Mock()
+        error_thread.start = Mock()
+        output_thread.join = Mock()
+        error_thread.join = Mock()
+        process.wait = Mock()
+
+        return_code = scanner.process_output(output_thread, error_thread, process)
+
+        output_thread.start.assert_called_once()
+        error_thread.start.assert_called_once()
+        output_thread.join.assert_called_once()
+        error_thread.join.assert_called_once()
+        process.wait.assert_called_once()
+
+    def test_scan(self):
+        scanner = Scanner(Configuration())
+
+        scanner.execute_command = Mock()
+        scanner.process_output = Mock()
+
+        scanner.scan()
+
+        scanner.execute_command.assert_called_once()
+        scanner.process_output.assert_called_once()