From 9940f959357cb25c63e8dab6581b488c3a76dcba Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C3=B6rg=20Thalheim?= Date: Sat, 11 Jun 2022 12:23:05 +0200 Subject: [PATCH] split test into smaller files --- tests/random_service.py | 25 ++++++++ tests/test_blocking_secret.py | 41 ++++++++++++ tests/test_service.py | 108 -------------------------------- tests/test_socket_activation.py | 54 ++++++++++++++++ tests/test_vault.py | 39 ++++++++++++ 5 files changed, 159 insertions(+), 108 deletions(-) create mode 100644 tests/random_service.py create mode 100644 tests/test_blocking_secret.py delete mode 100644 tests/test_service.py create mode 100644 tests/test_socket_activation.py create mode 100644 tests/test_vault.py diff --git a/tests/random_service.py b/tests/random_service.py new file mode 100644 index 0000000..1a322a5 --- /dev/null +++ b/tests/random_service.py @@ -0,0 +1,25 @@ +#!/usr/bin/env python3 + +import random +import string +from dataclasses import dataclass +from pathlib import Path + + +def rand_word(n: int) -> str: + return "".join(random.choices(string.ascii_uppercase + string.digits, k=n)) + + +@dataclass +class Service: + name: str + secret_name: str + secret_path: Path + + +def random_service(secrets_dir: Path) -> Service: + service = f"test-service-{rand_word(8)}.service" + secret_name = "foo" + secret = f"{service}-{secret_name}" + secret_path = secrets_dir / secret + return Service(service, secret_name, secret_path) diff --git a/tests/test_blocking_secret.py b/tests/test_blocking_secret.py new file mode 100644 index 0000000..84e21ca --- /dev/null +++ b/tests/test_blocking_secret.py @@ -0,0 +1,41 @@ +import subprocess +from pathlib import Path +import time + +from command import Command, run +from random_service import random_service + + +def test_blocking_secret( + systemd_vaultd: Path, command: Command, tempdir: Path +) -> None: + secrets_dir = tempdir / "secrets" + sock = tempdir / "sock" + command.run([str(systemd_vaultd), "-secrets", str(secrets_dir), "-sock", str(sock)]) + + while not sock.exists(): + time.sleep(0.1) + + service = random_service(secrets_dir) + + proc = command.run( + [ + "systemd-run", + "-u", + service.name, + "--collect", + "--user", + "-p", + f"LoadCredential={service.secret_name}:{sock}", + "--wait", + "--pipe", + "cat", + "${CREDENTIALS_DIRECTORY}/" + service.secret_name, + ], + stdout=subprocess.PIPE, + ) + time.sleep(0.1) + assert proc.poll() is None, "service should block for secret" + service.secret_path.write_text("foo") + assert proc.stdout is not None and proc.stdout.read() == "foo" + assert proc.wait() == 0 diff --git a/tests/test_service.py b/tests/test_service.py deleted file mode 100644 index fc36a73..0000000 --- a/tests/test_service.py +++ /dev/null @@ -1,108 +0,0 @@ -import subprocess - -from dataclasses import dataclass -from command import Command, run -from pathlib import Path -import time - -import string -import random - - -def rand_word(n: int) -> str: - return "".join(random.choices(string.ascii_uppercase + string.digits, k=n)) - - -@dataclass -class Service: - name: str - secret_name: str - secret_path: Path - - -def random_service(secrets_dir: Path) -> Service: - service = f"test-service-{rand_word(8)}.service" - secret_name = "foo" - secret = f"{service}-{secret_name}" - secret_path = secrets_dir / secret - return Service(service, secret_name, secret_path) - - -def test_socket_activation( - systemd_vaultd: Path, command: Command, tempdir: Path -) -> None: - secrets_dir = tempdir / "secrets" - secrets_dir.mkdir() - sock = tempdir / "sock" - - command.run( - [ - "systemd-socket-activate", - "--listen", - str(sock), - str(systemd_vaultd), - "-secrets", - str(secrets_dir), - "-sock", - str(sock), - ] - ) - - while not sock.exists(): - time.sleep(0.1) - - service = random_service(secrets_dir) - service.secret_path.write_text("foo") - - # should not block - out = run( - [ - "systemd-run", - "-u", - service.name, - "--collect", - "--user", - "-p", - f"LoadCredential={service.secret_name}:{sock}", - "--wait", - "--pipe", - "cat", - "${CREDENTIALS_DIRECTORY}/" + service.secret_name, - ], - stdout=subprocess.PIPE, - ) - assert out.stdout == "foo" - assert out.returncode == 0 - - -def test_blocking_secret(systemd_vaultd: Path, command: Command, tempdir: Path) -> None: - secrets_dir = tempdir / "secrets" - sock = tempdir / "sock" - command.run([str(systemd_vaultd), "-secrets", str(secrets_dir), "-sock", str(sock)]) - - while not sock.exists(): - time.sleep(0.1) - - service = random_service(secrets_dir) - - proc = command.run( - [ - "systemd-run", - "-u", - service.name, - "--collect", - "--user", - "-p", - f"LoadCredential={service.secret_name}:{sock}", - "--wait", - "--pipe", - "cat", - "${CREDENTIALS_DIRECTORY}/" + service.secret_name, - ], - stdout=subprocess.PIPE, - ) - time.sleep(0.1) - assert proc.poll() is None, "service should block for secret" - service.secret_path.write_text("foo") - assert proc.stdout is not None and proc.stdout.read() == "foo" - assert proc.wait() == 0 diff --git a/tests/test_socket_activation.py b/tests/test_socket_activation.py new file mode 100644 index 0000000..62b3d35 --- /dev/null +++ b/tests/test_socket_activation.py @@ -0,0 +1,54 @@ +#!/usr/bin/env python3 + +import time +import subprocess +from pathlib import Path + +from command import Command, run +from random_service import random_service + +def test_socket_activation( + systemd_vaultd: Path, command: Command, tempdir: Path, +) -> None: + secrets_dir = tempdir / "secrets" + secrets_dir.mkdir() + sock = tempdir / "sock" + + command.run( + [ + "systemd-socket-activate", + "--listen", + str(sock), + str(systemd_vaultd), + "-secrets", + str(secrets_dir), + "-sock", + str(sock), + ] + ) + + while not sock.exists(): + time.sleep(0.1) + + service = random_service(secrets_dir) + service.secret_path.write_text("foo") + + # should not block + out = run( + [ + "systemd-run", + "-u", + service.name, + "--collect", + "--user", + "-p", + f"LoadCredential={service.secret_name}:{sock}", + "--wait", + "--pipe", + "cat", + "${CREDENTIALS_DIRECTORY}/" + service.secret_name, + ], + stdout=subprocess.PIPE, + ) + assert out.stdout == "foo" + assert out.returncode == 0 diff --git a/tests/test_vault.py b/tests/test_vault.py new file mode 100644 index 0000000..7837023 --- /dev/null +++ b/tests/test_vault.py @@ -0,0 +1,39 @@ +#!/usr/bin/env python3 + +from command import Command, run +from pathlib import Path + +def test_blocking_secret( + systemd_vaultd: Path, command: Command, tempdir: Path +) -> None: + secrets_dir = tempdir / "secrets" + command.run(["vault", "server", "-dev"]) + #sock = tempdir / "sock" + #command.run([str(systemd_vaultd), "-secrets", str(secrets_dir), "-sock", str(sock)]) + + #while not sock.exists(): + # time.sleep(0.1) + + #service = random_service(secrets_dir) + + #proc = command.run( + # [ + # "systemd-run", + # "-u", + # service.name, + # "--collect", + # "--user", + # "-p", + # f"LoadCredential={service.secret_name}:{sock}", + # "--wait", + # "--pipe", + # "cat", + # "${CREDENTIALS_DIRECTORY}/" + service.secret_name, + # ], + # stdout=subprocess.PIPE, + #) + #time.sleep(0.1) + #assert proc.poll() is None, "service should block for secret" + #service.secret_path.write_text("foo") + #assert proc.stdout is not None and proc.stdout.read() == "foo" + #assert proc.wait() == 0