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
Showing only changes of commit f5f3effce5 - Show all commits

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 `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@<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 |
+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
`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."""
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
+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]:
"""VMIDs no backup job covers (Proxmox's own answer, incl. pool-based jobs)."""
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 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
+116
View File
@@ -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