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