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.
93 lines
3.5 KiB
Python
93 lines
3.5 KiB
Python
"""Minimal PBM (Profile-Based Management) SOAP endpoint for Terraform/govmomi."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from xml.sax.saxutils import escape
|
|
|
|
from fastapi import APIRouter, Request, Response
|
|
from fastapi.responses import PlainTextResponse
|
|
|
|
router = APIRouter(tags=["vSphere PBM"])
|
|
|
|
NS_SOAP = "http://schemas.xmlsoap.org/soap/envelope/"
|
|
|
|
|
|
@router.get("/pbm")
|
|
@router.get("/pbm/")
|
|
@router.get("/pbm/sdk")
|
|
@router.get("/pbm/sdk/")
|
|
async def pbm_get() -> PlainTextResponse:
|
|
return PlainTextResponse("VMware PBM SDK simulator — POST SOAP to /pbm/sdk")
|
|
|
|
|
|
@router.post("/pbm")
|
|
@router.post("/pbm/")
|
|
@router.post("/pbm/sdk")
|
|
@router.post("/pbm/sdk/")
|
|
async def pbm_post(request: Request) -> Response:
|
|
body = (await request.body()).decode("utf-8", errors="replace")
|
|
if "PbmRetrieveServiceContent" in body or "RetrieveContent" in body:
|
|
xml = _wrap(
|
|
"PbmRetrieveServiceContentResponse",
|
|
"""<returnval>
|
|
<about>
|
|
<name>VMware vCenter Profile-Driven Storage Service</name>
|
|
<version>2.0</version>
|
|
</about>
|
|
<sessionManager type="PbmSessionManager">SessionManager</sessionManager>
|
|
<capabilityMetadataManager type="PbmCapabilityMetadataManager">CapabilityMetadataManager</capabilityMetadataManager>
|
|
<profileManager type="PbmProfileProfileManager">ProfileManager</profileManager>
|
|
<complianceManager type="PbmComplianceManager">ComplianceManager</complianceManager>
|
|
<placementSolver type="PbmPlacementSolver">PlacementSolver</placementSolver>
|
|
</returnval>""",
|
|
)
|
|
return Response(content=xml, media_type='text/xml; charset="utf-8"')
|
|
if "PbmQueryProfile" in body or "PbmQueryDefaultRequirementProfile" in body:
|
|
xml = _wrap(
|
|
"PbmQueryProfileResponse",
|
|
"""<returnval>
|
|
<uniqueId>com.vmware.storage.default</uniqueId>
|
|
<name>vSAN Default Storage Policy</name>
|
|
</returnval>
|
|
<returnval>
|
|
<uniqueId>policy-thin</uniqueId>
|
|
<name>Thin provision</name>
|
|
</returnval>""",
|
|
)
|
|
return Response(content=xml, media_type='text/xml; charset="utf-8"')
|
|
if "PbmQueryAssociatedProfile" in body:
|
|
# Empty association list (no storage policy) — must use the correct response
|
|
# tag so govmomi unmarshals []types.PbmProfileId instead of panicking.
|
|
xml = _wrap("PbmQueryAssociatedProfileResponse", "")
|
|
return Response(content=xml, media_type='text/xml; charset="utf-8"')
|
|
if "PbmRetrieveContent" in body:
|
|
xml = _wrap("PbmRetrieveContentResponse", "<returnval></returnval>")
|
|
return Response(content=xml, media_type='text/xml; charset="utf-8"')
|
|
# Unknown PBM op — empty success so clients continue.
|
|
op = "PbmMethod"
|
|
match = re.search(r"<(?:\w+:)?([A-Za-z0-9_]+)(?:\s|>)", body)
|
|
if match:
|
|
op = match.group(1)
|
|
return Response(
|
|
content=_wrap(f"{op}Response", "<returnval></returnval>"),
|
|
media_type='text/xml; charset="utf-8"',
|
|
)
|
|
|
|
|
|
def _wrap(response_tag: str, inner: str) -> str:
|
|
return f"""<?xml version="1.0" encoding="UTF-8"?>
|
|
<soapenv:Envelope xmlns:soapenv="{NS_SOAP}">
|
|
<soapenv:Body>
|
|
<{response_tag} xmlns="urn:pbm">
|
|
{inner}
|
|
</{response_tag}>
|
|
</soapenv:Body>
|
|
</soapenv:Envelope>
|
|
"""
|
|
|
|
|
|
# silence unused escape import warning by using it in fault helper
|
|
def _fault(message: str) -> str:
|
|
return f"<faultstring>{escape(message)}</faultstring>"
|