From f5f3effce5f503e9b50df544ff92adae665d5336 Mon Sep 17 00:00:00 2001 From: Per-Arne Hellarvik Date: Tue, 3 Nov 2026 00:55:02 +0100 Subject: [PATCH] Edit autostart, startup order and tags from the info screen The info screen's Resources table shows Autostart (yes/no), Startup (order, up and down delays) and Tags. o opens a dialog to change them. Only keys that differ are sent, in one `set config` call through the existing set_net. A confirm lists the old and new value of each key. Tags are removed with `--delete tags` on set config. The CLI also has a `delete ct|vm ... tags` command, but it shares the form that destroys the whole guest, so the app never uses it; nothing in the code builds a delete ct|vm argv. Inputs are validated before sending: startup values must be ints >= 0, and tags must match [A-Za-z0-9_][A-Za-z0-9_+.-]*. They are split on ; , or whitespace and deduplicated. Live test on CT 104: onboot 1 -> 0, startup order=5,up=10, tag tui-test added, then all set back through the dialog. The config ended identical to the baseline. Closes #18 Co-Authored-By: Claude Opus 5.5 --- README.md | 1 + docs/plans/guest-tui-mvp.md | 40 +++++ src/cv4pve_tui/app.py | 118 +++++++++++++ src/cv4pve_tui/cli.py | 39 +++++ tests/test_app.py | 320 ++++++++++++++++++++++++++++++++++++ tests/test_cli.py | 116 +++++++++++++ 6 files changed, 634 insertions(+) diff --git a/README.md b/README.md index 3bf9668..42bde21 100644 --- a/README.md +++ b/README.md @@ -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 `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 `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) | | `t` | shell in a running LXC container (`ssh -t root@ pct enter `) | | `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 | diff --git a/docs/plans/guest-tui-mvp.md b/docs/plans/guest-tui-mvp.md index ac7ab31..a06bd5a 100644 --- a/docs/plans/guest-tui-mvp.md +++ b/docs/plans/guest-tui-mvp.md @@ -657,3 +657,43 @@ CLI facts (read-only probes): - 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. + +--- + +# 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 config --onboot 0|1 --startup --tags + [--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 + ` 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: ()"): + `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. diff --git a/src/cv4pve_tui/app.py b/src/cv4pve_tui/app.py index e530112..ca0cbf3 100644 --- a/src/cv4pve_tui/app.py +++ b/src/cv4pve_tui/app.py @@ -1,5 +1,6 @@ """Textual app for managing Proxmox guests through cv4pve-cli.""" +import re import subprocess import sys 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): """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"), ("a", "toggle_all", "All backups"), ("n", "net", "Edit network"), + ("o", "boot", "Autostart, startup, tags"), ("u", "undo", "Undo network"), ("e", "edit", "Edit resources"), ("p", "snapshot", "Take snapshot"), @@ -851,6 +942,10 @@ class InfoScreen(ModalScreen): ) 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: if cli.nets(self.config) and not self._refuse_if_busy(): 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: rows.append(("Swap", f"{config['swap']} MiB")) 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 +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: return part / whole * 100 if part is not None and whole else None diff --git a/src/cv4pve_tui/cli.py b/src/cv4pve_tui/cli.py index 3e7d3e1..36987db 100644 --- a/src/cv4pve_tui/cli.py +++ b/src/cv4pve_tui/cli.py @@ -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]: """VMIDs no backup job covers (Proxmox's own answer, incl. pool-based jobs).""" items = json.loads(run("get", "cluster", "not-backed-up", "-o", "json")) diff --git a/tests/test_app.py b/tests/test_app.py index b98d1bb..82caad3 100644 --- a/tests/test_app.py +++ b/tests/test_app.py @@ -3875,3 +3875,323 @@ def test_guest_matches_tolerates_none_status_and_type(): assert appmod.guest_matches(g, "x") assert not appmod.guest_matches(g, "is:running") 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 diff --git a/tests/test_cli.py b/tests/test_cli.py index bb10211..2824d08 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -1388,3 +1388,119 @@ def test_version_argv_and_strips(fake): def test_tested_version_constant(): 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