diff --git a/nsc/cli/handlers.py b/nsc/cli/handlers.py index 0c59f55..16e882c 100644 --- a/nsc/cli/handlers.py +++ b/nsc/cli/handlers.py @@ -66,6 +66,33 @@ def _out(stream: TextIO | None) -> TextIO: return stream if stream is not None else sys.stdout +def _custom_field_header_labels( + ctx: RuntimeContext, operation: Operation, columns: list[str] | None +) -> dict[str, str] | None: + """Map ``custom_fields.`` columns to their NetBox labels, or None. + + Only fetches definitions when a custom-field column is actually selected, so + ordinary lists incur no extra request. Display-only: the raw key stays the + selector everywhere else. + """ + if not columns or not any(c.startswith("custom_fields.") for c in columns): + return None + from nsc.savedfilters.custom_fields import ( # noqa: PLC0415 + CustomFieldResolver, + custom_field_labels, + ) + + # Custom-field definitions are keyed by the collection (list) endpoint, so a + # detail path like /api/dcim/devices/{id}/ must reduce to /api/dcim/devices/. + list_path = operation.path + if list_path.endswith("{id}/"): + list_path = list_path[: -len("{id}/")] + defs = CustomFieldResolver(ctx.client).resolve(list_path) + if defs is None: + return None + return custom_field_labels(columns, defs) + + def parse_filters(raw: list[str]) -> dict[str, str]: out: dict[str, str] = {} for item in raw: @@ -97,10 +124,12 @@ def handle_list( page_size=ctx.page_size, ) ) + columns = ctx.resolve_columns(op_tag, op_resource, operation) render( rows, format=ctx.output_format, - columns=ctx.resolve_columns(op_tag, op_resource, operation), + columns=columns, + header_labels=_custom_field_header_labels(ctx, operation, columns), stream=_out(stream), compact=ctx.compact, color=ctx.color, @@ -124,10 +153,12 @@ def handle_get( try: params, path_vars = _split_params(operation, kwargs) obj = ctx.client.get(operation.path.format(**path_vars), params) + columns = ctx.resolve_columns(op_tag, op_resource, operation) render( obj, format=ctx.output_format, - columns=ctx.resolve_columns(op_tag, op_resource, operation), + columns=columns, + header_labels=_custom_field_header_labels(ctx, operation, columns), stream=_out(stream), compact=ctx.compact, color=ctx.color, @@ -611,10 +642,12 @@ def _render_response( if is_delete: _render_delete_ok(ctx, stream=stream) return + columns = ctx.resolve_columns(op_tag, op_resource, operation) render( response, format=ctx.output_format, - columns=ctx.resolve_columns(op_tag, op_resource, operation), + columns=columns, + header_labels=_custom_field_header_labels(ctx, operation, columns), stream=stream, compact=ctx.compact, color=ctx.color, diff --git a/nsc/output/csv_.py b/nsc/output/csv_.py index d6010ca..01dd542 100644 --- a/nsc/output/csv_.py +++ b/nsc/output/csv_.py @@ -14,6 +14,7 @@ def render( *, stream: TextIO = sys.stdout, columns: list[str] | None = None, + header_labels: dict[str, str] | None = None, ) -> None: records = [data] if isinstance(data, dict) else list(data) if not records: @@ -21,7 +22,12 @@ def render( flat_records = [flatten(r, columns=columns) for r in records] fieldnames = columns if columns is not None else _gather_fieldnames(flat_records) writer = csv.DictWriter(stream, fieldnames=fieldnames, extrasaction="ignore") - writer.writeheader() + if header_labels: + # DictWriter keys data rows by raw fieldname, so emit a relabeled header + # row manually (instead of writeheader) to keep header/data aligned. + csv.writer(stream).writerow([header_labels.get(f, f) for f in fieldnames]) + else: + writer.writeheader() for row in flat_records: writer.writerow({k: ("" if v is None else v) for k, v in row.items()}) diff --git a/nsc/output/render.py b/nsc/output/render.py index dc76b23..9f8f584 100644 --- a/nsc/output/render.py +++ b/nsc/output/render.py @@ -14,6 +14,7 @@ def render( *, format: OutputFormat, columns: list[str] | None = None, + header_labels: dict[str, str] | None = None, stream: TextIO = sys.stdout, compact: bool = False, color: bool = False, @@ -26,9 +27,16 @@ def render( elif format is OutputFormat.YAML: yaml_.render(data, stream=stream) elif format is OutputFormat.CSV: - csv_.render(data, stream=stream, columns=columns) + csv_.render(data, stream=stream, columns=columns, header_labels=header_labels) elif format is OutputFormat.TABLE: - table.render(data, stream=stream, columns=columns, color=color, object_colors=object_colors) + table.render( + data, + stream=stream, + columns=columns, + color=color, + object_colors=object_colors, + header_labels=header_labels, + ) else: # pragma: no cover (StrEnum exhaustively covered above) raise ValueError(f"unknown output format: {format!r}") diff --git a/nsc/output/table.py b/nsc/output/table.py index af94909..0fcdf59 100644 --- a/nsc/output/table.py +++ b/nsc/output/table.py @@ -36,6 +36,7 @@ def render( columns: list[str] | None = None, color: bool = False, object_colors: bool = False, + header_labels: dict[str, str] | None = None, ) -> None: records = [data] if isinstance(data, dict) else list(data) if not records: @@ -48,7 +49,7 @@ def render( table = Table(show_header=True, header_style="bold") for col in fieldnames: - table.add_column(col) + table.add_column(header_labels.get(col, col) if header_labels else col) for r in flat_records: table.add_row(*[_format_cell(r.get(col, ""), color=color) for col in fieldnames]) diff --git a/nsc/savedfilters/custom_fields.py b/nsc/savedfilters/custom_fields.py new file mode 100644 index 0000000..1e690e1 --- /dev/null +++ b/nsc/savedfilters/custom_fields.py @@ -0,0 +1,151 @@ +"""Resolve a model's custom-field definitions (name -> label, type, choices). + +The list payload's ``custom_fields`` dict is keyed by field *name* only; the +human label and type live in the definitions at ``/api/extras/custom-fields/``, +filtered by ``object_type`` (e.g. ``dcim.device``). This mirrors +:class:`~nsc.savedfilters.objecttypes.ObjectTypeResolver`: fetch once per object +type, cache, and return ``None`` (never raise) when the registry can't be reached +so callers fall back to the raw key. Used to show custom-field column labels +(#132) and to build per-field edit widgets (#134). + +Choice values for ``select``/``multiselect`` fields come from a referenced +choice set; they are fetched best-effort and left empty on any failure. +""" + +from __future__ import annotations + +from collections.abc import Iterable +from dataclasses import dataclass +from typing import Any, Protocol + +from nsc.http.errors import NetBoxAPIError, NetBoxClientError +from nsc.savedfilters.objecttypes import ObjectTypeResolver + +_CUSTOM_FIELDS_PATH = "/api/extras/custom-fields/" +_CHOICE_SETS_PATH = "/api/extras/custom-field-choice-sets/" +_CF_PREFIX = "custom_fields." + + +class _ClientLike(Protocol): + def paginate( + self, path: str, params: dict[str, Any] | None = ..., *, limit: int | None = ... + ) -> Any: ... + + def get(self, path: str, params: dict[str, Any] | None = ...) -> dict[str, Any]: ... + + +@dataclass(frozen=True) +class CustomFieldDef: + name: str + label: str + type: str = "text" + choices: tuple[str, ...] = () + required: bool = False + + +def humanize(name: str) -> str: + """``site_contact`` -> ``Site Contact`` (fallback when a label is blank).""" + cleaned = name.replace("_", " ").strip() + return cleaned.title() if cleaned else name + + +def custom_field_labels( + columns: Iterable[str], defs: dict[str, CustomFieldDef] | None +) -> dict[str, str]: + """Map every column key to its display label. + + Non-custom-field keys map to themselves. ``custom_fields.`` keys map to + the field's resolved label. When two visible custom-field columns resolve to + the same label they are ambiguous, so both fall back to their raw key. Unknown + custom fields and a missing ``defs`` also fall back to the raw key. + """ + cols = list(columns) + if not defs: + return {col: col for col in cols} + resolved: dict[str, str] = {} + for col in cols: + if col.startswith(_CF_PREFIX) and (cf := defs.get(col[len(_CF_PREFIX) :])) is not None: + resolved[col] = cf.label + seen: dict[str, list[str]] = {} + for col, label in resolved.items(): + seen.setdefault(label, []).append(col) + labels: dict[str, str] = {} + for col in cols: + unique = resolved.get(col) + if unique is not None and len(seen[unique]) == 1: + labels[col] = unique + else: + labels[col] = col + return labels + + +class CustomFieldResolver: + """Looks up custom-field definitions from the live API, cached per object type.""" + + def __init__(self, client: _ClientLike, object_types: ObjectTypeResolver | None = None) -> None: + self._client = client + self._object_types = object_types or ObjectTypeResolver(client) + self._cache: dict[str, dict[str, CustomFieldDef]] = {} + self._choice_cache: dict[int, tuple[str, ...]] = {} + + def resolve(self, list_path: str) -> dict[str, CustomFieldDef] | None: + """``{name: CustomFieldDef}`` for a list path, or ``None`` if unresolvable.""" + object_type = self._object_types.resolve(list_path) + if object_type is None: + return None + if object_type in self._cache: + return self._cache[object_type] + try: + records = list(self._client.paginate(_CUSTOM_FIELDS_PATH, {"object_type": object_type})) + except (NetBoxAPIError, NetBoxClientError): + return None + defs = {d.name: d for d in (self._to_def(rec) for rec in records)} + self._cache[object_type] = defs + return defs + + def _to_def(self, record: dict[str, Any]) -> CustomFieldDef: + name = str(record.get("name", "")) + raw_label = record.get("label") + label = raw_label if isinstance(raw_label, str) and raw_label else humanize(name) + cf_type = record.get("type") + type_value = cf_type.get("value") if isinstance(cf_type, dict) else cf_type + type_str = type_value if isinstance(type_value, str) else "text" + choices = self._choices_for(record.get("choice_set")) + return CustomFieldDef( + name=name, + label=label, + type=type_str, + choices=choices, + required=bool(record.get("required", False)), + ) + + def _choices_for(self, choice_set: Any) -> tuple[str, ...]: + if not isinstance(choice_set, dict): + return () + cs_id = choice_set.get("id") + if not isinstance(cs_id, int): + return () + if cs_id in self._choice_cache: + return self._choice_cache[cs_id] + try: + payload = self._client.get(f"{_CHOICE_SETS_PATH}{cs_id}/") + except (NetBoxAPIError, NetBoxClientError): + return () + values = _extract_choice_values(payload) + self._choice_cache[cs_id] = values + return values + + +def _extract_choice_values(payload: dict[str, Any]) -> tuple[str, ...]: + raw = payload.get("choices") + if not isinstance(raw, list): + raw = payload.get("extra_choices") + if not isinstance(raw, list): + return () + values: list[str] = [] + for item in raw: + if isinstance(item, (list, tuple)) and item: + values.append(str(item[0])) + elif isinstance(item, str): + values.append(item) + return tuple(values) diff --git a/nsc/savedfilters/tags.py b/nsc/savedfilters/tags.py new file mode 100644 index 0000000..48d9cf0 --- /dev/null +++ b/nsc/savedfilters/tags.py @@ -0,0 +1,62 @@ +"""Resolve the available NetBox tags for tag-picker widgets (#134). + +Tags are global (not object-type scoped), so a single fetch of +``/api/extras/tags/`` is cached for the session. Mirrors the other resolvers: +returns ``None`` (never raises) when the API can't be reached, so the tag widget +degrades to free-text input. A writable ``tags`` PATCH wants a list of +``{name, slug}`` objects, so both are carried. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any, Protocol + +from nsc.http.errors import NetBoxAPIError, NetBoxClientError +from nsc.output.colors import normalize_hex + +_TAGS_PATH = "/api/extras/tags/" + + +class _ClientLike(Protocol): + def paginate( + self, path: str, params: dict[str, Any] | None = ..., *, limit: int | None = ... + ) -> Any: ... + + +@dataclass(frozen=True) +class TagDef: + name: str + slug: str + color: str | None = None + + @property + def label(self) -> str: + return self.name + + +class TagsResolver: + """Fetches the global tag list once and caches it for the session.""" + + def __init__(self, client: _ClientLike) -> None: + self._client = client + self._cache: tuple[TagDef, ...] | None = None + + def resolve(self) -> tuple[TagDef, ...] | None: + """All tags, or ``None`` if the API can't be reached.""" + if self._cache is not None: + return self._cache + try: + records = list(self._client.paginate(_TAGS_PATH)) + except (NetBoxAPIError, NetBoxClientError): + return None + tags = tuple( + TagDef( + name=str(rec.get("name", "")), + slug=str(rec.get("slug", "")), + color=normalize_hex(rec.get("color")), + ) + for rec in records + ) + self._cache = tags + return tags diff --git a/nsc/tui/app.py b/nsc/tui/app.py index b7b3371..5d5da51 100644 --- a/nsc/tui/app.py +++ b/nsc/tui/app.py @@ -53,6 +53,19 @@ 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 + from nsc.savedfilters.custom_fields import CustomFieldResolver # noqa: PLC0415 + from nsc.savedfilters.tags import TagsResolver # noqa: PLC0415 + + self._custom_field_resolver = CustomFieldResolver(client) + self._tags_resolver = TagsResolver(client) + + def custom_field_defs_for(self, tag: str, resource: str) -> dict[str, Any] | None: + """Custom-field definitions for a resource (read by list/forms), or None.""" + return self._custom_field_resolver.resolve(self._list_path(tag, resource)) + + def available_tags(self) -> Any | None: + """All NetBox tags (read by the tag-picker widget), or None if unavailable.""" + return self._tags_resolver.resolve() def columns_for(self, tag: str, resource: str) -> list[str] | None: """Saved visible columns for a resource, if any (read by ListScreen).""" diff --git a/nsc/tui/forms.py b/nsc/tui/forms.py index fa8eed5..9a21930 100644 --- a/nsc/tui/forms.py +++ b/nsc/tui/forms.py @@ -7,13 +7,18 @@ from __future__ import annotations import enum -from typing import Literal +from collections.abc import Iterable +from typing import Any, Literal from pydantic import BaseModel, ConfigDict from nsc.model.command_model import FieldShape, PrimitiveType +from nsc.savedfilters.custom_fields import CustomFieldDef +from nsc.savedfilters.tags import TagDef -WidgetKind = Literal["select", "switch", "number", "text", "masked"] +WidgetKind = Literal["select", "switch", "number", "text", "masked", "multi_select"] + +_CF_PREFIX = "custom_fields." class _SetNull(enum.Enum): @@ -34,6 +39,9 @@ class WidgetSpec(BaseModel): nullable: bool = False sensitive: bool = False is_float: bool = False + # multi_select only: (label, value) options and the currently-selected values. + options: tuple[tuple[str, str], ...] = () + selected: tuple[str, ...] = () def field_to_widget(name: str, field: FieldShape, sensitive_paths: tuple[str, ...]) -> WidgetSpec: @@ -56,6 +64,130 @@ def field_to_widget(name: str, field: FieldShape, sensitive_paths: tuple[str, .. return WidgetSpec(kind="text", name=name, nullable=field.nullable) +_CF_KINDS: dict[str, WidgetKind] = { + "integer": "number", + "decimal": "number", + "boolean": "switch", + "select": "select", + "multiselect": "multi_select", +} + + +def custom_field_widget(cf: CustomFieldDef) -> WidgetSpec: + """Widget spec for one custom field, keyed by its ``custom_fields.`` path. + + The type drives the widget; ``select``/``multiselect`` carry the choice set. + Unknown/object types fall back to a text input. + """ + name = f"{_CF_PREFIX}{cf.name}" + kind = _CF_KINDS.get(cf.type, "text") + nullable = not cf.required + if kind == "select": + return WidgetSpec(kind="select", name=name, choices=cf.choices, nullable=nullable) + if kind == "multi_select": + return WidgetSpec( + kind="multi_select", + name=name, + options=tuple((c, c) for c in cf.choices), + nullable=nullable, + ) + return WidgetSpec(kind=kind, name=name, nullable=nullable, is_float=cf.type == "decimal") + + +def expand_custom_fields(defs: dict[str, CustomFieldDef]) -> list[WidgetSpec]: + """One widget spec per custom field, in definition order.""" + return [custom_field_widget(cf) for cf in defs.values()] + + +def encode_field_id(name: str) -> str: + """Make a field name safe for a Textual widget id (dots are invalid there). + + NetBox field and custom-field names are ``[a-z0-9_]+``, so the only dot is the + ``custom_fields.`` separator; ``.`` -> ``-`` round-trips losslessly. + """ + return name.replace(".", "-") + + +def decode_field_id(token: str) -> str: + """Inverse of :func:`encode_field_id` (applied after stripping the id prefix).""" + return token.replace("-", ".") + + +def tag_slugs(value: object) -> tuple[str, ...]: + """Slugs of a record's ``tags`` list (falling back to name), or empty.""" + if not isinstance(value, list): + return () + slugs: list[str] = [] + for item in value: + if isinstance(item, dict): + slug = item.get("slug") or item.get("name") + if slug: + slugs.append(str(slug)) + return tuple(slugs) + + +def tags_widget_spec( + name: str, tags: tuple[TagDef, ...], current_slugs: tuple[str, ...] +) -> WidgetSpec: + """Multi-select spec offering every tag (label/slug) with current ones selected.""" + return WidgetSpec( + kind="multi_select", + name=name, + options=tuple((t.label, t.slug) for t in tags), + selected=current_slugs, + nullable=True, + ) + + +def tags_payload(slugs: Iterable[str], tags: tuple[TagDef, ...]) -> list[dict[str, str]]: + """Build a NetBox writable ``tags`` payload (list of ``{name, slug}``). + + Unknown slugs (a tag list that couldn't be resolved) fall back to using the + slug as both name and slug so the payload is still well-formed. + """ + by_slug = {t.slug: t for t in tags} + payload: list[dict[str, str]] = [] + for slug in slugs: + tag = by_slug.get(slug) + if tag is not None: + payload.append({"name": tag.name, "slug": tag.slug}) + else: + payload.append({"name": slug, "slug": slug}) + return payload + + +def flatten_custom_fields(record: dict[str, Any]) -> dict[str, Any]: + """Copy of ``record`` with ``custom_fields`` exploded into dotted keys. + + Lets :func:`compute_patch` diff each custom field individually against the + widget-staged ``custom_fields.`` values. The original is not mutated. + """ + flat = dict(record) + cf = record.get("custom_fields") + if isinstance(cf, dict): + for key, value in cf.items(): + flat[f"{_CF_PREFIX}{key}"] = value + return flat + + +def nest_custom_fields(patch: dict[str, Any]) -> dict[str, Any]: + """Fold ``custom_fields.`` patch entries back into one nested dict. + + NetBox expects custom fields nested under ``custom_fields``; the widgets stage + them flat. Non-custom-field entries pass through unchanged. + """ + nested: dict[str, Any] = {} + result: dict[str, Any] = {} + for key, value in patch.items(): + if key.startswith(_CF_PREFIX): + nested[key[len(_CF_PREFIX) :]] = value + else: + result[key] = value + if nested: + result["custom_fields"] = nested + return result + + class DiffRow(BaseModel): model_config = ConfigDict(frozen=True, extra="forbid") diff --git a/nsc/tui/screens/bulk_edit_form.py b/nsc/tui/screens/bulk_edit_form.py index 6d8833f..6351fb4 100644 --- a/nsc/tui/screens/bulk_edit_form.py +++ b/nsc/tui/screens/bulk_edit_form.py @@ -16,13 +16,38 @@ from textual.binding import BindingType from textual.containers import Horizontal, VerticalScroll from textual.screen import Screen -from textual.widgets import Button, Footer, Header, Input, Label, ProgressBar, Select, Switch +from textual.widgets import ( + Button, + Footer, + Header, + Input, + Label, + ProgressBar, + Select, + SelectionList, + Switch, +) +from textual.widgets.selection_list import Selection from nsc.model.command_model import CommandModel, Operation +from nsc.savedfilters.custom_fields import CustomFieldDef +from nsc.savedfilters.tags import TagDef from nsc.tui._bindings import textual_bindings from nsc.tui.bulk import RecordChange, shared_values from nsc.tui.fk import is_fk_value, resolve_fk_target -from nsc.tui.forms import SET_NULL, WidgetSpec, field_to_widget, fk_display +from nsc.tui.forms import ( + SET_NULL, + WidgetSpec, + decode_field_id, + encode_field_id, + expand_custom_fields, + field_to_widget, + fk_display, + flatten_custom_fields, + nest_custom_fields, + tags_payload, + tags_widget_spec, +) from nsc.tui.view import detail_path # Sentinel: a picker FK opted in with neither a pick nor a shared value @@ -56,6 +81,8 @@ def __init__( resource_name: str, update_op: Operation, selected_records: list[dict[str, Any]], + custom_field_defs: dict[str, CustomFieldDef] | None = None, + available_tags: tuple[TagDef, ...] | None = None, ) -> None: super().__init__() self._model = model @@ -64,6 +91,8 @@ def __init__( self._resource_name = resource_name self._op = update_op self._selected = selected_records + self._cf_defs = custom_field_defs + self._tags = available_tags self._specs: dict[str, WidgetSpec] = {} self._values: dict[str, Any] = {} self._included: set[str] = set() @@ -89,22 +118,34 @@ def compose(self) -> ComposeResult: fields = body.fields if body is not None else {} sensitive = body.sensitive_paths if body is not None else () for name, field in fields.items(): - spec = field_to_widget(name, field, sensitive) - self._specs[name] = spec - yield from self._compose_field(name, spec) + for spec in self._specs_for(name, field, sensitive): + self._specs[spec.name] = spec + yield from self._compose_field(spec.name, spec) yield Button("Preview", id="preview", classes="bulk-preview") progress = ProgressBar(id="bulk-progress", show_eta=False) progress.display = False yield progress yield Footer() + def _specs_for(self, name: str, field: Any, sensitive: tuple[str, ...]) -> list[WidgetSpec]: + """Widget spec(s) for a body field — expanding custom_fields and tags. + + Each expanded custom field gets its own include toggle, so users opt in + only the fields they intend to set across the selection. + """ + if name == "custom_fields" and self._cf_defs: + return expand_custom_fields(self._cf_defs) + if name == "tags" and self._tags is not None: + return [tags_widget_spec(name, self._tags, ())] + return [field_to_widget(name, field, sensitive)] + def _compose_field(self, name: str, spec: WidgetSpec) -> ComposeResult: with Horizontal(classes="bulk-field"): - yield Switch(value=False, id=f"include-{name}", classes="bulk-include") + yield Switch(value=False, id=f"include-{encode_field_id(name)}", classes="bulk-include") yield Label(name, classes="bulk-label") yield from self._compose_widget(name, spec) - if spec.nullable: - yield Button("∅", id=f"setnull-{name}", classes="bulk-setnull") + if spec.nullable and spec.kind != "multi_select": + yield Button("∅", id=f"setnull-{encode_field_id(name)}", classes="bulk-setnull") def _is_fk(self, name: str, spec: WidgetSpec) -> bool: if spec.kind in ("select", "switch", "masked") or spec.is_float: @@ -137,13 +178,25 @@ def _compose_fk(self, name: str) -> ComposeResult: shared_id = self._shared.get(name) if target.kind == "raw_id": text = "" if shared_id is None else str(shared_id) - yield Input(value=text, id=f"field-{name}") + yield Input(value=text, id=f"field-{encode_field_id(name)}") if target.hint: yield Label(target.hint, classes="bulk-fk-hint") return - yield Button(f"{name}: {self._fk_seed_label(name)}", id=f"fk-{name}", classes="bulk-fk") + yield Button( + f"{name}: {self._fk_seed_label(name)}", + id=f"fk-{encode_field_id(name)}", + classes="bulk-fk", + ) def _compose_widget(self, name: str, spec: WidgetSpec) -> ComposeResult: + wid = f"field-{encode_field_id(name)}" + if spec.kind == "multi_select": + yield SelectionList[str]( + *(Selection(label, val, val in spec.selected) for label, val in spec.options), + id=wid, + classes="bulk-multiselect", + ) + return if self._is_fk(name, spec): yield from self._compose_fk(name) return @@ -151,19 +204,19 @@ def _compose_widget(self, name: str, spec: WidgetSpec) -> ComposeResult: if spec.kind == "select": options = [(choice, choice) for choice in spec.choices] value = shared if shared in spec.choices else Select.NULL - yield Select(options, value=value, id=f"field-{name}", allow_blank=True) + yield Select(options, value=value, id=wid, allow_blank=True) return if spec.kind == "switch": - yield Switch(value=bool(shared), id=f"field-{name}") + yield Switch(value=bool(shared), id=wid) return text = "" if shared is None else str(shared) - yield Input(value=text, password=spec.sensitive, id=f"field-{name}") + yield Input(value=text, password=spec.sensitive, id=wid) @staticmethod def _strip(ident: str | None, prefix: str) -> str | None: if ident is None or not ident.startswith(prefix): return None - return ident.removeprefix(prefix) + return decode_field_id(ident.removeprefix(prefix)) def on_input_changed(self, event: Input.Changed) -> None: name = self._strip(event.input.id, "field-") @@ -203,15 +256,24 @@ def on_switch_changed(self, event: Switch.Changed) -> None: def _read_widget_value(self, name: str) -> Any: spec = self._specs.get(name) + wid = f"#field-{encode_field_id(name)}" if self._fk_kinds.get(name) == "picker": shared_id = self._shared.get(name) return shared_id if shared_id is not None else _NO_SEED + if spec is not None and spec.kind == "multi_select": + return self._multiselect_value(name, self.query_one(wid, SelectionList).selected) if spec is not None and spec.kind == "select": - value = self.query_one(f"#field-{name}", Select).value + value = self.query_one(wid, Select).value return None if value is Select.NULL else value if spec is not None and spec.kind == "switch": - return self.query_one(f"#field-{name}", Switch).value - return self._coerce_input(name, self.query_one(f"#field-{name}", Input).value) + return self.query_one(wid, Switch).value + return self._coerce_input(name, self.query_one(wid, Input).value) + + def _multiselect_value(self, name: str, selected: list[Any]) -> Any: + slugs = tuple(str(v) for v in selected) + if name == "tags": + return tags_payload(slugs, self._tags or ()) + return list(slugs) def on_select_changed(self, event: Select.Changed) -> None: name = self._strip(event.select.id, "field-") @@ -219,6 +281,12 @@ def on_select_changed(self, event: Select.Changed) -> None: return self._values[name] = None if event.value is Select.NULL else event.value + def on_selection_list_selected_changed(self, event: SelectionList.SelectedChanged[str]) -> None: + name = self._strip(event.selection_list.id, "field-") + if name is None: + return + self._values[name] = self._multiselect_value(name, event.selection_list.selected) + def on_button_pressed(self, event: Button.Pressed) -> None: ident = event.button.id if ident is None: @@ -247,7 +315,9 @@ def _stage(result: tuple[int, str] | None) -> None: if result is not None: self._values[name] = result[0] self._fk_labels[name] = result[1] - self.query_one(f"#fk-{name}", Button).label = f"{name}: {result[1]}" + self.query_one( + f"#fk-{encode_field_id(name)}", Button + ).label = f"{name}: {result[1]}" self.app.push_screen(RecordPicker(self._client, target.list_op, target.current_id), _stage) @@ -257,7 +327,9 @@ def action_preview(self) -> None: body = self._op.request_body sensitive = body.sensitive_paths if body is not None else () - changes = bulk_diff(self._selected, self.bulk_set, sensitive, self._fk_labels) + # Flatten custom_fields so each chosen custom_fields. diffs per record. + flattened = [flatten_custom_fields(record) for record in self._selected] + changes = bulk_diff(flattened, self.bulk_set, sensitive, self._fk_labels) def _on_confirm(confirmed: bool | None) -> None: if confirmed: @@ -282,7 +354,7 @@ async def _run_bulk( def _patch(change: RecordChange) -> None: self._client.patch( detail_path(self._op.path, change.record_id), - json=change.patch, + json=nest_custom_fields(change.patch), operation_id=self._op.operation_id, sensitive_paths=sensitive_paths, ) diff --git a/nsc/tui/screens/columns.py b/nsc/tui/screens/columns.py index b98fc0b..f7e4a48 100644 --- a/nsc/tui/screens/columns.py +++ b/nsc/tui/screens/columns.py @@ -24,9 +24,16 @@ class ColumnChooserScreen(ModalScreen[list[str]]): ("escape", "cancel", "Cancel"), ] - def __init__(self, available: list[str], visible: list[str]) -> None: + def __init__( + self, + available: list[str], + visible: list[str], + *, + labels: dict[str, str] | None = None, + ) -> None: super().__init__() self.selection = ColumnSelection(available, visible) + self._labels = labels or {} def compose(self) -> ComposeResult: with Vertical(id="columns-box"): @@ -42,7 +49,7 @@ def _rebuild(self, index: int) -> None: listing.clear() for name in self.selection.items: mark = "x" if self.selection.is_visible(name) else " " - listing.append(ListItem(Label(f"[{mark}] {name}"))) + listing.append(ListItem(Label(f"[{mark}] {self._labels.get(name, name)}"))) if self.selection.items: listing.index = max(0, min(index, len(self.selection.items) - 1)) diff --git a/nsc/tui/screens/detail.py b/nsc/tui/screens/detail.py index c8597ff..935ce72 100644 --- a/nsc/tui/screens/detail.py +++ b/nsc/tui/screens/detail.py @@ -12,6 +12,7 @@ from dataclasses import dataclass from typing import Any, ClassVar +from rich.text import Text from textual.app import ComposeResult from textual.binding import BindingType from textual.containers import Horizontal @@ -26,7 +27,7 @@ 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 -from nsc.tui.view import detail_path +from nsc.tui.view import detail_path, render_cell @dataclass @@ -88,7 +89,7 @@ def _tabs(self) -> Tabs: return self.query_one(Tabs) @property - def _table(self) -> DataTable[str]: + def _table(self) -> DataTable[str | Text]: return self.query_one("#fields", DataTable) def on_mount(self) -> None: @@ -124,28 +125,48 @@ def _refresh_rows(self) -> None: if table.row_count: table.move_cursor(row=min(cursor, table.row_count - 1)) - def _value_display(self, row: _FieldRow, flat: dict[str, Any]) -> str: - sensitive = row.spec is not None and row.spec.sensitive + def _value_display(self, row: _FieldRow, flat: dict[str, Any]) -> str | Text: if row.editable and row.name in self.staged: - staged = self.staged[row.name] - if staged is SET_NULL: - return "(null)" - # Never echo a just-typed secret; a picked FK stages a bare id, so - # show its chosen label rather than the number. - return ( - "****" if sensitive and staged != "" else self._fk_labels.get(row.name, str(staged)) - ) + return self._staged_display(row) + sensitive = row.spec is not None and row.spec.sensitive if row.editable: - value = self._record.get(row.name) - if isinstance(value, dict): - display = value.get("display") - return str(display if display is not None else value.get("id", "")) - if sensitive and value not in (None, ""): - return "****" - return "" if value is None else str(value) + return self._editable_display(row.name, sensitive=sensitive) + if isinstance(self._record.get(row.name), list): + return self._render_list(row.name) value = flat.get(row.name) return "" if value is None else str(value) + def _staged_display(self, row: _FieldRow) -> str: + staged = self.staged[row.name] + if staged is SET_NULL: + return "(null)" + # Never echo a just-typed secret; a picked FK stages a bare id, so show + # its chosen label rather than the number. + sensitive = row.spec is not None and row.spec.sensitive + return "****" if sensitive and staged != "" else self._fk_labels.get(row.name, str(staged)) + + def _editable_display(self, name: str, *, sensitive: bool) -> str | Text: + value = self._record.get(name) + if isinstance(value, dict): + display = value.get("display") + return str(display if display is not None else value.get("id", "")) + if sensitive and value not in (None, ""): + return "****" + if isinstance(value, list): + return self._render_list(name) + return "" if value is None else str(value) + + def _render_list(self, name: str) -> str | Text: + """Render a list-of-object field (tags, …) like the list table does. + + Routes through the same flatten/colour pipeline so the cell shows a + comma-joined display string — colored when object colors are on — rather + than a raw list-of-dicts repr. + """ + object_colors = bool(getattr(self.app, "object_colors", False)) + rendered = flatten(self._record, columns=[name], with_colors=object_colors).get(name) + return render_cell(rendered) + # --- editing --------------------------------------------------------- def action_edit_field(self) -> None: if self._editing is not None: diff --git a/nsc/tui/screens/edit_form.py b/nsc/tui/screens/edit_form.py index 284b7b3..0604643 100644 --- a/nsc/tui/screens/edit_form.py +++ b/nsc/tui/screens/edit_form.py @@ -13,10 +13,13 @@ from textual.binding import BindingType from textual.containers import Horizontal, VerticalScroll from textual.screen import Screen -from textual.widgets import Button, Footer, Header, Input, Label, Select, Switch +from textual.widgets import Button, Footer, Header, Input, Label, Select, SelectionList, Switch +from textual.widgets.selection_list import Selection from nsc.http.errors import NetBoxAPIError, NetBoxClientError from nsc.model.command_model import CommandModel, Operation +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.fk import is_fk_value, resolve_fk_target @@ -24,9 +27,17 @@ SET_NULL, WidgetSpec, compute_patch, + decode_field_id, diff_rows, + encode_field_id, + expand_custom_fields, field_to_widget, fk_display, + flatten_custom_fields, + nest_custom_fields, + tag_slugs, + tags_payload, + tags_widget_spec, ) from nsc.tui.view import detail_path @@ -73,6 +84,8 @@ def __init__( resource_name: str, operation: Operation, record: dict[str, Any], + custom_field_defs: dict[str, CustomFieldDef] | None = None, + available_tags: tuple[TagDef, ...] | None = None, ) -> None: super().__init__() self._model = model @@ -80,7 +93,11 @@ def __init__( self._tag = tag self._resource_name = resource_name self._op = operation - self._record = record + # Explode custom_fields into custom_fields. keys so the per-field + # widgets seed and diff individually; the patch is re-nested before send. + self._record = flatten_custom_fields(record) + self._cf_defs = custom_field_defs + self._tags = available_tags self._specs: dict[str, WidgetSpec] = {} self._fk_labels: dict[str, str] = {} self.staged: dict[str, Any] = {} @@ -97,34 +114,56 @@ def compose(self) -> ComposeResult: fields = body.fields if body is not None else {} sensitive = body.sensitive_paths if body is not None else () for name, field in fields.items(): - spec = field_to_widget(name, field, sensitive) - self._specs[name] = spec - yield from self._compose_field(name, spec) + for spec in self._specs_for(name, field, sensitive): + self._specs[spec.name] = spec + yield from self._compose_field(spec.name, spec) yield Button("Save", id="save", classes="edit-save") yield Footer() + def _specs_for(self, name: str, field: Any, sensitive: tuple[str, ...]) -> list[WidgetSpec]: + """Widget spec(s) for a body field — expanding custom_fields and tags.""" + if name == "custom_fields" and self._cf_defs: + return expand_custom_fields(self._cf_defs) + if name == "tags" and self._tags is not None: + current = tag_slugs(self._record.get("tags")) + return [tags_widget_spec(name, self._tags, current)] + return [field_to_widget(name, field, sensitive)] + def _compose_field(self, name: str, spec: WidgetSpec) -> ComposeResult: value = _record_value(self._record, name) with Horizontal(classes="edit-field"): yield Label(name, classes="edit-label") yield from self._compose_widget(name, spec, value) - if spec.nullable: - yield Button("∅", id=f"setnull-{name}", classes="edit-setnull") + if spec.nullable and spec.kind != "multi_select": + yield Button("∅", id=f"setnull-{encode_field_id(name)}", classes="edit-setnull") def _compose_widget(self, name: str, spec: WidgetSpec, value: Any) -> ComposeResult: + wid = f"field-{encode_field_id(name)}" + if spec.kind == "multi_select": + # Tags seed via spec.selected; a custom-field multiselect seeds from + # the record's current list value. + current = set(spec.selected) + if not current and isinstance(value, list): + current = {str(v) for v in value} + yield SelectionList[str]( + *(Selection(label, val, val in current) for label, val in spec.options), + id=wid, + classes="edit-multiselect", + ) + return if self._is_fk(name, spec): yield from self._compose_fk(name, value) return if spec.kind == "select": options = [(choice, choice) for choice in spec.choices] select_value = value if value in spec.choices else Select.NULL - yield Select(options, value=select_value, id=f"field-{name}", allow_blank=True) + yield Select(options, value=select_value, id=wid, allow_blank=True) return if spec.kind == "switch": - yield Switch(value=bool(value), id=f"field-{name}") + yield Switch(value=bool(value), id=wid) return text = "" if value is None else str(value) - yield Input(value=text, password=spec.sensitive, id=f"field-{name}") + yield Input(value=text, password=spec.sensitive, id=wid) def _is_fk(self, name: str, spec: WidgetSpec) -> bool: # Writable FK fields type as oneOf[int, brief] -> UNKNOWN -> `text`, so a @@ -154,7 +193,7 @@ def _compose_fk(self, name: str, value: Any) -> ComposeResult: def _field_name(ident: str | None) -> str | None: if ident is None or not ident.startswith("field-"): return None - return ident.removeprefix("field-") + return decode_field_id(ident.removeprefix("field-")) def on_input_changed(self, event: Input.Changed) -> None: name = self._field_name(event.input.id) @@ -197,7 +236,7 @@ def on_button_pressed(self, event: Button.Pressed) -> None: self.action_save() return if ident.startswith("setnull-"): - field = ident.removeprefix("setnull-") + field = decode_field_id(ident.removeprefix("setnull-")) self.staged[field] = SET_NULL # Drop any picked label so the diff can't show a name while nulling. self._fk_labels.pop(field, None) @@ -205,6 +244,27 @@ def on_button_pressed(self, event: Button.Pressed) -> None: if ident.startswith("fk-"): self._open_picker(ident.removeprefix("fk-")) + def on_selection_list_selected_changed(self, event: SelectionList.SelectedChanged[str]) -> None: + name = self._field_name(event.selection_list.id) + if name is None: + return + selected = tuple(str(v) for v in event.selection_list.selected) + # Only stage when the selection actually changed (by set), so an + # unchanged pre-seeded selection — regardless of order — doesn't force a + # spurious patch. + if name == "tags": + if set(selected) == set(tag_slugs(self._record.get("tags"))): + self.staged.pop(name, None) + else: + self.staged[name] = tags_payload(selected, self._tags or ()) + return + original = self._record.get(name) + original_set = {str(v) for v in original} if isinstance(original, list) else set() + if set(selected) == original_set: + self.staged.pop(name, None) + else: + self.staged[name] = list(selected) + def _open_picker(self, name: str) -> None: target = resolve_fk_target(name, self._record.get(name), self._model) if target.list_op is None: @@ -237,6 +297,7 @@ def _on_confirm(confirmed: bool | None) -> None: ) def _apply_patch(self, patch: dict[str, Any], sensitive_paths: tuple[str, ...]) -> None: + patch = nest_custom_fields(patch) try: if self._create_mode: self._client.post( diff --git a/nsc/tui/screens/list.py b/nsc/tui/screens/list.py index b057442..a63c029 100644 --- a/nsc/tui/screens/list.py +++ b/nsc/tui/screens/list.py @@ -137,7 +137,8 @@ def _populate(self, records: list[dict[str, Any]]) -> None: table.clear(columns=True) sample = records[0] if records else None columns = choose_columns(self._op, self._columns_config, sample) - table.add_columns(_MARKER_HEADER, *columns) + labels = self._header_labels(columns) + table.add_columns(_MARKER_HEADER, *(labels.get(c, c) for c in columns)) object_colors = bool(getattr(self.app, "object_colors", False)) rows = build_rows(records, columns, object_colors=object_colors) for record, row in zip(records, rows, strict=True): @@ -145,6 +146,20 @@ def _populate(self, records: list[dict[str, Any]]) -> None: self._records = records self._columns = columns + def _header_labels(self, columns: list[str]) -> dict[str, str]: + """Display labels for custom-field columns; raw key for everything else. + + Only resolves definitions when a ``custom_fields.`` column is shown, + so ordinary lists incur no extra request. The raw key stays the selector. + """ + if not any(c.startswith("custom_fields.") for c in columns): + return {} + from nsc.savedfilters.custom_fields import custom_field_labels # noqa: PLC0415 + + defs_fn = getattr(self.app, "custom_field_defs_for", None) + defs = defs_fn(self._tag, self._resource_name) if callable(defs_fn) else None + return custom_field_labels(columns, defs) + def _prune_selection(self, records: list[dict[str, Any]]) -> None: present = {record.get("id") for record in records} for record_id in self._selection.ids(): @@ -197,8 +212,14 @@ def _apply(columns: list[str] | None) -> None: if callable(saver): saver(self._tag, self._resource_name, columns) + available = available_columns(self._records) self.app.push_screen( - ColumnChooserScreen(available_columns(self._records), list(self._columns)), _apply + ColumnChooserScreen( + available, + list(self._columns), + labels=self._header_labels(available), + ), + _apply, ) def action_cursor_down(self) -> None: @@ -250,6 +271,14 @@ def action_go_back(self) -> None: else: self.notify("Nothing to go back to — press q to quit or ctrl+p to switch resource.") + def _form_data_sources(self) -> tuple[dict[str, Any] | None, tuple[Any, ...] | None]: + """Custom-field defs + available tags from the app, for editing widgets.""" + defs_fn = getattr(self.app, "custom_field_defs_for", None) + tags_fn = getattr(self.app, "available_tags", None) + defs = defs_fn(self._tag, self._resource_name) if callable(defs_fn) else None + tags = tags_fn() if callable(tags_fn) else None + return defs, tags + def action_create_record(self) -> None: resource = self._model.tags[self._tag].resources[self._resource_name] create_op = resource.create_op @@ -260,6 +289,7 @@ def action_create_record(self) -> None: def _after(_: None) -> None: self.reload() + defs, tags = self._form_data_sources() self.app.push_screen( EditForm( self._model, @@ -268,6 +298,8 @@ def _after(_: None) -> None: self._resource_name, create_op, {}, + custom_field_defs=defs, + available_tags=tags, ), _after, ) @@ -287,6 +319,7 @@ def _after(_: None) -> None: self._selection.clear() self.reload() + defs, tags = self._form_data_sources() self.app.push_screen( BulkEditForm( self._model, @@ -295,6 +328,8 @@ def _after(_: None) -> None: self._resource_name, update_op, selected, + custom_field_defs=defs, + available_tags=tags, ), _after, ) diff --git a/nsc/tui/styles.tcss b/nsc/tui/styles.tcss index 62012a1..ea4c086 100644 --- a/nsc/tui/styles.tcss +++ b/nsc/tui/styles.tcss @@ -50,6 +50,39 @@ EditForm #edit-form-body { color: $text-muted; } +.edit-multiselect { + height: auto; + max-height: 8; + border: round $panel; +} + +/* ---- bulk edit: include-toggle + label + widget rows, aligned to single edit ---- */ +BulkEditForm #bulk-form-body { + height: 1fr; +} + +.bulk-field { + height: auto; + min-height: 3; + padding: 0 1; +} + +.bulk-include { + width: 6; +} + +.bulk-label { + width: 20; + content-align: left middle; + color: $text-muted; +} + +.bulk-multiselect { + height: auto; + max-height: 8; + border: round $panel; +} + /* ---- generic modal dialogs: centre, box, dim the backdrop ---- */ ModalScreen { align: center middle; diff --git a/nsc/tui/view.py b/nsc/tui/view.py index db2987b..9ae3555 100644 --- a/nsc/tui/view.py +++ b/nsc/tui/view.py @@ -44,7 +44,7 @@ def build_rows( rows: list[list[str | Text]] = [] for record in records: flat = flatten(record, columns=columns, with_colors=object_colors) - rows.append([_cell(flat.get(col)) for col in columns]) + rows.append([render_cell(flat.get(col)) for col in columns]) return rows @@ -54,7 +54,13 @@ def _colored_text(value: ColoredValue) -> Text: return Text(value.text, style=f"#{value.color}") -def _cell(value: Any) -> str | Text: +def render_cell(value: Any) -> str | Text: + """Render a flattened cell value as plain text or colored Rich ``Text``. + + Shared by the list table and the detail view so list-of-object fields (tags) + render identically in both — colored chips or a comma-joined display string — + rather than a raw repr. + """ if value is None: return "" if isinstance(value, ColoredValue): diff --git a/tests/cli/test_handlers.py b/tests/cli/test_handlers.py index fe2cbe6..b2d906b 100644 --- a/tests/cli/test_handlers.py +++ b/tests/cli/test_handlers.py @@ -1,5 +1,6 @@ from __future__ import annotations +import csv import io import json from typing import Any @@ -196,3 +197,50 @@ def test_handle_list_emits_envelope_on_api_error(monkeypatch: pytest.MonkeyPatch with pytest.raises(typer.Exit) as ei: handle_list(operation, "dcim", "devices", ctx, stream=buf) assert ei.value.exit_code == 9 # EXIT_CODES[ErrorType.NOT_FOUND] + + +class _CfFakeClient: + """Responds per-path for the custom-field label resolution chain.""" + + def __init__(self, obj: dict[str, Any]) -> None: + self._obj = obj + + def get(self, path: str, params: Any = None) -> dict[str, Any]: + return self._obj + + def paginate(self, path: str, params: Any = None, *, limit: Any = None) -> Any: + if path == "/api/core/object-types/": + yield { + "app_label": "dcim", + "model": "device", + "rest_api_endpoint": "/api/dcim/devices/", + } + elif path == "/api/extras/custom-fields/": + yield {"name": "rack_role", "label": "Rack Role", "type": {"value": "text"}} + + +def test_handle_get_relabels_custom_field_header_in_csv() -> None: + client = _CfFakeClient({"id": 7, "name": "sw1", "custom_fields": {"rack_role": "gold"}}) + op = Operation( + operation_id="dcim_devices_retrieve", + http_method=HttpMethod.GET, + path="/api/dcim/devices/{id}/", + parameters=[ + Parameter( + name="id", + location=ParameterLocation.PATH, + primitive=PrimitiveType.INTEGER, + required=True, + ), + ], + ) + buf = io.StringIO() + ctx = _ctx( + client, + output_format=OutputFormat.CSV, + columns_override=["name", "custom_fields.rack_role"], + ) + handle_get(op, "dcim", "devices", ctx, stream=buf, id=7) + rows = list(csv.reader(io.StringIO(buf.getvalue()))) + assert rows[0] == ["name", "Rack Role"] + assert rows[1] == ["sw1", "gold"] diff --git a/tests/output/test_csv.py b/tests/output/test_csv.py index 34a08c3..b7fe536 100644 --- a/tests/output/test_csv.py +++ b/tests/output/test_csv.py @@ -73,3 +73,16 @@ def test_csv_quotes_joined_list_cell_containing_comma() -> None: # The joined cell ("edge, dc1, prod") survives a CSV round-trip as one field. rows = list(csv.reader(io.StringIO(buf.getvalue()))) assert rows[1] == ["1", "edge, dc1, prod"] + + +def test_csv_relabels_header_keeps_data_keyed_by_raw_column() -> None: + buf = io.StringIO() + render_csv( + [{"name": "a", "custom_fields": {"rack_role": "gold"}}], + stream=buf, + columns=["name", "custom_fields.rack_role"], + header_labels={"name": "name", "custom_fields.rack_role": "Rack Role"}, + ) + rows = list(csv.reader(io.StringIO(buf.getvalue()))) + assert rows[0] == ["name", "Rack Role"] + assert rows[1] == ["a", "gold"] diff --git a/tests/output/test_table.py b/tests/output/test_table.py index 213ef4b..e1880f0 100644 --- a/tests/output/test_table.py +++ b/tests/output/test_table.py @@ -57,3 +57,17 @@ def test_table_renders_placeholder_for_empty_list() -> None: buf = io.StringIO() render_table([], stream=buf) assert "(no records)" in buf.getvalue() + + +def test_table_relabels_headers_with_header_labels() -> None: + buf = io.StringIO() + render_table( + [{"name": "a", "custom_fields": {"rack_role": "gold"}}], + stream=buf, + columns=["name", "custom_fields.rack_role"], + header_labels={"name": "name", "custom_fields.rack_role": "Rack Role"}, + ) + text = buf.getvalue() + assert "Rack Role" in text + assert "custom_fields.rack_role" not in text + assert "gold" in text diff --git a/tests/savedfilters/test_custom_fields.py b/tests/savedfilters/test_custom_fields.py new file mode 100644 index 0000000..a01d327 --- /dev/null +++ b/tests/savedfilters/test_custom_fields.py @@ -0,0 +1,157 @@ +from __future__ import annotations + +from collections.abc import Iterator +from typing import Any + +from nsc.http.errors import NetBoxAPIError +from nsc.savedfilters.custom_fields import ( + CustomFieldDef, + CustomFieldResolver, + custom_field_labels, + humanize, +) + +_OBJECT_TYPES = [ + {"app_label": "dcim", "model": "device", "rest_api_endpoint": "/api/dcim/devices/"}, +] + +_CUSTOM_FIELDS = [ + { + "name": "site_contact", + "label": "Site Contact", + "type": {"value": "text", "label": "Text"}, + "required": False, + }, + { + # blank label -> humanized fallback + "name": "rack_role", + "label": "", + "type": {"value": "text", "label": "Text"}, + }, + { + "name": "tier", + "label": "Tier", + "type": {"value": "select", "label": "Selection"}, + "choice_set": {"id": 7, "url": "/api/extras/custom-field-choice-sets/7/"}, + }, +] + +_CHOICE_SETS = { + 7: {"id": 7, "extra_choices": [["gold", "Gold"], ["silver", "Silver"]]}, +} + + +class _FakeClient: + def __init__( + self, + object_types: list[dict[str, Any]], + custom_fields: list[dict[str, Any]], + *, + fail: bool = False, + ) -> None: + self._object_types = object_types + self._custom_fields = custom_fields + self._fail = fail + self.calls: list[tuple[str, dict[str, Any] | None]] = [] + self.gets: list[str] = [] + + 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={}) + if path == "/api/core/object-types/": + app = (params or {}).get("app_label") + for rec in self._object_types: + if app is None or rec["app_label"] == app: + yield rec + return + if path == "/api/extras/custom-fields/": + yield from self._custom_fields + return + + def get(self, path: str, params: dict[str, Any] | None = None) -> dict[str, Any]: + self.gets.append(path) + cs_id = int(path.rstrip("/").rsplit("/", 1)[-1]) + return _CHOICE_SETS[cs_id] + + +def _resolver(**kw: Any) -> CustomFieldResolver: + client = _FakeClient(_OBJECT_TYPES, _CUSTOM_FIELDS, **kw) + return CustomFieldResolver(client) + + +def test_humanize() -> None: + assert humanize("site_contact") == "Site Contact" + assert humanize("tier") == "Tier" + + +def test_resolve_maps_name_to_label_and_type() -> None: + defs = _resolver().resolve("/api/dcim/devices/") + assert defs is not None + assert defs["site_contact"].label == "Site Contact" + assert defs["site_contact"].type == "text" + + +def test_resolve_humanizes_blank_label() -> None: + defs = _resolver().resolve("/api/dcim/devices/") + assert defs is not None + assert defs["rack_role"].label == "Rack Role" + + +def test_resolve_populates_choices_from_choice_set() -> None: + defs = _resolver().resolve("/api/dcim/devices/") + assert defs is not None + assert defs["tier"].type == "select" + assert defs["tier"].choices == ("gold", "silver") + + +def test_resolve_caches_per_object_type() -> None: + client = _FakeClient(_OBJECT_TYPES, _CUSTOM_FIELDS) + resolver = CustomFieldResolver(client) + resolver.resolve("/api/dcim/devices/") + resolver.resolve("/api/dcim/devices/") + cf_calls = [c for c in client.calls if c[0] == "/api/extras/custom-fields/"] + assert len(cf_calls) == 1 + + +def test_resolve_scopes_by_object_type() -> None: + client = _FakeClient(_OBJECT_TYPES, _CUSTOM_FIELDS) + CustomFieldResolver(client).resolve("/api/dcim/devices/") + assert ("/api/extras/custom-fields/", {"object_type": "dcim.device"}) in client.calls + + +def test_resolve_none_on_api_error() -> None: + assert _resolver(fail=True).resolve("/api/dcim/devices/") is None + + +def test_resolve_none_when_object_type_unresolvable() -> None: + client = _FakeClient([], _CUSTOM_FIELDS) + assert CustomFieldResolver(client).resolve("/api/dcim/devices/") is None + + +def test_custom_field_labels_maps_keys_and_passes_through_others() -> None: + defs = {"site_contact": CustomFieldDef(name="site_contact", label="Site Contact")} + labels = custom_field_labels(["name", "custom_fields.site_contact"], defs) + assert labels == {"name": "name", "custom_fields.site_contact": "Site Contact"} + + +def test_custom_field_labels_falls_back_to_raw_on_collision() -> None: + defs = { + "a": CustomFieldDef(name="a", label="Dup"), + "b": CustomFieldDef(name="b", label="Dup"), + } + labels = custom_field_labels(["custom_fields.a", "custom_fields.b"], defs) + assert labels == {"custom_fields.a": "custom_fields.a", "custom_fields.b": "custom_fields.b"} + + +def test_custom_field_labels_identity_when_defs_none() -> None: + cols = ["name", "custom_fields.site_contact"] + assert custom_field_labels(cols, None) == {c: c for c in cols} + + +def test_custom_field_labels_unknown_cf_falls_back_to_raw() -> None: + defs: dict[str, CustomFieldDef] = {} + labels = custom_field_labels(["custom_fields.ghost"], defs) + assert labels == {"custom_fields.ghost": "custom_fields.ghost"} diff --git a/tests/savedfilters/test_tags.py b/tests/savedfilters/test_tags.py new file mode 100644 index 0000000..43817d2 --- /dev/null +++ b/tests/savedfilters/test_tags.py @@ -0,0 +1,54 @@ +from __future__ import annotations + +from collections.abc import Iterator +from typing import Any + +from nsc.http.errors import NetBoxAPIError +from nsc.savedfilters.tags import TagDef, TagsResolver + +_TAGS = [ + {"id": 1, "name": "prod", "slug": "prod", "color": "4caf50"}, + {"id": 2, "name": "Edge", "slug": "edge", "color": "#FF0000"}, + {"id": 3, "name": "blank", "slug": "blank", "color": ""}, +] + + +class _FakeClient: + def __init__(self, tags: list[dict[str, Any]], *, fail: bool = False) -> None: + self._tags = tags + self._fail = fail + self.calls: list[str] = [] + + def paginate( + self, path: str, params: dict[str, Any] | None = None, *, limit: int | None = None + ) -> Iterator[dict[str, Any]]: + self.calls.append(path) + if self._fail: + raise NetBoxAPIError(status_code=404, url=path, body_snippet="", headers={}) + yield from self._tags + + +def test_resolve_returns_tag_defs() -> None: + tags = TagsResolver(_FakeClient(_TAGS)).resolve() + assert tags is not None + assert TagDef(name="prod", slug="prod", color="4caf50") in tags + + +def test_resolve_normalizes_color() -> None: + tags = TagsResolver(_FakeClient(_TAGS)).resolve() + assert tags is not None + by_slug = {t.slug: t for t in tags} + assert by_slug["edge"].color == "ff0000" + assert by_slug["blank"].color is None + + +def test_resolve_caches() -> None: + client = _FakeClient(_TAGS) + resolver = TagsResolver(client) + resolver.resolve() + resolver.resolve() + assert client.calls == ["/api/extras/tags/"] + + +def test_resolve_none_on_api_error() -> None: + assert TagsResolver(_FakeClient(_TAGS, fail=True)).resolve() is None diff --git a/tests/tui/test_bulk_edit_form.py b/tests/tui/test_bulk_edit_form.py index 9dffd1c..279502d 100644 --- a/tests/tui/test_bulk_edit_form.py +++ b/tests/tui/test_bulk_edit_form.py @@ -4,7 +4,7 @@ import pytest from textual.app import App, ComposeResult -from textual.widgets import Button, Input, ListView, Select, Static, Switch +from textual.widgets import Button, Input, ListView, Select, SelectionList, Static, Switch from nsc.http.errors import NetBoxAPIError from nsc.model.command_model import ( @@ -16,7 +16,9 @@ Resource, Tag, ) -from nsc.tui.forms import SET_NULL +from nsc.savedfilters.custom_fields import CustomFieldDef +from nsc.savedfilters.tags import TagDef +from nsc.tui.forms import SET_NULL, tags_payload from nsc.tui.screens.bulk_edit_form import BulkEditForm from nsc.tui.screens.list import ListScreen from nsc.tui.screens.record_picker import RecordPicker @@ -805,3 +807,152 @@ async def on_mount(self) -> None: await pilot.pause() assert screen.bulk_set == {"role": 7} assert isinstance(screen.bulk_set["role"], int) + + +# --- #134: per-field custom fields + tags multi-select in bulk edit --- + +_CF_DEFS = { + "tier": CustomFieldDef("tier", "Tier", type="select", choices=("gold", "silver")), + "count": CustomFieldDef("count", "Count", type="integer"), +} +_TAGS = (TagDef("Prod", "prod", "ff0000"), TagDef("Edge", "edge", None)) + + +def _cf_model() -> CommandModel: + update_op = Operation( + operation_id="dcim_devices_partial_update", + http_method="PATCH", + path="/api/dcim/devices/{id}/", + request_body=RequestBodyShape( + top_level="object", + fields={ + "name": FieldShape(primitive=PrimitiveType.STRING), + "custom_fields": FieldShape(primitive=PrimitiveType.OBJECT), + "tags": FieldShape(primitive=PrimitiveType.ARRAY), + }, + ), + ) + devices = Resource(name="devices", update_op=update_op) + return CommandModel( + info_title="t", + info_version="1", + schema_hash="h", + tags={"dcim": Tag(name="dcim", resources={"devices": devices})}, + ) + + +class _CfBulkApp(App[None]): + def __init__( + self, + client: _SpyClient, + *, + defs: dict[str, Any] | None = _CF_DEFS, + tags: tuple[TagDef, ...] | None = _TAGS, + selected: list[dict[str, Any]] | None = None, + ) -> None: + super().__init__() + self._client = client + self._defs = defs + self._tags = tags + self._sel = selected or [ + {"id": 1, "name": "sw1", "custom_fields": {"tier": "silver", "count": 1}, "tags": []}, + {"id": 2, "name": "sw2", "custom_fields": {"tier": "silver", "count": 2}, "tags": []}, + ] + + def compose(self) -> ComposeResult: + yield Static("") + + async def on_mount(self) -> None: + model = _cf_model() + op = model.tags["dcim"].resources["devices"].update_op + assert op is not None + await self.push_screen( + BulkEditForm( + model, + self._client, + "dcim", + "devices", + op, + self._sel, + custom_field_defs=self._defs, + available_tags=self._tags, + ) + ) + + +def _cf_screen(app: _CfBulkApp) -> BulkEditForm: + screen = app.screen + assert isinstance(screen, BulkEditForm) + return screen + + +@pytest.mark.asyncio +async def test_custom_fields_expand_into_per_field_widgets_with_toggles() -> None: + app = _CfBulkApp(_SpyClient([])) + async with app.run_test(size=(120, 60)) as pilot: + await pilot.pause() + screen = _cf_screen(app) + assert isinstance(screen.query_one("#field-custom_fields-tier"), Select) + assert isinstance(screen.query_one("#field-custom_fields-count"), Input) + assert isinstance(screen.query_one("#include-custom_fields-tier"), Switch) + # No single opaque custom_fields widget remains. + assert not screen.query("#field-custom_fields") + + +@pytest.mark.asyncio +async def test_tags_render_as_selection_list() -> None: + app = _CfBulkApp(_SpyClient([])) + async with app.run_test(size=(120, 60)) as pilot: + await pilot.pause() + screen = _cf_screen(app) + assert isinstance(screen.query_one("#field-tags"), SelectionList) + + +@pytest.mark.asyncio +async def test_bulk_apply_nests_custom_field_value() -> None: + client = _SpyClient([]) + app = _CfBulkApp(client) + async with app.run_test(size=(120, 60)) as pilot: + await pilot.pause() + screen = _cf_screen(app) + screen._included.add("custom_fields.tier") + screen._values["custom_fields.tier"] = "gold" + screen.action_preview() + await pilot.pause() + await pilot.press("enter") + await app.workers.wait_for_complete() + await pilot.pause() + assert client.patch_calls + for call in client.patch_calls: + assert call["json"] == {"custom_fields": {"tier": "gold"}} + + +@pytest.mark.asyncio +async def test_bulk_apply_sets_tags_as_name_slug_list() -> None: + client = _SpyClient([]) + app = _CfBulkApp(client) + async with app.run_test(size=(120, 60)) as pilot: + await pilot.pause() + screen = _cf_screen(app) + screen._included.add("tags") + screen._values["tags"] = tags_payload(("prod",), _TAGS) + screen.action_preview() + await pilot.pause() + await pilot.press("enter") + await app.workers.wait_for_complete() + await pilot.pause() + assert client.patch_calls + for call in client.patch_calls: + assert call["json"] == {"tags": [{"name": "Prod", "slug": "prod"}]} + + +@pytest.mark.asyncio +async def test_bulk_falls_back_to_text_inputs_without_defs() -> None: + app = _CfBulkApp(_SpyClient([]), defs=None, tags=None) + async with app.run_test(size=(120, 60)) as pilot: + await pilot.pause() + screen = _cf_screen(app) + # Degrades to the previous opaque single inputs, no crash. + assert isinstance(screen.query_one("#field-custom_fields"), Input) + assert isinstance(screen.query_one("#field-tags"), Input) + assert not screen.query("#field-custom_fields-tier") diff --git a/tests/tui/test_column_chooser.py b/tests/tui/test_column_chooser.py index 5e5c97c..2936f4c 100644 --- a/tests/tui/test_column_chooser.py +++ b/tests/tui/test_column_chooser.py @@ -2,18 +2,24 @@ import pytest from textual.app import App, ComposeResult -from textual.widgets import Static +from textual.widgets import Label, ListView, Static from nsc.tui.screens.columns import ColumnChooserScreen class _ChooserApp(App[None]): - def __init__(self, available: list[str], visible: list[str]) -> None: + def __init__( + self, + available: list[str], + visible: list[str], + labels: dict[str, str] | None = None, + ) -> None: super().__init__() self.result: list[str] | None = None self._sentinel = object() self._available = available self._visible = visible + self._chooser_labels = labels def compose(self) -> ComposeResult: yield Static("") @@ -22,7 +28,9 @@ async def on_mount(self) -> None: def _cb(cols: list[str] | None) -> None: self.result = cols - await self.push_screen(ColumnChooserScreen(self._available, self._visible), _cb) + await self.push_screen( + ColumnChooserScreen(self._available, self._visible, labels=self._chooser_labels), _cb + ) def _chooser(app: _ChooserApp) -> ColumnChooserScreen: @@ -115,3 +123,22 @@ async def test_empty_selection_is_not_applied() -> None: await pilot.pause() assert isinstance(app.screen, ColumnChooserScreen) # still open assert app.result is None + + +@pytest.mark.asyncio +async def test_chooser_renders_labels_but_applies_raw_keys() -> None: + app = _ChooserApp( + available=["name", "custom_fields.rack_role"], + visible=["name", "custom_fields.rack_role"], + labels={"name": "name", "custom_fields.rack_role": "Rack Role"}, + ) + async with app.run_test() as pilot: + await pilot.pause() + screen = _chooser(app) + shown = [label.render().plain for label in screen.query(ListView).first().query(Label)] + assert any("Rack Role" in s for s in shown) + assert not any("custom_fields.rack_role" in s for s in shown) + screen.action_apply() + await pilot.pause() + # The applied/visible columns keep the raw selector key, not the label. + assert app.result == ["name", "custom_fields.rack_role"] diff --git a/tests/tui/test_detail_screen.py b/tests/tui/test_detail_screen.py index 827e073..fc0871a 100644 --- a/tests/tui/test_detail_screen.py +++ b/tests/tui/test_detail_screen.py @@ -3,6 +3,7 @@ from typing import Any import pytest +from rich.text import Text from textual.app import App, ComposeResult from textual.coordinate import Coordinate from textual.widgets import DataTable, Input, ListView, Static, Tab, Tabs @@ -631,3 +632,92 @@ async def test_text_kind_fk_opens_picker_and_shows_chosen_name() -> None: role_row = next(r for r in rows if r.field == "role") assert role_row.old_display == "Top of Rack Switch" assert role_row.new_display == "Leaf Switch" + + +def _tags_model(*, editable_tags: bool = False) -> CommandModel: + fields: dict[str, FieldShape] = {"name": FieldShape(primitive=PrimitiveType.STRING)} + if editable_tags: + fields["tags"] = FieldShape(primitive=PrimitiveType.ARRAY) + devices = Resource( + name="devices", + list_op=Operation(operation_id="d_list", http_method="GET", path="/api/dcim/devices/"), + get_op=Operation(operation_id="d_get", http_method="GET", path="/api/dcim/devices/{id}/"), + update_op=Operation( + operation_id="d_update", + http_method="PATCH", + path="/api/dcim/devices/{id}/", + request_body=RequestBodyShape(top_level="object", fields=fields), + ), + ) + return CommandModel( + info_title="t", + info_version="1", + schema_hash="h", + tags={"dcim": Tag(name="dcim", resources={"devices": devices})}, + ) + + +_TAGS_RECORD = { + "id": 7, + "name": "sw1", + "tags": [ + {"display": "prod", "name": "prod", "slug": "prod", "color": "ff0000"}, + {"display": "edge", "name": "edge", "slug": "edge", "color": "00ff00"}, + ], +} + + +class _TagsDetailApp(App[None]): + def __init__(self, *, object_colors: bool = False, editable_tags: bool = False) -> None: + super().__init__() + self.client = _SpyClient() + self.object_colors = object_colors + self._editable_tags = editable_tags + + def compose(self) -> ComposeResult: + yield Static("") + + async def on_mount(self) -> None: + model = _tags_model(editable_tags=self._editable_tags) + resource = model.tags["dcim"].resources["devices"] + await self.push_screen( + DetailScreen(model, self.client, "dcim", "devices", resource, dict(_TAGS_RECORD)) + ) + + +def _tags_cell(screen: DetailScreen) -> Any: + row = next(i for i, r in enumerate(screen._rows) if r.name == "tags") + return screen._table.get_cell_at(Coordinate(row, 1)) + + +@pytest.mark.asyncio +async def test_readonly_tag_list_renders_joined_not_raw_repr() -> None: + app = _TagsDetailApp() + async with app.run_test() as pilot: + await pilot.pause() + cell = _tags_cell(_detail(app)) + assert str(cell) == "prod, edge" + assert "{" not in str(cell) + + +@pytest.mark.asyncio +async def test_editable_tag_list_renders_joined_not_raw_repr() -> None: + app = _TagsDetailApp(editable_tags=True) + async with app.run_test() as pilot: + await pilot.pause() + cell = _tags_cell(_detail(app)) + assert str(cell) == "prod, edge" + assert "{" not in str(cell) + + +@pytest.mark.asyncio +async def test_tag_list_colored_when_object_colors_enabled() -> None: + app = _TagsDetailApp(object_colors=True) + async with app.run_test() as pilot: + await pilot.pause() + cell = _tags_cell(_detail(app)) + assert isinstance(cell, Text) + assert cell.plain == "prod, edge" + styles = {str(span.style) for span in cell.spans} + assert "#ff0000" in styles + assert "#00ff00" in styles diff --git a/tests/tui/test_edit_form.py b/tests/tui/test_edit_form.py index 01dab8e..74b5ddc 100644 --- a/tests/tui/test_edit_form.py +++ b/tests/tui/test_edit_form.py @@ -4,7 +4,7 @@ import pytest from textual.app import App, ComposeResult -from textual.widgets import Button, Input, ListView, Select, Static, Switch +from textual.widgets import Button, Input, ListView, Select, SelectionList, Static, Switch from nsc.http.errors import NetBoxAPIError from nsc.model.command_model import ( @@ -16,7 +16,9 @@ Resource, Tag, ) -from nsc.tui.forms import SET_NULL, compute_patch +from nsc.savedfilters.custom_fields import CustomFieldDef +from nsc.savedfilters.tags import TagDef +from nsc.tui.forms import SET_NULL, compute_patch, tags_payload from nsc.tui.screens.edit_form import EditForm from nsc.tui.screens.record_picker import RecordPicker from nsc.tui.widgets.diff import DiffModal @@ -606,3 +608,180 @@ async def test_go_back_without_changes_dismisses_immediately() -> None: screen.action_go_back() await pilot.pause() assert app.screen is not screen + + +# --- #134: per-field custom fields + tags multi-select in single edit --- + +_CF_DEFS = {"tier": CustomFieldDef("tier", "Tier", type="select", choices=("gold", "silver"))} +_TAGS = (TagDef("Prod", "prod", "ff0000"), TagDef("Edge", "edge", None)) + + +def _cf_model() -> CommandModel: + update_op = Operation( + operation_id="dcim_devices_partial_update", + http_method="PATCH", + path="/api/dcim/devices/{id}/", + request_body=RequestBodyShape( + top_level="object", + fields={ + "name": FieldShape(primitive=PrimitiveType.STRING), + "custom_fields": FieldShape(primitive=PrimitiveType.OBJECT), + "tags": FieldShape(primitive=PrimitiveType.ARRAY), + }, + ), + ) + devices = Resource(name="devices", update_op=update_op) + return CommandModel( + info_title="t", + info_version="1", + schema_hash="h", + tags={"dcim": Tag(name="dcim", resources={"devices": devices})}, + ) + + +class _CfEditApp(App[None]): + def __init__(self, client: _SpyClient, record: dict[str, Any]) -> None: + super().__init__() + self._client = client + self._record = record + + def compose(self) -> ComposeResult: + yield Static("") + + async def on_mount(self) -> None: + model = _cf_model() + op = model.tags["dcim"].resources["devices"].update_op + assert op is not None + await self.push_screen( + EditForm( + model, + self._client, + "dcim", + "devices", + op, + self._record, + custom_field_defs=_CF_DEFS, + available_tags=_TAGS, + ) + ) + + +def _cf_edit_screen(app: _CfEditApp) -> EditForm: + screen = app.screen + assert isinstance(screen, EditForm) + return screen + + +@pytest.mark.asyncio +async def test_edit_expands_custom_field_widget() -> None: + record = {"id": 5, "name": "sw1", "custom_fields": {"tier": "silver"}, "tags": []} + app = _CfEditApp(_SpyClient([]), record) + async with app.run_test(size=(120, 60)) as pilot: + await pilot.pause() + screen = _cf_edit_screen(app) + tier = screen.query_one("#field-custom_fields-tier", Select) + assert tier.value == "silver" + assert isinstance(screen.query_one("#field-tags"), SelectionList) + + +@pytest.mark.asyncio +async def test_edit_saves_custom_field_nested() -> None: + client = _SpyClient([]) + record = {"id": 5, "name": "sw1", "custom_fields": {"tier": "silver"}, "tags": []} + app = _CfEditApp(client, record) + async with app.run_test(size=(120, 60)) as pilot: + await pilot.pause() + screen = _cf_edit_screen(app) + screen.staged["custom_fields.tier"] = "gold" + screen.action_save() + await pilot.pause() + await pilot.press("y") + await pilot.pause() + assert len(client.patch_calls) == 1 + assert client.patch_calls[0]["json"] == {"custom_fields": {"tier": "gold"}} + + +@pytest.mark.asyncio +async def test_edit_saves_tags_as_name_slug_list() -> None: + client = _SpyClient([]) + record = { + "id": 5, + "name": "sw1", + "custom_fields": {}, + "tags": [{"slug": "prod", "name": "Prod"}], + } + app = _CfEditApp(client, record) + async with app.run_test(size=(120, 60)) as pilot: + await pilot.pause() + screen = _cf_edit_screen(app) + screen.staged["tags"] = tags_payload(("prod", "edge"), _TAGS) + screen.action_save() + await pilot.pause() + await pilot.press("y") + await pilot.pause() + assert len(client.patch_calls) == 1 + assert client.patch_calls[0]["json"] == { + "tags": [{"name": "Prod", "slug": "prod"}, {"name": "Edge", "slug": "edge"}] + } + + +@pytest.mark.asyncio +async def test_edit_unchanged_tag_selection_stages_nothing() -> None: + record = { + "id": 5, + "name": "sw1", + "custom_fields": {}, + "tags": [{"slug": "prod", "name": "Prod"}], + } + app = _CfEditApp(_SpyClient([]), record) + async with app.run_test(size=(120, 60)) as pilot: + await pilot.pause() + screen = _cf_edit_screen(app) + # The pre-seeded selection firing on mount must not stage a tags change. + assert "tags" not in screen.staged + + +_CF_DEFS_MS = { + "envs": CustomFieldDef("envs", "Envs", type="multiselect", choices=("dev", "prod", "qa")), +} + + +class _CfMsEditApp(App[None]): + def __init__(self, client: _SpyClient, record: dict[str, Any]) -> None: + super().__init__() + self._client = client + self._record = record + + def compose(self) -> ComposeResult: + yield Static("") + + async def on_mount(self) -> None: + model = _cf_model() + op = model.tags["dcim"].resources["devices"].update_op + assert op is not None + await self.push_screen( + EditForm( + model, + self._client, + "dcim", + "devices", + op, + self._record, + custom_field_defs=_CF_DEFS_MS, + available_tags=_TAGS, + ) + ) + + +@pytest.mark.asyncio +async def test_edit_multiselect_custom_field_preseeds_current_values() -> None: + record = {"id": 5, "name": "sw1", "custom_fields": {"envs": ["dev", "qa"]}, "tags": []} + app = _CfMsEditApp(_SpyClient([]), record) + async with app.run_test(size=(120, 60)) as pilot: + await pilot.pause() + screen = app.screen + assert isinstance(screen, EditForm) + widget = screen.query_one("#field-custom_fields-envs", SelectionList) + assert set(widget.selected) == {"dev", "qa"} + # Pre-seeded selection that is unchanged stages nothing. + assert "custom_fields.envs" not in screen.staged diff --git a/tests/tui/test_forms.py b/tests/tui/test_forms.py index 6463a60..ff7ed9e 100644 --- a/tests/tui/test_forms.py +++ b/tests/tui/test_forms.py @@ -130,3 +130,88 @@ def test_diff_rows_new_display_override_used_for_new_value() -> None: def test_diff_rows_override_absent_falls_back_to_str() -> None: rows = diff_rows({"status": "active"}, {"status": "offline"}, (), {"role": "x"}) assert rows == [DiffRow(field="status", old_display="active", new_display="offline")] + + +# --- #134: custom-field expansion, tags multi-select, payload nesting --- + +from nsc.savedfilters.custom_fields import CustomFieldDef # noqa: E402 +from nsc.savedfilters.tags import TagDef # noqa: E402 +from nsc.tui.forms import ( # noqa: E402 + custom_field_widget, + expand_custom_fields, + flatten_custom_fields, + nest_custom_fields, + tag_slugs, + tags_payload, + tags_widget_spec, +) + + +def test_custom_field_widget_maps_types() -> None: + assert custom_field_widget(CustomFieldDef("a", "A", type="text")).kind == "text" + assert custom_field_widget(CustomFieldDef("a", "A", type="integer")).kind == "number" + dec = custom_field_widget(CustomFieldDef("a", "A", type="decimal")) + assert dec.kind == "number" and dec.is_float is True + assert custom_field_widget(CustomFieldDef("a", "A", type="boolean")).kind == "switch" + sel = custom_field_widget(CustomFieldDef("a", "A", type="select", choices=("x", "y"))) + assert sel.kind == "select" and sel.choices == ("x", "y") + ms = custom_field_widget(CustomFieldDef("a", "A", type="multiselect", choices=("x", "y"))) + assert ms.kind == "multi_select" + + +def test_custom_field_widget_names_are_dotted() -> None: + spec = custom_field_widget(CustomFieldDef("rack_role", "Rack Role")) + assert spec.name == "custom_fields.rack_role" + + +def test_expand_custom_fields_one_spec_each() -> None: + defs = { + "a": CustomFieldDef("a", "A", type="text"), + "b": CustomFieldDef("b", "B", type="integer"), + } + specs = expand_custom_fields(defs) + assert [s.name for s in specs] == ["custom_fields.a", "custom_fields.b"] + + +def test_tags_widget_spec_carries_options_and_selected() -> None: + tags = (TagDef("Prod", "prod", "ff0000"), TagDef("Edge", "edge", None)) + spec = tags_widget_spec("tags", tags, ("prod",)) + assert spec.kind == "multi_select" + assert spec.options == (("Prod", "prod"), ("Edge", "edge")) + assert spec.selected == ("prod",) + + +def test_tag_slugs_extracts_from_record_list() -> None: + value = [{"slug": "prod", "name": "Prod"}, {"name": "edge"}] + assert tag_slugs(value) == ("prod", "edge") + assert tag_slugs(None) == () + + +def test_tags_payload_builds_name_slug_objects() -> None: + tags = (TagDef("Prod", "prod", None), TagDef("Edge", "edge", None)) + assert tags_payload(("prod",), tags) == [{"name": "Prod", "slug": "prod"}] + # Unknown slug falls back to itself. + assert tags_payload(("ghost",), tags) == [{"name": "ghost", "slug": "ghost"}] + + +def test_flatten_custom_fields_adds_dotted_keys() -> None: + rec = {"name": "x", "custom_fields": {"tier": "gold", "n": 3}} + flat = flatten_custom_fields(rec) + assert flat["custom_fields.tier"] == "gold" + assert flat["custom_fields.n"] == 3 + # Original is not mutated. + assert "custom_fields.tier" not in rec + + +def test_flatten_custom_fields_noop_without_cf() -> None: + assert flatten_custom_fields({"name": "x"}) == {"name": "x"} + + +def test_nest_custom_fields_folds_dotted_keys() -> None: + patch = {"name": "x", "custom_fields.tier": "silver", "custom_fields.n": None} + nested = nest_custom_fields(patch) + assert nested == {"name": "x", "custom_fields": {"tier": "silver", "n": None}} + + +def test_nest_custom_fields_passthrough_without_cf() -> None: + assert nest_custom_fields({"name": "x"}) == {"name": "x"} diff --git a/tests/tui/test_list_screen.py b/tests/tui/test_list_screen.py index 0c45d63..abffc81 100644 --- a/tests/tui/test_list_screen.py +++ b/tests/tui/test_list_screen.py @@ -16,6 +16,7 @@ Resource, Tag, ) +from nsc.savedfilters.custom_fields import CustomFieldDef from nsc.tui.app import NscTuiApp from nsc.tui.screens.bulk_edit_form import BulkEditForm from nsc.tui.screens.columns import ColumnChooserScreen @@ -694,3 +695,67 @@ async def test_bulk_edit_reloads_list_after_form_dismisses() -> None: await app.workers.wait_for_complete() assert isinstance(app.screen, ListScreen) assert len(client.calls) > load_count + + +def _cf_model() -> CommandModel: + op = Operation( + operation_id="devices_list", + http_method="GET", + path="/api/dcim/devices/", + default_columns=["id", "name", "custom_fields.rack_role"], + ) + tag = Tag(name="dcim", resources={"devices": Resource(name="devices", list_op=op)}) + return CommandModel(info_title="t", info_version="1", schema_hash="h", tags={"dcim": tag}) + + +class _CfListApp(App[None]): + def __init__(self, client: _FakeClient, defs: dict[str, Any] | None) -> None: + super().__init__() + self._client = client + self._defs = defs + + def custom_field_defs_for(self, tag: str, resource: str) -> dict[str, Any] | None: + return self._defs + + def compose(self) -> ComposeResult: + yield Static("") + + async def on_mount(self) -> None: + model = _cf_model() + op = model.tags["dcim"].resources["devices"].list_op + assert op is not None + await self.push_screen(ListScreen(model, self._client, "dcim", "devices", op)) + + +@pytest.mark.asyncio +async def test_list_header_shows_custom_field_label_but_keeps_raw_key() -> None: + client = _FakeClient([{"id": 1, "name": "sw1", "custom_fields": {"rack_role": "gold"}}]) + defs = {"rack_role": CustomFieldDef(name="rack_role", label="Rack Role")} + app = _CfListApp(client, defs) + async with app.run_test() as pilot: + await pilot.pause() + await app.workers.wait_for_complete() + await pilot.pause() + screen = app.screen + assert isinstance(screen, ListScreen) + table = screen.query_one(DataTable) + headers = [str(col.label) for col in table.ordered_columns] + assert "Rack Role" in headers + assert "custom_fields.rack_role" not in headers + # The selector key is preserved for cell lookup / persistence. + assert "custom_fields.rack_role" in screen._columns + + +@pytest.mark.asyncio +async def test_list_header_falls_back_to_raw_key_when_defs_unavailable() -> None: + client = _FakeClient([{"id": 1, "name": "sw1", "custom_fields": {"rack_role": "gold"}}]) + app = _CfListApp(client, None) + async with app.run_test() as pilot: + await pilot.pause() + await app.workers.wait_for_complete() + await pilot.pause() + screen = app.screen + assert isinstance(screen, ListScreen) + table = screen.query_one(DataTable) + headers = [str(col.label) for col in table.ordered_columns] + assert "custom_fields.rack_role" in headers