From 5038148686b0c5265604fca287703f91b89c3a61 Mon Sep 17 00:00:00 2001 From: Per-Arne Hellarvik Date: Mon, 2 Nov 2026 21:45:06 +0100 Subject: [PATCH] Add network editing with VLAN, firewall, IPv4/IPv6 and undo Press n in the info screen to edit one interface: bridge, VLAN tag, firewall, IPv4 (DHCP, static or none) and IPv6 (SLAAC, DHCP, static or none). The netN string is parsed, the chosen fields are changed and the string is written back, so hwaddr, model=MAC, mtu, rate, trunks and type are kept. For VMs, IP settings go to cloud-init ipconfigN and only when the VM has a cloud-init drive. A confirm shows the old and new raw values before anything is sent. Press u to restore the values from before the last change to that guest in this session. Review fixes: - Undo entries are cleared on a context switch, because VMIDs repeat across clusters. - has_cloudinit only looks at disks. - CIDR needs a decimal prefix: no netmask form and no /0. - Fields you did not touch are not rewritten. Live test on CT 104: firewall, VLAN 99, IPv6 DHCP and static IPv4 were each applied and undone, and net0 ended identical to the baseline. Going from static back to DHCP left the CT without IPv4 until a reboot, so the confirm now warns that IP changes on a running container may need one. Co-Authored-By: Claude Opus 5.5 --- .gitignore | 1 + README.md | 3 + docs/plans/guest-tui-mvp.md | 100 +++++++ pyproject.toml | 6 +- src/cv4pve_tui/app.py | 274 +++++++++++++++++- src/cv4pve_tui/cli.py | 105 +++++++ tests/test_app.py | 550 +++++++++++++++++++++++++++++++++++- tests/test_cli.py | 302 ++++++++++++++++++++ uv.lock | 92 ++++++ 9 files changed, 1422 insertions(+), 11 deletions(-) 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"