From 701157f570e3154da8149edf745ebb4c1fe073a5 Mon Sep 17 00:00:00 2001 From: Sergey Antropoff Date: Mon, 13 Jul 2026 01:54:04 +0300 Subject: [PATCH] feat: add durable QEMU create update delete --- README.md | 13 +- app/api/registry.py | 14 +- app/handlers/access.py | 7 +- app/handlers/qemu.py | 170 ++++++++++++++++++++++- app/main.py | 8 +- app/security/acl.py | 9 +- app/tasks/qemu.py | 92 ++++++++++++- docs/compatibility-0.1.0.md | 4 +- docs/original-prompt-gap-plan.md | 3 +- evidence/pve-9.2.3-0.1.0.json | 24 ++++ tests/compatibility/test_proxmoxer.py | 40 +++++- tests/unit/test_acl.py | 16 +++ tests/unit/test_core_handlers.py | 5 +- tests/unit/test_qemu_handlers.py | 187 ++++++++++++++++++++++++++ tests/unit/test_qemu_task.py | 93 +++++++++++++ 15 files changed, 659 insertions(+), 26 deletions(-) create mode 100644 tests/unit/test_qemu_handlers.py diff --git a/README.md b/README.md index e8e7d56..ebea553 100644 --- a/README.md +++ b/README.md @@ -10,9 +10,16 @@ limits are recorded in [the 0.1.0 compatibility report](docs/compatibility-0.1.0 The bundled PVE 9.2.3 declared contract contains 444 paths and 675 methods. Implemented semantics currently include version, ticket login, node listing and status, cluster resources, basic QEMU list/config/status/start/stop, and task -list/status/log, plus API-token list/create/read/update/delete. Mutations require -the ticket-bound CSRF header and execute through PostgreSQL-leased workers; all -other declared methods return an explicit unsupported error. +list/status/log, QEMU create/sync-update/async-update/delete, plus API-token +list/create/read/update/delete. Mutations require the ticket-bound CSRF header +and execute through PostgreSQL-leased workers; all other declared methods return +an explicit unsupported error. + +QEMU create, asynchronous config update, and delete return durable UPIDs and use +the same PostgreSQL resource lock as lifecycle operations. Synchronous config +PUT uses optimistic versioning. Common fields and unknown version-dependent +parameters are retained in JSONB; duplicate VMIDs and overlapping operations +fail with 409. ## Development diff --git a/app/api/registry.py b/app/api/registry.py index 851c866..9a53c68 100644 --- a/app/api/registry.py +++ b/app/api/registry.py @@ -14,7 +14,7 @@ from app.api.errors import ApiError, ContractValidationError from app.config import Settings from app.contracts.model import Method, Schema, Snapshot from app.db.pool import AsyncpgDatabase -from app.security.acl import AclEntry, authorize, requirement_from_contract +from app.security.acl import AclEntry, CapabilityRequirement, authorize, requirement_from_contract from app.security.auth import parse_api_token, verify_csrf, verify_secret, verify_ticket Handler = Callable[[Request, dict[str, Any]], Awaitable[Any]] @@ -159,13 +159,14 @@ async def _authenticate( if principal == "root@pam" and token_privileges is None: return database = cast(AsyncpgDatabase, request.app.state.database) - await _authorize(database, principal, token_privileges, method, inputs) + await _authorize(database, principal, token_privileges, semantic_path, method, inputs) async def _authorize( database: AsyncpgDatabase, principal: str, token_privileges: frozenset[str] | None, + semantic_path: str, method: Method, inputs: dict[str, Any], ) -> None: @@ -173,6 +174,8 @@ async def _authorize( requirement = requirement_from_contract( method.permissions, {name: str(value) for name, value in values.items()} ) + if requirement is None and semantic_path == "/nodes/{node}/qemu" and method.verb == "POST": + requirement = CapabilityRequirement(f"/vms/{values['vmid']}", frozenset({"VM.Allocate"})) if requirement is None: return rows = await database.pool.fetch( @@ -201,6 +204,7 @@ async def _authorize( requirement.privileges, entries, token_privileges=token_privileges, + require_all=requirement.require_all, ): raise ApiError(403, "permission check failed") @@ -241,7 +245,11 @@ async def _parse_inputs(request: Request, method: Method) -> dict[str, Any]: errors[name] = "property is not defined in schema" if errors: raise ContractValidationError(dict(sorted(errors.items()))) - return {"values": parsed, "path": dict(request.path_params)} + return { + "values": parsed, + "path": dict(request.path_params), + "provided": tuple(sorted(supplied)), + } def _coerce(value: Any, schema: Schema) -> Any: diff --git a/app/handlers/access.py b/app/handlers/access.py index aada5a4..48b955c 100644 --- a/app/handlers/access.py +++ b/app/handlers/access.py @@ -101,6 +101,7 @@ def register_access_handlers(registry: HandlerRegistry) -> None: async def token_update(request: Request, inputs: dict[str, Any]) -> dict[str, Any]: values = _values(inputs) + provided = frozenset(str(item) for item in inputs.get("provided", values)) userid, tokenid = str(values["userid"]), str(values["tokenid"]) _require_owner(request, userid) regenerate = bool(values.get("regenerate", False)) @@ -117,9 +118,9 @@ def register_access_handlers(registry: HandlerRegistry) -> None: extract(epoch from t.expires_at)::bigint AS expire""", userid, tokenid, - values.get("comment"), - _expire_value(values), - values.get("privsep"), + values.get("comment") if "comment" in provided else None, + _expire_value(values) if "expire" in provided else None, + values.get("privsep") if "privsep" in provided else None, hash_secret(secret) if secret is not None else None, ) if row is None: diff --git a/app/handlers/qemu.py b/app/handlers/qemu.py index 2d79d5f..bb1cfa6 100644 --- a/app/handlers/qemu.py +++ b/app/handlers/qemu.py @@ -11,6 +11,7 @@ from fastapi import Request from app.api.errors import ApiError from app.api.registry import HandlerRegistry from app.db.pool import AsyncpgDatabase +from app.db.primitives import ConflictError from app.tasks.repository import TaskRepository from app.tasks.upid import Upid @@ -43,14 +44,16 @@ def register_qemu_handlers(registry: HandlerRegistry) -> None: async def qemu_config(request: Request, inputs: dict[str, Any]) -> dict[str, Any]: node, vmid = str(_values(inputs)["node"]), str(_values(inputs)["vmid"]) row = await _database(request).pool.fetchrow( - """SELECT r.state FROM resources r JOIN nodes n ON n.id=r.node_id + """SELECT r.state, v.config FROM resources r + JOIN nodes n ON n.id=r.node_id + JOIN virtual_machines v ON v.resource_id=r.id WHERE n.name=$1 AND r.kind='qemu' AND r.external_id=$2""", node, vmid, ) if row is None: raise ApiError(404, "virtual machine does not exist") - return {"vmid": int(vmid), **_state(row["state"])} + return {"vmid": int(vmid), **_state(row["config"]), **_state(row["state"])} async def qemu_status(request: Request, inputs: dict[str, Any]) -> dict[str, Any]: return await qemu_config(request, inputs) @@ -74,7 +77,17 @@ def register_qemu_handlers(registry: HandlerRegistry) -> None: raise ApiError(409, f"cannot {operation} VM while it is {current}") timestamp = int(await database.pool.fetchval("SELECT extract(epoch from now())::bigint")) pid = int(await database.pool.fetchval("SELECT pg_backend_pid()")) - upid = str(Upid(node, pid, pid, timestamp, f"qm{operation}", vmid, "root@pam")) + upid = str( + Upid( + node, + pid, + pid, + timestamp, + f"qm{operation}", + vmid, + str(request.state.principal), + ) + ) task = await TaskRepository(database.pool).create( upid=upid, task_type=f"qemu-{operation}", @@ -84,6 +97,123 @@ def register_qemu_handlers(registry: HandlerRegistry) -> None: ) return task.upid + async def create(request: Request, inputs: dict[str, Any]) -> str: + values = _values(inputs) + node, vmid = str(values["node"]), int(values["vmid"]) + database = _database(request) + if not await database.pool.fetchval( + "SELECT EXISTS(SELECT 1 FROM nodes WHERE name=$1)", node + ): + raise ApiError(404, "node does not exist") + if await database.pool.fetchval( + """SELECT EXISTS(SELECT 1 FROM resources + WHERE external_id=$1 AND kind IN ('qemu','lxc'))""", + str(vmid), + ): + raise ApiError(409, "VMID already exists") + config = { + key: value + for key, value in values.items() + if key not in {"node", "vmid", "force", "archive", "start"} + } + return await _create_task( + request, + node=node, + vmid=str(vmid), + task_type="qemu-create", + payload={"node": node, "vmid": vmid, "config": config}, + ) + + async def update(request: Request, inputs: dict[str, Any], *, asynchronous: bool) -> str | None: + values = _values(inputs) + node, vmid = str(values["node"]), str(values["vmid"]) + database = _database(request) + row = await database.pool.fetchrow( + """SELECT r.id, r.version, r.state, v.config FROM resources r + JOIN nodes n ON n.id=r.node_id + JOIN virtual_machines v ON v.resource_id=r.id + WHERE n.name=$1 AND r.kind='qemu' AND r.external_id=$2""", + node, + vmid, + ) + if row is None: + raise ApiError(404, "virtual machine does not exist") + control = {"node", "vmid", "digest", "delete", "revert", "skiplock", "background_delay"} + provided = frozenset(str(item) for item in inputs.get("provided", values)) + changes = { + key: value for key, value in values.items() if key in provided and key not in control + } + delete = str(values.get("delete", "")) if "delete" in provided else "" + if asynchronous: + return await _create_task( + request, + node=node, + vmid=vmid, + task_type="qemu-update", + payload={ + "node": node, + "vmid": vmid, + "resource_id": str(row["id"]), + "changes": changes, + "delete": delete, + }, + ) + state = _state(row["state"]) + config = _state(row["config"]) + state.update(changes) + config.update(changes) + for key in delete.split(","): + if key: + state.pop(key, None) + config.pop(key, None) + status = await database.pool.execute( + """UPDATE resources SET state=$3::jsonb, version=version+1, + updated_at=now() WHERE id=$1 AND version=$2""", + row["id"], + row["version"], + json.dumps(state, sort_keys=True), + ) + if status != "UPDATE 1": + raise ApiError(409, "configuration changed concurrently") + await database.pool.execute( + """UPDATE virtual_machines SET config=$2::jsonb + WHERE resource_id=$1""", + row["id"], + json.dumps(config, sort_keys=True), + ) + return None + + async def update_async(request: Request, inputs: dict[str, Any]) -> str: + result = await update(request, inputs, asynchronous=True) + if not isinstance(result, str): + raise RuntimeError("async QEMU update did not create a task") + return result + + async def update_sync(request: Request, inputs: dict[str, Any]) -> None: + await update(request, inputs, asynchronous=False) + + async def delete(request: Request, inputs: dict[str, Any]) -> str: + values = _values(inputs) + node, vmid = str(values["node"]), str(values["vmid"]) + database = _database(request) + row = await database.pool.fetchrow( + """SELECT r.id, r.state FROM resources r JOIN nodes n ON n.id=r.node_id + WHERE n.name=$1 AND r.kind='qemu' AND r.external_id=$2""", + node, + vmid, + ) + if row is None: + raise ApiError(404, "virtual machine does not exist") + if str(_state(row["state"]).get("status")) != "stopped": + raise ApiError(409, "cannot delete a running virtual machine") + return await _create_task( + request, + node=node, + vmid=vmid, + task_type="qemu-delete", + payload={"node": node, "vmid": vmid, "resource_id": str(row["id"])}, + ) + async def start(request: Request, inputs: dict[str, Any]) -> str: return await mutate("start", request, inputs) @@ -124,10 +254,44 @@ def register_qemu_handlers(registry: HandlerRegistry) -> None: ] registry.register("/nodes/{node}/qemu", "GET", qemu_list) + registry.register("/nodes/{node}/qemu", "POST", create) + registry.register("/nodes/{node}/qemu/{vmid}", "DELETE", delete) registry.register("/nodes/{node}/qemu/{vmid}/config", "GET", qemu_config) + registry.register("/nodes/{node}/qemu/{vmid}/config", "POST", update_async) + registry.register("/nodes/{node}/qemu/{vmid}/config", "PUT", update_sync) registry.register("/nodes/{node}/qemu/{vmid}/status/current", "GET", qemu_status) registry.register("/nodes/{node}/qemu/{vmid}/status/start", "POST", start) registry.register("/nodes/{node}/qemu/{vmid}/status/stop", "POST", stop) registry.register("/nodes/{node}/tasks", "GET", task_list) registry.register("/nodes/{node}/tasks/{upid}/status", "GET", task_status) registry.register("/nodes/{node}/tasks/{upid}/log", "GET", task_log) + + +async def _create_task( + request: Request, + *, + node: str, + vmid: str, + task_type: str, + payload: dict[str, Any], +) -> str: + database = _database(request) + timestamp = int(await database.pool.fetchval("SELECT extract(epoch from now())::bigint")) + pid = int(await database.pool.fetchval("SELECT pg_backend_pid()")) + worker_type = { + "qemu-create": "qmcreate", + "qemu-delete": "qmdestroy", + "qemu-update": "qmconfig", + }[task_type] + upid = str(Upid(node, pid, pid, timestamp, worker_type, vmid, str(request.state.principal))) + try: + task = await TaskRepository(database.pool).create( + upid=upid, + task_type=task_type, + payload=payload, + resource_key=f"qemu:{vmid}", + idempotency_key=request.headers.get("Idempotency-Key"), + ) + except ConflictError as error: + raise ApiError(409, str(error)) from error + return task.upid diff --git a/app/main.py b/app/main.py index 72f4b3a..a06a27b 100644 --- a/app/main.py +++ b/app/main.py @@ -43,7 +43,13 @@ def create_app( return TaskWorker( repository, "simulator-worker", - {"qemu-start": handler, "qemu-stop": handler}, + { + "qemu-create": handler, + "qemu-delete": handler, + "qemu-start": handler, + "qemu-stop": handler, + "qemu-update": handler, + }, concurrency=resolved.task_worker_concurrency, lease_seconds=resolved.task_lease_seconds, ) diff --git a/app/security/acl.py b/app/security/acl.py index 6556384..5b7b128 100644 --- a/app/security/acl.py +++ b/app/security/acl.py @@ -57,17 +57,19 @@ def authorize( entries: tuple[AclEntry, ...], *, token_privileges: frozenset[str] | None = None, + require_all: bool = True, ) -> bool: privileges = effective_privileges(principal, path, entries) if token_privileges is not None: privileges &= token_privileges - return required <= privileges + return required <= privileges if require_all else bool(required & privileges) @dataclass(frozen=True, slots=True) class CapabilityRequirement: path: str privileges: frozenset[str] + require_all: bool = True def requirement_from_contract( @@ -84,4 +86,7 @@ def requirement_from_contract( raw_privileges = check[2] if not isinstance(raw_privileges, list): return None - return CapabilityRequirement(raw_path, frozenset(str(item) for item in raw_privileges)) + require_all = not (len(check) >= 4 and check[3] == "any") + return CapabilityRequirement( + raw_path, frozenset(str(item) for item in raw_privileges), require_all + ) diff --git a/app/tasks/qemu.py b/app/tasks/qemu.py index c03ce56..dac90ff 100644 --- a/app/tasks/qemu.py +++ b/app/tasks/qemu.py @@ -4,6 +4,8 @@ from __future__ import annotations import json import uuid +from collections.abc import Mapping +from typing import Any, cast from app.simulation.clock import Clock from app.simulation.transitions import VmState, plan_transition @@ -12,9 +14,15 @@ from app.tasks.worker import TaskHandler def qemu_handler(repository: TaskRepository, clock: Clock) -> TaskHandler: - async def execute(task: Task) -> dict[str, str]: + async def execute(task: Task) -> dict[str, Any]: operation = task.task_type.removeprefix("qemu-") + if operation == "create": + return await _create(repository, task) resource_id = uuid.UUID(str(task.payload["resource_id"])) + if operation == "update": + return await _update(repository, task, resource_id) + if operation == "delete": + return await _delete(repository, task, resource_id) async with repository.pool.acquire() as connection: row = await connection.fetchrow("SELECT state FROM resources WHERE id=$1", resource_id) if row is None: @@ -41,3 +49,85 @@ def qemu_handler(repository: TaskRepository, clock: Clock) -> TaskHandler: return {"status": str(transition.after)} return execute + + +async def _create(repository: TaskRepository, task: Task) -> dict[str, Any]: + node, vmid = str(task.payload["node"]), int(task.payload["vmid"]) + config = dict(task.payload.get("config", {})) + resource_id = uuid.uuid4() + state = {"status": "stopped", **config} + async with repository.pool.acquire() as connection: + async with connection.transaction(): + node_row = await connection.fetchrow( + "SELECT id, cluster_id FROM nodes WHERE name=$1", node + ) + if node_row is None: + raise ValueError("node disappeared") + await connection.execute( + """INSERT INTO resources( + id, node_id, cluster_id, kind, external_id, state, metadata + ) VALUES($1, $2, $3, 'qemu', $4, $5::jsonb, '{}'::jsonb)""", + resource_id, + node_row["id"], + node_row["cluster_id"], + str(vmid), + json.dumps(state, sort_keys=True), + ) + await connection.execute( + """INSERT INTO virtual_machines(resource_id, cluster_id, vmid, config) + VALUES($1, $2, $3, $4::jsonb)""", + resource_id, + node_row["cluster_id"], + vmid, + json.dumps(config, sort_keys=True), + ) + await repository.append_log(task.id, f"VM {vmid} created") + return {"vmid": vmid, "status": "stopped"} + + +async def _update(repository: TaskRepository, task: Task, resource_id: uuid.UUID) -> dict[str, Any]: + changes = dict(task.payload.get("changes", {})) + delete_keys = tuple(str(task.payload.get("delete", "")).split(",")) + async with repository.pool.acquire() as connection: + async with connection.transaction(): + row = await connection.fetchrow( + """SELECT r.state, v.config FROM resources r + JOIN virtual_machines v ON v.resource_id=r.id WHERE r.id=$1""", + resource_id, + ) + if row is None: + raise ValueError("resource disappeared") + state = _object(row["state"]) + config = _object(row["config"]) + config.update(changes) + for key in delete_keys: + if key: + config.pop(key, None) + state.pop(key, None) + state.update(changes) + await connection.execute( + """UPDATE resources SET state=$2::jsonb, version=version+1, + updated_at=now() WHERE id=$1""", + resource_id, + json.dumps(state, sort_keys=True), + ) + await connection.execute( + "UPDATE virtual_machines SET config=$2::jsonb WHERE resource_id=$1", + resource_id, + json.dumps(config, sort_keys=True), + ) + await repository.append_log(task.id, "VM configuration updated") + return {"updated": sorted(changes), "deleted": sorted(key for key in delete_keys if key)} + + +async def _delete(repository: TaskRepository, task: Task, resource_id: uuid.UUID) -> dict[str, Any]: + async with repository.pool.acquire() as connection: + status = await connection.execute("DELETE FROM resources WHERE id=$1", resource_id) + if status != "DELETE 1": + raise ValueError("resource disappeared") + await repository.append_log(task.id, "VM deleted") + return {"deleted": True} + + +def _object(value: object) -> dict[str, Any]: + return json.loads(value) if isinstance(value, str) else dict(cast(Mapping[str, Any], value)) diff --git a/docs/compatibility-0.1.0.md b/docs/compatibility-0.1.0.md index bf35bed..3fa5927 100644 --- a/docs/compatibility-0.1.0.md +++ b/docs/compatibility-0.1.0.md @@ -9,8 +9,8 @@ Proxmox compatibility. | Level | Methods | Contract share | Evidence | |---|---:|---:|---| | Declared and dynamically routed | 675 | 100% | Imported immutable API Viewer artifact | -| Stateful semantics implemented on current main | 18 | 2.67% | Handler registry and unit/integration tests | -| Schema-only or explicitly unsupported | 657 | 97.33% | Default 501 fallback | +| Stateful semantics implemented on current main | 22 | 3.26% | Handler registry and unit/integration tests | +| Schema-only or explicitly unsupported | 653 | 96.74% | Default 501 fallback | | proxmoxer smoke exercised | 9 | 1.33% | Unmodified proxmoxer 2.3 compatibility test | The smoke set is `POST /access/ticket`, `GET /version`, `GET /nodes`, diff --git a/docs/original-prompt-gap-plan.md b/docs/original-prompt-gap-plan.md index 0e6475c..ff9307b 100644 --- a/docs/original-prompt-gap-plan.md +++ b/docs/original-prompt-gap-plan.md @@ -55,7 +55,8 @@ plaintext password, ticket, CSRF token, or token secret reaches storage/logs. ## G4 — QEMU 0.2 verticals -- [ ] Create, update, delete, shutdown, reboot, reset, suspend and resume. +- [x] Create, synchronous/asynchronous update, and delete. +- [ ] Shutdown, reboot, reset, suspend and resume. - [ ] Snapshots and rollback, clone, local/remote migration, resize and move disk, selected agent endpoints, pending/status data. - [ ] Persist normalized CPU/memory/common fields plus unknown PVE parameters in diff --git a/evidence/pve-9.2.3-0.1.0.json b/evidence/pve-9.2.3-0.1.0.json index 1793bd7..b0a7288 100644 --- a/evidence/pve-9.2.3-0.1.0.json +++ b/evidence/pve-9.2.3-0.1.0.json @@ -50,6 +50,30 @@ "verb": "GET", "dimensions": ["input_parameters", "parameter_requiredness", "types_constraints", "http_status", "json_structure", "response_field_types", "response_required_fields", "long_task_behavior", "permissions"], "sources": ["tests/compatibility/test_proxmoxer.py", "tests/unit/test_task_worker.py", "tests/unit/test_upid.py"] + }, + { + "path": "/nodes/{node}/qemu", + "verb": "POST", + "dimensions": ["input_parameters", "parameter_requiredness", "types_constraints", "http_status", "json_structure", "state_semantics", "long_task_behavior", "errors_prohibitions", "permissions"], + "sources": ["tests/compatibility/test_proxmoxer.py", "tests/unit/test_qemu_handlers.py", "tests/unit/test_qemu_task.py"] + }, + { + "path": "/nodes/{node}/qemu/{vmid}", + "verb": "DELETE", + "dimensions": ["input_parameters", "parameter_requiredness", "types_constraints", "http_status", "json_structure", "state_semantics", "long_task_behavior", "errors_prohibitions", "permissions"], + "sources": ["tests/compatibility/test_proxmoxer.py", "tests/unit/test_qemu_handlers.py", "tests/unit/test_qemu_task.py"] + }, + { + "path": "/nodes/{node}/qemu/{vmid}/config", + "verb": "POST", + "dimensions": ["input_parameters", "parameter_requiredness", "types_constraints", "http_status", "json_structure", "state_semantics", "long_task_behavior", "errors_prohibitions", "permissions"], + "sources": ["tests/compatibility/test_proxmoxer.py", "tests/unit/test_qemu_handlers.py", "tests/unit/test_qemu_task.py", "tests/unit/test_acl.py"] + }, + { + "path": "/nodes/{node}/qemu/{vmid}/config", + "verb": "PUT", + "dimensions": ["input_parameters", "parameter_requiredness", "types_constraints", "http_status", "json_structure", "state_semantics", "errors_prohibitions", "permissions"], + "sources": ["tests/compatibility/test_proxmoxer.py", "tests/unit/test_qemu_handlers.py", "tests/unit/test_acl.py"] } ] } diff --git a/tests/compatibility/test_proxmoxer.py b/tests/compatibility/test_proxmoxer.py index 661e896..ea93217 100644 --- a/tests/compatibility/test_proxmoxer.py +++ b/tests/compatibility/test_proxmoxer.py @@ -2,6 +2,7 @@ import os from threading import Event +from typing import Any, cast import pytest from proxmoxer import ProxmoxAPI, ResourceException # type: ignore[import-untyped] @@ -12,6 +13,15 @@ pytestmark = [ ] +def wait_task(proxmox: Any, upid: str) -> dict[str, object]: + for _attempt in range(100): + task = proxmox.nodes("pve1").tasks(upid).status.get() + if task["status"] == "stopped": + return cast(dict[str, object], task) + Event().wait(0.05) + raise AssertionError("task did not finish") + + def test_proxmoxer_read_and_qemu_task_flow() -> None: proxmox = ProxmoxAPI( os.environ["PROXMOXER_HOST"], @@ -83,6 +93,29 @@ def test_proxmoxer_read_and_qemu_task_flow() -> None: storage_api.nodes("pve1").qemu(vmid).config.get() assert hidden.value.status_code == 403 + create_upid = proxmox.nodes("pve1").qemu.post( + vmid=150, name="created-by-proxmoxer", cores=2, memory=1024 + ) + with pytest.raises(ResourceException) as duplicate_create: + proxmox.nodes("pve1").qemu.post(vmid=150, name="duplicate") + assert duplicate_create.value.status_code == 409 + assert wait_task(proxmox, create_upid)["exitstatus"] == "OK" + created_config = proxmox.nodes("pve1").qemu("150").config.get() + assert created_config["name"] == "created-by-proxmoxer" + assert created_config["cores"] == 2 + + assert proxmox.nodes("pve1").qemu("150").config.put(name="sync-update", cores=3) is None + assert proxmox.nodes("pve1").qemu("150").config.get()["name"] == "sync-update" + update_upid = proxmox.nodes("pve1").qemu("150").config.post(name="async-update", memory=2048) + assert wait_task(proxmox, update_upid)["exitstatus"] == "OK" + assert proxmox.nodes("pve1").qemu("150").config.get()["name"] == "async-update" + + delete_upid = proxmox.nodes("pve1").qemu("150").delete() + assert wait_task(proxmox, delete_upid)["exitstatus"] == "OK" + with pytest.raises(ResourceException) as deleted_vm: + proxmox.nodes("pve1").qemu("150").config.get() + assert deleted_vm.value.status_code == 404 + if os.getenv("PROXMOXER_MUTATION_TEST") == "1": operator_api = ProxmoxAPI( os.environ["PROXMOXER_HOST"], @@ -96,10 +129,5 @@ def test_proxmoxer_read_and_qemu_task_flow() -> None: operation = "start" if status["status"] == "stopped" else "stop" endpoint = operator_api.nodes("pve1").qemu("101").status(operation) upid = endpoint.post() - for _attempt in range(100): - task = operator_api.nodes("pve1").tasks(upid).status.get() - if task["status"] == "stopped": - break - Event().wait(0.05) - assert task["status"] == "stopped" + task = wait_task(operator_api, upid) assert task["exitstatus"] == "OK" diff --git a/tests/unit/test_acl.py b/tests/unit/test_acl.py index 1ca87cc..d345662 100644 --- a/tests/unit/test_acl.py +++ b/tests/unit/test_acl.py @@ -43,3 +43,19 @@ def test_contract_permission_maps_to_capability_requirement() -> None: assert requirement is not None assert requirement.path == "/vms/100" assert requirement.privileges == frozenset({"VM.PowerMgmt"}) + + any_permission = Permissions( + expression={ + "check": ["perm", "/vms/{vmid}", ["VM.Config.CPU", "VM.Config.Memory"], "any", 1] + } + ) + any_requirement = requirement_from_contract(any_permission, {"vmid": "100"}) + assert any_requirement is not None + assert not any_requirement.require_all + assert authorize( + "alice@pve", + "/vms/100", + any_requirement.privileges, + (AclEntry("alice@pve", "/vms", frozenset({"VM.Config.CPU"})),), + require_all=any_requirement.require_all, + ) diff --git a/tests/unit/test_core_handlers.py b/tests/unit/test_core_handlers.py index bdd14a2..903e5df 100644 --- a/tests/unit/test_core_handlers.py +++ b/tests/unit/test_core_handlers.py @@ -29,7 +29,10 @@ class FakePool: "id": uuid.UUID("00000000-0000-0000-0000-000000000100"), "state": '{"name":"demo","status":"stopped"}', } - return {"state": '{"name":"demo","status":"stopped"}'} + return { + "config": '{"name":"demo"}', + "state": '{"name":"demo","status":"stopped"}', + } return None async def fetch(self, sql: str, *args: object) -> list[dict[str, object]]: diff --git a/tests/unit/test_qemu_handlers.py b/tests/unit/test_qemu_handlers.py new file mode 100644 index 0000000..a288009 --- /dev/null +++ b/tests/unit/test_qemu_handlers.py @@ -0,0 +1,187 @@ +"""Persistent QEMU CRUD semantic handler tests.""" + +import uuid +from typing import Any, cast + +import pytest +from fastapi import FastAPI, Request + +from app.api.errors import ApiError +from app.api.registry import HandlerRegistry +from app.db.pool import AsyncpgDatabase +from app.db.primitives import ConflictError +from app.handlers.qemu import register_qemu_handlers +from app.tasks.repository import Task + + +class QemuPool: + def __init__(self) -> None: + self.resource_exists = False + self.missing = False + self.running = False + self.commands: list[str] = [] + self.resource_id = uuid.uuid4() + + async def fetchval(self, sql: str, *args: object) -> bool | int: + del args + if "pg_backend_pid" in sql: + return 123 + if "extract(epoch" in sql: + return 1_700_000_000 + if "FROM nodes" in sql: + return True + if "FROM resources" in sql: + return self.resource_exists + raise AssertionError(sql) + + async def fetchrow(self, sql: str, *args: object) -> dict[str, object] | None: + del args + if self.missing: + return None + if "SELECT r.id, r.version" in sql: + return { + "id": self.resource_id, + "version": 1, + "state": '{"name":"old","status":"stopped"}', + "config": '{"name":"old"}', + } + if "SELECT r.id, r.state" in sql: + status = "running" if self.running else "stopped" + return {"id": self.resource_id, "state": f'{{"status":"{status}"}}'} + raise AssertionError(sql) + + async def execute(self, sql: str, *args: object) -> str: + del args + self.commands.append(sql) + return "UPDATE 1" + + +class FakeDatabase: + def __init__(self, pool: QemuPool) -> None: + self.pool = pool + + +def request(pool: QemuPool) -> Request: + app = FastAPI() + app.state.database = cast(AsyncpgDatabase, FakeDatabase(pool)) + result = Request( + { + "type": "http", + "app": app, + "method": "POST", + "path": "/", + "headers": [], + "query_string": b"", + "server": ("test", 80), + "client": ("test", 123), + "scheme": "http", + } + ) + result.state.principal = "root@pam" + return result + + +def inputs(**values: object) -> dict[str, Any]: + return {"values": values, "provided": tuple(values)} + + +async def test_qemu_create_sync_async_update_and_delete( + monkeypatch: pytest.MonkeyPatch, +) -> None: + created_payloads: list[dict[str, Any]] = [] + + class FakeTaskRepository: + def __init__(self, pool: object) -> None: + del pool + + async def create(self, **kwargs: Any) -> Task: + created_payloads.append(kwargs) + return Task( + uuid.uuid4(), + str(kwargs["upid"]), + str(kwargs["task_type"]), + "queued", + dict(kwargs["payload"]), + 0, + False, + 0, + ) + + monkeypatch.setattr("app.handlers.qemu.TaskRepository", FakeTaskRepository) + registry = HandlerRegistry() + register_qemu_handlers(registry) + pool = QemuPool() + http_request = request(pool) + create = registry.get("/nodes/{node}/qemu", "POST") + update_sync = registry.get("/nodes/{node}/qemu/{vmid}/config", "PUT") + update_async = registry.get("/nodes/{node}/qemu/{vmid}/config", "POST") + delete = registry.get("/nodes/{node}/qemu/{vmid}", "DELETE") + assert create and update_sync and update_async and delete + + create_upid = await create( + http_request, + inputs(node="pve1", vmid=150, name="new", cores=2), + ) + assert create_upid.startswith("UPID:pve1:") + assert created_payloads[-1]["task_type"] == "qemu-create" + + assert ( + await update_sync( + http_request, + inputs(node="pve1", vmid=150, name="sync", delete="unused"), + ) + is None + ) + assert len(pool.commands) == 2 + + update_upid = await update_async( + http_request, + inputs(node="pve1", vmid=150, memory="2048"), + ) + assert update_upid.startswith("UPID:pve1:") + assert created_payloads[-1]["task_type"] == "qemu-update" + + delete_upid = await delete(http_request, inputs(node="pve1", vmid=150)) + assert delete_upid.startswith("UPID:pve1:") + assert created_payloads[-1]["task_type"] == "qemu-delete" + + +async def test_qemu_crud_conflicts_and_missing_resources( + monkeypatch: pytest.MonkeyPatch, +) -> None: + class ConflictingRepository: + def __init__(self, pool: object) -> None: + del pool + + async def create(self, **_kwargs: object) -> Task: + raise ConflictError("resource is locked") + + monkeypatch.setattr("app.handlers.qemu.TaskRepository", ConflictingRepository) + registry = HandlerRegistry() + register_qemu_handlers(registry) + pool = QemuPool() + http_request = request(pool) + create = registry.get("/nodes/{node}/qemu", "POST") + update = registry.get("/nodes/{node}/qemu/{vmid}/config", "PUT") + delete = registry.get("/nodes/{node}/qemu/{vmid}", "DELETE") + assert create and update and delete + + with pytest.raises(ApiError) as locked: + await create(http_request, inputs(node="pve1", vmid=150)) + assert locked.value.status_code == 409 + + pool.resource_exists = True + with pytest.raises(ApiError) as duplicate: + await create(http_request, inputs(node="pve1", vmid=150)) + assert duplicate.value.status_code == 409 + + pool.missing = True + with pytest.raises(ApiError) as missing: + await update(http_request, inputs(node="pve1", vmid=150, name="missing")) + assert missing.value.status_code == 404 + + pool.missing = False + pool.running = True + with pytest.raises(ApiError) as running: + await delete(http_request, inputs(node="pve1", vmid=150)) + assert running.value.status_code == 409 diff --git a/tests/unit/test_qemu_task.py b/tests/unit/test_qemu_task.py index ce929d7..202f452 100644 --- a/tests/unit/test_qemu_task.py +++ b/tests/unit/test_qemu_task.py @@ -61,6 +61,45 @@ class Repository: self.logs.append(message) +class Transaction: + async def __aenter__(self) -> None: + return None + + async def __aexit__(self, *args: object) -> None: + return None + + +class CrudConnection: + def __init__(self) -> None: + self.commands: list[str] = [] + + def transaction(self) -> Transaction: + return Transaction() + + async def fetchrow(self, sql: str, *args: object) -> dict[str, object] | None: + del args + if "FROM nodes" in sql: + return {"id": uuid.uuid4(), "cluster_id": uuid.uuid4()} + if "JOIN virtual_machines" in sql: + return {"state": '{"status":"stopped","name":"old"}', "config": '{"name":"old"}'} + return None + + async def execute(self, sql: str, *args: object) -> str: + del args + self.commands.append(sql) + return "DELETE 1" if sql.startswith("DELETE") else "UPDATE 1" + + +class CrudRepository: + def __init__(self) -> None: + self.connection = CrudConnection() + self.pool = Pool(cast(Connection, self.connection)) + self.logs: list[str] = [] + + async def append_log(self, _task_id: uuid.UUID, message: str) -> None: + self.logs.append(message) + + async def test_qemu_worker_applies_intermediate_and_final_states() -> None: repository = Repository() task = Task( @@ -82,3 +121,57 @@ async def test_qemu_worker_applies_intermediate_and_final_states() -> None: assert '"starting"' in repository.connection.states[0] assert '"running"' in repository.connection.states[1] assert repository.logs == ["VM start started", "VM start completed"] + + +async def test_qemu_worker_create_update_and_delete_are_persistent() -> None: + repository = CrudRepository() + handler = qemu_handler(cast(TaskRepository, repository), cast(Clock, ImmediateClock())) + resource_id = uuid.uuid4() + + created = await handler( + Task( + uuid.uuid4(), + "UPID:create", + "qemu-create", + "running", + {"node": "pve1", "vmid": 150, "config": {"name": "new"}}, + 0, + False, + 1, + ) + ) + updated = await handler( + Task( + uuid.uuid4(), + "UPID:update", + "qemu-update", + "running", + { + "resource_id": str(resource_id), + "changes": {"name": "changed", "cores": 4}, + "delete": "unused", + }, + 0, + False, + 1, + ) + ) + deleted = await handler( + Task( + uuid.uuid4(), + "UPID:delete", + "qemu-delete", + "running", + {"resource_id": str(resource_id)}, + 0, + False, + 1, + ) + ) + + assert created == {"vmid": 150, "status": "stopped"} + assert updated == {"updated": ["cores", "name"], "deleted": ["unused"]} + assert deleted == {"deleted": True} + assert any("INSERT INTO resources" in command for command in repository.connection.commands) + assert any("UPDATE virtual_machines" in command for command in repository.connection.commands) + assert any("DELETE FROM resources" in command for command in repository.connection.commands)