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.
451 lines
17 KiB
Python
451 lines
17 KiB
Python
"""Full vSphere REST + SOAP + major-matrix surface tests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import re
|
|
|
|
import pytest
|
|
from httpx import ASGITransport, AsyncClient
|
|
|
|
from app.config import Settings
|
|
from app.main import create_app
|
|
from app.vsphere.contracts.matrix import VERSIONS, catalog_entries_for_major
|
|
from app.vsphere.rest.coverage import IMPLEMENTED, catalog_entries
|
|
from app.vsphere.seed import seed_vsphere_inventory
|
|
|
|
pytestmark = pytest.mark.integration
|
|
|
|
_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",
|
|
"{session_id}": "session-lab-1",
|
|
"{folder}": "group-v23",
|
|
"{datacenter}": "datacenter-21",
|
|
"{cluster}": "domain-c21",
|
|
"{resource_pool}": "resgroup-22",
|
|
"{permission_id}": "1",
|
|
"{policy}": "policy-default",
|
|
"{supervisor}": "supervisor-1",
|
|
"{namespace}": "ns-lab-1",
|
|
"{provider}": "vsphere.local",
|
|
"{interface}": "nic0",
|
|
"{network}": "network-41",
|
|
"{cdrom}": "3000",
|
|
"{disk}": "2000",
|
|
"{nic}": "4000",
|
|
"{adapter}": "1000",
|
|
"{service}": "vsphere-ui",
|
|
"{domain}": "lab.local",
|
|
}
|
|
|
|
|
|
def _concrete(path: str) -> str:
|
|
out = path
|
|
for key, value in _PATH_SUBS.items():
|
|
out = out.replace(key, value)
|
|
return out
|
|
|
|
|
|
@pytest.fixture
|
|
async def client() -> AsyncClient:
|
|
database_url = os.getenv("TEST_DATABASE_URL") or os.getenv("DATABASE_URL")
|
|
if not database_url:
|
|
pytest.skip("TEST_DATABASE_URL / DATABASE_URL required")
|
|
settings = Settings(
|
|
database_url=database_url, # type: ignore[arg-type]
|
|
contract_snapshot=None,
|
|
enable_pve_stub=False,
|
|
)
|
|
app = create_app(settings=settings, worker_factories=())
|
|
async with app.router.lifespan_context(app):
|
|
await seed_vsphere_inventory(app.state.database, force=True, profile="small")
|
|
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as http:
|
|
yield http
|
|
|
|
|
|
async def _session(
|
|
client: AsyncClient, user: str = "administrator@vsphere.local"
|
|
) -> dict[str, str]:
|
|
login = await client.post("/api/session", auth=(user, "VMware1!"))
|
|
assert login.status_code == 201, login.text
|
|
sid = login.json()
|
|
assert isinstance(sid, str) and sid
|
|
return {"vmware-api-session-id": sid}
|
|
|
|
|
|
@pytest.mark.parametrize("major", sorted(VERSIONS))
|
|
async def test_ui_catalog_and_method_fields_for_every_major(
|
|
client: AsyncClient, major: int
|
|
) -> None:
|
|
catalog = await client.get("/ui/api/catalog", params={"major": major})
|
|
assert catalog.status_code == 200
|
|
body = catalog.json()
|
|
assert body["plane"] == "vsphere-rest"
|
|
assert body["method_count"] == len(catalog_entries_for_major(major))
|
|
# Spot-check a path with params on majors that include VM get.
|
|
method = await client.get(
|
|
"/ui/api/method",
|
|
params={"major": major, "path": "/api/vcenter/vm/{vm}", "verb": "GET"},
|
|
)
|
|
assert method.status_code == 200
|
|
payload = method.json()
|
|
if payload.get("implemented"):
|
|
assert any(f["name"] == "vm" for f in payload["path_fields"])
|
|
|
|
|
|
async def test_all_coverage_routes_no_server_error(client: AsyncClient) -> None:
|
|
headers = await _session(client)
|
|
failures: list[str] = []
|
|
for entry in catalog_entries():
|
|
verb = entry["verb"]
|
|
path = entry["path"]
|
|
if verb == "DELETE" and path == "/api/session":
|
|
continue
|
|
url = _concrete(path)
|
|
kwargs: dict = {"headers": headers}
|
|
if verb in {"POST", "PATCH", "PUT"}:
|
|
kwargs["headers"] = {**headers, "Content-Type": "application/json"}
|
|
if path.endswith("/power"):
|
|
url = f"{url}?action=start"
|
|
kwargs["json"] = {}
|
|
elif "tag-association" in path:
|
|
kwargs["json"] = {
|
|
"action": "list-attached-tags",
|
|
"tag_id": "x",
|
|
"object_id": {"type": "VirtualMachine", "id": "vm-101"},
|
|
}
|
|
else:
|
|
kwargs["json"] = {}
|
|
response = await client.request(verb, url, **kwargs)
|
|
if response.status_code >= 500:
|
|
failures.append(f"{verb} {path} -> {response.status_code} {response.text[:160]}")
|
|
assert failures == [], "\n".join(failures)
|
|
|
|
|
|
async def test_rest_inventory_returns_seed_data(client: AsyncClient) -> None:
|
|
headers = await _session(client)
|
|
for path, min_count in (
|
|
("/api/vcenter/vm", 5),
|
|
("/api/vcenter/host", 3),
|
|
("/api/vcenter/datastore", 1),
|
|
("/api/vcenter/network", 1),
|
|
("/api/vcenter/datacenter", 1),
|
|
("/api/vcenter/cluster", 1),
|
|
("/api/vcenter/folder", 1),
|
|
):
|
|
response = await client.get(path, headers=headers)
|
|
assert response.status_code == 200, path
|
|
assert len(response.json()) >= min_count, path
|
|
|
|
|
|
async def test_legacy_rest_wrappers(client: AsyncClient) -> None:
|
|
headers = await _session(client)
|
|
for path in (
|
|
"/rest/vcenter/vm",
|
|
"/rest/vcenter/host",
|
|
"/rest/vcenter/datastore",
|
|
"/rest/vcenter/network",
|
|
"/rest/vcenter/datacenter",
|
|
"/rest/vcenter/cluster",
|
|
"/rest/appliance/system/version",
|
|
):
|
|
response = await client.get(path, headers=headers)
|
|
assert response.status_code == 200, path
|
|
body = response.json()
|
|
assert "value" in body
|
|
|
|
|
|
async def test_session_contracts(client: AsyncClient) -> None:
|
|
headers = await _session(client)
|
|
get_session = await client.get("/api/session", headers=headers)
|
|
assert get_session.status_code == 200
|
|
assert get_session.content in (b"", b"null") or not get_session.text.strip()
|
|
assert "Administrator" in (get_session.headers.get("x-vmware-session-roles") or "")
|
|
|
|
legacy = await client.post(
|
|
"/rest/com/vmware/cis/session",
|
|
auth=("administrator@vsphere.local", "VMware1!"),
|
|
)
|
|
assert legacy.status_code in {200, 201}
|
|
assert legacy.json()["value"]
|
|
legacy_get = await client.get(
|
|
"/rest/com/vmware/cis/session",
|
|
headers={"vmware-api-session-id": legacy.json()["value"]},
|
|
)
|
|
assert legacy_get.status_code == 200
|
|
assert legacy_get.json()["value"]
|
|
|
|
|
|
async def test_authz_readonly_forbidden_on_power(client: AsyncClient) -> None:
|
|
headers = await _session(client, "readonly@vsphere.local")
|
|
vms = await client.get("/api/vcenter/vm", headers=headers)
|
|
assert vms.status_code == 200
|
|
vm = vms.json()[0]["vm"]
|
|
power = await client.post(
|
|
f"/api/vcenter/vm/{vm}/power",
|
|
params={"action": "start"},
|
|
headers=headers,
|
|
)
|
|
assert power.status_code == 403
|
|
|
|
|
|
async def test_appliance_version_public_and_health(client: AsyncClient) -> None:
|
|
version = await client.get("/api/appliance/system/version")
|
|
assert version.status_code == 200
|
|
assert version.json()["version"]
|
|
headers = await _session(client)
|
|
health = await client.get("/api/appliance/health/system", headers=headers)
|
|
assert health.status_code == 200
|
|
assert health.json()["status"] == "green"
|
|
networking = await client.get("/api/appliance/networking", headers=headers)
|
|
assert networking.status_code == 200
|
|
assert networking.json()["hostname"]
|
|
|
|
|
|
async def test_soap_login_service_content_and_inventory(client: AsyncClient) -> None:
|
|
login = await client.post(
|
|
"/sdk",
|
|
content="""<?xml version="1.0"?>
|
|
<Envelope xmlns="http://schemas.xmlsoap.org/soap/envelope/">
|
|
<Body>
|
|
<Login xmlns="urn:vim25">
|
|
<_this type="SessionManager">SessionManager</_this>
|
|
<userName>administrator@vsphere.local</userName>
|
|
<password>VMware1!</password>
|
|
</Login>
|
|
</Body>
|
|
</Envelope>""",
|
|
headers={"Content-Type": "text/xml"},
|
|
)
|
|
assert login.status_code == 200
|
|
assert "LoginResponse" in login.text
|
|
cookie = (login.headers.get("set-cookie") or "").split(";")[0]
|
|
assert "vmware_soap_session" in cookie
|
|
|
|
content = await client.post(
|
|
"/sdk",
|
|
content="""<?xml version="1.0"?>
|
|
<Envelope xmlns="http://schemas.xmlsoap.org/soap/envelope/">
|
|
<Body>
|
|
<RetrieveServiceContent xmlns="urn:vim25">
|
|
<_this type="ServiceInstance">ServiceInstance</_this>
|
|
</RetrieveServiceContent>
|
|
</Body>
|
|
</Envelope>""",
|
|
headers={"Content-Type": "text/xml", "Cookie": cookie},
|
|
)
|
|
assert content.status_code == 200
|
|
assert "propertyCollector" in content.text
|
|
assert "eventManager" in content.text
|
|
|
|
props = await client.post(
|
|
"/sdk",
|
|
content="""<?xml version="1.0"?>
|
|
<Envelope xmlns="http://schemas.xmlsoap.org/soap/envelope/">
|
|
<Body>
|
|
<RetrieveProperties xmlns="urn:vim25">
|
|
<_this type="PropertyCollector">propertyCollector</_this>
|
|
<specSet>
|
|
<propSet><type>Folder</type><pathSet>childEntity</pathSet><pathSet>name</pathSet></propSet>
|
|
<objectSet>
|
|
<obj type="Folder">group-d1</obj>
|
|
<selectSet xsi:type="TraversalSpec">
|
|
<type>Folder</type><path>childEntity</path>
|
|
</selectSet>
|
|
</objectSet>
|
|
</specSet>
|
|
</RetrieveProperties>
|
|
</Body>
|
|
</Envelope>""",
|
|
headers={"Content-Type": "text/xml", "Cookie": cookie},
|
|
)
|
|
assert props.status_code == 200
|
|
assert "datacenter-21" in props.text or "Datacenter" in props.text
|
|
|
|
events = await client.post(
|
|
"/sdk",
|
|
content="""<?xml version="1.0"?>
|
|
<Envelope xmlns="http://schemas.xmlsoap.org/soap/envelope/">
|
|
<Body>
|
|
<QueryEvents xmlns="urn:vim25">
|
|
<_this type="EventManager">EventManager</_this>
|
|
</QueryEvents>
|
|
</Body>
|
|
</Envelope>""",
|
|
headers={"Content-Type": "text/xml", "Cookie": cookie},
|
|
)
|
|
assert events.status_code == 200
|
|
assert "QueryEventsResponse" in events.text
|
|
|
|
|
|
async def test_vm_lifecycle_and_power(client: AsyncClient) -> None:
|
|
headers = await _session(client)
|
|
created = await client.post(
|
|
"/api/vcenter/vm",
|
|
headers={**headers, "Content-Type": "application/json"},
|
|
json={
|
|
"name": "full-api-lifecycle",
|
|
"placement": {"folder": "group-v23", "host": "host-11", "datastore": "datastore-31"},
|
|
"cpu_count": 1,
|
|
"memory_size_MiB": 512,
|
|
},
|
|
)
|
|
assert created.status_code in {200, 201}, created.text
|
|
vm = created.json()
|
|
if isinstance(vm, dict):
|
|
vm = vm.get("vm") or vm.get("value") or vm
|
|
assert isinstance(vm, str)
|
|
detail = await client.get(f"/api/vcenter/vm/{vm}", headers=headers)
|
|
assert detail.status_code == 200
|
|
power = await client.post(
|
|
f"/api/vcenter/vm/{vm}/power",
|
|
params={"action": "start"},
|
|
headers=headers,
|
|
)
|
|
assert power.status_code in {200, 204}, power.text
|
|
# powered-on VMs cannot be deleted — stop first (vSphere semantics)
|
|
stop = await client.post(
|
|
f"/api/vcenter/vm/{vm}/power",
|
|
params={"action": "stop"},
|
|
headers=headers,
|
|
)
|
|
assert stop.status_code in {200, 204}, stop.text
|
|
deleted = await client.delete(f"/api/vcenter/vm/{vm}", headers=headers)
|
|
assert deleted.status_code in {200, 204}, deleted.text
|
|
|
|
|
|
async def test_coverage_registry_matches_implemented_constant() -> None:
|
|
assert len(catalog_entries()) == len(IMPLEMENTED)
|
|
for verb, path in IMPLEMENTED:
|
|
assert re.match(r"^/(api|rest)/", path), path
|
|
assert verb in {"GET", "POST", "PUT", "PATCH", "DELETE"}
|
|
|
|
|
|
@pytest.mark.parametrize("major", sorted(VERSIONS))
|
|
async def test_major_matrix_all_verbs_no_server_error(client: AsyncClient, major: int) -> None:
|
|
"""Apply each catalog major and exercise every registered GET/POST/PATCH/DELETE."""
|
|
|
|
headers = await _session(client)
|
|
apply = await client.post("/ui/api/contract/apply", params={"major": major})
|
|
assert apply.status_code == 200, apply.text
|
|
failures: list[str] = []
|
|
order = {"GET": 0, "PUT": 1, "PATCH": 2, "POST": 3, "DELETE": 4}
|
|
entries = sorted(
|
|
catalog_entries_for_major(major),
|
|
key=lambda item: (order.get(item["verb"], 9), item["path"]),
|
|
)
|
|
for entry in entries:
|
|
verb = entry["verb"]
|
|
path = entry["path"]
|
|
if verb == "DELETE" and path in {"/api/session", "/rest/com/vmware/cis/session"}:
|
|
continue
|
|
url = _concrete(path)
|
|
kwargs: dict = {"headers": {**headers}}
|
|
if verb in {"POST", "PATCH", "PUT"}:
|
|
kwargs["headers"] = {**headers, "Content-Type": "application/json"}
|
|
if path.endswith("/power") and "/guest/" not in path:
|
|
url = f"{url}?action=start"
|
|
kwargs["json"] = {}
|
|
elif path.endswith("/guest/power"):
|
|
url = f"{url}?action=reboot"
|
|
kwargs["json"] = {}
|
|
elif path.endswith("/maintenance"):
|
|
url = f"{url}?action=enter"
|
|
kwargs["json"] = {}
|
|
elif path == "/api/vcenter/folder/{folder}":
|
|
url = f"{url}?action=rename"
|
|
kwargs["json"] = {"name": "renamed-by-matrix"}
|
|
elif path == "/api/content/local-library":
|
|
kwargs["json"] = {"create_spec": {"name": f"lib-m{major}-{os.urandom(3).hex()}"}}
|
|
elif path == "/api/cis/tagging/category":
|
|
kwargs["json"] = {
|
|
"create_spec": {
|
|
"name": f"cat-m{major}-{os.urandom(3).hex()}",
|
|
"cardinality": "MULTIPLE",
|
|
"associable_types": [],
|
|
}
|
|
}
|
|
elif path == "/api/cis/tagging/tag":
|
|
kwargs["json"] = {
|
|
"create_spec": {
|
|
"name": f"tag-m{major}-{os.urandom(3).hex()}",
|
|
"category_id": "missing-category",
|
|
}
|
|
}
|
|
elif "tag-association" in path:
|
|
kwargs["json"] = {
|
|
"action": "list-attached-tags",
|
|
"tag_id": "x",
|
|
"object_id": {"type": "VirtualMachine", "id": "vm-101"},
|
|
}
|
|
elif path == "/api/vcenter/network/dvpg":
|
|
kwargs["json"] = {
|
|
"name": f"dvpg-m{major}-{os.urandom(2).hex()}",
|
|
"dvs": "dvs-51",
|
|
"vlan_id": 20,
|
|
}
|
|
elif path == "/api/content/library/item":
|
|
kwargs["json"] = {
|
|
"create_spec": {
|
|
"library_id": "lib-missing",
|
|
"name": f"item-m{major}-{os.urandom(2).hex()}",
|
|
"type": "ovf",
|
|
}
|
|
}
|
|
elif path == "/api/vcenter/authorization/permissions":
|
|
kwargs["json"] = {
|
|
"principal": "readonly@vsphere.local",
|
|
"role": "ReadOnly",
|
|
"entity": "datacenter-21",
|
|
}
|
|
elif path == "/api/vcenter/datastore/{datastore}/files":
|
|
kwargs["json"] = {
|
|
"path": f"/probe-m{major}-{os.urandom(2).hex()}.txt",
|
|
"size": 1,
|
|
"type": "FILE",
|
|
}
|
|
elif path.endswith("/hardware/cpu"):
|
|
kwargs["json"] = {"count": 2}
|
|
elif path.endswith("/hardware/memory"):
|
|
kwargs["json"] = {"size_MiB": 1024}
|
|
elif path.endswith("/hardware/disk"):
|
|
kwargs["json"] = {"type": "SCSI", "new_vmdk": {"capacity": 1024}}
|
|
elif path.endswith("/hardware/ethernet"):
|
|
kwargs["json"] = {
|
|
"type": "VMXNET3",
|
|
"backing": {"type": "STANDARD_PORTGROUP", "network": "network-41"},
|
|
}
|
|
elif path.endswith("/snapshots") and verb == "POST":
|
|
kwargs["json"] = {"name": f"snap-m{major}-{os.urandom(2).hex()}"}
|
|
elif "/snapshots/" in path and verb == "POST":
|
|
kwargs["json"] = {"action": "revert"}
|
|
elif path.endswith("/clone"):
|
|
kwargs["json"] = {
|
|
"name": f"clone-m{major}-{os.urandom(2).hex()}",
|
|
"placement": {"folder": "group-v23", "host": "host-11"},
|
|
}
|
|
elif path.endswith("/relocate"):
|
|
kwargs["json"] = {"placement": {"host": "host-12"}}
|
|
elif path.endswith("/console/tickets"):
|
|
kwargs["json"] = {"type": "WEBMKS"}
|
|
elif path.endswith("/guest/customization"):
|
|
kwargs["json"] = {"name": {"name": f"guest-m{major}"}}
|
|
elif path == "/api/vcenter/vm/{vm}" and verb == "POST":
|
|
kwargs["json"] = {"action": "unregister"}
|
|
else:
|
|
kwargs["json"] = {"name": f"probe-{major}-{os.urandom(2).hex()}"}
|
|
if verb == "DELETE" and path.endswith("{vm}"):
|
|
url = "/api/vcenter/vm/vm-missing-matrix"
|
|
response = await client.request(verb, url, **kwargs)
|
|
if response.status_code >= 500:
|
|
failures.append(f"{verb} {path} -> {response.status_code} {response.text[:160]}")
|
|
assert failures == [], "\n".join(failures)
|