diff options
Diffstat (limited to 'test/apiv2/python/rest_api/fixtures')
| -rw-r--r-- | test/apiv2/python/rest_api/fixtures/__init__.py | 3 | ||||
| -rw-r--r-- | test/apiv2/python/rest_api/fixtures/api_testcase.py | 103 | ||||
| -rw-r--r-- | test/apiv2/python/rest_api/fixtures/podman.py | 136 |
3 files changed, 242 insertions, 0 deletions
diff --git a/test/apiv2/python/rest_api/fixtures/__init__.py b/test/apiv2/python/rest_api/fixtures/__init__.py new file mode 100644 index 000000000..5d763e454 --- /dev/null +++ b/test/apiv2/python/rest_api/fixtures/__init__.py @@ -0,0 +1,3 @@ +from .api_testcase import APITestCase + +__all__ = ["APITestCase"] diff --git a/test/apiv2/python/rest_api/fixtures/api_testcase.py b/test/apiv2/python/rest_api/fixtures/api_testcase.py new file mode 100644 index 000000000..8b771774b --- /dev/null +++ b/test/apiv2/python/rest_api/fixtures/api_testcase.py @@ -0,0 +1,103 @@ +import json +import subprocess +import unittest + +import requests +import sys +import time + +from .podman import Podman + + +class APITestCase(unittest.TestCase): + PODMAN_URL = "http://localhost:8080" + podman = None # initialized podman configuration for tests + service = None # podman service instance + + @classmethod + def setUpClass(cls): + super().setUpClass() + + APITestCase.podman = Podman() + APITestCase.service = APITestCase.podman.open( + "system", "service", "tcp:localhost:8080", "--time=0" + ) + # give the service some time to be ready... + time.sleep(2) + + returncode = APITestCase.service.poll() + if returncode is not None: + raise subprocess.CalledProcessError(returncode, "podman system service") + + r = requests.post( + APITestCase.uri("/images/pull?reference=quay.io%2Flibpod%2Falpine%3Alatest") + ) + if r.status_code != 200: + raise subprocess.CalledProcessError( + r.status_code, f"podman images pull quay.io/libpod/alpine:latest {r.text}" + ) + + @classmethod + def tearDownClass(cls): + APITestCase.service.terminate() + stdout, stderr = APITestCase.service.communicate(timeout=0.5) + if stdout: + sys.stdout.write("\nService Stdout:\n" + stdout.decode("utf-8")) + if stderr: + sys.stderr.write("\nService Stderr:\n" + stderr.decode("utf-8")) + return super().tearDownClass() + + def setUp(self): + super().setUp() + APITestCase.podman.run("run", "alpine", "/bin/ls", check=True) + + def tearDown(self) -> None: + APITestCase.podman.run("pod", "rm", "--all", "--force", check=True) + APITestCase.podman.run("rm", "--all", "--force", check=True) + super().tearDown() + + @property + def podman_url(self): + return "http://localhost:8080" + + @staticmethod + def uri(path): + return APITestCase.PODMAN_URL + "/v2.0.0/libpod" + path + + def resolve_container(self, path): + """Find 'first' container and return 'Id' formatted into given URI path.""" + + try: + r = requests.get(self.uri("/containers/json?all=true")) + containers = r.json() + except Exception as e: + msg = f"Bad container response: {e}" + if r is not None: + msg += ": " + r.text + raise self.failureException(msg) + return path.format(containers[0]["Id"]) + + def assertContainerExists(self, member, msg=None): # pylint: disable=invalid-name + r = requests.get(self.uri(f"/containers/{member}/exists")) + if r.status_code == 404: + if msg is None: + msg = f"Container '{member}' does not exist." + self.failureException(msg) + + def assertContainerNotExists(self, member, msg=None): # pylint: disable=invalid-name + r = requests.get(self.uri(f"/containers/{member}/exists")) + if r.status_code == 204: + if msg is None: + msg = f"Container '{member}' exists." + self.failureException(msg) + + def assertId(self, content): # pylint: disable=invalid-name + objects = json.loads(content) + try: + if isinstance(objects, dict): + _ = objects["Id"] + else: + for item in objects: + _ = item["Id"] + except KeyError: + self.failureException("Failed in find 'Id' in return value.") diff --git a/test/apiv2/python/rest_api/fixtures/podman.py b/test/apiv2/python/rest_api/fixtures/podman.py new file mode 100644 index 000000000..bae04f87d --- /dev/null +++ b/test/apiv2/python/rest_api/fixtures/podman.py @@ -0,0 +1,136 @@ +import configparser +import json +import os +import shutil +import subprocess +import sys +import tempfile + + +class Podman: + """ + Instances hold the configuration and setup for running podman commands + """ + + def __init__(self): + """Initialize a Podman instance with global options""" + binary = os.getenv("PODMAN", "bin/podman") + self.cmd = [binary, "--storage-driver=vfs"] + + cgroupfs = os.getenv("CGROUP_MANAGER", "systemd") + self.cmd.append(f"--cgroup-manager={cgroupfs}") + + if os.getenv("DEBUG"): + self.cmd.append("--log-level=debug") + self.cmd.append("--syslog=true") + + self.anchor_directory = tempfile.mkdtemp(prefix="podman_restapi_") + self.cmd.append("--root=" + os.path.join(self.anchor_directory, "crio")) + self.cmd.append("--runroot=" + os.path.join(self.anchor_directory, "crio-run")) + + os.environ["CONTAINERS_REGISTRIES_CONF"] = os.path.join( + self.anchor_directory, "registry.conf" + ) + p = configparser.ConfigParser() + p.read_dict( + { + "registries.search": {"registries": "['quay.io']"}, + "registries.insecure": {"registries": "[]"}, + "registries.block": {"registries": "[]"}, + } + ) + with open(os.environ["CONTAINERS_REGISTRIES_CONF"], "w") as w: + p.write(w) + + os.environ["CNI_CONFIG_PATH"] = os.path.join(self.anchor_directory, "cni", "net.d") + os.makedirs(os.environ["CNI_CONFIG_PATH"], exist_ok=True) + self.cmd.append("--cni-config-dir=" + os.environ["CNI_CONFIG_PATH"]) + cni_cfg = os.path.join(os.environ["CNI_CONFIG_PATH"], "87-podman-bridge.conflist") + # json decoded and encoded to ensure legal json + buf = json.loads( + """ + { + "cniVersion": "0.3.0", + "name": "podman", + "plugins": [{ + "type": "bridge", + "bridge": "cni0", + "isGateway": true, + "ipMasq": true, + "ipam": { + "type": "host-local", + "subnet": "10.88.0.0/16", + "routes": [{ + "dst": "0.0.0.0/0" + }] + } + }, + { + "type": "portmap", + "capabilities": { + "portMappings": true + } + } + ] + } + """ + ) + with open(cni_cfg, "w") as w: + json.dump(buf, w) + + def open(self, command, *args, **kwargs): + """Podman initialized instance to run a given command + + :param self: Podman instance + :param command: podman sub-command to run + :param args: arguments and options for command + :param kwargs: See subprocess.Popen() for shell keyword + :return: subprocess.Popen() instance configured to run podman instance + """ + cmd = self.cmd.copy() + cmd.append(command) + cmd.extend(args) + + shell = kwargs.get("shell", False) + + return subprocess.Popen( + cmd, + shell=shell, + stdin=subprocess.DEVNULL, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + + def run(self, command, *args, **kwargs): + """Run given podman command + + :param self: Podman instance + :param command: podman sub-command to run + :param args: arguments and options for command + :param kwargs: See subprocess.Popen() for shell and check keywords + :return: subprocess.Popen() instance configured to run podman instance + """ + cmd = self.cmd.copy() + cmd.append(command) + cmd.extend(args) + + check = kwargs.get("check", False) + shell = kwargs.get("shell", False) + + try: + return subprocess.run( + cmd, + shell=shell, + check=check, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + ) + except subprocess.CalledProcessError as e: + if e.stdout: + sys.stdout.write("\nRun Stdout:\n" + e.stdout.decode("utf-8")) + if e.stderr: + sys.stderr.write("\nRun Stderr:\n" + e.stderr.decode("utf-8")) + raise + + def tear_down(self): + shutil.rmtree(self.anchor_directory, ignore_errors=True) |