vmware-api-simulator/tests/integration/test_vsphere_api.py

"""Integration smoke for native vSphere REST + SOAP."""

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 _session(client: AsyncClient) -> dict[str, str]:
    login = await client.post(
        "/api/session",
        auth=("administrator@vsphere.local", "VMware1!"),
    )
    assert login.status_code == 201
    return {"vmware-api-session-id": login.json()}


async def test_rest_session_inventory_power_clone_tags(client: AsyncClient) -> None:
    headers = await _session(client)

    vms = await client.get("/api/vcenter/vm", headers=headers)
    assert vms.status_code == 200
    assert len(vms.json()) >= 5

    vm_id = next(item["vm"] for item in vms.json() if item["name"] == "app-01")
    power = await client.post(
        f"/api/vcenter/vm/{vm_id}/power",
        params={"action": "start"},
        headers=headers,
    )
    assert power.status_code == 200
    assert str(power.json().get("task") or "").startswith("task-")

    snap = await client.post(
        f"/api/vcenter/vm/{vm_id}/snapshots",
        headers=headers,
        json={"name": "pre-update", "description": "lab"},
    )
    assert snap.status_code == 200
    assert "snapshot" in snap.json()

    clone = await client.post(
        f"/api/vcenter/vm/{vm_id}/clone",
        headers=headers,
        json={"name": "app-01-clone"},
    )
    assert clone.status_code == 200
    assert clone.json()["vm"].startswith("vm-")

    cat = await client.post(
        "/api/cis/tagging/category",
        headers=headers,
        json={"name": "Owner-api-test", "associable_types": ["VirtualMachine"]},
    )
    assert cat.status_code == 200, cat.text
    tag = await client.post(
        "/api/cis/tagging/tag",
        headers=headers,
        json={"category_id": cat.json(), "name": "team-a"},
    )
    assert tag.status_code == 200

    libs = await client.get("/api/content/library", headers=headers)
    assert libs.status_code == 200
    assert len(libs.json()) >= 1

    versions = await client.get("/ui/api/versions")
    assert versions.status_code == 200
    assert versions.json()["plane"] == "vsphere-rest"


async def test_readonly_cannot_mutate(client: AsyncClient) -> None:
    login = await client.post(
        "/api/session",
        auth=("readonly@vsphere.local", "VMware1!"),
    )
    assert login.status_code == 201
    headers = {"vmware-api-session-id": login.json()}
    session = await client.get("/api/session", headers=headers)
    assert session.status_code == 200
    assert session.content in (b"", b"null") or not session.text.strip()
    assert "ReadOnly" in (session.headers.get("x-vmware-session-roles") or "")
    vms = await client.get("/api/vcenter/vm", headers=headers)
    assert vms.status_code == 200
    assert len(vms.json()) >= 5
    power = await client.post(
        f"/api/vcenter/vm/{vms.json()[0]['vm']}/power",
        params={"action": "start"},
        headers=headers,
    )
    assert power.status_code == 403


async def test_soap_property_collector_and_wsdl(client: AsyncClient) -> None:
    content = await client.post(
        "/sdk",
        content="""<?xml version="1.0"?>
        <soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/">
          <soapenv:Body><RetrieveServiceContent xmlns="urn:vim25"><_this type="ServiceInstance">ServiceInstance</_this></RetrieveServiceContent></soapenv:Body>
        </soapenv:Envelope>""",
        headers={"Content-Type": "text/xml"},
    )
    assert content.status_code == 200
    assert "SessionManager" in content.text

    login = await client.post(
        "/sdk",
        content="""<?xml version="1.0"?>
        <soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/">
          <soapenv:Body>
            <Login xmlns="urn:vim25">
              <_this type="SessionManager">SessionManager</_this>
              <userName>administrator@vsphere.local</userName>
              <password>VMware1!</password>
            </Login>
          </soapenv:Body>
        </soapenv:Envelope>""",
        headers={"Content-Type": "text/xml"},
    )
    assert login.status_code == 200
    cookie = login.headers.get("set-cookie") or ""
    assert "vmware_soap_session" in cookie

    updates = await client.post(
        "/sdk",
        content="""<?xml version="1.0"?>
        <soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/">
          <soapenv:Body>
            <WaitForUpdatesEx xmlns="urn:vim25">
              <_this type="PropertyCollector">propertyCollector</_this>
              <version></version>
            </WaitForUpdatesEx>
          </soapenv:Body>
        </soapenv:Envelope>""",
        headers={"Content-Type": "text/xml", "Cookie": cookie.split(";")[0]},
    )
    assert updates.status_code == 200
    assert "WaitForUpdatesExResponse" in updates.text
    assert "vm-101" in updates.text

    wsdl = await client.get("/sdk/vimService.wsdl")
    assert wsdl.status_code == 200
    assert "VimService" in wsdl.text