diff options
Diffstat (limited to 'test/src/lib')
| -rw-r--r-- | test/src/lib/db.py | 6 | ||||
| -rw-r--r-- | test/src/lib/logging.py | 18 | ||||
| -rw-r--r-- | test/src/lib/net.py | 2 | ||||
| -rw-r--r-- | test/src/lib/process.py | 34 | ||||
| -rw-r--r-- | test/src/lib/test_repo.py | 62 | ||||
| -rw-r--r-- | test/src/lib/tests.py | 8 |
6 files changed, 65 insertions, 65 deletions
diff --git a/test/src/lib/db.py b/test/src/lib/db.py index de6960a..0e47901 100644 --- a/test/src/lib/db.py +++ b/test/src/lib/db.py @@ -10,8 +10,8 @@ import sqlite3 class Database: def __init__(self, path): - logging.info('Opening SQLite database: %s', path) - self.conn = sqlite3.connect(f'file:{path}?mode=ro', uri=True) + logging.info("Opening SQLite database: %s", path) + self.conn = sqlite3.connect(f"file:{path}?mode=ro", uri=True) def __enter__(self): return self @@ -26,5 +26,5 @@ class Database: def get_all_runs(self): with self.get_cursor() as cur: - cur.execute('SELECT * FROM cimple_runs_view') + cur.execute("SELECT * FROM cimple_runs_view") return cur.fetchall() diff --git a/test/src/lib/logging.py b/test/src/lib/logging.py index 663eb3a..aed0eeb 100644 --- a/test/src/lib/logging.py +++ b/test/src/lib/logging.py @@ -13,7 +13,7 @@ import multiprocessing as mp @contextmanager def child_logging_thread(): # Delegating logging to the parent logger. - ctx = mp.get_context('spawn') + ctx = mp.get_context("spawn") queue = ctx.Queue() listener = logging.handlers.QueueListener(queue, logging.getLogger()) listener.start() @@ -26,16 +26,16 @@ def child_logging_thread(): @contextmanager def configure_logging_in_child(queue): config = { - 'version': 1, - 'handlers': { - 'sink': { - 'class': 'logging.handlers.QueueHandler', - 'queue': queue, + "version": 1, + "handlers": { + "sink": { + "class": "logging.handlers.QueueHandler", + "queue": queue, }, }, - 'root': { - 'handlers': ['sink'], - 'level': 'DEBUG', + "root": { + "handlers": ["sink"], + "level": "DEBUG", }, } logging.config.dictConfig(config) diff --git a/test/src/lib/net.py b/test/src/lib/net.py index 06bfda0..96979f7 100644 --- a/test/src/lib/net.py +++ b/test/src/lib/net.py @@ -10,6 +10,6 @@ import socket def random_unused_port(): with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as sock: sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) - sock.bind(('0.0.0.0', 0)) + sock.bind(("0.0.0.0", 0)) port = sock.getsockname()[1] return port diff --git a/test/src/lib/process.py b/test/src/lib/process.py index cea109d..72a13c6 100644 --- a/test/src/lib/process.py +++ b/test/src/lib/process.py @@ -21,7 +21,7 @@ class LoggingEvent(Event): def wait(self): if not super().wait(self.timeout): - raise RuntimeError('timed out while waiting for an event') + raise RuntimeError("timed out while waiting for an event") class LoggingThread(Thread): @@ -41,8 +41,8 @@ class LoggingThread(Thread): def process_output_lines(self): for line in self.process.stdout: - line = line.removesuffix('\n') - logging.info('%s: %s', self.process.log_id, line) + line = line.removesuffix("\n") + logging.info("%s: %s", self.process.log_id, line) with self.events_lock: for event in self.events: if event.is_set(): @@ -66,7 +66,7 @@ class CmdLine: @staticmethod def unbuffered(): - return CmdLine('stdbuf', '-o0') + return CmdLine("stdbuf", "-o0") def __init__(self, binary, *args, name=None): binary = self.which(binary) @@ -104,7 +104,7 @@ class LoggingEventProcessReady(LoggingEvent): super().__init__() def set(self): - logging.info('Process %s is ready', self.process.log_id) + logging.info("Process %s is ready", self.process.log_id) super().set() def log_line_matches(self, line): @@ -113,15 +113,15 @@ class LoggingEventProcessReady(LoggingEvent): class Process(subprocess.Popen): _COMMON_ARGS = { - 'text': True, - 'stdin': subprocess.DEVNULL, - 'stdout': subprocess.PIPE, - 'stderr': subprocess.STDOUT, + "text": True, + "stdin": subprocess.DEVNULL, + "stdout": subprocess.PIPE, + "stderr": subprocess.STDOUT, } @staticmethod def _log_process_start(argv): - logging.info('Executing command: %s', argv) + logging.info("Executing command: %s", argv) @staticmethod def _log_process_end(argv, ec, output): @@ -129,11 +129,11 @@ class Process(subprocess.Popen): if ec: log = logging.error if ec: - log('Command %s exited with code %s', argv, ec) + log("Command %s exited with code %s", argv, ec) else: - log('Command %s completed successfully', argv) + log("Command %s completed successfully", argv) if output: - log('Output:\n%s', output) + log("Output:\n%s", output) @staticmethod def run(*args, **kwargs): @@ -163,7 +163,7 @@ class Process(subprocess.Popen): self._log_process_start(argv) super().__init__(argv, **Process._COMMON_ARGS) - logging.info('Process %s has started', self.log_id) + logging.info("Process %s has started", self.log_id) ready_event = LoggingEventProcessReady(self) self.logger = LoggingThread(self, [ready_event]) @@ -171,7 +171,7 @@ class Process(subprocess.Popen): @property def log_id(self): - return f'{self.pid}/{self.cmd_line.process_name}' + return f"{self.pid}/{self.cmd_line.process_name}" def __exit__(self, *args): try: @@ -189,13 +189,13 @@ class Process(subprocess.Popen): ec = self.poll() if ec is not None: return - logging.info('Terminating process %s', self.log_id) + logging.info("Terminating process %s", self.log_id) self.terminate() try: self.wait(timeout=Process.SHUT_DOWN_TIMEOUT_SEC) return except subprocess.TimeoutExpired: pass - logging.info('Process %s failed to terminate in time, killing it', self.log_id) + logging.info("Process %s failed to terminate in time, killing it", self.log_id) self.kill() self.wait(timeout=Process.SHUT_DOWN_TIMEOUT_SEC) diff --git a/test/src/lib/test_repo.py b/test/src/lib/test_repo.py index f295e91..063043d 100644 --- a/test/src/lib/test_repo.py +++ b/test/src/lib/test_repo.py @@ -15,43 +15,43 @@ from .process import Process class Repo: - BRANCH = 'main' + BRANCH = "main" def __init__(self, path): self.path = os.path.abspath(path) os.makedirs(path, exist_ok=True) - self.run('git', 'init', '-q', f'--initial-branch={Repo.BRANCH}') - self.run('git', 'config', 'user.name', 'Test User') - self.run('git', 'config', 'user.email', 'test@example.com') + self.run("git", "init", "-q", f"--initial-branch={Repo.BRANCH}") + self.run("git", "config", "user.name", "Test User") + self.run("git", "config", "user.email", "test@example.com") def run(self, *args, **kwargs): Process.run(*args, cwd=self.path, **kwargs) -CI_SCRIPT = r'''#!/usr/bin/env bash +CI_SCRIPT = r"""#!/usr/bin/env bash set -o errexit -o nounset -o pipefail readonly runs_dir={runs_dir} readonly run_output_template=run_XXXXXX run_output_path="$( mktemp --tmpdir="$runs_dir" "$run_output_template" )" touch -- "$run_output_path" -''' +""" class TestRepo(Repo): # Prevent Pytest from discovering test cases in this class: __test__ = False - def __init__(self, path, ci_script='ci'): + def __init__(self, path, ci_script="ci"): super().__init__(path) - self.runs_dir = os.path.join(self.path, 'runs') + self.runs_dir = os.path.join(self.path, "runs") os.makedirs(self.runs_dir, exist_ok=True) self.ci_script_path = os.path.join(self.path, ci_script) self.write_ci_script() - self.run('git', 'add', '--', ci_script) - self.run('git', 'commit', '-q', '-m', 'add CI script') + self.run("git", "add", "--", ci_script) + self.run("git", "commit", "-q", "-m", "add CI script") @staticmethod @abc.abstractmethod @@ -67,7 +67,7 @@ class TestRepo(Repo): pass def write_ci_script(self): - with open(self.ci_script_path, mode='x') as f: + with open(self.ci_script_path, mode="x") as f: f.write(self.format_ci_script()) os.chmod(self.ci_script_path, 0o755) @@ -91,25 +91,25 @@ class TestRepo(Repo): class TestRepoOutput(TestRepo, abc.ABC): __test__ = False - OUTPUT_SCRIPT_NAME = 'generate-output' + OUTPUT_SCRIPT_NAME = "generate-output" def __init__(self, path): self.output_script_path = os.path.join(path, TestRepoOutput.OUTPUT_SCRIPT_NAME) super().__init__(path) self.write_output_script() - self.run('git', 'add', '--', TestRepoOutput.OUTPUT_SCRIPT_NAME) - self.run('git', 'commit', '-q', '-m', 'add output script') + self.run("git", "add", "--", TestRepoOutput.OUTPUT_SCRIPT_NAME) + self.run("git", "commit", "-q", "-m", "add output script") def format_ci_script(self): script = super().format_ci_script() added = r'{output_script} | tee -a "$run_output_path"'.format( output_script=shlex.quote(self.output_script_path) ) - return f'{script}\n{added}\n' + return f"{script}\n{added}\n" def write_output_script(self): - with open(self.output_script_path, mode='x') as f: + with open(self.output_script_path, mode="x") as f: f.write(self.format_output_script()) os.chmod(self.output_script_path, 0o755) @@ -121,10 +121,10 @@ class TestRepoOutput(TestRepo, abc.ABC): return ec == 0 -OUTPUT_SCRIPT_SIMPLE = r'''#!/bin/sh -e +OUTPUT_SCRIPT_SIMPLE = r"""#!/bin/sh -e timestamp="$( date --iso-8601=ns )" echo "A CI run happened at $timestamp" -''' +""" class TestRepoOutputSimple(TestRepoOutput): @@ -132,17 +132,17 @@ class TestRepoOutputSimple(TestRepoOutput): @staticmethod def codename(): - return 'output_simple' + return "output_simple" def format_output_script(self): return OUTPUT_SCRIPT_SIMPLE def run_output_matches(self, output): - return output.decode().startswith('A CI run happened at ') + return output.decode().startswith("A CI run happened at ") -OUTPUT_SCRIPT_EMPTY = r'''#!/bin/sh -''' +OUTPUT_SCRIPT_EMPTY = r"""#!/bin/sh +""" class TestRepoOutputEmpty(TestRepoOutput): @@ -150,7 +150,7 @@ class TestRepoOutputEmpty(TestRepoOutput): @staticmethod def codename(): - return 'output_empty' + return "output_empty" def format_output_script(self): return OUTPUT_SCRIPT_EMPTY @@ -159,9 +159,9 @@ class TestRepoOutputEmpty(TestRepoOutput): return len(output) == 0 -OUTPUT_SCRIPT_LONG = r'''#!/bin/sh -e +OUTPUT_SCRIPT_LONG = r"""#!/bin/sh -e dd if=/dev/urandom count={output_len_kb} bs=1024 | base64 -''' +""" class TestRepoOutputLong(TestRepoOutput): @@ -171,7 +171,7 @@ class TestRepoOutputLong(TestRepoOutput): @staticmethod def codename(): - return 'output_long' + return "output_long" def format_output_script(self): output_len = TestRepoOutputLong.OUTPUT_LEN_KB @@ -182,18 +182,18 @@ class TestRepoOutputLong(TestRepoOutput): return len(output) > TestRepoOutputLong.OUTPUT_LEN_KB * 1024 -OUTPUT_SCRIPT_NULL = r'''#!/usr/bin/env python3 +OUTPUT_SCRIPT_NULL = r"""#!/usr/bin/env python3 output = {output} import sys sys.stdout.buffer.write(output) sys.exit(-2) -''' +""" class TestRepoOutputNull(TestRepoOutput): __test__ = False - OUTPUT = b'123\x00456\x00789' + OUTPUT = b"123\x00456\x00789" def __init__(self, *args, **kwargs): assert len(TestRepoOutputNull.OUTPUT) == 11 @@ -202,7 +202,7 @@ class TestRepoOutputNull(TestRepoOutput): @staticmethod def codename(): - return 'output_null' + return "output_null" def format_output_script(self): return OUTPUT_SCRIPT_NULL.format(output=repr(self.output)) @@ -223,7 +223,7 @@ class TestRepoSegfault(TestRepo): @staticmethod def codename(): - return 'segfault' + return "segfault" def write_ci_script(self): shutil.copy(self.prog_path, self.ci_script_path) diff --git a/test/src/lib/tests.py b/test/src/lib/tests.py index b20ecc5..c71a3ea 100644 --- a/test/src/lib/tests.py +++ b/test/src/lib/tests.py @@ -9,17 +9,17 @@ import pytest # Reference: https://github.com/pytest-dev/pytest/issues/3628 # Automatic generation of readable test IDs. def my_parametrize(names, values, ids=None, **kwargs): - _names = names.split(',') if isinstance(names, str) else names + _names = names.split(",") if isinstance(names, str) else names if not ids: if len(_names) == 1: - ids = [f'{names}={v}' for v in values] + ids = [f"{names}={v}" for v in values] else: _values = [ - combination.values if hasattr(combination, 'values') else combination + combination.values if hasattr(combination, "values") else combination for combination in values ] ids = [ - '-'.join(f'{k}={v}' for k, v in zip(_names, combination)) + "-".join(f"{k}={v}" for k, v in zip(_names, combination)) for combination in _values ] return pytest.mark.parametrize(names, values, ids=ids, **kwargs) |