1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
|
# SPDX-License-Identifier: GPL-2.0
import os
import re
import subprocess
import time
from pathlib import Path
def base_path() -> Path:
"""Returns the base path for the project"""
return Path(__file__).parent.parent.parent
def test_path() -> str:
"""Returns the path to the tests directory"""
return os.path.join(base_path(), "tests/integration")
def podman_compose_path() -> str:
"""Returns the path to the podman compose script"""
return os.path.join(base_path(), "podman_compose.py")
def is_systemd_available() -> bool:
try:
with open("/proc/1/comm", "r", encoding="utf-8") as fh:
return fh.read().strip() == "systemd"
except FileNotFoundError:
return False
class RunSubprocessMixin:
def is_debug_enabled(self) -> bool:
return "TESTS_DEBUG" in os.environ
def run_subprocess(self, args: list[str], env: dict[str, str] = {}) -> tuple[bytes, bytes, int]:
begin = time.time()
if self.is_debug_enabled():
print("TEST_CALL", args)
proc = subprocess.Popen(
args,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
env=os.environ | env,
)
out, err = proc.communicate()
if self.is_debug_enabled():
print("TEST_CALL completed", time.time() - begin)
print("STDOUT:", out.decode('utf-8'))
print("STDERR:", err.decode('utf-8'))
return out, err, proc.returncode
def run_subprocess_assert_returncode(
self, args: list[str], expected_returncode: int = 0, env: dict[str, str] = {}
) -> tuple[bytes, bytes]:
out, err, returncode = self.run_subprocess(args, env=env)
decoded_out = out.decode('utf-8')
decoded_err = err.decode('utf-8')
self.assertEqual( # type: ignore[attr-defined]
returncode,
expected_returncode,
f"Invalid return code of process {returncode} != {expected_returncode}\n"
f"stdout: {decoded_out}\nstderr: {decoded_err}\n",
)
return out, err
class PodmanAwareRunSubprocessMixin(RunSubprocessMixin):
def retrieve_podman_version(self) -> tuple[int, int, int]:
out, _ = self.run_subprocess_assert_returncode(["podman", "--version"])
matcher = re.match(r"\D*(\d+)\.(\d+)\.(\d+)", out.decode('utf-8'))
if matcher:
major = int(matcher.group(1))
minor = int(matcher.group(2))
patch = int(matcher.group(3))
return (major, minor, patch)
raise RuntimeError("Unable to retrieve podman version")
|
