feat(integration-vm): driver skeleton + CLI dispatch

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 @<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 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())

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")