inecs
f8d3cbdd59
Add the FastAPI app, PostgreSQL migrations, Docker/Helm packaging, API contracts, docs, client examples, and the unit/integration/compatibility test suite for local client and tooling labs without a real vCenter.
103 lines
3.8 KiB
Python
103 lines
3.8 KiB
Python
#!/usr/bin/env python3
|
|
"""Spot-check that critical Automation API GETs return real seeded payloads."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import ssl
|
|
import sys
|
|
import urllib.error
|
|
import urllib.request
|
|
from base64 import b64encode
|
|
|
|
BASE = os.getenv("VSPHERE_BASE", "https://localhost")
|
|
USER = os.getenv("VSPHERE_USER", "administrator@vsphere.local")
|
|
PASSWORD = os.getenv("VSPHERE_PASSWORD", "VMware1!")
|
|
|
|
SPOTS = [
|
|
("/api/vcenter/vm", "list"),
|
|
("/api/vcenter/host", "list"),
|
|
("/api/vcenter/datastore", "list"),
|
|
("/api/vcenter/network", "list"),
|
|
("/api/vcenter/cluster", "list"),
|
|
("/api/content/library", "list"),
|
|
("/api/cis/tagging/category", "list"),
|
|
("/api/appliance/access/ssh", "object"),
|
|
("/api/appliance/services", "list"),
|
|
("/api/esx/settings/clusters/domain-c21/software", "object"),
|
|
("/api/vcenter/namespace-management/supervisors/supervisor-1/summary", "object"),
|
|
("/api/vcenter/crypto-manager/kms/providers", "list"),
|
|
("/api/vcenter/vm/vm-101/hardware/cdrom", "list"),
|
|
("/api/vcenter/vm/vm-101/hardware/disk", "list"),
|
|
("/api/vcenter/host/host-11/networking", "object"),
|
|
("/api/vcenter/host/host-11/storage/storage-device", "list"),
|
|
("/api/vcenter/storage/policies", "list"),
|
|
("/api/vcenter/guest/customization-specs", "list"),
|
|
("/api/vcenter/identity/providers", "list"),
|
|
("/api/vcenter/certificate-management/vcenter/tls", "object"),
|
|
]
|
|
|
|
|
|
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, str]:
|
|
req = urllib.request.Request(f"{BASE}{path}", method=method, headers=headers)
|
|
try:
|
|
with urllib.request.urlopen(req, context=_ctx()) as resp: # noqa: S310
|
|
return int(resp.status), resp.read().decode("utf-8", errors="replace")
|
|
except urllib.error.HTTPError as error:
|
|
return int(error.code), error.read().decode("utf-8", errors="replace")
|
|
|
|
|
|
def main() -> int:
|
|
basic = b64encode(f"{USER}:{PASSWORD}".encode()).decode()
|
|
code, body = _request("POST", "/api/session", headers={"Authorization": f"Basic {basic}"})
|
|
if code not in {200, 201}:
|
|
print(json.dumps({"error": f"session failed {code}", "body": body[:200]}))
|
|
return 1
|
|
session = json.loads(body)
|
|
headers = {"vmware-api-session-id": session, "Accept": "application/json"}
|
|
failures: list[dict[str, object]] = []
|
|
ok = 0
|
|
for path, kind in SPOTS:
|
|
status, raw = _request("GET", path, headers=headers)
|
|
if status != 200:
|
|
failures.append({"path": path, "status": status, "body": raw[:160]})
|
|
continue
|
|
if '"stub": true' in raw or '"stub":true' in raw:
|
|
failures.append(
|
|
{"path": path, "status": status, "reason": "stub marker", "body": raw[:160]}
|
|
)
|
|
continue
|
|
try:
|
|
payload = json.loads(raw)
|
|
except json.JSONDecodeError:
|
|
failures.append(
|
|
{"path": path, "status": status, "reason": "non-json", "body": raw[:160]}
|
|
)
|
|
continue
|
|
if kind == "list":
|
|
if not isinstance(payload, list) or len(payload) < 1:
|
|
failures.append(
|
|
{"path": path, "status": status, "reason": "empty list", "body": raw[:160]}
|
|
)
|
|
continue
|
|
else:
|
|
if not isinstance(payload, dict) or not payload:
|
|
failures.append(
|
|
{"path": path, "status": status, "reason": "empty object", "body": raw[:160]}
|
|
)
|
|
continue
|
|
ok += 1
|
|
print(json.dumps({"checked": len(SPOTS), "ok": ok, "failures": failures}, indent=2))
|
|
return 1 if failures else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|