From cbb9d95781c297b87af6049cf6fb263e8c3ab36d Mon Sep 17 00:00:00 2001 From: Thomas Christory Date: Mon, 29 Jun 2026 00:42:53 +0200 Subject: [PATCH] feat(saved-searches): back saved searches with NetBox native SavedFilter Saved searches in the TUI (Ctrl+W/Ctrl+O) and the CLI `--saved` option now resolve against NetBox's native saved filters (`extras.saved-filters`), scoped to the resource's object type, so they are interchangeable with the web UI: a search saved in nsc appears in the web UI's filter dropdown and vice versa. The object type is resolved exactly from NetBox's own registry (`/api/core/object-types/` `rest_api_endpoint`) rather than guessed by depluralizing the URL, and the endpoint match tolerates sub-path installs, absolute URLs, and casing. Filter params translate between nsc's flat `dict[str, str]` and NetBox's list-valued `parameters`. Saves upsert by name (create, else PATCH, preserving an existing slug). When NetBox is unreachable or the user lacks permission on `extras.savedfilter`, saves/loads transparently fall back to the previous local `config.yaml` store and surface a notice instead of failing. TUI store calls run off the Textual event loop (threaded, like ListScreen) so the UI never freezes. Closes #129. Co-Authored-By: Claude Opus 4.8 (1M context) --- CHANGELOG.md | 16 ++ README.md | 16 +- docs/guides/interactive-tui.md | 16 ++ nsc/cli/registration.py | 32 ++-- nsc/cli/tui_commands.py | 54 +----- nsc/config/saved_searches.py | 58 +++++++ nsc/savedfilters/__init__.py | 4 + nsc/savedfilters/objecttypes.py | 103 +++++++++++ nsc/savedfilters/params.py | 55 ++++++ nsc/savedfilters/store.py | 140 +++++++++++++++ nsc/tui/__init__.py | 2 + nsc/tui/app.py | 35 +++- nsc/tui/screens/filter.py | 36 +++- tests/cli/test_registration.py | 26 ++- tests/cli/test_tui_command.py | 44 ++--- tests/config/test_saved_search_fallback.py | 55 ++++++ tests/savedfilters/__init__.py | 0 tests/savedfilters/test_objecttypes.py | 123 +++++++++++++ tests/savedfilters/test_params.py | 64 +++++++ tests/savedfilters/test_store.py | 190 +++++++++++++++++++++ tests/tui/test_filter_screen.py | 6 + tests/tui/test_saved_filter_integration.py | 98 +++++++++++ 22 files changed, 1063 insertions(+), 110 deletions(-) create mode 100644 nsc/savedfilters/__init__.py create mode 100644 nsc/savedfilters/objecttypes.py create mode 100644 nsc/savedfilters/params.py create mode 100644 nsc/savedfilters/store.py create mode 100644 tests/config/test_saved_search_fallback.py create mode 100644 tests/savedfilters/__init__.py create mode 100644 tests/savedfilters/test_objecttypes.py create mode 100644 tests/savedfilters/test_params.py create mode 100644 tests/savedfilters/test_store.py create mode 100644 tests/tui/test_saved_filter_integration.py diff --git a/CHANGELOG.md b/CHANGELOG.md index dab04f2..cb5237b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,22 @@ All notable changes to netbox-super-cli are tracked here. Format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/) loosely. From v1.0.0 onward, releases follow [Semantic Versioning](https://semver.org/) and the version in `pyproject.toml` matches the git tag. Pre-1.0 milestones (Phase 1-5) were pinned by tag while `pyproject.toml` stayed at `0.0.1`. +## [Unreleased] + +### Changed + +- **Saved searches are now NetBox-native** ([#129]). Filter sets saved from the + TUI (`Ctrl+W`) and applied via `--saved` are stored as NetBox saved filters + (`extras.saved-filters`), scoped to the resource's object type, so they are + fully interchangeable with the web UI's filter dropdown — a search saved in + `nsc` appears in the web UI and vice versa. The object type is resolved from + NetBox's own object-type registry (`/api/core/object-types/`), so the mapping + is exact rather than guessed. When NetBox is unreachable or the user lacks + permission on `extras.savedfilter`, saves/loads transparently fall back to the + previous local `config.yaml` store and surface a notice instead of failing. + +[#129]: https://github.com/thomaschristory/netbox-super-cli/issues/129 + ## v1.4.0 — 2026-06-26 Minor release. Three user-facing features for the interactive TUI and table diff --git a/README.md b/README.md index 0e674d7..e953fa6 100644 --- a/README.md +++ b/README.md @@ -85,9 +85,12 @@ uv run nsc ipam prefixes list --filter created__gte=2026-01-01 --output yaml ### Saved searches -Filter sets you save from the TUI (Ctrl+W in the filter builder) are stored -locally under `saved_searches` in `~/.nsc/config.yaml`, keyed by tag/resource. -Re-apply one on the CLI with `--saved`: +Filter sets you save from the TUI (Ctrl+W in the filter builder) are stored as +**NetBox native saved filters** (`extras.saved-filters`), scoped to the resource's +object type. A search you save in `nsc` shows up in the NetBox web UI's filter +dropdown for that object — and a filter saved in the web UI loads in the TUI +(Ctrl+O) — so the two are fully interchangeable. Re-apply one on the CLI with +`--saved`: ```sh uv run nsc dcim devices list --saved active-switches @@ -95,9 +98,10 @@ uv run nsc dcim devices list --saved active-switches uv run nsc dcim devices list --saved active-switches --status offline ``` -This is a local, per-config convenience and is unrelated to NetBox's -server-side saved filters (`extras saved-filters`), which store web-UI query -strings. +When NetBox is unreachable (offline, or missing permission on +`extras.savedfilter`), `nsc` transparently falls back to a local store under +`saved_searches` in `~/.nsc/config.yaml`, keyed by tag/resource, and tells you it +did so. ### Object colors diff --git a/docs/guides/interactive-tui.md b/docs/guides/interactive-tui.md index 155db89..42f85bb 100644 --- a/docs/guides/interactive-tui.md +++ b/docs/guides/interactive-tui.md @@ -81,11 +81,27 @@ NetBox list endpoint exposes hundreds of query parameters. | ↓ | Move to the next field (from a text field) | | Tab / Shift+Tab | Move between fields | | Ctrl+S | Apply | +| Ctrl+W | Save the current filters as a named search | +| Ctrl+O | Load (or delete) a saved search | | Esc | Cancel | Applying re-queries the list. Reopening the builder shows your current filters so you can refine them. +### Saved searches + +Ctrl+W stores the current filter set as a NetBox **native +saved filter** (`extras.saved-filters`), scoped to the current resource's object +type. It then appears in the NetBox web UI's filter dropdown for that object, and +filters created in the web UI load here with Ctrl+O — the two +surfaces are interchangeable. In the load picker, Enter applies the +highlighted search and d deletes it. Re-apply one non-interactively +with `nsc list --saved `. + +If NetBox is unreachable (offline, or you lack permission on +`extras.savedfilter`), saves and loads transparently fall back to a local store in +`~/.nsc/config.yaml` and a toast tells you the fallback was used. + ## Viewing and editing a record The detail view **is** the edit surface — there is no separate "edit window". diff --git a/nsc/cli/registration.py b/nsc/cli/registration.py index 59fb835..d3a15c4 100644 --- a/nsc/cli/registration.py +++ b/nsc/cli/registration.py @@ -128,22 +128,33 @@ def _resolve_saved_filters( ctx: RuntimeContext, tag_name: str, resource_name: str, + list_path: str, saved_name: str | None, explicit_filters: list[tuple[str, str]], typed_kwargs: dict[str, Any], ) -> list[tuple[str, str]]: """Layer a `--saved` filter set under explicit `--filter` and typed options. - Saved params are the base; explicit `--filter` pairs and any non-None typed - query option (already in ``typed_kwargs``) win. Order matters: ``handle_list`` - overlays ``ctx.filters`` onto the typed-option params, so saved keys that the - user also set via a typed flag must be dropped here to let the flag win. + The set is resolved against NetBox's native saved filters (interchangeable + with the web UI), falling back to the local `config.yaml` map when the API is + unavailable. Saved params are the base; explicit `--filter` pairs and any + non-None typed query option (already in ``typed_kwargs``) win. Order matters: + ``handle_list`` overlays ``ctx.filters`` onto the typed-option params, so + saved keys that the user also set via a typed flag must be dropped here to let + the flag win. """ if saved_name is None: return explicit_filters - from nsc.config.saved_searches import get_saved_search # noqa: PLC0415 + from nsc.config.saved_searches import ConfigFileSavedSearchStore # noqa: PLC0415 + from nsc.savedfilters.store import NativeSavedFilterStore # noqa: PLC0415 - saved = get_saved_search(ctx.config, tag_name, resource_name, saved_name) + store = NativeSavedFilterStore( + ctx.client, + ConfigFileSavedSearchStore(ctx.config), + on_error=lambda message: typer.echo(f"Warning: {message}", err=True), + ) + available = store.list(list_path, tag_name, resource_name) + saved = available.get(saved_name) if saved is None: raise typer.BadParameter( f"no saved search named {saved_name!r} for {tag_name}/{resource_name}" @@ -191,7 +202,7 @@ def impl(**kwargs: Any) -> None: "limit": limit, "fetch_all": fetch_all, "filters": _resolve_saved_filters( - ctx, tag_name, resource_name, saved_name, explicit_filters, kwargs + ctx, tag_name, resource_name, operation.path, saved_name, explicit_filters, kwargs ), } if output: @@ -350,9 +361,10 @@ def _build_global_flag_params() -> tuple[inspect.Parameter, ...]: None, "--saved", help=( - "Apply a LOCAL named filter set saved from the TUI (config " - "`saved_searches`). Explicit --filter/typed options override it. " - "Unrelated to NetBox server-side saved filters." + "Apply a named saved filter for this resource. Resolves against " + "NetBox's native saved filters (interchangeable with the web UI), " + "falling back to the local config `saved_searches` when the API is " + "unavailable. Explicit --filter/typed options override it." ), ), ), diff --git a/nsc/cli/tui_commands.py b/nsc/cli/tui_commands.py index cfb877a..202cc44 100644 --- a/nsc/cli/tui_commands.py +++ b/nsc/cli/tui_commands.py @@ -41,44 +41,13 @@ def _save_columns(tag: str, resource: str, columns: list[str]) -> None: atomic_write(path, dump_round_trip(doc)) -def _save_search(tag: str, resource: str, name: str, params: dict[str, str]) -> None: - """Persist a filter set to ``saved_searches...``.""" - from nsc.config.saved_searches import validate_saved_search_name # noqa: PLC0415 - from nsc.config.settings import default_paths # noqa: PLC0415 - from nsc.config.writer import ( # noqa: PLC0415 - acquire_lock, - atomic_write, - dump_round_trip, - load_round_trip, - set_path, - ) - - # Defensive: a dotted name would split into a nested map under set_path and - # corrupt the file. Validate before touching disk so the file is untouched. - validate_saved_search_name(name) - path = default_paths().config_file - with acquire_lock(path): - doc = load_round_trip(path) - set_path(doc, f"saved_searches.{tag}.{resource}.{name}", dict(params)) - atomic_write(path, dump_round_trip(doc)) +def _build_saved_filter_store(runtime: RuntimeContext) -> object: + """Native NetBox SavedFilter store with a config.yaml offline fallback.""" + from nsc.config.saved_searches import ConfigFileSavedSearchStore # noqa: PLC0415 + from nsc.savedfilters.store import NativeSavedFilterStore # noqa: PLC0415 - -def _delete_search(tag: str, resource: str, name: str) -> None: - """Remove ``saved_searches...`` and prune empty parents.""" - from nsc.config.settings import default_paths # noqa: PLC0415 - from nsc.config.writer import ( # noqa: PLC0415 - acquire_lock, - atomic_write, - dump_round_trip, - load_round_trip, - unset_path, - ) - - path = default_paths().config_file - with acquire_lock(path): - doc = load_round_trip(path) - unset_path(doc, f"saved_searches.{tag}.{resource}.{name}") - atomic_write(path, dump_round_trip(doc)) + fallback = ConfigFileSavedSearchStore(runtime.config) + return NativeSavedFilterStore(runtime.client, fallback) def register(app: typer.Typer) -> None: @@ -94,13 +63,6 @@ def tui( from nsc.tui import run_tui # noqa: PLC0415 # deferred: keeps Textual lazy. column_prefs = {tag: dict(resources) for tag, resources in runtime.config.columns.items()} - saved_searches = { - tag: { - res: {name: dict(params) for name, params in sets.items()} - for res, sets in resources.items() - } - for tag, resources in runtime.config.saved_searches.items() - } run_tui( runtime.command_model, runtime.client, @@ -108,9 +70,7 @@ def tui( save_columns=_save_columns, column_prefs=column_prefs, object_colors=runtime.object_colors, - saved_searches=saved_searches, - save_search=_save_search, - delete_search=_delete_search, + saved_filter_store=_build_saved_filter_store(runtime), ) # `tui` is canonical; `interactive` and `i` are hidden aliases for the same diff --git a/nsc/config/saved_searches.py b/nsc/config/saved_searches.py index c2080b5..50dc528 100644 --- a/nsc/config/saved_searches.py +++ b/nsc/config/saved_searches.py @@ -6,6 +6,8 @@ from __future__ import annotations +from pathlib import Path + from nsc.config.models import Config _MIN_PRINTABLE = 0x20 @@ -52,3 +54,59 @@ def get_saved_search(config: Config, tag: str, resource: str, name: str) -> dict def list_saved_searches(config: Config, tag: str, resource: str) -> list[str]: """Sorted names of saved searches for `.` (empty if none).""" return sorted(config.saved_searches.get(tag, {}).get(resource, {})) + + +class ConfigFileSavedSearchStore: + """Offline fallback: saved searches in `config.yaml`, keyed by tag/resource. + + Used when NetBox's native saved filters can't be reached. Reads come from the + in-memory `Config`; writes round-trip `config.yaml` (preserving comments) and + mutate the in-memory config so a later read in the same session is consistent. + """ + + def __init__(self, config: Config, *, config_file: Path | None = None) -> None: + self._config = config + if config_file is None: + from nsc.config.settings import default_paths # noqa: PLC0415 + + config_file = default_paths().config_file + self._config_file = config_file + + def list(self, tag: str, resource: str) -> dict[str, dict[str, str]]: + sets = self._config.saved_searches.get(tag, {}).get(resource, {}) + return {name: dict(params) for name, params in sets.items()} + + def save(self, tag: str, resource: str, name: str, params: dict[str, str]) -> None: + # Validate before touching disk: a dotted name would split into a nested + # map under set_path and corrupt the file. + validate_saved_search_name(name) + from nsc.config.writer import ( # noqa: PLC0415 + acquire_lock, + atomic_write, + dump_round_trip, + load_round_trip, + set_path, + ) + + with acquire_lock(self._config_file): + doc = load_round_trip(self._config_file) + set_path(doc, f"saved_searches.{tag}.{resource}.{name}", dict(params)) + atomic_write(self._config_file, dump_round_trip(doc)) + self._config.saved_searches.setdefault(tag, {}).setdefault(resource, {})[name] = dict( + params + ) + + def delete(self, tag: str, resource: str, name: str) -> None: + from nsc.config.writer import ( # noqa: PLC0415 + acquire_lock, + atomic_write, + dump_round_trip, + load_round_trip, + unset_path, + ) + + with acquire_lock(self._config_file): + doc = load_round_trip(self._config_file) + unset_path(doc, f"saved_searches.{tag}.{resource}.{name}") + atomic_write(self._config_file, dump_round_trip(doc)) + self._config.saved_searches.get(tag, {}).get(resource, {}).pop(name, None) diff --git a/nsc/savedfilters/__init__.py b/nsc/savedfilters/__init__.py new file mode 100644 index 0000000..74ed893 --- /dev/null +++ b/nsc/savedfilters/__init__.py @@ -0,0 +1,4 @@ +"""Native NetBox SavedFilter (``extras.saved-filters``) backing for nsc's +saved searches, plus the object-type resolution that makes them interchangeable +with the NetBox web UI. Framework-free: no Typer, no Textual. +""" diff --git a/nsc/savedfilters/objecttypes.py b/nsc/savedfilters/objecttypes.py new file mode 100644 index 0000000..d369c27 --- /dev/null +++ b/nsc/savedfilters/objecttypes.py @@ -0,0 +1,103 @@ +"""Resolve a NetBox list endpoint to its ``app_label.model`` object type. + +NetBox identifies a saved filter's target by object type (e.g. ``dcim.device``). +The only robust, version-stable mapping from one of nsc's list resources to that +string is NetBox's own object-type registry: ``/api/core/object-types/`` returns, +for every model, its ``app_label``, ``model``, and ``rest_api_endpoint`` (the +model's list URL). Matching a resource's list path against ``rest_api_endpoint`` +avoids fragile depluralization guesses (``ip-addresses`` -> ``ipaddress``) and the +schema-name noise (devices' schema is ``DeviceWithConfigContext``, not ``Device``). +""" + +from __future__ import annotations + +from typing import Any, Protocol +from urllib.parse import urlsplit + +from nsc.http.errors import NetBoxAPIError, NetBoxClientError + +_OBJECT_TYPES_PATH = "/api/core/object-types/" + + +class _ClientLike(Protocol): + def paginate( + self, path: str, params: dict[str, Any] | None = ..., *, limit: int | None = ... + ) -> Any: ... + + +def normalize_endpoint(path_or_url: str) -> str: + """A list path/URL reduced to a comparable ``/api/.../`` form. + + Strips any scheme+host and any deployment sub-path prefix, lowercases, and + ensures a single trailing slash, so a resource's ``Operation.path`` (always a + bare ``/api/...``) and an object type's server-built ``rest_api_endpoint`` + compare equal regardless of absolute-vs-relative, casing, *or* a NetBox + sub-path install (where ``rest_api_endpoint`` is e.g. ``/netbox/api/...``). + """ + path = urlsplit(path_or_url).path.lower() + marker = "/api/" + idx = path.find(marker) + if idx > 0: + path = path[idx:] + if not path.endswith("/"): + path += "/" + return path + + +def app_label_from_path(list_path: str) -> str | None: + """The NetBox app label a list path belongs to, or None if not derivable. + + ``/api/dcim/devices/`` -> ``dcim``; plugin routes + ``/api/plugins//...`` -> ````. + """ + parts = [p for p in urlsplit(list_path).path.split("/") if p] + if not parts or parts[0] != "api" or len(parts) < 2: # noqa: PLR2004 + return None + if parts[1] == "plugins": + return parts[2] if len(parts) >= 3 else None # noqa: PLR2004 + return parts[1] + + +def object_type_index(object_types: Any) -> dict[str, str]: + """Map normalized ``rest_api_endpoint`` -> ``app_label.model`` for each type.""" + index: dict[str, str] = {} + for ot in object_types: + endpoint = ot.get("rest_api_endpoint") + app_label = ot.get("app_label") + model = ot.get("model") + if endpoint and app_label and model: + index[normalize_endpoint(endpoint)] = f"{app_label}.{model}" + return index + + +class ObjectTypeResolver: + """Looks up object types from the live registry, cached per app label.""" + + def __init__(self, client: _ClientLike) -> None: + self._client = client + self._cache: dict[str, dict[str, str]] = {} + + def resolve(self, list_path: str) -> str | None: + """``app_label.model`` for a list path, or None if it can't be resolved. + + Returns None (rather than raising) when the path is unparseable or the + registry can't be reached, so callers can fall back to local storage. + """ + app_label = app_label_from_path(list_path) + if app_label is None: + return None + index = self._index_for(app_label) + if index is None: + return None + return index.get(normalize_endpoint(list_path)) + + def _index_for(self, app_label: str) -> dict[str, str] | None: + if app_label in self._cache: + return self._cache[app_label] + try: + records = list(self._client.paginate(_OBJECT_TYPES_PATH, {"app_label": app_label})) + except (NetBoxAPIError, NetBoxClientError): + return None + index = object_type_index(records) + self._cache[app_label] = index + return index diff --git a/nsc/savedfilters/params.py b/nsc/savedfilters/params.py new file mode 100644 index 0000000..b2faa79 --- /dev/null +++ b/nsc/savedfilters/params.py @@ -0,0 +1,55 @@ +"""Translation between nsc's flat filter params and NetBox SavedFilter shapes. + +NetBox's web UI persists a saved filter's form data as a QueryDict-shaped +mapping where every value is a *list* of strings (e.g. ``{"status": ["active"]}``) +and the object is keyed by a unique ``slug``. nsc models active filters as a flat +``dict[str, str]``. These helpers convert between the two so a filter saved in +either surface is usable in the other. +""" + +from __future__ import annotations + +import re + +_SLUG_STRIP = re.compile(r"[^a-z0-9_]+") + + +def to_netbox_parameters(params: dict[str, str]) -> dict[str, list[str]]: + """nsc flat params -> NetBox ``parameters`` (each value wrapped in a list).""" + return {key: [value] for key, value in params.items()} + + +def from_netbox_parameters(params: dict[str, object]) -> dict[str, str]: + """NetBox ``parameters`` -> nsc flat params. + + List values flatten to their first element (nsc holds one value per key, so + a web-UI multi-select degrades to its first value rather than being lost). + Empty lists, ``None``, and empty strings are skipped. + """ + out: dict[str, str] = {} + for key, value in params.items(): + if isinstance(value, list): + if not value: + continue + first = value[0] + if first is None or first == "": + continue + out[key] = str(first) + elif value is None or value == "": + continue + else: + out[key] = str(value) + return out + + +def slugify(name: str) -> str: + """A NetBox-safe slug for ``name`` (``^[-a-zA-Z0-9_]+$``). + + Lowercases, turns runs of unsafe characters into single hyphens, and strips + leading/trailing hyphens. Names with no slug-safe characters fall back to a + short hex digest so the result is always a non-empty, valid slug. + """ + slug = _SLUG_STRIP.sub("-", name.strip().lower()).strip("-") + if slug: + return slug + return "f-" + name.encode("utf-8").hex()[:12] diff --git a/nsc/savedfilters/store.py b/nsc/savedfilters/store.py new file mode 100644 index 0000000..52d80b3 --- /dev/null +++ b/nsc/savedfilters/store.py @@ -0,0 +1,140 @@ +"""A saved-search store backed by NetBox ``extras.saved-filters`` objects. + +Primary storage is the server, so a search saved here appears in the NetBox web +UI (and vice versa). Every operation degrades to a local fallback (config.yaml) +when the object type can't be resolved or the API is unreachable, and surfaces +that degradation through an optional ``on_error`` callback rather than failing +silently. +""" + +from __future__ import annotations + +from collections.abc import Callable +from typing import Any, Protocol, cast + +from nsc.http.errors import NetBoxAPIError, NetBoxClientError +from nsc.savedfilters.objecttypes import ObjectTypeResolver +from nsc.savedfilters.params import ( + from_netbox_parameters, + slugify, + to_netbox_parameters, +) + +SAVED_FILTERS_PATH = "/api/extras/saved-filters/" + + +class SavedSearchFallback(Protocol): + """Local, offline storage keyed by ``(tag, resource)`` — the config.yaml map.""" + + def list(self, tag: str, resource: str) -> dict[str, dict[str, str]]: ... + def save(self, tag: str, resource: str, name: str, params: dict[str, str]) -> None: ... + def delete(self, tag: str, resource: str, name: str) -> None: ... + + +class _ResolverLike(Protocol): + def resolve(self, list_path: str) -> str | None: ... + + +class NativeSavedFilterStore: + def __init__( + self, + client: Any, + fallback: SavedSearchFallback, + *, + resolver: _ResolverLike | None = None, + on_error: Callable[[str], None] | None = None, + ) -> None: + self._client = client + self._fallback = fallback + self._resolver = resolver or ObjectTypeResolver(client) + self.on_error = on_error + + def list(self, list_path: str, tag: str, resource: str) -> dict[str, dict[str, str]]: + object_type = self._resolver.resolve(list_path) + if object_type is None: + return self._fallback.list(tag, resource) + try: + records = self._client.paginate(SAVED_FILTERS_PATH, {"object_type": object_type}) + out: dict[str, dict[str, str]] = {} + for rec in records: + name = rec.get("name") + if not name: + continue + if name in out: + # NetBox allows same-named filters (e.g. a per-user one and a + # shared one). Keep the first deterministically — matching + # _find's save/delete target — and flag the ambiguity. + self._notify(f"Multiple NetBox saved filters named {name!r}; using the first.") + continue + out[name] = from_netbox_parameters(rec.get("parameters") or {}) + return out + except (NetBoxAPIError, NetBoxClientError) as exc: + self._report("load saved searches", exc) + return self._fallback.list(tag, resource) + + def save( + self, list_path: str, tag: str, resource: str, name: str, params: dict[str, str] + ) -> None: + object_type = self._resolver.resolve(list_path) + if object_type is None: + self._fallback.save(tag, resource, name, params) + return + try: + existing = self._find(object_type, name) + parameters = to_netbox_parameters(params) + if existing is None: + self._client.post( + SAVED_FILTERS_PATH, + json={ + "name": name, + "slug": slugify(f"{object_type}-{name}"), + "object_types": [object_type], + "parameters": parameters, + "enabled": True, + "shared": True, + }, + operation_id="extras_saved_filters_create", + ) + else: + # Update parameters only; leaving name/slug intact preserves a + # slug NetBox (or the web UI) already assigned to this filter. + self._client.patch( + f"{SAVED_FILTERS_PATH}{existing['id']}/", + json={"object_types": [object_type], "parameters": parameters}, + operation_id="extras_saved_filters_partial_update", + ) + except (NetBoxAPIError, NetBoxClientError) as exc: + self._report(f"save search {name!r}", exc) + self._fallback.save(tag, resource, name, params) + + def delete(self, list_path: str, tag: str, resource: str, name: str) -> None: + object_type = self._resolver.resolve(list_path) + if object_type is None: + self._fallback.delete(tag, resource, name) + return + try: + existing = self._find(object_type, name) + if existing is not None: + self._client.delete( + f"{SAVED_FILTERS_PATH}{existing['id']}/", + operation_id="extras_saved_filters_destroy", + ) + except (NetBoxAPIError, NetBoxClientError) as exc: + self._report(f"delete search {name!r}", exc) + self._fallback.delete(tag, resource, name) + + def _find(self, object_type: str, name: str) -> dict[str, Any] | None: + records = self._client.paginate( + SAVED_FILTERS_PATH, {"object_type": object_type, "name": name} + ) + for rec in records: + if rec.get("name") == name: + return cast(dict[str, Any], rec) + return None + + def _report(self, action: str, exc: Exception) -> None: + self._notify(f"Could not {action} via NetBox ({exc}); using local config.") + + def _notify(self, message: str) -> None: + if self.on_error is not None: + self.on_error(message) diff --git a/nsc/tui/__init__.py b/nsc/tui/__init__.py index abb5288..737cc2e 100644 --- a/nsc/tui/__init__.py +++ b/nsc/tui/__init__.py @@ -24,6 +24,7 @@ def run_tui( saved_searches: SavedSearchMap | None = None, save_search: Callable[[str, str, str, dict[str, str]], None] | None = None, delete_search: Callable[[str, str, str], None] | None = None, + saved_filter_store: Any | None = None, ) -> None: """Lazy entrypoint so importing `nsc.tui` never imports Textual eagerly.""" from nsc.tui.app import run_tui as _run # noqa: PLC0415 # deferred: keeps Textual lazy. @@ -38,4 +39,5 @@ def run_tui( saved_searches=saved_searches, save_search=save_search, delete_search=delete_search, + saved_filter_store=saved_filter_store, ) diff --git a/nsc/tui/app.py b/nsc/tui/app.py index ff11198..b7b3371 100644 --- a/nsc/tui/app.py +++ b/nsc/tui/app.py @@ -38,6 +38,7 @@ def __init__( saved_searches: SavedSearchMap | None = None, save_search: Callable[[str, str, str, dict[str, str]], None] | None = None, delete_search: Callable[[str, str, str], None] | None = None, + saved_filter_store: Any | None = None, ) -> None: super().__init__() self._model = model @@ -49,6 +50,9 @@ def __init__( self._saved_searches: SavedSearchMap = saved_searches or {} self._save_search = save_search self._delete_search = delete_search + self._saved_filter_store = saved_filter_store + if saved_filter_store is not None and getattr(saved_filter_store, "on_error", None) is None: + saved_filter_store.on_error = self._notify_saved_filter_issue def columns_for(self, tag: str, resource: str) -> list[str] | None: """Saved visible columns for a resource, if any (read by ListScreen).""" @@ -61,11 +65,35 @@ def save_columns(self, tag: str, resource: str, columns: list[str]) -> None: if self._save_columns is not None: self._save_columns(tag, resource, columns) + def _list_path(self, tag: str, resource: str) -> str: + """The NetBox list URL for a resource, used to resolve its object type.""" + try: + list_op = self._model.tags[tag].resources[resource].list_op + except KeyError: + return "" + return list_op.path if list_op is not None else "" + + def _notify_saved_filter_issue(self, message: str) -> None: + self.notify(message, severity="warning") + def saved_searches_for(self, tag: str, resource: str) -> dict[str, dict[str, str]]: - """Named saved filter sets for a resource (read by FilterScreen).""" + """Named saved filter sets for a resource (read by FilterScreen). + + Backed by NetBox's native saved filters when a store is wired (so the web + UI's filters appear here too); otherwise the in-memory config map. + """ + if self._saved_filter_store is not None: + result: dict[str, dict[str, str]] = self._saved_filter_store.list( + self._list_path(tag, resource), tag, resource + ) + return result return self._saved_searches.get(tag, {}).get(resource, {}) def save_search(self, tag: str, resource: str, name: str, params: dict[str, str]) -> None: + if self._saved_filter_store is not None: + path = self._list_path(tag, resource) + self._saved_filter_store.save(path, tag, resource, name, params) + return # Update the in-memory map too so the picker reflects the new entry # immediately, without a relaunch. self._saved_searches.setdefault(tag, {}).setdefault(resource, {})[name] = dict(params) @@ -73,6 +101,9 @@ def save_search(self, tag: str, resource: str, name: str, params: dict[str, str] self._save_search(tag, resource, name, params) def delete_search(self, tag: str, resource: str, name: str) -> None: + if self._saved_filter_store is not None: + self._saved_filter_store.delete(self._list_path(tag, resource), tag, resource, name) + return self._saved_searches.get(tag, {}).get(resource, {}).pop(name, None) if self._delete_search is not None: self._delete_search(tag, resource, name) @@ -129,6 +160,7 @@ def run_tui( saved_searches: SavedSearchMap | None = None, save_search: Callable[[str, str, str, dict[str, str]], None] | None = None, delete_search: Callable[[str, str, str], None] | None = None, + saved_filter_store: Any | None = None, ) -> None: NscTuiApp( model, @@ -140,4 +172,5 @@ def run_tui( saved_searches=saved_searches, save_search=save_search, delete_search=delete_search, + saved_filter_store=saved_filter_store, ).run() diff --git a/nsc/tui/screens/filter.py b/nsc/tui/screens/filter.py index a7addde..98b1bed 100644 --- a/nsc/tui/screens/filter.py +++ b/nsc/tui/screens/filter.py @@ -11,6 +11,7 @@ from __future__ import annotations +import asyncio from typing import Any, ClassVar from textual.app import ComposeResult @@ -271,6 +272,8 @@ def action_save_search(self) -> None: ) from nsc.tui.screens.saved_search_picker import SavedSearchNamePrompt # noqa: PLC0415 + tag, resource = self._tag, self._resource + def _save(name: str | None) -> None: if not name or not name.strip(): return @@ -280,19 +283,30 @@ def _save(name: str | None) -> None: except InvalidSavedSearchName as exc: self.notify(str(exc), severity="error") return - self.app.save_search( # type: ignore[attr-defined] - self._tag, self._resource, cleaned, self.state.as_params() - ) + # The store may hit the NetBox API; keep it off the event loop so the + # UI never freezes (mirrors ListScreen's threaded fetch). + self.run_worker(self._save_worker(tag, resource, cleaned, self.state.as_params())) self.app.push_screen(SavedSearchNamePrompt(), _save) + async def _save_worker( + self, tag: str, resource: str, name: str, params: dict[str, str] + ) -> None: + await asyncio.to_thread(self.app.save_search, tag, resource, name, params) # type: ignore[attr-defined] + def action_load_search(self) -> None: if self._tag is None or self._resource is None: return - reader = getattr(self.app, "saved_searches_for", None) - if not callable(reader): + if not callable(getattr(self.app, "saved_searches_for", None)): return - saved: dict[str, dict[str, str]] = reader(self._tag, self._resource) + self.run_worker(self._load_worker(self._tag, self._resource)) + + async def _load_worker(self, tag: str, resource: str) -> None: + saved: dict[str, dict[str, str]] = await asyncio.to_thread( + self.app.saved_searches_for, # type: ignore[attr-defined] + tag, + resource, + ) if not saved: self.notify("No saved searches for this resource yet.") return @@ -318,8 +332,12 @@ def _chosen(choice: SavedSearchChoice | None) -> None: self.app.push_screen(SavedSearchPicker(sorted(saved)), _chosen) def _delete_saved(self, name: str) -> None: - deleter = getattr(self.app, "delete_search", None) - if not callable(deleter): + if self._tag is None or self._resource is None: return - deleter(self._tag, self._resource, name) + if not callable(getattr(self.app, "delete_search", None)): + return + self.run_worker(self._delete_worker(self._tag, self._resource, name)) + + async def _delete_worker(self, tag: str, resource: str, name: str) -> None: + await asyncio.to_thread(self.app.delete_search, tag, resource, name) # type: ignore[attr-defined] self.notify(f"Deleted saved search {name!r}.") diff --git a/tests/cli/test_registration.py b/tests/cli/test_registration.py index 6c599e1..a6ad4eb 100644 --- a/tests/cli/test_registration.py +++ b/tests/cli/test_registration.py @@ -367,6 +367,21 @@ def _ctx_with_saved(client: Any, saved: dict[str, Any]) -> RuntimeContext: ) +def _data_list_params(client: MagicMock) -> dict[str, Any] | None: + """Params of the data-list paginate call (ignoring object-type resolution). + + ``--saved`` resolves through the native SavedFilter store, which probes + ``/api/core/object-types/`` (and ``/api/extras/saved-filters/``) before the + actual list fetch; with a MagicMock those probe empty and resolution falls + back to the local config ``saved_searches``. We assert on the final data call. + """ + for c in client.paginate.call_args_list: + path = c.args[0] if c.args else c.kwargs.get("path") + if path == "/api/dcim/devices/": + return c.args[1] if len(c.args) > 1 else c.kwargs.get("params") + return None + + def test_list_saved_resolves_into_filters() -> None: app = typer.Typer() client = MagicMock() @@ -376,9 +391,7 @@ def test_list_saved_resolves_into_filters() -> None: register_dynamic_commands(app, model, lambda: ctx) result = CliRunner().invoke(app, ["dcim", "devices", "list", "--saved", "active-sw"]) assert result.exit_code == 0, result.output - client.paginate.assert_called_once_with( - "/api/dcim/devices/", {"status": "active", "role": "switch"} - ) + assert _data_list_params(client) == {"status": "active", "role": "switch"} def test_explicit_filter_overrides_saved() -> None: @@ -392,7 +405,7 @@ def test_explicit_filter_overrides_saved() -> None: app, ["dcim", "devices", "list", "--saved", "active-sw", "--filter", "status=offline"] ) assert result.exit_code == 0, result.output - client.paginate.assert_called_once_with("/api/dcim/devices/", {"status": "offline"}) + assert _data_list_params(client) == {"status": "offline"} def test_explicit_typed_option_overrides_saved() -> None: @@ -408,7 +421,7 @@ def test_explicit_typed_option_overrides_saved() -> None: app, ["dcim", "devices", "list", "--saved", "active-sw", "--status", "offline"] ) assert result.exit_code == 0, result.output - client.paginate.assert_called_once_with("/api/dcim/devices/", {"status": "offline"}) + assert _data_list_params(client) == {"status": "offline"} def test_unknown_saved_name_exits_2() -> None: @@ -420,4 +433,5 @@ def test_unknown_saved_name_exits_2() -> None: register_dynamic_commands(app, model, lambda: ctx) result = CliRunner().invoke(app, ["dcim", "devices", "list", "--saved", "nope"]) assert result.exit_code == 2 - client.paginate.assert_not_called() + # An unknown saved name aborts before any data is fetched. + assert _data_list_params(client) is None diff --git a/tests/cli/test_tui_command.py b/tests/cli/test_tui_command.py index 838aba6..0e52ac6 100644 --- a/tests/cli/test_tui_command.py +++ b/tests/cli/test_tui_command.py @@ -12,7 +12,8 @@ from nsc.cli.tui_commands import _runtime_from_ctx, register from nsc.config import settings from nsc.config.loader import load_config -from nsc.config.saved_searches import InvalidSavedSearchName +from nsc.config.models import Config +from nsc.config.saved_searches import ConfigFileSavedSearchStore, InvalidSavedSearchName def test_tui_help_lists_resource_argument() -> None: @@ -56,6 +57,7 @@ def _fake_run_tui( saved_searches: Any = None, save_search: Any = None, delete_search: Any = None, + saved_filter_store: Any = None, ) -> None: calls.append((model, client, initial_resource)) @@ -103,18 +105,12 @@ def config_file(self) -> Any: assert "name" in text -def test_save_search_persists_to_config(tmp_path: Any, monkeypatch: pytest.MonkeyPatch) -> None: +def test_config_fallback_save_persists_to_config(tmp_path: Any) -> None: config_file = tmp_path / "config.yaml" config_file.write_text("") + store = ConfigFileSavedSearchStore(Config(), config_file=config_file) - class _Paths: - @property - def config_file(self) -> Any: - return config_file - - monkeypatch.setattr(settings, "default_paths", _Paths) - - tui_commands._save_search("dcim", "devices", "active-sw", {"status": "active"}) + store.save("dcim", "devices", "active-sw", {"status": "active"}) text = config_file.read_text() assert "saved_searches:" in text @@ -123,19 +119,13 @@ def config_file(self) -> Any: assert "active" in text -def test_delete_search_removes_from_config(tmp_path: Any, monkeypatch: pytest.MonkeyPatch) -> None: +def test_config_fallback_delete_removes_from_config(tmp_path: Any) -> None: config_file = tmp_path / "config.yaml" config_file.write_text("") + store = ConfigFileSavedSearchStore(Config(), config_file=config_file) - class _Paths: - @property - def config_file(self) -> Any: - return config_file - - monkeypatch.setattr(settings, "default_paths", _Paths) - - tui_commands._save_search("dcim", "devices", "active-sw", {"status": "active"}) - tui_commands._delete_search("dcim", "devices", "active-sw") + store.save("dcim", "devices", "active-sw", {"status": "active"}) + store.delete("dcim", "devices", "active-sw") text = config_file.read_text() assert "active-sw" not in text @@ -143,23 +133,15 @@ def config_file(self) -> Any: assert "saved_searches" not in text -def test_save_search_dotted_name_does_not_corrupt_config( - tmp_path: Any, monkeypatch: pytest.MonkeyPatch -) -> None: +def test_config_fallback_save_dotted_name_does_not_corrupt_config(tmp_path: Any) -> None: config_file = tmp_path / "config.yaml" config_file.write_text("") - - class _Paths: - @property - def config_file(self) -> Any: - return config_file - - monkeypatch.setattr(settings, "default_paths", _Paths) + store = ConfigFileSavedSearchStore(Config(), config_file=config_file) # A dotted name must be rejected at persist time, not split into a nested # map that later fails Config.model_validate and bricks every nsc run. with pytest.raises(InvalidSavedSearchName): - tui_commands._save_search("dcim", "devices", "prod.v2", {"status": "active"}) + store.save("dcim", "devices", "prod.v2", {"status": "active"}) # The config file is untouched and still loads cleanly. assert config_file.read_text() == "" diff --git a/tests/config/test_saved_search_fallback.py b/tests/config/test_saved_search_fallback.py new file mode 100644 index 0000000..fb60869 --- /dev/null +++ b/tests/config/test_saved_search_fallback.py @@ -0,0 +1,55 @@ +from __future__ import annotations + +from pathlib import Path + +import pytest + +from nsc.config.models import Config +from nsc.config.saved_searches import ConfigFileSavedSearchStore, InvalidSavedSearchName + + +def _store(tmp_path: Path, config: Config | None = None) -> ConfigFileSavedSearchStore: + return ConfigFileSavedSearchStore(config or Config(), config_file=tmp_path / "config.yaml") + + +def test_save_persists_to_file_and_in_memory_config(tmp_path: Path) -> None: + config = Config() + store = ConfigFileSavedSearchStore(config, config_file=tmp_path / "config.yaml") + store.save("dcim", "devices", "active", {"status": "active"}) + + assert config.saved_searches == {"dcim": {"devices": {"active": {"status": "active"}}}} + assert "status: active" in (tmp_path / "config.yaml").read_text() + + +def test_list_reads_back_what_was_saved(tmp_path: Path) -> None: + store = _store(tmp_path) + store.save("dcim", "devices", "active", {"status": "active"}) + store.save("dcim", "devices", "leafs", {"role": "leaf"}) + assert store.list("dcim", "devices") == { + "active": {"status": "active"}, + "leafs": {"role": "leaf"}, + } + + +def test_list_unknown_resource_is_empty(tmp_path: Path) -> None: + assert _store(tmp_path).list("dcim", "devices") == {} + + +def test_delete_removes_from_file_and_memory(tmp_path: Path) -> None: + config = Config() + store = ConfigFileSavedSearchStore(config, config_file=tmp_path / "config.yaml") + store.save("dcim", "devices", "active", {"status": "active"}) + store.delete("dcim", "devices", "active") + assert store.list("dcim", "devices") == {} + assert config.saved_searches.get("dcim", {}).get("devices", {}) == {} + + +def test_save_rejects_dotted_name(tmp_path: Path) -> None: + with pytest.raises(InvalidSavedSearchName): + _store(tmp_path).save("dcim", "devices", "a.b", {"x": "y"}) + + +def test_list_reflects_a_preloaded_config(tmp_path: Path) -> None: + config = Config(saved_searches={"ipam": {"prefixes": {"rfc1918": {"within": "10.0.0.0/8"}}}}) + store = ConfigFileSavedSearchStore(config, config_file=tmp_path / "config.yaml") + assert store.list("ipam", "prefixes") == {"rfc1918": {"within": "10.0.0.0/8"}} diff --git a/tests/savedfilters/__init__.py b/tests/savedfilters/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/savedfilters/test_objecttypes.py b/tests/savedfilters/test_objecttypes.py new file mode 100644 index 0000000..6e98fda --- /dev/null +++ b/tests/savedfilters/test_objecttypes.py @@ -0,0 +1,123 @@ +from __future__ import annotations + +from collections.abc import Iterator +from typing import Any + +import pytest + +from nsc.http.errors import NetBoxAPIError +from nsc.savedfilters.objecttypes import ( + ObjectTypeResolver, + app_label_from_path, + normalize_endpoint, + object_type_index, +) + +_OBJECT_TYPES = [ + {"app_label": "dcim", "model": "device", "rest_api_endpoint": "/api/dcim/devices/"}, + {"app_label": "dcim", "model": "devicetype", "rest_api_endpoint": "/api/dcim/device-types/"}, + {"app_label": "ipam", "model": "ipaddress", "rest_api_endpoint": "/api/ipam/ip-addresses/"}, +] + + +def test_normalize_endpoint_handles_full_urls_and_trailing_slash() -> None: + assert normalize_endpoint("https://nb.example.com/api/dcim/devices/") == "/api/dcim/devices/" + assert normalize_endpoint("/api/dcim/devices") == "/api/dcim/devices/" + assert normalize_endpoint("/API/DCIM/Devices/") == "/api/dcim/devices/" + + +def test_normalize_endpoint_strips_deployment_subpath_prefix() -> None: + # NetBox installed under a sub-path returns rest_api_endpoint with the prefix; + # Operation.path never has it, so both must reduce to the same /api/... tail. + assert normalize_endpoint("/netbox/api/dcim/devices/") == "/api/dcim/devices/" + assert ( + normalize_endpoint("https://nb.example.com/netbox/api/dcim/devices/") + == "/api/dcim/devices/" + ) + + +def test_resolver_matches_across_subpath_casing_and_trailing_slash() -> None: + # End-to-end: a sub-path, absolute-URL, mixed-case rest_api_endpoint still + # resolves the bare lowercase Operation.path that nsc passes in. + client = _FakeClient( + [ + { + "app_label": "dcim", + "model": "device", + "rest_api_endpoint": "https://nb.example.com/NetBox/API/DCIM/Devices/", + } + ] + ) + resolver = ObjectTypeResolver(client) + assert resolver.resolve("/api/dcim/devices/") == "dcim.device" + + +def test_app_label_from_path() -> None: + assert app_label_from_path("/api/dcim/devices/") == "dcim" + assert app_label_from_path("/api/ipam/ip-addresses/") == "ipam" + assert app_label_from_path("/api/plugins/myplugin/widgets/") == "myplugin" + assert app_label_from_path("/notapi/") is None + + +def test_object_type_index_maps_endpoint_to_dotted_type() -> None: + index = object_type_index(_OBJECT_TYPES) + assert index["/api/dcim/devices/"] == "dcim.device" + assert index["/api/ipam/ip-addresses/"] == "ipam.ipaddress" + + +class _FakeClient: + def __init__(self, records: list[dict[str, Any]], *, fail: bool = False) -> None: + self._records = records + self._fail = fail + self.calls: list[tuple[str, dict[str, Any] | None]] = [] + + def paginate( + self, path: str, params: dict[str, Any] | None = None, *, limit: int | None = None + ) -> Iterator[dict[str, Any]]: + self.calls.append((path, params)) + if self._fail: + raise NetBoxAPIError(status_code=404, url=path, body_snippet="", headers={}) + app = (params or {}).get("app_label") + for rec in self._records: + if app is None or rec["app_label"] == app: + yield rec + + +def test_resolver_returns_dotted_object_type_for_a_list_path() -> None: + client = _FakeClient(_OBJECT_TYPES) + resolver = ObjectTypeResolver(client) + assert resolver.resolve("/api/ipam/ip-addresses/") == "ipam.ipaddress" + + +def test_resolver_filters_object_types_by_app_label() -> None: + client = _FakeClient(_OBJECT_TYPES) + resolver = ObjectTypeResolver(client) + resolver.resolve("/api/dcim/devices/") + assert client.calls == [("/api/core/object-types/", {"app_label": "dcim"})] + + +def test_resolver_caches_per_app_label() -> None: + client = _FakeClient(_OBJECT_TYPES) + resolver = ObjectTypeResolver(client) + resolver.resolve("/api/dcim/devices/") + resolver.resolve("/api/dcim/device-types/") + assert len(client.calls) == 1 + + +def test_resolver_returns_none_on_api_error() -> None: + client = _FakeClient(_OBJECT_TYPES, fail=True) + resolver = ObjectTypeResolver(client) + assert resolver.resolve("/api/dcim/devices/") is None + + +def test_resolver_returns_none_for_unknown_endpoint() -> None: + client = _FakeClient(_OBJECT_TYPES) + resolver = ObjectTypeResolver(client) + assert resolver.resolve("/api/dcim/nonexistent/") is None + + +@pytest.mark.parametrize("bad_path", ["", "/", "notapath"]) +def test_resolver_returns_none_for_unparseable_path(bad_path: str) -> None: + client = _FakeClient(_OBJECT_TYPES) + resolver = ObjectTypeResolver(client) + assert resolver.resolve(bad_path) is None diff --git a/tests/savedfilters/test_params.py b/tests/savedfilters/test_params.py new file mode 100644 index 0000000..67a62c5 --- /dev/null +++ b/tests/savedfilters/test_params.py @@ -0,0 +1,64 @@ +from __future__ import annotations + +from nsc.savedfilters.params import ( + from_netbox_parameters, + slugify, + to_netbox_parameters, +) + + +def test_to_netbox_parameters_wraps_each_value_in_a_list() -> None: + # NetBox's web UI stores filter form data as a QueryDict-shaped mapping where + # every value is a list; matching that shape keeps filters interchangeable. + assert to_netbox_parameters({"status": "active", "site_id": "3"}) == { + "status": ["active"], + "site_id": ["3"], + } + + +def test_to_netbox_parameters_empty() -> None: + assert to_netbox_parameters({}) == {} + + +def test_from_netbox_parameters_flattens_single_element_lists() -> None: + assert from_netbox_parameters({"status": ["active"], "site_id": ["3"]}) == { + "status": "active", + "site_id": "3", + } + + +def test_from_netbox_parameters_takes_first_of_multi_value_lists() -> None: + # nsc's filter state is single-value per key; a web-UI multi-select degrades + # to its first value rather than being dropped. + assert from_netbox_parameters({"status": ["active", "offline"]}) == {"status": "active"} + + +def test_from_netbox_parameters_accepts_scalar_values() -> None: + assert from_netbox_parameters({"q": "sw1", "limit": 50}) == {"q": "sw1", "limit": "50"} + + +def test_from_netbox_parameters_skips_empty_and_null() -> None: + assert from_netbox_parameters({"a": [], "b": None, "c": ["x"], "d": ""}) == {"c": "x"} + + +def test_round_trip_is_stable_for_single_values() -> None: + params = {"status": "active", "role": "leaf"} + assert from_netbox_parameters(to_netbox_parameters(params)) == params + + +def test_slugify_basic() -> None: + assert slugify("My Saved Search") == "my-saved-search" + + +def test_slugify_collapses_and_strips_separators() -> None: + assert slugify(" Edge // routers!! ") == "edge-routers" + + +def test_slugify_keeps_underscores_and_digits() -> None: + assert slugify("vlan_100 group") == "vlan_100-group" + + +def test_slugify_non_empty_fallback() -> None: + # A name with no slug-safe characters still yields a usable, valid slug. + assert slugify("✦✦✦") != "" + assert all(c.isalnum() or c in "-_" for c in slugify("✦✦✦")) diff --git a/tests/savedfilters/test_store.py b/tests/savedfilters/test_store.py new file mode 100644 index 0000000..c5f818f --- /dev/null +++ b/tests/savedfilters/test_store.py @@ -0,0 +1,190 @@ +from __future__ import annotations + +from collections.abc import Iterator +from typing import Any + +from nsc.http.errors import NetBoxAPIError +from nsc.savedfilters.store import SAVED_FILTERS_PATH, NativeSavedFilterStore + + +class _StubResolver: + def __init__(self, object_type: str | None) -> None: + self._object_type = object_type + self.calls: list[str] = [] + + def resolve(self, list_path: str) -> str | None: + self.calls.append(list_path) + return self._object_type + + +class _FakeFallback: + def __init__(self) -> None: + self.store: dict[tuple[str, str], dict[str, dict[str, str]]] = {} + self.list_calls: list[tuple[str, str]] = [] + self.save_calls: list[tuple[str, str, str, dict[str, str]]] = [] + self.delete_calls: list[tuple[str, str, str]] = [] + + def list(self, tag: str, resource: str) -> dict[str, dict[str, str]]: + self.list_calls.append((tag, resource)) + return self.store.get((tag, resource), {}) + + def save(self, tag: str, resource: str, name: str, params: dict[str, str]) -> None: + self.save_calls.append((tag, resource, name, params)) + self.store.setdefault((tag, resource), {})[name] = params + + def delete(self, tag: str, resource: str, name: str) -> None: + self.delete_calls.append((tag, resource, name)) + self.store.get((tag, resource), {}).pop(name, None) + + +class _FakeClient: + def __init__(self, records: list[dict[str, Any]] | None = None) -> None: + self.records = records or [] + self.posts: list[tuple[str, Any]] = [] + self.patches: list[tuple[str, Any]] = [] + self.deletes: list[str] = [] + self.paginate_calls: list[tuple[str, dict[str, Any] | None]] = [] + self.fail = False + + def paginate( + self, path: str, params: dict[str, Any] | None = None, *, limit: int | None = None + ) -> Iterator[dict[str, Any]]: + self.paginate_calls.append((path, params)) + if self.fail: + raise NetBoxAPIError(status_code=500, url=path, body_snippet="", headers={}) + name = (params or {}).get("name") + for rec in self.records: + if name is None or rec.get("name") == name: + yield rec + + def post(self, path: str, *, json: Any = None, **kw: Any) -> dict[str, Any]: + self.posts.append((path, json)) + return {"id": 99, **(json or {})} + + def patch(self, path: str, *, json: Any = None, **kw: Any) -> dict[str, Any]: + self.patches.append((path, json)) + return {} + + def delete(self, path: str, **kw: Any) -> dict[str, Any]: + self.deletes.append(path) + return {} + + +def _store(client: _FakeClient, fallback: _FakeFallback, ot: str | None = "dcim.device", **kw: Any): + return NativeSavedFilterStore(client, fallback, resolver=_StubResolver(ot), **kw) + + +def test_list_reads_saved_filters_for_the_object_type() -> None: + client = _FakeClient( + [ + {"id": 1, "name": "active", "parameters": {"status": ["active"]}}, + {"id": 2, "name": "leafs", "parameters": {"role": ["leaf"]}}, + ] + ) + store = _store(client, _FakeFallback()) + result = store.list("/api/dcim/devices/", "dcim", "devices") + assert result == {"active": {"status": "active"}, "leafs": {"role": "leaf"}} + assert client.paginate_calls == [(SAVED_FILTERS_PATH, {"object_type": "dcim.device"})] + + +def test_save_creates_a_new_saved_filter_when_absent() -> None: + client = _FakeClient([]) + store = _store(client, _FakeFallback()) + store.save("/api/dcim/devices/", "dcim", "devices", "active", {"status": "active"}) + assert len(client.posts) == 1 + path, body = client.posts[0] + assert path == SAVED_FILTERS_PATH + assert body["name"] == "active" + assert body["object_types"] == ["dcim.device"] + assert body["parameters"] == {"status": ["active"]} + assert body["slug"] + assert body["shared"] is True + assert body["enabled"] is True + + +def test_save_patches_existing_filter_matched_by_name() -> None: + client = _FakeClient([{"id": 7, "name": "active", "parameters": {"status": ["offline"]}}]) + store = _store(client, _FakeFallback()) + store.save("/api/dcim/devices/", "dcim", "devices", "active", {"status": "active"}) + assert client.posts == [] + assert len(client.patches) == 1 + path, body = client.patches[0] + assert path == f"{SAVED_FILTERS_PATH}7/" + assert body["parameters"] == {"status": ["active"]} + # name/slug are not rewritten on update, so a web-UI-created slug survives. + assert "slug" not in body + + +def test_delete_removes_filter_matched_by_name() -> None: + client = _FakeClient([{"id": 7, "name": "active", "parameters": {}}]) + store = _store(client, _FakeFallback()) + store.delete("/api/dcim/devices/", "dcim", "devices", "active") + assert client.deletes == [f"{SAVED_FILTERS_PATH}7/"] + + +def test_delete_is_a_noop_when_name_not_found() -> None: + client = _FakeClient([]) + store = _store(client, _FakeFallback()) + store.delete("/api/dcim/devices/", "dcim", "devices", "ghost") + assert client.deletes == [] + + +def test_object_type_unresolved_uses_fallback_for_all_ops() -> None: + client = _FakeClient([]) + fallback = _FakeFallback() + store = _store(client, fallback, ot=None) + store.save("/api/x/y/", "x", "y", "n", {"a": "b"}) + store.list("/api/x/y/", "x", "y") + store.delete("/api/x/y/", "x", "y", "n") + assert fallback.save_calls and fallback.list_calls and fallback.delete_calls + assert client.posts == [] and client.deletes == [] + + +def test_api_failure_falls_back_and_reports() -> None: + client = _FakeClient([]) + client.fail = True + fallback = _FakeFallback() + errors: list[str] = [] + store = _store(client, fallback, on_error=errors.append) + result = store.list("/api/dcim/devices/", "dcim", "devices") + assert result == {} + assert fallback.list_calls == [("dcim", "devices")] + assert errors # the failure was surfaced, not swallowed silently + + +def test_save_api_failure_falls_back_to_local_write() -> None: + client = _FakeClient([]) + client.fail = True # the name-lookup paginate raises before any write + fallback = _FakeFallback() + errors: list[str] = [] + store = _store(client, fallback, on_error=errors.append) + store.save("/api/dcim/devices/", "dcim", "devices", "active", {"status": "active"}) + assert fallback.save_calls == [("dcim", "devices", "active", {"status": "active"})] + assert errors + + +def test_delete_api_failure_falls_back_to_local_delete() -> None: + client = _FakeClient([]) + client.fail = True # the name-lookup paginate raises before any delete + fallback = _FakeFallback() + errors: list[str] = [] + store = _store(client, fallback, on_error=errors.append) + store.delete("/api/dcim/devices/", "dcim", "devices", "active") + assert fallback.delete_calls == [("dcim", "devices", "active")] + assert errors + + +def test_list_warns_on_duplicate_names_and_keeps_first() -> None: + # NetBox permits same-named filters; list() keeps the first deterministically + # (matching _find's save/delete target) and surfaces the ambiguity. + client = _FakeClient( + [ + {"id": 1, "name": "dup", "parameters": {"status": ["active"]}}, + {"id": 2, "name": "dup", "parameters": {"status": ["offline"]}}, + ] + ) + errors: list[str] = [] + store = _store(client, _FakeFallback(), on_error=errors.append) + result = store.list("/api/dcim/devices/", "dcim", "devices") + assert result == {"dup": {"status": "active"}} + assert any("dup" in e for e in errors) diff --git a/tests/tui/test_filter_screen.py b/tests/tui/test_filter_screen.py index 6e410ba..9cd6a73 100644 --- a/tests/tui/test_filter_screen.py +++ b/tests/tui/test_filter_screen.py @@ -418,6 +418,7 @@ async def test_save_search_invokes_app_callback() -> None: prompt = app.screen prompt.dismiss("active-sw") await pilot.pause() + await app.workers.wait_for_complete() assert app.save_calls == [("dcim", "devices", "active-sw", {"status": "active"})] @@ -462,6 +463,8 @@ async def test_load_saved_search_repopulates_state_and_chips() -> None: assert isinstance(screen, FilterScreen) screen.action_load_search() await pilot.pause() + await app.workers.wait_for_complete() + await pilot.pause() picker = app.screen assert isinstance(picker, SavedSearchPicker) await pilot.press("enter") @@ -481,9 +484,12 @@ async def test_delete_saved_search_from_picker_invokes_app_callback() -> None: assert isinstance(screen, FilterScreen) screen.action_load_search() await pilot.pause() + await app.workers.wait_for_complete() + await pilot.pause() picker = app.screen assert isinstance(picker, SavedSearchPicker) # The delete key binding is the only way to remove a saved search. await pilot.press("d") await pilot.pause() + await app.workers.wait_for_complete() assert app.delete_calls == [("dcim", "devices", "active-sw")] diff --git a/tests/tui/test_saved_filter_integration.py b/tests/tui/test_saved_filter_integration.py new file mode 100644 index 0000000..1853f4a --- /dev/null +++ b/tests/tui/test_saved_filter_integration.py @@ -0,0 +1,98 @@ +"""`NscTuiApp` saved-search methods routed through the native SavedFilter store.""" + +from __future__ import annotations + +from collections.abc import Iterator +from typing import Any + +from nsc.config.models import Config +from nsc.config.saved_searches import ConfigFileSavedSearchStore +from nsc.model.command_model import CommandModel, Operation, Resource, Tag +from nsc.savedfilters.store import SAVED_FILTERS_PATH, NativeSavedFilterStore +from nsc.tui.app import NscTuiApp + +_OBJECT_TYPES = [ + {"app_label": "dcim", "model": "device", "rest_api_endpoint": "/api/dcim/devices/"}, +] + + +def _model() -> CommandModel: + devices = Resource( + name="devices", + list_op=Operation( + operation_id="dcim_devices_list", + http_method="GET", + path="/api/dcim/devices/", + ), + ) + return CommandModel( + info_title="t", + info_version="1", + schema_hash="h", + tags={"dcim": Tag(name="dcim", resources={"devices": devices})}, + ) + + +class _FakeClient: + def __init__(self, saved: list[dict[str, Any]] | None = None) -> None: + self._saved = saved or [] + self.posts: list[tuple[str, Any]] = [] + self.deletes: list[str] = [] + + def paginate( + self, path: str, params: dict[str, Any] | None = None, *, limit: int | None = None + ) -> Iterator[dict[str, Any]]: + if path == "/api/core/object-types/": + yield from _OBJECT_TYPES + return + if path == SAVED_FILTERS_PATH: + name = (params or {}).get("name") + for rec in self._saved: + if name is None or rec.get("name") == name: + yield rec + + def post(self, path: str, *, json: Any = None, **kw: Any) -> dict[str, Any]: + self.posts.append((path, json)) + return {"id": 1, **(json or {})} + + def delete(self, path: str, **kw: Any) -> dict[str, Any]: + self.deletes.append(path) + return {} + + +def _app(client: _FakeClient, config: Config | None = None) -> NscTuiApp: + fallback = ConfigFileSavedSearchStore(config or Config(), config_file=None) + store = NativeSavedFilterStore(client, fallback, on_error=lambda _m: None) + return NscTuiApp(_model(), client, saved_filter_store=store) + + +def test_saved_searches_for_reads_native_filters() -> None: + client = _FakeClient([{"id": 5, "name": "active", "parameters": {"status": ["active"]}}]) + app = _app(client) + assert app.saved_searches_for("dcim", "devices") == {"active": {"status": "active"}} + + +def test_save_search_creates_native_filter_with_object_type() -> None: + client = _FakeClient([]) + app = _app(client) + app.save_search("dcim", "devices", "active", {"status": "active"}) + assert len(client.posts) == 1 + _path, body = client.posts[0] + assert body["object_types"] == ["dcim.device"] + assert body["parameters"] == {"status": ["active"]} + + +def test_delete_search_removes_native_filter() -> None: + client = _FakeClient([{"id": 5, "name": "active", "parameters": {}}]) + app = _app(client) + app.delete_search("dcim", "devices", "active") + assert client.deletes == [f"{SAVED_FILTERS_PATH}5/"] + + +def test_app_wires_store_error_notifier() -> None: + # The app gives the store a way to surface API failures rather than letting + # them fail silently. + store = NativeSavedFilterStore(_FakeClient(), ConfigFileSavedSearchStore(Config())) + app = NscTuiApp(_model(), _FakeClient(), saved_filter_store=store) + assert store.on_error is not None + assert app is not None