240 lines
8.5 KiB
Python
240 lines
8.5 KiB
Python
from __future__ import annotations
|
|
|
|
from contextlib import redirect_stdout
|
|
import io
|
|
import json
|
|
from pathlib import Path
|
|
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,
|
|
_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))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|