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.
107 lines
2.9 KiB
Python
107 lines
2.9 KiB
Python
#!/usr/bin/env python3
|
|
"""Raw `requests` cookbook against the vSphere REST gateway (no vsphere-automation-sdk)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import sys
|
|
import time
|
|
from typing import Any
|
|
|
|
import requests
|
|
import urllib3
|
|
|
|
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
|
|
|
|
BASE = os.environ.get("VSPHERE_BASE", "https://localhost").rstrip("/")
|
|
USER = os.environ.get("VSPHERE_USER", "administrator@vsphere.local")
|
|
PASSWORD = os.environ.get("VSPHERE_PASSWORD", "VMware1!")
|
|
VM_NAME = os.environ.get("VSPHERE_VM_NAME", "req-lab-01")
|
|
|
|
|
|
def api(
|
|
method: str,
|
|
path: str,
|
|
*,
|
|
headers: dict[str, str] | None = None,
|
|
json_body: dict[str, Any] | None = None,
|
|
) -> Any:
|
|
response = requests.request(
|
|
method,
|
|
f"{BASE}{path}",
|
|
headers=headers,
|
|
json=json_body,
|
|
verify=False, # local self-signed development certificate only
|
|
timeout=60,
|
|
)
|
|
response.raise_for_status()
|
|
if response.status_code == 204 or not response.content:
|
|
return None
|
|
return response.json()
|
|
|
|
|
|
def wait_task(headers: dict[str, str], task: str, timeout: float = 120.0) -> None:
|
|
deadline = time.time() + timeout
|
|
while time.time() < deadline:
|
|
status = api("GET", f"/api/cis/tasks/{task}", headers=headers)
|
|
if status.get("status") in {"SUCCEEDED", "FAILED"}:
|
|
return
|
|
time.sleep(0.3)
|
|
raise TimeoutError(task)
|
|
|
|
|
|
def session_headers() -> dict[str, str]:
|
|
response = requests.post(
|
|
f"{BASE}/api/session",
|
|
auth=(USER, PASSWORD),
|
|
verify=False,
|
|
timeout=60,
|
|
)
|
|
response.raise_for_status()
|
|
return {"vmware-api-session-id": response.json()}
|
|
|
|
|
|
def main() -> int:
|
|
headers = session_headers()
|
|
print("session:", headers["vmware-api-session-id"])
|
|
|
|
vms = api("GET", "/api/vcenter/vm", headers=headers)
|
|
print("vms before:", len(vms))
|
|
|
|
moid = api(
|
|
"POST",
|
|
"/api/vcenter/vm",
|
|
headers=headers,
|
|
json_body={
|
|
"name": VM_NAME,
|
|
"guest_OS": "OTHER_GUEST_64",
|
|
"placement": {
|
|
"folder": "group-v23",
|
|
"host": "host-11",
|
|
"datastore": "datastore-31",
|
|
"resource_pool": "resgroup-22",
|
|
},
|
|
"cpu": {"count": 1},
|
|
"memory": {"size_MiB": 512},
|
|
},
|
|
)
|
|
print("created:", moid)
|
|
|
|
power = api("POST", f"/api/vcenter/vm/{moid}/power?action=start", headers=headers)
|
|
wait_task(headers, power["task"])
|
|
|
|
detail = api("GET", f"/api/vcenter/vm/{moid}", headers=headers)
|
|
print("power_state:", detail["power_state"])
|
|
|
|
power = api("POST", f"/api/vcenter/vm/{moid}/power?action=stop", headers=headers)
|
|
wait_task(headers, power["task"])
|
|
|
|
api("DELETE", f"/api/vcenter/vm/{moid}", headers=headers)
|
|
api("DELETE", "/api/session", headers=headers)
|
|
print("ok")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|