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.
136 lines
4.6 KiB
Python
136 lines
4.6 KiB
Python
"""SOAP CreateVM / FindChild / guest filesystem REST parity."""
|
|
|
|
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, login.text
|
|
session = login.headers.get("vmware-api-session-id")
|
|
assert session
|
|
return session
|
|
|
|
|
|
async def test_soap_create_vm_and_find_child(client: AsyncClient) -> None:
|
|
session = await _soap_login(client)
|
|
headers = {
|
|
"Content-Type": "text/xml",
|
|
"vmware-api-session-id": session,
|
|
"Cookie": f'vmware_soap_session="{session}"',
|
|
}
|
|
create = await client.post(
|
|
"/sdk",
|
|
content="""<?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>soap-create-lab</name>
|
|
<guestId>otherGuest64</guestId>
|
|
<numCPUs>2</numCPUs>
|
|
<memoryMB>1024</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=headers,
|
|
)
|
|
assert create.status_code == 200, create.text
|
|
assert "CreateVM_TaskResponse" in create.text
|
|
assert "task-" in create.text
|
|
|
|
find = await client.post(
|
|
"/sdk",
|
|
content="""<?xml version="1.0"?>
|
|
<soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/">
|
|
<soapenv:Body>
|
|
<FindChild xmlns="urn:vim25">
|
|
<_this type="SearchIndex">SearchIndex</_this>
|
|
<entity type="Folder">group-v23</entity>
|
|
<name>soap-create-lab</name>
|
|
</FindChild>
|
|
</soapenv:Body>
|
|
</soapenv:Envelope>""",
|
|
headers=headers,
|
|
)
|
|
assert find.status_code == 200, find.text
|
|
assert "VirtualMachine" in find.text
|
|
|
|
|
|
async def test_rest_guest_filesystem_roundtrip(client: AsyncClient) -> None:
|
|
login = await client.post(
|
|
"/api/session",
|
|
auth=("administrator@vsphere.local", "VMware1!"),
|
|
)
|
|
assert login.status_code in {200, 201}
|
|
headers = {"vmware-api-session-id": login.json()}
|
|
vms = await client.get("/api/vcenter/vm", headers=headers)
|
|
assert vms.status_code == 200
|
|
vm = next(item["vm"] for item in vms.json() if item["name"] == "web-01")
|
|
put = await client.put(
|
|
f"/api/vcenter/vm/{vm}/guest/filesystem",
|
|
params={"path": "/tmp/fs-test"},
|
|
headers=headers,
|
|
json={"content": "hello-lab"},
|
|
)
|
|
assert put.status_code == 204
|
|
get = await client.get(
|
|
f"/api/vcenter/vm/{vm}/guest/filesystem",
|
|
params={"path": "/tmp/fs-test"},
|
|
headers=headers,
|
|
)
|
|
assert get.status_code == 200
|
|
assert get.json()["content"] == "hello-lab"
|
|
listing = await client.get(
|
|
f"/api/vcenter/vm/{vm}/guest/filesystem/files",
|
|
params={"path": "/tmp"},
|
|
headers=headers,
|
|
)
|
|
assert listing.status_code == 200
|
|
assert any(item["path"] == "/tmp/fs-test" for item in listing.json())
|