Browse Source

dedicated DNS suffix for each cookbook and test

main
Nicolas Massé 2 weeks ago
parent
commit
a46f36553d
  1. 13
      conftest.py
  2. 22
      cookbooks/traefik/tests/test_01_install.py
  3. 15
      tests/test_quadlet.py

13
conftest.py

@ -53,10 +53,13 @@ def pebble_server_ip(libvirt_network_if: str) -> str:
"""IP Address of the Pebble ACME server.""" """IP Address of the Pebble ACME server."""
return _get_libvirt_bridge_ip(libvirt_network_if) return _get_libvirt_bridge_ip(libvirt_network_if)
@pytest.fixture(scope="session") @pytest.fixture(scope="module")
def top_level_domain() -> str: def top_level_domain(request, keep) -> str:
"""Top-level domain for the test environment.""" """Top-level domain for the test environment."""
return "pytest.example.test" module_name = request.module.__name__.split(".")[-1].replace("test_", "").replace("_", "-")
cookbook_dir = Path(request.path).parent.parent
instance = "dev" if keep else f"pid-{os.getpid()}"
return f"{instance}.{module_name}.{cookbook_dir.name}.pytest.example.test"
@pytest.fixture(scope="session") @pytest.fixture(scope="session")
def dns_server_ip(libvirt_network_if: str) -> str: def dns_server_ip(libvirt_network_if: str) -> str:
@ -217,9 +220,9 @@ def pebble_acme_server(tmp_path_factory: pytest.TempPathFactory, pebble_server_i
yield data # <-- tests run here with access to the Pebble ACME server yield data # <-- tests run here with access to the Pebble ACME server
@pytest.fixture(scope="session") @pytest.fixture(scope="module")
def dns_server(libvirt_network: str, top_level_domain: str, keep: bool) -> DNSServer: def dns_server(libvirt_network: str, top_level_domain: str, keep: bool) -> DNSServer:
"""Session-scoped DNS server manager for the libvirt network.""" """Module-scoped DNS server manager for the libvirt network."""
srv = DNSServer(network=libvirt_network, persistent=keep) srv = DNSServer(network=libvirt_network, persistent=keep)
srv.set_domain(top_level_domain) srv.set_domain(top_level_domain)

22
cookbooks/traefik/tests/test_01_install.py

@ -57,12 +57,14 @@ def fcos_extra_files(pebble_acme_server, top_level_domain) -> dict:
10000, 10000,
0o644, 0o644,
), ),
# The Pebble CA certificate is needed for Traefik to trust the Pebble ACME server.
"/etc/quadlets/traefik/pebble.pem": ( "/etc/quadlets/traefik/pebble.pem": (
pebble_acme_server['ca_cert'], pebble_acme_server['ca_cert'],
10001, 10001,
10000, 10000,
0o644, 0o644,
), ),
# The main Traefik configuration file.
"/etc/quadlets/traefik/traefik.yaml": ( "/etc/quadlets/traefik/traefik.yaml": (
textwrap.dedent(f"""\ textwrap.dedent(f"""\
api: api:
@ -89,7 +91,7 @@ def fcos_extra_files(pebble_acme_server, top_level_domain) -> dict:
certificatesResolvers: certificatesResolvers:
le: le:
acme: acme:
email: "traefik@pytest.example.test" email: "traefik@{top_level_domain}"
caServer: "{pebble_acme_server['directory_url']}" caServer: "{pebble_acme_server['directory_url']}"
caCertificates: "/etc/traefik/pebble.pem" caCertificates: "/etc/traefik/pebble.pem"
keyType: "EC384" keyType: "EC384"
@ -147,17 +149,6 @@ class TestTraefikQuadlet(test_quadlet.TestQuadlet):
expected_main_service = "traefik.target" expected_main_service = "traefik.target"
expected_main_service_timeout = 300 expected_main_service_timeout = 300
def test_clean_traefik_state(self, fcos_host, keep):
if keep:
# Stop the traefik.target to ensure a clean state for the tests, but only if --keep is set because otherwise the VM is not reused across runs and is already in a clean state.
result = fcos_host.run("systemctl stop traefik.target")
assert result.rc == 0, f"Failed to stop traefik.target: {result.stderr}"
fcos_host.run("rm -rf /var/lib/quadlets/traefik/acme.json")
result = fcos_host.run("systemctl start traefik.target")
assert result.rc == 0, f"Failed to start traefik.target: {result.stderr}"
else:
pytest.skip("Skipping clean Traefik state test because --keep is not set.")
@pytest.mark.flaky(reruns=6, reruns_delay=5) @pytest.mark.flaky(reruns=6, reruns_delay=5)
def test_traefik_ping_localhost(self, fcos_host): def test_traefik_ping_localhost(self, fcos_host):
"""Traefik must respond to the ping endpoint with HTTP 200.""" """Traefik must respond to the ping endpoint with HTTP 200."""
@ -165,7 +156,7 @@ class TestTraefikQuadlet(test_quadlet.TestQuadlet):
assert result.rc == 0, f"curl failed with exit code {result.rc}: {result.stderr}" assert result.rc == 0, f"curl failed with exit code {result.rc}: {result.stderr}"
assert result.stdout.strip() == "200", f"Expected HTTP 200 from ping endpoint, got: {result.stdout.strip()}" assert result.stdout.strip() == "200", f"Expected HTTP 200 from ping endpoint, got: {result.stdout.strip()}"
def test_traefik_ping_external(self, fcos_vm, top_level_domain): def test_traefik_reject_ping_external(self, fcos_vm, top_level_domain):
"""Traefik must NOT respond to the ping endpoint outside localhost.""" """Traefik must NOT respond to the ping endpoint outside localhost."""
result = subprocess.run( result = subprocess.run(
[ [
@ -182,11 +173,14 @@ class TestTraefikQuadlet(test_quadlet.TestQuadlet):
assert result.returncode == 22, f"curl failed with exit code {result.returncode}: {result.stderr}" assert result.returncode == 22, f"curl failed with exit code {result.returncode}: {result.stderr}"
assert int(result.stdout.strip()) == 403, f"Expected HTTP 403 from ping endpoint, got: {result.stdout.strip()}" assert int(result.stdout.strip()) == 403, f"Expected HTTP 403 from ping endpoint, got: {result.stdout.strip()}"
# This test is flaky because it depends on ACME certificate issuance, which can take time.
# During ACME certificate issuance, Traefik responds with a self-signed certificate which causes curl to fail with a certificate error.
# We work around this by retrying the test several times with a delay between retries.
@pytest.mark.flaky(reruns=12, reruns_delay=5) @pytest.mark.flaky(reruns=12, reruns_delay=5)
def test_traefik_tls(self, fcos_vm, pebble_acme_server, top_level_domain): def test_traefik_tls(self, fcos_vm, pebble_acme_server, top_level_domain):
"""Traefik must respond to the secure endpoint with HTTP 200.""" """Traefik must respond to the secure endpoint with HTTP 200."""
# On the host running pytest, create a temporary dir in /tmp and write the Pebble CA certificate in the pebble.pem file. # Store the Pebble CA bundle in a temporary file so that curl can use it to verify the certificate presented by Traefik.
tmpdir = tempfile.TemporaryDirectory(delete=True) tmpdir = tempfile.TemporaryDirectory(delete=True)
d = Path(tmpdir.name) d = Path(tmpdir.name)
pebble_ca_bundle_path = d / "pebble.pem" pebble_ca_bundle_path = d / "pebble.pem"

15
tests/test_quadlet.py

@ -1,6 +1,7 @@
import socket import socket
import json import json
import time import time
import pytest
class TestQuadlet: class TestQuadlet:
""" """
@ -95,6 +96,20 @@ class TestQuadlet:
If expected_main_service is set, the number of seconds to wait for it to become active before giving up and failing the tests. If expected_main_service is set, the number of seconds to wait for it to become active before giving up and failing the tests.
""" """
def test_clean_traefik_state(self, fcos_host, keep):
traefik_unit_file = fcos_host.file("/etc/systemd/system/traefik.target")
if keep and traefik_unit_file.exists:
# Clean-up the ACME storage of Traefik since it may be out-of-sync with Pebble,
# but only if --keep is set because otherwise the VM is not reused across runs
# and is already in a clean state.
result = fcos_host.run("systemctl stop traefik.target")
assert result.rc == 0, f"Failed to stop traefik.target: {result.stderr}"
fcos_host.run("rm -f /var/lib/quadlets/traefik/acme.json")
result = fcos_host.run("systemctl start traefik.target")
assert result.rc == 0, f"Failed to start traefik.target: {result.stderr}"
else:
pytest.skip("No need to clean Traefik state.")
def test_wait_for_main_service(self, fcos_host): def test_wait_for_main_service(self, fcos_host):
"""Wait for the expected main service to become active before running any other tests.""" """Wait for the expected main service to become active before running any other tests."""
if self.expected_main_service is None: if self.expected_main_service is None:

Loading…
Cancel
Save