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.
166 lines
5.6 KiB
Python
166 lines
5.6 KiB
Python
"""SOAP PropertyCollector / TaskManager fidelity tests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
|
|
import pytest
|
|
from httpx import ASGITransport, AsyncClient
|
|
|
|
from app.config import Settings
|
|
from app.main import create_app
|
|
from app.vsphere.seed import seed_vsphere_inventory
|
|
|
|
pytestmark = pytest.mark.integration
|
|
|
|
|
|
@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 _soap_login(client: AsyncClient) -> str:
|
|
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 ""
|
|
assert "vmware_soap_session" in cookie
|
|
return cookie.split(";")[0]
|
|
|
|
|
|
async def test_folder_child_entity_and_path(client: AsyncClient) -> None:
|
|
cookie = await _soap_login(client)
|
|
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></objectSet>
|
|
</specSet>
|
|
</RetrieveProperties>
|
|
</Body>
|
|
</Envelope>""",
|
|
headers={"Content-Type": "text/xml", "Cookie": cookie},
|
|
)
|
|
assert props.status_code == 200
|
|
assert "childEntity" in props.text
|
|
assert "datacenter-21" in props.text
|
|
|
|
path = await client.post(
|
|
"/sdk",
|
|
content="""<?xml version="1.0"?>
|
|
<Envelope xmlns="http://schemas.xmlsoap.org/soap/envelope/">
|
|
<Body>
|
|
<FindByInventoryPath xmlns="urn:vim25">
|
|
<_this type="SearchIndex">SearchIndex</_this>
|
|
<inventoryPath>/Datacenters/Datacenter/vm/web-01</inventoryPath>
|
|
</FindByInventoryPath>
|
|
</Body>
|
|
</Envelope>""",
|
|
headers={"Content-Type": "text/xml", "Cookie": cookie},
|
|
)
|
|
assert path.status_code == 200
|
|
assert "vm-101" in path.text
|
|
|
|
|
|
async def test_wait_for_updates_version_and_power_task_id(client: AsyncClient) -> None:
|
|
cookie = await _soap_login(client)
|
|
first = await client.post(
|
|
"/sdk",
|
|
content="""<?xml version="1.0"?>
|
|
<Envelope xmlns="http://schemas.xmlsoap.org/soap/envelope/">
|
|
<Body>
|
|
<WaitForUpdatesEx xmlns="urn:vim25">
|
|
<_this type="PropertyCollector">propertyCollector</_this>
|
|
<version></version>
|
|
</WaitForUpdatesEx>
|
|
</Body>
|
|
</Envelope>""",
|
|
headers={"Content-Type": "text/xml", "Cookie": cookie},
|
|
)
|
|
assert first.status_code == 200
|
|
assert "<version>1</version>" in first.text
|
|
assert "enter" in first.text
|
|
|
|
second = await client.post(
|
|
"/sdk",
|
|
content="""<?xml version="1.0"?>
|
|
<Envelope xmlns="http://schemas.xmlsoap.org/soap/envelope/">
|
|
<Body>
|
|
<WaitForUpdatesEx xmlns="urn:vim25">
|
|
<_this type="PropertyCollector">propertyCollector</_this>
|
|
<version>1</version>
|
|
</WaitForUpdatesEx>
|
|
</Body>
|
|
</Envelope>""",
|
|
headers={"Content-Type": "text/xml", "Cookie": cookie},
|
|
)
|
|
assert second.status_code == 200
|
|
assert "enter" not in second.text
|
|
|
|
power = await client.post(
|
|
"/sdk",
|
|
content="""<?xml version="1.0"?>
|
|
<Envelope xmlns="http://schemas.xmlsoap.org/soap/envelope/">
|
|
<Body>
|
|
<PowerOnVM_Task xmlns="urn:vim25">
|
|
<_this type="VirtualMachine">vm-104</_this>
|
|
</PowerOnVM_Task>
|
|
</Body>
|
|
</Envelope>""",
|
|
headers={"Content-Type": "text/xml", "Cookie": cookie},
|
|
)
|
|
assert power.status_code == 200
|
|
assert "task-" in power.text
|
|
assert "task-1<" not in power.text
|
|
|
|
about = await client.get("/sdk/about.do")
|
|
assert about.status_code == 200
|
|
assert "vCenter" in about.text
|
|
|
|
wsdl = await client.get("/sdk/vimService.wsdl")
|
|
assert "WaitForUpdatesEx" in wsdl.text
|
|
assert "CancelTask" in wsdl.text
|
|
|
|
|
|
async def test_legacy_rest_value_wrapper(client: AsyncClient) -> None:
|
|
login = await client.post(
|
|
"/api/session",
|
|
auth=("administrator@vsphere.local", "VMware1!"),
|
|
)
|
|
headers = {"vmware-api-session-id": login.json()}
|
|
resp = await client.get("/rest/vcenter/vm", headers=headers, params={"limit": 2})
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert "value" in body
|
|
assert len(body["value"]) == 2
|