From 32584c8aad9b1365ce684b6bd12ea87539bc891d Mon Sep 17 00:00:00 2001 From: David White Date: Fri, 23 Oct 2026 08:01:38 +0100 Subject: [PATCH] Type every Proxmox response, check the tree with ty, release 3.0.0 ApiResponse[T] is ApiSuccess[T] | ApiFailure. A success always carries its payload, validated with Pydantic against the endpoint's shape; a failure carries status, error and timed_out, and `if not res:` narrows to it. An answer missing a field arcane-mage relies on is a failure naming that field, so no caller reads a null or partial payload as data. Fields the shapes do not name are kept, which the Ansible status module returns as Proxmox sent them. Reading the payloads typed exposed four faults, fixed here: the TUI accepted a hypervisor whose connection timed out, `ping` could never report a rejected token, discovery counted a node whose VMs could not be listed as running none, and download_iso sent a Python bool for verify-certificates. ty checks src, tests and the Ansible collection, which needs ansible-core (its own dependency group) and pyrage-stubs. aioresponses is dropped: it cannot build a response under the current aiohttp, and the response tests run against aiohttp's own test server. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 14 +- CONTRIBUTING.md | 3 +- README.md | 13 +- .../arcane_mage/plugins/modules/iso_info.py | 9 +- pyproject.toml | 11 +- src/arcane_mage/__init__.py | 28 +- src/arcane_mage/__main__.py | 36 +-- src/arcane_mage/arcane_mage.py | 22 +- src/arcane_mage/helpers.py | 15 +- src/arcane_mage/models/network.py | 2 +- src/arcane_mage/provisioner.py | 78 +++--- src/arcane_mage/proxmox.py | 263 +++++++++++++----- src/arcane_mage/screens/hypervisor.py | 37 ++- src/arcane_mage/screens/provision_node.py | 25 +- src/arcane_mage/screens/welcome_proxmox.py | 63 +++-- tests/test_models/test_fluxnode.py | 2 +- tests/test_models/test_hypervisor.py | 13 +- tests/test_models/test_installer.py | 2 +- tests/test_provisioner.py | 59 ++-- tests/test_proxmox.py | 116 ++++++-- uv.lock | 61 ++-- 21 files changed, 568 insertions(+), 304 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6c59d4a..3ebfb3a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,10 +1,22 @@ # Changelog -## Unreleased +## 3.0.0 + +### Breaking Changes + +- **`ApiResponse`** is the union `ApiSuccess[T] | ApiFailure`. `ApiSuccess` always carries its `payload`, typed per endpoint (for example `ApiResponse[list[VmSummary]]` from `get_vms`); `ApiFailure` carries `status`, `error` and `timed_out`. `if not res:` narrows to `ApiFailure`. Code that constructs `ApiResponse(...)` constructs `ApiSuccess(payload=...)` or `ApiFailure(...)` instead. +- **Response validation**: each endpoint's answer is validated against its shape (`VmSummary`, `NetworkIface`, `NodeStorage`, `StorageVolume`, `NodeVersion`, `TaskStatus`, `HypervisorNode`, `StorageConfig`, all exported). An answer missing a required field, or with `data: null` where a value is due, is an `ApiFailure` naming the field. Fields the shapes do not name are kept. ### Changed - **Build backend**: the package builds with `uv_build`. The wheel's contents are unchanged. +- **Discovery**: a hypervisor node whose VMs cannot be listed is left out of `discover_nodes`, rather than treated as running none. +- **`download_iso`** sends `verify-certificates` as `0`/`1`. The Ansible `iso_info` module downloads through it. + +### Fixed + +- **TUI**: adding a hypervisor whose connection times out reports the timeout instead of validating it. +- **CLI**: `arcane-mage ping` reports "Authentication failed" for a rejected token. ## 2.3.0 diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 278b8be..35973da 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -13,10 +13,11 @@ This installs the library, the CLI and TUI extras, and the dev tools. ```bash uv run ruff check uv run ruff format +uv run ty check uv run pytest ``` -All three must be clean. +All four must be clean. ## Pull requests diff --git a/README.md b/README.md index 7ebb5c3..78b341b 100644 --- a/README.md +++ b/README.md @@ -89,10 +89,19 @@ if creds: # Use as an async context manager async with api: - version = await api.get_api_version("pve1") - vms = await api.get_vms("pve1") + res = await api.get_vms("pve1") + + # A failed call is an ApiFailure (status, error, timed_out); past this check + # it is an ApiSuccess whose payload has the endpoint's type, here list[VmSummary]. + if not res: + print(f"Unable to list VMs: {res.error}") + else: + for vm in res.payload: + print(vm["vmid"], vm.get("name"), vm["status"]) ``` +Every call validates Proxmox's answer against the endpoint's shape. An answer missing a field arcane-mage relies on comes back as an `ApiFailure` saying which field; fields arcane-mage does not name are kept as Proxmox sent them. + #### Provisioning Nodes ```python diff --git a/ansible_collection/runonflux/arcane_mage/plugins/modules/iso_info.py b/ansible_collection/runonflux/arcane_mage/plugins/modules/iso_info.py index 58f55de..7fce05c 100644 --- a/ansible_collection/runonflux/arcane_mage/plugins/modules/iso_info.py +++ b/ansible_collection/runonflux/arcane_mage/plugins/modules/iso_info.py @@ -117,19 +117,14 @@ async def run_iso_info(params): if existing: return True, iso_name, False, "" - # Download via Proxmox API (direct POST without verify-certificates param) # URL format: https://images.runonflux.io/arcane/releases//FluxLive-.iso build_id = iso_name.replace("FluxLive-", "").replace(".iso", "") download_url = f"https://images.runonflux.io/arcane/releases/{build_id}/{iso_name}" - endpoint = f"nodes/{params['node']}/storage/{params['storage_iso']}/download-url" - data = {"content": "iso", "filename": iso_name, "url": download_url} - - res = await api._do_post(endpoint, data=data) + res = await api.download_iso(params["node"], download_url, iso_name, params["storage_iso"]) if not res: - error_detail = f"status={res.status} error={res.error}" if res else "no response" - return False, iso_name, False, f"Failed to initiate ISO download: {error_detail}" + return False, iso_name, False, f"Failed to initiate ISO download: status={res.status} error={res.error}" # Wait for download task ok = await api.wait_for_task(res.payload, params["node"], max_wait_s=600) diff --git a/pyproject.toml b/pyproject.toml index ed8e924..75b576a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "arcane-mage" -version = "2.3.0" +version = "3.0.0" description = "Automation helper for Fluxnodes" readme = "README.md" authors = [ @@ -39,8 +39,11 @@ requires = ["uv_build>=0.13.3,<0.14"] build-backend = "uv_build" [dependency-groups] +ansible = [ + "ansible-core>=2.21.5", +] dev = [ - "aioresponses>=0.7.8", + "pyrage-stubs>=1.4.0", "pytest>=9.0.2", "pytest-asyncio>=1.3.0", "ruff>=0.17.2", @@ -50,6 +53,10 @@ dev = [ "types-pyyaml>=6.0.12.20250915", ] +[tool.uv] +# The Ansible collection runs inside Ansible; ansible-core is here so ty can check it. +default-groups = ["dev", "ansible"] + [tool.ruff] target-version = "py313" line-length = 120 diff --git a/src/arcane_mage/__init__.py b/src/arcane_mage/__init__.py index 821f4ff..b8b9798 100644 --- a/src/arcane_mage/__init__.py +++ b/src/arcane_mage/__init__.py @@ -20,12 +20,30 @@ ) from .password import HashedPassword from .provisioner import TIER_CONFIG, HypervisorDiscovery, Provisioner, get_latest_iso_version -from .proxmox import ApiResponse, ParsedToken, ParsedUserPass, ProxmoxApi, ResolvedConnection +from .proxmox import ( + ApiFailure, + ApiResponse, + ApiSuccess, + HypervisorNode, + NetworkIface, + NodeStorage, + NodeVersion, + ParsedToken, + ParsedUserPass, + ProxmoxApi, + ResolvedConnection, + StorageConfig, + StorageVolume, + TaskStatus, + VmSummary, +) __all__ = [ "TIER_CONFIG", "AddressConfig", + "ApiFailure", "ApiResponse", + "ApiSuccess", "ArcaneCreatorConfig", "ArcaneOsConfig", "ArcaneOsConfigGroup", @@ -36,17 +54,25 @@ "Hypervisor", "HypervisorConfig", "HypervisorDiscovery", + "HypervisorNode", "Identifier", "Identity", "InstallerConfig", "MetricsAppConfig", "NetworkConfig", + "NetworkIface", + "NodeStorage", + "NodeVersion", "Notifications", "ParsedToken", "ParsedUserPass", "Provisioner", "ProxmoxApi", "ResolvedConnection", + "StorageConfig", + "StorageVolume", "SystemConfig", + "TaskStatus", + "VmSummary", "get_latest_iso_version", ] diff --git a/src/arcane_mage/__main__.py b/src/arcane_mage/__main__.py index 0b52fb8..a5ff10a 100644 --- a/src/arcane_mage/__main__.py +++ b/src/arcane_mage/__main__.py @@ -14,6 +14,7 @@ import sys from importlib.metadata import version as get_version from pathlib import Path +from typing import TypedDict from urllib.parse import urlparse from rich.console import Console @@ -21,7 +22,7 @@ from rich.table import Table from .models import ArcaneCreatorConfig, ArcaneOsConfigGroup -from .proxmox import ProxmoxApi, ResolvedConnection +from .proxmox import ProxmoxApi, ResolvedConnection, VmSummary app = typer.Typer( name="arcane-mage", @@ -96,8 +97,8 @@ async def _backfill_hypervisor_names(creator_config: ArcaneCreatorConfig) -> Non async with api: res = await api.get_hypervisor_nodes() - if res and isinstance(res.payload, list): - names = [n.get("node", "") for n in res.payload if n.get("node")] + if res: + names = [n["node"] for n in res.payload] if names: hyper.name = names[0] if len(names) == 1 else ",".join(sorted(names)) dirty = True @@ -171,6 +172,12 @@ def _resolve_connection( return ResolvedConnection(resolved_url, parsed) +class _Check(TypedDict): + check: str + ok: bool + message: str + + def _load_configs(config: str) -> ArcaneOsConfigGroup: configs = ArcaneOsConfigGroup.from_fs(Path(config)) if not configs: @@ -310,9 +317,6 @@ def deprovision( if use_json and not force: raise CliError("--json requires --force for destructive operations") - if not direct_vm and not config: - config = "fluxnodes.yaml" - if direct_vm and config: raise CliError("Cannot specify --config with --vm-name or --vm-id") @@ -330,21 +334,21 @@ async def run_direct(): provisioner = Provisioner(api) res = await api.get_hypervisor_nodes() - if not res or not isinstance(res.payload, list): + if not res: raise CliError("Unable to discover hypervisor nodes") - hv_nodes = [n.get("node") for n in res.payload if n.get("node")] + hv_nodes = [n["node"] for n in res.payload] if node_filter: hv_nodes = [n for n in hv_nodes if n == node_filter] - found_vm: dict | None = None + found_vm: VmSummary | None = None found_node: str | None = None for hv_node in hv_nodes: vms_res = await api.get_vms(hv_node) if not vms_res: continue for v in vms_res.payload: - if vm_id is not None and v.get("vmid") == vm_id: + if vm_id is not None and v["vmid"] == vm_id: found_vm = v found_node = hv_node break @@ -411,7 +415,7 @@ def callback(ok: bool, msg: str, steps: list[dict[str, object]] = steps): return try: - configs = _load_configs(config) + configs = _load_configs(config or "fluxnodes.yaml") except CliError as e: _handle_error(e, use_json) @@ -534,7 +538,7 @@ async def run(): all_ok = False continue - checks = [] + checks: list[_Check] = [] api_ok, api_msg = await provisioner.validate_api_version(hyper.node) checks.append( @@ -630,17 +634,17 @@ async def run(): res = await api.get_hypervisor_nodes() if not res: - if res and res.unauthorized: + if res.unauthorized: raise CliError("Authentication failed") raise CliError(f"Unable to connect to {conn.url}") - nodes = [n.get("node", "unknown") for n in res.payload] if isinstance(res.payload, list) else [] + nodes = [n["node"] for n in res.payload] version = None if nodes: ver_res = await api.get_api_version(nodes[0]) - if ver_res and ver_res.payload: - version = ver_res.payload.get("version") + if ver_res: + version = ver_res.payload["version"] data = { "url": conn.url, diff --git a/src/arcane_mage/arcane_mage.py b/src/arcane_mage/arcane_mage.py index c96c161..581ab06 100644 --- a/src/arcane_mage/arcane_mage.py +++ b/src/arcane_mage/arcane_mage.py @@ -1,5 +1,6 @@ from __future__ import annotations +from collections.abc import Callable from functools import partial from pathlib import Path from typing import ClassVar @@ -29,7 +30,7 @@ class ArcaneMage(App): CSS_PATH = "arcane_mage.tcss" AUTO_FOCUS = None - SCREENS: ClassVar[dict[str, type[Screen]]] = { + SCREENS: ClassVar[dict[str, Callable[[], Screen]]] = { "welcome": WelcomeScreen, } @@ -64,7 +65,7 @@ def hypervisor_callback(self, hypervisor: HypervisorConfig | None) -> None: screen = self.get_screen("welcome-proxmox", WelcomeScreenProxmox) screen.validate_hypervisors(hypervisor) - def del_hypervisor_callback(self, hypervisor: HypervisorConfig, delete: bool) -> None: + def del_hypervisor_callback(self, hypervisor: HypervisorConfig, delete: bool | None) -> None: if not delete: return @@ -150,9 +151,12 @@ def on_del_hypervisor(self, event: WelcomeScreenProxmox.DelHypervisor) -> None: def on_provision_node(self, event: WelcomeScreenProxmox.ProvisionNode) -> None: self.push_screen( ProvisionNodeScreen(event.fluxnode), - self.provision_node_callback, + self.provision_configured_node, ) + async def provision_configured_node(self, fluxnode: ArcaneOsConfig | None) -> None: + await self.provision_node_callback(fluxnode) + @on(WelcomeScreenProxmox.ProvisionAllNodes) def on_provision_all_nodes(self, event: WelcomeScreenProxmox.ProvisionAllNodes) -> None: self.provision_all_nodes(event.delay) @@ -188,8 +192,10 @@ async def provision_all_nodes(self, delay: int) -> None: needs_pop = len(provisionable_nodes) > 1 - if delay and configured_node != provisionable_nodes.last: - worker = self.screen.show_delay(delay) + info_screen = self.screen + + if delay and configured_node != provisionable_nodes.last and isinstance(info_screen, ProvisioningInfoScreen): + worker = info_screen.show_delay(delay) try: await worker.wait() @@ -218,8 +224,10 @@ async def provision_all_nodes(self, delay: int) -> None: if not result: break - if delay and not last: - worker = self.screen.show_delay(delay) + info_screen = self.screen + + if delay and not last and isinstance(info_screen, ProvisioningInfoScreen): + worker = info_screen.show_delay(delay) try: await worker.wait() diff --git a/src/arcane_mage/helpers.py b/src/arcane_mage/helpers.py index 66b6462..25ef8fb 100644 --- a/src/arcane_mage/helpers.py +++ b/src/arcane_mage/helpers.py @@ -82,10 +82,7 @@ async def exec_binary( if len(cmd) == 0: raise ChildProcessError("Cannot execute empty cmd") - extra_params = {} - - if cwd is not None: - extra_params["cwd"] = cwd + preexec_fn: Callable[[], None] | None = None if user is not None: try: @@ -93,17 +90,16 @@ async def exec_binary( except KeyError as e: raise ChildProcessError("User not found") from e - extra_params["preexec_fn"] = demote(pw_record.pw_uid, pw_record.pw_gid) - - if env: - extra_params["env"] = env + preexec_fn = demote(pw_record.pw_uid, pw_record.pw_gid) try: proc = await asyncio.create_subprocess_exec( *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, - **extra_params, + cwd=cwd, + preexec_fn=preexec_fn, + env=env or None, ) except FileNotFoundError as e: raise ChildProcessError(f"Binary not found: {cmd[0]}, is it installed?") from e @@ -169,7 +165,6 @@ async def do_http( json=data, ssl=verify_ssl, ) as resp: - resp: ClientResponse # type: ignore if resp.status in [429, 500, 502, 503, 504]: log.debug("bad response: %s", resp.status) if remaining or not max_tries: diff --git a/src/arcane_mage/models/network.py b/src/arcane_mage/models/network.py index de77918..7a71d44 100644 --- a/src/arcane_mage/models/network.py +++ b/src/arcane_mage/models/network.py @@ -197,7 +197,7 @@ class NetworkConfig: def from_dict(cls, data: dict) -> NetworkConfig: ip_allocation: str = data.get("ip_allocation", "") vlan: int | None = data.get("vlan") - rate_limit: int | None = data.get("rate_limit") + rate_limit: Literal[35, 75, 135, 250] | None = data.get("rate_limit") if ip_allocation not in ("dhcp", "static"): raise ValueError(f"Invalid ip allocation: {ip_allocation}") diff --git a/src/arcane_mage/provisioner.py b/src/arcane_mage/provisioner.py index b7e55bd..5ade0dc 100644 --- a/src/arcane_mage/provisioner.py +++ b/src/arcane_mage/provisioner.py @@ -14,7 +14,7 @@ from .fat_writer import FAT12Writer from .helpers import do_http from .models import ArcaneOsConfig, ArcaneOsConfigGroup, HypervisorConfig -from .proxmox import ProxmoxApi +from .proxmox import ApiResponse, HypervisorNode, ProxmoxApi, VmSummary log = logging.getLogger(__name__) @@ -108,7 +108,7 @@ class HypervisorDiscovery: """Result of discovering nodes and their provisioned VMs on a hypervisor.""" nodes: ArcaneOsConfigGroup - provisioned_vms: dict[str, list[dict]] + provisioned_vms: dict[str, list[VmSummary]] class Provisioner: @@ -150,11 +150,14 @@ async def discover_nodes(self, all_configs: ArcaneOsConfigGroup) -> HypervisorDi return None useable_nodes = ArcaneOsConfigGroup() - provisioned: dict[str, list[dict]] = {} - - async def handle_node(node: dict) -> None: - if name := node.get("node"): - vm_res = await self.api.get_vms(name) + provisioned: dict[str, list[VmSummary]] = {} + + # A node whose VMs cannot be listed is left out: its configs cannot be + # checked against what it already runs. + async def handle_node(node: HypervisorNode) -> None: + name = node["node"] + vm_res = await self.api.get_vms(name) + if vm_res: provisioned[name] = vm_res.payload useable_nodes.add_nodes(all_configs.get_nodes_by_hypervisor_name(name)) @@ -169,10 +172,7 @@ async def validate_api_version(self, node: str) -> tuple[bool, str]: if not res: return False, "Unable to get Proxmox api version" - version = res.payload.get("version") - - if not version: - return False, "Api payload missing version info" + version = res.payload["version"] if not is_api_min_version(version): return False, f"Api version too old. Got: {version}, Want: {MIN_API_VERSION}" @@ -195,18 +195,19 @@ async def validate_storage( if not res.payload: return False, "No Storage state available, did you forget API permissions?" - node_storage_iso = next(filter(lambda x: x.get("storage") == storage_iso, res.payload), None) - node_storage_images = next(filter(lambda x: x.get("storage") == storage_images, res.payload), None) - node_storage_import = next(filter(lambda x: x.get("storage") == storage_import, res.payload), None) + by_name = {x["storage"]: x for x in res.payload} + node_storage_iso = by_name.get(storage_iso) + node_storage_images = by_name.get(storage_images) + node_storage_import = by_name.get(storage_import) - if not all([node_storage_iso, node_storage_images, node_storage_import]): + if not node_storage_iso or not node_storage_images or not node_storage_import: return False, "Missing storage config item" - iso_content = node_storage_iso.get("content") - images_content = node_storage_images.get("content") - import_content = node_storage_import.get("content") - - if "iso" not in iso_content or "images" not in images_content or "import" not in import_content: + if ( + "iso" not in node_storage_iso["content"] + or "images" not in node_storage_images["content"] + or "import" not in node_storage_import["content"] + ): return False, "Storage type missing on hypervisor" import_available = node_storage_import.get("avail", 0) @@ -230,15 +231,7 @@ async def validate_iso_version(self, node: str, iso_name: str, storage_iso: str) if not res: return False - iso_exists = next( - filter( - lambda x: x.get("content") == "iso" and x.get("volid", "").endswith(iso_name), - res.payload, - ), - None, - ) - - return bool(iso_exists) + return any(x["content"] == "iso" and x["volid"].endswith(iso_name) for x in res.payload) async def validate_network(self, node: str, network: str) -> bool: """Validate that the specified network bridge exists on the hypervisor.""" @@ -247,9 +240,7 @@ async def validate_network(self, node: str, network: str) -> bool: if not res: return False - network_exists = next(filter(lambda x: x.get("iface") == network, res.payload), None) - - return bool(network_exists) + return any(x["iface"] == network for x in res.payload) async def stop_vm(self, vm_id: int, node: str) -> bool: """Stop a VM and wait for the task to complete.""" @@ -347,7 +338,7 @@ async def deprovision_node( return await self._stop_and_delete_vm( vm["vmid"], hyper.vm_name, - vm.get("status", "unknown"), + vm["status"], hyper.node, callback, ) @@ -370,9 +361,9 @@ async def deprovision_vm( callback(False, f"Unable to list VMs on {node}") return False - vm: dict | None = None + vm: VmSummary | None = None if vm_id is not None: - vm = next((v for v in vms_res.payload if v.get("vmid") == vm_id), None) + vm = next((v for v in vms_res.payload if v["vmid"] == vm_id), None) if not vm: if callback: callback(False, f"VM id={vm_id} not found on {node}") @@ -390,7 +381,7 @@ async def deprovision_vm( return await self._stop_and_delete_vm( vm["vmid"], vm.get("name", str(vm["vmid"])), - vm.get("status", "unknown"), + vm["status"], node, callback, ) @@ -419,22 +410,17 @@ async def delete_install_disks(self, vm_id: int, node: str, storage: str, delete efi_file = "arcane_efi.raw" config_file = f"{vm_id}_arcane_config.raw" + efi_res: ApiResponse[str] | None = None + if delete_efi: efi_res = await self.api.delete_file(efi_file, node, storage, content="import") - else: - efi_res = True config_res = await self.api.delete_file(config_file, node, storage, content="import") - if not efi_res or not config_res: + if efi_res is not None and (not efi_res or not await self.api.wait_for_task(efi_res.payload, node)): return False - if delete_efi: - efi_ok = await self.api.wait_for_task(efi_res.payload, node) - else: - efi_ok = True - - if not efi_ok: + if not config_res: return False return await self.api.wait_for_task(config_res.payload, node) @@ -511,8 +497,6 @@ async def create_vm_config( vm_id = vm_id_res.payload - assert vm_id - disk_rate = f"mbps_rd={disk_limit},mbps_wr={disk_limit}," if disk_limit else "" network_rate = f",rate={network_limit}" if network_limit else "" cpu_limit = cpu_limit or 0 diff --git a/src/arcane_mage/proxmox.py b/src/arcane_mage/proxmox.py index 0167884..2de7630 100644 --- a/src/arcane_mage/proxmox.py +++ b/src/arcane_mage/proxmox.py @@ -5,14 +5,14 @@ import json import types import urllib.parse -from collections.abc import AsyncIterator +from collections.abc import AsyncGenerator from contextlib import asynccontextmanager from dataclasses import dataclass from io import BufferedReader from pathlib import Path from socket import AF_INET from time import monotonic -from typing import Literal +from typing import Literal, NotRequired, TypedDict import aiohttp from aiohttp import ( @@ -23,6 +23,7 @@ ContentTypeError, TCPConnector, ) +from pydantic import ConfigDict, TypeAdapter, ValidationError, with_config # Concrete type for HTTP request data passed to aiohttp methods type HttpData = dict | aiohttp.MultipartWriter | None @@ -63,21 +64,125 @@ class ResolvedConnection: token: ParsedToken -@dataclass -class ApiResponse: - """Response wrapper for Proxmox API calls, with status, payload, and error info.""" +@dataclass(frozen=True) +class ApiSuccess[T]: + """A successful Proxmox API call: its `data`, validated against the endpoint's shape.""" + + payload: T + status: int = 200 + + @property + def unauthorized(self) -> Literal[False]: + return False + + def __bool__(self) -> Literal[True]: + return True + + +@dataclass(frozen=True) +class ApiFailure: + """A failed Proxmox API call: refused, unreachable, timed out, or answered with data + that does not match the endpoint's shape.""" status: int | None = None - payload: dict | list | str | None = None - timed_out: bool = False error: str | None = None + timed_out: bool = False @property def unauthorized(self) -> bool: return self.status == 401 - def __bool__(self) -> bool: - return self.status == 200 and not self.error + def __bool__(self) -> Literal[False]: + return False + + +# `if not res:` narrows to ApiFailure; past it, `res.payload` has the endpoint's type. +type ApiResponse[T] = ApiSuccess[T] | ApiFailure + +# Payload shapes name the fields arcane-mage reads; every other field Proxmox sends is kept. +_KEEP_ALL_FIELDS = ConfigDict(extra="allow") + + +@with_config(_KEEP_ALL_FIELDS) +class HypervisorNode(TypedDict): + node: str + status: NotRequired[str] + + +@with_config(_KEEP_ALL_FIELDS) +class ClusterConfigNode(TypedDict): + node: str + + +@with_config(_KEEP_ALL_FIELDS) +class NodeVersion(TypedDict): + version: str + release: NotRequired[str] + repoid: NotRequired[str] + + +@with_config(_KEEP_ALL_FIELDS) +class NodeStorage(TypedDict): + storage: str + content: str + avail: NotRequired[int] + total: NotRequired[int] + used: NotRequired[int] + + +@with_config(_KEEP_ALL_FIELDS) +class StorageConfig(TypedDict): + storage: str + type: str + content: NotRequired[str] + shared: NotRequired[int] + nodes: NotRequired[str] + + +@with_config(_KEEP_ALL_FIELDS) +class StorageVolume(TypedDict): + volid: str + content: str + + +@with_config(_KEEP_ALL_FIELDS) +class NetworkIface(TypedDict): + iface: str + type: NotRequired[str] + + +@with_config(_KEEP_ALL_FIELDS) +class VmSummary(TypedDict): + vmid: int + status: str + name: NotRequired[str] + cpu: NotRequired[float] + mem: NotRequired[int] + maxmem: NotRequired[int] + disk: NotRequired[int] + maxdisk: NotRequired[int] + + +@with_config(_KEEP_ALL_FIELDS) +class TaskStatus(TypedDict): + status: str + exitstatus: NotRequired[str] + + +# A task-starting call answers with the task's UPID. +_UPID = TypeAdapter(str) +_NONE = TypeAdapter(type(None)) +_HYPERVISOR_NODES = TypeAdapter(list[HypervisorNode]) +_CLUSTER_CONFIG_NODES = TypeAdapter(list[ClusterConfigNode]) +_NODE_VERSION = TypeAdapter(NodeVersion) +_NODE_STORAGE = TypeAdapter(list[NodeStorage]) +_STORAGE_CONFIG = TypeAdapter(list[StorageConfig]) +_STORAGE_VOLUMES = TypeAdapter(list[StorageVolume]) +_NETWORK_IFACES = TypeAdapter(list[NetworkIface]) +_VMS = TypeAdapter(list[VmSummary]) +_VM_CONFIG = TypeAdapter(dict[str, str | int | float]) +_TASK_STATUS = TypeAdapter(TaskStatus) +_NEXT_ID = TypeAdapter(int) class ProxmoxApi: @@ -203,23 +308,31 @@ def build_token_client(url: str, user: str, token_name: str, token_value: str) - ) @staticmethod - async def handle_api_response( + async def handle_api_response[T]( response: aiohttp.ClientResponse, - ) -> ApiResponse: + shape: TypeAdapter[T], + ) -> ApiResponse[T]: try: response.raise_for_status() except ClientResponseError as e: - return ApiResponse(status=e.status, error=e.message) + return ApiFailure(status=e.status, error=e.message) try: - payload: dict = await response.json() + body: dict = await response.json() except (json.JSONDecodeError, ContentTypeError) as e: - return ApiResponse(status=response.status, error=str(e)) + return ApiFailure(status=response.status, error=str(e)) - data = payload.get("data") - message = payload.get("message") + if message := body.get("message"): + return ApiFailure(status=response.status, error=message) + + try: + payload = shape.validate_python(body.get("data")) + except ValidationError as e: + first = e.errors()[0] + where = ".".join(str(part) for part in first["loc"]) or "data" + return ApiFailure(status=response.status, error=f"Unexpected response at {where}: {first['msg']}") - return ApiResponse(status=response.status, payload=data, error=message) + return ApiSuccess(payload=payload, status=response.status) def __init__( self, @@ -244,7 +357,7 @@ async def __aexit__( await self.close() @asynccontextmanager - async def session(self) -> AsyncIterator[None]: + async def session(self) -> AsyncGenerator[None]: try: yield finally: @@ -253,121 +366,122 @@ async def session(self) -> AsyncIterator[None]: async def close(self) -> None: await self.client.close() - async def _do_http( + async def _do_http[T]( self, verb: Literal["get", "put", "post", "delete"], path: str, + shape: TypeAdapter[T], data: HttpData = None, - ) -> ApiResponse: + ) -> ApiResponse[T]: method = getattr(self.client, verb) try: res = await method(path, ssl=self.verify_ssl, data=data) except ConnectionTimeoutError: - return ApiResponse(timed_out=True) + return ApiFailure(timed_out=True) except ClientError: - return ApiResponse(error="Connection Error") + return ApiFailure(error="Connection Error") - return await self.handle_api_response(res) + return await self.handle_api_response(res, shape) - async def _do_get(self, path: str) -> ApiResponse: - return await self._do_http("get", path) + async def _do_get[T](self, path: str, shape: TypeAdapter[T]) -> ApiResponse[T]: + return await self._do_http("get", path, shape) - async def _do_post(self, path: str, data: HttpData = None) -> ApiResponse: - return await self._do_http("post", path, data) + async def _do_post[T](self, path: str, shape: TypeAdapter[T], data: HttpData = None) -> ApiResponse[T]: + return await self._do_http("post", path, shape, data) - async def _do_put(self, path: str, data: HttpData = None) -> ApiResponse: - return await self._do_http("put", path, data) + async def _do_put[T](self, path: str, shape: TypeAdapter[T], data: HttpData = None) -> ApiResponse[T]: + return await self._do_http("put", path, shape, data) - async def _do_delete(self, path: str) -> ApiResponse: - return await self._do_http("delete", path) + async def _do_delete[T](self, path: str, shape: TypeAdapter[T]) -> ApiResponse[T]: + return await self._do_http("delete", path, shape) - async def get_api_version(self, node: str) -> ApiResponse: - res = await self._do_get(f"nodes/{node}/version") + async def get_api_version(self, node: str) -> ApiResponse[NodeVersion]: + res = await self._do_get(f"nodes/{node}/version", _NODE_VERSION) return res - async def get_cluster_nodes(self) -> ApiResponse: - res = await self._do_get("cluster/config/nodes") + async def get_cluster_nodes(self) -> ApiResponse[list[ClusterConfigNode]]: + res = await self._do_get("cluster/config/nodes", _CLUSTER_CONFIG_NODES) return res - async def get_hypervisor_nodes(self) -> ApiResponse: - res = await self._do_get("nodes") + async def get_hypervisor_nodes(self) -> ApiResponse[list[HypervisorNode]]: + res = await self._do_get("nodes", _HYPERVISOR_NODES) return res - async def get_networks(self, node: str) -> ApiResponse: + async def get_networks(self, node: str) -> ApiResponse[list[NetworkIface]]: endpoint = f"nodes/{node}/network" - res = await self._do_get(endpoint) + res = await self._do_get(endpoint, _NETWORK_IFACES) return res - async def get_storage_state(self, node: str) -> ApiResponse: + async def get_storage_state(self, node: str) -> ApiResponse[list[NodeStorage]]: endpoint = f"nodes/{node}/storage" - res = await self._do_get(endpoint) + res = await self._do_get(endpoint, _NODE_STORAGE) return res - async def get_storage_content(self, node: str, location: str) -> ApiResponse: + async def get_storage_content(self, node: str, location: str) -> ApiResponse[list[StorageVolume]]: endpoint = f"nodes/{node}/storage/{location}/content" - res = await self._do_get(endpoint) + res = await self._do_get(endpoint, _STORAGE_VOLUMES) return res - async def get_storage_config(self) -> ApiResponse: - res = await self._do_get("storage") + async def get_storage_config(self) -> ApiResponse[list[StorageConfig]]: + res = await self._do_get("storage", _STORAGE_CONFIG) return res - async def set_storage_content(self, storage_id: str, content: str) -> ApiResponse: + async def set_storage_content(self, storage_id: str, content: str) -> ApiResponse[None]: endpoint = f"storage/{storage_id}" data = {"content": content} - res = await self._do_put(endpoint, data=data) + res = await self._do_put(endpoint, _NONE, data=data) return res async def download_iso( self, node: str, url: str, filename: str, storage: str, verify_certs: bool = True - ) -> ApiResponse: + ) -> ApiResponse[str]: endpoint = f"nodes/{node}/storage/{storage}/download-url" data = { "content": "iso", "filename": filename, "url": url, - "verify-certificates": verify_certs, + "verify-certificates": int(verify_certs), } - res = await self._do_post(endpoint, data=data) + res = await self._do_post(endpoint, _UPID, data=data) return res - async def get_next_id(self) -> ApiResponse: - res = await self._do_get("cluster/nextid") + async def get_next_id(self) -> ApiResponse[int]: + res = await self._do_get("cluster/nextid", _NEXT_ID) return res - async def create_vm(self, config: dict, node: str) -> ApiResponse: - res = await self._do_post(f"nodes/{node}/qemu", data=config) + async def create_vm(self, config: dict, node: str) -> ApiResponse[str]: + res = await self._do_post(f"nodes/{node}/qemu", _UPID, data=config) return res - async def start_vm(self, vm_id: int, node: str) -> ApiResponse: + async def start_vm(self, vm_id: int, node: str) -> ApiResponse[str]: endpoint = f"nodes/{node}/qemu/{vm_id}/status/start" - res = await self._do_post(endpoint) + res = await self._do_post(endpoint, _UPID) return res - async def get_task(self, task_id: str, node: str) -> ApiResponse: + async def get_task(self, task_id: str, node: str) -> ApiResponse[TaskStatus]: quoted_task = urllib.parse.quote(task_id) endpoint = f"nodes/{node}/tasks/{quoted_task}/status" - res = await self._do_get(endpoint) + res = await self._do_get(endpoint, _TASK_STATUS) return res @@ -396,11 +510,11 @@ async def wait_for_task(self, task_id: str, node: str, max_wait_s: int = 10) -> return exit_status == "OK" - async def delete_file(self, file_name: str, node: str, storage: str, content: str) -> ApiResponse: + async def delete_file(self, file_name: str, node: str, storage: str, content: str) -> ApiResponse[str]: volume = f"{storage}:{content}/{file_name}" endpoint = f"nodes/{node}/storage/{storage}/content/{volume}" - res = await self._do_delete(endpoint) + res = await self._do_delete(endpoint, _UPID) return res @@ -410,11 +524,11 @@ async def upload_file( node: str, storage: str, file_name: str | None = None, - ) -> ApiResponse: + ) -> ApiResponse[str]: endpoint = f"nodes/{node}/storage/{storage}/upload" if isinstance(file, bytes) and not file_name: - return ApiResponse(error="Unuseable file") + return ApiFailure(error="Unuseable file") # aiohttp's MultipartWriter.append() requires a synchronous file-like # object, so we open with the stdlib and manage the handle with ExitStack @@ -427,7 +541,7 @@ async def upload_file( payload = file resolved_name = file_name # already validated non-None above else: - return ApiResponse(error="Unuseable file") + return ApiFailure(error="Unuseable file") # Why proxmox does this is beyond me. This took hours to figure out. Had to # packet capture and decode the ssl traffic with wireshark. The multipart data @@ -447,35 +561,34 @@ async def upload_file( content_type = part[0].headers.pop(aiohttp.hdrs.CONTENT_TYPE) part[0].headers[aiohttp.hdrs.CONTENT_TYPE] = content_type - res = await self._do_post( - endpoint, - data=mpwriter, - ) + res = await self._do_post(endpoint, _UPID, data=mpwriter) return res - async def get_vms(self, node: str) -> ApiResponse: + async def get_vms(self, node: str) -> ApiResponse[list[VmSummary]]: endpoint = f"nodes/{node}/qemu" - res = await self._do_get(endpoint) + res = await self._do_get(endpoint, _VMS) return res - async def get_vm(self, vm_id: int, node: str) -> ApiResponse: + async def get_vm(self, vm_id: int, node: str) -> ApiResponse[dict[str, str | int | float]]: endpoint = f"nodes/{node}/qemu/{vm_id}/config" - res = await self._do_get(endpoint) + res = await self._do_get(endpoint, _VM_CONFIG) return res - async def stop_vm(self, vm_id: int, node: str) -> ApiResponse: + async def stop_vm(self, vm_id: int, node: str) -> ApiResponse[str]: endpoint = f"nodes/{node}/qemu/{vm_id}/status/stop" - res = await self._do_post(endpoint) + res = await self._do_post(endpoint, _UPID) return res - async def delete_vm(self, vm_id: int, node: str, purge: bool = True, destroy_disks: bool = True) -> ApiResponse: + async def delete_vm( + self, vm_id: int, node: str, purge: bool = True, destroy_disks: bool = True + ) -> ApiResponse[str]: params = [] if purge: params.append("purge=1") @@ -484,6 +597,6 @@ async def delete_vm(self, vm_id: int, node: str, purge: bool = True, destroy_dis query = "&".join(params) endpoint = f"nodes/{node}/qemu/{vm_id}{'?' + query if query else ''}" - res = await self._do_delete(endpoint) + res = await self._do_delete(endpoint, _UPID) return res diff --git a/src/arcane_mage/screens/hypervisor.py b/src/arcane_mage/screens/hypervisor.py index b63d19c..978b0d7 100644 --- a/src/arcane_mage/screens/hypervisor.py +++ b/src/arcane_mage/screens/hypervisor.py @@ -182,21 +182,23 @@ async def validate_hypervisor(self) -> None: async with client.session(): res = await client.get_hypervisor_nodes() - if res.unauthorized: - self.set_message(msg_type="error", value="Unauthorized") - self.hypervisor_valid = False - elif res.error: - self.set_message(msg_type="error", value=res.error) + if not res: + if res.unauthorized: + message = "Unauthorized" + elif res.timed_out: + message = "Timed out" + else: + message = res.error or "Unable to connect" + self.set_message(msg_type="error", value=message) self.hypervisor_valid = False - else: - # Extract node name from API response - if isinstance(res.payload, list): - names = [n.get("node", "") for n in res.payload if n.get("node")] - if names: - self._resolved_name = names[0] if len(names) == 1 else ",".join(sorted(names)) + return - self.set_message(msg_type="info", value="Validated") - self.hypervisor_valid = True + names = [n["node"] for n in res.payload] + if names: + self._resolved_name = names[0] if len(names) == 1 else ",".join(sorted(names)) + + self.set_message(msg_type="info", value="Validated") + self.hypervisor_valid = True def clear_message(self) -> None: try: @@ -232,10 +234,15 @@ def on_select_changed(self, event: Select.Changed) -> None: select = event.select - if select.id != "auth-type" or select.value == self.auth_type: + value = select.value + + if select.id != "auth-type" or value == self.auth_type: return - self.auth_type = select.value + if value == "token": + self.auth_type = "token" + elif value == "userpass": + self.auth_type = "userpass" def watch_auth_type(self, new: Literal["token", "userpass"]) -> None: labels = {"token": "API Token:", "userpass": "User / Pass:"} diff --git a/src/arcane_mage/screens/provision_node.py b/src/arcane_mage/screens/provision_node.py index e5bdc3e..43938e0 100644 --- a/src/arcane_mage/screens/provision_node.py +++ b/src/arcane_mage/screens/provision_node.py @@ -19,22 +19,22 @@ def __init__(self) -> None: self.previous: str | None = None - def validate(self, password: str) -> ValidationResult: - if self.previous is not None and len(password) < len(self.previous): - self.previous = password + def validate(self, value: str) -> ValidationResult: + if self.previous is not None and len(value) < len(self.previous): + self.previous = value return self.success() - if len(password) < 12: - self.previous = password + if len(value) < 12: + self.previous = value return self.failure("Password must be a minimum of 12 characters.") - groups = groupby(password) + groups = groupby(value) consecutives = [letter for letter, group in groups if len(list(group)) > 3] if any(consecutives): - self.previous = password + self.previous = value return self.failure("Only 3 consecutive repeat characters allowed.") - self.previous = password + self.previous = value return self.success() @@ -51,7 +51,9 @@ def on_screen_resume(self) -> None: self.query_one(Input).focus() def compose(self) -> ComposeResult: - title = "Provision: All nodes" if self.all else f"Provision: {self.fluxnode.hypervisor.vm_name}" + hyper = self.fluxnode.hypervisor + name = hyper.vm_name if hyper else self.fluxnode.system.hostname + title = "Provision: All nodes" if self.all else f"Provision: {name}" container = Container() container.border_title = title @@ -105,13 +107,14 @@ def on_button_pressed(self, event: Button.Pressed) -> None: self.dismiss(None) def on_input_changed(self, event: Input.Changed) -> None: - display = bool(event.validation_result.failures) + result = event.validation_result + display = bool(result and result.failures) label = self.query_one("#error-label", Label) label.visible = display - msg = "\n".join(event.validation_result.failure_descriptions) if display else "" + msg = "\n".join(result.failure_descriptions) if result and display else "" label.update(msg) provision_disabled = display or len(event.input.value) < 12 diff --git a/src/arcane_mage/screens/welcome_proxmox.py b/src/arcane_mage/screens/welcome_proxmox.py index b60e92a..9a7e278 100644 --- a/src/arcane_mage/screens/welcome_proxmox.py +++ b/src/arcane_mage/screens/welcome_proxmox.py @@ -24,6 +24,7 @@ from ..messages import UpdateDefaultPage from ..models import ArcaneOsConfig, ArcaneOsConfigGroup, HypervisorConfig from ..provisioner import Provisioner, get_latest_iso_version +from ..proxmox import VmSummary class WelcomeScreenProxmox(Screen): @@ -99,7 +100,7 @@ def validate_hypervisors(self, hypervisor: HypervisorConfig | None = None) -> No @work(name="populate_hypervisor") async def populate_hypervisor_api(self, hypervisor: HypervisorConfig) -> None: empty = ArcaneOsConfigGroup() - empty_provisioned: dict[str, list[dict]] = {} + empty_provisioned: dict[str, list[VmSummary]] = {} provisioner = await Provisioner.from_hypervisor_config(hypervisor) @@ -122,40 +123,37 @@ async def populate_hypervisor_api(self, hypervisor: HypervisorConfig) -> None: self.build_fluxnode_table(discovery.nodes, discovery.provisioned_vms) self.hypervisor_populated = True - def build_fluxnode_table(self, fluxnodes: ArcaneOsConfigGroup, provisioned_nodes: dict[str, list[dict]]) -> None: + def build_fluxnode_table( + self, fluxnodes: ArcaneOsConfigGroup, provisioned_nodes: dict[str, list[VmSummary]] + ) -> None: try: table = self.query_one(DataTable) except NoMatches: return columns = [ - {"label": "Node", "key": "node"}, - {"label": "Hostname", "key": "hostname"}, - {"label": "Tier", "key": "tier"}, - {"label": "Network", "key": "network"}, - {"label": "Address", "key": "address"}, - {"label": "Provisioned", "key": "provisioned"}, + ("Node", "node"), + ("Hostname", "hostname"), + ("Tier", "tier"), + ("Network", "network"), + ("Address", "address"), + ("Provisioned", "provisioned"), ] table.clear(columns=True) - for column in columns: - table.add_column(**column) + for label, key in columns: + table.add_column(label, key=key) for fluxnode in fluxnodes: hyper = fluxnode.hypervisor + if not hyper: + continue + row_key = f"{hyper.node}:{hyper.vm_name}" hypervisor_nodes = provisioned_nodes.get(hyper.node, []) - is_provisioned = bool( - next( - filter( - lambda x: x.get("name") == hyper.vm_name, - hypervisor_nodes, - ), - None, - ) - ) + is_provisioned = any(x.get("name") == hyper.vm_name for x in hypervisor_nodes) table.add_row(*fluxnode.as_row(), is_provisioned, key=row_key) def compose(self) -> ComposeResult: @@ -211,17 +209,17 @@ def on_button_pressed(self, event: Button.Pressed) -> None: if event.button.id == "add-hypervisor": self.post_message(WelcomeScreenProxmox.AddHypervisor()) elif event.button.id == "edit-hypervisor": - select = self.query_one(Select) - if select.value == Select.BLANK: + url = self.query_one(Select).value + if not isinstance(url, str): return - if hyper := self.get_hypervisor_by_url(select.value): + if hyper := self.get_hypervisor_by_url(url): self.post_message(WelcomeScreenProxmox.EditHypervisor(hyper)) elif event.button.id == "del-hypervisor": - select = self.query_one(Select) - if select.value == Select.BLANK: + url = self.query_one(Select).value + if not isinstance(url, str): return - if hyper := self.get_hypervisor_by_url(select.value): + if hyper := self.get_hypervisor_by_url(url): self.post_message(WelcomeScreenProxmox.DelHypervisor(hyper)) elif event.button.id == "back": @@ -251,7 +249,7 @@ def get_provisionable_nodes(self) -> ArcaneOsConfigGroup: row = dt.get_row(row_key) if not row[-1]: # not provisioned - fluxnode = self.fluxnodes.get_node_by_vm_name(*row_key.value.split(":")) + fluxnode = self._fluxnode_for_row(row_key.value) if fluxnode: nodes.append(fluxnode) @@ -268,18 +266,27 @@ def on_data_table_row_selected(self, event: DataTable.RowSelected) -> None: self.notify(f"Fluxnode: {name} already provisioned") return - fluxnode = self.fluxnodes.get_node_by_vm_name(*event.row_key.value.split(":")) + fluxnode = self._fluxnode_for_row(event.row_key.value) if not fluxnode: return self.post_message(WelcomeScreenProxmox.ProvisionNode(fluxnode)) + def _fluxnode_for_row(self, row_key: str | None) -> ArcaneOsConfig | None: + """The config behind a table row, keyed `:`.""" + if not row_key: + return None + + node, _, vm_name = row_key.partition(":") + + return self.fluxnodes.get_node_by_vm_name(node, vm_name) + def get_hypervisor_by_url(self, url: str) -> HypervisorConfig | None: return next(filter(lambda x: x.url == url, self.hypervisors), None) def on_select_changed(self, event: Select.Changed) -> None: - if (hyper_url := event.select.value) == Select.BLANK: + if not isinstance(hyper_url := event.select.value, str): return if hypervisor := self.get_hypervisor_by_url(hyper_url): diff --git a/tests/test_models/test_fluxnode.py b/tests/test_models/test_fluxnode.py index b4c7edb..d85580f 100644 --- a/tests/test_models/test_fluxnode.py +++ b/tests/test_models/test_fluxnode.py @@ -19,7 +19,7 @@ def test_from_dict(self): assert config.development is True def test_has_no_network_switch(self): - assert "testnet" not in GravityConfig(**{"testnet": True}).to_dict() + assert "testnet" not in GravityConfig(**{"testnet": True}).to_dict() # ty: ignore[unknown-argument] def test_to_dict_excludes_false(self): config = GravityConfig() diff --git a/tests/test_models/test_hypervisor.py b/tests/test_models/test_hypervisor.py index 0c1c2b5..d188f01 100644 --- a/tests/test_models/test_hypervisor.py +++ b/tests/test_models/test_hypervisor.py @@ -10,13 +10,12 @@ class TestHypervisorConfig: def test_from_dict(self): - data = { - "url": "https://pve.local:8006", - "auth_type": "token", - "credential": "user@pam!token=secret", - "keychain": False, - } - config = HypervisorConfig(**data) + config = HypervisorConfig( + url="https://pve.local:8006", + auth_type="token", + credential="user@pam!token=secret", + keychain=False, + ) assert config.url == "https://pve.local:8006" assert config.auth_type == "token" diff --git a/tests/test_models/test_installer.py b/tests/test_models/test_installer.py index 66e8df9..e4d7a3b 100644 --- a/tests/test_models/test_installer.py +++ b/tests/test_models/test_installer.py @@ -25,7 +25,7 @@ def test_from_dict(self): def test_from_dict_ignores_unknown_keys(self): data = {"auto_reboot": True, "unknown_key": "value"} - config = InstallerConfig(**data) + config = InstallerConfig(**data) # ty: ignore[invalid-argument-type] assert config.auto_reboot is True diff --git a/tests/test_provisioner.py b/tests/test_provisioner.py index 73a47f9..eae0bb4 100644 --- a/tests/test_provisioner.py +++ b/tests/test_provisioner.py @@ -5,7 +5,7 @@ import pytest from arcane_mage.provisioner import TIER_CONFIG, Provisioner, is_api_min_version -from arcane_mage.proxmox import ApiResponse +from arcane_mage.proxmox import ApiFailure, ApiSuccess class TestIsApiMinVersion: @@ -63,8 +63,7 @@ def provisioner(self, mock_api: AsyncMock) -> Provisioner: return Provisioner(api=mock_api) async def test_validate_api_version_success(self, provisioner: Provisioner, mock_api: AsyncMock): - mock_api.get_api_version.return_value = ApiResponse( - status=200, + mock_api.get_api_version.return_value = ApiSuccess( payload={"version": "8.4.1"}, ) @@ -74,8 +73,7 @@ async def test_validate_api_version_success(self, provisioner: Provisioner, mock assert msg == "8.4.1" async def test_validate_api_version_too_old(self, provisioner: Provisioner, mock_api: AsyncMock): - mock_api.get_api_version.return_value = ApiResponse( - status=200, + mock_api.get_api_version.return_value = ApiSuccess( payload={"version": "7.0.0"}, ) @@ -85,7 +83,7 @@ async def test_validate_api_version_too_old(self, provisioner: Provisioner, mock assert "too old" in msg async def test_validate_api_version_unreachable(self, provisioner: Provisioner, mock_api: AsyncMock): - mock_api.get_api_version.return_value = ApiResponse(error="Connection refused") + mock_api.get_api_version.return_value = ApiFailure(error="Connection refused") ok, msg = await provisioner.validate_api_version("node1") @@ -93,8 +91,7 @@ async def test_validate_api_version_unreachable(self, provisioner: Provisioner, assert "Unable to get" in msg async def test_validate_network_success(self, provisioner: Provisioner, mock_api: AsyncMock): - mock_api.get_networks.return_value = ApiResponse( - status=200, + mock_api.get_networks.return_value = ApiSuccess( payload=[{"iface": "vmbr0"}, {"iface": "vmbr1"}], ) @@ -103,8 +100,7 @@ async def test_validate_network_success(self, provisioner: Provisioner, mock_api assert result is True async def test_validate_network_missing(self, provisioner: Provisioner, mock_api: AsyncMock): - mock_api.get_networks.return_value = ApiResponse( - status=200, + mock_api.get_networks.return_value = ApiSuccess( payload=[{"iface": "vmbr0"}], ) @@ -113,8 +109,7 @@ async def test_validate_network_missing(self, provisioner: Provisioner, mock_api assert result is False async def test_validate_iso_version_found(self, provisioner: Provisioner, mock_api: AsyncMock): - mock_api.get_storage_content.return_value = ApiResponse( - status=200, + mock_api.get_storage_content.return_value = ApiSuccess( payload=[ {"content": "iso", "volid": "local:iso/FluxLive-1749291196.iso"}, ], @@ -125,8 +120,7 @@ async def test_validate_iso_version_found(self, provisioner: Provisioner, mock_a assert result is True async def test_validate_iso_version_not_found(self, provisioner: Provisioner, mock_api: AsyncMock): - mock_api.get_storage_content.return_value = ApiResponse( - status=200, + mock_api.get_storage_content.return_value = ApiSuccess( payload=[ {"content": "iso", "volid": "local:iso/other.iso"}, ], @@ -137,7 +131,7 @@ async def test_validate_iso_version_not_found(self, provisioner: Provisioner, mo assert result is False async def test_start_vm(self, provisioner: Provisioner, mock_api: AsyncMock): - mock_api.start_vm.return_value = ApiResponse(status=200, payload="UPID:task123") + mock_api.start_vm.return_value = ApiSuccess(payload="UPID:task123") mock_api.wait_for_task.return_value = True result = await provisioner.start_vm(100, "node1") @@ -146,7 +140,7 @@ async def test_start_vm(self, provisioner: Provisioner, mock_api: AsyncMock): mock_api.wait_for_task.assert_called_once_with("UPID:task123", "node1", 20) async def test_create_vm_config_cumulus(self, provisioner: Provisioner, mock_api: AsyncMock): - mock_api.get_next_id.return_value = ApiResponse(status=200, payload=100) + mock_api.get_next_id.return_value = ApiSuccess(payload=100) config = await provisioner.create_vm_config( vm_name="test-vm", @@ -163,14 +157,14 @@ async def test_create_vm_config_cumulus(self, provisioner: Provisioner, mock_api async def test_create_vm_config_invalid_tier(self, provisioner: Provisioner, mock_api: AsyncMock): config = await provisioner.create_vm_config( vm_name="test-vm", - tier="invalid", + tier="invalid", # ty: ignore[invalid-argument-type] network_bridge="vmbr0", ) assert config is None async def test_stop_vm_success(self, provisioner: Provisioner, mock_api: AsyncMock): - mock_api.stop_vm.return_value = ApiResponse(status=200, payload="UPID:stop123") + mock_api.stop_vm.return_value = ApiSuccess(payload="UPID:stop123") mock_api.wait_for_task.return_value = True result = await provisioner.stop_vm(100, "node1") @@ -180,14 +174,14 @@ async def test_stop_vm_success(self, provisioner: Provisioner, mock_api: AsyncMo mock_api.wait_for_task.assert_called_once_with("UPID:stop123", "node1", 30) async def test_stop_vm_failure(self, provisioner: Provisioner, mock_api: AsyncMock): - mock_api.stop_vm.return_value = ApiResponse(error="Connection refused") + mock_api.stop_vm.return_value = ApiFailure(error="Connection refused") result = await provisioner.stop_vm(100, "node1") assert result is False async def test_delete_vm_success(self, provisioner: Provisioner, mock_api: AsyncMock): - mock_api.delete_vm.return_value = ApiResponse(status=200, payload="UPID:del123") + mock_api.delete_vm.return_value = ApiSuccess(payload="UPID:del123") mock_api.wait_for_task.return_value = True result = await provisioner.delete_vm(100, "node1") @@ -197,7 +191,7 @@ async def test_delete_vm_success(self, provisioner: Provisioner, mock_api: Async mock_api.wait_for_task.assert_called_once_with("UPID:del123", "node1", 30) async def test_delete_vm_failure(self, provisioner: Provisioner, mock_api: AsyncMock): - mock_api.delete_vm.return_value = ApiResponse(error="Connection refused") + mock_api.delete_vm.return_value = ApiFailure(error="Connection refused") result = await provisioner.delete_vm(100, "node1") @@ -205,12 +199,11 @@ async def test_delete_vm_failure(self, provisioner: Provisioner, mock_api: Async async def test_deprovision_node_success(self, provisioner: Provisioner, mock_api: AsyncMock): """VM found, running, stopped, then deleted.""" - mock_api.get_vms.return_value = ApiResponse( - status=200, + mock_api.get_vms.return_value = ApiSuccess( payload=[{"vmid": 100, "name": "graham", "status": "running"}], ) - mock_api.stop_vm.return_value = ApiResponse(status=200, payload="UPID:stop1") - mock_api.delete_vm.return_value = ApiResponse(status=200, payload="UPID:del1") + mock_api.stop_vm.return_value = ApiSuccess(payload="UPID:stop1") + mock_api.delete_vm.return_value = ApiSuccess(payload="UPID:del1") mock_api.wait_for_task.return_value = True fluxnode = AsyncMock() @@ -224,8 +217,7 @@ async def test_deprovision_node_success(self, provisioner: Provisioner, mock_api mock_api.delete_vm.assert_called_once_with(100, "bigchug") async def test_deprovision_node_vm_not_found(self, provisioner: Provisioner, mock_api: AsyncMock): - mock_api.get_vms.return_value = ApiResponse( - status=200, + mock_api.get_vms.return_value = ApiSuccess( payload=[{"vmid": 200, "name": "other-vm", "status": "running"}], ) @@ -250,11 +242,10 @@ async def test_deprovision_node_no_hypervisor(self, provisioner: Provisioner, mo assert any("No hypervisor" in msg for _, msg in messages) async def test_deprovision_vm_by_name_success(self, provisioner: Provisioner, mock_api: AsyncMock): - mock_api.get_vms.return_value = ApiResponse( - status=200, + mock_api.get_vms.return_value = ApiSuccess( payload=[{"vmid": 100, "name": "graham", "status": "stopped"}], ) - mock_api.delete_vm.return_value = ApiResponse(status=200, payload="UPID:del1") + mock_api.delete_vm.return_value = ApiSuccess(payload="UPID:del1") mock_api.wait_for_task.return_value = True result = await provisioner.deprovision_vm("bigchug", vm_name="graham") @@ -263,11 +254,10 @@ async def test_deprovision_vm_by_name_success(self, provisioner: Provisioner, mo mock_api.delete_vm.assert_called_once_with(100, "bigchug") async def test_deprovision_vm_by_id_success(self, provisioner: Provisioner, mock_api: AsyncMock): - mock_api.get_vms.return_value = ApiResponse( - status=200, + mock_api.get_vms.return_value = ApiSuccess( payload=[{"vmid": 100, "name": "graham", "status": "stopped"}], ) - mock_api.delete_vm.return_value = ApiResponse(status=200, payload="UPID:del1") + mock_api.delete_vm.return_value = ApiSuccess(payload="UPID:del1") mock_api.wait_for_task.return_value = True result = await provisioner.deprovision_vm("bigchug", vm_id=100) @@ -276,8 +266,7 @@ async def test_deprovision_vm_by_id_success(self, provisioner: Provisioner, mock mock_api.delete_vm.assert_called_once_with(100, "bigchug") async def test_deprovision_vm_not_found(self, provisioner: Provisioner, mock_api: AsyncMock): - mock_api.get_vms.return_value = ApiResponse( - status=200, + mock_api.get_vms.return_value = ApiSuccess( payload=[{"vmid": 200, "name": "other", "status": "running"}], ) diff --git a/tests/test_proxmox.py b/tests/test_proxmox.py index 7d8e1a8..cbcc258 100644 --- a/tests/test_proxmox.py +++ b/tests/test_proxmox.py @@ -1,35 +1,111 @@ from __future__ import annotations +from collections.abc import AsyncGenerator +from contextlib import asynccontextmanager + import pytest +from aiohttp import web +from aiohttp.test_utils import TestServer -from arcane_mage.proxmox import ApiResponse, ParsedToken, ParsedUserPass, ProxmoxApi, ResolvedConnection +from arcane_mage.proxmox import ApiFailure, ApiSuccess, ParsedToken, ParsedUserPass, ProxmoxApi, ResolvedConnection class TestApiResponse: - def test_truthy_on_200(self): - resp = ApiResponse(status=200, payload={"data": "ok"}) - assert resp - - def test_falsy_on_error(self): - resp = ApiResponse(status=500, error="Internal Server Error") - assert not resp + def test_ok_is_truthy(self): + assert bool(ApiSuccess(payload="UPID:task")) is True - def test_falsy_on_200_with_error(self): - resp = ApiResponse(status=200, error="Something went wrong") - assert not resp + def test_error_is_falsy(self): + assert bool(ApiFailure(status=500, error="Internal Server Error")) is False def test_unauthorized(self): - resp = ApiResponse(status=401) - assert resp.unauthorized + assert ApiFailure(status=401).unauthorized - def test_not_unauthorized(self): - resp = ApiResponse(status=200) - assert not resp.unauthorized + def test_error_not_unauthorized(self): + assert not ApiFailure(status=500).unauthorized + + def test_ok_never_unauthorized(self): + assert not ApiSuccess(payload=None).unauthorized def test_timed_out(self): - resp = ApiResponse(timed_out=True) + resp = ApiFailure(timed_out=True) assert resp.timed_out - assert not resp + assert bool(resp) is False + + +TOKEN = ParsedToken("user@pam", "t", "v") + +type Route = tuple[str, str, int, dict[str, object]] + + +@asynccontextmanager +async def proxmox_answering(*routes: Route) -> AsyncGenerator[ProxmoxApi]: + """A ProxmoxApi talking to a local server that answers each (method, path) with (status, body).""" + app = web.Application() + + for method, path, status, body in routes: + + async def handler(_: web.Request, status: int = status, body: dict[str, object] = body) -> web.Response: + return web.json_response(body, status=status) + + app.router.add_route(method, f"/api2/json/{path}", handler) + + async with TestServer(app) as server, ProxmoxApi.from_token(str(server.make_url("")).rstrip("/"), TOKEN) as api: + yield api + + +class TestResponseShapes: + async def test_payload_validated(self): + async with proxmox_answering( + ("GET", "nodes/pve1/qemu", 200, {"data": [{"vmid": "101", "status": "running"}]}) + ) as api: + res = await api.get_vms("pve1") + + assert res + assert res.payload == [{"vmid": 101, "status": "running"}] + + async def test_unlisted_fields_kept(self): + vm = {"vmid": 101, "status": "running", "name": "a", "uptime": 42, "tags": "flux"} + async with proxmox_answering(("GET", "nodes/pve1/qemu", 200, {"data": [vm]})) as api: + res = await api.get_vms("pve1") + + assert res + assert res.payload == [vm] + + async def test_shape_mismatch_is_error(self): + async with proxmox_answering(("GET", "nodes/pve1/qemu", 200, {"data": None})) as api: + res = await api.get_vms("pve1") + + assert not res + assert res.status == 200 + assert res.error and res.error.startswith("Unexpected response") + + async def test_missing_required_field_is_error(self): + async with proxmox_answering(("GET", "nodes/pve1/network", 200, {"data": [{"type": "bridge"}]})) as api: + res = await api.get_networks("pve1") + + assert not res + assert res.error == "Unexpected response at 0.iface: Field required" + + async def test_message_is_error(self): + async with proxmox_answering(("GET", "cluster/nextid", 200, {"data": None, "message": "no permission"})) as api: + res = await api.get_next_id() + + assert not res + assert res.error == "no permission" + + async def test_http_error(self): + async with proxmox_answering(("GET", "nodes", 401, {"data": None})) as api: + res = await api.get_hypervisor_nodes() + + assert not res + assert res.unauthorized + + async def test_null_payload_endpoint(self): + async with proxmox_answering(("PUT", "storage/local", 200, {"data": None})) as api: + res = await api.set_storage_content("local", "iso,import") + + assert res + assert res.payload is None class TestProxmoxApiParsing: @@ -90,7 +166,7 @@ def test_frozen(self): conn = ResolvedConnection(url="https://pve.local:8006", token=token) with pytest.raises(AttributeError): - conn.url = "https://other.local:8006" + conn.url = "https://other.local:8006" # ty: ignore[invalid-assignment] with pytest.raises(AttributeError): - conn.token = ParsedToken("other@pam", "t", "v") + conn.token = ParsedToken("other@pam", "t", "v") # ty: ignore[invalid-assignment] diff --git a/uv.lock b/uv.lock index 104d29a..237ae1f 100644 --- a/uv.lock +++ b/uv.lock @@ -156,19 +156,6 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/eb/90/65238d4246307195411b87a07d03539049819b022c01bcc773826f600138/aiohttp_jinja2-1.6-py3-none-any.whl", hash = "sha256:0df405ee6ad1b58e5a068a105407dc7dcc1704544c559f1938babde954f945c7", size = 11736, upload-time = "2023-11-18T15:30:50.743Z" }, ] -[[package]] -name = "aioresponses" -version = "0.7.9" -source = { registry = "https://pypi.org/simple" } -dependencies = [ - { name = "aiohttp" }, - { name = "packaging" }, -] -sdist = { url = "https://files.pythonhosted.org/packages/28/fb/e3f08af812b3e66fca511ea1babb9dfddeca5965dea2a4d13b6926e0b1c2/aioresponses-0.7.9.tar.gz", hash = "sha256:1dcfa28938fc006f046a98383a7c07ac180be7a492c1ed557f5cd7b0805357d3", size = 34072, upload-time = "2026-06-23T21:23:23.828Z" } -wheels = [ - { url = "https://files.pythonhosted.org/packages/71/55/4c77cda7e69c1ac81a32e6895a361e0da9350eb7835a2ddb161a37ef1ce9/aioresponses-0.7.9-py2.py3-none-any.whl", hash = "sha256:94f9617f841c5bd7ee088ed783284f2cf4e6acc85d3933d92fc2fc7bd572a1b0", size = 12832, upload-time = "2026-06-23T21:23:22.426Z" }, -] - [[package]] name = "aiosignal" version = "1.4.0" @@ -199,10 +186,27 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/99/91/8acff4f5e50511b911bbccb72b8628a49c68ce14148cd9f6431094859a90/annotated_types-0.8.0-py3-none-any.whl", hash = "sha256:f072f4d804ea359e4eaf198b1af7a8b0943881a87f31bb764f8bf219bb9419e0", size = 13427, upload-time = "2026-07-23T20:16:12.938Z" }, ] +[[package]] +name = "ansible-core" +version = "2.21.5" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "cryptography" }, + { name = "jinja2" }, + { name = "packaging" }, + { name = "pyyaml" }, + { name = "resolvelib" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/0b/27/d02c3f75e9f1468b5efbe49667574f68331b17e8a9f2bb99e1d18be70052/ansible_core-2.21.5.tar.gz", hash = "sha256:65bb3148c58ebb843cd5074004c805f404f152114d98476dd823324616c9c6cb", size = 3395026, upload-time = "2026-10-05T15:11:00.15Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/57/6f/7321059b2d1a5ea3c7e60807a5269942d93b2c11a2fc89ba8e4a3e827891/ansible_core-2.21.5-py3-none-any.whl", hash = "sha256:08886f9f3e23d129aa98a01271c4a47833516e1cb7c737a246788cb474f41222", size = 2457267, upload-time = "2026-10-05T15:10:58.444Z" }, +] + [[package]] name = "arcane-mage" -version = "2.3.0" +version = "3.0.0" source = { editable = "." } +default-groups = ["ansible", "dev"] dependencies = [ { name = "aiofiles" }, { name = "aiohttp" }, @@ -225,8 +229,11 @@ tui = [ ] [package.dev-dependencies] +ansible = [ + { name = "ansible-core" }, +] dev = [ - { name = "aioresponses" }, + { name = "pyrage-stubs" }, { name = "pytest" }, { name = "pytest-asyncio" }, { name = "ruff" }, @@ -254,8 +261,9 @@ requires-dist = [ provides-extras = ["cli", "tui"] [package.metadata.requires-dev] +ansible = [{ name = "ansible-core", specifier = ">=2.21.5" }] dev = [ - { name = "aioresponses", specifier = ">=0.7.8" }, + { name = "pyrage-stubs", specifier = ">=1.4.0" }, { name = "pytest", specifier = ">=9.0.2" }, { name = "pytest-asyncio", specifier = ">=1.3.0" }, { name = "ruff", specifier = ">=0.17.2" }, @@ -1217,6 +1225,18 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/d3/8c/354fdb9ee7b45a2060272c623536b2cf9f0996de2242eaac177cff3f8003/pyrage-1.4.0-cp310-abi3-win_amd64.whl", hash = "sha256:c4a5dcd24a14b5e0d4cccff13649a2b8f1609ad45d6438bb85f35d84b0c8155e", size = 848219, upload-time = "2026-08-23T22:48:58.702Z" }, ] +[[package]] +name = "pyrage-stubs" +version = "1.4.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "pyrage" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/ae/50/f87f8cc62b1198840fa686bae94fad9543d1d7f8f7b058193d17f6d2771d/pyrage_stubs-1.4.0.tar.gz", hash = "sha256:9a44b0c1d2c169c67745ccb5790fcd1dd716941e9c4c4a66ecbd0862584766d9", size = 3400, upload-time = "2026-08-23T22:50:30.372Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/23/f5/e9edcdd379d819a5725b8b2b1dcfc0c7795b8bf1901867123302ce0a90e6/pyrage_stubs-1.4.0-py3-none-any.whl", hash = "sha256:033d58ee47f1fe294e5c576167dc679270557d63daaf79859a88896662eb0ec2", size = 4869, upload-time = "2026-08-23T22:50:29.223Z" }, +] + [[package]] name = "pytest" version = "9.1.1" @@ -1290,6 +1310,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/f1/12/de94a39c2ef588c7e6455cfbe7343d3b2dc9d6b6b2f40c4c6565744c873d/pyyaml-6.0.3-cp314-cp314t-win_arm64.whl", hash = "sha256:ebc55a14a21cb14062aa4162f906cd962b28e2e9ea38f9b4391244cd8de4ae0b", size = 149341, upload-time = "2025-09-25T21:32:56.828Z" }, ] +[[package]] +name = "resolvelib" +version = "1.2.1" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/1d/14/4669927e06631070edb968c78fdb6ce8992e27c9ab2cde4b3993e22ac7af/resolvelib-1.2.1.tar.gz", hash = "sha256:7d08a2022f6e16ce405d60b68c390f054efcfd0477d4b9bd019cc941c28fad1c", size = 24575, upload-time = "2025-10-11T01:07:44.582Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/e2/23/c941a0d0353681ca138489983c4309e0f5095dfd902e1357004f2357ddf2/resolvelib-1.2.1-py3-none-any.whl", hash = "sha256:fb06b66c8da04172d9e72a21d7d06186d8919e32ae5ab5cdf5b9d920be805ac2", size = 18737, upload-time = "2025-10-11T01:07:43.081Z" }, +] + [[package]] name = "rich" version = "15.0.0"