boma/scripts/integration-vm.py

#!/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 @<file>` 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 = """<network>
  <name>boma-it</name>
  <forward mode='nat'/>
  <bridge name='virbr-boma' stp='on' delay='0'/>
  <ip address='192.168.150.1' netmask='255.255.255.0'>
    <dhcp><range start='192.168.150.10' end='192.168.150.254'/></dhcp>
  </ip>
</network>
"""
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 dict.fromkeys(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())