diff --git a/CHANGELOG.md b/CHANGELOG.md index d35f632..bd37919 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,21 @@ All notable changes to netbox-super-cli are tracked here. Format follows [Keep a `X-Netbox-Branch` schema_id header, including under `audit_redaction: full`. Lines for requests on main are unchanged (no `branch` key). +### Fixed + +- **The TUI explains a branch that became unusable mid-session** ([#176]). If + the active branch is merged, archived or deleted while `nsc tui` is open, + NetBox Branching refuses later requests with a plain-text 400. List, detail, + edit, record-picker and bulk-edit errors now read + `branch '' is no longer usable (status: …)` and point back to main, + instead of `API 400 — Selected branch is not ready.`. Global search stops at + the first such failure instead of silently reporting no matches. Saved + searches that fall back to `config.yaml` (in the TUI or with `--saved`) now + say the branch is the reason, including when the object-type lookup is the + request that fails. Only the + Branching middleware's exact messages are matched, and only while a branch is + active, so other 400s such as validation errors look the same as before. + ## v1.9.0 — 2026-10-25 Minor release. Plugin endpoints get correct per-resource commands (**plugin command paths change**). Branch conflicts render as a table, and server data can no longer drive your terminal. @@ -241,6 +256,7 @@ notices, and a fix for `nsc login` reporting the wrong user. [#196]: https://github.com/thomaschristory/netbox-super-cli/issues/196 [#202]: https://github.com/thomaschristory/netbox-super-cli/issues/202 [#206]: https://github.com/thomaschristory/netbox-super-cli/issues/206 +[#176]: https://github.com/thomaschristory/netbox-super-cli/issues/176 [#171]: https://github.com/thomaschristory/netbox-super-cli/issues/171 [#179]: https://github.com/thomaschristory/netbox-super-cli/issues/179 [#173]: https://github.com/thomaschristory/netbox-super-cli/issues/173 diff --git a/docs/guides/branching.md b/docs/guides/branching.md index dd68145..0e70b82 100644 --- a/docs/guides/branching.md +++ b/docs/guides/branching.md @@ -57,7 +57,9 @@ When a branch is active, every command prints one line to **stderr**: ``` stdout is untouched, so `-o json` output piped elsewhere stays clean. The TUI -shows the branch in its header in the warning colour. `nsc branch current` +shows the branch in its header in the warning colour, and says so if the +branch is merged or deleted while it is open (see +[Interactive TUI](interactive-tui.md#working-in-a-branch)). `nsc branch current` prints the active branch and where it came from (`flag`, `env` or `profile`), or `main`. diff --git a/docs/guides/interactive-tui.md b/docs/guides/interactive-tui.md index 0790d70..033bf5e 100644 --- a/docs/guides/interactive-tui.md +++ b/docs/guides/interactive-tui.md @@ -107,7 +107,8 @@ 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. +`~/.nsc/config.yaml` and a toast tells you the fallback was used. If the active +branch has been merged or deleted, the toast names the branch as the cause. ## Viewing and editing a record @@ -201,7 +202,18 @@ profile's sticky branch), the header reads `nsc ⎇ ` on the warning colour, and every read and edit happens inside that branch. The TUI checks the branch before it opens. If the branch is merged, missing or not ready, `nsc tui` exits with the same error as any other command instead of showing -main's data. See [NetBox Branching](branching.md). +main's data. + +If the branch is merged, archived or deleted while the TUI is open, NetBox +refuses every later request. Instead of a bare `API 400`, the TUI then says +`branch '' is no longer usable (status: …)` and suggests switching back +to main. Global search stops after the first such failure rather than trying +every resource, and a bulk edit lists the same message for each failed record. +Saving, loading or deleting a saved search still falls back to +`~/.nsc/config.yaml`, and the warning says the branch is the reason. +Quit and relaunch on another branch (or main) to carry on. Other 400s, such +as a bad filter value, still show NetBox's own message. See +[NetBox Branching](branching.md). ## Loading and errors diff --git a/nsc/branching/errors.py b/nsc/branching/errors.py index 123e0da..3aa3f4b 100644 --- a/nsc/branching/errors.py +++ b/nsc/branching/errors.py @@ -48,6 +48,22 @@ def __init__(self, name: str, status: str) -> None: self.status = status +class BranchUnusableError(BranchError): + """The server refused a request because the active branch went away mid-session. + + `status` is the branch status when the server names it, else None. + """ + + def __init__(self, name: str, *, status: str | None, missing: bool = False) -> None: + state = "deleted or not found" if missing else f"status: {status or 'not ready'}" + super().__init__( + f"branch {name!r} is no longer usable ({state}); {_ESCAPE_HINT}", + branch=name, + ) + self.status = status + self.missing = missing + + class BranchConflictError(BranchError): def __init__(self, name: str, conflicts: list[dict[str, Any]]) -> None: what = ( diff --git a/nsc/branching/unusable.py b/nsc/branching/unusable.py new file mode 100644 index 0000000..a788acd --- /dev/null +++ b/nsc/branching/unusable.py @@ -0,0 +1,43 @@ +"""Recognize the NetBox Branching middleware's "this branch is unusable" 400s. + +The middleware answers a request whose `X-NetBox-Branch` names a merged, +archived or deleted branch with a plain-text 400 (Django's +`HttpResponseBadRequest`), before any view runs. Only those exact bodies are +matched, so a validation 400 (JSON) is never mistaken for one. +""" + +from __future__ import annotations + +import re + +from nsc.branching.errors import BranchUnusableError +from nsc.http.errors import NetBoxAPIError + +_BAD_REQUEST = 400 +_MISSING = re.compile(r"Invalid branch identifier") +_NOT_READY = re.compile(r"Selected branch is not ready\.?") +# The plugin's own BranchNotReady text, which names the status; current +# releases replace it with the generic message above, but it is matched in case +# a release passes it through. +_NOT_READY_WITH_STATUS = re.compile( + r"Branch .+ is not ready for use \(status: (?P[\w-]+)\)" +) + + +def branch_unusable_error(exc: NetBoxAPIError, branch: str | None) -> BranchUnusableError | None: + """The branch-state error behind `exc`, or None if it is any other failure. + + `branch` is the active branch's name; without one no request carried the + branch header, so the middleware cannot have produced the error. + """ + if branch is None or exc.status_code != _BAD_REQUEST: + return None + body = (exc.body or "").strip() + if _MISSING.fullmatch(body): + return BranchUnusableError(branch, status=None, missing=True) + if _NOT_READY.fullmatch(body): + return BranchUnusableError(branch, status=None) + match = _NOT_READY_WITH_STATUS.fullmatch(body) + if match: + return BranchUnusableError(branch, status=match["status"]) + return None diff --git a/nsc/cli/registration.py b/nsc/cli/registration.py index 2485fc6..f2a75b9 100644 --- a/nsc/cli/registration.py +++ b/nsc/cli/registration.py @@ -153,6 +153,7 @@ def _resolve_saved_filters( ctx.client, ConfigFileSavedSearchStore(ctx.config), on_error=lambda message: typer.echo(f"Warning: {message}", err=True), + branch=ctx.branch.name if ctx.branch else None, ) available = store.list(list_path, tag_name, resource_name) saved = available.get(saved_name) diff --git a/nsc/savedfilters/objecttypes.py b/nsc/savedfilters/objecttypes.py index d369c27..7084677 100644 --- a/nsc/savedfilters/objecttypes.py +++ b/nsc/savedfilters/objecttypes.py @@ -11,6 +11,7 @@ from __future__ import annotations +from collections.abc import Callable from typing import Any, Protocol from urllib.parse import urlsplit @@ -73,9 +74,15 @@ def object_type_index(object_types: Any) -> dict[str, str]: class ObjectTypeResolver: """Looks up object types from the live registry, cached per app label.""" - def __init__(self, client: _ClientLike) -> None: + def __init__( + self, + client: _ClientLike, + *, + on_error: Callable[[NetBoxAPIError | NetBoxClientError], None] | None = None, + ) -> None: self._client = client self._cache: dict[str, dict[str, str]] = {} + self._on_error = on_error def resolve(self, list_path: str) -> str | None: """``app_label.model`` for a list path, or None if it can't be resolved. @@ -96,7 +103,9 @@ def _index_for(self, app_label: str) -> dict[str, str] | None: return self._cache[app_label] try: records = list(self._client.paginate(_OBJECT_TYPES_PATH, {"app_label": app_label})) - except (NetBoxAPIError, NetBoxClientError): + except (NetBoxAPIError, NetBoxClientError) as exc: + if self._on_error is not None: + self._on_error(exc) return None index = object_type_index(records) self._cache[app_label] = index diff --git a/nsc/savedfilters/store.py b/nsc/savedfilters/store.py index 52d80b3..c014336 100644 --- a/nsc/savedfilters/store.py +++ b/nsc/savedfilters/store.py @@ -12,6 +12,8 @@ from collections.abc import Callable from typing import Any, Protocol, cast +from nsc.branching.errors import BranchUnusableError +from nsc.branching.unusable import branch_unusable_error from nsc.http.errors import NetBoxAPIError, NetBoxClientError from nsc.savedfilters.objecttypes import ObjectTypeResolver from nsc.savedfilters.params import ( @@ -43,11 +45,13 @@ def __init__( *, resolver: _ResolverLike | None = None, on_error: Callable[[str], None] | None = None, + branch: str | None = None, ) -> None: self._client = client self._fallback = fallback - self._resolver = resolver or ObjectTypeResolver(client) + self._resolver = resolver or ObjectTypeResolver(client, on_error=self._lookup_failed) self.on_error = on_error + self.branch = branch def list(self, list_path: str, tag: str, resource: str) -> dict[str, dict[str, str]]: object_type = self._resolver.resolve(list_path) @@ -133,8 +137,24 @@ def _find(self, object_type: str, name: str) -> dict[str, Any] | None: return None def _report(self, action: str, exc: Exception) -> None: + unusable = self._branch_state(exc) + if unusable is not None: + self._notify(f"Could not {action} via NetBox: {unusable}; using local config.") + return self._notify(f"Could not {action} via NetBox ({exc}); using local config.") + def _lookup_failed(self, exc: NetBoxAPIError | NetBoxClientError) -> None: + # Other registry failures stay silent (a missing object type is routine); + # a dead branch would otherwise quietly divert every save to config.yaml. + unusable = self._branch_state(exc) + if unusable is not None: + self._notify(f"Could not use NetBox saved filters: {unusable}; using local config.") + + def _branch_state(self, exc: Exception) -> BranchUnusableError | None: + if isinstance(exc, NetBoxAPIError): + return branch_unusable_error(exc, self.branch) + return None + def _notify(self, message: str) -> None: if self.on_error is not None: self.on_error(message) diff --git a/nsc/tui/app.py b/nsc/tui/app.py index 0b8165d..2c2f5d1 100644 --- a/nsc/tui/app.py +++ b/nsc/tui/app.py @@ -42,6 +42,7 @@ def __init__( branch: str | None = None, ) -> None: super().__init__() + self.active_branch = branch if branch is not None: self.title = f"nsc ⎇ {branch}" self.add_class("-branch") @@ -57,6 +58,12 @@ def __init__( 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 + if ( + saved_filter_store is not None + and branch is not None + and getattr(saved_filter_store, "branch", ...) is None + ): + saved_filter_store.branch = branch from nsc.savedfilters.custom_fields import CustomFieldResolver # noqa: PLC0415 from nsc.savedfilters.tags import TagsResolver # noqa: PLC0415 diff --git a/nsc/tui/bulk.py b/nsc/tui/bulk.py index 4d6a213..6ac00ef 100644 --- a/nsc/tui/bulk.py +++ b/nsc/tui/bulk.py @@ -90,6 +90,8 @@ def apply_bulk( changes: list[RecordChange], patch_fn: Callable[[RecordChange], None], on_progress: Callable[[int, int], None] | None = None, + *, + describe: Callable[[NetBoxAPIError | NetBoxClientError], str] | None = None, ) -> BulkResult: """Apply ``patch_fn`` to each change, aggregating per-record outcomes. @@ -112,7 +114,8 @@ def apply_bulk( try: patch_fn(change) except (NetBoxAPIError, NetBoxClientError) as exc: - failures.append(BulkFailure(record_id=change.record_id, error=exc.render_for_cli())) + error = describe(exc) if describe is not None else exc.render_for_cli() + failures.append(BulkFailure(record_id=change.record_id, error=error)) else: successes.append(change.record_id) return BulkResult(successes=successes, failures=failures, skipped=skipped) diff --git a/nsc/tui/errors.py b/nsc/tui/errors.py index 112cfe4..7a8e14f 100644 --- a/nsc/tui/errors.py +++ b/nsc/tui/errors.py @@ -9,17 +9,44 @@ import json +from nsc.branching.errors import BranchUnusableError +from nsc.branching.unusable import branch_unusable_error from nsc.http.errors import NetBoxAPIError, NetBoxClientError from nsc.output.sanitize import safe_markup -def api_error_message(exc: Exception) -> str: +def active_branch(app: object) -> str | None: + """The branch the TUI session runs on (`NscTuiApp.active_branch`), if any.""" + branch = getattr(app, "active_branch", None) + return branch if isinstance(branch, str) else None + + +def api_error_message(exc: Exception, *, branch: str | None = None) -> str: """The message as Textual markup: `notify` parses markup, and the text is server data, so it is passed through `safe_markup` (#206).""" - return safe_markup(_message(exc)) + return safe_markup(_message(exc, branch)) + + +def api_error_text(exc: Exception, *, branch: str | None = None) -> str: + """Plain (unescaped) one-line text, for callers that sanitize it themselves.""" + unusable = branch_state_error(exc, branch) + if unusable is not None: + return str(unusable) + if isinstance(exc, NetBoxAPIError | NetBoxClientError): + return exc.render_for_cli() + return str(exc) + + +def branch_state_error(exc: Exception, branch: str | None) -> BranchUnusableError | None: + if isinstance(exc, NetBoxAPIError): + return branch_unusable_error(exc, branch) + return None -def _message(exc: Exception) -> str: +def _message(exc: Exception, branch: str | None) -> str: + unusable = branch_state_error(exc, branch) + if unusable is not None: + return str(unusable) if isinstance(exc, NetBoxAPIError): detail = _format_body(exc.body_snippet) return f"API {exc.status_code} — {detail}" if detail else f"API {exc.status_code}" diff --git a/nsc/tui/screens/bulk_edit_form.py b/nsc/tui/screens/bulk_edit_form.py index 2e4c5c0..7afc971 100644 --- a/nsc/tui/screens/bulk_edit_form.py +++ b/nsc/tui/screens/bulk_edit_form.py @@ -35,6 +35,7 @@ from nsc.savedfilters.tags import TagDef from nsc.tui._bindings import textual_bindings from nsc.tui.bulk import RecordChange, shared_values +from nsc.tui.errors import active_branch, api_error_text from nsc.tui.fk import is_fk_value, resolve_fk_target from nsc.tui.forms import ( SET_NULL, @@ -408,7 +409,14 @@ def _on_progress(index: int, total: int) -> None: # apply_bulk runs in a worker thread; bump the bar on the UI thread. self.app.call_from_thread(_advance, index, total) - result = await asyncio.to_thread(apply_bulk, changes, _patch, _on_progress) + branch = active_branch(self.app) + result = await asyncio.to_thread( + apply_bulk, + changes, + _patch, + _on_progress, + describe=lambda exc: api_error_text(exc, branch=branch), + ) self.app.push_screen(BulkSummaryModal(result), self._on_summary_dismissed) def _on_summary_dismissed(self, _: None) -> None: diff --git a/nsc/tui/screens/detail.py b/nsc/tui/screens/detail.py index 5cf01b7..2304393 100644 --- a/nsc/tui/screens/detail.py +++ b/nsc/tui/screens/detail.py @@ -24,7 +24,7 @@ from nsc.output.flatten import flatten from nsc.output.sanitize import safe_text from nsc.tui._bindings import textual_bindings -from nsc.tui.errors import api_error_message +from nsc.tui.errors import active_branch, api_error_message from nsc.tui.fk import is_fk_value, resolve_fk_target from nsc.tui.forms import SET_NULL, WidgetSpec, compute_patch, diff_rows, field_to_widget from nsc.tui.relations import RelatedView, related_views @@ -305,7 +305,9 @@ def _apply_patch(self, patch: dict[str, Any]) -> None: sensitive_paths=self._sensitive, ) except (NetBoxAPIError, NetBoxClientError) as exc: - self.notify(api_error_message(exc), severity="error", timeout=8) + self.notify( + api_error_message(exc, branch=active_branch(self.app)), severity="error", timeout=8 + ) return self._record.update(patch) self.staged.clear() @@ -388,7 +390,11 @@ def _on_confirm(confirmed: bool | None) -> None: try: self._client.delete(path, operation_id=delete_op.operation_id) except (NetBoxAPIError, NetBoxClientError) as exc: - self.notify(api_error_message(exc), severity="error", timeout=8) + self.notify( + api_error_message(exc, branch=active_branch(self.app)), + severity="error", + timeout=8, + ) return self.app.pop_screen() self._reload_underlying_list() diff --git a/nsc/tui/screens/edit_form.py b/nsc/tui/screens/edit_form.py index 6586adf..3fe8639 100644 --- a/nsc/tui/screens/edit_form.py +++ b/nsc/tui/screens/edit_form.py @@ -22,7 +22,7 @@ from nsc.savedfilters.custom_fields import CustomFieldDef from nsc.savedfilters.tags import TagDef from nsc.tui._bindings import textual_bindings -from nsc.tui.errors import api_error_message +from nsc.tui.errors import active_branch, api_error_message from nsc.tui.fk import is_fk_value, resolve_fk_target from nsc.tui.forms import ( SET_NULL, @@ -329,7 +329,9 @@ def _apply_patch(self, patch: dict[str, Any], sensitive_paths: tuple[str, ...]) sensitive_paths=sensitive_paths, ) except (NetBoxAPIError, NetBoxClientError) as exc: - self.notify(api_error_message(exc), severity="error", timeout=8) + self.notify( + api_error_message(exc, branch=active_branch(self.app)), severity="error", timeout=8 + ) return self.dismiss() diff --git a/nsc/tui/screens/global_search.py b/nsc/tui/screens/global_search.py index faa9e54..d9e0d6a 100644 --- a/nsc/tui/screens/global_search.py +++ b/nsc/tui/screens/global_search.py @@ -23,6 +23,7 @@ from nsc.model.command_model import CommandModel from nsc.output.sanitize import safe_text from nsc.tui.catalog import ResourceRef +from nsc.tui.errors import active_branch, api_error_message, branch_state_error from nsc.tui.search import global_search_targets, search_target from nsc.tui.widgets.nav_tree import NavTree @@ -104,7 +105,15 @@ async def _run_search(self, term: str) -> None: rows = await asyncio.to_thread( search_target, self._client, ref, term, _PER_TYPE_LIMIT ) - except (NetBoxAPIError, NetBoxClientError): + except (NetBoxAPIError, NetBoxClientError) as exc: + # Every other target would fail the same way on a dead branch. + if branch_state_error(exc, active_branch(self.app)) is not None: + self.notify( + api_error_message(exc, branch=active_branch(self.app)), + severity="error", + timeout=8, + ) + break continue # one bad endpoint must not abort the whole search if rows: self._add_group(ref, rows) diff --git a/nsc/tui/screens/list.py b/nsc/tui/screens/list.py index aa2d2e2..23d7ca7 100644 --- a/nsc/tui/screens/list.py +++ b/nsc/tui/screens/list.py @@ -15,7 +15,7 @@ from nsc.http.errors import NetBoxAPIError, NetBoxClientError from nsc.model.command_model import CommandModel, Operation from nsc.tui._bindings import textual_bindings -from nsc.tui.errors import api_error_message +from nsc.tui.errors import active_branch, api_error_message from nsc.tui.nav import can_go_back from nsc.tui.selection import Selection from nsc.tui.view import build_rows, choose_columns, count_label @@ -162,7 +162,9 @@ def _first_chunk() -> tuple[Any, list[dict[str, Any]]]: try: pager, records = await asyncio.to_thread(_first_chunk) except (NetBoxAPIError, NetBoxClientError) as exc: - self.notify(api_error_message(exc), severity="error", timeout=8) + self.notify( + api_error_message(exc, branch=active_branch(self.app)), severity="error", timeout=8 + ) table.loading = False self._loading = False return @@ -198,7 +200,11 @@ async def _load_more(self) -> None: self._has_more = _more_remaining(pager, got=len(records), wanted=want) else: self._has_more = getattr(pager, "has_more", True) is not False - self.notify(api_error_message(error), severity="error", timeout=8) + self.notify( + api_error_message(error, branch=active_branch(self.app)), + severity="error", + timeout=8, + ) if not self._has_more: self._prune_selection(self._records) self._update_count() diff --git a/nsc/tui/screens/record_picker.py b/nsc/tui/screens/record_picker.py index a5f19ee..f4c38ac 100644 --- a/nsc/tui/screens/record_picker.py +++ b/nsc/tui/screens/record_picker.py @@ -15,7 +15,7 @@ from nsc.http.errors import NetBoxAPIError, NetBoxClientError from nsc.model.command_model import Operation from nsc.output.sanitize import safe_text -from nsc.tui.errors import api_error_message +from nsc.tui.errors import active_branch, api_error_message _PAGE_LIMIT = 50 _DEBOUNCE = 0.25 @@ -76,7 +76,9 @@ async def _query(self, search: str) -> None: lambda: list(self._client.paginate(self._op.path, params, limit=_PAGE_LIMIT)) ) except (NetBoxAPIError, NetBoxClientError) as exc: - self.notify(api_error_message(exc), severity="error", timeout=8) + self.notify( + api_error_message(exc, branch=active_branch(self.app)), severity="error", timeout=8 + ) await self._populate([]) listing.loading = False return diff --git a/tests/branching/test_unusable.py b/tests/branching/test_unusable.py new file mode 100644 index 0000000..35eb889 --- /dev/null +++ b/tests/branching/test_unusable.py @@ -0,0 +1,66 @@ +from __future__ import annotations + +import pytest + +from nsc.branching.errors import BranchUnusableError +from nsc.branching.unusable import branch_unusable_error +from nsc.http.errors import NetBoxAPIError + + +def _err(status_code: int, body: str) -> NetBoxAPIError: + return NetBoxAPIError( + status_code=status_code, + url="https://nb/api/dcim/devices/", + body_snippet=body, + headers={"Content-Type": "text/html; charset=utf-8"}, + ) + + +def test_status_from_not_ready_for_use_message() -> None: + exc = _err(400, "Branch my-change is not ready for use (status: merged)") + err = branch_unusable_error(exc, "my-change") + assert isinstance(err, BranchUnusableError) + assert err.branch == "my-change" + assert err.status == "merged" + assert "branch 'my-change' is no longer usable (status: merged)" in str(err) + + +def test_generic_not_ready_message_has_no_status() -> None: + err = branch_unusable_error(_err(400, "Selected branch is not ready."), "my-change") + assert err is not None + assert err.status is None + assert "branch 'my-change' is no longer usable (status: not ready)" in str(err) + + +def test_invalid_identifier_means_branch_is_gone() -> None: + err = branch_unusable_error(_err(400, "Invalid branch identifier\n"), "my-change") + assert err is not None + assert err.status is None + assert "no longer usable (deleted or not found)" in str(err) + + +def test_message_points_back_to_main() -> None: + err = branch_unusable_error(_err(400, "Selected branch is not ready."), "b") + assert err is not None + assert "--branch main" in str(err) + + +@pytest.mark.parametrize( + "body", + [ + '{"name": ["This field is required."]}', + '{"detail": "Invalid branch identifier"}', + "Invalid branch identifier for site", + "", + ], +) +def test_other_400s_are_not_branch_state_errors(body: str) -> None: + assert branch_unusable_error(_err(400, body), "my-change") is None + + +def test_non_400_is_not_a_branch_state_error() -> None: + assert branch_unusable_error(_err(404, "Invalid branch identifier"), "b") is None + + +def test_no_active_branch_never_classifies() -> None: + assert branch_unusable_error(_err(400, "Selected branch is not ready."), None) is None diff --git a/tests/savedfilters/test_objecttypes.py b/tests/savedfilters/test_objecttypes.py index 6e98fda..0665484 100644 --- a/tests/savedfilters/test_objecttypes.py +++ b/tests/savedfilters/test_objecttypes.py @@ -110,6 +110,15 @@ def test_resolver_returns_none_on_api_error() -> None: assert resolver.resolve("/api/dcim/devices/") is None +def test_resolver_hands_registry_failures_to_on_error() -> None: + client = _FakeClient(_OBJECT_TYPES, fail=True) + seen: list[Exception] = [] + resolver = ObjectTypeResolver(client, on_error=seen.append) + assert resolver.resolve("/api/dcim/devices/") is None + assert len(seen) == 1 + assert isinstance(seen[0], NetBoxAPIError) + + def test_resolver_returns_none_for_unknown_endpoint() -> None: client = _FakeClient(_OBJECT_TYPES) resolver = ObjectTypeResolver(client) diff --git a/tests/savedfilters/test_store.py b/tests/savedfilters/test_store.py index c5f818f..9c3d21f 100644 --- a/tests/savedfilters/test_store.py +++ b/tests/savedfilters/test_store.py @@ -174,6 +174,88 @@ def test_delete_api_failure_falls_back_to_local_delete() -> None: assert errors +_NOT_READY = "Selected branch is not ready." + + +class _DeadBranchClient(_FakeClient): + """Every request answered with the branching middleware's plain-text 400.""" + + 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)) + raise NetBoxAPIError(status_code=400, url=path, body_snippet=_NOT_READY, headers={}) + + +def _assert_names_branch(errors: list[str]) -> None: + assert len(errors) == 1, errors + assert "branch 'b1' is no longer usable" in errors[0] + assert "NetBox API 400" not in errors[0] + assert "using local config" in errors[0] + + +def test_list_on_unusable_branch_names_the_branch() -> None: + fallback = _FakeFallback() + errors: list[str] = [] + store = _store(_DeadBranchClient(), fallback, on_error=errors.append, branch="b1") + store.list("/api/dcim/devices/", "dcim", "devices") + assert fallback.list_calls == [("dcim", "devices")] + _assert_names_branch(errors) + + +def test_save_on_unusable_branch_names_the_branch() -> None: + fallback = _FakeFallback() + errors: list[str] = [] + store = _store(_DeadBranchClient(), fallback, on_error=errors.append, branch="b1") + store.save("/api/dcim/devices/", "dcim", "devices", "x", {"status": "active"}) + assert fallback.save_calls == [("dcim", "devices", "x", {"status": "active"})] + _assert_names_branch(errors) + + +def test_delete_on_unusable_branch_names_the_branch() -> None: + fallback = _FakeFallback() + errors: list[str] = [] + store = _store(_DeadBranchClient(), fallback, on_error=errors.append, branch="b1") + store.delete("/api/dcim/devices/", "dcim", "devices", "x") + assert fallback.delete_calls == [("dcim", "devices", "x")] + _assert_names_branch(errors) + + +def test_unusable_branch_during_object_type_lookup_is_reported() -> None: + fallback = _FakeFallback() + errors: list[str] = [] + store = NativeSavedFilterStore( + _DeadBranchClient(), fallback, on_error=errors.append, branch="b1" + ) + store.save("/api/dcim/devices/", "dcim", "devices", "x", {"status": "active"}) + assert fallback.save_calls == [("dcim", "devices", "x", {"status": "active"})] + _assert_names_branch(errors) + + +def test_object_type_lookup_failure_off_branch_stays_quiet() -> None: + errors: list[str] = [] + store = NativeSavedFilterStore(_DeadBranchClient(), _FakeFallback(), on_error=errors.append) + store.list("/api/dcim/devices/", "dcim", "devices") + assert errors == [] + + +def test_validation_400_on_a_branch_keeps_the_generic_message() -> None: + class _BadRequestClient(_FakeClient): + def paginate( + self, path: str, params: dict[str, Any] | None = None, *, limit: int | None = None + ) -> Iterator[dict[str, Any]]: + raise NetBoxAPIError( + status_code=400, url=path, body_snippet='{"name": ["bad"]}', headers={} + ) + + errors: list[str] = [] + store = _store(_BadRequestClient(), _FakeFallback(), on_error=errors.append, branch="b1") + store.list("/api/dcim/devices/", "dcim", "devices") + assert len(errors) == 1 + assert "no longer usable" not in errors[0] + assert "NetBox API 400" in errors[0] + + 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. diff --git a/tests/tui/test_branch_unusable.py b/tests/tui/test_branch_unusable.py new file mode 100644 index 0000000..73455cb --- /dev/null +++ b/tests/tui/test_branch_unusable.py @@ -0,0 +1,294 @@ +"""A branch merged/deleted mid-session is reported as such, not as a bare 400 (#176).""" + +from __future__ import annotations + +from typing import Any + +import pytest +from textual.app import App, ComposeResult +from textual.widgets import Input, Static + +from nsc.http.errors import NetBoxAPIError +from nsc.tui.app import NscTuiApp +from nsc.tui.bulk import RecordChange, apply_bulk +from nsc.tui.screens.bulk_edit_form import BulkEditForm +from nsc.tui.screens.detail import DetailScreen +from nsc.tui.screens.edit_form import EditForm +from nsc.tui.screens.global_search import GlobalSearchScreen +from nsc.tui.screens.list import ListScreen +from nsc.tui.screens.record_picker import RecordPicker +from nsc.tui.widgets.bulk_summary import BulkSummaryModal +from tests.tui.test_app import _model +from tests.tui.test_bulk_edit_form import _BulkApp3, _selected_all_active, _stage_status_offline +from tests.tui.test_bulk_edit_form import _SpyClient as _BulkSpyClient +from tests.tui.test_detail_screen import _DetailApp +from tests.tui.test_detail_screen import _SpyClient as _DetailSpyClient +from tests.tui.test_edit_form import _EditApp +from tests.tui.test_edit_form import _SpyClient as _EditSpyClient +from tests.tui.test_global_search import _model as _search_model +from tests.tui.test_list_screen import _GatedPagedClient, _ListApp, _settle +from tests.tui.test_record_picker import _FakeClient as _PickerFakeClient +from tests.tui.test_record_picker import _PickerApp + +_MERGED = "Branch b1 is not ready for use (status: merged)" + + +class _RaisingClient: + def __init__(self, body: str) -> None: + self.body = body + self.calls: list[str] = [] + + def paginate( + self, path: str, params: dict[str, Any] | None = None, *, limit: int | None = None + ) -> Any: + self.calls.append(path) + raise NetBoxAPIError(status_code=400, url=path, body_snippet=self.body, headers={}) + + +def _messages(app: App[Any]) -> list[str]: + return [n.message for n in app._notifications] + + +@pytest.mark.asyncio +async def test_list_load_explains_unusable_branch() -> None: + app = NscTuiApp(_model(), _RaisingClient(_MERGED), initial_resource="devices", branch="b1") + async with app.run_test() as pilot: + await pilot.pause() + await app.workers.wait_for_complete() + await pilot.pause() + messages = _messages(app) + assert any("branch 'b1' is no longer usable (status: merged)" in m for m in messages) + assert not any(m.startswith("API 400") for m in messages) + + +@pytest.mark.asyncio +async def test_list_load_keeps_validation_400_unchanged_on_a_branch() -> None: + client = _RaisingClient('{"name": ["Bad filter."]}') + app = NscTuiApp(_model(), client, initial_resource="devices", branch="b1") + async with app.run_test() as pilot: + await pilot.pause() + await app.workers.wait_for_complete() + await pilot.pause() + assert "API 400 — name: Bad filter." in _messages(app) + + +@pytest.mark.asyncio +async def test_list_load_without_branch_renders_raw_body() -> None: + app = NscTuiApp( + _model(), _RaisingClient("Invalid branch identifier"), initial_resource="devices" + ) + async with app.run_test() as pilot: + await pilot.pause() + await app.workers.wait_for_complete() + await pilot.pause() + assert "API 400 — Invalid branch identifier" in _messages(app) + + +class _SearchApp(App[None]): + active_branch = "b1" + + def __init__(self, client: _RaisingClient) -> None: + super().__init__() + self.client = client + + def compose(self) -> ComposeResult: + yield Static("") + + async def on_mount(self) -> None: + await self.push_screen(GlobalSearchScreen(_search_model(), self.client)) + + +@pytest.mark.asyncio +async def test_global_search_stops_and_explains_unusable_branch() -> None: + client = _RaisingClient("Invalid branch identifier") + app = _SearchApp(client) + async with app.run_test() as pilot: + await pilot.pause() + screen = app.screen + assert isinstance(screen, GlobalSearchScreen) + await screen._run_search("sw") + await pilot.pause() + assert len(client.calls) == 1 + assert any( + "branch 'b1' is no longer usable (deleted or not found)" in m for m in _messages(app) + ) + + +def test_apply_bulk_uses_describe_for_failures() -> None: + def _patch(change: RecordChange) -> None: + raise NetBoxAPIError(status_code=400, url="u", body_snippet=_MERGED, headers={}) + + result = apply_bulk( + [RecordChange(record_id=1, patch={"status": "active"}, rows=[])], + _patch, + describe=lambda exc: f"described: {type(exc).__name__}", + ) + assert [f.error for f in result.failures] == ["described: NetBoxAPIError"] + + +def _merged_error(path: str) -> NetBoxAPIError: + return NetBoxAPIError(status_code=400, url=path, body_snippet=_MERGED, headers={}) + + +def _assert_explains_merged(messages: list[str]) -> None: + assert any("branch 'b1' is no longer usable (status: merged)" in m for m in messages), messages + assert not any("API 400" in m for m in messages), messages + + +class _DeadBranchDetailClient(_DetailSpyClient): + def patch(self, path: str, **kwargs: Any) -> dict[str, Any]: + self.patch_calls.append({"path": path}) + raise _merged_error(path) + + def delete(self, path: str, **kwargs: Any) -> Any: + self.delete_calls.append({"path": path}) + raise _merged_error(path) + + +class _BranchDetailApp(_DetailApp): + active_branch = "b1" + + def __init__(self) -> None: + super().__init__() + self.client = _DeadBranchDetailClient() + + +@pytest.mark.asyncio +async def test_detail_save_explains_unusable_branch() -> None: + app = _BranchDetailApp() + async with app.run_test() as pilot: + await pilot.pause() + screen = app.screen + assert isinstance(screen, DetailScreen) + screen.staged = {"name": "sw1-renamed"} + screen.action_save_all() + await pilot.pause() + await pilot.press("y") + await pilot.pause() + assert app.client.patch_calls + _assert_explains_merged(_messages(app)) + + +@pytest.mark.asyncio +async def test_detail_delete_explains_unusable_branch() -> None: + app = _BranchDetailApp() + async with app.run_test() as pilot: + await pilot.pause() + await pilot.press("d") + await pilot.pause() + await pilot.press("y") + await pilot.pause() + assert app.client.delete_calls + _assert_explains_merged(_messages(app)) + + +class _DeadBranchEditClient(_EditSpyClient): + def patch(self, path: str, **kwargs: Any) -> dict[str, Any]: + self.patch_calls.append({"path": path}) + raise _merged_error(path) + + +class _BranchEditApp(_EditApp): + active_branch = "b1" + + +@pytest.mark.asyncio +async def test_edit_form_save_explains_unusable_branch() -> None: + client = _DeadBranchEditClient([]) + app = _BranchEditApp(client) + async with app.run_test(size=(120, 60)) as pilot: + await pilot.pause() + screen = app.screen + assert isinstance(screen, EditForm) + screen.query_one("#field-name", Input).value = "sw2" + await pilot.pause() + screen.action_save() + await pilot.pause() + await pilot.press("y") + await pilot.pause() + assert client.patch_calls + _assert_explains_merged(_messages(app)) + + +class _DeadBranchBulkClient(_BulkSpyClient): + def patch(self, path: str, **kwargs: Any) -> dict[str, Any]: + self.patch_calls.append({"path": path}) + raise _merged_error(path) + + +class _BranchBulkApp(_BulkApp3): + active_branch = "b1" + + +@pytest.mark.asyncio +async def test_bulk_edit_summary_explains_unusable_branch() -> None: + client = _DeadBranchBulkClient([]) + app = _BranchBulkApp(client, _selected_all_active()) + async with app.run_test(size=(120, 60)) as pilot: + await pilot.pause() + screen = app.screen + assert isinstance(screen, BulkEditForm) + await _stage_status_offline(pilot, screen) + screen.action_preview() + await pilot.pause() + await pilot.press("enter") + await app.workers.wait_for_complete() + await pilot.pause() + summary = app.screen + assert isinstance(summary, BulkSummaryModal) + text = summary.render_text() + assert "3 failed" in text + assert "branch 'b1' is no longer usable (status: merged)" in text + assert "API 400" not in text + + +class _DeadBranchPickerClient(_PickerFakeClient): + def paginate( + self, path: str, params: dict[str, Any] | None = None, *, limit: int | None = None + ) -> Any: + self.calls.append((path, params)) + raise _merged_error(path) + + +class _BranchPickerApp(_PickerApp): + active_branch = "b1" + + +@pytest.mark.asyncio +async def test_record_picker_load_explains_unusable_branch() -> None: + client = _DeadBranchPickerClient([]) + app = _BranchPickerApp(client) + async with app.run_test() as pilot: + await pilot.pause() + screen = app.screen + assert isinstance(screen, RecordPicker) + await screen._query("anything") + await pilot.pause() + _assert_explains_merged(_messages(app)) + + +class _DeadBranchPagedClient(_GatedPagedClient): + dead_from = 10**9 + + def _fetch(self, path: str, params: dict[str, Any] | None) -> dict[str, Any]: + if int((params or {}).get("offset", 0)) >= self.dead_from: + raise _merged_error(path) + return super()._fetch(path, params) + + +class _BranchListApp(_ListApp): + active_branch = "b1" + + +@pytest.mark.asyncio +async def test_list_load_more_explains_unusable_branch() -> None: + client = _DeadBranchPagedClient(300) + app = _BranchListApp(client) + async with app.run_test() as pilot: + await _settle(app, pilot) + screen = app.screen + assert isinstance(screen, ListScreen) + client.dead_from = 100 + screen.action_load_more() + await _settle(app, pilot) + _assert_explains_merged(_messages(app)) diff --git a/tests/tui/test_errors.py b/tests/tui/test_errors.py index acf2a01..4d311fc 100644 --- a/tests/tui/test_errors.py +++ b/tests/tui/test_errors.py @@ -1,7 +1,7 @@ from __future__ import annotations from nsc.http.errors import NetBoxAPIError, NetBoxClientError -from nsc.tui.errors import api_error_message +from nsc.tui.errors import api_error_message, api_error_text def test_api_error_formats_netbox_field_validation_body() -> None: @@ -27,3 +27,39 @@ def test_api_error_with_empty_body_shows_only_status() -> None: def test_client_error_reports_unreachable() -> None: exc = NetBoxClientError(url="https://nb/api/x/", cause=OSError("name resolution failed")) assert api_error_message(exc) == "Could not reach NetBox: name resolution failed" + + +def _branch_400(body: str) -> NetBoxAPIError: + return NetBoxAPIError(status_code=400, url="https://nb/api/x/", body_snippet=body, headers={}) + + +def test_branch_state_error_is_explained_when_a_branch_is_active() -> None: + exc = _branch_400("Branch b1 is not ready for use (status: archived)") + message = api_error_message(exc, branch="b1") + assert "branch 'b1' is no longer usable (status: archived)" in message + assert not message.startswith("API 400") + + +def test_branch_state_body_without_active_branch_renders_unchanged() -> None: + exc = _branch_400("Invalid branch identifier") + assert api_error_message(exc) == "API 400 — Invalid branch identifier" + + +def test_validation_400_with_active_branch_renders_unchanged() -> None: + exc = _branch_400('{"name": ["This field is required."]}') + assert api_error_message(exc, branch="b1") == "API 400 — name: This field is required." + + +def test_branch_state_message_sanitizes_branch_name() -> None: + message = api_error_message(_branch_400("Invalid branch identifier"), branch="[bold]x\x1b]0;t") + assert "\x1b" not in message + assert "\\[bold]" in message + + +def test_api_error_text_is_plain_not_markup() -> None: + exc = _branch_400("Invalid branch identifier") + assert api_error_text(exc, branch="[b]").startswith("branch '[b]' is no longer usable") + other = NetBoxAPIError( + status_code=500, url="https://nb/api/x/", body_snippet="boom", headers={} + ) + assert api_error_text(other, branch="b") == "NetBox API 500 on https://nb/api/x/: boom" diff --git a/tests/tui/test_saved_filter_integration.py b/tests/tui/test_saved_filter_integration.py index 1853f4a..fc2f822 100644 --- a/tests/tui/test_saved_filter_integration.py +++ b/tests/tui/test_saved_filter_integration.py @@ -96,3 +96,17 @@ def test_app_wires_store_error_notifier() -> None: app = NscTuiApp(_model(), _FakeClient(), saved_filter_store=store) assert store.on_error is not None assert app is not None + + +def test_app_hands_its_branch_to_the_store() -> None: + store = NativeSavedFilterStore(_FakeClient(), ConfigFileSavedSearchStore(Config())) + NscTuiApp(_model(), _FakeClient(), saved_filter_store=store, branch="b1") + assert store.branch == "b1" + + +def test_app_keeps_a_branch_the_store_already_has() -> None: + store = NativeSavedFilterStore( + _FakeClient(), ConfigFileSavedSearchStore(Config()), branch="other" + ) + NscTuiApp(_model(), _FakeClient(), saved_filter_store=store, branch="b1") + assert store.branch == "other"