From 62b79ab29efb3dab5287ab3d7fc5bf532bee9a49 Mon Sep 17 00:00:00 2001 From: Per-Arne Hellarvik Date: Mon, 2 Nov 2026 20:22:53 +0100 Subject: [PATCH] Add context listing and switching to the TUI Press c to list cv4pve-cli contexts and switch between them. The active context is always shown in the header. - Switching runs `cv4pve-cli config use`, which changes the global current context. cv4pve-cli 2.4.0 has no per-call context option. - `config list` has no JSON output, so its text lines are parsed. - The guest table is cleared as soon as another context is chosen, so the old cluster's guests cannot be acted on. - A generation counter drops guest listings that started before a switch, and g and c are refused while a switch runs. Without this, a reload already in flight could refill the table with the old cluster's guests, and VMIDs collide across clusters. 103 tests pass. The switch was tested live from mainframe to mainframe-edge and back. Co-Authored-By: Claude Opus 5.5 --- README.md | 4 + docs/plans/guest-tui-mvp.md | 63 ++++++++ src/cv4pve_tui/app.py | 130 +++++++++++++++- src/cv4pve_tui/cli.py | 18 +++ tests/test_app.py | 288 +++++++++++++++++++++++++++++++++++- tests/test_cli.py | 49 ++++++ 6 files changed, 543 insertions(+), 9 deletions(-) diff --git a/README.md b/README.md index 1a47505..7d83837 100644 --- a/README.md +++ b/README.md @@ -21,8 +21,12 @@ uv run cv4pve-tui | `r` | reboot (confirm) | | `e` | edit cores / memory / grow a disk (disk growth asks again; it cannot be undone) | | `g` | refresh | +| `c` | switch cv4pve-cli context (cluster) | | `q` | quit | +Switching context runs `cv4pve-cli config use`, which changes cv4pve-cli's global current +context: other shells and scripts follow, and it stays after the app exits. + Empty fields in the edit dialog are left unchanged. If a CPU/memory change cannot be hot-plugged, the app tells you it applies at the next reboot. diff --git a/docs/plans/guest-tui-mvp.md b/docs/plans/guest-tui-mvp.md index 2411be2..3f73265 100644 --- a/docs/plans/guest-tui-mvp.md +++ b/docs/plans/guest-tui-mvp.md @@ -116,3 +116,66 @@ sockets/vCPU/hotplug settings, live auto-refresh. ## Open questions None blocking. + +--- + +# Context switching + +## Problem + +Several cv4pve-cli contexts (clusters) exist; switch between them inside the TUI +instead of quitting and running `cv4pve-cli config use`. + +## Decision + +Option A: switching runs `cv4pve-cli config use `, which changes the **global** +current context (all shells and scripts follow) and stays after the app exits. cv4pve-cli +2.4.0 has no per-call context flag or env var (verified), so there is no app-local option. + +Observed CLI behaviour: + +- `config list` has no `-o json`. Text output, one line per context; current one + prefixed with `* `, others with two spaces: + `* mainframe mainframe.iot.home.arpa:8006 (api-token)` + ` mainframe-edge mainframe-edge.home.arpa:8006 (api-token)` +- `config use ` -> exit 0 on success; unknown name -> exit 3, + `Error: Context '' not found.`, current context unchanged. + +### `cli.py` + +- `contexts() -> list[dict]` - `run("config", "list")`, each line parsed to + `{"name", "host", "auth", "current": bool}` (regex + `^(\*)?\s*(\S+)\s+(\S+)\s+\((.*)\)\s*$`); non-matching lines are ignored. +- `use_context(name)` - `run("config", "use", name)`. + +### `app.py` (UI contract) + +- The active context is always visible: `app.sub_title == f"context: {name}"` + (shown in the Header). Set on mount from `cli.contexts()` and after every switch. + If `contexts()` fails: error notify, sub_title `context: ?`. +- Key `c` opens `ContextScreen(ModalScreen[str | None])` with `OptionList#contexts`, + one option per context, option id = context name, current one marked with `*`. + Selecting an option dismisses with its name; `escape` dismisses with `None`. +- `c` while any guest is busy -> warning notify, no screen (VMIDs collide across + clusters, so the busy guard would point at the wrong cluster). +- Selecting the current context or dismissing -> no CLI call. +- A reload started before a switch must not refill the table (generation counter); + `g` and `c` are refused while a switch runs; the post-action reload finishes before + the guest stops being busy. +- Selecting another context: table cleared immediately (old cluster's guests must not + stay actionable), then in a worker `cli.use_context(name)`, sub_title updated, guests + reloaded. On `CliError`: error notify with summary, sub_title re-read from + `cli.contexts()`, guests reloaded. + +## Out of scope + +Adding/editing/deleting contexts, `config verify`, restoring the original context on exit. + +## Steps + +- [x] Tests for `contexts`/`use_context` and the UI contract above - red first. +- [x] Implementation - full suite green, ruff clean. +- [x] Live: open the list, switch to `mainframe-edge` and back to `mainframe` + (only `config use` and read-only listing; no guest actions); final context + is `mainframe`. +- [ ] Push `feat/guest-tui-mvp`, merge to `main` (user-approved). diff --git a/src/cv4pve_tui/app.py b/src/cv4pve_tui/app.py index e2c3df7..7d6bb6b 100644 --- a/src/cv4pve_tui/app.py +++ b/src/cv4pve_tui/app.py @@ -4,7 +4,17 @@ from textual import work from textual.app import App, ComposeResult from textual.containers import Horizontal, Vertical from textual.screen import ModalScreen -from textual.widgets import Button, DataTable, Footer, Header, Input, Select, Static +from textual.widgets import ( + Button, + DataTable, + Footer, + Header, + Input, + OptionList, + Select, + Static, +) +from textual.widgets.option_list import Option from cv4pve_tui import cli @@ -89,6 +99,32 @@ class EditScreen(ModalScreen): return int(text) +class ContextScreen(ModalScreen[str | None]): + """Dismisses with the chosen context name, or None when cancelled.""" + + BINDINGS = [("escape", "dismiss(None)", "Cancel")] # noqa: RUF012 - Textual idiom + + def __init__(self, contexts: list[dict]) -> None: + super().__init__() + self.contexts = contexts + + def compose(self) -> ComposeResult: + with Vertical(): + yield OptionList( + *( + Option( + f"{'*' if c['current'] else ' '} {c['name']} {c['host']}", + id=c["name"], + ) + for c in self.contexts + ), + id="contexts", + ) + + def on_option_list_option_selected(self, event: OptionList.OptionSelected) -> None: + self.dismiss(event.option.id) + + class GuestApp(App): BINDINGS = [ # noqa: RUF012 - Textual idiom ("s", "act('start')", "Start"), @@ -97,6 +133,7 @@ class GuestApp(App): ("r", "act('reboot')", "Reboot"), ("e", "edit", "Edit"), ("g", "reload", "Refresh"), + ("c", "context", "Context"), ("q", "quit", "Quit"), ] @@ -104,6 +141,9 @@ class GuestApp(App): super().__init__() self.guests: dict[str, dict] = {} self.busy: set[int] = set() + # bumped on every context switch; replies started earlier come from the old cluster + self.generation = 0 + self.switching = False def compose(self) -> ComposeResult: yield Header() @@ -122,8 +162,29 @@ class GuestApp(App): "Memory (MiB)", "Disk (GiB)", ) + self._show_context() self.action_reload() + @work(thread=True) + def _show_context(self) -> None: + self._read_context() + + def _read_context(self) -> None: + """Runs in a worker thread: show the CLI's current context in the Header.""" + gen = self.generation + try: + name = next(c["name"] for c in cli.contexts() if c["current"]) + except StopIteration: + name = "?" + except Exception as exc: # noqa: BLE001 - any failure must reach the user, not kill the app + self.call_from_thread(self.notify, _message(exc), severity="error") + name = "?" + self.call_from_thread(self._set_context_title, name, gen) + + def _set_context_title(self, name: str, gen: int) -> None: + if gen == self.generation: + self.sub_title = f"context: {name}" + def _selected(self) -> dict | None: table = self.query_one("#guests", DataTable) if not table.row_count: @@ -132,7 +193,9 @@ class GuestApp(App): table.coordinate_to_cell_key(table.cursor_coordinate).row_key.value ] - def _fill(self, guests: list[dict]) -> None: + def _fill(self, guests: list[dict], gen: int) -> None: + if gen != self.generation: # listed before a context switch: old cluster + return table = self.query_one("#guests", DataTable) current = self._selected() table.clear() @@ -148,6 +211,7 @@ class GuestApp(App): def _reload(self, acted: dict | None = None) -> None: """Runs in a worker thread. `acted` gets its live status: the listing lags.""" + gen = self.generation try: guests = cli.list_guests() for g in guests: @@ -156,12 +220,66 @@ class GuestApp(App): except Exception as exc: # noqa: BLE001 - any failure must reach the user, not kill the app self.call_from_thread(self.notify, _message(exc), severity="error") return - self.call_from_thread(self._fill, guests) + self.call_from_thread(self._fill, guests, gen) + + def action_reload(self) -> None: + if self.switching: + self.notify("Switching context; refresh after it ends", severity="warning") + return + self._reload_worker() @work(thread=True) - def action_reload(self) -> None: + def _reload_worker(self) -> None: self._reload() + def action_context(self) -> None: + if self.busy or self.switching: + self.notify( + "Busy (guest action or context switch); try again when it ends", + severity="warning", + ) + return + self._open_context() + + @work(thread=True) + def _open_context(self) -> None: + try: + contexts = cli.contexts() + except Exception as exc: # noqa: BLE001 - surface every failure, never kill the app + self.call_from_thread(self.notify, _message(exc), severity="error") + return + current = next((c["name"] for c in contexts if c["current"]), None) + self.call_from_thread( + self.push_screen, + ContextScreen(contexts), + lambda name: self._context_chosen(name, current), + ) + + def _context_chosen(self, name: str | None, current: str | None) -> None: + if name is None or name == current: + return + self.query_one( + "#guests", DataTable + ).clear() # old cluster's guests: not actionable + self.guests = {} + self.generation += 1 + self.switching = True + self._switch_context(name) + + @work(thread=True) + def _switch_context(self, name: str) -> None: + try: + try: + cli.use_context(name) + except Exception as exc: # noqa: BLE001 - surface every failure, never kill the app + self.call_from_thread(self.notify, _message(exc), severity="error") + self._read_context() + else: + self.call_from_thread(setattr, self, "sub_title", f"context: {name}") + self._reload() + finally: + self.call_from_thread(setattr, self, "switching", False) + def _start_job(self, guest: dict, label: str, job) -> None: if guest["vmid"] in self.busy: self.notify( @@ -185,8 +303,10 @@ class GuestApp(App): self.notify, f"{label} done: {guest['name']} ({guest['vmid']})" ) finally: + self._reload( + acted=guest + ) # before un-busying, so no context switch can start mid-reload self.busy.discard(guest["vmid"]) - self._reload(acted=guest) def action_act(self, name: str) -> None: guest = self._selected() diff --git a/src/cv4pve_tui/cli.py b/src/cv4pve_tui/cli.py index d665612..a90aecf 100644 --- a/src/cv4pve_tui/cli.py +++ b/src/cv4pve_tui/cli.py @@ -138,3 +138,21 @@ def status(guest: dict) -> str: return { i["key"]: i["value"] for i in json.loads(run("api", "get", path, "-o", "json")) }["status"] + + +_CONTEXT_LINE = re.compile(r"^(\*)?\s*(\S+)\s+(\S+)\s+\((.*)\)\s*$") + + +def contexts() -> list[dict]: + """Parse `config list` (text only); lines that do not look like a context are skipped.""" + found = (_CONTEXT_LINE.match(line) for line in run("config", "list").splitlines()) + return [ + {"name": m[2], "host": m[3], "auth": m[4], "current": bool(m[1])} + for m in found + if m + ] + + +def use_context(name: str) -> None: + """Changes cv4pve-cli's GLOBAL current context.""" + run("config", "use", name) diff --git a/tests/test_app.py b/tests/test_app.py index 3ddd6a1..dde40c9 100644 --- a/tests/test_app.py +++ b/tests/test_app.py @@ -8,10 +8,10 @@ import threading import pytest from textual.screen import ModalScreen -from textual.widgets import Button, DataTable, Input, Select, Static +from textual.widgets import Button, DataTable, Input, OptionList, Select, Static from cv4pve_tui import cli -from cv4pve_tui.app import ConfirmScreen, EditScreen, GuestApp +from cv4pve_tui.app import ConfirmScreen, ContextScreen, EditScreen, GuestApp VM = {"vmid": 108, "name": "docker", "node": "mainframe", "type": "qemu", "status": "running", "maxcpu": 4, "maxmem": 8192 * 2**20, "maxdisk": 10 * 2**30} # fmt: skip @@ -31,10 +31,24 @@ class Fake: self.action_error = None self.action_hook = None self.live_status, self.status_calls = {}, [] + self.current = "mainframe" + self.names = ["mainframe", "mainframe-edge"] + self.contexts_error = None + self.contexts_calls = 0 + self.use_calls = [] + self.use_error = None + self.use_hook = None + self.guests_by_context = {} + self.list_hook = None def list_guests(self): self.list_calls += 1 - return list(self.guests) + snapshot = list(self.guests_by_context.get(self.current, self.guests)) + if ( + self.list_hook + ): # runs after reading: simulates a slow reply from that cluster + self.list_hook() + return snapshot def guest_config(self, guest): return {"cores": guest["maxcpu"]} @@ -64,12 +78,30 @@ class Fake: self.status_calls.append(guest["vmid"]) return self.live_status.get(guest["vmid"], guest["status"]) + def contexts(self): + self.contexts_calls += 1 + if self.contexts_error: + raise self.contexts_error + return [ + {"name": n, "host": f"{n}.home.arpa:8006", "auth": "api-token", + "current": n == self.current} + for n in self.names + ] # fmt: skip + + def use_context(self, name): + self.use_calls.append(name) + if self.use_hook: + self.use_hook() + if self.use_error: + raise self.use_error + self.current = name + @pytest.fixture def fake(monkeypatch): f = Fake() for name in ("list_guests", "guest_config", "disks", "action", "set_resources", - "resize_disk", "pending_keys", "status"): # fmt: skip + "resize_disk", "pending_keys", "status", "contexts", "use_context"): # fmt: skip monkeypatch.setattr(cli, name, getattr(f, name)) return f @@ -447,3 +479,251 @@ async def test_table_shows_live_status_after_action_not_lagging_list(fake): assert "stopped" in [ str(c) for c in app.query_one("#guests", DataTable).get_row("108") ] + + +# --- context switching ------------------------------------------------------ +# spec: docs/plans/guest-tui-mvp.md ("Context switching" > "app.py (UI contract)") + + +async def open_contexts(app, pilot): + await settle(app, pilot) + await pilot.press("c") + await settle(app, pilot) + assert isinstance(app.screen, ContextScreen) + return app.screen.query_one("#contexts", OptionList) + + +async def choose(app, pilot, index): + ol = app.screen.query_one("#contexts", OptionList) + ol.focus() + ol.highlighted = index + await pilot.pause() + await pilot.press("enter") + await settle(app, pilot) + + +async def test_sub_title_shows_active_context_after_mount(fake): + app = make_app() + async with app.run_test() as pilot: + await settle(app, pilot) + assert app.sub_title == "context: mainframe" + + +async def test_contexts_failure_notifies_and_sub_title_is_unknown(fake): + fake.contexts_error = cli.CliError("Error: no config") + app = make_app() + async with app.run_test() as pilot: + await settle(app, pilot) + assert errors(app) + assert app.sub_title == "context: ?" + + +async def test_c_opens_context_screen_listing_contexts_with_current_marked(fake): + app = make_app() + async with app.run_test() as pilot: + ol = await open_contexts(app, pilot) + assert ol.option_count == 2 + options = [ol.get_option_at_index(i) for i in range(2)] + assert [o.id for o in options] == ["mainframe", "mainframe-edge"] + assert "*" in str(options[0].prompt) + assert "*" not in str(options[1].prompt) + assert "mainframe-edge" in str(options[1].prompt) + + +async def test_selecting_other_context_switches_and_reloads(fake): + app = make_app() + async with app.run_test() as pilot: + await open_contexts(app, pilot) + before = fake.list_calls + await choose(app, pilot, 1) + assert fake.use_calls == ["mainframe-edge"] + assert app.sub_title == "context: mainframe-edge" + assert fake.list_calls > before + assert not isinstance(app.screen, ModalScreen) + assert app.query_one("#guests", DataTable).row_count == 2 + + +async def test_table_is_cleared_while_switch_is_in_progress(fake): + started, release = threading.Event(), threading.Event() + fake.use_hook = lambda: (started.set(), release.wait(10)) + app = make_app() + async with app.run_test() as pilot: + await open_contexts(app, pilot) + assert app.query_one("#guests", DataTable).row_count == 2 + try: + ol = app.screen.query_one("#contexts", OptionList) + ol.focus() + ol.highlighted = 1 + await pilot.pause() + await pilot.press("enter") + for _ in range(200): + if started.is_set(): + break + await pilot.pause(0.01) + assert started.is_set() + await pilot.pause() + assert app.query_one("#guests", DataTable).row_count == 0 + finally: + release.set() + await settle(app, pilot) + assert app.query_one("#guests", DataTable).row_count == 2 + + +async def test_selecting_current_context_makes_no_cli_call(fake): + app = make_app() + async with app.run_test() as pilot: + await open_contexts(app, pilot) + before = fake.list_calls + await choose(app, pilot, 0) + assert fake.use_calls == [] + assert not isinstance(app.screen, ModalScreen) + assert app.sub_title == "context: mainframe" + assert fake.list_calls == before + + +async def test_escape_closes_context_screen_without_cli_call(fake): + app = make_app() + async with app.run_test() as pilot: + await open_contexts(app, pilot) + await pilot.press("escape") + await settle(app, pilot) + assert not isinstance(app.screen, ModalScreen) + assert fake.use_calls == [] + assert app.sub_title == "context: mainframe" + + +async def test_c_while_guest_busy_warns_and_opens_nothing(fake): + started, release = threading.Event(), threading.Event() + fake.action_hook = lambda: (started.set(), release.wait(10)) + app = make_app() + async with app.run_test() as pilot: + await settle(app, pilot) + try: + await pilot.press("s") + for _ in range(200): + if started.is_set(): + break + await pilot.pause(0.01) + assert started.is_set() + await pilot.press("c") + await pilot.pause() + assert not isinstance(app.screen, ContextScreen) + assert any(sev == "warning" for _, sev in app.seen) + finally: + release.set() + await settle(app, pilot) + assert fake.use_calls == [] + + +async def test_use_context_failure_notifies_rereads_context_and_reloads(fake): + err = cli.CliError("Error: Context 'mainframe-edge' not found.") + + def hook(): # the CLI state differs from what the app assumed + fake.current = "mainframe-edge" + + fake.use_hook, fake.use_error = hook, err + app = make_app() + async with app.run_test() as pilot: + await open_contexts(app, pilot) + contexts_before, list_before = fake.contexts_calls, fake.list_calls + await choose(app, pilot, 1) + assert any(err.summary in m for m in errors(app)) + assert fake.contexts_calls > contexts_before + assert app.sub_title == "context: mainframe-edge" # whatever contexts() says + assert fake.list_calls > list_before + assert app.query_one("#guests", DataTable).row_count == 2 + + +# --- review finding: no old-cluster rows after a switch ---------------------- + +EDGE = {"vmid": 300, "name": "edge-vm", "node": "edge", "type": "qemu", "status": "running", + "maxcpu": 2, "maxmem": 2**30, "maxdisk": 2**31} # fmt: skip + + +async def until(pilot, cond): + for _ in range(300): + if cond(): + return + await pilot.pause(0.01) + raise AssertionError("condition not reached") + + +def blocking_once(): + started, release = threading.Event(), threading.Event() + + def hook(): + if not started.is_set(): + started.set() + release.wait(10) + + return hook, started, release + + +def rows(app): + return [k.value for k in app.query_one("#guests", DataTable).rows] + + +async def switch_to_edge(app, pilot): + await pilot.press("c") + await until(pilot, lambda: isinstance(app.screen, ContextScreen)) + ol = app.screen.query_one("#contexts", OptionList) + ol.focus() + ol.highlighted = 1 + await pilot.pause() + await pilot.press("enter") + + +async def test_reload_in_flight_before_switch_cannot_restore_old_rows(fake): + fake.guests_by_context = {"mainframe-edge": [EDGE]} + app = make_app() + async with app.run_test() as pilot: + await settle(app, pilot) + hook, started, release = blocking_once() + fake.list_hook = hook + try: + await pilot.press("g") # reads mainframe, then stalls + await until(pilot, started.is_set) + await switch_to_edge(app, pilot) + await until(pilot, lambda: rows(app) == ["300"]) + finally: + release.set() + await settle(app, pilot) + assert rows(app) == ["300"] + assert app.sub_title == "context: mainframe-edge" + + +async def test_refresh_refused_while_switching(fake): + started, release = threading.Event(), threading.Event() + fake.use_hook = lambda: (started.set(), release.wait(10)) + app = make_app() + async with app.run_test() as pilot: + await settle(app, pilot) + try: + await switch_to_edge(app, pilot) + await until(pilot, started.is_set) + before = fake.list_calls + await pilot.press("g") + await pilot.pause() + assert fake.list_calls == before + assert any(sev == "warning" for _, sev in app.seen) + finally: + release.set() + await settle(app, pilot) + + +async def test_context_refused_while_post_action_reload_runs(fake): + app = make_app() + async with app.run_test() as pilot: + await settle(app, pilot) + hook, started, release = blocking_once() + fake.list_hook = hook + try: + await pilot.press("s") # start is instant; the reload after it stalls + await until(pilot, started.is_set) + await pilot.press("c") + await pilot.pause(0.1) + assert not isinstance(app.screen, ContextScreen) + assert any(sev == "warning" for _, sev in app.seen) + finally: + release.set() + await settle(app, pilot) diff --git a/tests/test_cli.py b/tests/test_cli.py index db4e3c4..98b1ac8 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -370,3 +370,52 @@ def test_status_reads_live_status_current(fake): "-o", "json", ] + + +# --- contexts / use_context (spec: "Context switching" > cli.py) ------------- + +CONTEXT_LIST = ( + "* mainframe mainframe.iot.home.arpa:8006 (api-token)\n" + " mainframe-edge mainframe-edge.home.arpa:8006 (api-token)\n" +) + + +def test_contexts_parses_current_name_host_auth(fake): + fake.result = (CONTEXT_LIST, "", 0) + assert cli.contexts() == [ + {"name": "mainframe", "host": "mainframe.iot.home.arpa:8006", + "auth": "api-token", "current": True}, + {"name": "mainframe-edge", "host": "mainframe-edge.home.arpa:8006", + "auth": "api-token", "current": False}, + ] # fmt: skip + + +def test_contexts_runs_config_list(fake): + fake.result = (CONTEXT_LIST, "", 0) + cli.contexts() + assert fake.argv == ["cv4pve-cli", "config", "list"] + + +def test_contexts_ignores_junk_and_empty_lines(fake): + fake.result = ( + "\nContexts:\n" + CONTEXT_LIST + "\n \nnot a context line\n", + "", + 0, + ) + assert [c["name"] for c in cli.contexts()] == ["mainframe", "mainframe-edge"] + + +def test_contexts_empty_output_gives_empty_list(fake): + assert cli.contexts() == [] + + +def test_use_context_runs_config_use_with_name(fake): + cli.use_context("mainframe-edge") + assert fake.argv == ["cv4pve-cli", "config", "use", "mainframe-edge"] + + +def test_use_context_unknown_name_raises_clierror(monkeypatch): + FakeRun(monkeypatch, "Error: Context 'x' not found.\n", returncode=3) + with pytest.raises(cli.CliError) as exc: + cli.use_context("x") + assert "Context 'x' not found" in str(exc.value)