Edit autostart, startup order and tags #30

Merged
Duckboot merged 1 commits from feat/boot-options-18 into main 2026-11-03 01:33:14 +01:00
6 changed files with 634 additions and 0 deletions

No files matched your search

+1
View File
@@ -44,6 +44,7 @@ uv run cv4pve-tui
| `Enter` then `e` | in info: edit cores / memory / grow a disk (disk growth asks again; it cannot be undone) | | `Enter` then `e` | in info: edit cores / memory / grow a disk (disk growth asks again; it cannot be undone) |
| `Enter` then `b` | in info: backup now (snapshot, zstd, storage choice); `a` there = also show backups of earlier guests that had the same VMID | | `Enter` then `b` | in info: backup now (snapshot, zstd, storage choice); `a` there = also show backups of earlier guests that had the same VMID |
| `Enter` 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) | | `Enter` 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) |
| `Enter` then `o` | in info: edit autostart, startup order (order, up and down delay) and tags (one confirmed `set config` call; empty fields remove the setting) |
| `Enter` then `p` / `R` / `D` | in info: take a snapshot (name, description, VM: include RAM) / roll back to the snapshot under the cursor (confirms; starts the guest again if it is running) / delete the snapshot under the cursor (confirms) | | `Enter` then `p` / `R` / `D` | in info: take a snapshot (name, description, VM: include RAM) / roll back to the snapshot under the cursor (confirms; starts the guest again if it is running) / delete the snapshot under the cursor (confirms) |
| `t` | shell in a running LXC container (`ssh -t root@<node> pct enter <vmid>`) | | `t` | shell in a running LXC container (`ssh -t root@<node> pct enter <vmid>`) |
| `l` | tasks: recent cluster tasks (start, duration, node, type, VMID, user, status; `!` marks failures); `Enter` shows the log of the task under the cursor, `g` reloads. In info: `l` = only this guest's tasks | | `l` | tasks: recent cluster tasks (start, duration, node, type, VMID, user, status; `!` marks failures); `Enter` shows the log of the task under the cursor, `g` reloads. In info: `l` = only this guest's tasks |
+40
View File
@@ -657,3 +657,43 @@ CLI facts (read-only probes):
- Pure function `guest_matches(guest: dict, query: str) -> bool` in `app.py` (or - Pure function `guest_matches(guest: dict, query: str) -> bool` in `app.py` (or
`cli.py`-free module-level helper) so it can be tested without the UI. `cli.py`-free module-level helper) so it can be tested without the UI.
---
# Autostart, startup order and tags (issue #18)
## Decision
Shown on the info screen; edited with key `o` (options) there. LXC and VM.
CLI facts (read-only probes, PVE 9.2):
- Config keys: `onboot` (0/1, absent = 0), `startup` (`[[order=]N][,up=S][,down=S]`,
absent = none), `tags` (`;`-separated, e.g. `172.16.41.200;community-script;os`).
- Write with `set <kind> config <node> <vmid> --onboot 0|1 --startup <s> --tags <t>
[--delete startup,tags] --wait` - i.e. the existing `cli.set_net(guest, changes)` (one
call; `None` values go to `--delete`).
- DANGER: `delete ct|vm ... tags` exists but shares the guest-destroying `delete ct|vm
<node> <vmid>` form. Never use it; removing tags = `--delete tags` on `set config`.
### `cli.py`
- `startup_value(order, up, down) -> str | None` - ints >= 0 or None; builds
`order=N[,up=S][,down=S]` (only the given parts); all None -> None. ValueError for
negatives/non-ints. `parse_startup(s) -> (order, up, down)` (accepts a bare leading
number as order).
- `tags_value(text) -> str | None` - split on `;`, `,` or whitespace, drop empties and
duplicates (order kept); each tag must match `[A-Za-z0-9_][A-Za-z0-9_+.-]*` else
ValueError; joined with `;`; none -> None.
### `app.py` (UI contract)
- Info screen Resources table gets rows "Autostart" (`yes`/`no`), "Startup" (e.g.
`order 2, up 10 s, down 30 s` or `-`), "Tags" (`a, b, c` or `-`); In use `-`.
- `o`: `BootScreen` dialog (DialogScreen style, title "Boot and tags: <name> (<vmid>)"):
`Checkbox#onboot`, `Input#order`, `Input#up`, `Input#down` (empty = not set),
`Input#tags` (prefilled `a; b; c`), buttons `#apply`/`#cancel`, escape = cancel.
Apply: build changes for keys whose value differs (`onboot` "1"/"0", `startup` string
or None, `tags` string or None); invalid -> error notify, dialog stays, no call; no
change -> info notify; else `ConfirmScreen` titled "Boot and tags" listing old -> new
per key; yes -> busy guard, `cli.set_net(guest, changes)` via start_job, info reloads.
+118
View File
@@ -1,5 +1,6 @@
"""Textual app for managing Proxmox guests through cv4pve-cli.""" """Textual app for managing Proxmox guests through cv4pve-cli."""
import re
import subprocess import subprocess
import sys import sys
from datetime import datetime from datetime import datetime
@@ -419,6 +420,95 @@ class NetScreen(DialogScreen):
) )
class BootScreen(DialogScreen):
"""Edit autostart, startup order and tags; confirms, then sets them in one call."""
BINDINGS = [("escape", "dismiss", "Cancel")] # noqa: RUF012 - Textual idiom
def __init__(self, guest: dict, config: dict) -> None:
super().__init__()
self.guest, self.config = guest, config
def compose(self) -> ComposeResult:
with self.box(f"Boot and tags: {self.guest['name']} ({self.guest['vmid']})"):
with Horizontal(classes="row"):
yield Label("Autostart")
yield Checkbox("", id="onboot")
for wid, label, hint in (
("order", "Order", "start order (empty = not set)"),
("up", "Up delay", "seconds to wait after start (empty = not set)"),
("down", "Down delay", "seconds to wait on shutdown (empty = not set)"),
("tags", "Tags", "a; b; c (empty = none)"),
):
with Horizontal(classes="row"):
yield Label(label)
yield Input(placeholder=hint, id=wid)
with Horizontal(classes="buttons"):
yield Button("Apply", id="apply", variant="primary")
yield Button("Cancel", id="cancel")
def on_mount(self) -> None:
self.query_one("#onboot", Checkbox).value = _autostart(self.config)
order, up, down = cli.parse_startup(self.config.get("startup"))
for wid, value in (("order", order), ("up", up), ("down", down)):
self.query_one(f"#{wid}", Input).value = "" if value is None else str(value)
self.query_one("#tags", Input).value = "; ".join(_tags(self.config))
def _number(self, wid: str) -> int | None:
text = self.query_one(f"#{wid}", Input).value.strip()
if text and not text.isdecimal():
raise ValueError(f"{wid}: a whole number >= 0, or empty")
return int(text) if text else None
def _changes(self) -> dict[str, str | None]:
"""Keys whose value differs from the config; ValueError on bad input."""
startup = cli.startup_value(*(self._number(w) for w in ("order", "up", "down")))
tags = cli.tags_value(self.query_one("#tags", Input).value)
onboot = self.query_one("#onboot", Checkbox).value
changes: dict[str, str | None] = {}
if onboot != _autostart(self.config):
changes["onboot"] = "1" if onboot else "0"
if startup != cli.startup_value(*cli.parse_startup(self.config.get("startup"))):
changes["startup"] = startup
if tags != cli.tags_value(";".join(_tags(self.config))):
changes["tags"] = tags
return changes
def on_button_pressed(self, event: Button.Pressed) -> None:
if event.button.id == "cancel":
self.dismiss()
return
try:
changes = 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 {self.config.get(k, '(not set)')}\n to {v or '(removed)'}"
for k, v in changes.items()
]
who = f"{self.guest['name']} ({self.guest['vmid']})"
self.app.push_screen(
ConfirmScreen(
"\n".join([f"Change boot options of {who}?", *lines]), "Boot and tags"
),
self._confirmed(changes),
)
def _confirmed(self, changes: dict[str, str | None]):
def done(ok: bool) -> None:
if ok:
self.dismiss()
self.app.start_job(
self.guest, "Boot options", lambda: cli.set_net(self.guest, changes)
)
return done
class TaskLogScreen(DialogScreen): class TaskLogScreen(DialogScreen):
"""The log of one task, as plain text (task output is never parsed as markup).""" """The log of one task, as plain text (task output is never parsed as markup)."""
@@ -621,6 +711,7 @@ class InfoScreen(ModalScreen):
("b", "backup", "Backup now"), ("b", "backup", "Backup now"),
("a", "toggle_all", "All backups"), ("a", "toggle_all", "All backups"),
("n", "net", "Edit network"), ("n", "net", "Edit network"),
("o", "boot", "Autostart, startup, tags"),
("u", "undo", "Undo network"), ("u", "undo", "Undo network"),
("e", "edit", "Edit resources"), ("e", "edit", "Edit resources"),
("p", "snapshot", "Take snapshot"), ("p", "snapshot", "Take snapshot"),
@@ -851,6 +942,10 @@ class InfoScreen(ModalScreen):
) )
return busy return busy
def action_boot(self) -> None:
if self.config and not self._refuse_if_busy():
self.app.push_screen(BootScreen(self.guest, self.config))
def action_net(self) -> None: def action_net(self) -> None:
if cli.nets(self.config) and not self._refuse_if_busy(): if cli.nets(self.config) and not self._refuse_if_busy():
self._open_net() self._open_net()
@@ -1587,9 +1682,32 @@ def _resources(config: dict, kind: str) -> list[tuple[str, str]]:
if kind == "ct" and config.get("swap") is not None: if kind == "ct" and config.get("swap") is not None:
rows.append(("Swap", f"{config['swap']} MiB")) rows.append(("Swap", f"{config['swap']} MiB"))
rows += [(f"Disk {name}", size) for name, size in cli.disks(config, kind).items()] rows += [(f"Disk {name}", size) for name, size in cli.disks(config, kind).items()]
order, up, down = cli.parse_startup(config.get("startup"))
startup = [
f"{label} {v}{unit}"
for label, v, unit in (
("order", order, ""),
("up", up, " s"),
("down", down, " s"),
)
if v is not None
]
rows += [
("Autostart", "yes" if _autostart(config) else "no"),
("Startup", ", ".join(startup) or "-"),
("Tags", ", ".join(_tags(config)) or "-"),
]
return rows return rows
def _autostart(config: dict) -> bool:
return str(config.get("onboot", 0)) == "1" # int or str, depending on the source
def _tags(config: dict) -> list[str]:
return [t for t in re.split(r"[;,\s]+", str(config.get("tags") or "")) if t]
def _pct(part, whole) -> float | None: def _pct(part, whole) -> float | None:
return part / whole * 100 if part is not None and whole else None return part / whole * 100 if part is not None and whole else None
+39
View File
@@ -390,6 +390,45 @@ def set_net(guest: dict, changes: dict[str, str | None]) -> None:
) )
def startup_value(order: int | None, up: int | None, down: int | None) -> str | None:
"""`order=N[,up=S][,down=S]` from the given parts; None when none is given."""
parts = []
for name, value in (("order", order), ("up", up), ("down", down)):
if value is None:
continue
if type(value) is not int or value < 0:
raise ValueError(f"startup {name} must be a whole number >= 0")
parts.append(f"{name}={value}")
return ",".join(parts) or None
def parse_startup(text: str | None) -> tuple[int | None, int | None, int | None]:
"""(order, up, down) of a startup string; a bare leading number is the order."""
got: dict[str, int] = {}
for i, part in enumerate((text or "").split(",")):
key, sep, value = part.partition("=")
if not sep and i == 0 and key.strip().isdecimal():
key, value = "order", key
if key.strip() in ("order", "up", "down") and value.strip().isdecimal():
got[key.strip()] = int(value)
return got.get("order"), got.get("up"), got.get("down")
_TAG = re.compile(r"[A-Za-z0-9_][A-Za-z0-9_+.-]*")
def tags_value(text: str) -> str | None:
"""Tags split on `;`, `,` or whitespace, deduplicated, `;`-joined; None when empty."""
tags = list(dict.fromkeys(re.split(r"[;,\s]+", text.strip())))
tags = [t for t in tags if t]
for tag in tags:
if not _TAG.fullmatch(tag):
raise ValueError(
f"invalid tag {tag!r}: letters, digits, _ + . - (not starting with + . -)"
)
return ";".join(tags) or None
def not_backed_up() -> set[int]: def not_backed_up() -> set[int]:
"""VMIDs no backup job covers (Proxmox's own answer, incl. pool-based jobs).""" """VMIDs no backup job covers (Proxmox's own answer, incl. pool-based jobs)."""
items = json.loads(run("get", "cluster", "not-backed-up", "-o", "json")) items = json.loads(run("get", "cluster", "not-backed-up", "-o", "json"))
+320
View File
@@ -3875,3 +3875,323 @@ def test_guest_matches_tolerates_none_status_and_type():
assert appmod.guest_matches(g, "x") assert appmod.guest_matches(g, "x")
assert not appmod.guest_matches(g, "is:running") assert not appmod.guest_matches(g, "is:running")
assert not appmod.guest_matches(g, "type:lxc") assert not appmod.guest_matches(g, "type:lxc")
# --- #18: autostart, startup order and tags (spec: "Autostart, startup order and tags") ---
CT_BOOT = {"onboot": 1, "startup": "order=2,up=10",
"tags": "172.16.41.200;community-script;os"} # fmt: skip
async def open_boot(app, pilot, down=1):
await open_info(app, pilot, down=down) # down=1: CT 100 "backrest"
await pilot.press("o")
await settle(app, pilot)
assert isinstance(app.screen, appmod.BootScreen)
def setb(app, **values):
for key, value in values.items():
w = app.screen.query_one(f"#{key}")
w.value = value
async def boot_to_confirm(app, pilot, **values):
setb(app, **values)
await press_button(app, pilot, "#apply")
assert isinstance(app.screen, ConfirmScreen)
return text_of(app.screen)
async def boot_change(app, pilot, **values):
await boot_to_confirm(app, pilot, **values)
await press_button(app, pilot, "#yes")
def any_text(screen, needle):
titles = [str(w.border_title or "") for w in screen.query("*")]
return needle in text_of(screen) or any(needle in t for t in titles)
@pytest.mark.parametrize("onboot", [1, "1"])
async def test_info_resources_show_autostart_startup_tags(
fake, onboot
): # Resources rows
fake.net_configs[100] = {**CT_BOOT, "onboot": onboot}
app = make_app()
async with app.run_test(size=SIZE) as pilot:
await open_info(app, pilot, down=1)
rows = resource_rows(app)
assert rows["Autostart"] == "yes"
assert rows["Startup"] == "order 2, up 10 s"
assert rows["Tags"] == "172.16.41.200, community-script, os"
used = in_use_rows(app)
assert used["Autostart"] == used["Startup"] == used["Tags"] == "-"
async def test_info_shows_full_startup_with_down(fake):
fake.net_configs[100] = {"startup": "order=2,up=10,down=30"}
app = make_app()
async with app.run_test(size=SIZE) as pilot:
await open_info(app, pilot, down=1)
assert resource_rows(app)["Startup"] == "order 2, up 10 s, down 30 s"
async def test_info_shows_defaults_for_guest_without_boot_keys(fake):
app = make_app()
async with app.run_test(size=SIZE) as pilot:
await open_info(app, pilot, down=1)
rows = resource_rows(app)
assert (rows["Autostart"], rows["Startup"], rows["Tags"]) == ("no", "-", "-")
async def test_info_shows_boot_rows_for_vm_too(fake): # LXC and VM
fake.net_configs[108] = {"onboot": 1, "startup": "up=5", "tags": "web"}
app = make_app()
async with app.run_test(size=SIZE) as pilot:
await open_info(app, pilot, down=0)
rows = resource_rows(app)
assert rows["Autostart"] == "yes"
assert rows["Startup"] == "up 5 s"
assert rows["Tags"] == "web"
async def test_o_opens_boot_screen_prefilled(fake):
fake.net_configs[100] = dict(CT_BOOT)
app = make_app()
async with app.run_test(size=SIZE) as pilot:
await open_boot(app, pilot)
s = app.screen
assert any_text(s, "Boot and tags: backrest (100)")
assert s.query_one("#onboot", Checkbox).value is True
assert s.query_one("#order", Input).value == "2"
assert s.query_one("#up", Input).value == "10"
assert s.query_one("#down", Input).value == "" # empty = not set
assert s.query_one("#tags", Input).value == (
"172.16.41.200; community-script; os"
)
s.query_one("#apply", Button)
s.query_one("#cancel", Button)
async def test_o_opens_boot_screen_empty_for_guest_without_keys(fake):
app = make_app()
async with app.run_test(size=SIZE) as pilot:
await open_boot(app, pilot)
s = app.screen
assert s.query_one("#onboot", Checkbox).value is False
for sel in ("#order", "#up", "#down", "#tags"):
assert s.query_one(sel, Input).value == ""
async def test_change_only_onboot_confirms_then_sets_once(fake):
fake.net_configs[100] = dict(CT_BOOT)
app = make_app()
busy_seen = []
fake.set_hook = lambda: busy_seen.append(set(app.busy))
async with app.run_test(size=SIZE) as pilot:
await open_boot(app, pilot)
before = len(fake.backups_calls)
text = await boot_to_confirm(app, pilot, onboot=False)
assert any_text(app.screen, "Boot and tags")
assert "onboot" in text.lower() or "autostart" in text.lower()
assert fake.set_calls == [] # nothing before "yes"
await press_button(app, pilot, "#yes")
assert fake.set_calls == [(100, {"onboot": "0"})]
assert busy_seen and busy_seen[0] # guest was busy during the call
assert app.busy == set() and errors(app) == []
assert isinstance(app.screen, appmod.InfoScreen)
assert len(fake.backups_calls) > before # info reloaded
assert resource_rows(app)["Autostart"] == "no"
async def test_enable_onboot_on_guest_without_keys_sends_string_one(fake):
app = make_app()
async with app.run_test(size=SIZE) as pilot:
await open_boot(app, pilot)
await boot_change(app, pilot, onboot=True)
assert fake.set_calls == [(100, {"onboot": "1"})]
async def test_change_order_keeps_other_startup_parts(fake):
fake.net_configs[100] = dict(CT_BOOT)
app = make_app()
async with app.run_test(size=SIZE) as pilot:
await open_boot(app, pilot)
await boot_change(app, pilot, order="5")
assert fake.set_calls == [(100, {"startup": "order=5,up=10"})]
assert resource_rows(app)["Startup"] == "order 5, up 10 s"
async def test_set_startup_on_guest_without_keys(fake):
app = make_app()
async with app.run_test(size=SIZE) as pilot:
await open_boot(app, pilot)
await boot_change(app, pilot, order="1", down="20")
assert fake.set_calls == [(100, {"startup": "order=1,down=20"})]
async def test_clearing_all_startup_fields_deletes_startup(fake):
fake.net_configs[100] = dict(CT_BOOT)
app = make_app()
async with app.run_test(size=SIZE) as pilot:
await open_boot(app, pilot)
await boot_change(app, pilot, order="", up="", down="")
assert fake.set_calls == [(100, {"startup": None})]
assert resource_rows(app)["Startup"] == "-"
async def test_adding_a_tag_sends_normalised_tag_string(fake):
fake.net_configs[100] = dict(CT_BOOT)
app = make_app()
async with app.run_test(size=SIZE) as pilot:
await open_boot(app, pilot)
text = await boot_to_confirm(
app, pilot, tags="172.16.41.200; community-script; os; tui-test"
)
assert "tui-test" in text
await press_button(app, pilot, "#yes")
assert fake.set_calls == [
(100, {"tags": "172.16.41.200;community-script;os;tui-test"})
]
assert resource_rows(app)["Tags"].endswith("tui-test")
async def test_clearing_tags_uses_set_config_delete_never_a_delete_command(
fake, monkeypatch
):
fake.net_configs[100] = dict(CT_BOOT)
app = make_app()
async with app.run_test(size=SIZE) as pilot:
await open_boot(app, pilot)
def forbidden(*args, **kwargs): # any CLI call outside the fake would land here
raise AssertionError(f"unexpected CLI call: {args!r}")
monkeypatch.setattr(cli, "run", forbidden)
monkeypatch.setattr(cli.subprocess, "run", forbidden)
await boot_change(app, pilot, tags="")
assert fake.set_calls == [
(100, {"tags": None})
] # only set_net, None -> --delete
assert fake.delete_calls == [] and fake.actions == [] and fake.sets == []
assert errors(app) == []
assert resource_rows(app)["Tags"] == "-"
async def test_combined_changes_go_in_one_set_net_call(fake):
fake.net_configs[100] = dict(CT_BOOT)
app = make_app()
async with app.run_test(size=SIZE) as pilot:
await open_boot(app, pilot)
await boot_change(app, pilot, onboot=False, order="7", tags="")
assert fake.set_calls == [
(100, {"onboot": "0", "startup": "order=7,up=10", "tags": None})
]
@pytest.mark.parametrize(
"values",
[
{"tags": "-x"},
{"tags": "a;b!"},
{"order": "-1"},
{"order": "abc"},
{"up": "-5"},
{"down": "1.5"},
],
)
async def test_invalid_boot_input_errors_without_call_and_dialog_stays(fake, values):
fake.net_configs[100] = dict(CT_BOOT)
app = make_app()
async with app.run_test(size=SIZE) as pilot:
await open_boot(app, pilot)
setb(app, **values)
await press_button(app, pilot, "#apply")
assert errors(app) and fake.set_calls == []
assert isinstance(app.screen, appmod.BootScreen)
assert app.busy == set()
async def test_boot_no_change_gives_info_notify_no_confirm_no_call(fake):
fake.net_configs[100] = dict(CT_BOOT)
app = make_app()
async with app.run_test(size=SIZE) as pilot:
await open_boot(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)
@pytest.mark.parametrize("how", ["cancel", "escape"])
async def test_cancel_or_escape_in_boot_screen_makes_no_call(fake, how):
fake.net_configs[100] = dict(CT_BOOT)
app = make_app()
async with app.run_test(size=SIZE) as pilot:
await open_boot(app, pilot)
setb(app, onboot=False, tags="x")
if how == "cancel":
await press_button(app, pilot, "#cancel")
else:
await pilot.press("escape")
await settle(app, pilot)
assert isinstance(app.screen, appmod.InfoScreen)
assert fake.set_calls == [] and app.busy == set()
async def test_declining_boot_confirm_makes_no_call(fake):
fake.net_configs[100] = dict(CT_BOOT)
app = make_app()
async with app.run_test(size=SIZE) as pilot:
await open_boot(app, pilot)
await boot_to_confirm(app, pilot, onboot=False)
await press_button(app, pilot, "#no")
assert fake.set_calls == [] and app.busy == set()
async def test_boot_apply_while_guest_busy_warns_and_makes_no_second_call(fake):
fake.net_configs[100] = dict(CT_BOOT)
started, release = threading.Event(), threading.Event()
fake.set_hook = lambda: (started.set(), release.wait(10))
app = make_app()
async with app.run_test(size=SIZE) as pilot:
await open_boot(app, pilot)
try:
await boot_to_confirm(app, pilot, onboot=False)
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 to_info(app, pilot)
await pilot.press("o")
await pilot.pause()
if isinstance(
app.screen, appmod.BootScreen
): # refusal may come at apply/yes
setb(app, tags="x")
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_boot_set_net_failure_notifies_error_and_releases_busy(fake):
fake.net_configs[100] = dict(CT_BOOT)
fake.set_error = cli_error("Error: unable to apply onboot")
app = make_app()
async with app.run_test(size=SIZE) as pilot:
await open_boot(app, pilot)
await boot_change(app, pilot, onboot=False)
assert any("unable to apply onboot" in m for m in errors(app))
assert app.busy == set() and app.is_running
assert len(fake.set_calls) == 1
+116
View File
@@ -1388,3 +1388,119 @@ def test_version_argv_and_strips(fake):
def test_tested_version_constant(): def test_tested_version_constant():
assert cli.TESTED_VERSION == "2.4.0" assert cli.TESTED_VERSION == "2.4.0"
# --- #18: startup order and tags (spec: "Autostart, startup order and tags") ----------
@pytest.mark.parametrize(
("args", "expected"),
[
((2, 10, 30), "order=2,up=10,down=30"),
((2, 10, None), "order=2,up=10"),
((2, None, None), "order=2"),
((None, 10, None), "up=10"),
((None, None, 30), "down=30"),
((None, 10, 30), "up=10,down=30"),
((0, 0, 0), "order=0,up=0,down=0"), # 0 is a value, not "unset"
((None, None, None), None),
],
)
def test_startup_value_builds_only_the_given_parts(args, expected):
assert cli.startup_value(*args) == expected
@pytest.mark.parametrize(
"args", [(-1, None, None), (None, -5, None), (None, None, -30), (1, 2, -1)]
)
def test_startup_value_rejects_negatives(args):
with pytest.raises(ValueError):
cli.startup_value(*args)
@pytest.mark.parametrize("bad", ["x", 1.5, [1]])
def test_startup_value_rejects_non_ints(bad):
for args in ((bad, None, None), (None, bad, None), (None, None, bad)):
with pytest.raises(ValueError):
cli.startup_value(*args)
@pytest.mark.parametrize(
("text", "expected"),
[
("order=2,up=10,down=30", (2, 10, 30)),
("3", (3, None, None)), # a bare leading number is the order
("up=5", (None, 5, None)),
("order=2,up=10", (2, 10, None)),
("3,up=5", (3, 5, None)),
("down=7", (None, None, 7)),
(None, (None, None, None)),
("", (None, None, None)),
],
)
def test_parse_startup(text, expected):
assert cli.parse_startup(text) == expected
@given(
st.tuples(*[st.one_of(st.none(), st.integers(min_value=0, max_value=10**6))] * 3)
)
def test_parse_startup_inverts_startup_value_property(parts):
assert cli.parse_startup(cli.startup_value(*parts)) == parts
@pytest.mark.parametrize(
("text", "expected"),
[
("os", "os"),
("a;b;c", "a;b;c"),
("a,b", "a;b"),
("a b", "a;b"), # whitespace separates: two tags, not one with a space
("a;b, c d", "a;b;c;d"),
("a; b; c", "a;b;c"),
("a;b;a;c;b", "a;b;c"), # duplicates dropped, first occurrence wins
("c;a;b", "c;a;b"), # order kept, not sorted
(";;a;;,, b;", "a;b"), # empties dropped
("172.16.41.200", "172.16.41.200"), # IP-like tag is valid
("172.16.41.200;community-script;os", "172.16.41.200;community-script;os"),
("_x;a+b;a.b;a-b;9", "_x;a+b;a.b;a-b;9"),
],
)
def test_tags_value_normalises(text, expected):
assert cli.tags_value(text) == expected
@pytest.mark.parametrize("text", ["", " ", ";", " ; , ;", "\n\t"])
def test_tags_value_empty_is_none(text):
assert cli.tags_value(text) is None
@pytest.mark.parametrize(
"bad", ["-x", "a;b!", "!", "a;-b", ".a", "+a", "a/b", "a;b@c", "ä"]
)
def test_tags_value_rejects_invalid_tags(bad):
with pytest.raises(ValueError):
cli.tags_value(bad)
_TAG = st.from_regex(r"[A-Za-z0-9_][A-Za-z0-9_+.-]*", fullmatch=True)
_SEP = st.sampled_from([";", ",", " ", "; ", ", ", " ", ";;"])
@st.composite
def _valid_tag_text(draw):
tags = draw(st.lists(_TAG, max_size=6))
out = ""
for tag in tags:
out += tag + draw(_SEP)
return out
@given(_valid_tag_text())
def test_tags_value_is_idempotent_property(text):
once = cli.tags_value(text)
if once is None:
assert text.replace(";", "").replace(",", "").strip() == ""
else:
assert cli.tags_value(once) == once
assert len(set(once.split(";"))) == len(once.split(";")) # no duplicates