#!/usr/bin/env python3 """Audit seeded inventory against live Automation API responses. Compares the declarative profile (small / large / big) to live GET /api/vcenter/* dumps: counts, MOID/name/power for every VM and host, per-host placement via filter, and a canonical AND-filter that must hit on all profiles (web-01 @ host-11, POWERED_ON). Run against a freshly seeded lab (``make seed``) before matrix probes mutate inventory. Exit 0 only on a 100% dump match. """ from __future__ import annotations import argparse import json import os import ssl import sys import urllib.error import urllib.request from base64 import b64encode from collections import defaultdict from typing import Any from urllib.parse import urlencode from app.vsphere.profiles import build_vsphere_profile BASE = os.getenv("VSPHERE_BASE", "https://localhost") USER = os.getenv("VSPHERE_USER", "administrator@vsphere.local") PASSWORD = os.getenv("VSPHERE_PASSWORD", "VMware1!") # Works on small (3h), large (10h), and big (20h). CANONICAL_FILTER = { "names": "web-01", "hosts": "host-11", "power_states": "POWERED_ON", } def _ctx() -> ssl.SSLContext | None: if not BASE.startswith("https://"): return None return ssl._create_unverified_context() # noqa: S323 def _request(method: str, path: str, *, headers: dict[str, str]) -> tuple[int, Any]: req = urllib.request.Request(f"{BASE}{path}", method=method, headers=headers) try: with urllib.request.urlopen(req, context=_ctx(), timeout=120) as resp: # noqa: S310 raw = resp.read().decode("utf-8", errors="replace") return int(resp.status), json.loads(raw) if raw.strip() else None except urllib.error.HTTPError as error: raw = error.read().decode("utf-8", errors="replace") try: body = json.loads(raw) if raw.strip() else raw except json.JSONDecodeError: body = raw return int(error.code), body def _session() -> str: basic = b64encode(f"{USER}:{PASSWORD}".encode()).decode() code, body = _request("POST", "/api/session", headers={"Authorization": f"Basic {basic}"}) if code not in {200, 201} or not isinstance(body, str): raise SystemExit(json.dumps({"error": "session failed", "status": code, "body": body})) return body def _by_type(profile_objects: tuple[Any, ...]) -> dict[str, list[Any]]: out: dict[str, list[Any]] = {} for obj in profile_objects: out.setdefault(obj.type, []).append(obj) return out def audit_profile(profile_name: str, *, hosts: int | None, vms: int | None) -> dict[str, Any]: profile = build_vsphere_profile(profile_name, large_hosts=hosts, large_vms=vms) expected = _by_type(profile.objects) session = _session() headers = {"vmware-api-session-id": session, "Accept": "application/json"} failures: list[dict[str, Any]] = [] live_vms_code, live_vms = _request("GET", "/api/vcenter/vm", headers=headers) live_hosts_code, live_hosts = _request("GET", "/api/vcenter/host", headers=headers) live_ds_code, live_ds = _request("GET", "/api/vcenter/datastore", headers=headers) live_net_code, live_net = _request("GET", "/api/vcenter/network", headers=headers) live_cl_code, live_cl = _request("GET", "/api/vcenter/cluster", headers=headers) live_dc_code, live_dc = _request("GET", "/api/vcenter/datacenter", headers=headers) live_folder_code, live_folder = _request("GET", "/api/vcenter/folder", headers=headers) live_rp_code, live_rp = _request("GET", "/api/vcenter/resource-pool", headers=headers) for label, code, payload in ( ("vm", live_vms_code, live_vms), ("host", live_hosts_code, live_hosts), ("datastore", live_ds_code, live_ds), ("network", live_net_code, live_net), ("cluster", live_cl_code, live_cl), ("datacenter", live_dc_code, live_dc), ("folder", live_folder_code, live_folder), ("resource-pool", live_rp_code, live_rp), ): if code != 200 or not isinstance(payload, list): failures.append({"check": f"list/{label}", "status": code, "body": str(payload)[:160]}) exp_vms = expected.get("VirtualMachine", []) exp_hosts = expected.get("HostSystem", []) exp_ds = expected.get("Datastore", []) exp_nets = [ *expected.get("Network", []), *expected.get("DistributedVirtualPortgroup", []), ] exp_clusters = expected.get("ClusterComputeResource", []) exp_dcs = expected.get("Datacenter", []) exp_folders = expected.get("Folder", []) exp_rps = expected.get("ResourcePool", []) def _count(label: str, got: Any, want: int) -> None: if not isinstance(got, list): return if len(got) != want: failures.append({"check": f"count/{label}", "expected": want, "actual": len(got)}) _count("vm", live_vms, len(exp_vms)) _count("host", live_hosts, len(exp_hosts)) _count("datastore", live_ds, len(exp_ds)) _count("network", live_net, len(exp_nets)) _count("cluster", live_cl, len(exp_clusters)) _count("datacenter", live_dc, len(exp_dcs)) _count("folder", live_folder, len(exp_folders)) _count("resource-pool", live_rp, len(exp_rps)) if isinstance(live_hosts, list): live_host_map = {row.get("host"): row for row in live_hosts if isinstance(row, dict)} for obj in exp_hosts: row = live_host_map.get(obj.moid) if row is None: failures.append({"check": "host/missing", "host": obj.moid}) continue if row.get("name") != obj.name: failures.append( { "check": "host/name", "host": obj.moid, "expected": obj.name, "actual": row.get("name"), } ) if isinstance(live_vms, list): live_vm_map = {row.get("vm"): row for row in live_vms if isinstance(row, dict)} for obj in exp_vms: row = live_vm_map.get(obj.moid) if row is None: failures.append({"check": "vm/missing", "vm": obj.moid, "name": obj.name}) continue want = { "name": obj.name, "power_state": obj.props.get("power_state"), "cpu_count": obj.props.get("cpu_count"), "memory_size_MiB": obj.props.get("memory_size_mib"), } for key, expected_value in want.items(): if row.get(key) != expected_value: failures.append( { "check": f"vm/{key}", "vm": obj.moid, "expected": expected_value, "actual": row.get(key), } ) # Per-host placement dump via filter (O(hosts), not O(vms)). by_host: dict[str, set[str]] = defaultdict(set) for obj in exp_vms: by_host[str(obj.props.get("host"))].add(obj.moid) for host_moid, want_ids in sorted(by_host.items()): qs = urlencode({"hosts": host_moid}) code, filtered = _request("GET", f"/api/vcenter/vm?{qs}", headers=headers) if code != 200 or not isinstance(filtered, list): failures.append({"check": "host-filter", "host": host_moid, "status": code}) continue got_ids = {row.get("vm") for row in filtered if isinstance(row, dict)} missing = sorted(want_ids - got_ids) extra = sorted(got_ids - want_ids) if missing or extra: failures.append( { "check": "host-filter/mismatch", "host": host_moid, "missing": missing[:20], "extra": extra[:20], "expected": len(want_ids), "actual": len(got_ids), } ) qs = urlencode(CANONICAL_FILTER) code, filtered = _request("GET", f"/api/vcenter/vm?{qs}", headers=headers) if code != 200 or not isinstance(filtered, list) or len(filtered) != 1: failures.append( { "check": "canonical-filter", "query": CANONICAL_FILTER, "status": code, "hits": filtered if not isinstance(filtered, list) else len(filtered), "body": filtered[:3] if isinstance(filtered, list) else filtered, } ) elif filtered[0].get("name") != "web-01" or filtered[0].get("vm") != "vm-101": failures.append( { "check": "canonical-filter/identity", "expected": {"vm": "vm-101", "name": "web-01"}, "actual": filtered[0], } ) detail_code, detail = _request("GET", "/api/vcenter/vm/vm-101", headers=headers) if detail_code != 200 or not isinstance(detail, dict) or detail.get("name") != "web-01": failures.append({"check": "vm/detail", "status": detail_code, "body": str(detail)[:200]}) # --- proportional platform extras --- extras_scale = int(getattr(profile, "extras_scale", 1) or 1) want_libraries = 2 + max(0, extras_scale - 1) # local+published + scaled locals want_categories = 3 + max(0, extras_scale - 1) # Environment/Owner/Lab + scaled want_folders = 8 + max(0, (extras_scale - 1) * 2) lib_code, libraries = _request("GET", "/api/content/library", headers=headers) if lib_code != 200 or not isinstance(libraries, list): failures.append({"check": "extras/libraries", "status": lib_code}) elif len(libraries) < want_libraries: failures.append( { "check": "extras/libraries/count", "expected_min": want_libraries, "actual": len(libraries), "extras_scale": extras_scale, } ) cat_code, categories = _request("GET", "/api/cis/tagging/category", headers=headers) if cat_code != 200 or not isinstance(categories, list): failures.append({"check": "extras/categories", "status": cat_code}) elif len(categories) < want_categories: failures.append( { "check": "extras/categories/count", "expected_min": want_categories, "actual": len(categories), "extras_scale": extras_scale, } ) if isinstance(live_folder, list) and len(live_folder) != want_folders: failures.append( { "check": "count/folder-scaled", "expected": want_folders, "actual": len(live_folder), "extras_scale": extras_scale, } ) # Datastore references on VMs must exist in inventory. ds_ids = {obj.moid for obj in exp_ds} for obj in exp_vms: ds = obj.props.get("datastore") if ds and ds not in ds_ids: failures.append({"check": "vm/datastore-missing", "vm": obj.moid, "datastore": ds}) break return { "base": BASE, "profile": profile.name, "extras_scale": extras_scale, "expected": { "hosts": len(exp_hosts), "vms": len(exp_vms), "datastores": len(exp_ds), "networks": len(exp_nets), "clusters": len(exp_clusters), "datacenters": len(exp_dcs), "folders": want_folders, "resource_pools": len(exp_rps), "libraries_min": want_libraries, "categories_min": want_categories, }, "live": { "hosts": len(live_hosts) if isinstance(live_hosts, list) else None, "vms": len(live_vms) if isinstance(live_vms, list) else None, "datastores": len(live_ds) if isinstance(live_ds, list) else None, "networks": len(live_net) if isinstance(live_net, list) else None, "clusters": len(live_cl) if isinstance(live_cl, list) else None, "datacenters": len(live_dc) if isinstance(live_dc, list) else None, "folders": len(live_folder) if isinstance(live_folder, list) else None, "resource_pools": len(live_rp) if isinstance(live_rp, list) else None, "libraries": len(libraries) if isinstance(libraries, list) else None, "categories": len(categories) if isinstance(categories, list) else None, }, "failure_count": len(failures), "failures": failures[:100], "ok": len(failures) == 0, } def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument( "--profile", default=os.getenv("SEED_VSPHERE_PROFILE", "large"), help="small | large | big (demo-cluster aliases big)", ) parser.add_argument("--hosts", type=int, default=None) parser.add_argument("--vms", type=int, default=None) args = parser.parse_args() report = audit_profile(args.profile, hosts=args.hosts, vms=args.vms) print(json.dumps(report, indent=2, ensure_ascii=False)) return 0 if report["ok"] else 1 if __name__ == "__main__": raise SystemExit(main())