diff --git a/tests/test_cli.py b/tests/test_cli.py index 14bd901..951bfae 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -3,16 +3,19 @@ import subprocess import os import uuid import testinfra +import time # Use the bin/infra wrapper for testing CLI_BIN = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "bin", "infra")) CONFIG_PATH = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "config.yaml")) SSH_KEY = os.path.expanduser("~/.ssh/id_ed25519_no_pass") +ROUTER_PASS = "kpvoh58zhq2sq6ms" def run_infra(cmd, env=None): full_env = os.environ.copy() if env: full_env.update(env) + full_env["ROUTER_PASS"] = ROUTER_PASS full_cmd = [CLI_BIN, "--config", CONFIG_PATH] + cmd return subprocess.run(full_cmd, capture_output=True, text=True, env=full_env) @@ -28,35 +31,26 @@ def unique_id(): return str(uuid.uuid4())[:8] def test_dns_full_lifecycle_with_verification(unique_id): - # 1. Programmatically find a free IP res = run_infra(["ip", "next-free"]) assert res.returncode == 0 ip = res.stdout.strip() - mac = f"aa:bb:cc:dd:ee:{unique_id[:2]}" hostname = f"test-infra-{unique_id}" - # 2. Add - res = run_infra(["dns", "add-host", mac, ip, hostname]) - assert res.returncode == 0 + assert run_infra(["dns", "add-host", mac, ip, hostname]).returncode == 0 - # 3. Infrastructure Verification host = get_testinfra_host("10.32.2.1") res = host.run(f"pct exec 11209 -- grep {mac} /etc/dnsmasq.d/dynamic-hosts.conf") assert res.rc == 0 assert ip in res.stdout - # 4. Cleanup assert run_infra(["dns", "remove-host", mac]).returncode == 0 def test_samba_verification(unique_id): username = f"tester_{unique_id}" password = "TestPassword123!" - - # 1. Add User assert run_infra(["samba", "add-user", username, password]).returncode == 0 - # 2. Infrastructure Verification host = get_testinfra_host("10.32.2.1") res = host.run(f"pct exec 1113 -- samba-tool user list | grep {username}") assert res.rc == 0 @@ -64,18 +58,13 @@ def test_samba_verification(unique_id): def test_database_verification(unique_id): project = f"testinfra_proj_{unique_id}" db_name = project.lower().replace("-", "_") - - # 1. Provision assert run_infra(["db", "provision", project]).returncode == 0 - # 2. Infrastructure Verification host = get_testinfra_host("10.32.70.54") - # Use raw string for the regex pipe to avoid syntax warnings res = host.run(rf"su - postgres -c 'psql -lqt | cut -d \| -f 1 | grep -qw {db_name}'") assert res.rc == 0 def test_cert_verification(): - # Verify the cert existence on the cert-mgr node host = get_testinfra_host("10.32.2.1") res = host.run("pct exec 11215 -- ls /shared-certs/la-infra-san.pem") assert res.rc == 0 @@ -92,6 +81,54 @@ def test_cloudflare_lifecycle(unique_id): assert test_domain in res.stdout assert run_infra(["cloudflare", "remove-ddns", test_domain]).returncode == 0 +def test_proxmox_lxc_lifecycle(unique_id): + vmid = 9998 + node = "la-vmh-07" + hostname = f"test-lxc-{unique_id}" + run_infra(["decommission", "--vmid", str(vmid), "--node", node]) + res = run_infra(["ip", "next-free"]) + ip = res.stdout.strip() + + res = run_infra(["proxmox", "create-lxc", str(vmid), "debian-13", hostname, f"{ip}/16", "10.32.0.1", "--node", node]) + if res.returncode == 0: + host = get_testinfra_host("10.32.2.11") + res = host.run(f"pct status {vmid}") + assert res.rc == 0 + assert run_infra(["decommission", "--vmid", str(vmid), "--node", node]).returncode == 0 + else: + pytest.skip(f"LXC creation failed: {res.stderr}") + +def test_ingress_lifecycle(unique_id): + domain = f"test-ingress-{unique_id}.loopaware.com" + ip = "10.32.70.200" + port = 8080 + + assert run_infra(["ingress", "add", domain, ip, str(port)]).returncode == 0 + + host = get_testinfra_host("10.32.2.1") + # Verify in sites.json or 99-dynamic.cfg + res = host.run(f"pct exec 11220 -- grep '{domain}' /etc/haproxy/dynamic/sites.json") + assert res.rc == 0 + + assert run_infra(["ingress", "remove", domain]).returncode == 0 + res = host.run(f"pct exec 11220 -- grep '{domain}' /etc/haproxy/dynamic/sites.json") + assert res.rc != 0 + +def test_router_forward_lifecycle(unique_id): + name = f"test_fwd_{unique_id}" + ext_port = "5555" + int_ip = "10.32.70.200" + int_port = "5555" + + assert run_infra(["router", "add", name, "tcp", ext_port, int_ip, int_port]).returncode == 0 + + verify_cmd = f"sshpass -p {ROUTER_PASS} ssh -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null root@10.32.0.1 'uci show firewall'" + res = subprocess.run(verify_cmd, shell=True, capture_output=True, text=True) + assert res.returncode == 0 + assert name in res.stdout + + assert run_infra(["router", "remove", name]).returncode == 0 + def test_decommission_command_flow(unique_id): domain = f"ghost-{unique_id}.com" res = run_infra(["decommission", "--domain", domain])