From a90f719a0e6476dcd5bb7318752123e39c5aa015 Mon Sep 17 00:00:00 2001 From: Robin Krahl Date: Tue, 30 Apr 2024 13:30:31 +0200 Subject: [PATCH] tests: Move container setup into fixtures --- tests/conftest.py | 26 +++-- tests/test_nethsm_other.py | 18 +--- tests/test_nethsm_system.py | 28 ++--- tests/utilities.py | 203 ++++++++++++++++++++++-------------- 4 files changed, 163 insertions(+), 112 deletions(-) diff --git a/tests/conftest.py b/tests/conftest.py index 06950d6..cd85b03 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,6 +1,6 @@ from dataclasses import dataclass from os import environ -from typing import Iterator, Literal +from typing import TYPE_CHECKING, Iterator, Literal import pytest @@ -14,6 +14,9 @@ UnattendedBootStatus, ) +if TYPE_CHECKING: + from utilities import Container + @dataclass class UserData: @@ -127,19 +130,28 @@ class Constants: @pytest.fixture(scope="module") -def nethsm() -> Iterator[NetHSM]: +def container() -> Iterator["Container"]: + from utilities import KeyfenderManager + + container = KeyfenderManager.get().spawn() + try: + container.start() + container.wait() + yield container + finally: + container.kill() + + +@pytest.fixture(scope="module") +def nethsm(container: "Container") -> Iterator[NetHSM]: """Start Docker container with Nethsm image and connect to Nethsm This Pytest Fixture will run before the tests to provide the tests with a nethsm instance via Docker container, also the first provision of the NetHSM will be done in here""" - from utilities import connect, provision, start_nethsm - - container = start_nethsm() + from utilities import connect, provision with connect(Constants.ADMIN_USER) as nethsm: provision(nethsm) yield nethsm - - container.kill() diff --git a/tests/test_nethsm_other.py b/tests/test_nethsm_other.py index a38af42..6c9eecb 100644 --- a/tests/test_nethsm_other.py +++ b/tests/test_nethsm_other.py @@ -3,7 +3,7 @@ import docker # type: ignore import pytest from conftest import Constants as C -from utilities import add_user, connect, lock, provision, start_nethsm, unlock +from utilities import Container, add_user, connect, lock, provision, unlock import nethsm as nethsm_sdk from nethsm import NetHSM @@ -18,38 +18,26 @@ @pytest.fixture(scope="module") -def nethsm_no_provision() -> Iterator[NetHSM]: +def nethsm_no_provision(container: Container) -> Iterator[NetHSM]: """Start Docker container with Nethsm image and connect to Nethsm This Pytest Fixture will run before the tests to provide the tests with a nethsm instance via Docker container""" - container = start_nethsm() with connect(C.ADMIN_USER) as nethsm: yield nethsm - try: - container.kill() - except docker.errors.APIError: - pass - @pytest.fixture(scope="module") -def nethsm_no_provision_no_auth() -> Iterator[NetHSM]: +def nethsm_no_provision_no_auth(container: Container) -> Iterator[NetHSM]: """Start Docker container with Nethsm image and connect to Nethsm This Pytest Fixture will run before the tests to provide the tests with a nethsm instance via Docker container""" - container = start_nethsm() with nethsm_sdk.connect(C.HOST, verify_tls=C.VERIFY_TLS) as nethsm: yield nethsm - try: - container.kill() - except docker.errors.APIError: - pass - """######################### Start of Tests #########################""" diff --git a/tests/test_nethsm_system.py b/tests/test_nethsm_system.py index f92996b..95032df 100644 --- a/tests/test_nethsm_system.py +++ b/tests/test_nethsm_system.py @@ -5,12 +5,12 @@ from conftest import Constants as C from test_nethsm_keys import generate_key from utilities import ( + Container, add_user, connect, encrypt_rsa, provision, set_backup_passphrase, - start_nethsm, update, ) @@ -73,13 +73,13 @@ def test_passphrase_add_user_retrieve_backup(nethsm: NetHSM) -> None: assert False -def test_factory_reset(nethsm: NetHSM) -> None: +def test_factory_reset(container: Container, nethsm: NetHSM) -> None: """Perform a factory reset for a NetHSM instance. This command requires authentication as a user with the Administrator role.""" nethsm.factory_reset() - start_nethsm() + container.restart() # make sure that we really cleared the data assert nethsm.get_state().value == "Unprovisioned" @@ -87,7 +87,7 @@ def test_factory_reset(nethsm: NetHSM) -> None: assert nethsm.list_keys() == [] nethsm.factory_reset() - start_nethsm() + container.restart() def test_state_restore(nethsm: NetHSM) -> None: @@ -135,24 +135,24 @@ def test_state_restore(nethsm: NetHSM) -> None: assert decrypt.decode().decode() == C.DATA -def test_state_provision_update(nethsm: NetHSM) -> None: +def test_state_provision_update(container: Container, nethsm: NetHSM) -> None: """Load an update to a NetHSM instance. This command requires authentication as a user with the Administrator role.""" - start_nethsm() + container.restart() provision(nethsm) update(nethsm) -def test_state_provision_update_cancel_update(nethsm: NetHSM) -> None: +def test_state_provision_update_cancel_update(container: Container, nethsm: NetHSM) -> None: """Cancel a queued update on a NetHSM instance. This command requires authentication as a user with the Administrator role.""" - start_nethsm() + container.restart() provision(nethsm) @@ -160,12 +160,12 @@ def test_state_provision_update_cancel_update(nethsm: NetHSM) -> None: nethsm.cancel_update() -def test_update_commit_update(nethsm: NetHSM) -> None: +def test_update_commit_update(container: Container, nethsm: NetHSM) -> None: """Commit a queued update on a NetHSM instance. This command requires authentication as a user with the Administrator role.""" - start_nethsm() + container.restart() provision(nethsm) @@ -173,24 +173,24 @@ def test_update_commit_update(nethsm: NetHSM) -> None: nethsm.commit_update() -def test_provision_reboot(nethsm: NetHSM) -> None: +def test_provision_reboot(container: Container, nethsm: NetHSM) -> None: """Reboot a NetHSM instance. This command requires authentication as a user with the Administrator role.""" - start_nethsm() + container.restart() provision(nethsm) nethsm.reboot() -def test_provision_shutdown(nethsm: NetHSM) -> None: +def test_provision_shutdown(container: Container, nethsm: NetHSM) -> None: """Shutdown a NetHSM instance. This command requires authentication as a user with the Administrator role.""" - start_nethsm() + container.restart() provision(nethsm) diff --git a/tests/utilities.py b/tests/utilities.py index 5d97ab8..1bf9daf 100644 --- a/tests/utilities.py +++ b/tests/utilities.py @@ -2,8 +2,9 @@ import datetime import os import subprocess +from abc import ABC, abstractmethod from time import sleep -from typing import Iterator +from typing import Iterator, Optional import docker # type: ignore import podman # type: ignore @@ -25,12 +26,124 @@ from nethsm import Authentication, Base64, NetHSM, RsaPrivateKey -class KeyfenderManager: +class Container(ABC): + def restart(self) -> None: + self.kill() + self.start() + self.wait() + + def wait(self) -> None: + http = urllib3.PoolManager(cert_reqs="CERT_NONE") + print("Waiting for container to be ready") + while True: + try: + response = http.request("GET", f"https://{C.HOST}/api/v1/health/alive") + print(f"Response: {response.status}") + if response.status == 200: + break + except Exception as e: + print(e) + pass + sleep(0.5) + + @abstractmethod + def start(self) -> None: + ... + + @abstractmethod + def kill(self) -> None: + ... + + +class DockerContainer(Container): + def __init__(self, client: docker.client.DockerClient, image: docker.models.images.Image) -> None: + self.client = client + self.image = image + self.container = None + + def start(self) -> None: + self.container = self.client.containers.run( + self.image, + "", + ports={"8443": 8443}, + remove=True, + detach=True, + ) + + def kill(self) -> None: + if self.container: + try: + self.container.kill() + self.container.wait() + except docker.errors.APIError: + pass + + +class PodmanContainer(Container): + def __init__(self, client: podman.client.PodmanClient, image: podman.domain.images.Image) -> None: + self.client = client + self.image = image + self.container = None + + def start(self) -> None: + self.container = self.client.containers.run( + self.image, + "", + ports={"8443": 8443}, + remove=True, + detach=True, + ) + + def kill(self) -> None: + if self.container: + try: + self.container.kill() + self.container.wait() + except podman.errors.APIError: + pass + + +class CIContainer(Container): def __init__(self) -> None: - pass + self.process: Optional[subprocess.Popen[bytes]] = None + + def start(self) -> None: + os.system("pkill keyfender.unix") + os.system("pkill etcd") + + # Wait for everything to shut down, creates problems otherwise on the gitlab ci + sleep(1) + + os.system("rm -rf /data") + + self.process = subprocess.Popen( + [ + "/bin/sh", + "-c", + "/start.sh", + ] + ) def kill(self) -> None: - pass + if self.process: + self.process.kill() + + +class KeyfenderManager(ABC): + @abstractmethod + def spawn(self) -> Container: + ... + + @staticmethod + def get() -> "KeyfenderManager": + if C.TEST_MODE == "docker": + return KeyfenderDockerManager() + elif C.TEST_MODE == "podman": + return KeyfenderPodmanManager() + elif C.TEST_MODE == "ci": + return KeyfenderCIManager() + else: + raise Exception("Invalid Test Mode") class KeyfenderDockerManager(KeyfenderManager): @@ -56,20 +169,11 @@ def __init__(self) -> None: repository, tag = C.IMAGE.split(":") image = client.images.pull(repository, tag=tag) - container = client.containers.run( - image, - "", - ports={"8443": 8443}, - remove=True, - detach=True, - ) - self.container = container + self.client = client + self.image = image - def kill(self) -> None: - try: - self.container.kill() - except docker.errors.APIError: - pass + def spawn(self) -> Container: + return DockerContainer(self.client, self.image) class KeyfenderPodmanManager(KeyfenderManager): @@ -95,69 +199,16 @@ def __init__(self) -> None: repository, tag = C.IMAGE.split(":") image = client.images.pull(repository, tag=tag) - container = client.containers.run( - image, - "", - ports={"8443": 8443}, - remove=True, - detach=True, - ) - self.container = container + self.client = client + self.image = image - def kill(self) -> None: - try: - self.container.kill() - except podman.errors.APIError: - pass + def spawn(self) -> Container: + return PodmanContainer(self.client, self.image) class KeyfenderCIManager(KeyfenderManager): - def __init__(self) -> None: - os.system("pkill keyfender.unix") - os.system("pkill etcd") - - # Wait for everything to shut down, creates problems otherwise on the gitlab ci - sleep(1) - - os.system("rm -rf /data") - - self.process = subprocess.Popen( - [ - "/bin/sh", - "-c", - "/start.sh", - ] - ) - - def kill(self) -> None: - self.process.kill() - - -def start_nethsm() -> KeyfenderManager: - context: KeyfenderManager - if C.TEST_MODE == "docker": - context = KeyfenderDockerManager() - elif C.TEST_MODE == "podman": - context = KeyfenderPodmanManager() - elif C.TEST_MODE == "ci": - context = KeyfenderCIManager() - else: - raise Exception("Invalid Test Mode") - - http = urllib3.PoolManager(cert_reqs="CERT_NONE") - print("Waiting for container to be ready") - while True: - try: - response = http.request("GET", f"https://{C.HOST}/api/v1/health/alive") - print(f"Response: {response.status}") - if response.status == 200: - break - except Exception as e: - print(e) - pass - sleep(0.5) - - return context + def spawn(self) -> Container: + return CIContainer() @contextlib.contextmanager