diff --git a/test/common/base.py b/test/common/base.py index ffc339c9..8de63ff4 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""" @@ -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) diff --git a/test/common/codechecker_server.py b/test/common/codechecker_server.py new file mode 100644 index 00000000..b4c994dc --- /dev/null +++ b/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) diff --git a/test/unit/parse/test_parse.py b/test/unit/parse/test_parse.py index 0e6e137a..ba3a41d6 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,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(