diff --git a/scripts/integration-vm.py b/scripts/integration-vm.py new file mode 100644 index 0000000..36386ae --- /dev/null +++ b/scripts/integration-vm.py @@ -0,0 +1,126 @@ +#!/usr/bin/env python3 +"""boma local-VM integration test harness driver (ADR-025). + +Stdlib-only by convention (TODO-14): never imports a YAML library. The transient +inventory is emitted via string templates; stubs/cert-tiers reach Ansible as +`-e @` extra-vars; profile metadata is JSON. Talks to libvirt via `virsh`. +""" +import argparse +import hashlib +import json +import os +import pathlib +import re +import shutil +import subprocess +import sys +import time +import urllib.request +import uuid + +REPO_ROOT = pathlib.Path(__file__).resolve().parent.parent +CACHE_DIR = pathlib.Path(os.environ.get("BOMA_IT_CACHE", "/var/lib/boma-integration")) +IMAGE_URL = "https://cloud.debian.org/images/cloud/trixie/latest/debian-13-genericcloud-amd64.qcow2" +SHA_URL = "https://cloud.debian.org/images/cloud/trixie/latest/SHA512SUMS" +IMAGE_NAME = "debian-13-genericcloud-amd64.qcow2" +NET_NAME = "boma-it" +NET_XML = """ + boma-it + + + + + + +""" +NAME_PREFIX = "boma-it-" +RUN_DIR = REPO_ROOT / "tests" / "integration" / ".run" +DIAG_ROOT = pathlib.Path.home() / "integration-runs" +PROFILE_DIR = REPO_ROOT / "tests" / "integration" / "profiles" +INTEG_DIR = REPO_ROOT / "tests" / "integration" +CERT_DIR = REPO_ROOT / "tests" / "integration" / "certs" +DEFAULT_MEM_MIB = 3072 +DEFAULT_VCPUS = 2 +MIN_FREE_MIB = 4096 +VALID_TIERS = ("internal", "le-staging", "le-prod-wildcard") + +DISPATCH = {} # temporary; real dispatch added in a later task + + +def vm_name(host, suffix=None): + suffix = suffix or uuid.uuid4().hex[:8] + return f"{NAME_PREFIX}{host}-{suffix}" + + +def free_mib(meminfo_text): + m = re.search(r"^MemAvailable:\s+(\d+)\s+kB", meminfo_text, re.MULTILINE) + return int(m.group(1)) // 1024 if m else 0 + + +def parse_lease_ip(domifaddr_output): + m = re.search(r"ipv4\s+(\d+\.\d+\.\d+\.\d+)", domifaddr_output) + return m.group(1) if m else None + + +def render_meta_data(instance_id, hostname): + return f"instance-id: {instance_id}\nlocal-hostname: {hostname}\n" + + +def render_user_data(ssh_pubkey, ansible_user): + return ( + "#cloud-config\n" + "users:\n" + f" - name: {ansible_user}\n" + " sudo: 'ALL=(ALL) NOPASSWD:ALL'\n" + " shell: /bin/bash\n" + " ssh_authorized_keys:\n" + f" - {ssh_pubkey}\n" + "ssh_pwauth: false\n" + "package_update: false\n" + ) + + +def cert_file(tier): + if tier not in VALID_TIERS: + raise ValueError(f"unknown cert tier: {tier}") + return CERT_DIR / f"{tier}.yml" + + +def profile_path(host): + return PROFILE_DIR / f"{host}.json" + + +def render_run_hosts(name, ip, ansible_user, groups): + lines = [ + "# Generated by scripts/integration-vm.py — transient, gitignored. Do not edit.", + "# Single test host ONLY (safety invariant: no real host is ever in scope).", + "all:", + " children:", + ] + for g in groups: + lines += [ + f" {g}:", + " hosts:", + f" {name}:", + f" ansible_host: {ip}", + f" ansible_user: {ansible_user}", + ] + return "\n".join(lines) + "\n" + + +def main(argv=None): + p = argparse.ArgumentParser(prog="integration-vm", description=__doc__) + sub = p.add_subparsers(dest="cmd", required=True) + for c in ("up", "apply", "reboot", "assert", "cycle", "down", "console"): + sp = sub.add_parser(c) + sp.add_argument("--host", required=True) + sp.add_argument("--certs", choices=VALID_TIERS, default="internal") + sp.add_argument("--keep", action="store_true") + sp.add_argument("--no-reboot", action="store_true") + sub.add_parser("prune") + args = p.parse_args(argv) + return DISPATCH[args.cmd](args) + + +if __name__ == "__main__": # pragma: no cover + sys.exit(main()) diff --git a/tests/test_integration_vm.py b/tests/test_integration_vm.py new file mode 100644 index 0000000..4eab622 --- /dev/null +++ b/tests/test_integration_vm.py @@ -0,0 +1,11 @@ +import importlib.util +import pathlib + +_PATH = pathlib.Path(__file__).resolve().parent.parent / "scripts" / "integration-vm.py" +_spec = importlib.util.spec_from_file_location("integration_vm", _PATH) +ivm = importlib.util.module_from_spec(_spec) +_spec.loader.exec_module(ivm) + + +def test_valid_tiers(): + assert ivm.VALID_TIERS == ("internal", "le-staging", "le-prod-wildcard")