Network editing: bridge, VLAN, firewall, IPv4/IPv6 (SLAAC), undo #4

Merged
Duckboot merged 1 commits from feat/network-edit into main 2026-11-02 21:45:59 +01:00
9 changed files with 1422 additions and 11 deletions

No files matched your search

+1
View File
@@ -2,3 +2,4 @@
__pycache__/
*.pyc
.pytest_cache/
.claude/scratch/
+3
View File
@@ -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@<node> pct enter <vmid>`) |
| `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
+100
View File
@@ -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=<string>[,bridge][,firewall=<1|0>][,gw=<IPv4>][,gw6=<IPv6>]
[,host-managed][,hwaddr][,ip=<IPv4/CIDR|dhcp|manual>][,ip6=<IPv6/CIDR|auto|dhcp|manual>]
[,link_down][,mtu][,rate][,tag=<int>][,trunks][,type=veth]`. `ip6=auto` = SLAAC.
Applied immediately on a running CT.
- VM `net[n]`: `[model=]<enum>[,bridge][,firewall=<1|0>][,link_down][,macaddr][,mtu]
[,queues][,rate][,tag=<int>][,trunks][,<model>=<macaddr>]`. Applied immediately.
- VM `ipconfig[n]`: `[gw=<IPv4>][,gw6=<IPv6>][,ip=<IPv4/CIDR|dhcp>][,ip6=<IPv6/CIDR|dhcp|auto>]`
(cloud-init; "no explicit gateway with dhcp"). Applies at the next boot.
- `delete=<list>` removes settings (used when `ipconfigN` ends up empty).
- `get node network <node> -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]=<cidr>`,
`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 <kind> config <node> <vmid> [--<key> <value> ...] [--delete <k1,k2>] --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).
+5 -1
View File
@@ -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"]
+265 -9
View File
@@ -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:
+105
View File
@@ -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"
)
+549 -1
View File
@@ -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()
+302
View File
@@ -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)
Generated
+92
View File
@@ -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"