From d97ad22c78f5fe70f62fb608e9733e0a969fc4c0 Mon Sep 17 00:00:00 2001 From: Per-Arne Hellarvik Date: Tue, 3 Nov 2026 00:42:13 +0100 Subject: [PATCH] Add search and filter to the guest table MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit / opens a filter box under the guest table, and the table filters as you type. Terms are separated by spaces, all must match, and case is ignored: - a plain term matches part of the name, the VMID or a tag - is:running and is:stopped match the status - type:lxc/ct and type:vm/qemu match the guest type Enter keeps the filter and goes back to the table. Escape clears it, from the box or from the table. The frame shows "· filter: · N of M". Actions only ever see the visible rows, and the filter survives reloads and context switches. Review fixes: - Typing "[/b]" or "[/]" crashed the app, because the filter text went into the border subtitle, which Textual parses as markup. The subtitle is now plain text, so "[red]" is also shown literally. - Escape on the table now clears a filter that was applied with Enter. - guest_matches tolerates a None status or type. Read-only live check on mainframe: "debian" gives 104, "type:vm" gives 108, "community-script is:running" gives 14 of 19, and "is:stopped" gives 0 of 19. Closes #16 Co-Authored-By: Claude Opus 5.5 --- README.md | 1 + docs/plans/guest-tui-mvp.md | 28 +++ src/cv4pve_tui/app.py | 95 +++++++++- tests/test_app.py | 364 ++++++++++++++++++++++++++++++++++++ 4 files changed, 481 insertions(+), 7 deletions(-) diff --git a/README.md b/README.md index e400928..3bf9668 100644 --- a/README.md +++ b/README.md @@ -49,6 +49,7 @@ uv run cv4pve-tui | `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 | | `U` | package updates: pending updates per online node (node, package, current, new, origin, priority), read-only; `g` reloads. The list is as of the node's last `apt update`. The frame's bottom border shows the count | | `g` | refresh | +| `/` | filter the guest table as you type: space-separated terms, all must match (case-insensitive); plain term = part of name, VMID or a tag; `is:running` / `is:stopped`; `type:lxc` (or `ct`) / `type:vm` (or `qemu`). `Enter` keeps the filter and returns to the table, `Escape` clears it. The frame's bottom border shows `filter: · N of M` | | `c` | switch cv4pve-cli context (cluster) | | `q` | quit | diff --git a/docs/plans/guest-tui-mvp.md b/docs/plans/guest-tui-mvp.md index 93fa717..ac7ab31 100644 --- a/docs/plans/guest-tui-mvp.md +++ b/docs/plans/guest-tui-mvp.md @@ -629,3 +629,31 @@ CLI facts (read-only probes): is dropped after a switch) and is cleared when a switch starts. Failures leave the count unknown (nothing shown), no error notify on mount. - Main footer shows `U Updates`. + +--- + +# Search and filter in the guest table (issue #16) + +## Decision + +- `/` on the main screen shows `Input#filter` (hidden until then) below the guest table + and focuses it. The table filters as you type. +- Filter text = space-separated terms, all must match (AND), case-insensitive: + - a plain term matches a substring of the name, the VMID or one of the tags + (`tags` from `get guests`, `;`-separated); + - `is:running` / `is:stopped` match the status; + - `type:lxc` / `type:ct` and `type:vm` / `type:qemu` match the guest type. +- Enter in the filter keeps the filter and moves focus to the table. Escape in the + filter clears it, hides the input and moves focus to the table. +- The frame's bottom border gets "· filter: · N of M" while a filter is active, + after the context name and before the backup/package counts. +- Actions (s/S/x/r/t/Enter) act on the selected *visible* row only; the table only ever + contains the matching guests. A reload (g, after actions, context switch) keeps the + filter applied. A context switch keeps the filter text. +- Placeholder of the input shows the syntax: `name, vmid, tag · is:running · type:lxc`. +- No match: empty table, and the frame says "0 of M"; action keys do nothing. + +### Code + +- 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. diff --git a/src/cv4pve_tui/app.py b/src/cv4pve_tui/app.py index 62371af..e530112 100644 --- a/src/cv4pve_tui/app.py +++ b/src/cv4pve_tui/app.py @@ -23,6 +23,7 @@ from textual.widgets import ( Select, Static, ) +from textual.widgets._data_table import RowDoesNotExist from textual.widgets.option_list import Option from cv4pve_tui import cli @@ -991,12 +992,45 @@ def _job_text(job: dict) -> str: ) +_TYPES = {"lxc": "lxc", "ct": "lxc", "vm": "qemu", "qemu": "qemu"} + + +def guest_matches(guest: dict, query: str) -> bool: + """True if the guest matches every space-separated term (case-insensitive). + + Plain term: substring of the name, the VMID or one tag. `is:running|stopped` + matches the status, `type:lxc|ct|vm|qemu` the type; an unknown value matches nothing. + """ + tags = (guest.get("tags") or "").lower().split(";") + for term in query.lower().split(): + key, _, value = term.partition(":") + if key == "is" and value: + ok = (guest.get("status") or "").lower() == value + elif key == "type" and value: + ok = _TYPES.get(value) == (guest.get("type") or "").lower() + else: + ok = ( + term in str(guest.get("name", "")).lower() + or term in str(guest.get("vmid", "")) + or any(term in t for t in tags) + ) + if not ok: + return False + return True + + class GuestTable(DataTable): """The guest table; shows its Enter (row selected -> info screen) in the footer.""" BINDINGS = [Binding("enter", "select_cursor", "Info")] # noqa: RUF012 - Textual idiom +class FilterInput(Input): + """The filter box; Escape clears and hides it.""" + + BINDINGS = [Binding("escape", "app.clear_filter", "Clear filter")] # noqa: RUF012 - Textual idiom + + class GuestApp(App): BINDINGS = [ # noqa: RUF012 - Textual idiom ("s", "act('start')", "Start"), @@ -1006,9 +1040,12 @@ class GuestApp(App): ("t", "shell", "Shell"), ("g", "reload", "Refresh"), ("c", "context", "Context"), + ("slash", "filter", "Filter"), ("l", "tasks", "Tasks"), ("U", "updates", "Updates"), ("q", "quit", "Quit"), + # also on the table, once a filter was applied with Enter (focus left the input) + Binding("escape", "clear_filter", "Clear filter", show=False), ] def __init__(self) -> None: @@ -1028,6 +1065,7 @@ class GuestApp(App): border: round $accent; border-title-style: bold; padding: 0 1; } #guests_section DataTable { height: 1fr; } + #filter { display: none; } """ def compose(self) -> ComposeResult: @@ -1035,6 +1073,9 @@ class GuestApp(App): with Vertical(id="guests_section") as section: section.border_title = "Guests" yield GuestTable(id="guests", cursor_type="row") + yield FilterInput( + id="filter", placeholder="name, vmid, tag · is:running · type:lxc" + ) yield Footer() def on_mount(self) -> None: @@ -1111,10 +1152,17 @@ class GuestApp(App): def _update_frame(self) -> None: n = len(self.unprotected & {g["vmid"] for g in self.guests.values()}) count = f" · {n} without backup job" if n else "" + query = self.query_one("#filter", Input).value.strip() + if query: + shown = self.query_one("#guests", DataTable).row_count + count = f" · filter: {query} · {shown} of {len(self.guests)}" + count if self.package_count: n = self.package_count count += f" · {n} package update{'s' if n != 1 else ''}" - self.query_one("#guests_section").border_subtitle = self.context_name + count + # plain text: the filter is typed by the user and must never be parsed as markup + self.query_one("#guests_section").border_subtitle = Text( + self.context_name + count + ) @work(thread=True) def _show_context(self) -> None: @@ -1150,11 +1198,22 @@ class GuestApp(App): if gen != self.generation: # listed before a context switch: old cluster return self.unprotected = unprotected or set() - table = self.query_one("#guests", DataTable) - current = self._selected() - table.clear() self.guests = {str(g["vmid"]): g for g in guests} - for g in guests: + self._show_guests() + + def _show_guests(self) -> None: + """Rebuild the table from self.guests: only the guests matching the filter.""" + table = self.query_one("#guests", DataTable) + current = ( # the key, not _selected(): self.guests may already be the new listing + table.coordinate_to_cell_key(table.cursor_coordinate).row_key.value + if table.row_count + else None + ) + table.clear() + query = self.query_one("#filter", Input).value + for g in self.guests.values(): + if not guest_matches(g, query): + continue _add_plain(table, "!" if g["vmid"] in self.unprotected else "", g["vmid"], g["name"], g["type"], g["node"], g["status"], @@ -1162,10 +1221,32 @@ class GuestApp(App): *_percents(g), key=str(g["vmid"]), ) # fmt: skip - if current and str(current["vmid"]) in self.guests: - table.move_cursor(row=table.get_row_index(str(current["vmid"]))) + if current is not None: + try: + table.move_cursor(row=table.get_row_index(current)) + except RowDoesNotExist: # gone or filtered out: keep the default cursor + pass self._update_frame() + def action_filter(self) -> None: + box = self.query_one("#filter", Input) + box.display = True + box.focus() + + def action_clear_filter(self) -> None: + box = self.query_one("#filter", Input) + box.value = "" # Input.Changed rebuilds the table + box.display = False + self.query_one("#guests", DataTable).focus() + + def on_input_changed(self, event: Input.Changed) -> None: + if event.input.id == "filter": + self._show_guests() + + def on_input_submitted(self, event: Input.Submitted) -> None: + if event.input.id == "filter": + self.query_one("#guests", DataTable).focus() + def _reload(self, acted: dict | None = None) -> None: """Runs in a worker thread. `acted` gets its live status: the listing lags.""" gen = self.generation diff --git a/tests/test_app.py b/tests/test_app.py index 3610698..b98d1bb 100644 --- a/tests/test_app.py +++ b/tests/test_app.py @@ -9,6 +9,8 @@ import threading from datetime import datetime import pytest +from hypothesis import given +from hypothesis import strategies as st from textual.screen import ModalScreen from textual.widgets import ( Button, @@ -3511,3 +3513,365 @@ async def test_cli_not_runnable_is_an_error_not_a_crash(fake): async with app.run_test(size=SIZE) as pilot: await settle(app, pilot) assert any("cv4pve-cli" in m for m in errors(app)) and app.is_running + + +# --- #16: search and filter in the guest table -------------------------------------------- +# spec: docs/plans/guest-tui-mvp.md ("Search and filter in the guest table (issue #16)") + +TAGGED_VM = {**VM, "tags": "172.16.41.200;community-script;os"} +TAGGED_CT = {**CT, "tags": "backup;Nightly"} + + +def fguest(**over): + return {**TAGGED_VM, **over} + + +@pytest.mark.parametrize( + ("query", "expected"), + [ + ("dock", True), # plain term: name substring + ("DOCK", True), # case-insensitive + ("108", True), # plain term: vmid + ("10", True), # vmid substring + ("community", True), # plain term: tag substring + ("172.16.41.200", True), # a whole tag + ("COMMUNITY-script", True), # tag, case-insensitive + ("nomatch", False), + ("999", False), + ], +) +def test_guest_matches_plain_term_on_name_vmid_tag(query, expected): # plain term + assert appmod.guest_matches(fguest(), query) is expected + + +def test_guest_matches_terms_are_anded(): # all terms must match + g = fguest() + assert appmod.guest_matches(g, "dock community") is True + assert appmod.guest_matches(g, "dock nomatch") is False + assert appmod.guest_matches(g, "nomatch community") is False + assert appmod.guest_matches(g, "dock 108 os is:running type:vm") is True + + +@pytest.mark.parametrize( + ("status", "query", "expected"), + [ + ("running", "is:running", True), + ("running", "is:stopped", False), + ("stopped", "is:stopped", True), + ("stopped", "is:running", False), + ("running", "IS:Running", True), # case-insensitive + ], +) +def test_guest_matches_is_status(status, query, expected): # is:running / is:stopped + assert appmod.guest_matches(fguest(status=status), query) is expected + + +@pytest.mark.parametrize( + ("kind", "query", "expected"), + [ + ("lxc", "type:lxc", True), + ("lxc", "type:ct", True), + ("lxc", "type:vm", False), + ("lxc", "type:qemu", False), + ("qemu", "type:vm", True), + ("qemu", "type:qemu", True), + ("qemu", "type:lxc", False), + ("qemu", "type:ct", False), + ("qemu", "TYPE:VM", True), # case-insensitive + ], +) +def test_guest_matches_type(kind, query, expected): # type:lxc|ct / type:vm|qemu + assert appmod.guest_matches(fguest(type=kind), query) is expected + + +@pytest.mark.parametrize("query", ["", " "]) +def test_guest_matches_empty_query_matches_all(query): # empty filter = no filter + assert appmod.guest_matches(fguest(), query) is True + assert appmod.guest_matches(TAGGED_CT, query) is True + + +def test_guest_matches_combines_keyword_and_plain_terms(): # AND across term kinds + assert appmod.guest_matches(TAGGED_CT, "type:ct is:stopped back") is True + assert appmod.guest_matches(TAGGED_CT, "type:ct is:running back") is False + + +_text = st.text(alphabet="abcdXYZ0123;.-", max_size=8) +_guests = st.fixed_dictionaries( + { + "vmid": st.integers(min_value=100, max_value=999), + "name": _text, + "type": st.sampled_from(["qemu", "lxc"]), + "status": st.sampled_from(["running", "stopped"]), + "tags": _text, + } +) +_terms = st.one_of( + _text.filter(bool), + st.sampled_from(["is:running", "is:stopped", "type:lxc", "type:vm", "type:ct"]), +) + + +@given(g=_guests, terms=st.lists(_terms, max_size=4), extra=_terms) +def test_guest_matches_property_empty_matches_and_more_terms_only_narrow( + g, terms, extra +): # empty query matches all; AND is monotone + assert appmod.guest_matches(g, "") is True + base = " ".join(terms) + if not appmod.guest_matches(g, base): + assert not appmod.guest_matches(g, f"{base} {extra}") + + +# --- UI ------------------------------------------------------------------------------------- + + +def three_guests(fake): + fake.guests = [ + TAGGED_VM, + TAGGED_CT, + {**VM, "vmid": 200, "name": "web", "status": "stopped", "tags": "frontend"}, + ] + + +def filter_input(app): + return app.query_one("#filter", Input) + + +async def type_filter(app, pilot, text): + await pilot.press("slash") + await pilot.pause() + for ch in text: + await pilot.press("space" if ch == " " else ch) + await settle(app, pilot) + + +async def test_slash_shows_and_focuses_the_filter_input(fake): # '/' shows + focuses + three_guests(fake) + app = make_app() + async with app.run_test(size=SIZE) as pilot: + await settle(app, pilot) + assert not visible(app, filter_input(app)) # hidden until then + await pilot.press("slash") + await pilot.pause() + assert visible(app, filter_input(app)) + assert app.focused is filter_input(app) + assert "is:running" in filter_input(app).placeholder # syntax hint + + +async def test_typing_filters_rows_live(fake): # table filters as you type + three_guests(fake) + app = make_app() + async with app.run_test(size=SIZE) as pilot: + await settle(app, pilot) + assert sorted(rows(app)) == ["100", "108", "200"] + await pilot.press("slash") + await pilot.press("w", "e") + await settle(app, pilot) + assert rows(app) == ["200"] + await pilot.press("b") + await settle(app, pilot) + assert rows(app) == ["200"] + await pilot.press("backspace", "backspace", "backspace") + await settle(app, pilot) + assert sorted(rows(app)) == ["100", "108", "200"] + + +async def test_filter_keywords_in_the_ui( + fake, +): # is:/type: through the UI, AND of terms + three_guests(fake) + app = make_app() + async with app.run_test(size=SIZE) as pilot: + await settle(app, pilot) + await type_filter(app, pilot, "is:stopped type:vm") + assert rows(app) == ["200"] + + +async def test_subtitle_shows_filter_text_and_n_of_m( + fake, +): # "· filter: · N of M" + three_guests(fake) + app = make_app() + async with app.run_test(size=SIZE) as pilot: + await settle(app, pilot) + assert "filter" not in frame_subtitle(app) + assert " of " not in frame_subtitle(app) + await type_filter(app, pilot, "is:stopped") + sub = frame_subtitle(app) + assert "· filter: is:stopped · 2 of 3" in sub + + +async def test_subtitle_order_context_filter_then_counts( + fake, +): # context, filter, counts + three_guests(fake) + fake.unprotected = {100} + fake.updates = {"mainframe": many(3)} + app = make_app() + async with app.run_test(size=SIZE) as pilot: + await settle(app, pilot) + await type_filter(app, pilot, "back") + sub = frame_subtitle(app) + assert "mainframe" in sub and "filter: back" in sub + assert "without backup job" in sub and "package updates" in sub + assert ( + sub.index("mainframe") + < sub.index("filter: back") + < sub.index("without backup job") + < sub.index("package updates") + ) + + +async def test_enter_keeps_the_filter_and_focuses_the_table( + fake, +): # Enter in the filter + three_guests(fake) + app = make_app() + async with app.run_test(size=SIZE) as pilot: + await settle(app, pilot) + await type_filter(app, pilot, "web") + await pilot.press("enter") + await settle(app, pilot) + assert app.focused is app.query_one("#guests", DataTable) + assert rows(app) == ["200"] + assert "filter: web" in frame_subtitle(app) + assert not isinstance(app.screen, ModalScreen) # Enter did not open guest info + + +async def test_escape_clears_hides_and_shows_all_rows(fake): # Escape in the filter + three_guests(fake) + app = make_app() + async with app.run_test(size=SIZE) as pilot: + await settle(app, pilot) + await type_filter(app, pilot, "web") + assert rows(app) == ["200"] + await pilot.press("escape") + await settle(app, pilot) + assert sorted(rows(app)) == ["100", "108", "200"] + assert not visible(app, filter_input(app)) + assert filter_input(app).value == "" + assert app.focused is app.query_one("#guests", DataTable) + assert "filter" not in frame_subtitle(app) + assert " of " not in frame_subtitle(app) + + +async def test_action_after_filtering_acts_on_the_visible_guest( + fake, +): # actions: visible row + three_guests(fake) + app = make_app() + async with app.run_test(size=SIZE) as pilot: + await settle(app, pilot) + await type_filter(app, pilot, "web") + await pilot.press("enter") + await settle(app, pilot) + await pilot.press("s") + await settle(app, pilot) + assert fake.actions == [("start", 200)] # not the first, hidden guest (108) + + +async def test_reload_keeps_the_filter(fake): # reload (g) keeps the filter applied + three_guests(fake) + app = make_app() + async with app.run_test(size=SIZE) as pilot: + await settle(app, pilot) + await type_filter(app, pilot, "web") + await pilot.press("enter") + await settle(app, pilot) + before = fake.list_calls + await pilot.press("g") + await settle(app, pilot) + assert fake.list_calls > before + assert rows(app) == ["200"] + assert "· filter: web · 1 of 3" in frame_subtitle(app) + + +async def test_action_refresh_keeps_the_filter( + fake, +): # reload after actions keeps the filter + three_guests(fake) + app = make_app() + async with app.run_test(size=SIZE) as pilot: + await settle(app, pilot) + await type_filter(app, pilot, "web") + await pilot.press("enter") + await pilot.press("s") + await settle(app, pilot) + assert rows(app) == ["200"] + + +async def test_no_match_shows_empty_table_0_of_m_and_actions_do_nothing( + fake, +): # no match + three_guests(fake) + app = make_app() + async with app.run_test(size=SIZE) as pilot: + await settle(app, pilot) + await type_filter(app, pilot, "zzz") + assert rows(app) == [] + assert "0 of 3" in frame_subtitle(app) + await pilot.press("enter") # leave the filter, table focused and empty + await settle(app, pilot) + await pilot.press("s", "S", "x", "r", "t") + await settle(app, pilot) + assert fake.actions == [] + assert not isinstance(app.screen, ModalScreen) + assert app.is_running + + +async def test_context_switch_keeps_the_filter_text( + fake, +): # a context switch keeps the filter text + three_guests(fake) + fake.guests_by_context = { + "mainframe-edge": [{**VM, "vmid": 300, "name": "web-edge", "tags": ""}, + {**VM, "vmid": 301, "name": "db-edge", "tags": ""}] + } # fmt: skip + app = make_app() + async with app.run_test(size=SIZE) as pilot: + await settle(app, pilot) + await type_filter(app, pilot, "web") + await pilot.press("enter") + await settle(app, pilot) + await switch_to_edge(app, pilot) + await until(pilot, lambda: rows(app) == ["300"]) + assert "filter: web" in frame_subtitle(app) + assert "1 of 2" in frame_subtitle(app) + + +# --- #16 review findings ---------------------------------------------------------------- + + +@pytest.mark.parametrize("text", ["[/b]", "[/]", "[red]x"]) +async def test_markup_in_the_filter_is_shown_literally_and_does_not_crash(fake, text): + app = make_app() + async with app.run_test(size=SIZE) as pilot: + await settle(app, pilot) + await pilot.press("slash") + await pilot.pause() + app.query_one("#filter").value = text + await settle(app, pilot) + assert app.is_running + shown = Content.from_markup( + app.query_one("#guests_section").border_subtitle + ).plain + assert f"filter: {text}" in shown + + +async def test_escape_on_the_table_clears_an_applied_filter(fake): + app = make_app() + async with app.run_test(size=SIZE) as pilot: + await settle(app, pilot) + await type_filter(app, pilot, "is:stopped") + await pilot.press("enter") # filter applied, focus on the table + await settle(app, pilot) + await pilot.press("escape") + await settle(app, pilot) + assert "filter:" not in frame_subtitle(app) + assert app.query_one("#guests", DataTable).row_count == len(fake.guests) + + +def test_guest_matches_tolerates_none_status_and_type(): + g = {"vmid": 1, "name": "x", "status": None, "type": None, "tags": None} + assert appmod.guest_matches(g, "x") + assert not appmod.guest_matches(g, "is:running") + assert not appmod.guest_matches(g, "type:lxc") -- 2.39.5