Refactor codechecker_server

test/common/base.py

@@ -20,49 +20,12 @@
 import os
 import re
 import shlex
-import shutil
-import signal
-import socket
-import urllib.request
-import urllib.error
 import subprocess
-import tempfile
-import time
 import unittest
 import sys
 from typing import Optional
 
 
-def _get_free_port():
-    """
-    Return a port number that is free
-    """
-    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
-        s.bind(("", 0))
-        return s.getsockname()[1]
-
-def wait_codechecker_server(
-    product: str = "Default",
-    host: str = "localhost",
-    port: int = 8001,
-    timeout: int = 10000,
-    attempt_every: int = 100,
-) -> bool:
-    """
-    Wait until the product is available in the CodeChecker server
-    """
-    start = time.monotonic()
-    url = f"http://{host}:{port}/{product}"
-    while time.monotonic() - start < timeout:
-        try:
-            with urllib.request.urlopen(url, timeout=timeout / 1000) as resp:
-                if resp.getcode() == 200:
-                    return True
-        except (urllib.error.URLError, urllib.error.HTTPError):
-            pass
-        time.sleep(attempt_every / 1000)
-    return False
-
 
 class TestBase(unittest.TestCase):
     """Unittest base abstract class"""
@@ -110,10 +73,6 @@ def tearDownClass(cls):
         """Restore environment"""
         os.chdir(cls.save_cwd)
         os.environ = cls.save_env
-        try:
-            assert cls.server_process.poll() is not None, "Server not stopped"
-        except AttributeError:
-            pass  # if server_process is not set, everything is fine
 
     def setUp(self):
         """Before every test"""
@@ -174,67 +133,3 @@ def contains_regex_in_file(cls, file_path: str, regex: str) -> bool:
         Returns a boolean, whether the specified file contains the regex or not.
         """
         return bool(cls.grep_file(file_path, regex))
-
-    @classmethod
-    def start_codechecker_server(cls):
-        """
-        Starts a CodeChecker server instance on port 8001
-        This server must be shutdown with stop_codechecker_sever
-        """
-        cls.temp_workspace = tempfile.mkdtemp()
-        cls.port: int = _get_free_port()
-        server_command = [
-            "CodeChecker",
-            "server",
-            "--workspace",
-            cls.temp_workspace,
-            "--port",
-            str(cls.port),
-        ]
-        # pylint: disable=consider-using-with
-        cls.devnull = open(os.devnull, "w", encoding="utf-8")
-        # pylint: disable=consider-using-with
-        cls.server_process: subprocess.Popen = subprocess.Popen(
-            server_command, stdout=cls.devnull
-        )
-        assert wait_codechecker_server(
-            port=cls.port
-        ), "Failed to start CodeChecker server"
-
-    @classmethod
-    def stop_codechecker_server(cls):
-        """
-        Stops the CodeChecker server started by start_codechecker_server
-        """
-        os.kill(cls.server_process.pid, signal.SIGTERM)
-        cls.server_process.wait()
-        cls.devnull.close()
-        shutil.rmtree(cls.temp_workspace)
-
-    def check_store(self, path: str, name: str):
-        """
-        Tries to store the results on the codechecker server,
-        asserts for successful storing.
-
-        Args:
-            path - Path of the result files
-            name - name of the project to be saved under
-        """
-        port = getattr(self, 'port', 8001)
-        ret, stdout, stderr = self.run_command(
-            f"CodeChecker store {path} -n {name}"
-            f" --url=http://localhost:{port}/Default"
-        )
-        self.assertEqual(ret, 0, stdout + "\n" + stderr)
-
-    def check_parse(self, path: str, will_find_bug: bool = True):
-        """
-        Checks if the parse command finishes correctly on results.
-
-        Args:
-            path - Path of the result files
-            will_find_bug - Will there be a bug in the result files,
-            changes on what we assert
-        """
-        ret, _, _ = self.run_command(f"CodeChecker parse {path}")
-        self.assertEqual(ret, 2 if will_find_bug else 0)

test/common/codechecker_server.py

@@ -0,0 +1,113 @@
+# Copyright 2023 Ericsson AB
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""
+Codechecker server functionality, and related functions
+"""
+
+import os
+import shutil
+import signal
+import socket
+import subprocess
+import tempfile
+import time
+import urllib
+import urllib.request
+import urllib.error
+
+
+def _get_free_port():
+    """
+    Return a port number that is free
+    """
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+        s.bind(("", 0))
+        return s.getsockname()[1]
+
+
+def wait_codechecker_server(
+    product: str = "Default",
+    host: str = "localhost",
+    port: int = 8001,
+    timeout: int = 10000,
+    attempt_every: int = 100,
+) -> bool:
+    """
+    Wait until the product is available in the CodeChecker server
+    """
+    start = time.monotonic()
+    url = f"http://{host}:{port}/{product}"
+    while time.monotonic() - start < timeout:
+        try:
+            with urllib.request.urlopen(url, timeout=timeout / 1000) as resp:
+                if resp.getcode() == 200:
+                    return True
+        except (urllib.error.URLError, urllib.error.HTTPError):
+            pass
+        time.sleep(attempt_every / 1000)
+    return False
+
+
+class CodeCheckerServer:
+    """
+    CodeCheckerServer object for testing.
+    Cleans up after itself.
+    """
+
+    def __init__(self, port=None):
+        self.running = False
+        self.port = port if port else _get_free_port()
+        self.temp_workspace = tempfile.mkdtemp()
+        self.start_codechecker_server()
+
+    def __del__(self):
+        self.stop_codechecker_server()
+
+    def start_codechecker_server(self):
+        """
+        Starts a CodeChecker server instance on a free port 
+        This server must be shutdown with stop_codechecker_sever
+        """
+        if self.running:
+            return
+        server_command = [
+            "CodeChecker",
+            "server",
+            "--workspace",
+            self.temp_workspace,
+            "--port",
+            str(self.port),
+        ]
+        # These file/popen processes are closed when the object dies
+        # pylint: disable=consider-using-with
+        self.devnull = open(os.devnull, "w", encoding="utf-8")
+        # pylint: disable=consider-using-with
+        self.server_process: subprocess.Popen = subprocess.Popen(
+            server_command, stdout=self.devnull
+        )
+        assert wait_codechecker_server(
+            port=self.port, timeout=10000
+        ), "Failed to start CodeChecker server"
+        self.running = True
+
+    def stop_codechecker_server(self):
+        """
+        Stops the CodeChecker server started by start_codechecker_server
+        """
+        os.kill(self.server_process.pid, signal.SIGTERM)
+        self.server_process.wait()
+        self.running = False
+        self.devnull.close()
+        shutil.rmtree(self.temp_workspace)

test/unit/parse/test_parse.py

@@ -20,6 +20,7 @@
 import unittest
 from typing import final
 from common.base import TestBase
+from common.codechecker_server import CodeCheckerServer
 
 
 class TestTemplate(TestBase):
@@ -39,15 +40,43 @@ class TestTemplate(TestBase):
     def setUpClass(cls):
         """Start CodeChecker server"""
         super().setUpClass()
-        cls.start_codechecker_server()
+        cls.codechecker_server = CodeCheckerServer()
 
     @final
     @classmethod
     def tearDownClass(cls):
         """Stop CodeChecker server"""
-        cls.stop_codechecker_server()
+        del cls.codechecker_server
         super().tearDownClass()
 
+    def check_store(self, path: str, name: str):
+        """
+        Tries to store the results on the codechecker server,
+        asserts for successful storing.
+
+        Args:
+            path - Path of the result files
+            name - name of the project to be saved under
+        """
+        port = getattr(self.codechecker_server, 'port', 8001)
+        ret, stdout, stderr = self.run_command(
+            f"CodeChecker store {path} -n {name}"
+            f" --url=http://localhost:{port}/Default"
+        )
+        self.assertEqual(ret, 0, stdout + "\n" + stderr)
+
+    def check_parse(self, path: str, will_find_bug: bool = True):
+        """
+        Checks if the parse command finishes correctly on results.
+
+        Args:
+            path - Path of the result files
+            will_find_bug - Will there be a bug in the result files,
+            changes on what we assert
+        """
+        ret, _, _ = self.run_command(f"CodeChecker parse {path}")
+        self.assertEqual(ret, 2 if will_find_bug else 0)
+
     def test_parse_html(self):
         """Test: Parse results into html"""
         ret, _, stderr = self.run_command(