74 lines
3.3 KiB
Python
74 lines
3.3 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import sys
|
|
import unittest
|
|
from unittest import mock
|
|
from pathlib import Path
|
|
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[3]
|
|
CLI_PATH = REPO_ROOT / "pkgs" / "helpers" / "cli.py"
|
|
|
|
spec = importlib.util.spec_from_file_location("nodeiwest_cli", CLI_PATH)
|
|
cli = importlib.util.module_from_spec(spec)
|
|
assert spec.loader is not None
|
|
sys.modules[spec.name] = cli
|
|
spec.loader.exec_module(cli)
|
|
|
|
|
|
class HelperCliTests(unittest.TestCase):
|
|
def test_disk_from_device_supports_sd_and_nvme(self) -> None:
|
|
self.assertEqual(cli.disk_from_device("/dev/sda2"), "/dev/sda")
|
|
self.assertEqual(cli.disk_from_device("/dev/nvme0n1p2"), "/dev/nvme0n1")
|
|
|
|
def test_lookup_colmena_target_host_reads_existing_inventory(self) -> None:
|
|
flake_text = (REPO_ROOT / "flake.nix").read_text()
|
|
self.assertEqual(cli.lookup_colmena_target_host(flake_text, "vps1"), "100.101.167.118")
|
|
|
|
def test_parse_existing_vps1_configuration(self) -> None:
|
|
configuration = cli.parse_existing_configuration(REPO_ROOT / "hosts" / "vps1" / "configuration.nix")
|
|
self.assertEqual(configuration.host_name, "vps1")
|
|
self.assertEqual(configuration.boot_mode, "uefi")
|
|
self.assertTrue(configuration.tailscale_openbao)
|
|
self.assertEqual(configuration.state_version, "25.05")
|
|
self.assertTrue(configuration.user_ca_public_keys)
|
|
|
|
def test_parse_existing_vps1_disko(self) -> None:
|
|
disko = cli.parse_existing_disko(REPO_ROOT / "hosts" / "vps1" / "disko.nix")
|
|
self.assertEqual(disko.disk_device, "/dev/sda")
|
|
self.assertEqual(disko.boot_mode, "uefi")
|
|
self.assertEqual(disko.swap_size, "4GiB")
|
|
|
|
def test_render_bios_disko_uses_bios_partition(self) -> None:
|
|
rendered = cli.render_disko(boot_mode="bios", disk_device="/dev/vda", swap_size="8GiB")
|
|
self.assertIn('type = "EF02";', rendered)
|
|
self.assertIn('device = lib.mkDefault "/dev/vda";', rendered)
|
|
self.assertIn('size = "8GiB";', rendered)
|
|
|
|
def test_bao_kv_get_uses_explicit_kv_mount(self) -> None:
|
|
completed = mock.Mock()
|
|
completed.stdout = '{"data": {"data": {"CLIENT_ID": "x"}}}'
|
|
with mock.patch.object(cli, "run_command", return_value=completed) as run_command:
|
|
data = cli.bao_kv_get("it", "kv", "tailscale")
|
|
|
|
self.assertEqual(data["data"]["data"]["CLIENT_ID"], "x")
|
|
command = run_command.call_args.args[0]
|
|
self.assertEqual(command, ["bao", "kv", "get", "-mount=kv", "-format=json", "tailscale"])
|
|
self.assertEqual(run_command.call_args.kwargs["env"], {"BAO_NAMESPACE": "it"})
|
|
|
|
def test_derive_openbao_policy_uses_explicit_kv_mount(self) -> None:
|
|
completed = mock.Mock()
|
|
completed.stdout = 'path "kv/data/tailscale" { capabilities = ["read"] }\n'
|
|
with mock.patch.object(cli, "run_command", return_value=completed) as run_command:
|
|
policy = cli.derive_openbao_policy("it", "kv", "tailscale")
|
|
|
|
self.assertIn('path "kv/data/tailscale"', policy)
|
|
command = run_command.call_args.args[0]
|
|
self.assertEqual(command, ["bao", "kv", "get", "-mount=kv", "-output-policy", "tailscale"])
|
|
self.assertEqual(run_command.call_args.kwargs["env"], {"BAO_NAMESPACE": "it"})
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|