193 lines
7.7 KiB
Python
193 lines
7.7 KiB
Python
from __future__ import annotations
|
|
|
|
from urllib.parse import parse_qs
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from pve_vm_setup.errors import ProxmoxConnectError
|
|
from pve_vm_setup.models.workflow import VmConfig
|
|
from pve_vm_setup.services.proxmox import ProxmoxApiClient
|
|
from pve_vm_setup.settings import AppSettings
|
|
|
|
|
|
def build_settings() -> AppSettings:
|
|
return AppSettings.from_env(
|
|
{
|
|
"PROXMOX_URL": "https://proxmox.example.invalid:8006",
|
|
"PROXMOX_USER": "root",
|
|
"PROXMOX_PASSWORD": "secret",
|
|
"PROXMOX_REALM": "pam",
|
|
},
|
|
load_dotenv_file=False,
|
|
)
|
|
|
|
|
|
def test_client_uses_api_base_when_loading_realms() -> None:
|
|
recorded_urls: list[str] = []
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
recorded_urls.append(str(request.url))
|
|
return httpx.Response(200, json={"data": [{"realm": "pam", "comment": "Linux PAM"}]})
|
|
|
|
client = ProxmoxApiClient(build_settings(), transport=httpx.MockTransport(handler))
|
|
try:
|
|
realms = client.load_realms()
|
|
finally:
|
|
client.close()
|
|
|
|
assert realms[0].name == "pam"
|
|
assert recorded_urls == ["https://proxmox.example.invalid:8006/api2/json/access/domains"]
|
|
|
|
|
|
def test_client_maps_connect_errors() -> None:
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
raise httpx.ConnectError("boom", request=request)
|
|
|
|
client = ProxmoxApiClient(build_settings(), transport=httpx.MockTransport(handler))
|
|
try:
|
|
with pytest.raises(ProxmoxConnectError):
|
|
client.load_realms()
|
|
finally:
|
|
client.close()
|
|
|
|
|
|
def test_client_attaches_serial_device_without_switching_display_to_serial() -> None:
|
|
requests: list[tuple[str, str, bytes]] = []
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
requests.append((request.method, request.url.path, request.content))
|
|
path = request.url.path
|
|
if path.endswith("/nodes/fake-node-01/qemu") and request.method == "POST":
|
|
return httpx.Response(200, json={"data": "UPID:create"})
|
|
if path.endswith("/nodes/fake-node-01/tasks/UPID:create/status"):
|
|
return httpx.Response(200, json={"data": {"status": "stopped", "exitstatus": "OK"}})
|
|
if path.endswith("/nodes/fake-node-01/qemu/123/config") and request.method == "PUT":
|
|
return httpx.Response(200, json={"data": "UPID:serial"})
|
|
if path.endswith("/nodes/fake-node-01/tasks/UPID:serial/status"):
|
|
return httpx.Response(200, json={"data": {"status": "stopped", "exitstatus": "OK"}})
|
|
raise AssertionError(f"Unexpected request: {request.method} {request.url}")
|
|
|
|
client = ProxmoxApiClient(build_settings(), transport=httpx.MockTransport(handler))
|
|
client._ticket = "ticket"
|
|
client._csrf_token = "csrf"
|
|
client._client.cookies.set("PVEAuthCookie", "ticket")
|
|
|
|
config = VmConfig()
|
|
config.general.node = "fake-node-01"
|
|
config.general.vmid = 123
|
|
config.general.name = "demo"
|
|
config.general.ha_enabled = False
|
|
config.os.storage = "cephfs"
|
|
config.os.iso = "cephfs:iso/nixos.iso"
|
|
|
|
try:
|
|
client.create_vm(config)
|
|
finally:
|
|
client.close()
|
|
|
|
serial_request = next(
|
|
content
|
|
for method, path, content in requests
|
|
if method == "PUT" and path.endswith("/nodes/fake-node-01/qemu/123/config")
|
|
)
|
|
payload = parse_qs(serial_request.decode())
|
|
|
|
assert payload["serial0"] == ["socket"]
|
|
assert "vga" not in payload
|
|
|
|
|
|
def test_client_starts_vm_after_create_when_requested() -> None:
|
|
requests: list[tuple[str, str, bytes]] = []
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
requests.append((request.method, request.url.path, request.content))
|
|
path = request.url.path
|
|
if path.endswith("/nodes/fake-node-01/qemu") and request.method == "POST":
|
|
return httpx.Response(200, json={"data": "UPID:create"})
|
|
if path.endswith("/nodes/fake-node-01/tasks/UPID:create/status"):
|
|
return httpx.Response(200, json={"data": {"status": "stopped", "exitstatus": "OK"}})
|
|
if path.endswith("/nodes/fake-node-01/qemu/123/config") and request.method == "PUT":
|
|
return httpx.Response(200, json={"data": "UPID:serial"})
|
|
if path.endswith("/nodes/fake-node-01/tasks/UPID:serial/status"):
|
|
return httpx.Response(200, json={"data": {"status": "stopped", "exitstatus": "OK"}})
|
|
if path.endswith("/nodes/fake-node-01/qemu/123/status/start") and request.method == "POST":
|
|
return httpx.Response(200, json={"data": "UPID:start"})
|
|
if path.endswith("/nodes/fake-node-01/tasks/UPID:start/status"):
|
|
return httpx.Response(200, json={"data": {"status": "stopped", "exitstatus": "OK"}})
|
|
raise AssertionError(f"Unexpected request: {request.method} {request.url}")
|
|
|
|
client = ProxmoxApiClient(build_settings(), transport=httpx.MockTransport(handler))
|
|
client._ticket = "ticket"
|
|
client._csrf_token = "csrf"
|
|
client._client.cookies.set("PVEAuthCookie", "ticket")
|
|
|
|
config = VmConfig()
|
|
config.general.node = "fake-node-01"
|
|
config.general.vmid = 123
|
|
config.general.name = "demo"
|
|
config.general.ha_enabled = False
|
|
config.os.storage = "cephfs"
|
|
config.os.iso = "cephfs:iso/nixos.iso"
|
|
|
|
try:
|
|
client.create_vm(config, start_after_create=True)
|
|
finally:
|
|
client.close()
|
|
|
|
assert any(
|
|
method == "POST" and path.endswith("/nodes/fake-node-01/qemu/123/status/start")
|
|
for method, path, _ in requests
|
|
)
|
|
|
|
|
|
def test_client_registers_ha_without_start_when_auto_start_disabled() -> None:
|
|
requests: list[tuple[str, str, bytes]] = []
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
requests.append((request.method, request.url.path, request.content))
|
|
path = request.url.path
|
|
if path.endswith("/nodes/fake-node-01/qemu") and request.method == "POST":
|
|
return httpx.Response(200, json={"data": "UPID:create"})
|
|
if path.endswith("/nodes/fake-node-01/tasks/UPID:create/status"):
|
|
return httpx.Response(200, json={"data": {"status": "stopped", "exitstatus": "OK"}})
|
|
if path.endswith("/nodes/fake-node-01/qemu/123/config") and request.method == "PUT":
|
|
return httpx.Response(200, json={"data": "UPID:serial"})
|
|
if path.endswith("/nodes/fake-node-01/tasks/UPID:serial/status"):
|
|
return httpx.Response(200, json={"data": {"status": "stopped", "exitstatus": "OK"}})
|
|
if path.endswith("/cluster/ha/resources") and request.method == "POST":
|
|
return httpx.Response(200, json={"data": "UPID:ha"})
|
|
if path.endswith("/nodes/fake-node-01/tasks/UPID:ha/status"):
|
|
return httpx.Response(200, json={"data": {"status": "stopped", "exitstatus": "OK"}})
|
|
raise AssertionError(f"Unexpected request: {request.method} {request.url}")
|
|
|
|
client = ProxmoxApiClient(build_settings(), transport=httpx.MockTransport(handler))
|
|
client._ticket = "ticket"
|
|
client._csrf_token = "csrf"
|
|
client._client.cookies.set("PVEAuthCookie", "ticket")
|
|
|
|
config = VmConfig()
|
|
config.general.node = "fake-node-01"
|
|
config.general.vmid = 123
|
|
config.general.name = "demo"
|
|
config.general.ha_enabled = True
|
|
config.os.storage = "cephfs"
|
|
config.os.iso = "cephfs:iso/nixos.iso"
|
|
|
|
try:
|
|
client.create_vm(config, start_after_create=False)
|
|
finally:
|
|
client.close()
|
|
|
|
ha_request = next(
|
|
content
|
|
for method, path, content in requests
|
|
if method == "POST" and path.endswith("/cluster/ha/resources")
|
|
)
|
|
payload = parse_qs(ha_request.decode())
|
|
|
|
assert payload["state"] == ["stopped"]
|
|
assert not any(
|
|
method == "POST" and path.endswith("/nodes/fake-node-01/qemu/123/status/start")
|
|
for method, path, _ in requests
|
|
)
|