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"