From 438cb4cb313ab939b4e94f2f40fc64fd03705a12 Mon Sep 17 00:00:00 2001 From: "Commentator2.0" Date: Sat, 24 May 2025 12:01:16 +0200 Subject: [PATCH] tests/functional2: move commands to own lib file Currently the Command and CommandResult classes are mixed into the nix fixture file. This commit moves them out into their own lib file, to make it more obvious that they can be used standalone for other applications too Additionally improved documentation of said classes and bumped log level of stdout and err on unexpected exitcodes, as it is within an error context Change-Id: If2d554acde86fd54f2445fc46453f06923af5fe9 --- tests/functional2/testlib/commands.py | 138 ++++++++++++++++++++++ tests/functional2/testlib/fixtures/nix.py | 90 +------------- 2 files changed, 139 insertions(+), 89 deletions(-) create mode 100644 tests/functional2/testlib/commands.py diff --git a/tests/functional2/testlib/commands.py b/tests/functional2/testlib/commands.py new file mode 100644 index 000000000..7275c5e51 --- /dev/null +++ b/tests/functional2/testlib/commands.py @@ -0,0 +1,138 @@ +import dataclasses +import json +import logging +import subprocess +from pathlib import Path +from typing import Any + +from functional2.testlib.terminal_code_eater import eat_terminal_codes + +logger = logging.getLogger(__name__) + + +@dataclasses.dataclass +class CommandResult: + cmd: list[str] + """Command arguments which were run""" + rc: int + """Return code""" + stderr: bytes + """Outputted stderr""" + stdout: bytes + """Outputted stdout""" + + def ok(self) -> "CommandResult": + """ + assumes a return code of 0 + :raises CalledProcessError: if the return code wasn't 0 and logs the processes stdout and stderr + """ + if self.rc != 0: + logger.error("stdout: %s", self.stdout_s) + logger.error("stderr: %s", self.stderr_s) + raise subprocess.CalledProcessError( + returncode=self.rc, cmd=self.cmd, stderr=self.stderr, output=self.stdout + ) + return self + + def expect(self, rc: int) -> "CommandResult": + """ + assumes a return code of `rc` + :param rc: The expected return code + :raises CalledProcessError: if the return code wasn't `rc` and logs the processes stdout and stderr + """ + if self.rc != rc: + logger.error("stdout: %s", self.stdout_s) + logger.error("stderr: %s", self.stderr_s) + raise subprocess.CalledProcessError( + returncode=self.rc, cmd=self.cmd, stderr=self.stderr, output=self.stdout + ) + return self + + @property + def stdout_s(self) -> str: + """Command stdout as str""" + return self.stdout.decode("utf-8", errors="replace") + + @property + def stderr_s(self) -> str: + """Command stderr as str""" + return self.stderr.decode("utf-8", errors="replace") + + @property + def stdout_plain(self) -> str: + """Command stderr as str with terminal escape sequences eaten and whitespace stripped""" + return eat_terminal_codes(self.stdout).decode("utf-8", errors="replace").strip() + + @property + def stderr_plain(self) -> str: + """Command stderr as str with terminal escape sequences eaten and whitespace stripped""" + return eat_terminal_codes(self.stderr).decode("utf-8", errors="replace").strip() + + def json(self) -> Any: + """ + Assumes an ok() result and returns the Commands stdout parsed as json + :return: A parsed json object + """ + self.ok() + return json.loads(self.stdout) + + +@dataclasses.dataclass +class Command: + """ + Provides a way of configuring a shell command and then running it + calls Popen internally + """ + + argv: list[str] + """ + Arguments of the Process; argv[0] is the name of the binary + """ + env: dict[str, str] = dataclasses.field(default_factory=dict) + """ + environment variables; Note that `$PATH` is not added by default. + Use `.with_env(**os.environ.copy())` to add `$PATH` + """ + stdin: bytes | None = None + """ + Things to pipe into stdin of the process + """ + cwd: Path | None = None + """ + current-working-directory fo the process + """ + + def with_env(self, **kwargs) -> "Command": + """ + adds or updates environment and path variables + :param kwargs: new or updated variables + :return: self, command is chainable + """ + self.env.update(kwargs) + return self + + def with_stdin(self, stdin: bytes) -> "Command": + """ + sets the input provided to stdin of the commands + :param stdin: data to pipe into stdin + :return: self, command is chainable + """ + self.stdin = stdin + return self + + def run(self) -> CommandResult: + """ + Runs the configured command + :return: Information about the Result of the execution + """ + proc = subprocess.Popen( + self.argv, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + stdin=subprocess.PIPE if self.stdin else subprocess.DEVNULL, + cwd=self.cwd, + env=self.env, + ) + (stdout, stderr) = proc.communicate(input=self.stdin) + rc = proc.returncode + return CommandResult(cmd=self.argv, rc=rc, stdout=stdout, stderr=stderr) diff --git a/tests/functional2/testlib/fixtures/nix.py b/tests/functional2/testlib/fixtures/nix.py index 63ce1d176..558b62036 100644 --- a/tests/functional2/testlib/fixtures/nix.py +++ b/tests/functional2/testlib/fixtures/nix.py @@ -1,8 +1,5 @@ import dataclasses -import json -import logging import os -import subprocess from functools import partialmethod from pathlib import Path from typing import Any, AnyStr @@ -10,63 +7,7 @@ from collections.abc import Callable, Generator import pytest -from functional2.testlib.terminal_code_eater import eat_terminal_codes - - -logger = logging.getLogger(__name__) - - -@dataclasses.dataclass -class CommandResult: - cmd: list[str] - rc: int - """Return code""" - stderr: bytes - """Outputted stderr""" - stdout: bytes - """Outputted stdout""" - - def ok(self) -> "CommandResult": - if self.rc != 0: - logger.debug("stdout: %s", self.stderr_s) - logger.debug("stderr: %s", self.stderr_s) - raise subprocess.CalledProcessError( - returncode=self.rc, cmd=self.cmd, stderr=self.stderr, output=self.stdout - ) - return self - - def expect(self, rc: int) -> "CommandResult": - if self.rc != rc: - logger.debug("stdout: %s", self.stderr_s) - logger.debug("stderr: %s", self.stderr_s) - raise subprocess.CalledProcessError( - returncode=self.rc, cmd=self.cmd, stderr=self.stderr, output=self.stdout - ) - return self - - @property - def stdout_s(self) -> str: - """Command stdout as str""" - return self.stdout.decode("utf-8", errors="replace") - - @property - def stderr_s(self) -> str: - """Command stderr as str""" - return self.stderr.decode("utf-8", errors="replace") - - @property - def stdout_plain(self) -> str: - """Command stderr as str with terminal escape sequences eaten and whitespace stripped""" - return eat_terminal_codes(self.stdout).decode("utf-8", errors="replace").strip() - - @property - def stderr_plain(self) -> str: - """Command stderr as str with terminal escape sequences eaten and whitespace stripped""" - return eat_terminal_codes(self.stderr).decode("utf-8", errors="replace").strip() - - def json(self) -> Any: - self.ok() - return json.loads(self.stdout) +from functional2.testlib.commands import CommandResult, Command @dataclasses.dataclass @@ -123,35 +64,6 @@ class NixSettings: return ret -@dataclasses.dataclass -class Command: - argv: list[str] - env: dict[str, str] = dataclasses.field(default_factory=dict) - stdin: bytes | None = None - cwd: Path | None = None - - def with_env(self, **kwargs) -> "Command": - self.env.update(kwargs) - return self - - def with_stdin(self, stdin: bytes) -> "Command": - self.stdin = stdin - return self - - def run(self) -> CommandResult: - proc = subprocess.Popen( - self.argv, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - stdin=subprocess.PIPE if self.stdin else subprocess.DEVNULL, - cwd=self.cwd, - env=self.env, - ) - (stdout, stderr) = proc.communicate(input=self.stdin) - rc = proc.returncode - return CommandResult(cmd=self.argv, rc=rc, stdout=stdout, stderr=stderr) - - @dataclasses.dataclass class NixCommand(Command): settings: NixSettings = dataclasses.field(default_factory=NixSettings)