from __future__ import annotations from contextlib import redirect_stdout from dataclasses import replace import io import json from pathlib import Path import shutil import stat import subprocess import sys import tempfile import unittest from unittest.mock import MagicMock META_ROOT = Path(__file__).resolve().parents[1] LAB_TOOLS = META_ROOT / "tools" / "lab" if str(LAB_TOOLS) not in sys.path: sys.path.insert(0, str(LAB_TOOLS)) from govoplan_lab.cli import main # noqa: E402 from govoplan_lab.config import LabConfigError, load_config # noqa: E402 from govoplan_lab.lifecycle import ( # noqa: E402 CommandRunner, LabOperationError, _assert_domain_owned, _domain_description, _ensure_certificates, _render_kubectl_wrapper, destroy, ) from govoplan_lab.render import ( # noqa: E402 render_k3s_config, render_registry_config, render_state_compose, write_private, ) REHEARSAL_CONFIG = LAB_TOOLS / "govoplan-lab.example.toml" ACCEPTANCE_CONFIG = LAB_TOOLS / "govoplan-lab.acceptance.example.toml" class KubernetesLabTests(unittest.TestCase): def test_example_inventories_describe_their_evidence_boundary(self) -> None: rehearsal = load_config(REHEARSAL_CONFIG) acceptance = load_config(ACCEPTANCE_CONFIG) self.assertEqual("rehearsal", rehearsal.mode) self.assertFalse(rehearsal.evidence_capable) self.assertEqual(2, len(rehearsal.workers)) self.assertEqual("acceptance", acceptance.mode) self.assertTrue(acceptance.evidence_capable) self.assertEqual(3, len({node.hypervisor for node in acceptance.nodes})) self.assertEqual(3, len({node.failure_domain for node in acceptance.nodes})) def test_acceptance_inventory_rejects_collapsed_worker_failure_domains(self) -> None: source = ACCEPTANCE_CONFIG.read_text(encoding="utf-8") collapsed = source.replace( 'hypervisor = "lab-admin@hypervisor-b.example.org"', 'hypervisor = "lab-admin@hypervisor-a.example.org"', ).replace('failure_domain = "rack-b"', 'failure_domain = "rack-a"') with tempfile.TemporaryDirectory(prefix="govoplan-lab-config-") as directory: path = Path(directory) / "lab.toml" path.write_text(collapsed, encoding="utf-8") with self.assertRaisesRegex(LabConfigError, "acceptance mode"): load_config(path) def test_create_without_apply_is_a_non_mutating_preview(self) -> None: output = io.StringIO() with redirect_stdout(output): exit_code = main(["--config", str(REHEARSAL_CONFIG), "create"]) self.assertEqual(0, exit_code) self.assertIn("Dry run: create", output.getvalue()) self.assertIn("Re-run with --apply", output.getvalue()) def test_local_hypervisor_uses_system_libvirt_without_sudo(self) -> None: config = load_config(REHEARSAL_CONFIG) runner = CommandRunner(config) runner.run = MagicMock( return_value=subprocess.CompletedProcess([], 0, stdout=b"", stderr=b"") ) runner.hypervisor(config.nodes[0], ["virsh", "dominfo", "test-domain"]) runner.run.assert_called_once_with( [ "virsh", "--connect", "qemu:///system", "dominfo", "test-domain", ], capture=False, check=True, timeout=None, ) def test_local_hypervisor_file_operations_do_not_use_sudo(self) -> None: config = load_config(REHEARSAL_CONFIG) runner = CommandRunner(config) runner.run = MagicMock( return_value=subprocess.CompletedProcess([], 0, stdout=b"", stderr=b"") ) runner.hypervisor(config.nodes[0], ["install", "-d", "/tmp/lab"]) runner.run.assert_called_once_with( ["install", "-d", "/tmp/lab"], capture=False, check=True, timeout=None, ) def test_kubectl_wrapper_quotes_remote_arguments(self) -> None: wrapper = _render_kubectl_wrapper( ["ssh", "-i", "/tmp/lab key", "govoplan@example.test"] ) self.assertIn("shlex.join(_REMOTE)", wrapper) self.assertIn('["sudo", "--", "k3s", "kubectl", *sys.argv[1:]]', wrapper) self.assertNotIn('kubectl \"$@\"', wrapper) def test_destroy_requires_the_exact_lab_name(self) -> None: config = load_config(REHEARSAL_CONFIG) with self.assertRaisesRegex(LabOperationError, "--confirm"): destroy( config, apply=True, confirmation="wrong-lab", purge_local_state=False, ) def test_enroll_admin_without_apply_is_a_non_mutating_preview(self) -> None: output = io.StringIO() with redirect_stdout(output): exit_code = main( [ "--config", str(REHEARSAL_CONFIG), "enroll-admin", "--email", "owner@example.test", ] ) self.assertEqual(0, exit_code) self.assertIn("Dry run: enroll-admin", output.getvalue()) self.assertIn("owner@example.test", output.getvalue()) def test_domain_ownership_requires_marker_and_expected_disks(self) -> None: config = load_config(REHEARSAL_CONFIG) node = config.nodes[0] node_directory = f"{config.vm_image_directory}/{config.name}/{node.name}" runner = MagicMock() runner.hypervisor.side_effect = [ subprocess.CompletedProcess( [], 0, stdout=(_domain_description(config, node) + "\n").encode(), stderr=b"", ), subprocess.CompletedProcess( [], 0, stdout=( f"file disk vda {node_directory}/root.qcow2\n" f"file cdrom sda {node_directory}/seed.img\n" ).encode(), stderr=b"", ), ] _assert_domain_owned(config, runner, node) runner.hypervisor.side_effect = [ subprocess.CompletedProcess( [], 0, stdout=b"unrelated domain\n", stderr=b"", ) ] with self.assertRaisesRegex(LabOperationError, "ownership marker"): _assert_domain_owned(config, runner, node) def test_state_compose_uses_only_supplied_pinned_images(self) -> None: names = ("postgres", "redis", "garage", "managed_ingress", "test_mail") images = { name: f"registry.example.test/{name}@sha256:{index:064x}" for index, name in enumerate(names, start=1) } compose = json.loads(render_state_compose(images)) self.assertEqual(images["postgres"], compose["services"]["postgres"]["image"]) self.assertEqual(images["redis"], compose["services"]["redis"]["image"]) self.assertEqual(images["garage"], compose["services"]["garage"]["image"]) self.assertEqual( images["managed_ingress"], compose["services"]["s3-tls"]["image"] ) self.assertEqual( images["test_mail"], compose["services"]["test-mail"]["image"] ) def test_k3s_workers_join_the_primary_control_and_receive_failure_labels( self, ) -> None: config = load_config(ACCEPTANCE_CONFIG) worker = config.workers[0] rendered = render_k3s_config(config, worker, cluster_token="test-token") self.assertIn(f'server: "https://{config.primary_control.address}:6443"', rendered) self.assertIn( f'topology.govoplan.add-ideas.de/failure-domain={worker.failure_domain}', rendered, ) self.assertNotIn("cluster-init", rendered) def test_registry_credentials_are_all_or_nothing(self) -> None: self.assertEqual("", render_registry_config("", "")) with self.assertRaisesRegex(ValueError, "supplied together"): render_registry_config("publisher", "") rendered = render_registry_config("publisher", "secret") self.assertIn('"git.add-ideas.de"', rendered) self.assertIn("publisher", rendered) def test_private_writer_enforces_owner_only_permissions(self) -> None: with tempfile.TemporaryDirectory(prefix="govoplan-lab-private-") as directory: path = Path(directory) / "nested" / "secret.txt" write_private(path, "secret\n") self.assertEqual(0o600, stat.S_IMODE(path.stat().st_mode)) @unittest.skipUnless(shutil.which("openssl"), "openssl is required") def test_generated_lab_ca_passes_strict_chain_validation(self) -> None: with tempfile.TemporaryDirectory(prefix="govoplan-lab-pki-") as directory: config = replace( load_config(REHEARSAL_CONFIG), state_directory=Path(directory), ) _ensure_certificates(config, CommandRunner(config)) ca_certificate = config.state_directory / "pki" / "ca.crt" server_certificate = config.state_directory / "pki" / "server.crt" result = subprocess.run( [ "openssl", "verify", "-x509_strict", "-CAfile", str(ca_certificate), str(server_certificate), ], stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False, ) self.assertEqual( 0, result.returncode, (result.stdout + result.stderr).decode(errors="replace"), ) if __name__ == "__main__": unittest.main()