diff --git a/.gitignore b/.gitignore index 17cd2fc..afcaaed 100644 --- a/.gitignore +++ b/.gitignore @@ -2,3 +2,4 @@ __pycache__/ *.pyc .pytest_cache/ +.claude/scratch/ diff --git a/README.md b/README.md index 2d2d612..72a25b7 100644 --- a/README.md +++ b/README.md @@ -21,11 +21,14 @@ uv run cv4pve-tui | `r` | reboot (confirm) | | `e` | edit cores / memory / grow a disk (disk growth asks again; it cannot be undone) | | `i` | info: backups, backup job, network (config and runtime IPs); `b` there = backup now (snapshot, zstd, storage choice); `a` there = also show backups of earlier guests that had the same VMID | +| `i` then `n` | in info: edit network (bridge, VLAN, firewall, IPv4/IPv6; VM IP via cloud-init at next boot); `u` there = undo the last network change of this guest (this session only) | | `t` | shell in a running LXC container (`ssh -t root@ pct enter `) | | `g` | refresh | | `c` | switch cv4pve-cli context (cluster) | | `q` | quit | +Warning: link changes (bridge, VLAN, firewall) apply immediately on a running guest and can cut its network (and DNS for others if the guest is your DNS server). IP changes on a running container may need a reboot of the container (seen: static -> DHCP gave no IPv4 until a reboot). + The shell key needs key-based root SSH to the PVE node (LXC only). Switching context runs `cv4pve-cli config use`, which changes cv4pve-cli's global current diff --git a/docs/plans/guest-tui-mvp.md b/docs/plans/guest-tui-mvp.md index a4cee1c..2a3fcb8 100644 --- a/docs/plans/guest-tui-mvp.md +++ b/docs/plans/guest-tui-mvp.md @@ -346,3 +346,103 @@ Observed CLI behaviour (cv4pve-cli 2.4.0, all read-only probes): - [x] Read-only live check: open InfoScreen for CT 104 and VM 108, data shown. - [x] User tests "Backup now" on CT 104 (`debian (manual)`, 2026-11-02 21:09). - [ ] Push `feat/guest-info`, merge to `main` (user-approved after the test). + +--- + +# Network editing - delivery B + +## Problem + +Change a guest's network settings from the info screen: bridge, VLAN, firewall, IPv4 +and IPv6 (incl. SLAAC), without breaking what is not edited. + +## Decisions (user-approved) + +- VM IP settings go through cloud-init `ipconfigN`, only when the VM has a cloud-init + drive; they apply at the next boot. VMs without cloud-init: IP fields disabled with a + note (IP is configured inside the VM). +- IPv4 modes: DHCP / Static (CIDR + optional gateway) / None. + IPv6 modes: SLAAC / DHCP / Static (CIDR + optional gateway) / None. +- Undo key `u`: restores the values from before the last network change made in this + app session for that guest. +- Out of scope: adding/removing interfaces, MAC, mtu, rate, trunks, DNS. + +API facts (Proxmox VE 9.2, `api usage ... config set -v`): + +- LXC `net[n]`: `name=[,bridge][,firewall=<1|0>][,gw=][,gw6=] + [,host-managed][,hwaddr][,ip=][,ip6=] + [,link_down][,mtu][,rate][,tag=][,trunks][,type=veth]`. `ip6=auto` = SLAAC. + Applied immediately on a running CT. +- VM `net[n]`: `[model=][,bridge][,firewall=<1|0>][,link_down][,macaddr][,mtu] + [,queues][,rate][,tag=][,trunks][,=]`. Applied immediately. +- VM `ipconfig[n]`: `[gw=][,gw6=][,ip=][,ip6=]` + (cloud-init; "no explicit gateway with dhcp"). Applies at the next boot. +- `delete=` removes settings (used when `ipconfigN` ends up empty). +- `get node network -o json` -> list of dicts with `iface`, `type` + (`bridge`, `OVSBridge`, `eth`, ...), `active`. +- Cloud-init drive: a disk value containing `cloudinit` + (`ide2 = local-lvm:vm-108-cloudinit,media=cdrom`). + +### `cli.py` + +- `bridges(guest) -> list[str]` - `iface` of entries with type `bridge` or `OVSBridge` + on the guest's node, sorted. +- `format_net(d: dict) -> str` - inverse of `parse_net`: `k=v` joined by `,`, order + kept. `format_net(parse_net(s)) == s` for the real samples. +- `has_cloudinit(config) -> bool`. +- `apply_link(net, bridge, tag, firewall) -> dict` - a copy of `net`: `bridge` set; + `tag` int 1-4094 or None (None removes `tag`); `firewall` True -> `"1"`, False -> + `firewall` removed. Existing keys keep their position; new keys are appended. +- `ip_mode(cfg, family) -> tuple[str, str, str]` - `(mode, cidr, gw)` read from a + parsed LXC net or VM ipconfig dict; family 4 uses `ip`/`gw`, 6 uses `ip6`/`gw6`. + Modes: `dhcp`, `static`, `none`, and for 6 also `auto`; `manual` reads as `none`. +- `apply_ip(cfg, family, mode, cidr="", gw="") -> dict` - copy with the family's keys + set: `dhcp`/`auto` -> `ip[6]=dhcp|auto`, gateway removed; `static` -> `ip[6]=`, + `gw[6]` set if given else removed; `none` -> both removed. Validation (ValueError): + static needs a CIDR of the right family with a `/prefix` + (`ipaddress.ip_interface`); a gateway must be an address of the same family; `auto` + only for family 6. +- `set_net(guest, changes: dict[str, str | None])` -> one call: + `set config [-- ...] [--delete ] --wait`; + `None` values go to `--delete`. Empty `changes` -> ValueError. + +### `app.py` (UI contract) + +- Key `n` in InfoScreen opens `NetScreen` for the guest (needs the loaded config; + nothing if the guest has no `netN`). +- `NetScreen(ModalScreen)`: `Select#iface` (netN keys), `Select#bridge` + (`cli.bridges`), `Input#tag` (empty = none), `Checkbox#firewall`, `Select#ipv4` + (`dhcp`/`static`/`none`), `Input#ipv4_cidr`, `Input#ipv4_gw`, `Select#ipv6` + (`auto` labelled SLAAC / `dhcp` / `static` / `none`), `Input#ipv6_cidr`, + `Input#ipv6_gw`, buttons `#apply` / `#cancel`; escape = cancel. Prefilled from the + current values of the selected interface; changing `#iface` prefills again. + VM without cloud-init: IP widgets disabled and `Static#ip_note` explains why. +- Apply: build the new netN (and for VMs the new ipconfigN) with `apply_link` / + `apply_ip`; ValueError -> error notify, screen stays open, no CLI call. No change -> + info notify, nothing else. Otherwise `ConfirmScreen` listing each changed key with + its old and new value (raw strings); for a running guest it says link changes apply + immediately, and for VM ipconfig that it applies at next boot. +- On yes: guest busy (refuse if busy), worker runs `cli.set_net(guest, changes)`, + done/error notify, busy released, InfoScreen reloads. On success the old values + (`None` for keys that did not exist) are stored as the guest's undo entry. +- Live finding (CT 104): Proxmox applies a changed `ip` on a running CT, but static -> + dhcp left the CT without IPv4 (no DHCP client restart) until the CT was rebooted. + The confirm therefore says, for IP changes on a running container, that a reboot of + the container may be needed. Proxmox re-orders keys when it stores `netN`. +- Review fixes: undo entries are cleared on a context switch (VMIDs repeat across + clusters); `has_cloudinit` only looks at disk keys; CIDR needs a decimal prefix + 1-32 / 1-128 (no netmask form, no /0); fields the user did not touch are not rewritten + (e.g. `ip6=manual` stays). +- Key `u` in InfoScreen: no undo entry -> info notify. Otherwise `ConfirmScreen` + showing what will be restored, then `cli.set_net(guest, old_values)` the same way; + the undo entry is cleared after success. + +## Steps + +- [x] Tests - red first. +- [x] Implementation - suite green, ruff clean, README keys `n` / `u`. +- [x] Live on CT 104 (approved), each step reverted: firewall off->on->off; VLAN + none->99->none (undo key); IPv6 SLAAC->DHCP->SLAAC; IPv4 DHCP->static + 172.16.41.200/23 gw 172.16.40.1->DHCP. Bridge change cannot be tested live: the + node has a single bridge (`vmbr0`). Final net0 string identical to the baseline. +- [ ] Push `feat/network-edit`, merge to `main` (ask first). diff --git a/pyproject.toml b/pyproject.toml index 2fc7a53..30e0c2e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -10,7 +10,11 @@ dependencies = ["textual>=8.2"] cv4pve-tui = "cv4pve_tui.app:main" [dependency-groups] -dev = ["pytest>=8", "pytest-asyncio>=1.0"] +dev = [ + "hypothesis>=6.170.14", + "pytest>=8", + "pytest-asyncio>=1.0", +] [build-system] requires = ["hatchling"] diff --git a/src/cv4pve_tui/app.py b/src/cv4pve_tui/app.py index 36faee1..173da82 100644 --- a/src/cv4pve_tui/app.py +++ b/src/cv4pve_tui/app.py @@ -10,6 +10,7 @@ from textual.containers import Horizontal, Vertical from textual.screen import ModalScreen from textual.widgets import ( Button, + Checkbox, DataTable, Footer, Header, @@ -160,6 +161,168 @@ class BackupScreen(ModalScreen[str | None]): self.dismiss(None) +_IPV4_MODES = [("DHCP", "dhcp"), ("Static", "static"), ("None", "none")] +_IPV6_MODES = [("SLAAC", "auto"), *_IPV4_MODES] + + +class NetScreen(ModalScreen): + """Edit bridge, VLAN, firewall and IP of one interface; confirms, then applies.""" + + BINDINGS = [("escape", "dismiss", "Cancel")] # noqa: RUF012 - Textual idiom + DEFAULT_CSS = "NetScreen Horizontal { height: auto; }" + + def __init__( + self, guest: dict, config: dict, bridges: list[str], running: bool + ) -> None: + super().__init__() + self.guest, self.config, self.running = guest, config, running + self.nets = cli.nets(config) + self.is_vm = cli.kind(guest) == "vm" + self.ip_enabled = not self.is_vm or cli.has_cloudinit(config) + # a bridge the node no longer lists must stay selectable, or Select rejects the prefill + used = {n.get("bridge") for n in self.nets.values()} - {None} + self.bridges = sorted(set(bridges) | used) + + def compose(self) -> ComposeResult: + with Vertical(): + yield Static( + f"Network of {self.guest['name']} ({self.guest['vmid']})", markup=False + ) + yield Select( + [(k, k) for k in self.nets], + value=next(iter(self.nets)), + allow_blank=False, + id="iface", + ) + with Horizontal(): + yield Select( + [(b, b) for b in self.bridges], allow_blank=False, id="bridge" + ) + yield Input(placeholder="VLAN tag (empty = none)", id="tag") + yield Checkbox("Firewall", id="firewall") + if not self.ip_enabled: + yield Static( + "IP is configured inside the VM (no cloud-init drive)", id="ip_note" + ) + for fam, modes in (("ipv4", _IPV4_MODES), ("ipv6", _IPV6_MODES)): + with Horizontal(): + yield Select(modes, allow_blank=False, id=fam) + yield Input(placeholder="address/prefix", id=f"{fam}_cidr") + yield Input(placeholder="gateway", id=f"{fam}_gw") + with Horizontal(): + yield Button("Apply", id="apply", variant="primary") + yield Button("Cancel", id="cancel") + + def on_mount(self) -> None: + for wid in ("ipv4", "ipv4_cidr", "ipv4_gw", "ipv6", "ipv6_cidr", "ipv6_gw"): + self.query_one(f"#{wid}").disabled = not self.ip_enabled + self._prefill() + + def on_select_changed(self, event: Select.Changed) -> None: + if event.select.id == "iface": + self._prefill() + + def _ip_cfg(self, iface: str) -> dict: + """The dict holding the IP keys: the VM's ipconfigN, or the LXC net itself.""" + if self.is_vm: + return cli.parse_net(str(self.config.get(f"ipconfig{iface[3:]}", ""))) + return self.nets[iface] + + def _prefill(self) -> None: + iface = self.query_one("#iface", Select).value + net = self.nets[iface] + self.query_one("#bridge", Select).value = net.get("bridge", self.bridges[0]) + self.query_one("#tag", Input).value = net.get("tag", "") + self.query_one("#firewall", Checkbox).value = net.get("firewall") == "1" + for fam, family in (("ipv4", 4), ("ipv6", 6)): + mode, cidr, gw = cli.ip_mode(self._ip_cfg(iface), family) + self.query_one(f"#{fam}", Select).value = mode + self.query_one(f"#{fam}_cidr", Input).value = cidr + self.query_one(f"#{fam}_gw", Input).value = gw + + def _changes(self) -> tuple[dict, dict]: + """(changes, old) as raw netN/ipconfigN strings, None = absent; ValueError on bad input.""" + iface = self.query_one("#iface", Select).value + text = self.query_one("#tag", Input).value.strip() + if text and not text.isdecimal(): + raise ValueError("VLAN tag must be 1-4094") + current = self.nets[iface] + link = ( + self.query_one("#bridge", Select).value, + int(text) if text else None, + self.query_one("#firewall", Checkbox).value, + ) + was = (current.get("bridge"), int(current["tag"]) if current.get("tag") else None, + current.get("firewall") == "1") # fmt: skip + # untouched fields are not rewritten (e.g. firewall=0 stays as it is) + net = cli.apply_link(current, *link) if link != was else dict(current) + ip_key = f"ipconfig{iface[3:]}" + ip_old = self._ip_cfg(iface) if self.is_vm else {} + ip = ip_old if self.is_vm else net # the dict that holds the IP keys + if self.ip_enabled: + for fam, family in (("ipv4", 4), ("ipv6", 6)): + mode, cidr, gw = ( + self.query_one(f"#{fam}{s}").value for s in ("", "_cidr", "_gw") + ) + wanted = (mode, cidr.strip(), gw.strip()) + if wanted != cli.ip_mode( + ip, family + ): # untouched: e.g. ip6=manual stays + ip = cli.apply_ip(ip, family, *wanted) + if not self.is_vm: + net = ip + changes, old = {}, {} + if net != self.nets[iface]: + changes[iface] = cli.format_net(net) + old[iface] = str(self.config[iface]) + if self.is_vm and ip != ip_old: + changes[ip_key] = cli.format_net(ip) or None + old[ip_key] = self.config.get(ip_key) + return changes, old + + def on_button_pressed(self, event: Button.Pressed) -> None: + if event.button.id == "cancel": + self.dismiss() + return + try: + changes, old = self._changes() + except ValueError as exc: + self.app.notify(str(exc), severity="error") + return + if not changes: + self.app.notify("No change") + return + lines = [ + f"{k}\n from {old[k] or '(not set)'}\n to {v or '(removed)'}" + for k, v in changes.items() + ] + notes = [] + if self.running and any(k.startswith("net") for k in changes): + notes.append( + "Link changes apply immediately and can cut the guest's network." + ) + if ( + self.running + and not self.is_vm + and any(_ip_changed(old[k], v) for k, v in changes.items()) + ): # seen live: static -> dhcp left the CT without IPv4 until it was rebooted + notes.append( + "IP changes on a running container may need a reboot of the container " + "(e.g. DHCP is not restarted)." + ) + if any(k.startswith("ipconfig") for k in changes): + notes.append("IP settings (cloud-init) apply at next boot.") + who = f"{self.guest['name']} ({self.guest['vmid']})" + self.app.push_screen( + ConfirmScreen("\n".join([f"Change network of {who}?", *lines, *notes])), + lambda ok: ( + self.app.start_net_change(self.guest, changes, old, self) + if ok + else None + ), + ) + + class InfoScreen(ModalScreen): """Backups, backup job and network of one guest (read-only), plus backup now.""" @@ -168,6 +331,8 @@ class InfoScreen(ModalScreen): ("g", "reload", "Refresh"), ("b", "backup", "Backup now"), ("a", "toggle_all", "All backups"), + ("n", "net", "Edit network"), + ("u", "undo", "Undo network"), ] DEFAULT_CSS = "InfoScreen DataTable { height: auto; max-height: 12; }" @@ -178,6 +343,8 @@ class InfoScreen(ModalScreen): self.jobs: list[dict] = [] self.backups: list[dict] = [] self.show_all = False + self.config: dict = {} + self.running = False def compose(self) -> ComposeResult: with Vertical(): @@ -231,6 +398,7 @@ class InfoScreen(ModalScreen): addresses = cli.interfaces(guest) except Exception as exc: # noqa: BLE001 - shown inline, the rest of the screen still works unavailable = f"unavailable: {_message(exc)}" + self.running = running self.app.call_from_thread( self._show, config, nets, storages, jobs, backups, addresses, unavailable ) @@ -250,7 +418,7 @@ class InfoScreen(ModalScreen): def _fill_screen( self, config, nets, storages, jobs, backups, addresses, unavailable ) -> None: - self.storages, self.jobs = storages, jobs + self.storages, self.jobs, self.config = storages, jobs, config covering = cli.covering_jobs(jobs, self.guest) self.query_one("#jobs", Static).update( "; ".join(_job_text(j) for j in covering) @@ -309,6 +477,46 @@ class InfoScreen(ModalScreen): self.show_all = not self.show_all self._show_backups() + def _refuse_if_busy(self) -> bool: + busy = self.guest["vmid"] in self.app.busy + if busy: + self.app.notify( + f"{self.guest['name']} ({self.guest['vmid']}) is busy", + severity="warning", + ) + return busy + + def action_net(self) -> None: + if cli.nets(self.config) and not self._refuse_if_busy(): + self._open_net() + + @work(thread=True) + def _open_net(self) -> None: + try: + bridges = cli.bridges(self.guest) + except Exception as exc: # noqa: BLE001 - surface every failure, never kill the app + self.app.call_from_thread(self.app.notify, _message(exc), severity="error") + return + self.app.call_from_thread( + self.app.push_screen, + NetScreen(self.guest, self.config, bridges, self.running), + ) + + def action_undo(self) -> None: + old = self.app.undo.get(self.guest["vmid"]) + if old is None: + self.app.notify("No network change to undo") + return + if self._refuse_if_busy(): + return + lines = [f"{k} = {v or '(remove)'}" for k, v in old.items()] + self.app.push_screen( + ConfirmScreen( + "\n".join(["Restore the previous network settings?", *lines]) + ), + lambda ok: self.app.start_net_change(self.guest, old, None) if ok else None, + ) + def action_backup(self) -> None: if not self.storages: self.app.notify("No backup storage available", severity="warning") @@ -324,6 +532,11 @@ class InfoScreen(ModalScreen): ) +def _ip_changed(old: str | None, new: str | None) -> bool: + before, after = cli.parse_net(old or ""), cli.parse_net(new or "") + return any(before.get(k) != after.get(k) for k in ("ip", "gw", "ip6", "gw6")) + + def _earlier_guest(backup: dict, name: str) -> bool: """VMIDs are reused: notes (the job's {{guestname}}) naming another guest mean a deleted one.""" notes = backup.get("notes") or "" @@ -355,6 +568,7 @@ class GuestApp(App): super().__init__() self.guests: dict[str, dict] = {} self.busy: set[int] = set() + self.undo: dict[int, dict] = {} # vmid -> values before the last network change # bumped on every context switch; replies started earlier come from the old cluster self.generation = 0 self.switching = False @@ -476,6 +690,7 @@ class GuestApp(App): "#guests", DataTable ).clear() # old cluster's guests: not actionable self.guests = {} + self.undo.clear() # VMIDs repeat across clusters: never undo onto another cluster self.generation += 1 self.switching = True self._switch_context(name) @@ -576,16 +791,57 @@ class GuestApp(App): else: self.call_from_thread(self.notify, f"Backup done: {who}") finally: - for screen in list( - self.screen_stack - ): # reload while still busy, like _run_job - if ( - isinstance(screen, InfoScreen) - and screen.guest["vmid"] == guest["vmid"] - ): - screen.load() + self._reload_infos(guest) self.busy.discard(guest["vmid"]) + def _reload_infos(self, guest: dict) -> None: + """Runs in a worker thread: reload the guest's open InfoScreens (call while still busy).""" + for screen in list(self.screen_stack): + if isinstance(screen, InfoScreen) and screen.guest["vmid"] == guest["vmid"]: + screen.load() + + def start_net_change( + self, guest: dict, changes: dict, entry: dict | None, screen=None + ) -> None: + """`entry` becomes the guest's undo entry on success (None clears it). + + `screen` (the NetScreen) is closed on success and stays open on failure. + """ + who = f"{guest['name']} ({guest['vmid']})" + if guest["vmid"] in self.busy: + self.notify(f"{who} is busy", severity="warning") + return + self.busy.add(guest["vmid"]) + self.notify(f"Network change started: {who}") + self._run_net(guest, changes, entry, screen) + + @work(thread=True) + def _run_net(self, guest: dict, changes: dict, entry: dict | None, screen) -> None: + who = f"{guest['name']} ({guest['vmid']})" + done = False + try: + cli.set_net(guest, changes) + except Exception as exc: # noqa: BLE001 - surface every failure, never kill the app + self.call_from_thread( + self.notify, f"Network change failed: {_message(exc)}", severity="error" + ) + else: + done = True + if entry is None: + self.undo.pop(guest["vmid"], None) + else: + self.undo[guest["vmid"]] = entry + self.call_from_thread(self.notify, f"Network change done: {who}") + finally: + self._reload_infos(guest) + if done and screen is not None: + self.call_from_thread(self._close_net, screen) + self.busy.discard(guest["vmid"]) + + def _close_net(self, screen) -> None: + if self.screen is screen: # the user may have left it already + screen.dismiss() + def action_shell(self) -> None: guest = self._selected() if guest is None: diff --git a/src/cv4pve_tui/cli.py b/src/cv4pve_tui/cli.py index 9b074a8..d3cd2c5 100644 --- a/src/cv4pve_tui/cli.py +++ b/src/cv4pve_tui/cli.py @@ -1,5 +1,6 @@ """Thin subprocess wrapper around cv4pve-cli.""" +import ipaddress import json import re import subprocess @@ -274,3 +275,107 @@ def interfaces(guest: dict) -> list[dict]: ) found.append({"name": i["name"], "mac": i.get("hardware-address"), **addrs}) return [i for i in found if i["name"] != "lo"] + + +def bridges(guest: dict) -> list[str]: + items = json.loads(run("get", "node", "network", guest["node"], "-o", "json")) + return sorted(i["iface"] for i in items if i.get("type") in ("bridge", "OVSBridge")) + + +def format_net(d: dict) -> str: + return ",".join( + f"{k}={v}" if v else k for k, v in d.items() + ) # bare flags stay bare + + +def has_cloudinit(config: dict) -> bool: + """A cloud-init drive: a disk (not a name or description) whose volume says so.""" + return any( + _VM_DISK.fullmatch(k) and "cloudinit" in str(v) for k, v in config.items() + ) + + +def apply_link(net: dict, bridge: str, tag: int | None, firewall: bool) -> dict: + """Copy of `net` with bridge, VLAN tag and firewall set; other keys stay in place.""" + if tag is not None and not 1 <= tag <= 4094: + raise ValueError("VLAN tag must be 1-4094") + out = dict(net) + out["bridge"] = bridge + if tag is None: + out.pop("tag", None) + else: + out["tag"] = str(tag) + if firewall: + out["firewall"] = "1" + else: + out.pop("firewall", None) + return out + + +def ip_mode(cfg: dict, family: int) -> tuple[str, str, str]: + """`(mode, cidr, gw)` of one family from an LXC net or VM ipconfig dict.""" + ip, gw = cfg.get(_IP_KEYS[family][0], ""), cfg.get(_IP_KEYS[family][1], "") + if ip in ("dhcp", "auto"): + return ip, "", "" + if ip in ("", "manual"): + return "none", "", "" + return "static", ip, gw + + +_IP_KEYS = {4: ("ip", "gw"), 6: ("ip6", "gw6")} + + +def apply_ip(cfg: dict, family: int, mode: str, cidr: str = "", gw: str = "") -> dict: + """Copy of `cfg` with one family's address keys set; ValueError on bad input.""" + ip_key, gw_key = _IP_KEYS[family] + out = dict(cfg) + if mode == "static": + try: + prefix = cidr.partition("/")[2] + if not ( + prefix.isdecimal() and 1 <= int(prefix) <= (32 if family == 4 else 128) + ): + raise ValueError # CIDR only: no netmask form, no /0 + if ipaddress.ip_interface(cidr).version != family: + raise ValueError + except ValueError: + raise ValueError(f"IPv{family} needs an address like 10.0.0.5/24") from None + try: + if gw and ipaddress.ip_address(gw).version != family: + raise ValueError + except ValueError: + raise ValueError( + f"IPv{family} gateway must be an IPv{family} address" + ) from None + out[ip_key] = cidr + if gw: + out[gw_key] = gw + else: + out.pop(gw_key, None) + elif mode in ("dhcp", "auto"): + if mode == "auto" and family != 6: + raise ValueError("SLAAC (auto) exists only for IPv6") + out[ip_key] = mode + out.pop(gw_key, None) + elif mode == "none": + out.pop(ip_key, None) + out.pop(gw_key, None) + else: + raise ValueError(f"unknown mode: {mode}") + return out + + +def set_net(guest: dict, changes: dict[str, str | None]) -> None: + """Set the given keys; `None` values are deleted. One call.""" + if not changes: + raise ValueError("nothing to set") + opts = [] + for key, value in changes.items(): + if value is not None: + opts += [f"--{key}", value] + gone = [k for k, v in changes.items() if v is None] + if gone: + opts += ["--delete", ",".join(gone)] + run( + "set", kind(guest), "config", guest["node"], str(guest["vmid"]), *opts, "--wait" + ) diff --git a/tests/test_app.py b/tests/test_app.py index d3c9ae8..6f56ba0 100644 --- a/tests/test_app.py +++ b/tests/test_app.py @@ -10,7 +10,14 @@ from datetime import datetime import pytest from textual.screen import ModalScreen -from textual.widgets import Button, DataTable, Input, OptionList, Select, Static +from textual.widgets import ( + Button, + DataTable, + Input, + OptionList, + Select, + Static, +) from cv4pve_tui import app as appmod from cv4pve_tui import cli @@ -52,6 +59,9 @@ class Fake: self.iface_calls, self.backups_calls, self.backup_runs = [], [], [] self.backup_error = None self.backup_hook = None + # delivery B (network editing) + self.bridge_list = ["vmbr0"] + self.set_calls, self.set_error, self.set_hook = [], None, None def list_guests(self): self.list_calls += 1 @@ -134,6 +144,25 @@ class Fake: if self.backup_error: raise self.backup_error + def bridges(self, guest): + return list(self.bridge_list) + + def set_net(self, guest, changes): + self.set_calls.append((guest["vmid"], dict(changes))) + if self.set_hook: + self.set_hook() + if self.set_error: + raise self.set_error + config = self.net_configs.setdefault(guest["vmid"], {}) + for ( + key, + value, + ) in changes.items(): # mirror the server so reloads see the change + if value is None: + config.pop(key, None) + else: + config[key] = value + BACKUP_JOB = {"id": "backup-1", "enabled": 1, "storage": "truenas-backup", "all": 1, "vmid": None, "exclude": None, "node": None, "schedule": "mon,wed,fri 04:00", @@ -164,6 +193,8 @@ def fake(monkeypatch): "backup_jobs", "backups", "backup_now", + "bridges", + "set_net", ): monkeypatch.setattr(cli, name, getattr(f, name), raising=False) for name in ("list_guests", "guest_config", "disks", "action", "set_resources", @@ -1316,3 +1347,520 @@ async def test_no_hint_when_nothing_hidden(fake): async with app.run_test() as pilot: await open_info(app, pilot) assert str(app.screen.query_one("#hidden", Static).render()).strip() == "" + + +# --- delivery B: network editing ---------------------------------------------- +# spec: docs/plans/guest-tui-mvp.md ("Network editing - delivery B" > app.py) + +CT_OLD = CT_CONFIG["net0"] +VM_CI = { + **VM_CONFIG, + "ide2": "local-lvm:vm-108-cloudinit,media=cdrom", +} # VM with cloud-init +CT_STATIC6 = ("name=eth0,bridge=vmbr0,hwaddr=BC:24:11:36:DE:78,ip=dhcp," + "ip6=2001:db8::5/64,gw6=2001:db8::1,type=veth") # fmt: skip + + +async def open_net(app, pilot, down=1): + """Open NetScreen; default row is CT 100 (stopped), down=0 is VM 108 (running).""" + await open_info(app, pilot, down=down) + await pilot.press("n") + await settle(app, pilot) + assert isinstance(app.screen, appmod.NetScreen) + + +def val(app, sel): + return app.screen.query_one(f"#{sel}").value + + +async def setv(app, pilot, **values): + for key, value in values.items(): + app.screen.query_one(f"#{key}").value = value + await pilot.pause() + + +async def to_info(app, pilot): + for _ in range(4): + if isinstance(app.screen, appmod.InfoScreen): + return + await pilot.press("escape") + await settle(app, pilot) + assert isinstance(app.screen, appmod.InfoScreen) + + +async def apply_to_confirm(app, pilot): + await press_button(app, pilot, "#apply") + assert isinstance(app.screen, ConfirmScreen) + return text_of(app.screen) + + +async def change_and_confirm(app, pilot, **values): + await setv(app, pilot, **values) + text = await apply_to_confirm(app, pilot) + await press_button(app, pilot, "#yes") + return text + + +def keys_of(changes): + return {k: cli.parse_net(v) if v is not None else None for k, v in changes.items()} + + +async def test_n_opens_net_screen_prefilled_for_ct_sample(fake): + fake.net_configs[100] = dict(CT_CONFIG) + app = make_app() + async with app.run_test() as pilot: + await open_net(app, pilot) + assert val(app, "iface") == "net0" + assert val(app, "bridge") == "vmbr0" + assert val(app, "tag") == "" + assert val(app, "firewall") is False + assert val(app, "ipv4") == "dhcp" + assert val(app, "ipv6") == "auto" + + +async def test_net_screen_prefilled_for_vm_from_ipconfig(fake): + fake.net_configs[108] = dict(VM_CI) + app = make_app() + async with app.run_test() as pilot: + await open_net(app, pilot, down=0) + assert val(app, "bridge") == "vmbr0" + assert val(app, "tag") == "20" + assert val(app, "firewall") is True + assert val(app, "ipv4") == "dhcp" + assert val(app, "ipv6") == "none" + assert not app.screen.query_one("#ipv4", Select).disabled + + +async def test_n_on_guest_without_nets_opens_nothing(fake): + app = make_app() # default Fake config has no netN + async with app.run_test() as pilot: + await open_info(app, pilot) + await pilot.press("n") + await settle(app, pilot) + assert isinstance(app.screen, appmod.InfoScreen) + + +async def test_changing_iface_prefills_again(fake): + fake.bridge_list = ["vmbr0", "vmbr1"] + fake.net_configs[100] = { + **CT_CONFIG, + "net1": "name=eth1,bridge=vmbr1,hwaddr=BC:24:11:36:DE:79,tag=7,firewall=1," + "ip=10.1.1.5/24,gw=10.1.1.1,ip6=manual,type=veth", + } # fmt: skip + app = make_app() + async with app.run_test() as pilot: + await open_net(app, pilot) + await setv(app, pilot, iface="net1") + assert val(app, "bridge") == "vmbr1" and val(app, "tag") == "7" + assert val(app, "firewall") is True + assert val(app, "ipv4") == "static" + assert ( + val(app, "ipv4_cidr") == "10.1.1.5/24" and val(app, "ipv4_gw") == "10.1.1.1" + ) + assert val(app, "ipv6") == "none" + + +async def test_vm_without_cloudinit_disables_ip_widgets_and_explains(fake): + fake.net_configs[108] = dict(VM_CONFIG) # no cloud-init drive + app = make_app() + async with app.run_test() as pilot: + await open_net(app, pilot, down=0) + for wid in ("ipv4", "ipv4_cidr", "ipv4_gw", "ipv6", "ipv6_cidr", "ipv6_gw"): + assert app.screen.query_one(f"#{wid}").disabled, wid + assert str(app.screen.query_one("#ip_note", Static).render()).strip() != "" + assert not app.screen.query_one("#tag").disabled + + +async def test_vm_without_cloudinit_link_change_keeps_key_position_and_no_ipconfig( + fake, +): + fake.net_configs[108] = dict(VM_CONFIG) + app = make_app() + async with app.run_test() as pilot: + await open_net(app, pilot, down=0) + await change_and_confirm(app, pilot, tag="30") + assert fake.set_calls == [ + (108, {"net0": "virtio=02:24:A6:49:A8:25,bridge=vmbr0,tag=30,firewall=1"}) + ] + + +async def test_set_vlan_99_confirm_shows_old_and_new_then_sets_once_busy_and_reloads( + fake, +): + fake.net_configs[100] = dict(CT_CONFIG) + app = make_app() + busy_seen = [] + fake.set_hook = lambda: busy_seen.append(set(app.busy)) + new = CT_OLD + ",tag=99" + async with app.run_test() as pilot: + await open_net(app, pilot) + before = len(fake.backups_calls) + await setv(app, pilot, tag="99") + text = await apply_to_confirm(app, pilot) + assert CT_OLD in text and new in text + assert fake.set_calls == [] # nothing before "yes" + await press_button(app, pilot, "#yes") + assert fake.set_calls == [(100, {"net0": new})] + assert busy_seen and busy_seen[0] + assert app.busy == set() and errors(app) == [] + assert len(fake.backups_calls) > before # InfoScreen reloaded + assert isinstance(app.screen, appmod.InfoScreen) + assert cells(app.screen.query_one("#nets", DataTable), 0)[2] == "99" + + +@pytest.mark.parametrize("bad", ["5000", "0", "abc"]) +async def test_invalid_tag_errors_without_call_and_screen_stays(fake, bad): + fake.net_configs[100] = dict(CT_CONFIG) + app = make_app() + async with app.run_test() as pilot: + await open_net(app, pilot) + await setv(app, pilot, tag=bad) + await press_button(app, pilot, "#apply") + assert errors(app) and fake.set_calls == [] + assert isinstance(app.screen, appmod.NetScreen) + + +async def test_static_ipv4_without_prefix_errors_without_call(fake): + fake.net_configs[100] = dict(CT_CONFIG) + app = make_app() + async with app.run_test() as pilot: + await open_net(app, pilot) + await setv(app, pilot, ipv4="static", ipv4_cidr="10.0.0.5") + await press_button(app, pilot, "#apply") + assert errors(app) and fake.set_calls == [] + assert isinstance(app.screen, appmod.NetScreen) + + +async def test_no_change_gives_info_notify_no_confirm_no_call(fake): + fake.net_configs[100] = dict(CT_CONFIG) + app = make_app() + async with app.run_test() as pilot: + await open_net(app, pilot) + await press_button(app, pilot, "#apply") + assert not isinstance(app.screen, ConfirmScreen) + assert fake.set_calls == [] and errors(app) == [] + assert any(sev == "information" for _, sev in app.seen) + + +async def test_static_ipv4_with_gateway_keeps_hwaddr_and_other_keys(fake): + fake.net_configs[100] = dict(CT_CONFIG) + app = make_app() + async with app.run_test() as pilot: + await open_net(app, pilot) + await change_and_confirm( + app, + pilot, + ipv4="static", + ipv4_cidr="172.16.41.200/23", + ipv4_gw="172.16.40.1", + ) + assert len(fake.set_calls) == 1 + vmid, changes = fake.set_calls[0] + assert vmid == 100 and set(changes) == {"net0"} + assert cli.parse_net(changes["net0"]) == { + "name": "eth0", "bridge": "vmbr0", "hwaddr": "BC:24:11:36:DE:78", + "ip": "172.16.41.200/23", "gw": "172.16.40.1", "ip6": "auto", "type": "veth", + } # fmt: skip + + +async def test_ipv6_static_to_dhcp_removes_gw6(fake): + fake.net_configs[100] = {"net0": CT_STATIC6} + app = make_app() + async with app.run_test() as pilot: + await open_net(app, pilot) + assert val(app, "ipv6") == "static" and val(app, "ipv6_gw") == "2001:db8::1" + await change_and_confirm(app, pilot, ipv6="dhcp") + new = cli.parse_net(fake.set_calls[0][1]["net0"]) + assert new["ip6"] == "dhcp" and "gw6" not in new + assert new["hwaddr"] == "BC:24:11:36:DE:78" and new["ip"] == "dhcp" + + +async def test_vm_ipconfig_change_goes_into_ipconfig0_and_confirm_mentions_next_boot( + fake, +): + fake.net_configs[108] = dict(VM_CI) + app = make_app() + async with app.run_test() as pilot: + await open_net(app, pilot, down=0) + await setv( + app, pilot, ipv4="static", ipv4_cidr="10.0.0.5/24", ipv4_gw="10.0.0.1" + ) + text = await apply_to_confirm(app, pilot) + assert "next boot" in text.lower() + assert "ip=dhcp" in text and "10.0.0.5/24" in text # old and new raw strings + await press_button(app, pilot, "#yes") + assert [v for v, _ in fake.set_calls] == [108] + changes = fake.set_calls[0][1] + assert set(changes) == {"ipconfig0"} # net0 is unchanged and not sent + assert cli.parse_net(changes["ipconfig0"]) == { + "ip": "10.0.0.5/24", + "gw": "10.0.0.1", + } + + +async def test_vm_ipconfig_none_none_deletes_ipconfig0(fake): + fake.net_configs[108] = dict(VM_CI) + app = make_app() + async with app.run_test() as pilot: + await open_net(app, pilot, down=0) + await change_and_confirm(app, pilot, ipv4="none") # ipv6 already none + assert fake.set_calls == [(108, {"ipconfig0": None})] + + +async def test_link_confirm_for_running_guest_says_applies_immediately(fake): + fake.net_configs[108] = dict(VM_CI) + app = make_app() + async with app.run_test() as pilot: + await open_net(app, pilot, down=0) + await setv(app, pilot, firewall=False) + text = await apply_to_confirm(app, pilot) + assert "immediately" in text.lower() + + +@pytest.mark.parametrize("how", ["cancel", "escape"]) +async def test_cancel_or_escape_in_net_screen_makes_no_call(fake, how): + fake.net_configs[100] = dict(CT_CONFIG) + app = make_app() + async with app.run_test() as pilot: + await open_net(app, pilot) + await setv(app, pilot, tag="99") + if how == "cancel": + await press_button(app, pilot, "#cancel") + else: + await pilot.press("escape") + await settle(app, pilot) + assert not isinstance(app.screen, appmod.NetScreen) + assert fake.set_calls == [] and app.busy == set() + + +async def test_declining_confirm_makes_no_call(fake): + fake.net_configs[100] = dict(CT_CONFIG) + app = make_app() + async with app.run_test() as pilot: + await open_net(app, pilot) + await setv(app, pilot, tag="99") + await apply_to_confirm(app, pilot) + await press_button(app, pilot, "#no") + assert fake.set_calls == [] and app.busy == set() + + +async def test_apply_while_guest_busy_warns_and_makes_no_second_call(fake): + fake.net_configs[100] = dict(CT_CONFIG) + started, release = threading.Event(), threading.Event() + fake.set_hook = lambda: (started.set(), release.wait(10)) + app = make_app() + async with app.run_test() as pilot: + await open_net(app, pilot) + try: + await setv(app, pilot, tag="99") + await apply_to_confirm(app, pilot) + app.screen.query_one("#yes", Button).press() + for _ in range(200): + if started.is_set(): + break + await pilot.pause(0.01) + assert started.is_set() + await pilot.press( + "escape" + ) # leave whatever is on top; the guest is still busy + await pilot.pause() + await to_info(app, pilot) + await pilot.press("n") + await pilot.pause() + if isinstance( + app.screen, appmod.NetScreen + ): # refusal may come at apply/yes + await setv(app, pilot, tag="55") + app.screen.query_one("#apply", Button).press() + await pilot.pause() + if isinstance(app.screen, ConfirmScreen): + app.screen.query_one("#yes", Button).press() + await pilot.pause() + assert any(sev == "warning" for _, sev in app.seen) + assert len(fake.set_calls) == 1 + finally: + release.set() + await settle(app, pilot) + assert len(fake.set_calls) == 1 + + +async def test_set_net_failure_notifies_error_releases_busy_and_stores_no_undo(fake): + fake.net_configs[100] = dict(CT_CONFIG) + fake.set_error = cli_error("Error: unable to apply net0") + app = make_app() + async with app.run_test() as pilot: + await open_net(app, pilot) + await change_and_confirm(app, pilot, tag="99") + assert any("unable to apply net0" in m for m in errors(app)) + assert app.busy == set() and app.is_running + await to_info(app, pilot) + calls = len(fake.set_calls) + await pilot.press("u") + await settle(app, pilot) + assert not isinstance(app.screen, ConfirmScreen) # no undo entry was stored + assert len(fake.set_calls) == calls + + +# --- undo --------------------------------------------------------------------- + + +async def test_u_without_undo_entry_gives_info_notify_and_no_call(fake): + fake.net_configs[100] = dict(CT_CONFIG) + app = make_app() + async with app.run_test() as pilot: + await open_info(app, pilot, down=1) + await pilot.press("u") + await settle(app, pilot) + assert isinstance(app.screen, appmod.InfoScreen) + assert any(sev == "information" for _, sev in app.seen) + assert fake.set_calls == [] + + +async def test_u_after_change_confirms_then_restores_old_net_string_and_clears_entry( + fake, +): + fake.net_configs[100] = dict(CT_CONFIG) + app = make_app() + async with app.run_test() as pilot: + await open_net(app, pilot) + await change_and_confirm(app, pilot, tag="99") + await to_info(app, pilot) + await pilot.press("u") + await settle(app, pilot) + assert isinstance(app.screen, ConfirmScreen) + assert CT_OLD in text_of(app.screen) # what will be restored + assert len(fake.set_calls) == 1 # nothing before "yes" + await press_button(app, pilot, "#yes") + assert fake.set_calls[1] == (100, {"net0": CT_OLD}) + assert app.busy == set() and errors(app) == [] + assert isinstance(app.screen, appmod.InfoScreen) + assert cells(app.screen.query_one("#nets", DataTable), 0)[2] == "-" + await pilot.press("u") # entry cleared after success + await settle(app, pilot) + assert not isinstance(app.screen, ConfirmScreen) and len(fake.set_calls) == 2 + + +async def test_undo_declined_makes_no_call_and_keeps_entry(fake): + fake.net_configs[100] = dict(CT_CONFIG) + app = make_app() + async with app.run_test() as pilot: + await open_net(app, pilot) + await change_and_confirm(app, pilot, tag="99") + await to_info(app, pilot) + await pilot.press("u") + await settle(app, pilot) + await press_button(app, pilot, "#no") + assert len(fake.set_calls) == 1 + await pilot.press("u") + await settle(app, pilot) + assert isinstance(app.screen, ConfirmScreen) # entry still there + + +async def test_undo_of_key_that_did_not_exist_deletes_it(fake): + ci_no_ipconfig = {k: v for k, v in VM_CI.items() if k != "ipconfig0"} + fake.net_configs[108] = ci_no_ipconfig + app = make_app() + async with app.run_test() as pilot: + await open_net(app, pilot, down=0) + await change_and_confirm(app, pilot, ipv4="dhcp") + assert fake.set_calls == [(108, {"ipconfig0": "ip=dhcp"})] + await to_info(app, pilot) + await pilot.press("u") + await settle(app, pilot) + await press_button(app, pilot, "#yes") + assert fake.set_calls[1] == (108, {"ipconfig0": None}) + + +async def test_undo_restores_both_net_and_ipconfig_of_one_change(fake): + fake.net_configs[108] = dict(VM_CI) + app = make_app() + async with app.run_test() as pilot: + await open_net(app, pilot, down=0) + await change_and_confirm(app, pilot, tag="30", ipv4="none") + assert set(fake.set_calls[0][1]) == {"net0", "ipconfig0"} + await to_info(app, pilot) + await pilot.press("u") + await settle(app, pilot) + await press_button(app, pilot, "#yes") + assert fake.set_calls[1] == ( + 108, + {"net0": VM_CONFIG["net0"], "ipconfig0": VM_CONFIG["ipconfig0"]}, + ) + + +async def test_undo_entry_is_per_guest(fake): + fake.net_configs[100] = dict(CT_CONFIG) + fake.net_configs[108] = dict(VM_CI) + app = make_app() + async with app.run_test() as pilot: + await open_net(app, pilot) # CT 100 + await change_and_confirm(app, pilot, tag="99") + await pilot.press("escape") # close InfoScreen + await settle(app, pilot) + await pilot.press("up") + await pilot.press("i") # VM 108: no change was made there + await settle(app, pilot) + assert isinstance(app.screen, appmod.InfoScreen) + await pilot.press("u") + await settle(app, pilot) + assert not isinstance(app.screen, ConfirmScreen) + + +# --- review findings: network editing ------------------------------------------- + + +async def test_context_switch_clears_undo_entries(fake): + # VMIDs repeat across clusters: an undo from one cluster must never hit the other + app = make_app() + async with app.run_test() as pilot: + await settle(app, pilot) + app.undo[108] = {"net0": "virtio=02:24:A6:49:A8:25,bridge=vmbr0"} + await switch_to_edge(app, pilot) + await settle(app, pilot) + assert app.undo == {} + + +async def test_untouched_fields_are_not_rewritten(fake): + # ip6=manual is left alone when only the VLAN changes; firewall=0 == absent (spec: off removes it) + fake.net_configs[100] = { + "net0": "name=eth0,bridge=vmbr0,firewall=0,hwaddr=BC:24:11:36:DE:78,ip=dhcp,ip6=manual,type=veth" + } + app = make_app() + async with app.run_test() as pilot: + await open_net(app, pilot) + await press_button(app, pilot, "#apply") + assert not isinstance( + app.screen, ConfirmScreen + ) # nothing touched -> "no change" + await setv(app, pilot, tag="99") + await press_button(app, pilot, "#apply") + await press_button(app, pilot, "#yes") + assert fake.set_calls[-1][1] == { + "net0": "name=eth0,bridge=vmbr0,hwaddr=BC:24:11:36:DE:78,ip=dhcp,ip6=manual,type=veth,tag=99" + } + + +# --- live finding: a running CT does not follow IP mode changes ------------------- + + +async def test_ip_change_on_running_ct_warns_reboot_may_be_needed(fake): + # live on CT 104: static -> dhcp left the CT without IPv4 until a reboot + fake.guests = [RUNNING_CT] + fake.net_configs[104] = dict(CT_CONFIG) + app = make_app() + async with app.run_test() as pilot: + await open_net(app, pilot, down=0) + await setv(app, pilot, ipv4="static", ipv4_cidr="172.16.41.200/23") + text = await apply_to_confirm(app, pilot) + assert "reboot" in text.lower() + + +async def test_link_only_change_on_running_ct_does_not_mention_reboot(fake): + fake.guests = [RUNNING_CT] + fake.net_configs[104] = dict(CT_CONFIG) + app = make_app() + async with app.run_test() as pilot: + await open_net(app, pilot, down=0) + await setv(app, pilot, tag="99") + text = await apply_to_confirm(app, pilot) + assert "reboot" not in text.lower() diff --git a/tests/test_cli.py b/tests/test_cli.py index 4bf5526..a3d8208 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -7,6 +7,8 @@ import json import subprocess import pytest +from hypothesis import given +from hypothesis import strategies as st from cv4pve_tui import cli @@ -683,3 +685,303 @@ def test_interfaces_ct_lists_all_addresses_incl_global_ipv6(fake): found = cli.interfaces({"vmid": 104, "node": "mainframe", "type": "lxc"}) assert found == [{"name": "eth0", "mac": "bc:24:11:36:de:78", "ipv4": ["172.16.41.200/23"], "ipv6": ["2a02:fe1:e77e:6004:be24:11ff:fe36:de78/64", "fe80::be24:11ff:fe36:de78/64"]}] # fmt: skip + + +# --- delivery B: network editing ---------------------------------------------- +# spec: docs/plans/guest-tui-mvp.md ("Network editing - delivery B" > cli.py) + +CT_NET = "name=eth0,bridge=vmbr0,hwaddr=BC:24:11:36:DE:78,ip=dhcp,ip6=auto,type=veth" +VM_NET_LINK = "virtio=02:24:A6:49:A8:25,bridge=vmbr0" + + +def test_bridges_filters_bridge_types_sorted_and_uses_node_network_argv(fake): + fake.result = ( + json.dumps( + [ + {"iface": "vmbr1", "type": "bridge", "active": 1}, + {"iface": "eno1", "type": "eth", "active": 1}, + {"iface": "ovs0", "type": "OVSBridge", "active": 1}, + {"iface": "bond0", "type": "bond", "active": 1}, + {"iface": "vmbr0", "type": "bridge", "active": 1}, + ] + ), + "", + 0, + ) + assert cli.bridges(VM) == ["ovs0", "vmbr0", "vmbr1"] + assert fake.argv == [ + "cv4pve-cli", + "get", + "node", + "network", + "mainframe", + "-o", + "json", + ] + + +def test_bridges_none_present_gives_empty_list(fake): + fake.result = (json.dumps([{"iface": "eno1", "type": "eth", "active": 1}]), "", 0) + assert cli.bridges(CT) == [] + + +@pytest.mark.parametrize("sample", [CT_NET, VM_NET_LINK, "ip=dhcp"]) +def test_format_net_round_trips_real_samples(sample): + assert cli.format_net(cli.parse_net(sample)) == sample + + +_token = st.text( + alphabet=st.characters( + blacklist_characters=",=", blacklist_categories=("Cs", "Cc") + ), + min_size=1, +) + + +@given( + st.lists( + st.tuples(_token, _token), min_size=1, max_size=8, unique_by=lambda kv: kv[0] + ) +) +def test_format_net_parse_net_round_trip_property(pairs): + s = ",".join(f"{k}={v}" for k, v in pairs) + assert cli.format_net(cli.parse_net(s)) == s + + +def test_format_net_keeps_dict_order(): + assert cli.format_net({"b": "1", "a": "2"}) == "b=1,a=2" + + +def test_has_cloudinit_true_when_a_disk_value_mentions_cloudinit(): + assert cli.has_cloudinit( + {"ide2": "local-lvm:vm-108-cloudinit,media=cdrom", "cores": 2} + ) + + +def test_has_cloudinit_false_without_cloudinit_drive(): + assert not cli.has_cloudinit( + {"scsi0": "local-lvm:vm-108-disk-0,size=10G", "cores": 2} + ) + + +# apply_link + + +def test_apply_link_sets_bridge_and_returns_a_copy(): + net = cli.parse_net(CT_NET) + out = cli.apply_link(net, "vmbr1", None, False) + assert out["bridge"] == "vmbr1" + assert net == cli.parse_net(CT_NET) # input untouched + + +def test_apply_link_new_tag_and_firewall_are_appended_after_existing_keys(): + out = cli.apply_link(cli.parse_net(CT_NET), "vmbr0", 99, True) + assert list(out)[-2:] == ["tag", "firewall"] or list(out)[-2:] == [ + "firewall", + "tag", + ] + assert out["tag"] == "99" and out["firewall"] == "1" + assert list(out)[:6] == ["name", "bridge", "hwaddr", "ip", "ip6", "type"] + + +def test_apply_link_none_tag_and_false_firewall_remove_the_keys(): + net = cli.parse_net("virtio=02:24:A6:49:A8:25,bridge=vmbr0,tag=20,firewall=1") + out = cli.apply_link(net, "vmbr0", None, False) + assert out == {"virtio": "02:24:A6:49:A8:25", "bridge": "vmbr0"} + + +def test_apply_link_false_firewall_removes_firewall_zero_too(): + out = cli.apply_link(cli.parse_net(CT_NET + ",firewall=0"), "vmbr0", None, False) + assert "firewall" not in out + + +def test_apply_link_existing_keys_keep_position_and_unrelated_keys_untouched(): + src = "e1000=02:24:A6:49:A8:25,bridge=vmbr0,tag=20,mtu=1400,rate=10,trunks=10;20,firewall=1" + out = cli.apply_link(cli.parse_net(src), "vmbr2", 30, True) + assert cli.format_net(out) == ( + "e1000=02:24:A6:49:A8:25,bridge=vmbr2,tag=30,mtu=1400,rate=10,trunks=10;20,firewall=1" + ) + + +def test_apply_link_keeps_lxc_hwaddr_and_ip_keys(): + out = cli.apply_link(cli.parse_net(CT_NET), "vmbr1", 5, False) + for key in ("hwaddr", "ip", "ip6", "name", "type"): + assert out[key] == cli.parse_net(CT_NET)[key] + + +@pytest.mark.parametrize("tag", [1, 4094]) +def test_apply_link_accepts_tag_boundaries(tag): + assert cli.apply_link({"bridge": "vmbr0"}, "vmbr0", tag, False)["tag"] == str(tag) + + +@pytest.mark.parametrize("tag", [0, -1, 4095, 5000]) +def test_apply_link_rejects_tag_outside_1_to_4094(tag): + with pytest.raises(ValueError): + cli.apply_link({"bridge": "vmbr0"}, "vmbr0", tag, False) + + +# ip_mode + + +@pytest.mark.parametrize( + ("cfg", "family", "expected"), + [ + ({"ip": "dhcp", "ip6": "auto"}, 4, ("dhcp", "", "")), + ({"ip": "dhcp", "ip6": "auto"}, 6, ("auto", "", "")), + ( + {"ip": "10.0.0.5/24", "gw": "10.0.0.1"}, + 4, + ("static", "10.0.0.5/24", "10.0.0.1"), + ), + ({"ip": "10.0.0.5/24"}, 4, ("static", "10.0.0.5/24", "")), + ( + {"ip6": "2001:db8::5/64", "gw6": "2001:db8::1"}, + 6, + ("static", "2001:db8::5/64", "2001:db8::1"), + ), + ({"ip6": "dhcp"}, 6, ("dhcp", "", "")), + ({"ip": "manual", "ip6": "manual"}, 4, ("none", "", "")), + ({"ip": "manual", "ip6": "manual"}, 6, ("none", "", "")), + ({"ip": "dhcp"}, 6, ("none", "", "")), # key absent reads as none + ({"ip6": "auto"}, 4, ("none", "", "")), # families are independent + ({"ip": "10.0.0.5/24", "gw": "10.0.0.1"}, 6, ("none", "", "")), + ], +) +def test_ip_mode_reads_mode_cidr_and_gateway(cfg, family, expected): + assert cli.ip_mode(cfg, family) == expected + + +# apply_ip + + +def test_apply_ip_dhcp_sets_dhcp_and_removes_gateway_keeping_other_keys(): + cfg = {"name": "eth0", "ip": "10.0.0.5/24", "gw": "10.0.0.1", "ip6": "auto"} + out = cli.apply_ip(cfg, 4, "dhcp") + assert out == {"name": "eth0", "ip": "dhcp", "ip6": "auto"} + assert cfg["ip"] == "10.0.0.5/24" # input untouched + + +def test_apply_ip_auto_for_family_6_removes_gw6(): + out = cli.apply_ip( + {"ip": "dhcp", "ip6": "2001:db8::5/64", "gw6": "2001:db8::1"}, 6, "auto" + ) + assert out == {"ip": "dhcp", "ip6": "auto"} + + +def test_apply_ip_dhcp_for_family_6(): + assert cli.apply_ip({"ip6": "auto"}, 6, "dhcp") == {"ip6": "dhcp"} + + +def test_apply_ip_static_v4_with_gateway(): + out = cli.apply_ip( + {"ip": "dhcp", "ip6": "auto"}, 4, "static", "172.16.41.200/23", "172.16.40.1" + ) + assert out == {"ip": "172.16.41.200/23", "gw": "172.16.40.1", "ip6": "auto"} + + +def test_apply_ip_static_v4_without_gateway_removes_old_gateway(): + out = cli.apply_ip( + {"ip": "10.0.0.5/24", "gw": "10.0.0.1"}, 4, "static", "10.0.0.9/24" + ) + assert out == {"ip": "10.0.0.9/24"} + + +def test_apply_ip_static_v6_with_and_without_gateway(): + with_gw = cli.apply_ip( + {"ip6": "auto"}, 6, "static", "2001:db8::5/64", "2001:db8::1" + ) + assert with_gw == {"ip6": "2001:db8::5/64", "gw6": "2001:db8::1"} + no_gw = cli.apply_ip(with_gw, 6, "static", "2001:db8::6/64") + assert no_gw == {"ip6": "2001:db8::6/64"} + + +def test_apply_ip_none_removes_ip_and_gateway_of_that_family_only(): + cfg = { + "ip": "10.0.0.5/24", + "gw": "10.0.0.1", + "ip6": "2001:db8::5/64", + "gw6": "2001:db8::1", + } + assert cli.apply_ip(cfg, 4, "none") == { + "ip6": "2001:db8::5/64", + "gw6": "2001:db8::1", + } + assert cli.apply_ip(cfg, 6, "none") == {"ip": "10.0.0.5/24", "gw": "10.0.0.1"} + + +def test_apply_ip_none_on_vm_ipconfig_can_leave_empty_dict(): + assert cli.apply_ip({"ip": "dhcp"}, 4, "none") == {} + + +@pytest.mark.parametrize( + ("family", "mode", "cidr", "gw"), + [ + (4, "static", "2001:db8::5/64", ""), # wrong family + (6, "static", "10.0.0.5/24", ""), # wrong family + (4, "static", "10.0.0.5", ""), # missing prefix + (6, "static", "2001:db8::5", ""), # missing prefix + (4, "static", "", ""), # missing CIDR + (4, "static", "not-an-ip/24", ""), + (4, "static", "10.0.0.5/24", "nonsense"), # bad gateway + (4, "static", "10.0.0.5/24", "2001:db8::1"), # gateway family mismatch + (6, "static", "2001:db8::5/64", "10.0.0.1"), # gateway family mismatch + (4, "auto", "", ""), # SLAAC exists only for IPv6 + ], +) +def test_apply_ip_validation_errors(family, mode, cidr, gw): + with pytest.raises(ValueError): + cli.apply_ip({"ip": "dhcp"}, family, mode, cidr, gw) + + +# set_net + + +def test_set_net_sets_only(fake): + cli.set_net(CT, {"net0": CT_NET + ",tag=99"}) + assert fake.argv == ["cv4pve-cli", "set", "ct", "config", "mainframe", "100", + "--net0", CT_NET + ",tag=99", "--wait"] # fmt: skip + + +def test_set_net_deletes_only(fake): + cli.set_net(VM, {"ipconfig0": None}) + assert fake.argv == ["cv4pve-cli", "set", "vm", "config", "mainframe", "108", + "--delete", "ipconfig0", "--wait"] # fmt: skip + + +def test_set_net_sets_then_deletes_joined_with_comma_and_wait_last(fake): + cli.set_net( + VM, {"net0": VM_NET_LINK + ",tag=5", "ipconfig0": None, "ipconfig1": None} + ) + assert fake.argv == ["cv4pve-cli", "set", "vm", "config", "mainframe", "108", + "--net0", VM_NET_LINK + ",tag=5", "--delete", "ipconfig0,ipconfig1", "--wait"] # fmt: skip + + +def test_set_net_empty_changes_is_value_error_without_call(fake): + with pytest.raises(ValueError): + cli.set_net(VM, {}) + assert fake.calls == [] + + +def test_set_net_failure_raises_clierror(monkeypatch): + FakeRun(monkeypatch, "Error: boom", returncode=1) + with pytest.raises(cli.CliError): + cli.set_net(CT, {"net0": CT_NET}) + + +# --- review findings: network editing ----------------------------------------- + + +def test_has_cloudinit_only_looks_at_disks(): + assert cli.has_cloudinit({"ide2": "local-lvm:vm-108-cloudinit,media=cdrom"}) + assert not cli.has_cloudinit( + {"name": "ubuntu-cloudinit-tpl", "description": "cloudinit base"} + ) + + +@pytest.mark.parametrize( + "cidr", ["10.0.0.5/255.255.255.0", "10.0.0.5/0", "2001:db8::5/0"] +) +def test_apply_ip_rejects_netmask_form_and_zero_prefix(cidr): + family = 6 if ":" in cidr else 4 + with pytest.raises(ValueError): + cli.apply_ip({}, family, "static", cidr) diff --git a/uv.lock b/uv.lock index 194b9dc..a09eb80 100644 --- a/uv.lock +++ b/uv.lock @@ -21,6 +21,7 @@ dependencies = [ [package.dev-dependencies] dev = [ + { name = "hypothesis" }, { name = "pytest" }, { name = "pytest-asyncio" }, ] @@ -30,10 +31,92 @@ requires-dist = [{ name = "textual", specifier = ">=8.2" }] [package.metadata.requires-dev] dev = [ + { name = "hypothesis", specifier = ">=6.170.14" }, { name = "pytest", specifier = ">=8" }, { name = "pytest-asyncio", specifier = ">=1.0" }, ] +[[package]] +name = "hypothesis" +version = "6.170.14" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "sortedcontainers" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/d6/0f/10b2076a1314c554eee58c26e1dbe630fd26a32773df2a63bd5dba5d9947/hypothesis-6.170.14.tar.gz", hash = "sha256:8272913eb34c4399a88cf8a9ecdb351c9ef8f621c061f2ac323362d4a66b755d", size = 521490, upload-time = "2026-10-30T06:51:55.943Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/c9/d6/eae1191f3398b8cc9f10b20c9b93dd1ceae037565e94ababf521730e59a4/hypothesis-6.170.14-cp311-abi3-macosx_10_12_x86_64.whl", hash = "sha256:b1c4990f9fafad83cdac33bc564d41066f5824a92ddd45ded7e0e84277a3ec0e", size = 800123, upload-time = "2026-10-30T06:50:33.682Z" }, + { url = "https://files.pythonhosted.org/packages/56/78/434915150cecd5b9ba41837106a3da2782f515a9aed583d25860bfa6d71c/hypothesis-6.170.14-cp311-abi3-macosx_11_0_arm64.whl", hash = "sha256:ab414b45a56853fcdd6b602d9c5033e3c7d127ee8b8e7e3397223b746440b37c", size = 795278, upload-time = "2026-10-30T06:50:22.237Z" }, + { url = "https://files.pythonhosted.org/packages/95/6b/07f5e5e667b4b3e90bea16b4db09367b5bb1dfae63eb75a7152b053b0c50/hypothesis-6.170.14-cp311-abi3-manylinux1_i686.manylinux_2_5_i686.whl", hash = "sha256:cceb1b3c19ea0a146c7bb2fb2b44e3ef231bcecd42c97d95154e5d233e209572", size = 1162031, upload-time = "2026-10-30T06:50:08.828Z" }, + { url = "https://files.pythonhosted.org/packages/22/ba/51e0dc10befcc4ca9b7fb81b76ca6aef4ffc9c7a741191a8b41e878ee2f2/hypothesis-6.170.14-cp311-abi3-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:376dbf5ffa1e3ad71af2a1af0877176faabe3e63d345324a33c82e408dbfd604", size = 1124181, upload-time = "2026-10-30T06:51:51.244Z" }, + { url = "https://files.pythonhosted.org/packages/ee/37/77834f102b0bca702f723dd519b8f590e83ec6392e59565384cfdc5bbec7/hypothesis-6.170.14-cp311-abi3-manylinux2014_armv7l.manylinux_2_17_armv7l.whl", hash = "sha256:5f0f6f6666fa15ec13ca1d8430f70660fce8b3f4f133e25fdf674be6ad77295e", size = 1154664, upload-time = "2026-10-30T06:51:50.027Z" }, + { url = "https://files.pythonhosted.org/packages/12/76/83edc11f33319a2febd025fb504aa1ed8e245b7e25905da80513615e4b5e/hypothesis-6.170.14-cp311-abi3-manylinux2014_ppc64le.manylinux_2_17_ppc64le.whl", hash = "sha256:c4ff70fbb94f54cd35045a40e5a77dd8f876d4cc6a39dcdc4ac1af6f8f5cc6fe", size = 1148576, upload-time = "2026-10-30T06:50:18.505Z" }, + { url = "https://files.pythonhosted.org/packages/ce/35/d4086d546c0abe406f0ece17e33eca7051e87dc7501485a696385aed34fd/hypothesis-6.170.14-cp311-abi3-manylinux2014_s390x.manylinux_2_17_s390x.whl", hash = "sha256:8ecec6be3a02e22c70bc7dfb15afeeeb708aeccb994c9d57db438199cad9bfbb", size = 1201336, upload-time = "2026-10-30T06:50:17.221Z" }, + { url = "https://files.pythonhosted.org/packages/dd/85/117b0dde708240854ce0f1ad95d43bc132f59e6bffc82c2a20e76b984020/hypothesis-6.170.14-cp311-abi3-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:3bbf06c4f8e3aab37c144fe3bc1d89a051a2dfbefef8c2a25160b11da854f949", size = 1167592, upload-time = "2026-10-30T06:50:23.393Z" }, + { url = "https://files.pythonhosted.org/packages/29/bb/105ee91bf92e3b7a5bc388827e774b7c5744f39ce36f50f94918b584813c/hypothesis-6.170.14-cp311-abi3-manylinux_2_31_riscv64.whl", hash = "sha256:07869d66bf1e92e39447808fcc1977bfd10bfdff4970d6969a49e89a84cf04a7", size = 1123748, upload-time = "2026-10-30T06:51:07.034Z" }, + { url = "https://files.pythonhosted.org/packages/cd/9b/c60e868ba0f93ff5bc7cec3e413c5e824863bbd9bd45dcc86bc18adb4266/hypothesis-6.170.14-cp311-abi3-musllinux_1_2_aarch64.whl", hash = "sha256:0ce9ddf0d0e99c532c2cf00d97b63316830810d18f11c80c7bcf0f771969f3a4", size = 1299121, upload-time = "2026-10-30T06:51:00.068Z" }, + { url = "https://files.pythonhosted.org/packages/e9/f6/a2b9630ddc70aa7642931ac6c353fc57df38c6de655718a50337b5b7d4e0/hypothesis-6.170.14-cp311-abi3-musllinux_1_2_armv7l.whl", hash = "sha256:420b4a77157e94f124498fff639284d514b60926f61925dd7aafb75411e7ce92", size = 1427779, upload-time = "2026-10-30T06:50:15.652Z" }, + { url = "https://files.pythonhosted.org/packages/db/36/b3e7d3fef4ce3f44f70a61399a3549c02a71e430122f688880a24336e9a6/hypothesis-6.170.14-cp311-abi3-musllinux_1_2_i686.whl", hash = "sha256:2a8ed08715a9a27d3a4730f48c99133ef25b3561d8fe992d1eda91e22cf6be99", size = 1381908, upload-time = "2026-10-30T06:50:58.685Z" }, + { url = "https://files.pythonhosted.org/packages/ad/52/b044ccaf1cb9d89f319b737f1fee7cc83e16c2fd086e1bf1ce149092eb44/hypothesis-6.170.14-cp311-abi3-musllinux_1_2_ppc64le.whl", hash = "sha256:71081b46714e95e9a10d922bedd08d5a6e49835bbc2ef37a81d30d62f8596829", size = 1277802, upload-time = "2026-10-30T06:51:53.644Z" }, + { url = "https://files.pythonhosted.org/packages/27/2f/989ef755683c7cefd195670861a2fbd72b6b7952c124256152e94660e62c/hypothesis-6.170.14-cp311-abi3-musllinux_1_2_riscv64.whl", hash = "sha256:ce563cb34dedc9163811b03ce05a955eededd6f429db7c80c5c31d5836a9c05a", size = 1293571, upload-time = "2026-10-30T06:50:03.076Z" }, + { url = "https://files.pythonhosted.org/packages/a5/00/7426641c28dfc52e0a0d32748a1e47d3c189f9cf7eef81c0ecb7e90eb4cc/hypothesis-6.170.14-cp311-abi3-musllinux_1_2_x86_64.whl", hash = "sha256:06a752e3e879f1b69dda9ea203d4e8e1562cf72f694513f01c8c99d2e5e3f3cc", size = 1333329, upload-time = "2026-10-30T06:50:50.362Z" }, + { url = "https://files.pythonhosted.org/packages/7b/55/88307f535bb0914824b0eb2c6b58aa4a482b483c8d25d8022e7b4f641f8c/hypothesis-6.170.14-cp311-abi3-win32.whl", hash = "sha256:0f418c6bff7cd8a117dd0f526c20f100c32df1ff9f7bc8e01a5eeb3855278039", size = 687305, upload-time = "2026-10-30T06:50:43.079Z" }, + { url = "https://files.pythonhosted.org/packages/25/61/ba5bb3e8593afa9212fad51dc70f219ef6da7527d806c62a42c1cf12247b/hypothesis-6.170.14-cp311-abi3-win_amd64.whl", hash = "sha256:1e9f7619dff3072685e4f36e8ae7b134a0c1ca41074fc54ade81e5355c72cd8c", size = 693378, upload-time = "2026-10-30T06:51:36.729Z" }, + { url = "https://files.pythonhosted.org/packages/ab/72/482a68ae0e94b24acd964196cca830d1dd65efbf3d4821599d1c9527f8e2/hypothesis-6.170.14-cp311-abi3-win_arm64.whl", hash = "sha256:7dbaec9a62b7792b0b04b7ebc933aeafd3ff019b81d67d763c36ebcf4d56f864", size = 691696, upload-time = "2026-10-30T06:50:32.53Z" }, + { url = "https://files.pythonhosted.org/packages/8e/41/aa2c8ffc1490f02ec2b0e283e0856a0822c2b0707a83831e286493f888f5/hypothesis-6.170.14-cp312-cp312-macosx_10_12_x86_64.whl", hash = "sha256:76b03ded37a7eb14d01b03edca02b9a6ce6b8ab0ad8d2f9ac5f2e9de6769a33a", size = 801893, upload-time = "2026-10-30T06:51:10.965Z" }, + { url = "https://files.pythonhosted.org/packages/81/45/a82212e5773214d3ad1e0cee626098c1068adc93cdb37449e3c0629aa591/hypothesis-6.170.14-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:a6055172afdca41cda869652bd2f2c24ec984005cd16bbec5a8f38483f23b9bc", size = 793892, upload-time = "2026-10-30T06:50:55.958Z" }, + { url = "https://files.pythonhosted.org/packages/8a/cc/1889082d37e946e006845163a91e4091d1cff4f1692e3580729454cb02f4/hypothesis-6.170.14-cp312-cp312-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:b6490c64811a2176a5af9fe5a5d59ab2aa20d35d691328cbb336f6fb398355c4", size = 1123577, upload-time = "2026-10-30T06:50:11.524Z" }, + { url = "https://files.pythonhosted.org/packages/7f/7a/311a4ec14dbde1ce82cbd4d8957665c4d68e9139524b10e0a622db98bc4e/hypothesis-6.170.14-cp312-cp312-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:cde7e98694a9baa1d9761c1b595b3d617467c876200c0798a40f2016a0e85304", size = 1167206, upload-time = "2026-10-30T06:50:05.235Z" }, + { url = "https://files.pythonhosted.org/packages/f5/0d/1afb5e9f470898cf296f10545cfbf76e6b8d4d7025e0c41a491bbe969a22/hypothesis-6.170.14-cp312-cp312-musllinux_1_2_aarch64.whl", hash = "sha256:7e39d46bdebb3853fb3000aab3c354e4d780705dc8b2674773ffcdfd04fd5122", size = 1298455, upload-time = "2026-10-30T06:51:30.733Z" }, + { url = "https://files.pythonhosted.org/packages/f5/f6/ff7e9568f7a7d9ec375c68636c66e1b1a3baa06f44652535ff434110f6cd/hypothesis-6.170.14-cp312-cp312-musllinux_1_2_x86_64.whl", hash = "sha256:1309925a71a2baacb85ed5c20291ddb4b2a4665bd33fb594697463d457374a1b", size = 1332876, upload-time = "2026-10-30T06:50:57.14Z" }, + { url = "https://files.pythonhosted.org/packages/cc/73/fd330eec15026dc30d0ec8c658517c99703774098305632efb3e523b2f38/hypothesis-6.170.14-cp312-cp312-win_amd64.whl", hash = "sha256:ac98e833bdefbcbebb9094180a7431a8946f1acc91a5b779c814aecca71c7345", size = 691003, upload-time = "2026-10-30T06:50:36.624Z" }, + { url = "https://files.pythonhosted.org/packages/88/95/9729cc49645d07211ce8158f0bb51860873b3b99fb1650c2dad78bd2e86c/hypothesis-6.170.14-cp313-cp313-macosx_10_12_x86_64.whl", hash = "sha256:7d5e5affa9e69b8673f15bef86421740945e74f3c31d0817141ba020c7e00705", size = 801888, upload-time = "2026-10-30T06:50:06.456Z" }, + { url = "https://files.pythonhosted.org/packages/6e/ac/687277e019d68b96d2c33b979ad8a955b5747e279891f8c335bbf58cd9bc/hypothesis-6.170.14-cp313-cp313-macosx_11_0_arm64.whl", hash = "sha256:e7ae9ff3220659c4721acb58d34e72227a0af7747fc903ae177dfdb0323796b4", size = 793879, upload-time = "2026-10-30T06:51:09.51Z" }, + { url = "https://files.pythonhosted.org/packages/b2/1a/b46a2959b41ff9409af8b43fd63b07c6ed51958db0e9d314db9ee6012f5a/hypothesis-6.170.14-cp313-cp313-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:857932ff6bba183af0236f659b042023ab8b4ad58725b9024bcaba941c8cd5d0", size = 1123483, upload-time = "2026-10-30T06:50:47.998Z" }, + { url = "https://files.pythonhosted.org/packages/cb/2a/6735b2edff23d045441b5917f35c01a4c139a186604847ca19bbd63e81e2/hypothesis-6.170.14-cp313-cp313-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:8126e8208acadfc3c92d0aedb8011cf724fb59e5523ceaca3f4612dcc409e632", size = 1167158, upload-time = "2026-10-30T06:51:26.528Z" }, + { url = "https://files.pythonhosted.org/packages/ea/f0/b5b4273568620080db24dda2008236bd6c64599ab03210e46977b44b54d5/hypothesis-6.170.14-cp313-cp313-musllinux_1_2_aarch64.whl", hash = "sha256:d28ab5c5b08b1de11bc022d576b20f9f585f7e7758778295461f80b9222d7a18", size = 1298440, upload-time = "2026-10-30T06:50:10.195Z" }, + { url = "https://files.pythonhosted.org/packages/2a/c7/38427a9803dac9bdd06581061a84672107d821f6d91b833dec467c37f5d5/hypothesis-6.170.14-cp313-cp313-musllinux_1_2_x86_64.whl", hash = "sha256:033aca79fd3f9a6e15402dfddbc988141d43eae38d26a5fc554f6879f7e0e5f9", size = 1332909, upload-time = "2026-10-30T06:51:17.244Z" }, + { url = "https://files.pythonhosted.org/packages/7a/4f/47c060c96d077750c8a7b83b2e50b1add2050b391e77f49126828c125264/hypothesis-6.170.14-cp313-cp313-win_amd64.whl", hash = "sha256:03831d833b161b9b7a32f0b607205293954f3a651196c80138804b9479845d71", size = 690983, upload-time = "2026-10-30T06:51:04.201Z" }, + { url = "https://files.pythonhosted.org/packages/e8/3a/7ce36c91aafb192e08d990b3a8fd632be113eeb43973a94fddaa4172c570/hypothesis-6.170.14-cp314-cp314-macosx_10_12_x86_64.whl", hash = "sha256:a2410c5a6dcc011338e53cd381bd32d979b6f06d799c5cf01d76b923f0a0670b", size = 802098, upload-time = "2026-10-30T06:51:48.445Z" }, + { url = "https://files.pythonhosted.org/packages/c7/dc/17bc4ba00a48976944918bb5d7207519ed206fdf98af0d0319e5ba33afd2/hypothesis-6.170.14-cp314-cp314-macosx_11_0_arm64.whl", hash = "sha256:dcf6303c65e73acf134b7854cbdfec8f2108cfdc84806fa180de04d7946aab0c", size = 793921, upload-time = "2026-10-30T06:50:07.623Z" }, + { url = "https://files.pythonhosted.org/packages/1c/c3/252db9362dab21e3c3351abb89e715fa1ef598ebaf30abb9c9b1dd3b3aab/hypothesis-6.170.14-cp314-cp314-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:67d17e8b91d3913e7aa2b1179957df4e0634caea3868b5ba4fe2bf1945334134", size = 1123996, upload-time = "2026-10-30T06:51:05.628Z" }, + { url = "https://files.pythonhosted.org/packages/f8/41/573468979e9579eb6df937e6a96c1a944dfe2765c73624156571e3cc4b34/hypothesis-6.170.14-cp314-cp314-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:96f9f8a684c88ad2280129b28bc609adb6b31cd63b1b83b397b86d552391e947", size = 1167330, upload-time = "2026-10-30T06:51:23.741Z" }, + { url = "https://files.pythonhosted.org/packages/f6/1c/d1b790ec3247966ceb3d5729bb79f810e24b1684b49b6290c9ad5be05e33/hypothesis-6.170.14-cp314-cp314-musllinux_1_2_aarch64.whl", hash = "sha256:7c42d504d9bc0aef71fa88877c68ce202c02242c84f05791be2785e01e3a8699", size = 1298940, upload-time = "2026-10-30T06:51:27.68Z" }, + { url = "https://files.pythonhosted.org/packages/f6/a3/2fb6f7e825c9d943c5c35e0d9ef968e416267db2ef710e6d2d3cf7f36ab0/hypothesis-6.170.14-cp314-cp314-musllinux_1_2_x86_64.whl", hash = "sha256:ad84ca98f38e9f385b0f616df04bea4a9c6849333e221ce9289e9566ccd41058", size = 1333102, upload-time = "2026-10-30T06:51:20.993Z" }, + { url = "https://files.pythonhosted.org/packages/49/09/b533e4ac06dc317314d574dd0e3b548bac9cffe05e7e7a32f7b25aa7f052/hypothesis-6.170.14-cp314-cp314-pyemscripten_2026_0_wasm32.whl", hash = "sha256:f569bd9745835f9864d53c853e401c2021e796fdbb385497fdae633fe950b6ba", size = 633715, upload-time = "2026-10-30T06:50:25.715Z" }, + { url = "https://files.pythonhosted.org/packages/08/c6/9c47f183e5a8a4059d7c1c39e85f7b140f4416961b5ed0dac5d35dc6e1f7/hypothesis-6.170.14-cp314-cp314-win_amd64.whl", hash = "sha256:31c6a4015c02c45ba0285ff48538f4eb2c2de7e836f313bb8832fd8d86f71d62", size = 690702, upload-time = "2026-10-30T06:50:41.998Z" }, + { url = "https://files.pythonhosted.org/packages/05/47/623b189cc8d9e360bfbb69828fb88ca4d0d4996b3975f3191b49a50f04ce/hypothesis-6.170.14-cp314-cp314t-macosx_10_12_x86_64.whl", hash = "sha256:744c587791b2c5e38d1da2eaca617b1a82073832dfcf42975ddb03e93f187919", size = 800266, upload-time = "2026-10-30T06:50:29.462Z" }, + { url = "https://files.pythonhosted.org/packages/ea/af/7ec2697869df2c7f5d1c414b1a0a21746aa0298d41d7358e45fb7ed83992/hypothesis-6.170.14-cp314-cp314t-macosx_11_0_arm64.whl", hash = "sha256:d0daded6c2f3e05f6bd989059bb798e570722802e05eb3543c6f5d7a1e7e6500", size = 792500, upload-time = "2026-10-30T06:50:51.878Z" }, + { url = "https://files.pythonhosted.org/packages/e0/cf/2c7b13b5e56cd42789dbef4948f21e685c4a4db7dd6daffa253f261a9193/hypothesis-6.170.14-cp314-cp314t-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:986eaec8ca5a22710b1086b88734035c7d7ffcdb4ce02341b9f7069cc1bdfc5e", size = 1122469, upload-time = "2026-10-30T06:51:31.981Z" }, + { url = "https://files.pythonhosted.org/packages/63/17/aab31d83a3195edf21ff0b05ea8fabf1fcf723517a6ba7bd1aee3b2713e3/hypothesis-6.170.14-cp314-cp314t-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:44c481115770bb3df5014b0369ea1aaf7319b81b3346a744a9265fc933190c3d", size = 1166369, upload-time = "2026-10-30T06:50:54.572Z" }, + { url = "https://files.pythonhosted.org/packages/8b/ce/88ab370d5c1e06c74b87995381c28f5c1456bcd832db339bb93b6b15cae5/hypothesis-6.170.14-cp314-cp314t-musllinux_1_2_aarch64.whl", hash = "sha256:261ae20a5573cf59e5dbf931c7fb9abcb012695adccb61fdab108cbba1772ee5", size = 1297198, upload-time = "2026-10-30T06:51:22.283Z" }, + { url = "https://files.pythonhosted.org/packages/06/03/5c8d82f48d3a7adb593d1c141fdae597ef874e27dabd94ede9edd5c36c29/hypothesis-6.170.14-cp314-cp314t-musllinux_1_2_x86_64.whl", hash = "sha256:833f7e6d93c03f3c00583a126e61f631d6b91a84e157a30da2d1c786ce5658b7", size = 1332114, upload-time = "2026-10-30T06:51:34.61Z" }, + { url = "https://files.pythonhosted.org/packages/25/24/b08d06042af15e1a5fd2fe378feaa2336983f4ecc2bf645b38c6accf7d7d/hypothesis-6.170.14-cp314-cp314t-win_amd64.whl", hash = "sha256:224cad0e7ba8bfc6ef708652a3cc2148161e3157d191eb77ecc85dc5d16afb70", size = 690815, upload-time = "2026-10-30T06:51:43.881Z" }, + { url = "https://files.pythonhosted.org/packages/ee/72/bbd7a5341e23b56759937cef4fe83ea9c907f15ffccee1c2e8e2c8937546/hypothesis-6.170.14-cp315-abi3.abi3t-macosx_10_12_x86_64.whl", hash = "sha256:e76877e7c7a5cf8c4fbf98280dc6582f5d6450a73494490e8b48b7d03fb5e12f", size = 799866, upload-time = "2026-10-30T06:50:12.929Z" }, + { url = "https://files.pythonhosted.org/packages/df/f0/be3b05a4c57bbb91b0048c37a5f3a278accb1ec51e853bb87037379c3f93/hypothesis-6.170.14-cp315-abi3.abi3t-macosx_11_0_arm64.whl", hash = "sha256:5a613867bd2675101dda8fda380c3ce6be47842bed5640e1ba7b6b6fd0ffe8a3", size = 792315, upload-time = "2026-10-30T06:50:37.847Z" }, + { url = "https://files.pythonhosted.org/packages/aa/2b/f0a1509d5d4ad2e3b3c834bebd8a4b1f430d01d350710f45338342756b91/hypothesis-6.170.14-cp315-abi3.abi3t-manylinux1_i686.manylinux_2_5_i686.whl", hash = "sha256:a2249f2428ea94c47bccceb9fb8beada242c577dd4b3cd2edef5fb7160d1b71b", size = 1158319, upload-time = "2026-10-30T06:50:46.745Z" }, + { url = "https://files.pythonhosted.org/packages/6c/5c/eea759905d4b7d4d9a750b9b61a419120742143288c48c7d0ba633efc5f1/hypothesis-6.170.14-cp315-abi3.abi3t-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:20df34cca1265c1683ccdcab5872cf0a023f40235a3092a96d4bd81e4fc57986", size = 1121221, upload-time = "2026-10-30T06:50:28.317Z" }, + { url = "https://files.pythonhosted.org/packages/93/b0/3c90a9e734aa0a4aef0fa9004dd7c2bcd27a6e66bc32a8690647639cba88/hypothesis-6.170.14-cp315-abi3.abi3t-manylinux2014_armv7l.manylinux_2_17_armv7l.whl", hash = "sha256:4a7f8023efbec61296fd20edb35d638ec89ff761d3f9d3dc8128ac8f204652ec", size = 1152108, upload-time = "2026-10-30T06:50:35.02Z" }, + { url = "https://files.pythonhosted.org/packages/63/fd/4b7e7f69ee248547f6fe69475197ffbb5ae9563ab49c1430f020bd1d5493/hypothesis-6.170.14-cp315-abi3.abi3t-manylinux2014_ppc64le.manylinux_2_17_ppc64le.whl", hash = "sha256:3faa2b45f30320fb57ce1c30f1bf9f08b1fd1cc1c606ae2a3e1efd0256fd7c77", size = 1145869, upload-time = "2026-10-30T06:51:54.841Z" }, + { url = "https://files.pythonhosted.org/packages/15/de/dc96bc0646e8543386e3c539fea154075b2d0f40e6b3fb19022c43058dac/hypothesis-6.170.14-cp315-abi3.abi3t-manylinux2014_s390x.manylinux_2_17_s390x.whl", hash = "sha256:5be7a2239adde6c72001eb42cdb924d3c3856b1ee80342049961efdd6c72d064", size = 1199697, upload-time = "2026-10-30T06:51:40.612Z" }, + { url = "https://files.pythonhosted.org/packages/f5/fd/363472210e3ed2a356b7bc6ca8ca405c0dae6531ad43c8f76ca20a946884/hypothesis-6.170.14-cp315-abi3.abi3t-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:1a7aa883cd30c6c0c7259f34c1e3f37617c0d52fa6718e93c9926cc49b8047d6", size = 1165533, upload-time = "2026-10-30T06:50:14.401Z" }, + { url = "https://files.pythonhosted.org/packages/37/3f/989ab6da164efe574399debdf64370d153f17d4cfda51c8d7e14781dc92b/hypothesis-6.170.14-cp315-abi3.abi3t-manylinux_2_31_riscv64.whl", hash = "sha256:79360d1dd671e602d2327e26fb9179645511adcfb9ec9d28c5d6bda313920aa2", size = 1119844, upload-time = "2026-10-30T06:51:08.317Z" }, + { url = "https://files.pythonhosted.org/packages/5d/c4/7758868c29d3fbcdd0face967a2d7e3af221a1c8d57802c7ac6cb49becd9/hypothesis-6.170.14-cp315-abi3.abi3t-musllinux_1_2_aarch64.whl", hash = "sha256:e4c45885e931a55c5097ebf0a9b65459878ad5e1c69c0448ae37e03fd764ddfc", size = 1296008, upload-time = "2026-10-30T06:50:40.598Z" }, + { url = "https://files.pythonhosted.org/packages/c7/b2/dc88869b6f8508fbfb290c4a9b03328142743d041ae4a144870b904b69bd/hypothesis-6.170.14-cp315-abi3.abi3t-musllinux_1_2_armv7l.whl", hash = "sha256:a6d09601401c019366826effa98ae64393dd149b50d7d6ea202ad472784278a8", size = 1425148, upload-time = "2026-10-30T06:51:18.492Z" }, + { url = "https://files.pythonhosted.org/packages/f2/f2/f6dde97e8a1d017d9663164f48f600f65e9d2ade37c43447421f6a369cbe/hypothesis-6.170.14-cp315-abi3.abi3t-musllinux_1_2_i686.whl", hash = "sha256:05e2a645b94fd43107a3360fe0aaf6db268199c80a43cf375fac5f1f9047745f", size = 1378730, upload-time = "2026-10-30T06:51:16.063Z" }, + { url = "https://files.pythonhosted.org/packages/9b/42/e2aa93586527428b51c019a4cedf28e1d2f10dfec5f9acfa3e1a22cd2e86/hypothesis-6.170.14-cp315-abi3.abi3t-musllinux_1_2_ppc64le.whl", hash = "sha256:cb54946f70ae5008ea8cd58ce9a27b2126506732b53b83a6d117371d7c755362", size = 1274346, upload-time = "2026-10-30T06:51:14.2Z" }, + { url = "https://files.pythonhosted.org/packages/c2/22/17ee8d38c5cb0a948ba9bb6cd7be2351a021cf1ea6686aa4fac16c29c45a/hypothesis-6.170.14-cp315-abi3.abi3t-musllinux_1_2_riscv64.whl", hash = "sha256:b72805cf56012a3d96f4a1fbb869f178d448f242b4f64310956134a3c952113f", size = 1291604, upload-time = "2026-10-30T06:50:53.304Z" }, + { url = "https://files.pythonhosted.org/packages/0c/c8/c97cdaa1191519fadcc06bc118f271910d7d7972caeecb70b7003a62303c/hypothesis-6.170.14-cp315-abi3.abi3t-musllinux_1_2_x86_64.whl", hash = "sha256:a17500680ec2c3cf006c12d3832a6c8daba8a18386f93b3d1828de07b07910aa", size = 1331507, upload-time = "2026-10-30T06:51:02.784Z" }, + { url = "https://files.pythonhosted.org/packages/cf/c7/1007bdaf517b086df3c54d06de6c47f099704fd8bd6c2b5013e749833ee4/hypothesis-6.170.14-cp315-abi3.abi3t-win32.whl", hash = "sha256:b1d9baaef909100fd1a6ba5457ecffa4a184e820c7f0bd3d4589f14870f5cade", size = 684627, upload-time = "2026-10-30T06:50:49.194Z" }, + { url = "https://files.pythonhosted.org/packages/51/d1/68539ecbe240de4afc03c5fa120ecf0875690708f2a97914352ae3172d36/hypothesis-6.170.14-cp315-abi3.abi3t-win_amd64.whl", hash = "sha256:a725b854c5df088c5a579f1c323657258abe5d8e32e16cd07beb48960f31c070", size = 690617, upload-time = "2026-10-30T06:51:42.268Z" }, + { url = "https://files.pythonhosted.org/packages/66/d7/9671c708cdab1c49ff96344f6730ccd76a7e1c2bdbbb2bed93bc9d6a18b4/hypothesis-6.170.14-cp315-abi3.abi3t-win_arm64.whl", hash = "sha256:75f60ef38c010742339a9b2ec2a44cb0aed6c6c8f1455ad6a6de48f5216a72b2", size = 688595, upload-time = "2026-10-30T06:51:37.945Z" }, + { url = "https://files.pythonhosted.org/packages/ba/ae/822ac5f03f0f13e96b4a7b380ede8fa4ae2a46730f4e5c6d7eca016147bb/hypothesis-6.170.14-pp312-pypy312_pp80-macosx_10_12_x86_64.whl", hash = "sha256:c7b5dee8b87f8a78c1786eb1f96a6c0c91fcc2e10b2e9d876dcbc8e3f65dffc3", size = 801000, upload-time = "2026-10-30T06:50:44.17Z" }, + { url = "https://files.pythonhosted.org/packages/fa/b2/e4a0901adb647ca375a794cbf1a23e5e49e59c77d191d1922e3d3c4ce6b6/hypothesis-6.170.14-pp312-pypy312_pp80-macosx_11_0_arm64.whl", hash = "sha256:e99d1bc7d8e8fc17cb67a38e003cbf4a17f41f7851c1f205efd35b6f3ba93da5", size = 792896, upload-time = "2026-10-30T06:51:25.028Z" }, + { url = "https://files.pythonhosted.org/packages/d1/9c/ac50eed9becb0de590d8af7bd1b67546e58b84dda4abb89ca563eed1babe/hypothesis-6.170.14-pp312-pypy312_pp80-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:d66dab3b79eda78964665ab4f9d7a269137c14cd6ec7ea4fb27002a8786452b9", size = 1122853, upload-time = "2026-10-30T06:51:33.262Z" }, + { url = "https://files.pythonhosted.org/packages/c4/7f/bc41905c23c74c3db4620a5382977b168156f357e8a421c2b9f66f764979/hypothesis-6.170.14-pp312-pypy312_pp80-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:88d829c29907e7d86f8b1d33041bb73814062a3681ec51dcb231f64874f88b14", size = 1167065, upload-time = "2026-10-30T06:51:45.442Z" }, + { url = "https://files.pythonhosted.org/packages/0f/17/b28b4348a95694562357dc0c8ecb521dd607051dddd7af158c754b895710/hypothesis-6.170.14-pp312-pypy312_pp80-win_amd64.whl", hash = "sha256:90d4e85f22c3154a59df851bb2c59c89c129fba3155125d98ebad80bf69aad26", size = 692119, upload-time = "2026-10-30T06:51:19.8Z" }, +] + [[package]] name = "iniconfig" version = "2.3.1" @@ -168,6 +251,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/82/3b/64d4899d73f91ba49a8c18a8ff3f0ea8f1c1d75481760df8c68ef5235bf5/rich-15.0.0-py3-none-any.whl", hash = "sha256:33bd4ef74232fb73fe9279a257718407f169c09b78a87ad3d296f548e27de0bb", size = 310654, upload-time = "2026-04-12T08:24:02.83Z" }, ] +[[package]] +name = "sortedcontainers" +version = "2.4.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/e8/c4/ba2f8066cceb6f23394729afe52f3bf7adec04bf9ed2c820b39e19299111/sortedcontainers-2.4.0.tar.gz", hash = "sha256:25caa5a06cc30b6b83d11423433f65d1f9d76c4c6a0c90e3379eaa43b9bfdb88", size = 30594, upload-time = "2021-05-16T22:03:42.897Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/32/46/9cb0e58b2deb7f82b84065f37f3bffeb12413f947f9388e4cac22c4621ce/sortedcontainers-2.4.0-py2.py3-none-any.whl", hash = "sha256:a163dcaede0f1c021485e957a39245190e74249897e2ae4b2aa38595db237ee0", size = 29575, upload-time = "2021-05-16T22:03:41.177Z" }, +] + [[package]] name = "textual" version = "8.2.8"