DevOpsTools/vmware-api-simulator
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Initial commit: VMware vSphere API simulator scaffold.

Browse Source 
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.
This commit is contained in:
Sergey Antropoff
2026-07-18 04:42:11 +03:00
commit f8d3cbdd59
422 changed files with 361335 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/generate_vsphere_universe.py
+253
View File
@@ -0,0 +1,253 @@
#!/usr/bin/env python3
"""Generate REST universe stubs from Broadcom vSphere Automation operations index.
Source: contracts/vsphere/broadcom-9.1-operations-index.txt
(scraped from https://developer.broadcom.com/xapis/vsphere-automation-api/latest/operation-index/)
Output: app/vsphere/rest/universe.json
"""
from __future__ import annotations
import json
import re
import sys
from collections import Counter, defaultdict
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
INDEX = ROOT / "contracts" / "vsphere" / "broadcom-9.1-operations-index.txt"
OUT = ROOT / "app" / "vsphere" / "rest" / "universe.json"
# Title-Case service tokens that already imply a hyphenated REST segment.
_SERVICE_ALIAS: dict[tuple[str, ...], tuple[str, ...]] = {
("Cis", "Session"): ("session",),
("Content", "LocalLibrary"): ("content", "local-library"),
("Content", "SubscribedLibrary"): ("content", "subscribed-library"),
("Vcenter", "VM"): ("vcenter", "vm"),
("Vcenter", "ResourcePool"): ("vcenter", "resource-pool"),
("Vcenter", "Authorization", "Privileges"): ("vcenter", "privilege"),
("Appliance", "Timesync"): ("appliance", "timesync"),
}
# Segments that carry a resource id when used as a parent, or as a collection leaf.
_ID_NAMES: dict[str, str] = {
"vm": "vm",
"host": "host",
"hosts": "host",
"cluster": "cluster",
"clusters": "cluster",
"datacenter": "datacenter",
"datacenters": "datacenter",
"folder": "folder",
"folders": "folder",
"datastore": "datastore",
"datastores": "datastore",
"network": "network",
"networks": "network",
"resource-pool": "resource_pool",
"library": "library_id",
"libraries": "library_id",
"local-library": "library_id",
"subscribed-library": "library_id",
"item": "item_id",
"items": "item_id",
"category": "category_id",
"categories": "category_id",
"tag": "tag_id",
"tags": "tag_id",
"policy": "policy",
"policies": "policy",
"snapshot": "snapshot",
"snapshots": "snapshot",
"disk": "disk",
"disks": "disk",
"ethernet": "nic",
"cdrom": "cdrom",
"cdroms": "cdrom",
"serial": "port",
"parallel": "port",
"floppy": "floppy",
"nvme": "adapter",
"sata": "adapter",
"scsi": "adapter",
"provider": "provider",
"providers": "provider",
"task": "task",
"tasks": "task",
"permission": "permission_id",
"permissions": "permission_id",
"role": "role",
"roles": "role",
"zone": "zone",
"zones": "zone",
"project": "project",
"projects": "project",
"domain": "domain",
"domains": "domain",
"service": "service",
"services": "service",
"supervisor": "supervisor",
"supervisors": "supervisor",
"namespace": "namespace",
"namespaces": "namespace",
"depot": "depot",
"depots": "depot",
"component": "component",
"components": "component",
"image": "image",
"images": "image",
"draft": "draft",
"drafts": "draft",
"connection": "connection",
"connections": "connection",
"vpc": "vpc",
"vpcs": "vpc",
"subnet": "subnet",
"subnets": "subnet",
"download-session": "download_session_id",
"update-session": "update_session_id",
"subscription": "subscription_id",
"subscriptions": "subscription_id",
"usage": "usage_id",
"usages": "usage_id",
"library-items": "item_id",
"versions": "version",
"check-outs": "vm",
"trusted-root-chains": "chain",
"nodes": "node",
"profiles": "profile",
"interfaces": "interface",
"cores": "core",
"commit": "commit",
"commits": "commit",
}
_LEAF_ID_ACTIONS = {
"get",
"delete",
"update",
"set",
"remove",
"forceddelete",
"forcedDelete",
"forceDelete",
}
def _to_kebab(token: str) -> str:
token = token.replace("_", "-")
s = re.sub(r"([a-z0-9])([A-Z])", r"\1-\2", token)
s = re.sub(r"([A-Z]+)([A-Z][a-z])", r"\1-\2", s)
return s.lower()
def _parse_ops(text: str) -> list[tuple[str, tuple[str, ...], str]]:
ops: list[tuple[str, tuple[str, ...], str]] = []
for raw in text.splitlines():
line = raw.strip()
match = re.match(r"^(GET|POST|PUT|PATCH|DELETE)\s+(.+)$", line)
if not match:
continue
method, rest = match.group(1), match.group(2)
parts = rest.split()
if len(parts) < 2:
continue
action = parts[-1]
service = tuple(parts[:-1])
ops.append((method, service, action))
return ops
def _segments(service: tuple[str, ...]) -> list[str]:
if service in _SERVICE_ALIAS:
return list(_SERVICE_ALIAS[service])
# Prefer Vm over VM when both appear in aliases above.
return [_to_kebab(part) for part in service]
def _action_base(action: str) -> str:
return action.split("$", 1)[0]
def _build_path(service: tuple[str, ...], action: str, actions_for_service: set[str]) -> str:
action_base = _action_base(action)
segs = _segments(service)
out: list[str] = []
for index, seg in enumerate(segs):
out.append(seg)
id_name = _ID_NAMES.get(seg)
if not id_name:
continue
is_last = index == len(segs) - 1
if not is_last:
out.append("{" + id_name + "}")
continue
leaf_actions = {_action_base(a).lower() for a in actions_for_service}
collection = bool(leaf_actions & {"list", "create", "add"})
if collection and action_base.lower() in {a.lower() for a in _LEAF_ID_ACTIONS}:
out.append("{" + id_name + "}")
return "/api/" + "/".join(out)
def generate() -> dict:
text = INDEX.read_text(encoding="utf-8")
ops = _parse_ops(text)
by_service: dict[tuple[str, ...], set[str]] = defaultdict(set)
for _method, service, action in ops:
by_service[service].add(action)
routes: dict[str, dict[str, str]] = {}
# key: "VERB PATH"
for method, service, action in ops:
path = _build_path(service, action, by_service[service])
key = f"{method} {path}"
# Prefer keeping first-seen; all map to implemented stub.
routes.setdefault(
key,
{
"verb": method,
"path": path,
"service": " ".join(service),
"sample_action": action,
"status": "stub",
},
)
methods = sorted(routes.values(), key=lambda item: (item["path"], item["verb"]))
verb_counts = Counter(item["verb"] for item in methods)
payload = {
"source": "https://developer.broadcom.com/xapis/vsphere-automation-api/latest/operation-index/",
"source_label": "vSphere Automation API 9.1 (Latest) operations index",
"source_file": str(INDEX.relative_to(ROOT)),
"broadcom_operations": len(ops),
"broadcom_by_verb": dict(Counter(m for m, _s, _a in ops)),
"unique_routes": len(methods),
"unique_by_verb": dict(verb_counts),
"methods": methods,
}
OUT.parent.mkdir(parents=True, exist_ok=True)
OUT.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8")
return payload
def main() -> int:
if not INDEX.is_file():
print(f"missing index: {INDEX}", file=sys.stderr)
return 1
payload = generate()
print(
json.dumps(
{
"out": str(OUT.relative_to(ROOT)),
"broadcom_operations": payload["broadcom_operations"],
"unique_routes": payload["unique_routes"],
"unique_by_verb": payload["unique_by_verb"],
},
indent=2,
)
)
return 0
if __name__ == "__main__":
raise SystemExit(main())
scripts/probe_api_surface.py
+11
View File
@@ -0,0 +1,11 @@
#!/usr/bin/env python3
"""CLI entrypoint for the CI API surface probe."""
from __future__ import annotations
import asyncio
from app.surface_probe import main
if __name__ == "__main__":
raise SystemExit(asyncio.run(main()))
scripts/run_client_cookbooks.py
+380
View File
@@ -0,0 +1,380 @@
#!/usr/bin/env python3
"""Run Python / Ansible-uri / Terraform-data / Pulumi-style cookbooks against the simulator.
Uses only ``requests`` so it works inside the Compose ``dev`` image.
Terraform/Ansible CLIs are optional — when present they are invoked too.
"""
from __future__ import annotations
import json
import os
import shutil
import subprocess
import sys
import tempfile
import urllib3
from pathlib import Path
import requests
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
BASE = os.environ.get("VSPHERE_BASE", "https://localhost").rstrip("/")
USER = os.environ.get("VSPHERE_USER", "administrator@vsphere.local")
PASSWORD = os.environ.get("VSPHERE_PASSWORD", "VMware1!")
ROOT = Path(__file__).resolve().parents[1]
def _session() -> dict[str, str]:
response = requests.post(f"{BASE}/api/session", auth=(USER, PASSWORD), verify=False, timeout=60)
response.raise_for_status()
return {"vmware-api-session-id": response.json()}
def run_python_lifecycle(headers: dict[str, str]) -> dict[str, str]:
created = requests.post(
f"{BASE}/api/vcenter/vm",
headers=headers,
json={
"name": "cookbook-py-01",
"guest_OS": "OTHER_GUEST_64",
"placement": {
"folder": "group-v23",
"host": "host-11",
"datastore": "datastore-31",
"resource_pool": "resgroup-22",
},
"cpu": {"count": 1},
"memory": {"size_MiB": 512},
},
verify=False,
timeout=60,
)
created.raise_for_status()
vm = created.json()
power = requests.post(
f"{BASE}/api/vcenter/vm/{vm}/power",
params={"action": "start"},
headers=headers,
verify=False,
timeout=60,
)
power.raise_for_status()
assert power.json().get("task")
# Platform surfaces previously deferred
providers = requests.get(
f"{BASE}/api/vcenter/identity/providers", headers=headers, verify=False, timeout=60
)
providers.raise_for_status()
assert any(p.get("type_id") in {"Oidc", "Saml", "LocalOS"} for p in providers.json())
nsx = requests.get(
f"{BASE}/api/vcenter/namespace-management/nsx-tier0-gateway",
headers=headers,
verify=False,
timeout=60,
)
nsx.raise_for_status()
assert nsx.json()
nfc = requests.post(
f"{BASE}/sdk",
data="""<?xml version="1.0"?>
<soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/">
<soapenv:Body>
<ImportVApp_Task xmlns="urn:vim25">
<_this type="Folder">group-v23</_this>
<name>nfc-import-lab</name>
</ImportVApp_Task>
</soapenv:Body>
</soapenv:Envelope>""",
headers={
**headers,
"Content-Type": "text/xml",
"Cookie": f'vmware_soap_session="{headers["vmware-api-session-id"]}"',
},
verify=False,
timeout=60,
)
nfc.raise_for_status()
assert "task-" in nfc.text and "ImportVApp_TaskResponse" in nfc.text
requests.post(
f"{BASE}/api/vcenter/vm/{vm}/power",
params={"action": "stop"},
headers=headers,
verify=False,
timeout=60,
).raise_for_status()
requests.delete(
f"{BASE}/api/vcenter/vm/{vm}", headers=headers, verify=False, timeout=60
).raise_for_status()
return {"python": "ok", "vm": str(vm)}
def run_ansible_uri(headers: dict[str, str]) -> dict[str, str]:
"""Mirror examples/ansible/vsphere_playbook.yml using the same REST calls."""
created = requests.post(
f"{BASE}/api/vcenter/vm",
headers=headers,
json={
"name": "cookbook-ansible-01",
"guest_OS": "OTHER_GUEST_64",
"placement": {
"folder": "group-v23",
"host": "host-11",
"datastore": "datastore-31",
"resource_pool": "resgroup-22",
},
"cpu": {"count": 1},
"memory": {"size_MiB": 512},
},
verify=False,
timeout=60,
)
created.raise_for_status()
vm = created.json()
for action in ("start", "stop"):
requests.post(
f"{BASE}/api/vcenter/vm/{vm}/power",
params={"action": action},
headers=headers,
verify=False,
timeout=60,
).raise_for_status()
requests.put(
f"{BASE}/api/vcenter/vm/{vm}/guest/filesystem",
params={"path": "/tmp/ansible-marker"},
headers=headers,
json={"content": "ansible-ok"},
verify=False,
timeout=60,
).raise_for_status()
requests.delete(
f"{BASE}/api/vcenter/vm/{vm}", headers=headers, verify=False, timeout=60
).raise_for_status()
result = {"ansible_uri": "ok", "vm": str(vm)}
playbook = ROOT / "examples" / "ansible" / "vsphere_playbook.yml"
if shutil.which("ansible-playbook") and playbook.is_file():
# Prefer simulator HTTP inside compose if BASE is internal.
env_base = BASE.replace("https://localhost", "https://localhost")
proc = subprocess.run(
[
"ansible-playbook",
"-i",
str(ROOT / "examples" / "ansible" / "inventory.ini"),
str(playbook),
"-e",
f"vsphere_base={env_base}",
"-e",
"vm_name=cookbook-ansible-cli-01",
],
check=False,
capture_output=True,
text=True,
timeout=180,
)
result["ansible_cli"] = "ok" if proc.returncode == 0 else f"failed:{proc.returncode}"
if proc.returncode != 0:
result["ansible_cli_stderr"] = (proc.stderr or proc.stdout)[-500:]
else:
result["ansible_cli"] = "skipped"
return result
def run_pulumi_style(headers: dict[str, str]) -> dict[str, str]:
"""Mirror examples/pulumi/__main__.py REST ComponentResource flow."""
created = requests.post(
f"{BASE}/api/vcenter/vm",
headers=headers,
json={
"name": "cookbook-pulumi-01",
"guest_OS": "OTHER_GUEST_64",
"placement": {
"folder": "group-v23",
"host": "host-11",
"datastore": "datastore-31",
"resource_pool": "resgroup-22",
},
"cpu": {"count": 1},
"memory": {"size_MiB": 512},
},
verify=False,
timeout=60,
)
created.raise_for_status()
vm = created.json()
power = requests.post(
f"{BASE}/api/vcenter/vm/{vm}/power",
params={"action": "start"},
headers=headers,
verify=False,
timeout=60,
)
power.raise_for_status()
detail = requests.get(f"{BASE}/api/vcenter/vm/{vm}", headers=headers, verify=False, timeout=60)
detail.raise_for_status()
requests.post(
f"{BASE}/api/vcenter/vm/{vm}/power",
params={"action": "stop"},
headers=headers,
verify=False,
timeout=60,
).raise_for_status()
requests.delete(
f"{BASE}/api/vcenter/vm/{vm}", headers=headers, verify=False, timeout=60
).raise_for_status()
result = {"pulumi_style": "ok", "vm": str(vm), "name": detail.json().get("name")}
if shutil.which("pulumi"):
result["pulumi_cli"] = "available"
else:
result["pulumi_cli"] = "skipped"
return result
def run_terraform_style(headers: dict[str, str]) -> dict[str, str]:
"""Validate the inventory lookups Terraform data sources need (SOAP+REST)."""
# REST inventory used by many TF plans as complementary checks
for path in (
"/api/vcenter/datacenter",
"/api/vcenter/cluster",
"/api/vcenter/datastore",
"/api/vcenter/network",
"/api/vcenter/vm?names=web-01",
):
response = requests.get(f"{BASE}{path}", headers=headers, verify=False, timeout=60)
response.raise_for_status()
assert response.json(), path
# SOAP FindByInventoryPath + CreateVM (resource path)
sid = headers["vmware-api-session-id"]
soap_headers = {
**headers,
"Content-Type": "text/xml",
"Cookie": f'vmware_soap_session="{sid}"',
}
find = requests.post(
f"{BASE}/sdk",
data="""<?xml version="1.0"?>
<soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/">
<soapenv:Body>
<FindByInventoryPath xmlns="urn:vim25">
<_this type="SearchIndex">SearchIndex</_this>
<inventoryPath>/Datacenters/Datacenter/vm/web-01</inventoryPath>
</FindByInventoryPath>
</soapenv:Body>
</soapenv:Envelope>""",
headers=soap_headers,
verify=False,
timeout=60,
)
find.raise_for_status()
assert "VirtualMachine" in find.text
create = requests.post(
f"{BASE}/sdk",
data="""<?xml version="1.0"?>
<soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/">
<soapenv:Body>
<CreateVM_Task xmlns="urn:vim25">
<_this type="Folder">group-v23</_this>
<config>
<name>cookbook-tf-01</name>
<guestId>otherGuest64</guestId>
<numCPUs>1</numCPUs>
<memoryMB>512</memoryMB>
<files><vmPathName>[datastore1]</vmPathName></files>
</config>
<pool type="ResourcePool">resgroup-22</pool>
<host type="HostSystem">host-11</host>
</CreateVM_Task>
</soapenv:Body>
</soapenv:Envelope>""",
headers=soap_headers,
verify=False,
timeout=60,
)
create.raise_for_status()
assert "task-" in create.text
result = {"terraform_style": "ok"}
tf_dir = ROOT / "examples" / "terraform" / "vsphere"
tf_bin = shutil.which("terraform") or (
str(ROOT / ".tools" / "terraform") if (ROOT / ".tools" / "terraform").is_file() else None
)
if tf_bin and tf_dir.is_dir():
server = BASE.replace("https://", "").replace("http://", "")
# Prefer the checked-out example (keeps .terraform providers) when writable.
work = tf_dir if (tf_dir / ".terraform").is_dir() else None
tmp_ctx = None
if work is None:
tmp_ctx = tempfile.TemporaryDirectory()
work = Path(tmp_ctx.name)
for name in ("main.tf", "variables.tf"):
(work / name).write_text(
(tf_dir / name).read_text(encoding="utf-8"), encoding="utf-8"
)
env = {
**os.environ,
"TF_VAR_vsphere_server": server,
"TF_VAR_vsphere_user": USER,
"TF_VAR_vsphere_password": PASSWORD,
"TF_VAR_create_lab_vm": "false",
}
try:
if not (work / ".terraform").is_dir():
init = subprocess.run(
[tf_bin, "init", "-input=false", "-no-color"],
cwd=work,
env=env,
capture_output=True,
text=True,
timeout=180,
check=False,
)
if init.returncode != 0:
result["terraform_cli"] = f"init_failed:{(init.stderr or '')[-300:]}"
return result
plan = subprocess.run(
[tf_bin, "plan", "-input=false", "-no-color", "-detailed-exitcode"],
cwd=work,
env=env,
capture_output=True,
text=True,
timeout=180,
check=False,
)
# 0 = no changes, 2 = changes present — both OK for data sources
result["terraform_cli"] = (
"ok" if plan.returncode in {0, 2} else f"plan_failed:{plan.returncode}"
)
if plan.returncode not in {0, 2}:
result["terraform_cli_stderr"] = ((plan.stderr or "") + (plan.stdout or ""))[-500:]
finally:
if tmp_ctx is not None:
tmp_ctx.cleanup()
else:
result["terraform_cli"] = "skipped"
return result
def main() -> int:
headers = _session()
report: dict[str, object] = {"base": BASE}
failed = False
for name, fn in (
("python", run_python_lifecycle),
("ansible", run_ansible_uri),
("pulumi", run_pulumi_style),
("terraform", run_terraform_style),
):
try:
report[name] = fn(headers)
except Exception as error: # noqa: BLE001
report[name] = {"error": str(error)}
failed = True
print(json.dumps(report, indent=2))
return 1 if failed else 0
if __name__ == "__main__":
raise SystemExit(main())
scripts/run_vsphere_full_green.sh
Executable
+34
View File
@@ -0,0 +1,34 @@
#!/usr/bin/env bash
set -euo pipefail
cd "$(dirname "$0")/.."
export PYTHONPATH=/workspace
export VSPHERE_BASE="${VSPHERE_BASE:-http://simulator:8080}"
echo "== pytest =="
python -m pytest \
tests/unit/test_vsphere_universe.py \
tests/unit/test_vsphere_matrix.py \
tests/unit/test_vsphere_compatibility.py \
tests/unit/test_vsphere_catalog.py \
tests/unit/test_vsphere_profiles.py \
tests/unit/test_vsphere_mappers.py \
tests/unit/test_property_collector.py \
tests/unit/test_web_assets.py \
tests/unit/test_web_console.py \
tests/integration/test_vsphere_api_surface_data.py \
tests/integration/test_vsphere_api.py \
tests/integration/test_vsphere_soap_depth.py \
tests/integration/test_vsphere_full_api.py \
-q
echo "== surface =="
python scripts/vsphere_surface_probe.py
echo "== matrix =="
python scripts/vsphere_full_matrix_probe.py
echo "== real-data spotcheck =="
python scripts/vsphere_real_data_spotcheck.py
echo "ALL GREEN"
scripts/vsphere_full_matrix_probe.py
+439
View File
@@ -0,0 +1,439 @@
#!/usr/bin/env python3
"""Probe every registered vSphere REST method for majors 6–9 (GET/POST/PATCH/PUT/DELETE).
Acceptable statuses: 2xx, 400/404/405/409/422 (validation / missing id).
Fail on: 5xx, unexpected exceptions, empty inventory on seeded GETs.
Lab policy: catalog floors are browse-only — runtime never expects HTTP 501.
"""
from __future__ import annotations
import argparse
import json
import os
import secrets
import ssl
import sys
import urllib.error
import urllib.request
from base64 import b64encode
from collections import Counter
from typing import Any
from urllib.parse import urlencode
from app.vsphere.contracts.matrix import VERSIONS, catalog_entries_for_major, methods_for_major
from app.vsphere.rest.coverage import IMPLEMENTED
BASE = os.getenv("VSPHERE_BASE", "https://localhost")
USER = os.getenv("VSPHERE_USER", "administrator@vsphere.local")
PASSWORD = os.getenv("VSPHERE_PASSWORD", "VMware1!")
_PATH_SUBS = {
"{vm}": "vm-101",
"{host}": "host-11",
"{datastore}": "datastore-31",
"{task}": "task-1",
"{snapshot}": "snapshot-missing",
"{category_id}": "cat-lab-1",
"{tag_id}": "tag-lab-1",
"{item_id}": "item-ubuntu",
"{library_id}": "lib-local-1",
"{folder}": "group-v23",
"{datacenter}": "datacenter-21",
"{cluster}": "domain-c21",
"{resource_pool}": "resgroup-22",
"{permission_id}": "999999",
"{policy}": "policy-default",
"{disk}": "2000",
"{nic}": "4000",
"{cdrom}": "3000",
"{floppy}": "8000",
"{port}": "9000",
"{adapter}": "1000",
"{provider}": "vsphere.local",
"{supervisor}": "supervisor-1",
"{namespace}": "ns-lab-1",
"{role}": "ReadOnly",
"{zone}": "zone-1",
"{project}": "project-1",
"{domain}": "lab.local",
"{service}": "vsphere-ui",
"{depot}": "depot-1",
"{component}": "component-1",
"{image}": "image-1",
"{draft}": "draft-1",
"{connection}": "connection-1",
"{vpc}": "vpc-1",
"{subnet}": "subnet-1",
"{session_id}": "session-lab-1",
"{download_session_id}": "session-lab-1",
"{update_session_id}": "session-lab-1",
"{subscription_id}": "sub-1",
"{usage_id}": "usage-1",
"{version}": "1",
"{chain}": "chain-1",
"{node}": "node-1",
"{profile}": "profile-1",
"{interface}": "nic0",
"{core}": "core-1",
"{network}": "network-41",
"{commit}": "commit-lab-1",
}
_ACCEPT_CLIENT = {400, 401, 403, 404, 405, 409, 412, 422}
def _ctx() -> ssl.SSLContext | None:
if not BASE.startswith("https://"):
return None
return ssl._create_unverified_context() # noqa: S323
def _concrete(path: str) -> str:
import re
out = path
for key, value in _PATH_SUBS.items():
out = out.replace(key, value)
# Any remaining {param} tokens from the Broadcom universe.
return re.sub(r"\{([A-Za-z0-9_]+)\}", r"probe-\1", out)
def _request(
method: str,
path: str,
*,
headers: dict[str, str],
data: bytes | None = None,
) -> tuple[int, str]:
url = f"{BASE}{_concrete(path)}"
req = urllib.request.Request(url, data=data, method=method, headers=headers)
try:
with urllib.request.urlopen(req, context=_ctx()) as resp: # noqa: S310
body = resp.read().decode("utf-8", errors="replace")
return int(resp.status), body
except urllib.error.HTTPError as error:
body = error.read().decode("utf-8", errors="replace")
return int(error.code), body
def _login() -> 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}:
raise SystemExit(f"session failed: {code} {body[:200]}")
return json.loads(body)
def _payload_for(verb: str, path: str) -> tuple[str, bytes | None]:
"""Return (url_suffix_or_path, body_bytes). Path may gain query string."""
if verb not in {"POST", "PUT", "PATCH"}:
return path, None
if path.endswith("/power") and verb == "POST":
if "/guest/power" in path:
return f"{path}?action=reboot", b"{}"
return f"{path}?action=start", b"{}"
if path.endswith("/maintenance") and verb == "POST":
return f"{path}?action=enter", b"{}"
if path.endswith("/folder/{folder}") and verb == "POST":
return f"{path}?action=rename", json.dumps({"name": "folder-renamed-probe"}).encode()
suffix = secrets.token_hex(4)
bodies: dict[str, dict[str, Any]] = {
"/api/vcenter/vm": {
"name": f"matrix-probe-vm-{suffix}",
"placement": {"folder": "group-v23", "host": "host-11", "datastore": "datastore-31"},
"cpu_count": 1,
"memory_size_MiB": 512,
},
"/api/vcenter/datacenter": {"name": f"probe-dc-{suffix}"},
"/api/vcenter/cluster": {"name": f"probe-cluster-{suffix}"},
"/api/vcenter/folder": {"name": f"probe-folder-{suffix}", "parent": "group-v23"},
"/api/vcenter/resource-pool": {"name": f"probe-rp-{suffix}", "parent": "resgroup-22"},
"/api/vcenter/network/dvs": {"name": f"probe-dvs-{suffix}"},
"/api/vcenter/network/dvpg": {
"name": f"probe-dvpg-{suffix}",
"dvs": "dvs-51",
"vlan_id": 10,
},
"/api/cis/tagging/category": {
"create_spec": {
"name": f"probe-cat-{suffix}",
"description": "probe",
"cardinality": "MULTIPLE",
"associable_types": [],
}
},
"/api/cis/tagging/tag": {
"create_spec": {
"name": f"probe-tag-{suffix}",
"category_id": "missing-category",
"description": "x",
}
},
"/api/cis/tagging/tag-association": {
"action": "list-attached-tags",
"tag_id": "x",
"object_id": {"type": "VirtualMachine", "id": "vm-101"},
},
"/api/content/local-library": {"create_spec": {"name": f"probe-lib-{suffix}"}},
"/api/content/library/item": {
"create_spec": {
"library_id": "lib-missing",
"name": f"probe-item-{suffix}",
"type": "ovf",
}
},
"/api/vcenter/ovf/library-item/{item_id}": {
"deployment_spec": {"name": f"ovf-probe-{suffix}"},
"target": {"folder": "group-v23", "host": "host-11", "datastore": "datastore-31"},
},
"/api/vcenter/authorization/permissions": {
"principal": "readonly@vsphere.local",
"role": "ReadOnly",
"entity": "datacenter-21",
},
"/api/vcenter/datastore/{datastore}/files": {
"path": f"/probe-{suffix}.txt",
"size": 1,
"type": "FILE",
},
"/api/vcenter/vm/{vm}/hardware/cpu": {"count": 2},
"/api/vcenter/vm/{vm}/hardware/memory": {"size_MiB": 1024},
"/api/vcenter/vm/{vm}/hardware/disk": {"type": "SCSI", "new_vmdk": {"capacity": 1024}},
"/api/vcenter/vm/{vm}/hardware/ethernet": {
"type": "VMXNET3",
"backing": {"type": "STANDARD_PORTGROUP", "network": "network-41"},
},
"/api/vcenter/vm/{vm}/snapshots": {"name": f"probe-snap-{suffix}"},
"/api/vcenter/vm/{vm}/snapshots/{snapshot}": {"action": "revert"},
"/api/vcenter/vm/{vm}/clone": {
"name": f"probe-clone-{suffix}",
"placement": {"folder": "group-v23", "host": "host-11"},
},
"/api/vcenter/vm/{vm}/relocate": {"placement": {"host": "host-12"}},
"/api/vcenter/vm/{vm}/tools": {"action": "upgrade"},
"/api/vcenter/vm/{vm}/console/tickets": {"type": "WEBMKS"},
"/api/vcenter/vm/{vm}/guest/customization": {"name": {"name": f"guest-probe-{suffix}"}},
"/api/vcenter/vm/{vm}": {"action": "unregister"},
}
body = bodies.get(path, {})
return path, json.dumps(body).encode()
def _apply_major(major: int, headers: dict[str, str]) -> dict[str, Any]:
params = urlencode({"major": major})
code, body = _request(
"POST",
f"/ui/api/contract/apply?{params}",
headers=headers,
)
if code >= 400:
raise SystemExit(f"contract apply major={major} failed: {code} {body[:200]}")
return json.loads(body)
def probe_major(major: int, session: str) -> dict[str, Any]:
headers = {
"vmware-api-session-id": session,
"Content-Type": "application/json",
"Accept": "application/json",
}
applied = _apply_major(major, headers)
active = methods_for_major(major)
verb_order = {"GET": 0, "PUT": 1, "PATCH": 2, "POST": 3, "DELETE": 4}
entries = sorted(
catalog_entries_for_major(major),
key=lambda e: (verb_order.get(e["verb"], 9), e["path"]),
)
buckets: Counter[str] = Counter()
failures: list[dict[str, Any]] = []
probed = 0
for entry in entries:
verb = entry["verb"]
path = entry["path"]
# Don't tear down the probe session mid-run.
if verb == "DELETE" and path in {"/api/session", "/rest/com/vmware/cis/session"}:
continue
# Don't destroy seeded datacenter/cluster/folder parents.
if verb == "DELETE" and path in {
"/api/vcenter/datacenter/{datacenter}",
"/api/vcenter/cluster/{cluster}",
"/api/vcenter/folder/{folder}",
"/api/vcenter/resource-pool/{resource_pool}",
"/api/vcenter/vm/{vm}",
}:
# Still hit the route, but against missing id → expect 4xx.
if path.endswith("{vm}"):
url_path = path.replace("{vm}", "vm-missing-matrix")
elif path.endswith("{datacenter}"):
url_path = path.replace("{datacenter}", "dc-missing")
elif path.endswith("{cluster}"):
url_path = path.replace("{cluster}", "cluster-missing")
elif path.endswith("{folder}"):
url_path = path.replace("{folder}", "folder-missing")
else:
url_path = path.replace("{resource_pool}", "rp-missing")
code, body = _request(verb, url_path, headers=headers)
else:
url_path, data = _payload_for(verb, path)
if verb == "GET" and path == "/api/content/library/item":
url_path = f"{url_path}?library_id=lib-local-1"
code, body = _request(verb, url_path, headers=headers, data=data)
probed += 1
if 200 <= code < 300:
buckets["success_2xx"] += 1
# Major 9 must return real seeded payloads, not synthetic stub markers.
if major == 9 and verb == "GET" and body:
if '"stub": true' in body or '"stub":true' in body:
buckets["stub_marker"] += 1
failures.append(
{
"major": major,
"verb": verb,
"path": path,
"status": code,
"body": body[:200],
"expected": "non-stub JSON from DB/inventory",
}
)
elif path in {
"/api/vcenter/vm",
"/api/vcenter/host",
"/api/vcenter/datastore",
"/api/vcenter/network",
"/api/vcenter/cluster",
"/api/cis/tagging/category",
"/api/content/library",
"/api/esx/settings/clusters/{cluster}/software",
"/api/vcenter/namespace-management/supervisors/{supervisor}/summary",
"/api/appliance/access/ssh",
"/api/appliance/services",
}:
try:
parsed = json.loads(body)
except json.JSONDecodeError:
parsed = None
empty = parsed in ([], {}, None) or parsed == ""
if empty or (isinstance(parsed, dict) and parsed == {}):
buckets["empty_inventory"] += 1
failures.append(
{
"major": major,
"verb": verb,
"path": path,
"status": code,
"body": body[:200],
"expected": "non-empty seeded data",
}
)
elif code in _ACCEPT_CLIENT:
buckets["client_4xx"] += 1
elif code == 501:
buckets["unexpected_501"] += 1
failures.append(
{"major": major, "verb": verb, "path": path, "status": code, "body": body[:200]}
)
elif code >= 500:
buckets["server_5xx"] += 1
failures.append(
{"major": major, "verb": verb, "path": path, "status": code, "body": body[:200]}
)
else:
buckets[f"other_{code}"] += 1
failures.append(
{"major": major, "verb": verb, "path": path, "status": code, "body": body[:200]}
)
# Paths above this major's catalog floor still must serve real data (no 501).
above_floor = 0
for (verb, path), _status in sorted(IMPLEMENTED.items()):
if (verb, path) in active:
continue
if verb == "DELETE" and path in {"/api/session", "/rest/com/vmware/cis/session"}:
continue
url_path, data = _payload_for(verb, path)
code, body = _request(verb, url_path, headers=headers, data=data)
above_floor += 1
probed += 1
if code == 501:
buckets["unexpected_501"] += 1
failures.append(
{
"major": major,
"verb": verb,
"path": path,
"status": code,
"body": body[:200],
"expected": "2xx/4xx (version gate disabled)",
}
)
elif 200 <= code < 300:
buckets["success_2xx"] += 1
elif code in _ACCEPT_CLIENT:
buckets["client_4xx"] += 1
elif code >= 500:
buckets["server_5xx"] += 1
failures.append(
{"major": major, "verb": verb, "path": path, "status": code, "body": body[:200]}
)
else:
buckets[f"other_{code}"] += 1
failures.append(
{"major": major, "verb": verb, "path": path, "status": code, "body": body[:200]}
)
by_verb = Counter(e["verb"] for e in entries)
return {
"major": major,
"version": applied.get("runtime_version"),
"method_count": len(entries),
"by_verb": dict(by_verb),
"probed": probed,
"above_floor_checked": above_floor,
"buckets": dict(buckets),
"failures": failures,
}
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("--majors", default="6,7,8,9", help="Comma-separated majors")
args = parser.parse_args()
majors = [int(x) for x in args.majors.split(",") if x.strip()]
for major in majors:
if major not in VERSIONS:
raise SystemExit(f"unknown major {major}")
session = _login()
reports = []
all_failures: list[dict[str, Any]] = []
for major in majors:
report = probe_major(major, session)
reports.append(report)
all_failures.extend(report["failures"])
# Refresh session between majors (logout delete skipped during probe).
session = _login()
# Restore latest floor for the lab UI.
_apply_major(9, {"vmware-api-session-id": session, "Content-Type": "application/json"})
summary = {
"base": BASE,
"majors": reports,
"total_failures": len(all_failures),
"failures": all_failures[:80],
}
print(json.dumps(summary, indent=2))
return 1 if all_failures else 0
if __name__ == "__main__":
raise SystemExit(main())
scripts/vsphere_nonempty_probe.py
+163
View File
@@ -0,0 +1,163 @@
#!/usr/bin/env python3
"""Fail if any registered GET returns 501, stub marker, or empty JSON body."""
from __future__ import annotations
import json
import os
import re
import ssl
import sys
import urllib.error
import urllib.request
from base64 import b64encode
from app.vsphere.rest.coverage import IMPLEMENTED
BASE = os.getenv("VSPHERE_BASE", "https://localhost")
USER = os.getenv("VSPHERE_USER", "administrator@vsphere.local")
PASSWORD = os.getenv("VSPHERE_PASSWORD", "VMware1!")
_SUBS = {
"{vm}": "vm-101",
"{host}": "host-11",
"{datastore}": "datastore-31",
"{task}": "task-1",
"{snapshot}": "snapshot-missing",
"{category_id}": "cat-lab-1",
"{tag_id}": "tag-lab-1",
"{item_id}": "item-ubuntu",
"{library_id}": "lib-local-1",
"{folder}": "group-v23",
"{datacenter}": "datacenter-21",
"{cluster}": "domain-c21",
"{resource_pool}": "resgroup-22",
"{permission_id}": "1",
"{policy}": "policy-default",
"{disk}": "2000",
"{nic}": "4000",
"{cdrom}": "3000",
"{adapter}": "1000",
"{network}": "network-41",
"{supervisor}": "supervisor-1",
"{commit}": "commit-lab-1",
"{domain}": "lab.local",
"{interface}": "nic0",
"{service}": "vpxd",
"{session_id}": "session-lab-1",
"{download_session_id}": "session-lab-1",
"{update_session_id}": "session-lab-1",
"{provider}": "vsphere.local",
"{role}": "ReadOnly",
"{chain}": "chain-1",
"{floppy}": "8000",
"{port}": "9000",
}
def _ctx() -> ssl.SSLContext | None:
if not BASE.startswith("https://"):
return None
return ssl._create_unverified_context() # noqa: S323
def _concrete(path: str) -> str:
out = path
for key, value in _SUBS.items():
out = out.replace(key, value)
return re.sub(r"\{([A-Za-z0-9_]+)\}", r"lab-\1", out)
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(), timeout=60) 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
checked = 0
for (verb, path), _status in sorted(IMPLEMENTED.items()):
if verb != "GET" or path in {"/api/session", "/rest/com/vmware/cis/session"}:
continue
checked += 1
concrete = _concrete(path)
if path == "/api/content/library/item":
concrete = f"{concrete}?library_id=lib-local-1"
status, raw = _request("GET", concrete, headers=headers)
if status == 501:
failures.append({"path": path, "status": status, "reason": "version gate 501"})
continue
if status >= 500:
failures.append(
{"path": path, "status": status, "reason": "server error", "body": raw[:160]}
)
continue
if status not in {200, 201}:
# Missing probe ids may 404 — still require a JSON error body.
if not raw.strip():
failures.append({"path": path, "status": status, "reason": "empty error body"})
continue
if not raw.strip():
failures.append({"path": path, "status": status, "reason": "empty body"})
continue
if '"stub": true' in raw or '"stub":true' in raw:
failures.append({"path": path, "status": status, "reason": "stub marker"})
continue
try:
payload = json.loads(raw)
except json.JSONDecodeError:
failures.append(
{"path": path, "status": status, "reason": "non-json", "body": raw[:160]}
)
continue
if payload in ([], {}, None, "") or (
isinstance(payload, (list, dict)) and len(payload) == 0
):
failures.append(
{"path": path, "status": status, "reason": "empty json", "body": raw[:160]}
)
continue
if isinstance(payload, dict):
for key in ("data", "value", "messages", "items", "results"):
if key in payload and payload[key] in ([], None, {}):
failures.append(
{
"path": path,
"status": status,
"reason": f"empty nested {key}",
"body": raw[:160],
}
)
break
else:
ok += 1
continue
ok += 1
print(
json.dumps(
{
"checked": checked,
"ok": ok,
"failure_count": len(failures),
"failures": failures[:80],
},
indent=2,
)
)
return 1 if failures else 0
if __name__ == "__main__":
raise SystemExit(main())
scripts/vsphere_real_data_spotcheck.py
+102
View File
@@ -0,0 +1,102 @@
#!/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())
scripts/vsphere_surface_probe.py
+170
View File
@@ -0,0 +1,170 @@
#!/usr/bin/env python3
"""Probe every implemented REST path from the coverage registry."""
from __future__ import annotations
import json
import os
import ssl
import sys
import urllib.error
import urllib.request
from base64 import b64encode
from app.vsphere.rest.coverage import catalog_entries
BASE = os.getenv("VSPHERE_BASE", "https://localhost")
USER = "administrator@vsphere.local"
PASSWORD = "VMware1!"
def _ctx() -> ssl.SSLContext | None:
if not BASE.startswith("https://"):
return None
return ssl._create_unverified_context() # noqa: S323
def _concrete(path: str) -> str:
import re
subs = {
"{vm}": "vm-101",
"{host}": "host-11",
"{datastore}": "datastore-31",
"{task}": "task-missing",
"{snapshot}": "snapshot-missing",
"{category_id}": "missing",
"{tag_id}": "missing",
"{item_id}": "missing",
"{library_id}": "lib-missing",
"{folder}": "group-v23",
"{datacenter}": "datacenter-21",
"{cluster}": "domain-c21",
"{resource_pool}": "resgroup-22",
"{permission_id}": "1",
"{policy}": "policy-default",
"{disk}": "2000",
"{nic}": "4000",
"{cdrom}": "3000",
"{adapter}": "1000",
"{network}": "network-41",
}
out = path
for key, value in subs.items():
out = out.replace(key, value)
return re.sub(r"\{([A-Za-z0-9_]+)\}", r"probe-\1", out)
def _request(method: str, path: str, *, headers: dict[str, str], data: bytes | None = None) -> int:
concrete = _concrete(path)
req = urllib.request.Request(f"{BASE}{concrete}", data=data, method=method, headers=headers)
try:
with urllib.request.urlopen(req, context=_ctx()) as resp: # noqa: S310
return int(resp.status)
except urllib.error.HTTPError as error:
return int(error.code)
def main() -> int:
basic = b64encode(f"{USER}:{PASSWORD}".encode()).decode()
status = _request("POST", "/api/session", headers={"Authorization": f"Basic {basic}"})
if status not in {200, 201}:
print(f"session failed: {status}", file=sys.stderr)
return 1
# Re-login to capture body
req = urllib.request.Request(
f"{BASE}/api/session",
method="POST",
headers={"Authorization": f"Basic {basic}"},
)
with urllib.request.urlopen(req, context=_ctx()) as resp: # noqa: S310
session = json.loads(resp.read().decode())
headers = {"vmware-api-session-id": session, "Content-Type": "application/json"}
failures: list[str] = []
probed = 0
for entry in catalog_entries():
verb = entry["verb"]
path = entry["path"]
if verb == "DELETE" and path in {"/api/session", "/rest/com/vmware/cis/session"}:
continue
if (
"{" in path
and verb in {"POST", "PATCH", "DELETE"}
and "missing" in (path.replace("{vm}", "vm-101"))
):
# skip destructive ops on missing ids except GET
pass
data = b"{}" if verb in {"POST", "PUT", "PATCH"} else None
if path.endswith("/power") and verb == "POST":
code = _request(verb, path + "?action=start", headers=headers)
elif "tag-association" in path and verb == "POST":
data = json.dumps(
{
"action": "list-attached-tags",
"tag_id": "x",
"object_id": {"type": "VirtualMachine", "id": "vm-101"},
}
).encode()
code = _request(verb, path, headers=headers, data=data)
else:
code = _request(verb, path, headers=headers, data=data)
probed += 1
# Accept success, not-found for missing substitutions, or validation errors.
if code >= 500:
failures.append(f"{verb} {path} -> {code}")
continue
if verb == "GET" and 200 <= code < 300:
# Surface probe reads body via a second request-sized check only for markers.
# Re-fetch is avoided: empty GET bodies for session are OK.
pass
# Re-auth in case any probe request invalidated the session cookie.
req = urllib.request.Request(
f"{BASE}/api/session",
method="POST",
headers={"Authorization": f"Basic {basic}"},
)
with urllib.request.urlopen(req, context=_ctx()) as resp: # noqa: S310
session = json.loads(resp.read().decode())
headers = {"vmware-api-session-id": session, "Content-Type": "application/json"}
# Spot-check critical inventory payloads are non-empty / non-stub.
spot = [
"/api/vcenter/vm",
"/api/vcenter/host",
"/api/content/library",
"/api/cis/tagging/category",
"/api/esx/settings/clusters/domain-c21/software",
"/api/vcenter/namespace-management/supervisors/supervisor-1/summary",
"/api/appliance/access/ssh",
"/api/appliance/services",
"/api/vcenter/vm/vm-101/hardware/cdrom",
]
for path in spot:
url = f"{BASE}{_concrete(path)}"
req = urllib.request.Request(url, method="GET", headers=headers)
try:
with urllib.request.urlopen(req, context=_ctx()) as resp: # noqa: S310
body = resp.read().decode("utf-8", errors="replace")
code = int(resp.status)
except urllib.error.HTTPError as error:
body = error.read().decode("utf-8", errors="replace")
code = int(error.code)
if code >= 400:
failures.append(f"GET {path} spot -> {code}")
continue
if '"stub": true' in body or '"stub":true' in body:
failures.append(f"GET {path} spot -> stub marker")
continue
try:
parsed = json.loads(body)
except json.JSONDecodeError:
failures.append(f"GET {path} spot -> non-json")
continue
if parsed in ([], {}, None):
failures.append(f"GET {path} spot -> empty")
print(json.dumps({"probed": probed, "failures": failures}, indent=2))
return 1 if failures else 0
if __name__ == "__main__":
raise SystemExit(main())
scripts/write_vsphere_bundles.py
+17
View File
@@ -0,0 +1,17 @@
#!/usr/bin/env python3
"""Regenerate contracts/vsphere/*/manifest.json stub OpenAPI matrices."""
from __future__ import annotations
from app.vsphere.contracts.matrix import write_bundles
def main() -> int:
written = write_bundles()
for path in written:
print(path)
return 0
if __name__ == "__main__":
raise SystemExit(main())
scripts/write_vsphere_evidence.py
+30
View File
@@ -0,0 +1,30 @@
#!/usr/bin/env python3
"""Regenerate evidence/vsphere-*.json ledgers from the live coverage matrix."""
from __future__ import annotations
import json
from pathlib import Path
from app.vsphere.contracts.compatibility import evidence_ledger
from app.vsphere.contracts.matrix import VERSIONS
ROOT = Path(__file__).resolve().parents[1] / "evidence"
def main() -> int:
ROOT.mkdir(parents=True, exist_ok=True)
for major, meta in VERSIONS.items():
payload = evidence_ledger(major)
path = ROOT / f"vsphere-{meta['version']}.json"
path.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8")
summary = payload["summary"]
print(
f"{path} implemented={summary['implemented_methods']}/"
f"{summary['universe_methods']} coverage={summary['coverage']}"
)
return 0
if __name__ == "__main__":
raise SystemExit(main())