From 939d260b5385baefaaec4305935f0ed19e001b8c Mon Sep 17 00:00:00 2001 From: "F.Tibor" Date: Fri, 30 Jan 2026 14:06:37 +0100 Subject: [PATCH 1/3] Create a self cleaning codechecker server --- test/common/base.py | 77 ---------------------- test/common/codechecker_server.py | 105 ++++++++++++++++++++++++++++++ test/unit/parse/test_parse.py | 5 +- 3 files changed, 108 insertions(+), 79 deletions(-) create mode 100644 test/common/codechecker_server.py diff --git a/test/common/base.py b/test/common/base.py index ffc339c9..e9a5c299 100644 --- a/test/common/base.py +++ b/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""" @@ -175,42 +134,6 @@ def contains_regex_in_file(cls, file_path: str, regex: str) -> bool: """ 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, diff --git a/test/common/codechecker_server.py b/test/common/codechecker_server.py new file mode 100644 index 00000000..395675eb --- /dev/null +++ b/test/common/codechecker_server.py @@ -0,0 +1,105 @@ +# 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 + + +# Based on: +# https://dev.to/farcellier/wait-for-a-server-to-respond-in-python-488e +def wait_port( + port: int, + host: str = "localhost", + timeout: int = 3000, + attempt_every: int = 100, +) -> bool: + """ + Wait until a port would be open, + for example the port 8001 for CodeChecker server + """ + start = time.monotonic() + while True: + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: + try: + s.connect((host, port)) + s.close() + return True + except ConnectionRefusedError: + if timeout is not None and time.monotonic() - start > ( + timeout / 1000 + ): + return False + + time.sleep(attempt_every / 1000) + + +class CodeCheckerServer: + """ + CodeCheckerServer object for testing. + Cleans up after itself. + """ + def __init__(self, port="8001"): + self.running = False + self.port = 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 port 8001 + This server must be shutdown with stop_codechecker_sever + """ + if self.running: + return + server_command = [ + "CodeChecker", + "server", + "--workspace", + self.temp_workspace, + "--port", + 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_port( + port=8001, 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) diff --git a/test/unit/parse/test_parse.py b/test/unit/parse/test_parse.py index 0e6e137a..75780bec 100644 --- a/test/unit/parse/test_parse.py +++ b/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,13 +40,13 @@ 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 test_parse_html(self): From 46e4852c4f02ed8db9628b58af44be9e4eaaf253 Mon Sep 17 00:00:00 2001 From: "F.Tibor" Date: Tue, 5 May 2026 12:39:09 +0200 Subject: [PATCH 2/3] Update --- test/common/codechecker_server.py | 58 ++++++++++++++++++------------- 1 file changed, 33 insertions(+), 25 deletions(-) diff --git a/test/common/codechecker_server.py b/test/common/codechecker_server.py index 395675eb..b4c994dc 100644 --- a/test/common/codechecker_server.py +++ b/test/common/codechecker_server.py @@ -23,34 +23,41 @@ import subprocess import tempfile import time +import urllib +import urllib.request +import urllib.error -# Based on: -# https://dev.to/farcellier/wait-for-a-server-to-respond-in-python-488e -def wait_port( - port: int, +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", - timeout: int = 3000, + port: int = 8001, + timeout: int = 10000, attempt_every: int = 100, ) -> bool: """ - Wait until a port would be open, - for example the port 8001 for CodeChecker server + Wait until the product is available in the CodeChecker server """ start = time.monotonic() - while True: - with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: - try: - s.connect((host, port)) - s.close() - return True - except ConnectionRefusedError: - if timeout is not None and time.monotonic() - start > ( - timeout / 1000 - ): - return False - + 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: @@ -58,9 +65,10 @@ class CodeCheckerServer: CodeCheckerServer object for testing. Cleans up after itself. """ - def __init__(self, port="8001"): + + def __init__(self, port=None): self.running = False - self.port = port + self.port = port if port else _get_free_port() self.temp_workspace = tempfile.mkdtemp() self.start_codechecker_server() @@ -69,7 +77,7 @@ def __del__(self): def start_codechecker_server(self): """ - Starts a CodeChecker server instance on port 8001 + Starts a CodeChecker server instance on a free port This server must be shutdown with stop_codechecker_sever """ if self.running: @@ -80,7 +88,7 @@ def start_codechecker_server(self): "--workspace", self.temp_workspace, "--port", - self.port, + str(self.port), ] # These file/popen processes are closed when the object dies # pylint: disable=consider-using-with @@ -89,8 +97,8 @@ def start_codechecker_server(self): self.server_process: subprocess.Popen = subprocess.Popen( server_command, stdout=self.devnull ) - assert wait_port( - port=8001, timeout=10000 + assert wait_codechecker_server( + port=self.port, timeout=10000 ), "Failed to start CodeChecker server" self.running = True From 0eeedc35e2b2e87410aec7f7cc759ea054ccba5d Mon Sep 17 00:00:00 2001 From: "F.Tibor" Date: Tue, 5 May 2026 14:01:20 +0200 Subject: [PATCH 3/3] Move helper funcions to the test, since they are specific to this tests --- test/common/base.py | 28 ---------------------------- test/unit/parse/test_parse.py | 28 ++++++++++++++++++++++++++++ 2 files changed, 28 insertions(+), 28 deletions(-) diff --git a/test/common/base.py b/test/common/base.py index e9a5c299..8de63ff4 100644 --- a/test/common/base.py +++ b/test/common/base.py @@ -133,31 +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)) - - 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) diff --git a/test/unit/parse/test_parse.py b/test/unit/parse/test_parse.py index 75780bec..ba3a41d6 100644 --- a/test/unit/parse/test_parse.py +++ b/test/unit/parse/test_parse.py @@ -49,6 +49,34 @@ def tearDownClass(cls): 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(