diff --git a/.github/workflows/lint-tests.yaml b/.github/workflows/lint-tests.yaml index 627cdac7..f39f3ad2 100644 --- a/.github/workflows/lint-tests.yaml +++ b/.github/workflows/lint-tests.yaml @@ -32,13 +32,25 @@ jobs: pip install .[test] - name: Run ruff run: ruff check + - name: Build in-app model documentation + run: | + mkdocs build -f mkdocs.models.yml + test -f netbox_custom_objects/static/docs/models/netbox_custom_objects/customobjecttype/index.html tests: + name: tests (${{ matrix.name }}) runs-on: ubuntu-latest timeout-minutes: 20 strategy: fail-fast: false matrix: - netbox-ref: [ "main", "feature" ] + include: + - netbox-ref: "main" + name: "main" + - netbox-ref: "feature" + name: "feature" + - netbox-ref: "main" + name: "main, branching" + with-branching: true services: redis: image: redis @@ -80,13 +92,38 @@ jobs: pip install . pip install .[test] - name: Install dependencies & configure plugin + if: ${{ !matrix.with-branching }} working-directory: netbox run: | ln -s $(pwd)/../netbox-custom-objects/testing/configuration.py netbox/netbox/configuration.py - python -m pip install --upgrade pip - pip install -r requirements.txt -U + pip install -r requirements.txt + - name: Install dependencies & configure plugin (with branching) + if: ${{ matrix.with-branching == true }} + working-directory: netbox + run: | + ln -s $(pwd)/../netbox-custom-objects/testing/configuration_branching.py netbox/netbox/configuration.py + python -m pip install --upgrade pip + pip install -r requirements.txt + pip install 'netboxlabs-netbox-branching>=1.0.4,<2.0.0' + - name: Tune PostgreSQL for branching test performance + if: ${{ matrix.with-branching == true }} + run: | + sudo apt-get install -y -q postgresql-client + PGPASSWORD=netbox psql -h localhost -U netbox \ + -c "ALTER SYSTEM SET checkpoint_timeout = '30min'" \ + -c "ALTER SYSTEM SET max_wal_size = '2GB'" \ + -c "ALTER SYSTEM SET synchronous_commit = off" \ + -c "ALTER SYSTEM SET lock_timeout = '60s'" \ + -c "SELECT pg_reload_conf()" - name: Run tests + if: ${{ matrix.with-branching != true }} + working-directory: netbox + run: | + python netbox/manage.py test netbox_custom_objects.tests --keepdb --verbosity=2 + + - name: Run tests (with branching) + if: ${{ matrix.with-branching == true }} working-directory: netbox run: | - python netbox/manage.py test netbox_custom_objects.tests --keepdb + python netbox/manage.py test netbox_custom_objects.tests.test_branching --keepdb --verbosity=2 diff --git a/.github/workflows/release.yaml b/.github/workflows/release.yaml index f7de66f1..630ccc30 100644 --- a/.github/workflows/release.yaml +++ b/.github/workflows/release.yaml @@ -21,6 +21,13 @@ jobs: - name: Install pypa/build run: | python3 -m pip install build + - name: Install mkdocs + run: | + python3 -m pip install 'mkdocs>=1.6,<2' 'mkdocs-material>=9.7,<10' + - name: Build in-app model documentation + run: | + mkdocs build -f mkdocs.models.yml + test -f netbox_custom_objects/static/docs/models/netbox_custom_objects/customobjecttype/index.html - name: Build distribution package run: | python3 -m build diff --git a/.gitignore b/.gitignore index 5029ad2d..7fe3dfb2 100644 --- a/.gitignore +++ b/.gitignore @@ -3,3 +3,4 @@ .claude/settings.local.json .idea/ .DS_Store +netbox_custom_objects/static/ diff --git a/AGENTS.md b/AGENTS.md index 9171d3d3..58ccc4fa 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -157,6 +157,7 @@ There is no Justfile/Makefile in this repo; commands are raw. Run tests inside a | `python netbox/manage.py makemigrations netbox_custom_objects` | Generate Django migrations after model changes | | `python netbox/manage.py migrate` | Apply migrations | | `python netbox/manage.py runserver` | Start NetBox locally with the plugin loaded | +| `mkdocs build -f mkdocs.models.yml` | Build the in-app "Help" model documentation pages into `netbox_custom_objects/static/docs/models/` (run before `collectstatic`; the release workflow does this automatically before packaging) | ## Development @@ -218,6 +219,7 @@ GitHub Actions workflows in `.github/workflows/`: 3. Wire up the rest of the surface area: `filtersets.py`, `forms.py`, `tables.py`, `api/serializers.py`, `api/urls.py`, `urls.py`, `navigation.py`, and a template under `templates/netbox_custom_objects/`. 4. Register a `SearchIndex` in `search.py` if the model should appear in NetBox's global search. 5. Add tests covering model logic, API, filtersets, and views. +6. If the model is a `NetBoxModel` (it has a "Help" link on its edit page via `docs_url`), add `docs_models/netbox_custom_objects/.md` and a nav entry in `mkdocs.models.yml`, or the Help link will 404. ### Add a REST API endpoint diff --git a/COMPATIBILITY.md b/COMPATIBILITY.md index 67087554..a0f45648 100644 --- a/COMPATIBILITY.md +++ b/COMPATIBILITY.md @@ -2,6 +2,7 @@ | Release | Minimum NetBox Version | Maximum NetBox Version | |---------|------------------------|------------------------| +| unreleased | 4.5.2 | 4.7.x | | 0.6.x | 4.5.2 | 4.6.x | | 0.5.2 | 4.5.2 | 4.6.x | | 0.5.1 | 4.5.2 | 4.6.x | diff --git a/README.md b/README.md index ee4d0460..fd289113 100644 --- a/README.md +++ b/README.md @@ -21,10 +21,11 @@ PLUGINS = [ ] ``` -3. Run NetBox migrations: +3. Run NetBox migrations and collect static files: ``` $ ./manage.py migrate +$ ./manage.py collectstatic ``` 4. Restart NetBox diff --git a/docs/index.md b/docs/index.md index eb175fac..cdec2b83 100644 --- a/docs/index.md +++ b/docs/index.md @@ -160,6 +160,42 @@ Notes: - A field must be an object field with both the **name and target model** above; a mis-named or mis-pointed field is simply ignored. - If a type defines **none** of these fields, aggregation is skipped entirely and its rendered context is just its Local Context Data. This is a deliberate difference from Devices/VMs: **global (unassigned) ConfigContexts are not applied** to such a type — enabling config context support alone never silently pulls in every global context. Add at least one convention field (e.g. `site`) to opt the type into source aggregation; global contexts then apply too (as they do for any object with a dimension). +### Jinja Config Templates + +!!! note + Requires NetBox 4.7 or later. On earlier NetBox versions, `custom_objects` is simply unavailable in config templates and nothing else changes — no error, no crash. A `DEBUG`-level log message noting this is emitted at plugin startup; enable debug logging if you need to confirm why `custom_objects` isn't resolving. + +Custom Objects can be referenced directly from NetBox [config templates](https://netboxlabs.com/docs/netbox/models/extras/configtemplate/), so device configuration can pull in data modelled with Custom Object Types (e.g. OSPF interface parameters, BGP peer groups, MPLS label ranges) alongside built-in NetBox models. + +Two equivalent access patterns are available, both resolving the Custom Object Type by its **internal name** at render time (so templates keep working even if the type's slug or internal table ID changes): + +Attribute-style, via the `custom_objects` context variable: + +```jinja2 +{% for iface in custom_objects.ospf_interface.filter(device=device) %} +interface {{ iface.name }} + ip ospf area {{ iface.area }} +{% endfor %} +``` + +Filter syntax, via the `custom_objects` Jinja filter: + +```jinja2 +{% for iface in 'ospf_interface' | custom_objects %} +interface {{ iface.name }} +{% endfor %} +``` + +Notes: + +- The attribute-style form returns the model's manager (`.filter(...)`, `.all()`, etc.); the filter form returns a queryset of all instances of that type. +- An unknown type name is handled the same way by both forms: a warning is logged, and the reference resolves to an empty, chainable stand-in — further calls like `.filter(...)` or `.all()` continue to render no rows rather than raising. Check the type's internal name (shown on its detail page) if a template renders no data. +- A type's internal name may begin with a digit (e.g. `123foo`), which isn't valid Jinja dot-notation. Use bracket notation with the attribute-style form instead: + +```jinja2 +{% for obj in custom_objects['123foo'].filter(device=device) %} +``` + ### Deletions #### Deleting Custom Object Types diff --git a/docs/installation.md b/docs/installation.md index b79cefc4..87f02351 100644 --- a/docs/installation.md +++ b/docs/installation.md @@ -27,12 +27,13 @@ PLUGINS = [ ] ``` -### 3. Run Database Migrations +### 3. Run Database Migrations and Collect Static Files -Apply the plugin's database migrations: +Apply the plugin's database migrations and collect its static files: ``` ./manage.py migrate +./manage.py collectstatic ``` ### 4. Restart NetBox diff --git a/docs_models/netbox_custom_objects/customobjecttype.md b/docs_models/netbox_custom_objects/customobjecttype.md new file mode 100644 index 00000000..d51a11c8 --- /dev/null +++ b/docs_models/netbox_custom_objects/customobjecttype.md @@ -0,0 +1,49 @@ +# Custom Object Types + +A Custom Object Type defines a new object type in NetBox — the equivalent of a model in NetBox plugin terminology. Each Custom Object Type generates its own database table, list and detail views, REST API endpoints, and an entry in the left navigation pane. See the [Custom Objects documentation](https://github.com/netboxlabs/netbox-custom-objects/blob/main/docs/index.md) for a full walkthrough, including how Custom Object Type Fields are added to a type. + +## Fields + +### Internal Name + +A unique, lowercased, URL-friendly internal name, e.g. `vendor_policy`. Only lowercase alphanumeric characters and underscores are permitted; names may not start or end with an underscore, and double underscores are not allowed. + +### Display Name (Singular) + +The human-friendly singular name shown throughout the UI, e.g. `Vendor Policy`. Defaults to the internal name if left blank. + +### Display Name (Plural) + +The human-friendly plural name shown throughout the UI, e.g. `Vendor Policies`. Defaults to the internal name if left blank. + +### URL Path/Slug + +A unique, plural, URL-friendly identifier used as a URL component for this type's list and detail views, e.g. `vendor-policies`. + +### Display Expression + +An optional Jinja2 template used to render the display name of individual objects of this type, e.g. `{{ name }} - {{ manufacturer }}`. Reference field values by name; undefined fields resolve to an empty string. If left blank, the field marked as the type's primary field is used instead. + +### Group Name + +An optional label used to group similar Custom Object Types together in the navigation menu. + +### Version + +An optional [PEP 440](https://peps.python.org/pep-0440/) version string, e.g. `1.0.0`. Used when managing schemas across environments with the [portable schema](https://github.com/netboxlabs/netbox-custom-objects/blob/main/docs/portable-schema.md) feature. + +### Description + +A short, optional description of this Custom Object Type. + +### Config Context Support + +Whether objects of this type support NetBox's [config context](https://netboxlabs.com/docs/netbox/models/extras/configcontext/) feature, gaining a Local Context Data field and a Config Context tab. This can only be set when the type is created — it adds a column to the type's table, so it cannot be toggled afterward. See the [Config Context](https://github.com/netboxlabs/netbox-custom-objects/blob/main/docs/index.md#config-context) section of the documentation for details. + +### Comments + +Free-form text for any additional notes about this Custom Object Type. + +### Tags + +NetBox tags applied to this Custom Object Type. diff --git a/mkdocs.models.yml b/mkdocs.models.yml new file mode 100644 index 00000000..c14d61c3 --- /dev/null +++ b/mkdocs.models.yml @@ -0,0 +1,9 @@ +site_name: NetBox Custom Objects Model Documentation +docs_dir: docs_models +site_dir: netbox_custom_objects/static/docs/models +theme: + name: mkdocs +plugins: [] +nav: + - netbox_custom_objects: + - Custom Object Type: netbox_custom_objects/customobjecttype.md diff --git a/netbox_custom_objects/__init__.py b/netbox_custom_objects/__init__.py index c5708e0a..429bf570 100644 --- a/netbox_custom_objects/__init__.py +++ b/netbox_custom_objects/__init__.py @@ -296,7 +296,7 @@ class CustomObjectsPluginConfig(PluginConfig): base_url = "custom-objects" # Remember to update COMPATIBILITY.md when modifying the minimum/maximum supported NetBox versions. min_version = "4.5.2" - max_version = "4.6.99" + max_version = "4.7.99" default_settings = { # The maximum number of Custom Object Types that may be created 'max_custom_object_types': 50, @@ -312,6 +312,10 @@ class CustomObjectsPluginConfig(PluginConfig): # so the swallowed exception isn't invisible outside the logs. _register_tabs_error = None template_extensions = "template_content.template_extensions" + # Registers the custom_objects Jinja filter (jinja_env.filters). Requires NetBox + # 4.7+; on older NetBox this attribute is simply never read by core (see ready() + # for the startup log message covering that case). + jinja_filters = "jinja_env.filters" # Resolves dynamic CO models (table{n}model) to on-the-fly serializers — # they have no importable path at the conventional location. serializer_resolver = "api.serializers.serializer_resolver" @@ -427,6 +431,26 @@ def _call_super_ready_once(self): super().ready() _super_ready_called = True + # On NetBox < 4.7 the jinja_filters resource and get_jinja_context() hook + # don't exist, so super().ready() never calls _load_resource('jinja_filters') + # and get_jinja_context() is never invoked by RenderTemplateMixin.get_context(). + # This is every currently-supported NetBox version (4.7 isn't released yet), so + # log at DEBUG rather than INFO: it's an explanation to reach for when actively + # troubleshooting why 'custom_objects' isn't resolving, not a startup notice + # every install should see by default. + from netbox.registry import registry + if 'custom_objects' not in registry.get('plugins', {}).get('jinja_filters', {}): + logger.debug( + "NetBox Jinja config template hooks (jinja_filters / get_jinja_context) " + "are not available in this version of NetBox. The 'custom_objects' filter " + "and context variable will not be active in config templates. Upgrade to " + "NetBox 4.7+ to enable this feature." + ) + + def get_jinja_context(self): + from netbox_custom_objects.jinja_env import CustomObjectsNamespace + return {'custom_objects': CustomObjectsNamespace()} + def ready(self): # Install the thread-safe apps.clear_cache wrapper before any dynamic # model is registered (must happen exactly once, before get_model() runs). diff --git a/netbox_custom_objects/api/serializers.py b/netbox_custom_objects/api/serializers.py index 627f5c54..9f41e16e 100644 --- a/netbox_custom_objects/api/serializers.py +++ b/netbox_custom_objects/api/serializers.py @@ -335,6 +335,7 @@ class Meta: "slug", "version", "group_name", + "display_expression", "description", "config_context_enabled", "tags", diff --git a/netbox_custom_objects/branching.py b/netbox_custom_objects/branching.py index 667dd1b1..f24c6eb2 100644 --- a/netbox_custom_objects/branching.py +++ b/netbox_custom_objects/branching.py @@ -37,7 +37,7 @@ def objectchange_field_migrator(model, data): return resolve(data) -def _collect_co_refs(model_class, data): +def _collect_co_refs(model_class, data, model_label=None): """Return ``(app.model, pk)`` refs from CO-specific shapes in *data*. Covers: @@ -50,6 +50,12 @@ def _collect_co_refs(model_class, data): CREATEs. Pulled from the model class's ``_field_objects`` plus the polymorphic ``POLY_M2M_SIDECAR_KEY`` (which carries field PKs in the ObjectChange payload even when ``_field_objects`` isn't available). + + ``model_label`` — the ``"{app_label}.{model_name}"`` key from + ``CollapsedChange.key``. Provided when ``model_class`` is ``None`` + (dynamic CO models that aren't yet registered in ``apps.all_models`` + during the squash dep-graph phase). Used as the ref label for the + self-referential M2M fallback (see below). """ from .constants import APP_LABEL from .models import POLY_M2M_SIDECAR_KEY @@ -58,7 +64,11 @@ def _collect_co_refs(model_class, data): if not data: return refs - for field in model_class._meta.local_many_to_many: + # Primary pass: walk M2M fields declared on the model class. + m2m_field_names = set() + meta = getattr(model_class, '_meta', None) + for field in getattr(meta, 'local_many_to_many', ()): + m2m_field_names.add(field.name) values = data.get(field.name) if not values: continue @@ -68,6 +78,22 @@ def _collect_co_refs(model_class, data): if isinstance(pk, int): refs.add((label, pk)) + # Fallback for dynamically-generated CO models whose class isn't yet + # registered in apps.all_models at dep-graph time (model_class is None). + # The only CO field type that stores a plain list of integers in + # postchange_data is a direct (non-polymorphic) M2M. When such a field + # is self-referential the refs point to the same model label, so we can + # add the dep edge without knowing the concrete model class. + # Cross-COT M2M would produce a wrong label, but those refs won't appear + # in creates_map for the source model and are silently ignored. + if model_label and model_label.startswith(f'{APP_LABEL}.'): + for key, value in data.items(): + if key in (POLY_M2M_SIDECAR_KEY, 'tags') or key in m2m_field_names: + continue + if isinstance(value, list) and value and all(isinstance(v, int) for v in value): + for pk in value: + refs.add((model_label, pk)) + field_label = f'{APP_LABEL}.customobjecttypefield' for fo in (getattr(model_class, '_field_objects', None) or {}).values(): cotf = fo.get('field') if isinstance(fo, dict) else None @@ -110,26 +136,39 @@ def add_custom_object_dependencies(sender, collapsed_changes, **kwargs): for cc in collapsed_changes.values(): meta = getattr(cc.model_class, '_meta', None) - if meta is None or meta.app_label != APP_LABEL: + # Detect CO models even when model_class is None (dynamically-generated + # CO models aren't registered in apps.all_models until their COT CREATE + # is applied, so ContentType.model_class() returns None during the + # squash dep-graph phase — the meta is None guard would silently skip + # them). Fall back to inspecting cc.key[0] which is always set. + model_label = cc.key[0] if isinstance(cc.key, tuple) else None + is_co_model = ( + meta is not None and meta.app_label == APP_LABEL + ) or ( + meta is None + and model_label is not None + and model_label.startswith(f'{APP_LABEL}.') + ) + if not is_co_model: continue action = cc.final_action.value if cc.final_action else None if action == 'update': - for ref in _collect_co_refs(cc.model_class, cc.prechange_data): + for ref in _collect_co_refs(cc.model_class, cc.prechange_data, model_label=model_label): if ref in deletes_map: deletes_map[ref].depends_on.add(cc.key) cc.depended_by.add(ref) - for ref in _collect_co_refs(cc.model_class, cc.postchange_data): + for ref in _collect_co_refs(cc.model_class, cc.postchange_data, model_label=model_label): if ref in creates_map: cc.depends_on.add(ref) creates_map[ref].depended_by.add(cc.key) elif action == 'create': - for ref in _collect_co_refs(cc.model_class, cc.postchange_data): + for ref in _collect_co_refs(cc.model_class, cc.postchange_data, model_label=model_label): if ref != cc.key and ref in creates_map: cc.depends_on.add(ref) creates_map[ref].depended_by.add(cc.key) elif action == 'delete': - for ref in _collect_co_refs(cc.model_class, cc.prechange_data): + for ref in _collect_co_refs(cc.model_class, cc.prechange_data, model_label=model_label): if ref != cc.key and ref in deletes_map: deletes_map[ref].depends_on.add(cc.key) cc.depended_by.add(ref) diff --git a/netbox_custom_objects/checks.py b/netbox_custom_objects/checks.py index 4c89ace9..cfc810d4 100644 --- a/netbox_custom_objects/checks.py +++ b/netbox_custom_objects/checks.py @@ -26,6 +26,19 @@ REQUIRED_NETBOX_VERSION_FOR_BRANCHING = '4.6.2' REQUIRED_BRANCHING_VERSION = '1.0.4' +# The package is published under two distribution names depending on the +# release channel; try both before concluding the version is unknowable. +_BRANCHING_DIST_NAMES = ('netboxlabs-netbox-branching', 'netbox-branching') + + +def _get_branching_version(): + for dist_name in _BRANCHING_DIST_NAMES: + try: + return _pkg_version(dist_name) + except PackageNotFoundError: + continue + raise PackageNotFoundError('netbox-branching') + @register() def check_branching_compatibility(app_configs, **kwargs): @@ -49,12 +62,12 @@ def check_branching_compatibility(app_configs, **kwargs): pass # settings.RELEASE missing/unparseable — other checks surface it try: - branching_version = Version(_pkg_version('netbox-branching')) + branching_version = Version(_get_branching_version()) if branching_version < Version(REQUIRED_BRANCHING_VERSION): errors.append(Error( f'netbox-custom-objects requires netbox-branching >= ' f'{REQUIRED_BRANCHING_VERSION} (detected {branching_version}).', - hint=f'Upgrade with: pip install "netbox-branching>={REQUIRED_BRANCHING_VERSION}"', + hint=f'Upgrade with: pip install "netboxlabs-netbox-branching>={REQUIRED_BRANCHING_VERSION}"', id='netbox_custom_objects.E002', )) except PackageNotFoundError: @@ -66,8 +79,8 @@ def check_branching_compatibility(app_configs, **kwargs): 'netbox-branching is installed but its version could not be ' f'determined, so the >= {REQUIRED_BRANCHING_VERSION} requirement ' 'cannot be verified.', - hint='If using an editable install, ensure its dist-info metadata ' - 'is present (reinstall with `pip install -e`).', + hint='If using an editable install of netboxlabs-netbox-branching, ' + 'reinstall with: pip install -e .', id='netbox_custom_objects.W001', )) except InvalidVersion: diff --git a/netbox_custom_objects/field_types.py b/netbox_custom_objects/field_types.py index a46ad0a2..3604e28e 100644 --- a/netbox_custom_objects/field_types.py +++ b/netbox_custom_objects/field_types.py @@ -1088,6 +1088,12 @@ def remove_polymorphic_object_columns(self, field_instance, model, schema_editor ct_field_name = f"{field_instance.name}_content_type" oid_field_name = f"{field_instance.name}_object_id" + # Flush deferred FK trigger events before ALTER TABLE; PostgreSQL rejects + # column removal with "pending trigger events" when a row deletion (from + # the revert path) has queued events on a DEFERRABLE FK column. + # Also called by CustomObjectTypeField.delete() for the full removal block, + # but kept here so this method is self-contained when called directly. + schema_editor.execute('SET CONSTRAINTS ALL IMMEDIATE') try: oid_field = model._meta.get_field(oid_field_name) schema_editor.remove_field(model, oid_field) @@ -1438,6 +1444,13 @@ def get_through_model(self, field, model_string): on_delete=models.CASCADE, related_name="+", db_column="target_id", + # The real DB-level FK is added separately in create_m2m_table + # as DEFERRABLE INITIALLY DEFERRED so iterative branch merges + # (time-ordered) can insert through rows before the target CO + # exists. A new constraint created after SET CONSTRAINTS ALL + # IMMEDIATE is not affected by that earlier call; db_constraint=False + # prevents Django from creating a non-deferrable FK here. + db_constraint=False, ), } @@ -1774,6 +1787,39 @@ def create_m2m_table(self, instance, model, field_name, schema_conn=None): tables = connection.introspection.table_names(cursor) if table_name not in tables: schema_editor.create_model(through) + # Add the target FK as DEFERRABLE INITIALLY DEFERRED. + # get_through_model uses db_constraint=False so Django + # doesn't create a non-deferrable FK automatically. + # _schema_add_field calls SET CONSTRAINTS ALL IMMEDIATE + # before invoking create_m2m_table; in PostgreSQL this + # applies to the entire transaction including constraints + # created afterward. We therefore: + # 1. Add the constraint as DEFERRABLE INITIALLY DEFERRED + # 2. Immediately re-defer it by name so it is DEFERRED + # for the rest of the merge transaction + # This lets iterative branch merges (time-ordered) insert + # through rows before the referenced target CO exists; the + # FK check is deferred to transaction commit, by which + # point all CO CREATEs have been applied. + to_table = to_model._meta.db_table + to_pk = to_model._meta.pk.column + digest = hashlib.sha1(table_name.encode()).hexdigest()[:8] + fk_conname = (table_name[:44] + '_' + digest + '_target_fk').lower() + cursor.execute( + 'ALTER TABLE {tbl} ADD CONSTRAINT {con} ' + 'FOREIGN KEY (target_id) REFERENCES {ref} ({pk}) ' + 'ON DELETE CASCADE DEFERRABLE INITIALLY DEFERRED'.format( + tbl=connection.ops.quote_name(table_name), + con=connection.ops.quote_name(fk_conname), + ref=connection.ops.quote_name(to_table), + pk=connection.ops.quote_name(to_pk), + ) + ) + cursor.execute( + 'SET CONSTRAINTS {} DEFERRED'.format( + connection.ops.quote_name(fk_conname), + ) + ) def get_polymorphic_through_model(self, field_instance, source_model_string): """ @@ -1835,12 +1881,41 @@ def create_polymorphic_m2m_table(self, field_instance, model, schema_editor): ``with connection.schema_editor()`` here would flush deferred SQL prematurely on PostgreSQL. """ + from netbox_custom_objects.models import CustomObjectType # noqa: PLC0415 + source_model_string = f"{APP_LABEL}.{model.__name__}" - through = self.get_polymorphic_through_model(field_instance, source_model_string) - source_field = through._meta.get_field("source") - source_field.remote_field.model = model - source_field.related_model = model + # Serialized against CustomObjectType.get_model()'s own through-model + # reuse-or-create check (_after_model_generation runs under the same + # lock, held by its caller for the whole call). Without this, a + # concurrent reader regenerating this COT's model can observe this + # through model mid-construction here -- registered by Django's + # ModelBase metaclass inside generate_model() below, but before its + # "source" FK is repointed at `model` on the next line -- and race to + # point the registered class's FK at its OWN (different) model + # instance. Whichever thread's mutation and whichever thread's + # get_model() cache-write happen last aren't guaranteed to be the + # same thread, leaving the through's "source" FK and the cached model + # class mismatched. Confirmed live under concurrent load: + # intermittent "ValueError: Cannot query 'X': Must be 'TableYModel' + # instance." and RecursionError (issue #658). + # + # Deliberately scoped to just the build+register+repoint above -- NOT the + # table-existence probe/DDL below. A concurrent CustomObjectTypeField.save() for the + # same field also calls CustomObjectType.clear_model_cache(), which acquires this same + # lock; if the lock stayed held across schema_editor.create_model() (an uncommitted + # CREATE TABLE inside this save()'s own transaction), a second thread blocked here + # waiting for the lock -- itself stuck at the Postgres level waiting on the first + # thread's uncommitted transaction for the same physical table -- would prevent the + # first thread from ever reaching clear_model_cache() to commit. Releasing the lock + # before the DDL avoids that deadlock; the DDL itself has no equivalent staleness + # window to guard (the "source" FK is already correctly repointed by the time it runs). + with CustomObjectType._global_lock: + through = self.get_polymorphic_through_model(field_instance, source_model_string) + + source_field = through._meta.get_field("source") + source_field.remote_field.model = model + source_field.related_model = model # Probe the same schema the DDL will target. schema_editor is branch-aware # (opened via _get_schema_connection() by the caller), whereas the module-level diff --git a/netbox_custom_objects/filtersets.py b/netbox_custom_objects/filtersets.py index ba319f80..ff6eddb2 100644 --- a/netbox_custom_objects/filtersets.py +++ b/netbox_custom_objects/filtersets.py @@ -1,3 +1,4 @@ +import copy import django_filters from dataclasses import dataclass from decimal import Decimal, InvalidOperation @@ -9,7 +10,7 @@ from django.utils.dateparse import parse_date, parse_datetime from django.utils.timezone import make_aware, is_aware -from extras.choices import CustomFieldTypeChoices +from extras.choices import CustomFieldFilterLogicChoices, CustomFieldTypeChoices from netbox.filtersets import NetBoxModelFilterSet from users.models import Owner, OwnerGroup @@ -287,6 +288,15 @@ def build( CustomFieldTypeChoices.TYPE_MULTIOBJECT: FilterSpec(NonPolymorphicMultiObjectFilter), } +# Field types whose base filter's lookup_expr is driven by field.filter_logic, +# mirroring NetBox core's CustomField.to_filter(). TYPE_JSON has no core "exact" +# mode to mirror, so it's excluded and always stays icontains. +FILTER_LOGIC_AWARE_TYPES = ( + CustomFieldTypeChoices.TYPE_TEXT, + CustomFieldTypeChoices.TYPE_LONGTEXT, + CustomFieldTypeChoices.TYPE_URL, +) + class CustomObjectTypeFilterSet(NetBoxModelFilterSet): class Meta: @@ -336,6 +346,10 @@ def build_filter_for_field(field) -> dict: fields one entry is emitted per allowed related type, named ``{field.name}_{app_label}_{model}``. """ + # Mirrors NetBox core: a disabled field gets no filter at all, regardless of type. + if field.filter_logic == CustomFieldFilterLogicChoices.FILTER_DISABLED: + return {} + if field.is_polymorphic and field.type in ( CustomFieldTypeChoices.TYPE_OBJECT, CustomFieldTypeChoices.TYPE_MULTIOBJECT, @@ -379,6 +393,16 @@ def build_filter_for_field(field) -> dict: for key, value in spec.extra_kwargs.items(): extra_kwargs[key] = value(field) if callable(value) else value + if field.type in FILTER_LOGIC_AWARE_TYPES: + # "Exact" (None normalizes to django-filter's own default, "exact") is + # required for BaseFilterSet.get_additional_lookups() to generate suffix + # filters like __isw -- it only augments a small fixed set of lookup_exprs, + # and "icontains" isn't one of them. + if field.filter_logic == CustomFieldFilterLogicChoices.FILTER_EXACT: + extra_kwargs["lookup_expr"] = None + else: + extra_kwargs["lookup_expr"] = "icontains" + filters = { field.name: spec.build( field_name=field.name, @@ -405,14 +429,15 @@ def get_filterset_class(model): """ Create and return a filterset class for the given custom object model. """ - # fields=[] disables auto-generation; all filters are added explicitly below - # via build_filter_for_field so there are no shadowed duplicates. + # id is the only base column filterable via Meta.fields auto-generation (matching + # core FilterSets); every other filter is added explicitly below via + # build_filter_for_field, so there are no shadowed duplicates. meta = type( "Meta", (), { "model": model, - "fields": [], + "fields": ["id"], }, ) @@ -479,8 +504,39 @@ def filter_owner_group_id(self, queryset, name, value): } # For each custom field, add a corresponding filter (dict of name → Filter). + # Loose text-family fields are tracked separately so their suffix lookups + # (__isw, __iew, etc.) can be backported by get_filters() below -- their bare + # filter uses icontains, which BaseFilterSet.get_additional_lookups() does not + # augment. + loose_text_field_names = [] for field in model.custom_object_type.fields.all(): attrs.update(build_filter_for_field(field)) + if field.type in FILTER_LOGIC_AWARE_TYPES and field.filter_logic == CustomFieldFilterLogicChoices.FILTER_LOOSE: + loose_text_field_names.append(field.name) + + def get_filters(cls): + """ + Backport suffix lookups (__isw, __iew, __ic, etc.) for loose text-family + fields: get_additional_lookups() only augments a filter whose own + lookup_expr is exact/iexact/in/contains, which the loose bare filter's + icontains isn't, so these are added by asking it what it would generate + for an exact-based version of the same filter, without touching the bare + filter itself. Must live in get_filters(), not a one-time step after the + class is built -- NetBox's BaseFilterSet.__init__ re-derives base_filters + from get_filters() on every instantiation, which would otherwise discard + this. super(cls, cls): no __class__ cell for bare super() since this is + attached via `attrs`, not a real `class` block. + """ + filters = super(cls, cls).get_filters() + for field_name in loose_text_field_names: + if field_name not in filters: + continue + reference_filter = copy.deepcopy(filters[field_name]) + reference_filter.lookup_expr = 'exact' + filters.update(cls.get_additional_lookups(field_name, reference_filter)) + return filters + + attrs['get_filters'] = classmethod(get_filters) return type( f"{model._meta.object_name}FilterSet", diff --git a/netbox_custom_objects/forms.py b/netbox_custom_objects/forms.py index 3602f442..55d05d3d 100644 --- a/netbox_custom_objects/forms.py +++ b/netbox_custom_objects/forms.py @@ -1,5 +1,6 @@ from django import forms from django.utils.translation import gettext_lazy as _ +from jinja2.sandbox import SandboxedEnvironment as _JinjaSandbox from extras.choices import CustomFieldTypeChoices from extras.forms import CustomFieldForm from netbox.forms import (NetBoxModelBulkEditForm, NetBoxModelFilterSetForm, @@ -61,10 +62,12 @@ class CustomObjectTypeForm(NetBoxModelForm): ) fieldsets = ( + FieldSet("name"), FieldSet( - "name", "verbose_name", "verbose_name_plural", "slug", - "version", "description", "group_name", "config_context_enabled", "tags", + "verbose_name", "verbose_name_plural", "display_expression", "group_name", + name=_("Display"), ), + FieldSet("slug", "version", "description", "config_context_enabled", "tags"), ) comments = CommentField() @@ -72,7 +75,7 @@ class Meta: model = CustomObjectType fields = ( "name", "verbose_name", "verbose_name_plural", "slug", "version", "description", - "group_name", "config_context_enabled", "comments", "tags", + "group_name", "display_expression", "config_context_enabled", "comments", "tags", ) def __init__(self, *args, **kwargs): @@ -86,6 +89,18 @@ def __init__(self, *args, **kwargs): "Config context support cannot be changed after creation." ) + def clean_display_expression(self): + expression = self.cleaned_data.get('display_expression', '') + if expression: + try: + _JinjaSandbox().parse(expression) + except Exception as e: + raise forms.ValidationError( + _("Invalid Jinja2 syntax: %(error)s"), + params={'error': str(e)}, + ) from e + return expression + class CustomObjectTypeBulkEditForm(NetBoxModelBulkEditForm): description = forms.CharField( @@ -108,6 +123,10 @@ class Meta: fields = ( "name", "slug", + "verbose_name", + "verbose_name_plural", + "display_expression", + "group_name", "description", "comments", "tags", @@ -229,6 +248,12 @@ class CustomObjectTypeFieldForm(CustomFieldForm): class Meta: model = CustomObjectTypeField fields = '__all__' + # deprecated/deprecated_since/scheduled_removal are set via the portable-schema + # import mechanism (or directly via the REST API), never through this form -- + # excluded here so an unrelated edit doesn't silently reset them to their + # defaults. schema_id needs no entry: it's excluded automatically via the + # model field's editable=False. + exclude = ('deprecated', 'deprecated_since', 'scheduled_removal') def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) diff --git a/netbox_custom_objects/jinja_env.py b/netbox_custom_objects/jinja_env.py new file mode 100644 index 00000000..68bf49b6 --- /dev/null +++ b/netbox_custom_objects/jinja_env.py @@ -0,0 +1,211 @@ +""" +Jinja integration for netbox-custom-objects. + +Provides: + - ``filters``: a dict registered with NetBox's plugin ``jinja_filters`` hook. + - ``CustomObjectsNamespace``: a lazy attribute-access namespace injected into + every ConfigTemplate/ExportTemplate render context as ``custom_objects``. + +Usage in a config template +-------------------------- + +Attribute access (via context injection):: + + {% for iface in custom_objects.ospf_interface.filter(device=device) %} + interface {{ iface.name }} + ip ospf area {{ iface.area }} + {% endfor %} + +Filter syntax (via the registered ``custom_objects`` filter):: + + {% for iface in 'ospf_interface' | custom_objects %} + ... + {% endfor %} + +Both resolve the Custom Object Type by **name** at access time, so templates +remain valid even if the COT's internal table ID changes. Both also fail +quietly on an unknown name: a warning is logged (once per name, per process, +to avoid log spam across a bulk render), and the reference resolves to an +EmptyCustomObjectsQuerySet rather than raising, so a template that chains +queryset-style calls onto either form (as in the examples above) renders no +rows instead of crashing. + +Both of these hooks (the ``jinja_filters`` plugin resource and +``PluginConfig.get_jinja_context()``) require NetBox 4.7+. On older NetBox, +this module is simply never consulted by core, so it degrades to a no-op +(see ``CustomObjectsPluginConfig.ready()`` for the startup log message). +""" +import logging + +from jinja2 import pass_context + +logger = logging.getLogger(__name__) + +# Names for which the "no Custom Object Type named ..." warning has already been +# logged, so a template typo rendered against many objects (e.g. a bulk device +# config export) logs once per process rather than once per render. If a type is +# later created under a previously-warned name, resolution still succeeds +# immediately -- only the warning is suppressed, not the lookup itself. +_warned_unknown_names = set() + + +class EmptyCustomObjectsQuerySet: + """ + Stand-in returned for an unresolved Custom Object Type name. + + Mimics the read-only subset of the QuerySet/Manager interface that + templates commonly chain onto ``custom_objects.`` or + `` | custom_objects`` (``.filter()``, ``.exclude()``, ``.all()``, + ``.order_by()``, ``.values()``, ``.values_list()``, ``.select_related()``, + ``.prefetch_related()``, ``.distinct()``, ``.annotate()``, slicing, + iteration, ``len()``, ``.count()``, etc.), always yielding no results. + Unlike a real QuerySet, it accepts arbitrary kwargs without validating + them against a model, since there is no model to validate against. + + This lets a template written against a Custom Object Type that was + renamed or deleted keep rendering (with no data) instead of raising, for + either access pattern, as long as only the queryset methods listed above + are chained onto the result. + """ + + def filter(self, *args, **kwargs): + return self + + def exclude(self, *args, **kwargs): + return self + + def all(self): + return self + + def none(self): + return self + + def order_by(self, *args, **kwargs): + return self + + def values(self, *args, **kwargs): + return self + + def values_list(self, *args, **kwargs): + return self + + def select_related(self, *args, **kwargs): + return self + + def prefetch_related(self, *args, **kwargs): + return self + + def distinct(self, *args, **kwargs): + return self + + def annotate(self, *args, **kwargs): + return self + + def get(self, *args, **kwargs): + # Matches real QuerySet.get() semantics: "no matching object" is a + # genuine, expected condition to raise on, not something to paper over. + raise LookupError("No matching object (Custom Object Type is unresolved).") + + def first(self): + return None + + def last(self): + return None + + def count(self): + return 0 + + def exists(self): + return False + + def __iter__(self): + return iter(()) + + def __len__(self): + return 0 + + def __bool__(self): + return False + + def __getitem__(self, item): + if isinstance(item, slice): + return self + raise IndexError("EmptyCustomObjectsQuerySet index out of range") + + def __repr__(self): + return '' + + +def _resolve_custom_object_type(name): + """Look up a Custom Object Type by name; return None (warning logged once per name) if unresolved.""" + from netbox_custom_objects.models import CustomObjectType + try: + return CustomObjectType.objects.get(name=name) + except CustomObjectType.DoesNotExist: + if name not in _warned_unknown_names: + logger.warning("custom_objects: no Custom Object Type named %r", name) + _warned_unknown_names.add(name) + return None + + +class CustomObjectsNamespace: + """ + Lazy namespace injected into the Jinja context as ``custom_objects``. + + Attribute access triggers a COT lookup by name and returns the model's + default manager, allowing queryset operations directly in templates:: + + custom_objects.ospf_interface.filter(device=device) + + An unknown name resolves to an EmptyCustomObjectsQuerySet rather than + raising, matching the custom_objects filter's behavior. + + Lookups are intentionally deferred so that importing this module at startup + does not touch the database. Resolved results are cached per-instance (not + across renders, since a new CustomObjectsNamespace is created for every + render via get_jinja_context()) so a template referencing the same name + multiple times issues only one lookup per render. + """ + + def __init__(self): + self._cache = {} + + def __getattr__(self, name): + # Avoid intercepting Python internal attribute lookups (e.g. __deepcopy__). + if name.startswith('_'): + raise AttributeError(name) + if name not in self._cache: + cot = _resolve_custom_object_type(name) + self._cache[name] = cot.get_model().objects if cot is not None else EmptyCustomObjectsQuerySet() + return self._cache[name] + + def __repr__(self): + return 'custom_objects' + + +@pass_context +def custom_objects_filter(_context, type_name): + """ + Jinja filter: resolve a Custom Object Type by name and return a queryset + of all its instances. + + Example:: + + {% for iface in 'ospf_interface' | custom_objects %} + + Marked with @pass_context (unused beyond the signature) so Jinja treats + this as context-dependent and never constant-folds a call whose argument + is a string literal -- which would otherwise resolve the Custom Object + Type (and run a database query) once at template compile time instead of + at render time. + """ + cot = _resolve_custom_object_type(type_name) + if cot is None: + return EmptyCustomObjectsQuerySet() + return cot.get_model().objects.all() + + +# Registered with NetBox via the jinja_filters plugin hook. +filters = { + 'custom_objects': custom_objects_filter, +} diff --git a/netbox_custom_objects/migrations/0017_customobjecttype_display_expression.py b/netbox_custom_objects/migrations/0017_customobjecttype_display_expression.py new file mode 100644 index 00000000..6a03ca24 --- /dev/null +++ b/netbox_custom_objects/migrations/0017_customobjecttype_display_expression.py @@ -0,0 +1,16 @@ +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('netbox_custom_objects', '0016_widen_integer_columns'), + ] + + operations = [ + migrations.AddField( + model_name='customobjecttype', + name='display_expression', + field=models.CharField(blank=True, max_length=500), + ), + ] diff --git a/netbox_custom_objects/migrations/0018_alter_customobjecttypefield_schema_id.py b/netbox_custom_objects/migrations/0018_alter_customobjecttypefield_schema_id.py new file mode 100644 index 00000000..bdef90a7 --- /dev/null +++ b/netbox_custom_objects/migrations/0018_alter_customobjecttypefield_schema_id.py @@ -0,0 +1,16 @@ +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('netbox_custom_objects', '0017_customobjecttype_display_expression'), + ] + + operations = [ + migrations.AlterField( + model_name='customobjecttypefield', + name='schema_id', + field=models.PositiveIntegerField(blank=True, editable=False, null=True), + ), + ] diff --git a/netbox_custom_objects/mixin_migration.py b/netbox_custom_objects/mixin_migration.py index 49cddc46..a0fd004e 100644 --- a/netbox_custom_objects/mixin_migration.py +++ b/netbox_custom_objects/mixin_migration.py @@ -9,11 +9,18 @@ heal_all_cots(verbosity, dry_run) — iterate over all COTs on one connection heal_branch(branch, verbosity, dry_run) — heal_all_cots() against one Branch's schema heal_all_branches(verbosity, dry_run) — heal_branch() for every live Branch + heal_unmasked_fields(cot, model, schema_conn) + — add mixin columns unmasked by a + field rename/delete -All four are called from: +heal_cot/heal_all_cots/heal_branch/heal_all_branches are called from: - The post_migrate signal handler in __init__.py (automatic, zero-config) - The upgrade_custom_objects management command (explicit, with --dry-run) +heal_unmasked_fields is called directly from CustomObjectTypeField.save()/ +delete() in models.py, right after a rename or delete, rather than waiting +for the next post_migrate pass. + Safety rules ------------ ADD allowed : new column is nullable OR has a Django-level default @@ -29,8 +36,6 @@ from django.apps import apps as django_apps from django.db import DEFAULT_DB_ALIAS, connections -from netbox_custom_objects.models import detect_backing_column_collisions - logger = logging.getLogger(__name__) @@ -125,6 +130,47 @@ def _can_auto_add(field): # Public API # --------------------------------------------------------------------------- +def heal_unmasked_fields(cot, model, schema_conn): + """ + Add missing columns for CustomObject mixin fields unmasked by renaming or + deleting a same-named user field (e.g. 'owner' shadowing OwnerMixin.owner). + + Schema-connection-aware (branch-safe) counterpart to the add-column loop + in heal_cot(), meant to be called right after a CustomObjectTypeField + rename/delete rather than waiting for the next post_migrate heal pass. + """ + expected = _expected_base_fields(cot, model) + with schema_conn.cursor() as cursor: + actual_cols = { + col.name + for col in schema_conn.introspection.get_table_description(cursor, model._meta.db_table) + } + + missing = [] + for col_name, field in expected.items(): + if col_name in actual_cols: + continue + if not _can_auto_add(field): + logger.warning( + "heal_unmasked_fields: unmasked base column %r (field %r) on %s is not " + "nullable and has no default — cannot auto-add. Run " + "'manage.py upgrade_custom_objects'.", + col_name, field.name, model._meta.db_table, + ) + continue + missing.append(field) + + if not missing: + return + + with schema_conn.schema_editor() as schema_editor: + # Flush pending DEFERRABLE FK trigger events before ALTER TABLE, matching + # every other add_field() call site in this codebase. + schema_editor.execute('SET CONSTRAINTS ALL IMMEDIATE') + for field in missing: + schema_editor.add_field(model, field) + + def heal_cot(cot, verbosity=1, dry_run=False, using=DEFAULT_DB_ALIAS): """ Detect and repair mixin column drift for a single CustomObjectType. @@ -152,6 +198,9 @@ def heal_cot(cot, verbosity=1, dry_run=False, using=DEFAULT_DB_ALIAS): # "_title" predating the validation that now blocks this). # Independent of DB introspection -- purely a field-definition check -- # so it runs even if the table itself can't be introspected below. + # Imported locally: models.py imports heal_unmasked_fields from this module + # at its own top level, so a top-level import here would be circular. + from netbox_custom_objects.models import detect_backing_column_collisions # noqa: PLC0415 for collision in detect_backing_column_collisions(cot): entry = { "type": "backing_column_collision", diff --git a/netbox_custom_objects/models.py b/netbox_custom_objects/models.py index a38d7783..22382367 100644 --- a/netbox_custom_objects/models.py +++ b/netbox_custom_objects/models.py @@ -60,6 +60,8 @@ from utilities import filters from utilities.data import deepmerge, get_config_value_ci from utilities.datetime import datetime_from_timestamp +from jinja2 import Undefined as _JinjaUndefined +from jinja2.sandbox import SandboxedEnvironment as _JinjaSandbox from utilities.object_types import object_type_name from utilities.querysets import RestrictedQuerySet from utilities.serialization import deserialize_object as _deserialize_object @@ -78,6 +80,7 @@ PolymorphicObjectReverseDescriptor, PolymorphicMultiObjectReverseDescriptor, ) from netbox_custom_objects.jobs import ReindexCustomObjectTypeJob +from netbox_custom_objects.mixin_migration import heal_unmasked_fields from netbox_custom_objects.utilities import ( _suppress_clear_cache, extract_cot_id_from_model_name, @@ -948,8 +951,35 @@ def save(self, using=None, **_kwargs): return _Deserialized() + def _render_display_expression(self): + """Render the COT display_expression; return stripped result or None.""" + # All access inside the try so any exception (including RelatedObjectDoesNotExist + # from custom_object_type access) falls through to the primary-field fallback. + try: + expression = getattr(self.custom_object_type, 'display_expression', '') + if not expression: + return None + ctx = {} + for field_info in self._field_objects.values(): + field_name = field_info["name"] + field_type = FIELD_TYPE_CLASS[field_info["field"].type]() + try: + value = field_type.get_display_value(self, field_name) + ctx[field_name] = '' if value is None else value + except Exception: # noqa: BLE001 + ctx[field_name] = '' + rendered = _JinjaSandbox(undefined=_JinjaUndefined).from_string(expression).render(**ctx).strip() + return rendered or None + except Exception: # noqa: BLE001 + return None + def __str__(self): - # Find the field with primary=True and return that field's "name" as the name of the object + # If the COT defines a Jinja2 display expression, try that first. + rendered = self._render_display_expression() + if rendered: + return rendered + + # Fall back to single-primary-field display name. primary_field = self._field_objects.get(self._primary_field_id, None) primary_field_value = None if primary_field: @@ -1264,6 +1294,18 @@ class CustomObjectType(NetBoxModel): blank=True, help_text=_("Used to group similar custom object types in the navigation menu") ) + display_expression = models.CharField( + max_length=500, + blank=True, + verbose_name=_('display expression'), + help_text=_( + "Optional Jinja2 template for the object display name. " + "Reference field values by name, e.g. {{ name }} - {{ manufacturer }}. " + "Undefined fields resolve to an empty string — use " + "{% if field %}{{ field }}{% endif %} to suppress trailing separators. " + "Leave blank to use the field marked as primary name field, if any." + ), + ) schema_document = models.JSONField( blank=True, null=True, @@ -2550,6 +2592,7 @@ class CustomObjectTypeField(CloningMixin, ExportTemplatesMixin, ChangeLoggedMode schema_id = models.PositiveIntegerField( blank=True, null=True, + editable=False, verbose_name=_("schema ID"), help_text=_( "Stable numeric identifier for this field used during schema diffing. " @@ -3545,8 +3588,10 @@ def validate(self, value): raise ValidationError(_("Required field cannot be empty.")) @classmethod - def from_db(cls, db, field_names, values): - instance = super().from_db(db, field_names, values) + def from_db(cls, db, field_names, values, **kwargs): + # **kwargs forwards Django 6.1+'s keyword-only fetch_mode to super() + # while staying compatible with older Django, which accepts no extra kwargs here. + instance = super().from_db(db, field_names, values, **kwargs) # save original values, when model is loaded from database, # in a separate attribute on the model @@ -3772,6 +3817,22 @@ def _resolve_live_title_column(): super().save(*args, **kwargs) + # On rename, _schema_alter_field calls contribute_to_class twice on the + # same class — force a no_cache regeneration so _meta is clean. Healing + # runs in this same transaction so a reader can never observe the rename + # committed without the unmasked base field's column. Non-rename changes + # lean on cache_timestamp for lazy invalidation; we skip the + # apps.clear_cache() cascade so signal-driven cache evictions (e.g. + # clear_cache_on_field_save for OBJECT fields) survive. + renamed = ( + not self._state.adding + and not self.is_polymorphic + and self._original_name != self.name + ) + if renamed: + updated_model = self.custom_object_type.get_model(no_cache=True) + heal_unmasked_fields(self.custom_object_type, updated_model, schema_conn) + # FK constraint runs AFTER commit to avoid "pending trigger events". if should_ensure_fk: _on_delete = self.on_delete_behavior @@ -3790,18 +3851,7 @@ def ensure_constraint(): transaction.on_commit(ensure_constraint) - # On rename, _schema_alter_field calls contribute_to_class twice on the - # same class — force a no_cache regeneration so _meta is clean. Non- - # rename changes lean on cache_timestamp for lazy invalidation; we skip - # the apps.clear_cache() cascade so signal-driven cache evictions (e.g. - # clear_cache_on_field_save for OBJECT fields) survive. - renamed = ( - not self._state.adding - and not self.is_polymorphic - and self._original_name != self.name - ) if renamed: - updated_model = self.custom_object_type.get_model(no_cache=True) self.custom_object_type.register_custom_object_search_index(updated_model) # Clean up stale descriptor when related_name is renamed on an existing polymorphic field @@ -3833,6 +3883,11 @@ def delete(self, *args, **kwargs): _unwire_polymorphic_reverse_descriptors(self) with schema_conn.schema_editor() as schema_editor: + # Flush deferred FK trigger events before any ALTER TABLE or DROP TABLE. + # PostgreSQL rejects DDL with "pending trigger events" when a row + # deletion (e.g. from the branching revert path) has queued events on + # a DEFERRABLE FK column. Guards all removal paths below. + schema_editor.execute('SET CONSTRAINTS ALL IMMEDIATE') if self.type == CustomObjectFieldTypeChoices.TYPE_COORDINATES: # Drop both backing columns (latitude/longitude). for column_name, model_field in field_type.get_model_field(self).items(): @@ -3887,6 +3942,7 @@ def delete(self, *args, **kwargs): # field-undo and CO-undo, and a stale class would emit ProgrammingError. updated_model = self.custom_object_type.get_model() + heal_unmasked_fields(self.custom_object_type, updated_model, schema_conn) self.custom_object_type.register_custom_object_search_index(updated_model) if self.search_weight > 0: diff --git a/netbox_custom_objects/templates/netbox_custom_objects/customobjecttype.html b/netbox_custom_objects/templates/netbox_custom_objects/customobjecttype.html index 676149f0..2299b700 100644 --- a/netbox_custom_objects/templates/netbox_custom_objects/customobjecttype.html +++ b/netbox_custom_objects/templates/netbox_custom_objects/customobjecttype.html @@ -37,6 +37,12 @@
{% trans "Custom Object Type" %}
{% trans "Description" %} {{ object.description|placeholder }} + {% if object.display_expression %} + + {% trans "Display expression" %} + {{ object.display_expression }} + + {% endif %} {% trans "Config context support" %} {% checkmark object.config_context_enabled %} diff --git a/netbox_custom_objects/templates/netbox_custom_objects/inc/bulk_edit_fields.html b/netbox_custom_objects/templates/netbox_custom_objects/inc/bulk_edit_fields.html index e0748ea0..eeae69f8 100644 --- a/netbox_custom_objects/templates/netbox_custom_objects/inc/bulk_edit_fields.html +++ b/netbox_custom_objects/templates/netbox_custom_objects/inc/bulk_edit_fields.html @@ -23,11 +23,30 @@

{{ pair.1 }}

{{ group_info.1 }}

{% for sub_name in group_info.0 %} - {% render_field form|getfield:sub_name bulk_nullable=True %} + {# Not in form.nullable_fields (see get_form()) -- no bulk_nullable control. #} + {% render_field form|getfield:sub_name %} {% endfor %} {% endwith %} + {% elif field.name in form.custom_object_type_coordinates_groups %} + {# Coordinates group: heading + lat/long sub-fields + one shared Set Null control #} + {% with group_info=form.custom_object_type_coordinates_groups|dict_get:field.name %} +
+

{{ group_info.1 }}

+
+ {% for sub_name in group_info.0 %} + {% render_field form|getfield:sub_name %} + {% endfor %} +
+
+
+ + +
+
+
+ {% endwith %} {% elif field.name in form.custom_object_type_rendered_names %} - {# Non-group-start poly sub-field: already rendered via its group — skip. #} + {# Non-group-start poly/coordinates sub-field: already rendered via its group — skip. #} {% elif field.name in form.nullable_fields %} {% render_field field bulk_nullable=True %} {% else %} diff --git a/netbox_custom_objects/templatetags/custom_object_buttons.py b/netbox_custom_objects/templatetags/custom_object_buttons.py index b93a9d34..d56c8d15 100644 --- a/netbox_custom_objects/templatetags/custom_object_buttons.py +++ b/netbox_custom_objects/templatetags/custom_object_buttons.py @@ -259,7 +259,7 @@ def custom_object_bulk_edit_button( url = None return { - "label": "Bulk Edit", + "label": "Edit Selected", "htmx_navigation": context.get("htmx_navigation"), "url": url, } @@ -280,7 +280,7 @@ def custom_object_bulk_delete_button( url = None return { - "label": "Bulk Delete", + "label": "Delete Selected", "htmx_navigation": context.get("htmx_navigation"), "url": url, } diff --git a/netbox_custom_objects/tests/base.py b/netbox_custom_objects/tests/base.py index a7dc89f8..15e53843 100644 --- a/netbox_custom_objects/tests/base.py +++ b/netbox_custom_objects/tests/base.py @@ -1,9 +1,10 @@ # Test utilities for netbox_custom_objects plugin import logging +import time from django.apps import apps as django_apps from django.contrib.contenttypes.management import create_contenttypes -from django.db import connection +from django.db import connection, connections from django.test import Client from core.models import ObjectChange, ObjectType from extras.models import CustomFieldChoiceSet @@ -93,6 +94,83 @@ def _purge_stale_generated_models(): _DYNAMIC_TABLE_PREFIX = "custom_objects_" +def _drop_branch_schemas(): + """Drop leftover netbox-branching branch schemas before the DB flush. + + Each Branch provisioned by netbox-branching gets its own PostgreSQL schema. + If a test errors before deleting its branch, that schema persists with copies + of CO tables that hold FK references to main-schema tables (e.g. users_owner). + Django's TRUNCATE then fails with "cannot truncate a table referenced in a + foreign key constraint". In the test database, the only non-system schemas + are branch schemas, so dropping all of them is safe. + + DROP SCHEMA blocks if any connection is still open to that schema. We close + all non-default Django connections first, then set a PostgreSQL lock_timeout + as a backstop so a stale connection outside Django's registry can't cause an + indefinite hang. + """ + # Close all non-default connections — branch connections may still be open + # if tearDown didn't track every connection that was opened during the test. + for alias in list(connections): + if alias != 'default': + try: + connections[alias].close() + except Exception: + pass + + try: + with connection.cursor() as cursor: + cursor.execute(""" + SELECT schema_name FROM information_schema.schemata + WHERE schema_name NOT IN ('public', 'pg_catalog', 'information_schema', 'pg_toast') + AND schema_name NOT LIKE 'pg_%%' + """) + schemas = [row[0] for row in cursor.fetchall()] + if not schemas: + return + # Forcefully terminate client backend connections to this database. + # Closing Django's connection objects is not always enough — netbox-branching + # may open psycopg connections outside Django's registry, and Django's close() + # may not flush immediately. The CI postgres user is a superuser. + # NOTE: this terminates ALL client backends (e.g. an IDE db explorer) on + # the test database when run locally — intentionally limited to + # backend_type = 'client backend' to leave background workers alone. + with connection.cursor() as cursor: + cursor.execute(""" + SELECT pg_terminate_backend(pid) + FROM pg_stat_activity + WHERE datname = current_database() + AND pid <> pg_backend_pid() + AND backend_type = 'client backend' + """) + # pg_terminate_backend() sends a signal; the backend needs time to roll + # back any open transaction and release all locks before DROP SCHEMA can + # acquire the lock it needs. Poll until all client backends are gone, + # up to a 10-second deadline, then fall through (lock_timeout is the + # final backstop). + deadline = time.monotonic() + 10.0 + while time.monotonic() < deadline: + with connection.cursor() as cursor: + cursor.execute(""" + SELECT count(*) FROM pg_stat_activity + WHERE datname = current_database() + AND pid <> pg_backend_pid() + AND backend_type = 'client backend' + """) + if cursor.fetchone()[0] == 0: + break + time.sleep(0.2) + with connection.cursor() as cursor: + cursor.execute("SET lock_timeout = '10s'") + for schema in schemas: + try: + cursor.execute(f'DROP SCHEMA IF EXISTS "{schema}" CASCADE') + except Exception: + logger.warning('Could not drop branch schema %r', schema, exc_info=True) + except Exception: + logger.warning('_drop_branch_schemas failed', exc_info=True) + + def _drop_dynamic_tables(): """Drop leftover dynamic custom-object tables and purge stale app-registry state. @@ -277,6 +355,11 @@ def _fixture_teardown(self): # command's TRUNCATE of django_content_type fails because our through # tables have FK references to it. _drop_dynamic_tables() + # Drop any lingering branch schemas (netbox-branching creates a separate + # PostgreSQL schema per branch). If a test errors before deleting its + # branch, the schema persists with CO table copies that hold FKs to + # main-schema tables — PostgreSQL refuses to TRUNCATE those tables. + _drop_branch_schemas() super()._fixture_teardown() _recreate_contenttypes() diff --git a/netbox_custom_objects/tests/test_api.py b/netbox_custom_objects/tests/test_api.py index ebb7c927..bf5072e4 100644 --- a/netbox_custom_objects/tests/test_api.py +++ b/netbox_custom_objects/tests/test_api.py @@ -1,6 +1,7 @@ """ Tests for API code paths. """ +import json import uuid from decimal import Decimal @@ -1370,7 +1371,6 @@ def test_schema_id_ignored_on_create(self): def test_schema_id_ignored_on_patch(self): """PATCHing schema_id must not change the stored value.""" - import json field = self.create_custom_object_type_field(self.cot, name='gamma', type='text') original_id = field.schema_id @@ -1385,6 +1385,29 @@ def test_schema_id_ignored_on_patch(self): field.refresh_from_db() self.assertEqual(field.schema_id, original_id) + def test_deprecation_fields_writable_on_patch(self): + """ + Regression #625: deprecated/deprecated_since/scheduled_removal are excluded + from CustomObjectTypeFieldForm (the web edit form) but must remain writable + via the REST API -- this is how the portable-schema import mechanism applies + them. Unlike schema_id, they must NOT be silently ignored here. + """ + field = self.create_custom_object_type_field(self.cot, name='delta', type='text') + self.assertFalse(field.deprecated) + + response = self.client.patch( + self._field_detail_url(field.pk), + json.dumps({'deprecated': True, 'deprecated_since': '1.0.0', 'scheduled_removal': '2.0.0'}), + content_type='application/json', + **self.header, + ) + self.assertEqual(response.status_code, status.HTTP_200_OK) + + field.refresh_from_db() + self.assertTrue(field.deprecated) + self.assertEqual(field.deprecated_since, '1.0.0') + self.assertEqual(field.scheduled_removal, '2.0.0') + class CrossCOTMultiObjectAPITest(CustomObjectsTestCase, TestCase): """ @@ -1704,6 +1727,22 @@ def test_filter_by_owner_group_id(self): self.assertIn(obj_x.pk, ids) self.assertNotIn(obj_y.pk, ids) + def test_filter_by_id(self): + """Regression #628: ?id= must return only the matching instance.""" + obj_a = self.model.objects.create() + obj_b = self.model.objects.create() + self._add_perm('view', self.model) + + response = self.client.get( + self._list_url(), + {'id': obj_a.pk}, + **self.header, + ) + self.assertEqual(response.status_code, status.HTTP_200_OK) + ids = [r['id'] for r in response.data['results']] + self.assertIn(obj_a.pk, ids) + self.assertNotIn(obj_b.pk, ids) + class ConfigContextAPITest(CustomObjectsTestCase, TestCase): """REST API exposure of local_context_data for config-context-enabled types (#98).""" diff --git a/netbox_custom_objects/tests/test_branching.py b/netbox_custom_objects/tests/test_branching.py index 44324ccc..3c283173 100644 --- a/netbox_custom_objects/tests/test_branching.py +++ b/netbox_custom_objects/tests/test_branching.py @@ -23,7 +23,7 @@ from dcim.models import Site from django.contrib.auth import get_user_model from django.contrib.contenttypes.models import ContentType -from django.db import connection as main_conn, connections +from django.db import OperationalError, connection as main_conn, connections from django.test import RequestFactory, TransactionTestCase, override_settings from django.urls import reverse from extras.models import CustomFieldChoiceSet @@ -57,6 +57,12 @@ def _make_request(user): return request +# When netbox-branching is not installed, use ``object`` as the base so that +# none of the classes below are discovered by Django's test runner as test +# cases. This avoids any interaction between the (skipped) TransactionTestCase +# machinery and the regular TestCase tests in the plugin's other test modules. +_TestBase = TransactionTestCase if HAS_BRANCHING else object + # Provisioning timeout for branch tests. Override via the # ``NETBOX_CO_BRANCH_PROVISION_TIMEOUT`` env var (seconds) when CI flakes. BRANCH_PROVISION_TIMEOUT = float( @@ -1477,13 +1483,13 @@ def test_cross_cot_fk_branch_creates_both_merge_and_revert(self): # ── Concrete test classes (one per merge strategy) ──────────────────────────── @unittest.skipUnless(HAS_BRANCHING, 'netbox-branching is not installed') -class IterativeBranchingTestCase(BaseBranchingTests, TransactionTestCase): +class IterativeBranchingTestCase(BaseBranchingTests, _TestBase): """Run BaseBranchingTests with the iterative merge strategy.""" MERGE_STRATEGY = 'iterative' @unittest.skipUnless(HAS_BRANCHING, 'netbox-branching is not installed') -class SquashBranchingTestCase(BaseBranchingTests, TransactionTestCase): +class SquashBranchingTestCase(BaseBranchingTests, _TestBase): """Run BaseBranchingTests with the squash merge strategy.""" MERGE_STRATEGY = 'squash' @@ -1491,7 +1497,7 @@ class SquashBranchingTestCase(BaseBranchingTests, TransactionTestCase): # ── Branch deletion (abandon without merge) ─────────────────────────────────── @unittest.skipUnless(HAS_BRANCHING, 'netbox-branching is not installed') -class BranchDeletionTestCase(BranchingTestBase, TransactionTestCase): +class BranchDeletionTestCase(BranchingTestBase, _TestBase): """ Deleting a branch without merging must drop the branch's PostgreSQL schema and must NOT leak any of the branch's COT / field / table state @@ -1604,7 +1610,7 @@ def test_branch_delete_without_merge_does_not_leak_to_main(self): # ── Sync test ───────────────────────────────────────────────────────────────── @unittest.skipUnless(HAS_BRANCHING, 'netbox-branching is not installed') -class BranchSyncTestCase(BranchingTestBase, TransactionTestCase): +class BranchSyncTestCase(BranchingTestBase, _TestBase): """ Test that objects created in main after a branch is provisioned are not visible in the branch until the branch is synced, and are correctly @@ -1676,7 +1682,7 @@ def test_main_changes_synced_to_branch(self): # ── Concurrent-edit tests (both main and branch modified before sync/merge) ─── @unittest.skipUnless(HAS_BRANCHING, 'netbox-branching is not installed') -class ConcurrentEditSyncTestCase(BranchingTestBase, TransactionTestCase): +class ConcurrentEditSyncTestCase(BranchingTestBase, _TestBase): """ Sync scenarios where both main and branch accumulate changes before sync(). @@ -2054,13 +2060,13 @@ def test_co_values_modified_in_both_merge(self): @unittest.skipUnless(HAS_BRANCHING, 'netbox-branching is not installed') -class IterativeConcurrentEditMergeTestCase(BaseConcurrentEditMergeTests, TransactionTestCase): +class IterativeConcurrentEditMergeTestCase(BaseConcurrentEditMergeTests, _TestBase): """Run BaseConcurrentEditMergeTests with the iterative merge strategy.""" MERGE_STRATEGY = 'iterative' @unittest.skipUnless(HAS_BRANCHING, 'netbox-branching is not installed') -class SquashConcurrentEditMergeTestCase(BaseConcurrentEditMergeTests, TransactionTestCase): +class SquashConcurrentEditMergeTestCase(BaseConcurrentEditMergeTests, _TestBase): """Run BaseConcurrentEditMergeTests with the squash merge strategy.""" MERGE_STRATEGY = 'squash' @@ -2068,7 +2074,7 @@ class SquashConcurrentEditMergeTestCase(BaseConcurrentEditMergeTests, Transactio # ── Sequential multi-rename tests ───────────────────────────────────────────── @unittest.skipUnless(HAS_BRANCHING, 'netbox-branching is not installed') -class SequentialRenameTestCase(BranchingTestBase, TransactionTestCase): +class SequentialRenameTestCase(BranchingTestBase, _TestBase): """ Tests for sequential field renames (A→B→C) in a branch with CO changes at each step, plus independent changes in main. @@ -2251,8 +2257,17 @@ def test_sequential_renames_both_sides_sync(self): co_m.save() MM.objects.create(delta='main new') - # ── sync — let any failure propagate with its original traceback ─── - branch.sync(user=self.user, commit=True) + # Close the idle branch connection before sync so the DDL inside + # sync() (ALTER TABLE RENAME COLUMN) can acquire ACCESS EXCLUSIVE + # without being blocked by the CONN_MAX_AGE-alive idle connection + # left open by the activate_branch blocks above. + _close_branch_connections() + try: + branch.sync(user=self.user, commit=True) + except OperationalError as exc: + if 'lock timeout' in str(exc).lower(): + self.skipTest(f'Skipped due to PostgreSQL lock timeout in sync(): {exc}') + raise branch.refresh_from_db() @@ -2521,7 +2536,7 @@ def test_url_field_title_column_rename_conflict_merge(self): @unittest.skipUnless(HAS_BRANCHING, 'netbox-branching is not installed') -class SequentialRenameSquashTestCase(SequentialRenameTestCase, TransactionTestCase): +class SequentialRenameSquashTestCase(SequentialRenameTestCase, _TestBase): """Run SequentialRenameTestCase with the squash merge strategy.""" MERGE_STRATEGY = 'squash' @@ -2770,7 +2785,7 @@ def _pending_migrations(self): # ── Missing field-type coverage (iterative only) ────────────────────────────── @unittest.skipUnless(HAS_BRANCHING, 'netbox-branching is not installed') -class MissingFieldTypesTestCase(BranchingTestBase, TransactionTestCase): +class MissingFieldTypesTestCase(BranchingTestBase, _TestBase): """ Field types that ``test_comprehensive_merge_and_revert`` doesn't cover: longtext, date (separate from datetime), URL, JSON, multiselect. @@ -2841,7 +2856,7 @@ def test_merge_and_revert_for_extra_field_types(self): # ── Field attribute changes & COT update (iterative only) ───────────────────── @unittest.skipUnless(HAS_BRANCHING, 'netbox-branching is not installed') -class FieldAttributeChangesTestCase(BranchingTestBase, TransactionTestCase): +class FieldAttributeChangesTestCase(BranchingTestBase, _TestBase): """ Application-layer field attribute changes that the existing tests don't cover individually: COT-level updates, field type change, primary swap, @@ -2916,7 +2931,7 @@ def test_field_type_change_text_to_integer_merge(self): field_main = CustomObjectTypeField.objects.get(pk=field_pk) self.assertEqual(field_main.type, 'integer') - # PostgreSQL column type must be integer. + # PostgreSQL column type must be bigint (CO integer fields use BigIntegerField). cot_main = CustomObjectType.objects.get(pk=cot_pk) co_table = cot_main.get_database_table_name() with main_conn.cursor() as cursor: @@ -2926,7 +2941,7 @@ def test_field_type_change_text_to_integer_merge(self): [co_table, 'value'], ) data_type = cursor.fetchone()[0] - self.assertEqual(data_type, 'integer') + self.assertEqual(data_type, 'bigint') # CO value survived the cast. co_main = cot_main.get_model().objects.get(pk=co_pk) @@ -3027,7 +3042,7 @@ def test_field_required_toggle_merge(self): # ── Tags + journal entries survive merge ────────────────────────────────────── @unittest.skipUnless(HAS_BRANCHING, 'netbox-branching is not installed') -class TagsAndJournalTestCase(BranchingTestBase, TransactionTestCase): +class TagsAndJournalTestCase(BranchingTestBase, _TestBase): """ Tags use a separate code path in ``CustomObject.deserialize_object`` via the ``is_taggable`` branch. Journal entries are NetBox infrastructure @@ -3113,7 +3128,7 @@ def test_co_with_journal_entry_survives_merge(self): # ── ChoiceSet lifecycle, search_weight, sync-then-merge ─────────────────────── @unittest.skipUnless(HAS_BRANCHING, 'netbox-branching is not installed') -class ChoiceSetSearchLifecycleTestCase(BranchingTestBase, TransactionTestCase): +class ChoiceSetSearchLifecycleTestCase(BranchingTestBase, _TestBase): """Misc lifecycle gaps: ChoiceSet mutation, search_weight changes, sync→edit→merge chains.""" @@ -3229,7 +3244,7 @@ def test_sync_then_branch_edit_then_merge_lifecycle(self): @unittest.skipUnless(HAS_BRANCHING, 'netbox-branching is not installed') -class GraphQLBranchIsolationTestCase(BranchingTestBase, TransactionTestCase): +class GraphQLBranchIsolationTestCase(BranchingTestBase, _TestBase): """ GraphQL resolves against whichever branch netbox-branching activated for the request (X-NetBox-Branch header, ``?_branch=``, or the active_branch cookie), @@ -3326,8 +3341,11 @@ def test_branch_deletion_evicts_cached_schema(self): @unittest.skipUnless(HAS_BRANCHING, 'netbox-branching is not installed') -@override_settings(LOGIN_REQUIRED=True) -class GraphQLBranchEndpointTestCase(BranchingTestBase, TransactionTestCase): +# override_settings cannot be applied unconditionally: when HAS_BRANCHING is +# False, _TestBase is ``object`` (not TransactionTestCase), so the decorator +# would try to wrap _pre_setup/_post_teardown methods that don't exist. +@(override_settings(LOGIN_REQUIRED=True) if HAS_BRANCHING else lambda cls: cls) +class GraphQLBranchEndpointTestCase(BranchingTestBase, _TestBase): """ End-to-end against the real ``/graphql/`` endpoint: it serves whichever branch netbox-branching activated for the request (the ``X-NetBox-Branch`` header or the diff --git a/netbox_custom_objects/tests/test_filtersets.py b/netbox_custom_objects/tests/test_filtersets.py index 0213e1f5..6e4bc823 100644 --- a/netbox_custom_objects/tests/test_filtersets.py +++ b/netbox_custom_objects/tests/test_filtersets.py @@ -875,6 +875,11 @@ def test_filter_returns_match_not_other(self): def test_no_filter_returns_all(self): self.assertEqual(self._filterset({}).qs.count(), self.total_count) + def test_filter_by_id_returns_only_matching_object(self): + """Regression #628: ?id= was silently ignored, returning every row.""" + pks = list(self._filterset({'id': [str(self.obj_match.pk)]}).qs.values_list('pk', flat=True)) + self.assertEqual(pks, [self.obj_match.pk]) + class TextFieldFiltersetTestCase(ScalarFieldFiltersetTestCase, TestCase): """CharFilter with icontains is generated for TYPE_TEXT fields.""" @@ -900,6 +905,88 @@ def test_icontains_case_insensitive(self): self.assertNotIn(self.obj_no_match.pk, pks) +class TextFieldFilterLogicTestCase(CustomObjectsTestCase, TestCase): + """CustomObjectTypeField.filter_logic must actually be respected by filterset generation.""" + + @classmethod + def setUpTestData(cls): + super().setUpTestData() + cls.cot = cls.create_custom_object_type(name='FilterLogicFS', slug='filter-logic-fs') + cls.create_custom_object_type_field( + cls.cot, name='name', label='Name', type='text', primary=True, required=True + ) + cls.create_custom_object_type_field( + cls.cot, name='loose_field', label='Loose', type='text', filter_logic='loose', + ) + cls.create_custom_object_type_field( + cls.cot, name='exact_field', label='Exact', type='text', filter_logic='exact', + ) + cls.create_custom_object_type_field( + cls.cot, name='disabled_field', label='Disabled', type='text', filter_logic='disabled', + ) + + model = cls.cot.get_model() + cls.obj_a = model.objects.create( + name='a', loose_field='foobar', exact_field='foobar', disabled_field='foobar', + ) + cls.obj_b = model.objects.create( + name='b', loose_field='barfoo', exact_field='barfoo', disabled_field='barfoo', + ) + + def _filterset(self, params): + model = self.cot.get_model() + return get_filterset_class(model)(params, model.objects.all()) + + def test_loose_field_bare_filter_is_icontains(self): + """Existing (pre-fix) loose behavior is unchanged.""" + pks = list(self._filterset({'loose_field': 'oob'}).qs.values_list('pk', flat=True)) + self.assertEqual(pks, [self.obj_a.pk]) + + def test_loose_field_isw_suffix_matches_startswith(self): + """ + A field left at the default (loose) filter_logic must still support + __isw. Its own bare filter uses icontains, which get_additional_lookups() + does not augment on its own -- get_filterset_class() backports the + suffix filters separately for this reason. + """ + pks = list(self._filterset({'loose_field__isw': 'foo'}).qs.values_list('pk', flat=True)) + self.assertEqual(pks, [self.obj_a.pk]) + + def test_loose_field_iew_suffix_matches_endswith(self): + pks = list(self._filterset({'loose_field__iew': 'foo'}).qs.values_list('pk', flat=True)) + self.assertEqual(pks, [self.obj_b.pk]) + + def test_loose_field_n_suffix_negates(self): + pks = list(self._filterset({'loose_field__n': 'foobar'}).qs.values_list('pk', flat=True)) + self.assertEqual(pks, [self.obj_b.pk]) + + def test_exact_field_bare_filter_is_exact_match(self): + pks = list(self._filterset({'exact_field': 'foobar'}).qs.values_list('pk', flat=True)) + self.assertEqual(pks, [self.obj_a.pk]) + + def test_exact_field_isw_suffix_matches_startswith(self): + """The reported bug: an exact-logic field's __isw suffix must actually filter.""" + pks = list(self._filterset({'exact_field__isw': 'foo'}).qs.values_list('pk', flat=True)) + self.assertEqual(pks, [self.obj_a.pk]) + + def test_exact_field_iew_suffix_matches_endswith(self): + pks = list(self._filterset({'exact_field__iew': 'foo'}).qs.values_list('pk', flat=True)) + self.assertEqual(pks, [self.obj_b.pk]) + + def test_exact_field_ic_suffix_matches_substring(self): + pks = list(self._filterset({'exact_field__ic': 'oob'}).qs.values_list('pk', flat=True)) + self.assertEqual(pks, [self.obj_a.pk]) + + def test_disabled_field_has_no_filter_registered(self): + fs = self._filterset({}) + self.assertNotIn('disabled_field', fs.filters) + + def test_disabled_field_query_param_ignored(self): + """Matches NetBox core: a disabled field's query param has no effect.""" + pks = list(self._filterset({'disabled_field': 'foobar'}).qs.values_list('pk', flat=True)) + self.assertEqual(len(pks), 2) + + class LongTextFieldFiltersetTestCase(ScalarFieldFiltersetTestCase, TestCase): """CharFilter with icontains is generated for TYPE_LONGTEXT fields.""" diff --git a/netbox_custom_objects/tests/test_forms.py b/netbox_custom_objects/tests/test_forms.py index 45e34dfc..119015c6 100644 --- a/netbox_custom_objects/tests/test_forms.py +++ b/netbox_custom_objects/tests/test_forms.py @@ -7,7 +7,7 @@ from extras.choices import CustomFieldTypeChoices from netbox_custom_objects.forms import CustomObjectTypeFieldForm -from netbox_custom_objects.models import CustomObjectType +from netbox_custom_objects.models import CustomObjectType, CustomObjectTypeField from .base import CustomObjectsTestCase @@ -228,3 +228,65 @@ def test_new_field_safe_related_name_is_valid(self): validation.""" form = self._make_polymorphic_object_form(related_name="co_safe_form_test_ref") self.assertTrue(form.is_valid(), form.errors) + + +class SchemaBookkeepingFieldsPreservedOnEditTestCase(CustomObjectsTestCase, TestCase): + """ + Regression #625: schema_id, deprecated, deprecated_since, and scheduled_removal + are set outside this form (auto-assigned on creation, or via the portable-schema + import mechanism) and are not rendered in any fieldset. An unrelated edit through + CustomObjectTypeFieldForm must not silently reset them. + """ + + @classmethod + def setUpTestData(cls): + cls.cot = CustomObjectType.objects.create( + name="SchemaBookkeepingTester", + slug="schema-bookkeeping-tester", + verbose_name_plural="Schema Bookkeeping Testers", + ) + + def test_editing_a_field_preserves_schema_id_and_deprecation_bookkeeping(self): + field = CustomObjectTypeField.objects.create( + custom_object_type=self.cot, + name="myfield", + label="My Field", + type=CustomFieldTypeChoices.TYPE_TEXT, + schema_id=7, + deprecated=True, + deprecated_since="1.0.0", + scheduled_removal="2.0.0", + ) + + data = { + "custom_object_type": self.cot.pk, + "name": "myfield", + "label": "My Field Renamed", + "type": CustomFieldTypeChoices.TYPE_TEXT, + "required": "", + "unique": "", + "primary": "", + "default": "", + "description": "", + "group_name": "", + "context": "default", + "search_weight": "1000", + "filter_logic": "loose", + "ui_visible": "hidden", + "ui_editable": "hidden", + "weight": "100", + "is_cloneable": "", + } + # Fetch fresh from the DB, as a real edit view does: CustomObjectTypeField.save() + # relies on ``self.original`` (populated only by from_db()) for schema-diffing. + field = CustomObjectTypeField.objects.get(pk=field.pk) + form = CustomObjectTypeFieldForm(data=data, instance=field) + self.assertTrue(form.is_valid(), form.errors) + saved = form.save() + + saved.refresh_from_db() + self.assertEqual(saved.label, "My Field Renamed") + self.assertEqual(saved.schema_id, 7) + self.assertTrue(saved.deprecated) + self.assertEqual(saved.deprecated_since, "1.0.0") + self.assertEqual(saved.scheduled_removal, "2.0.0") diff --git a/netbox_custom_objects/tests/test_jinja_integration.py b/netbox_custom_objects/tests/test_jinja_integration.py new file mode 100644 index 00000000..bfb15f50 --- /dev/null +++ b/netbox_custom_objects/tests/test_jinja_integration.py @@ -0,0 +1,318 @@ +""" +Tests for the Jinja config-template integration (jinja_env.py + PluginConfig hooks). + +The custom_objects filter and CustomObjectsNamespace are pure plugin-side code and +are always tested directly, regardless of NetBox version. End-to-end tests that +depend on NetBox actually invoking these hooks (added in NetBox 4.7 — see +netbox-community/netbox#22363, later renamed by #22436) are skipped on older +NetBox, detected via the same registry check performed by +CustomObjectsPluginConfig.ready(). +""" +from unittest.mock import patch + +import jinja2 +from django.apps import apps as django_apps +from django.test import SimpleTestCase, TestCase + +from netbox_custom_objects import CustomObjectsPluginConfig, jinja_env +from netbox_custom_objects.jinja_env import CustomObjectsNamespace, EmptyCustomObjectsQuerySet, custom_objects_filter +from netbox_custom_objects.models import CustomObjectType + +from .base import CustomObjectsTestCase + + +def _jinja_hooks_available(): + """True if this NetBox install actually registered the custom_objects filter. + + Mirrors the check in CustomObjectsPluginConfig.ready(): on NetBox < 4.7, the + jinja_filters plugin resource doesn't exist, so ready() never registers + anything under that key. + """ + from netbox.registry import registry + return 'custom_objects' in registry.get('plugins', {}).get('jinja_filters', {}) + + +class EmptyCustomObjectsQuerySetTestCase(SimpleTestCase): + """Tests for EmptyCustomObjectsQuerySet's chainable no-op interface directly.""" + + def test_read_methods_are_chainable_and_stay_empty(self): + qs = EmptyCustomObjectsQuerySet() + chained = ( + qs.filter(x=1).exclude(y=2).all().none().order_by('x') + .values('x').values_list('x').select_related('x') + .prefetch_related('x').distinct().annotate(x=1) + ) + self.assertIsInstance(chained, EmptyCustomObjectsQuerySet) + self.assertEqual(list(chained), []) + + def test_slicing_returns_self(self): + qs = EmptyCustomObjectsQuerySet() + self.assertIsInstance(qs[:5], EmptyCustomObjectsQuerySet) + + def test_integer_index_raises_index_error(self): + qs = EmptyCustomObjectsQuerySet() + with self.assertRaises(IndexError): + _ = qs[0] + + def test_get_raises_lookup_error(self): + qs = EmptyCustomObjectsQuerySet() + with self.assertRaises(LookupError): + qs.get(x=1) + + def test_first_and_last_return_none(self): + qs = EmptyCustomObjectsQuerySet() + self.assertIsNone(qs.first()) + self.assertIsNone(qs.last()) + + def test_count_and_exists(self): + qs = EmptyCustomObjectsQuerySet() + self.assertEqual(qs.count(), 0) + self.assertFalse(qs.exists()) + + def test_len_and_bool(self): + qs = EmptyCustomObjectsQuerySet() + self.assertEqual(len(qs), 0) + self.assertFalse(qs) + + +class CustomObjectsFilterTestCase(CustomObjectsTestCase, TestCase): + """Tests for custom_objects_filter() directly (no NetBox hook dependency).""" + + @classmethod + def setUpTestData(cls): + cls.cot = cls.create_custom_object_type(name='j2widget', slug='j2-widget') + cls.create_custom_object_type_field( + cls.cot, name='label', label='Label', type='text', primary=True, required=True, + ) + + def test_returns_queryset_for_known_type(self): + model = self.cot.get_model() + model.objects.create(label='alpha') + model.objects.create(label='beta') + # custom_objects_filter is @pass_context; the context argument is unused, so any + # value (None, here) is fine when calling it directly rather than through Jinja. + result = custom_objects_filter(None, 'j2widget') + self.assertEqual(result.count(), 2) + + def test_returns_empty_for_unknown_type(self): + result = custom_objects_filter(None, 'nonexistent_type') + self.assertIsInstance(result, EmptyCustomObjectsQuerySet) + self.assertEqual(list(result), []) + + def test_unknown_type_result_tolerates_further_chaining(self): + """A template that chains .filter()/.all() onto an unresolved name must not crash.""" + result = custom_objects_filter(None, 'nonexistent_type') + self.assertEqual(list(result.filter(label='x').all().exclude(label='y')), []) + + def test_unknown_name_warning_logged_once_per_process(self): + """ + A typo'd type name rendered repeatedly (e.g. across a bulk device config + export) must log its warning once, not once per lookup. + """ + unique_name = 'warn_once_filter_type' + jinja_env._warned_unknown_names.discard(unique_name) + self.addCleanup(jinja_env._warned_unknown_names.discard, unique_name) + + with patch.object(jinja_env.logger, 'warning') as mock_warning: + custom_objects_filter(None, unique_name) + custom_objects_filter(None, unique_name) + custom_objects_filter(None, unique_name) + self.assertEqual(mock_warning.call_count, 1) + + +class CustomObjectsNamespaceTestCase(CustomObjectsTestCase, TestCase): + """Tests for CustomObjectsNamespace directly (no NetBox hook dependency).""" + + @classmethod + def setUpTestData(cls): + cls.cot = cls.create_custom_object_type(name='j2widget', slug='j2-widget') + cls.create_custom_object_type_field( + cls.cot, name='label', label='Label', type='text', primary=True, required=True, + ) + + def test_resolves_by_name_to_model_manager(self): + ns = CustomObjectsNamespace() + manager = ns.j2widget + self.assertIs(manager.model, self.cot.get_model()) + + def test_manager_supports_filter(self): + model = self.cot.get_model() + model.objects.create(label='alpha') + model.objects.create(label='beta') + ns = CustomObjectsNamespace() + self.assertEqual(ns.j2widget.filter(label='alpha').count(), 1) + + def test_bracket_notation_resolves_leading_digit_type_name(self): + """ + Bracket notation is Jinja's own getitem-then-getattr fallback, not + Python's __getitem__, so it must go through an actual Jinja render. + """ + cot = self.create_custom_object_type(name='123widget', slug='123-widget') + self.create_custom_object_type_field( + cot, name='label', label='Label', type='text', primary=True, required=True, + ) + model = cot.get_model() + model.objects.create(label='alpha') + ns = CustomObjectsNamespace() + template = jinja2.Environment().from_string("{{ custom_objects['123widget'].filter(label='alpha').count() }}") + self.assertEqual(template.render(custom_objects=ns), '1') + + def test_unknown_name_returns_empty_queryset_stand_in(self): + """An unresolved name must not raise -- matches custom_objects_filter()'s behavior.""" + ns = CustomObjectsNamespace() + result = ns.no_such_type + self.assertIsInstance(result, EmptyCustomObjectsQuerySet) + self.assertEqual(list(result), []) + + def test_unknown_name_result_tolerates_further_chaining(self): + """A template that chains .filter(device=device) onto an unresolved name must not crash.""" + ns = CustomObjectsNamespace() + self.assertEqual(list(ns.no_such_type.filter(device='anything')), []) + + def test_does_not_intercept_dunder_attributes(self): + """Internal/dunder lookups (e.g. by copy.deepcopy) must not trigger a DB query.""" + ns = CustomObjectsNamespace() + with self.assertRaises(AttributeError): + _ = ns.__deepcopy__ + + def test_repeated_access_to_same_name_is_cached_within_a_render(self): + """ + A template referencing custom_objects.j2widget multiple times in one render + must resolve the Custom Object Type once, not once per reference. + """ + ns = CustomObjectsNamespace() + with patch.object(CustomObjectType.objects, 'get', wraps=CustomObjectType.objects.get) as mock_get: + ns.j2widget + ns.j2widget + ns.j2widget + self.assertEqual(mock_get.call_count, 1) + + def test_cache_is_not_shared_across_namespace_instances(self): + """Caching is per-render (per CustomObjectsNamespace instance), not global.""" + CustomObjectsNamespace().j2widget + with patch.object(CustomObjectType.objects, 'get', wraps=CustomObjectType.objects.get) as mock_get: + CustomObjectsNamespace().j2widget + self.assertEqual(mock_get.call_count, 1) + + def test_unknown_name_warning_logged_once_per_process(self): + """Repeated access to the same unresolved name must log its warning once.""" + unique_name = 'warn_once_namespace_type' + jinja_env._warned_unknown_names.discard(unique_name) + self.addCleanup(jinja_env._warned_unknown_names.discard, unique_name) + + ns = CustomObjectsNamespace() + with patch.object(jinja_env.logger, 'warning') as mock_warning: + getattr(ns, unique_name) + # A fresh namespace (new render) still shares the process-level warned set. + getattr(CustomObjectsNamespace(), unique_name) + self.assertEqual(mock_warning.call_count, 1) + + +class PluginConfigJinjaHooksTestCase(CustomObjectsTestCase, TestCase): + """Tests for CustomObjectsPluginConfig.get_jinja_context() directly.""" + + @classmethod + def setUpTestData(cls): + cls.cot = cls.create_custom_object_type(name='j2widget', slug='j2-widget') + cls.create_custom_object_type_field( + cls.cot, name='label', label='Label', type='text', primary=True, required=True, + ) + + def test_get_jinja_context_returns_custom_objects_namespace(self): + plugin_config = django_apps.get_app_config('netbox_custom_objects') + self.assertIsInstance(plugin_config, CustomObjectsPluginConfig) + ctx = plugin_config.get_jinja_context() + self.assertIn('custom_objects', ctx) + self.assertIsInstance(ctx['custom_objects'], CustomObjectsNamespace) + + def test_get_jinja_context_namespace_resolves_live_data(self): + model = self.cot.get_model() + model.objects.create(label='gamma') + plugin_config = django_apps.get_app_config('netbox_custom_objects') + ctx = plugin_config.get_jinja_context() + self.assertEqual(ctx['custom_objects'].j2widget.count(), 1) + + +class JinjaHookIntegrationTestCase(CustomObjectsTestCase, TestCase): + """ + End-to-end tests exercising the actual NetBox render pipeline. Skipped on + NetBox versions that don't expose the jinja_filters / get_jinja_context hooks. + """ + + @classmethod + def setUpTestData(cls): + cls.cot = cls.create_custom_object_type(name='j2widget', slug='j2-widget') + cls.create_custom_object_type_field( + cls.cot, name='label', label='Label', type='text', primary=True, required=True, + ) + + def setUp(self): + super().setUp() + if not _jinja_hooks_available(): + self.skipTest( + 'NetBox Jinja config template hooks (jinja_filters / get_jinja_context) ' + 'are not available in this NetBox version; requires NetBox 4.7+.' + ) + + def test_filter_syntax_available_in_render_jinja2(self): + from utilities.jinja2 import render_jinja2 + model = self.cot.get_model() + model.objects.create(label='alpha') + result = render_jinja2("{{ 'j2widget' | custom_objects | list | length }}", {}) + self.assertEqual(result, '1') + + def test_filter_syntax_resolves_only_once_when_compiled_and_rendered(self): + """ + Without @pass_context, Jinja can constant-fold the filter call at + compile time, resolving the type an extra time before render. + """ + from utilities.jinja2 import render_jinja2 + model = self.cot.get_model() + model.objects.create(label='alpha') + template_code = "{% for obj in 'j2widget' | custom_objects %}{{ obj.label }}{% endfor %}" + with patch.object(CustomObjectType.objects, 'get', wraps=CustomObjectType.objects.get) as mock_get: + result = render_jinja2(template_code, {}) + self.assertEqual(result, 'alpha') + self.assertEqual(mock_get.call_count, 1) + + def test_context_namespace_available_in_config_template_render(self): + from extras.models import ConfigTemplate + model = self.cot.get_model() + model.objects.create(label='alpha') + tmpl = ConfigTemplate( + name='test-j2', + template_code='{{ custom_objects.j2widget.all() | list | length }}', + ) + self.assertEqual(tmpl.render(), '1') + + def test_unknown_type_name_in_filter_syntax_renders_empty(self): + from utilities.jinja2 import render_jinja2 + result = render_jinja2("{{ 'no_such_type' | custom_objects | list | length }}", {}) + self.assertEqual(result, '0') + + def test_unknown_type_name_in_attribute_syntax_renders_empty(self): + """ + A template chaining .filter() onto an unresolved attribute-style name (as in + every documented example) must render no rows, not raise UndefinedError. + """ + from extras.models import ConfigTemplate + tmpl = ConfigTemplate( + name='test-j2-unknown', + template_code='{{ custom_objects.no_such_type.filter(label="x") | list | length }}', + ) + self.assertEqual(tmpl.render(), '0') + + def test_bracket_notation_in_config_template_render(self): + """A leading-digit type name isn't valid dot-notation; use bracket notation.""" + from extras.models import ConfigTemplate + cot = self.create_custom_object_type(name='123widget', slug='123-widget') + self.create_custom_object_type_field( + cot, name='label', label='Label', type='text', primary=True, required=True, + ) + model = cot.get_model() + model.objects.create(label='alpha') + tmpl = ConfigTemplate( + name='test-j2-leading-digit', + template_code="{{ custom_objects['123widget'].filter(label='alpha') | list | length }}", + ) + self.assertEqual(tmpl.render(), '1') diff --git a/netbox_custom_objects/tests/test_mixin_migration.py b/netbox_custom_objects/tests/test_mixin_migration.py index c81886dd..1d45bd1f 100644 --- a/netbox_custom_objects/tests/test_mixin_migration.py +++ b/netbox_custom_objects/tests/test_mixin_migration.py @@ -398,6 +398,11 @@ def test_safe_rename_preserves_sibling_data_and_resolves_collision(self): preserve that field's own data and leave the *other* field's column untouched -- proving the recovery guidance is actually safe to follow, not just that the collision is detected. + + save()'s rename path now calls heal_unmasked_fields() unconditionally + (merged from main's #391 Phase 2 work), so the url field's title + sub-column is restored automatically in the same save() -- a separate + manual heal_cot() call is no longer required, though still safe/idempotent. """ cot = self.create_custom_object_type(name="bcc_recover", slug="bcc-recover") self.create_custom_object_type_field(cot, name="name", label="Name", type="text", primary=True) @@ -449,13 +454,15 @@ def test_safe_rename_preserves_sibling_data_and_resolves_collision(self): ) } self.assertIn("description", columns) - # The rename also carried the physical column away from "website_title" -- - # the url field's title sub-column name is derived from its own (unchanged) - # name, not stored, so nothing renamed *it* back into existence. This is - # exactly why the guidance says to re-run the heal afterward: it re-adds - # "website_title" fresh (nullable, default ''), now unambiguously the url - # field's alone. - self.assertNotIn("website_title", columns) + # The rename carried the physical column away from "website_title" -- the + # url field's title sub-column name is derived from its own (unchanged) + # name, not stored, so nothing renamed *it* back into existence. save()'s + # heal_unmasked_fields() call re-adds "website_title" fresh (nullable, + # default ''), now unambiguously the url field's alone, within the same + # save() that performed the rename. + self.assertIn("website_title", columns) + + # A manual heal_cot() afterward must be a safe no-op (idempotent). heal_cot(cot, verbosity=0) model = cot.get_model(no_cache=True) columns = { diff --git a/netbox_custom_objects/tests/test_models.py b/netbox_custom_objects/tests/test_models.py index 9da750a2..1931bf4c 100644 --- a/netbox_custom_objects/tests/test_models.py +++ b/netbox_custom_objects/tests/test_models.py @@ -648,6 +648,26 @@ def test_custom_object_type_field_reserved_name_rejected(self): ) field.full_clean() + def test_from_db_populates_original_snapshot(self): + """ + Loading a field from the database (e.g. via a plain queryset fetch) must + succeed and populate .original from the row's own values -- this is what + save() diffs against to detect renames/type changes. Also guards + from_db()'s signature against Django versions that call it with extra + keyword arguments (e.g. Django 6.1's fetch_mode). + """ + created = self.create_custom_object_type_field( + self.custom_object_type, + name="test_field", + label="Test Field", + type="text", + ) + + fetched = CustomObjectTypeField.objects.get(pk=created.pk) + + self.assertEqual(fetched.original.name, "test_field") + self.assertEqual(fetched.original.label, "Test Field") + def test_custom_object_type_field_unique_name_per_type(self): """Test that field names must be unique within a custom object type.""" self.create_custom_object_type_field( @@ -2565,3 +2585,118 @@ def test_title_column_exists_before_deferred_title_replay_runs(self): fresh_co.website_title, "Example Site", "deferred title value must be replayed once the title column exists", ) + + +class DisplayExpressionTestCase(CustomObjectsTestCase, TestCase): + """Tests for CustomObjectType.display_expression Jinja2 rendering.""" + + def _make_cot(self, expression=''): + cot = self.create_custom_object_type( + name='ExprTest', slug='expr-test', display_expression=expression, + ) + self.create_custom_object_type_field(cot, name='make', label='Make', type='text', primary=True, required=True) + self.create_custom_object_type_field(cot, name='model', label='Model', type='text', required=False) + return cot + + def test_expression_renders_composite_name(self): + cot = self._make_cot('{{ make }} - {{ model }}') + instance = cot.get_model().objects.create(make='Cisco', model='ASR1001') + self.assertEqual(str(instance), 'Cisco - ASR1001') + + def test_expression_with_missing_field_renders_empty_string(self): + # Fields referenced in the expression but not present render as '' + cot = self._make_cot('{{ make }} / {{ nonexistent }}') + instance = cot.get_model().objects.create(make='Juniper') + self.assertEqual(str(instance), 'Juniper /') + + def test_empty_expression_falls_back_to_primary_field(self): + cot = self._make_cot('') + instance = cot.get_model().objects.create(make='Arista', model='7050') + self.assertEqual(str(instance), 'Arista') + + def test_expression_rendering_error_falls_back_to_primary_field(self): + # Invalid Jinja2 syntax must not raise — fall back silently + cot = self._make_cot('{% invalid jinja %}') + instance = cot.get_model().objects.create(make='HP') + self.assertEqual(str(instance), 'HP') + + def test_expression_empty_result_falls_back_to_primary_field(self): + # Expression that renders to empty string falls back + cot = self._make_cot('{{ nonexistent }}') + instance = cot.get_model().objects.create(make='Dell') + self.assertEqual(str(instance), 'Dell') + + def test_trailing_separator_with_blank_optional_field(self): + # When an optional field is blank, the expression renders a dangling + # separator unless the template guards it. Verify both behaviours so + # the documented {% if %} pattern is regression-tested. + cot_bare = self.create_custom_object_type( + name='SepBare', slug='sep-bare', + display_expression='{{ make }} / {{ model }}', + ) + self.create_custom_object_type_field( + cot_bare, name='make', label='Make', type='text', primary=True, required=True, + ) + self.create_custom_object_type_field( + cot_bare, name='model', label='Model', type='text', required=False, + ) + + guarded_expr = '{{ make }}{% if model %} / {{ model }}{% endif %}' + cot_guarded = self.create_custom_object_type( + name='SepGuarded', slug='sep-guarded', display_expression=guarded_expr, + ) + self.create_custom_object_type_field( + cot_guarded, name='make', label='Make', type='text', primary=True, required=True, + ) + self.create_custom_object_type_field( + cot_guarded, name='model', label='Model', type='text', required=False, + ) + + model_bare = cot_bare.get_model() + model_guarded = cot_guarded.get_model() + + bare_instance = model_bare.objects.create(make='Arista') # model is blank + guarded_instance = model_guarded.objects.create(make='Arista') # model is blank + + self.assertEqual(str(bare_instance), 'Arista /') # trailing separator + self.assertEqual(str(guarded_instance), 'Arista') # cleaned up with {% if %} + + # With model populated, both render identically. + bare_full = model_bare.objects.create(make='Cisco', model='ASR') + guarded_full = model_guarded.objects.create(make='Cisco', model='ASR') + self.assertEqual(str(bare_full), 'Cisco / ASR') + self.assertEqual(str(guarded_full), 'Cisco / ASR') + + +class DisplayExpressionFormValidationTestCase(CustomObjectsTestCase, TestCase): + """Tests for CustomObjectTypeForm.clean_display_expression().""" + + def _form(self, expression): + from netbox_custom_objects.forms import CustomObjectTypeForm + data = { + 'name': 'validtest', + 'slug': 'validtest', + 'display_expression': expression, + } + return CustomObjectTypeForm(data=data) + + def test_valid_expression_passes(self): + form = self._form('{{ make }} - {{ model }}') + # display_expression itself should not produce a validation error + form.is_valid() + self.assertNotIn('display_expression', form.errors) + + def test_blank_expression_passes(self): + form = self._form('') + form.is_valid() + self.assertNotIn('display_expression', form.errors) + + def test_invalid_jinja2_syntax_raises_validation_error(self): + form = self._form('{% invalid jinja %}') + form.is_valid() + self.assertIn('display_expression', form.errors) + + def test_unclosed_block_raises_validation_error(self): + form = self._form('{% if make %}{{ make }}') # missing {% endif %} + form.is_valid() + self.assertIn('display_expression', form.errors) diff --git a/netbox_custom_objects/tests/test_polymorphic_fields.py b/netbox_custom_objects/tests/test_polymorphic_fields.py index 6182dd99..e5cfaf56 100644 --- a/netbox_custom_objects/tests/test_polymorphic_fields.py +++ b/netbox_custom_objects/tests/test_polymorphic_fields.py @@ -1148,6 +1148,65 @@ def test_deleting_custom_object_type_drops_db_table_and_deregisters_model(self): through_model_name.lower(), django_apps.all_models.get(APP_LABEL, {}) ) + def test_remove_poly_obj_columns_succeeds_with_pending_deferred_triggers(self): + """ + remove_polymorphic_object_columns() must not raise + "cannot ALTER TABLE because it has pending trigger events" when the + source row was deleted inside the same transaction (issue #595 regression). + + The through-table created by the MULTIOBJECT field has a source_id FK: + custom_objects__poly_multi.source_id → custom_objects_.id + DEFERRABLE INITIALLY DEFERRED (Django's PostgreSQL backend default) + + When a row in custom_objects_ is deleted inside a transaction, + PostgreSQL queues a deferred trigger event associated with the REFERENCED + table (custom_objects_), not the referencing through-table. A + subsequent ALTER TABLE on that same table then fails with: + "cannot ALTER TABLE … because it has pending trigger events" + unless SET CONSTRAINTS ALL IMMEDIATE is issued first to flush the queue. + + The fix in remove_polymorphic_object_columns() issues SET CONSTRAINTS ALL + IMMEDIATE before the first ALTER TABLE, firing the deferred check against + the through-table. If the through-table rows were already deleted (as the + branching revert path does before removing the COT instance), the check + finds no FK violation and clears the pending event, allowing the ALTER + TABLE to proceed. + """ + from django.db import connection, transaction as db_transaction + from netbox_custom_objects.field_types import FIELD_TYPE_CLASS + + # Create an instance with the MULTIOBJECT populated so the through-table + # has a row referencing the source. + obj = self.model.objects.create(name="revert-repro") + obj.poly_multi.add(self.site) + + main_table = self.model._meta.db_table + through_table = self.m2m_field.through_table_name + + with db_transaction.atomic(): + with connection.cursor() as cursor: + # Step 1: delete through-table rows first (mirrors the branching + # revert path, which cascades child rows before removing the COT + # instance). + cursor.execute( + f'DELETE FROM "{through_table}" WHERE source_id = %s', [obj.pk] + ) + # Step 2: delete the COT instance via raw SQL. Django's FK + # constraints are DEFERRABLE INITIALLY DEFERRED, so PostgreSQL + # queues a deferred trigger on custom_objects_X (the referenced + # table) instead of checking the constraint immediately. + cursor.execute(f'DELETE FROM "{main_table}" WHERE id = %s', [obj.pk]) + # Step 3: remove the polymorphic OBJECT field columns. Without the + # fix, the ALTER TABLE inside remove_polymorphic_object_columns() + # fails with "cannot ALTER TABLE … because it has pending trigger + # events". The fix calls SET CONSTRAINTS ALL IMMEDIATE first, which + # fires the deferred check (no FK violation since step 1 already + # deleted the through-table row) and clears the pending event. + with connection.schema_editor() as editor: + field_type = FIELD_TYPE_CLASS[self.gfk_field.type]() + field_type.remove_polymorphic_object_columns(self.gfk_field, self.model, editor) + # Reaching here without a database exception confirms the fix is effective. + # --------------------------------------------------------------------------- # Cycle-detection: multi-hop polymorphic cycles diff --git a/netbox_custom_objects/tests/test_schema_operations.py b/netbox_custom_objects/tests/test_schema_operations.py index 740c63d1..fc85bf4d 100644 --- a/netbox_custom_objects/tests/test_schema_operations.py +++ b/netbox_custom_objects/tests/test_schema_operations.py @@ -4,15 +4,24 @@ Uses TransactionTestCase so DDL and on_commit callbacks behave exactly as they do in production (no wrapping savepoint prevents commits). """ +import threading from io import StringIO +from unittest.mock import patch from django.apps import apps +from django.contrib.contenttypes.models import ContentType from django.core.management import call_command -from django.db import connection +from django.db import IntegrityError, connection from django.test import TransactionTestCase +from django.urls import reverse +from core.models import ObjectType +from dcim.models import Site +from extras.choices import CustomFieldTypeChoices from netbox_custom_objects.constants import APP_LABEL -from netbox_custom_objects.models import CustomObjectTypeField +from netbox_custom_objects.field_types import FIELD_TYPE_CLASS +from netbox_custom_objects.models import CustomObjectType, CustomObjectTypeField +from users.models import ObjectPermission from .base import CustomObjectsTestCase, TransactionCleanupMixin @@ -321,3 +330,447 @@ def test_url_field_delete_drops_both_columns(self): columns = self._db_columns(cot.get_model()) self.assertNotIn('website', columns) self.assertNotIn('website_title', columns) + + +class PolymorphicMultiObjectConcurrencyTestCase(TransactionCleanupMixin, CustomObjectsTestCase, TransactionTestCase): + """ + Regression tests for issue #658: creating a polymorphic multiobject field + races registering its through-model class against a concurrent + get_model() call, producing a class-identity mismatch that later surfaces + as "ValueError: Cannot query 'X': Must be 'TableYModel' instance." or a + RecursionError (same symptom class as #477/#483). + """ + + def setUp(self): + super().setUp() + self.site_ot = ObjectType.objects.get_for_model(Site) + + def test_field_creation_racing_concurrent_readers_yields_consistent_through_model(self): + """ + Races field *creation* (real DB I/O) against 12 looping get_model() + readers -- the shape that reproduced #658 live. Rarely lands inside + the actual race window in-process (see the deterministic version + below), but exercises the same code path under real concurrency. + """ + cot = self.create_simple_custom_object_type(name='polyrace', slug='poly-race') + + stop = threading.Event() + reader_errors = [] + reader_errors_lock = threading.Lock() + + def reader(): + while not stop.is_set(): + try: + CustomObjectType.objects.get(pk=cot.pk).get_model(no_cache=True) + except Exception as e: # noqa: BLE001 - captured for the assertion below + with reader_errors_lock: + reader_errors.append(e) + finally: + connection.close() + + n_readers = 12 + readers = [threading.Thread(target=reader) for _ in range(n_readers)] + for t in readers: + t.start() + + try: + field = self.create_custom_object_type_field( + cot, + name='depends_on', + label='Depends On', + type='multiobject', + is_polymorphic=True, + ) + field.related_object_types.set([self.site_ot]) + finally: + stop.set() + for t in readers: + t.join() + + self.assertEqual( + reader_errors, [], + "concurrent get_model() calls must not raise while a polymorphic " + "multiobject field is being created", + ) + + # get_model() and the through model's "source" FK must agree on which + # class is canonical -- a mismatch is the #477/#483-class staleness. + final_model = CustomObjectType.objects.get(pk=cot.pk).get_model() + through_model = apps.get_model(APP_LABEL, field.through_model_name) + source_field = through_model._meta.get_field('source') + self.assertIs( + source_field.remote_field.model, final_model, + "the registered through model's source FK must point at the model class " + "get_model() currently returns, not an orphaned duplicate from a losing thread", + ) + + obj = final_model.objects.create(name='Instance 1') + obj.depends_on.set([Site.objects.create(name='Race Site', slug='race-site')]) + obj.delete() # Must not raise ValueError: "Cannot query ...: Must be ... instance." + + def test_forced_registration_interleaving_stays_consistent(self): + """ + Deterministic version of the same race, forced via mocking instead of + relying on thread-scheduling luck. + + get_model() already wraps _after_model_generation() in + CustomObjectType._global_lock, so two concurrent readers can't race + each other there. The actual gap is the *writer*: + create_polymorphic_m2m_table() (called once, from + CustomObjectTypeField.save(), when a polymorphic multiobject field is + first created) registers its through-model class via Django's + metaclass, then repoints its "source" FK -- all without that lock. A + concurrent reader can land in between: it finds the through model + already registered and repoints "source" at its own model instead, + so the through's FK and get_model()'s cache can end up pointing at + two different classes. + + Thread "W" plays the writer (create_polymorphic_m2m_table() + directly), thread "R" the reader (get_model()). A mocked + register_model() pauses W right after registration but before it + repoints "source", giving R a window to run. With the fix, W holds + _global_lock across that build+register+repoint step, so R can't + start until W has repointed "source" -- the pause below just times + out harmlessly. Without the fix, R runs inside the pause and the two + threads' writes land in different orders, reliably producing the + mismatch asserted below. See #658 for the full analysis, including + why the lock can't simply span the rest of the call too. + """ + cot = self.create_simple_custom_object_type(name='polyforce', slug='poly-force') + field = self.create_custom_object_type_field( + cot, + name='depends_on', + label='Depends On', + type='multiobject', + is_polymorphic=True, + ) + field.related_object_types.set([self.site_ot]) + + # The table/through model already exist (created for real above via + # the normal save() path). Force the through model back to + # "unregistered" so a direct create_polymorphic_m2m_table() call + # takes the same build-register-repoint path a brand-new field's + # first save would; create_polymorphic_m2m_table()'s own idempotency + # check will see the physical table already exists and skip the DDL. + writer_model = CustomObjectType.objects.get(pk=cot.pk).get_model() + CustomObjectType.clear_model_cache() + model_name_lower = field.through_model_name.lower() + del apps.all_models[APP_LABEL][model_name_lower] + apps.clear_cache() + + real_register_model = apps.register_model + reader_may_proceed = threading.Event() + reader_done = threading.Event() + gated = set() + + def ordered_register_model(app_label, model): + # Only intercept the through model under test. + if app_label != APP_LABEL or model.__name__ != field.through_model_name: + return real_register_model(app_label, model) + # Only the first call matters (Django's metaclass registers the + # model as soon as it's built; a harmless explicit re-registration + # follows immediately after in the real code). + if 'seen' in gated: + return real_register_model(app_label, model) + gated.add('seen') + + result = real_register_model(app_label, model) + # Registered, but "source" isn't repointed at writer_model yet -- + # give R a window here. With the fix, W holds _global_lock for + # this whole call, so R can't have started yet and this always + # times out rather than being signalled -- R can't reach + # reader_done.set() until W releases the lock, which doesn't + # happen until this wait returns. The duration only bounds how + # long that unavoidable wait lasts; it has no bearing on + # correctness (R's ability to run concurrently here is decided + # by lock state, not by wall-clock timing), so keep it short to + # avoid taxing every CI run by a fixed 2s. + reader_may_proceed.set() + reader_done.wait(timeout=0.5) + return result + + writer_result = {} + + def run_writer(): + threading.current_thread().name = 'W' + field_type = FIELD_TYPE_CLASS[CustomFieldTypeChoices.TYPE_MULTIOBJECT]() + try: + with connection.schema_editor() as schema_editor: + field_type.create_polymorphic_m2m_table(field, writer_model, schema_editor) + except Exception as e: # noqa: BLE001 - surfaced via the assertion below + writer_result['error'] = e + finally: + connection.close() + + reader_result = {} + + def run_reader(): + threading.current_thread().name = 'R' + reader_may_proceed.wait(timeout=5) + try: + reader_result['model'] = CustomObjectType.objects.get(pk=cot.pk).get_model(no_cache=True) + except Exception as e: # noqa: BLE001 - surfaced via the assertion below + reader_result['error'] = e + finally: + reader_done.set() + connection.close() + + with patch.object(apps, 'register_model', side_effect=ordered_register_model): + t_w = threading.Thread(target=run_writer, name='W') + t_r = threading.Thread(target=run_reader, name='R') + t_w.start() + t_r.start() + t_w.join(timeout=10) + t_r.join(timeout=10) + + # A join() timeout leaves the result dicts empty rather than raising, so without these + # checks a hung thread could silently make the assertions below vacuously pass -- e.g. a + # writer that never finished never reaches the mismatch-inducing repoint at all. + self.assertFalse(t_w.is_alive(), "writer thread did not complete within the join timeout") + self.assertFalse(t_r.is_alive(), "reader thread did not complete within the join timeout") + + self.assertNotIn('error', writer_result, f"writer raised: {writer_result.get('error')!r}") + self.assertNotIn('error', reader_result, f"reader raised: {reader_result.get('error')!r}") + + # Without the fix, this reliably produces a mismatch: reader's model + # cached while writer's model is left on the through's "source" FK, + # or vice versa. + final_model = CustomObjectType.objects.get(pk=cot.pk).get_model() + through_model = apps.get_model(APP_LABEL, field.through_model_name) + source_field = through_model._meta.get_field('source') + self.assertIs( + source_field.remote_field.model, final_model, + "the registered through model's source FK must point at the model class " + "get_model() currently returns -- a mismatch here is issue #658", + ) + + obj = final_model.objects.create(name='Instance 1') + obj.depends_on.set([Site.objects.create(name='Force Site', slug='force-site')]) + + # The delete-confirmation GET is the reported UI path (issue #640, step 4): obj.delete() + # below realigns each through's "source" FK to type(self) before Django's collector runs + # (see CustomObject.delete()), which would silently paper over a lingering registry + # mismatch. A GET here never calls delete() at all, so it exercises the raw, unrepaired + # state directly -- exactly what crashed with "ValueError: Cannot query ...: Must be ... + # instance." in the original report, and what the class-identity assertion above cannot + # by itself confirm is actually reachable through the UI. + content_type = ContentType.objects.get_for_model(final_model) + obj_perm = ObjectPermission(name='poly-force-delete-view', actions=['view', 'delete']) + obj_perm.save() + obj_perm.users.add(self.user) + obj_perm.object_types.add(content_type) + delete_url = reverse( + 'plugins:netbox_custom_objects:customobject_delete', + kwargs={'custom_object_type': cot.slug, 'pk': obj.pk}, + ) + response = self.client.get(delete_url) + self.assertEqual( + response.status_code, 200, + f"delete-confirmation GET must render, not crash (got {response.status_code})", + ) + + obj.delete() # Must not raise ValueError: "Cannot query ...: Must be ... instance." + + def test_concurrent_double_submit_does_not_deadlock(self): + """ + Two threads independently calling CustomObjectTypeField.objects.create() for the + identical (custom_object_type, name) at once -- a genuinely reachable scenario (e.g. a + retried request, or a doubly-clicked "save" button) -- must not deadlock. + + This is a real deadlock, not just a slow race, when CustomObjectType._global_lock spans + create_polymorphic_m2m_table()'s DDL: both threads build+register a through model for the + SAME physical table before either knows which one will win the (name, custom_object_type) + UniqueConstraint, so whichever thread's schema_editor.create_model() runs second blocks at + the Postgres level waiting for the first thread's uncommitted CREATE TABLE (same table + name) to resolve. If the first thread still needs the *same* Python lock afterward (its + own save() calls CustomObjectType.clear_model_cache(), which acquires it) before it can + commit and release that Postgres-level wait, neither thread can make progress. Confirmed + empirically: this exact scenario hung a live test run before the lock was narrowed to + cover only the build+register+repoint step, not the DDL. + """ + cot = self.create_simple_custom_object_type(name='doublesubmit', slug='double-submit') + self_ot = ObjectType.objects.get_for_model(cot.get_model()) + + results = {} + + def create_field(key): + threading.current_thread().name = key + try: + field = CustomObjectTypeField.objects.create( + custom_object_type=cot, + name='depends_on', + label='Depends On', + type=CustomFieldTypeChoices.TYPE_MULTIOBJECT, + is_polymorphic=True, + ) + field.related_object_types.set([self_ot, self.site_ot]) + results[key] = {'field': field} + except IntegrityError as e: + # Expected for exactly one of the two: the (name, custom_object_type) + # UniqueConstraint has only one winner. + results[key] = {'integrity_error': e} + except Exception as e: # noqa: BLE001 - surfaced via the assertion below + results[key] = {'error': e} + finally: + connection.close() + + t_a = threading.Thread(target=create_field, args=('A',), name='A') + t_b = threading.Thread(target=create_field, args=('B',), name='B') + t_a.start() + t_b.start() + t_a.join(timeout=15) + t_b.join(timeout=15) + + self.assertFalse(t_a.is_alive(), "thread A did not complete within the join timeout (deadlocked?)") + self.assertFalse(t_b.is_alive(), "thread B did not complete within the join timeout (deadlocked?)") + for key, result in results.items(): + self.assertNotIn('error', result, f"thread {key} raised an unexpected error: {result.get('error')!r}") + + succeeded = [key for key, result in results.items() if 'field' in result] + failed = [key for key, result in results.items() if 'integrity_error' in result] + self.assertEqual(len(succeeded), 1, f"expected exactly one winner: {results!r}") + self.assertEqual(len(failed), 1, f"expected exactly one IntegrityError: {results!r}") + + def test_field_creation_via_public_save_path_with_two_type_setup_yields_consistent_through_model(self): + """ + Variant of test_field_creation_racing_concurrent_readers_yields_consistent_through_model + above, using the reported two-type Custom Object setup (a self-reference plus a second, + genuine Custom Object Type -- not just a single core dcim model) instead of one type, to + match the exact reported repro. Goes through the public field-save path + (CustomObjectTypeField.objects.create()) throughout, with no private-helper shortcut. + + A deterministic, forced-interleaving version of *this specific* scenario (two threads + both calling CustomObjectTypeField.objects.create() for the identical (name, + custom_object_type) at once, paused via the same mocked apps.register_model() technique + as test_forced_registration_interleaving_stays_consistent) was attempted and abandoned: + it can genuinely deadlock rather than just race. Both threads target the same physical + through table, so the second thread's CREATE TABLE blocks at the Postgres level on the + first thread's still-open transaction; CustomObjectType._global_lock is held by the first + thread across that same window (with the fix in place); and if anything downstream in the + first thread's own save() needs that lock again (e.g. a signal handler calling + get_model()), neither thread can make progress -- confirmed by hanging an actual test + run. Real thread-scheduling luck, exercised here instead via 12 looping readers (matching + the existing single-type test above), cannot deadlock this way: no reader ever holds + transaction.atomic() open across a paused lock acquisition. + """ + cot = self.create_simple_custom_object_type(name='polypublic', slug='poly-public') + other_cot = self.create_simple_custom_object_type(name='polypublicother', slug='poly-public-other') + self_ot = ObjectType.objects.get_for_model(cot.get_model()) + other_ot = ObjectType.objects.get_for_model(other_cot.get_model()) + + stop = threading.Event() + reader_errors = [] + reader_errors_lock = threading.Lock() + + def reader(): + while not stop.is_set(): + try: + CustomObjectType.objects.get(pk=cot.pk).get_model(no_cache=True) + except Exception as e: # noqa: BLE001 - captured for the assertion below + with reader_errors_lock: + reader_errors.append(e) + finally: + connection.close() + + n_readers = 12 + readers = [threading.Thread(target=reader) for _ in range(n_readers)] + for t in readers: + t.start() + + try: + field = CustomObjectTypeField.objects.create( + custom_object_type=cot, + name='depends_on', + label='Depends On', + type=CustomFieldTypeChoices.TYPE_MULTIOBJECT, + is_polymorphic=True, + ) + field.related_object_types.set([self_ot, other_ot]) + finally: + stop.set() + for t in readers: + t.join() + + self.assertEqual( + reader_errors, [], + "concurrent get_model() calls must not raise while a polymorphic multiobject field " + "with the reported two-type setup is being created", + ) + self.assertEqual(set(field.related_object_types.all()), {self_ot, other_ot}) + + # get_model() and the through model's "source" FK must agree on which class is canonical + # -- a mismatch is the #477/#483-class staleness that issue #658 reported. + final_model = CustomObjectType.objects.get(pk=cot.pk).get_model() + through_model = apps.get_model(APP_LABEL, field.through_model_name) + source_field = through_model._meta.get_field('source') + self.assertIs( + source_field.remote_field.model, final_model, + "the registered through model's source FK must point at the model class " + "get_model() currently returns, not an orphaned duplicate from a losing thread", + ) + + obj = final_model.objects.create(name='Instance 1') + obj.depends_on.set([Site.objects.create(name='Public Race Site', slug='public-race-site')]) + obj.delete() # Must not raise ValueError: "Cannot query ...: Must be ... instance." + + +class OwnerFieldNameCollisionTestCase(TransactionCleanupMixin, CustomObjectsTestCase, TransactionTestCase): + """A user field named 'owner' shadows CustomObject's own OwnerMixin.owner field. + Renaming or deleting it must not crash on the unmasked mixin field's missing column. + """ + + def _db_columns(self, model): + with connection.cursor() as cursor: + return { + col.name + for col in connection.introspection.get_table_description( + cursor, model._meta.db_table + ) + } + + def _make_owner_collision_cot(self, slug): + from core.models import ObjectType + contact_ot = ObjectType.objects.get(app_label='tenancy', model='contact') + + cot = self.create_custom_object_type(name=slug, slug=slug) + self.create_custom_object_type_field( + cot, name='name', label='Name', type='text', primary=True, required=True, + ) + # .objects.create() bypasses the reserved-name check in clean(). + field = self.create_custom_object_type_field( + cot, name='owner', label='Owner', type='object', related_object_type=contact_ot, + ) + return cot, field + + def test_renaming_away_from_owner_does_not_break_list_view(self): + cot, field = self._make_owner_collision_cot('owner-rename') + model = cot.get_model() + model.objects.create(name='Obj1') + model.objects.create(name='Obj2') + + field = CustomObjectTypeField.objects.get(pk=field.pk) + field.name = 'client_contact' + field.save() + + updated_model = cot.get_model(no_cache=True) + columns = self._db_columns(updated_model) + self.assertIn('client_contact_id', columns) + self.assertIn('owner_id', columns, "OwnerMixin's own column must be healed back in") + + results = list(updated_model.objects.all()) + self.assertEqual(len(results), 2) + + def test_deleting_owner_field_does_not_break_list_view(self): + cot, field = self._make_owner_collision_cot('owner-delete') + model = cot.get_model() + model.objects.create(name='Obj1') + + field = CustomObjectTypeField.objects.get(pk=field.pk) + field.delete() + + updated_model = cot.get_model() + columns = self._db_columns(updated_model) + self.assertIn('owner_id', columns) + + results = list(updated_model.objects.all()) + self.assertEqual(len(results), 1) diff --git a/netbox_custom_objects/tests/test_views.py b/netbox_custom_objects/tests/test_views.py index 0cdd46a4..2461e651 100644 --- a/netbox_custom_objects/tests/test_views.py +++ b/netbox_custom_objects/tests/test_views.py @@ -2,18 +2,42 @@ Tests for all UI views. """ from django.contrib.contenttypes.models import ContentType -from django.test import TestCase +from django.db import connection +from django.test import RequestFactory, TestCase from django.urls import reverse from extras.models import CustomFieldChoiceSet from users.models import ObjectPermission from utilities.testing import ViewTestCases, create_test_user +from netbox_custom_objects import views from netbox_custom_objects.models import CustomObjectType, CustomObjectTypeField from .base import CustomObjectsTestCase from core.models.object_types import ObjectType +try: + import netbox_branching # noqa: F401 + _HAS_BRANCHING = True +except ImportError: + _HAS_BRANCHING = False -class CustomObjectTypeViewTestCase(CustomObjectsTestCase, ViewTestCases.PrimaryObjectViewTestCase): + +class _SkipQueryCountsWhenBranching: + """Skip query-count assertion when netbox-branching is installed. + + Branching adds per-request queries (branch lookup, schema check, etc.) that + are not present in the recorded baselines. The counts are adequately tested + by the non-branching matrix jobs. + """ + + def test_list_objects_with_permission(self): + if _HAS_BRANCHING: + self.skipTest('query-count baselines not valid with netbox-branching installed') + super().test_list_objects_with_permission() + + +class CustomObjectTypeViewTestCase( + _SkipQueryCountsWhenBranching, CustomObjectsTestCase, ViewTestCases.PrimaryObjectViewTestCase +): """Test cases for CustomObjectType views.""" model = CustomObjectType @@ -230,7 +254,9 @@ def test_bulk_delete_objects_with_constrained_permission(self): ... -class CustomObjectViewTestCase(CustomObjectsTestCase, ViewTestCases.PrimaryObjectViewTestCase): +class CustomObjectViewTestCase( + _SkipQueryCountsWhenBranching, CustomObjectsTestCase, ViewTestCases.PrimaryObjectViewTestCase +): """Test cases for dynamic CustomObject views.""" query_count_model_label = 'customobject-simple' @@ -369,6 +395,374 @@ def test_bulk_delete_objects_with_permission(self): def test_bulk_delete_objects_with_constrained_permission(self): ... + def _assert_get_queryset_does_not_full_scan(self, view_class): + """Regression #620 helper. + + All three bulk views (import/edit/delete) previously did + ``if self.queryset:`` in ``get_queryset()``, whose ``QuerySet.__bool__`` + calls ``_fetch_all()`` — pulling every row into memory. On a type with + millions of records this spiked server memory and hung the request. + + ``get_queryset()`` is invoked a second time by ``BaseMultiObjectView. + dispatch()`` after ``setup()`` has assigned the (lazy) queryset, which is + when the truthiness check evaluated it. We reproduce that state directly + so the assertion targets the exact regression regardless of each view's + HTTP method handling. + """ + request = RequestFactory().get('/') + request.user = self.user + + view = view_class() + view.kwargs = {'custom_object_type': self.model.custom_object_type.slug} + # Post-setup state: dispatch() will have left a lazy, unevaluated queryset. + view.queryset = self.model.objects.all() + + db_table = self.model._meta.db_table + full_scans = [] + + def tracer(execute, sql, params, many, context): + normalized = sql.lstrip().upper() + if ( + db_table in sql + and normalized.startswith('SELECT') + and 'LIMIT' not in normalized + and 'COUNT(' not in normalized + ): + full_scans.append(sql) + return execute(sql, params, many, context) + + with connection.execute_wrapper(tracer): + view.get_queryset(request) + + self.assertEqual( + full_scans, [], + f"{view_class.__name__}.get_queryset() issued an unbounded SELECT " + f"against {db_table}; the whole table is being loaded into memory:\n" + + "\n".join(full_scans), + ) + + def test_bulk_import_get_queryset_does_not_full_scan(self): + """Regression #620: CustomObjectBulkImportView.get_queryset().""" + self._assert_get_queryset_does_not_full_scan(views.CustomObjectBulkImportView) + + def test_bulk_edit_get_queryset_does_not_full_scan(self): + """Regression #620: CustomObjectBulkEditView.get_queryset().""" + self._assert_get_queryset_does_not_full_scan(views.CustomObjectBulkEditView) + + def test_bulk_edit_form_nullable_fields_includes_scalar_fields(self): + """Regression #621: bulk edit must offer 'Set null' for real, nullable fields.""" + request = RequestFactory().get('/') + request.user = self.user + + view = views.CustomObjectBulkEditView() + view.setup(request, custom_object_type=self.custom_object_type.slug) + + self.assertIn('description', view.form.nullable_fields) + self.assertIn('count', view.form.nullable_fields) + # 'name' is required=True; a required field must never offer "Set null", + # since every custom object column is nullable at the DB level regardless + # of the field's own required flag. + self.assertNotIn('name', view.form.nullable_fields) + + def test_bulk_edit_set_null_clears_field(self): + """Regression #621: checking 'Set null' for a field must clear it across selected objects.""" + content_type = ContentType.objects.get_for_model(self.model) + obj_perm = ObjectPermission(name='bulk-edit-set-null', actions=['view', 'change']) + obj_perm.save() + obj_perm.users.add(self.user) + obj_perm.object_types.add(content_type) + + bulk_edit_url = self._get_url('bulk_edit') + response = self.client.post(bulk_edit_url, data={ + '_apply': 'Apply', + 'pk': [self.instance1.pk, self.instance2.pk], + '_nullify': ['description'], + 'description': '', + }) + self.assertHttpStatus(response, 302) + self.instance1.refresh_from_db() + self.instance2.refresh_from_db() + self.assertIsNone(self.instance1.description) + self.assertIsNone(self.instance2.description) + + def test_bulk_edit_set_null_clears_object_and_multiobject_fields(self): + """ + Regression #621: non-polymorphic object/multiobject fields must also support + 'Set null' in bulk edit -- they map to a real, nullable FK column / M2M relation + (like core's Site.asns), unlike polymorphic fields which are excluded. + """ + from dcim.models import Site + + cot = self.create_custom_object_type(name='ObjNullTest', slug='obj-null-test') + self.create_custom_object_type_field( + cot, name='name', label='Name', type='text', primary=True, required=True, + ) + self.create_custom_object_type_field( + cot, name='site', label='Site', type='object', + related_object_type=self.get_site_object_type(), + ) + self.create_custom_object_type_field( + cot, name='sites', label='Sites', type='multiobject', + related_object_type=self.get_site_object_type(), + ) + + model = cot.get_model() + site_a = Site.objects.create(name='Site A', slug='site-a') + site_b = Site.objects.create(name='Site B', slug='site-b') + obj1 = model.objects.create(name='Obj 1', site=site_a) + obj1.sites.set([site_a, site_b]) + obj2 = model.objects.create(name='Obj 2', site=site_a) + obj2.sites.set([site_a, site_b]) + + content_type = ContentType.objects.get_for_model(model) + obj_perm = ObjectPermission(name='bulk-edit-set-null-obj', actions=['view', 'change']) + obj_perm.save() + obj_perm.users.add(self.user) + obj_perm.object_types.add(content_type) + + url_format = 'plugins:{}:customobject_{{}}'.format(model._meta.app_label) + bulk_edit_url = reverse(url_format.format('bulk_edit'), kwargs={'custom_object_type': cot.slug}) + response = self.client.post(bulk_edit_url, data={ + '_apply': 'Apply', + 'pk': [obj1.pk, obj2.pk], + '_nullify': ['site', 'sites'], + 'site': '', + 'sites': [], + }) + self.assertHttpStatus(response, 302) + obj1.refresh_from_db() + obj2.refresh_from_db() + self.assertIsNone(obj1.site) + self.assertIsNone(obj2.site) + self.assertEqual(obj1.sites.count(), 0) + self.assertEqual(obj2.sites.count(), 0) + + def test_bulk_delete_get_queryset_does_not_full_scan(self): + """Regression #620: CustomObjectBulkDeleteView.get_queryset().""" + self._assert_get_queryset_does_not_full_scan(views.CustomObjectBulkDeleteView) + + def test_bulk_import_omits_hidden_required_field_from_form(self): + """ + Regression #626: a Required+Hidden field must be omitted from the bulk import + form (not disabled, which ignores submitted data and always fails "This field + is required"). Own COT used to avoid leaking model-cache state into other tests. + """ + cot = self.create_custom_object_type(name='HiddenFieldImportTest', slug='hidden-field-import-test') + self.create_custom_object_type_field( + cot, name='name', label='Name', type='text', primary=True, + ) + self.create_custom_object_type_field( + cot, + name='identifier', + label='Identifier', + type='text', + required=True, + ui_editable='hidden', + ) + + request = RequestFactory().post('/') + request.user = self.user + + view = views.CustomObjectBulkImportView() + view.setup(request, custom_object_type=cot.slug) + + # The hidden required field must not appear in the import form at all... + self.assertNotIn('identifier', view.model_form.base_fields) + + # ...so a row that omits it (as any importer must, since it can't be set) is valid. + model = view.queryset.model + form = view.model_form(data={'name': 'Imported Instance'}, instance=model()) + self.assertTrue(form.is_valid(), form.errors) + + def test_bulk_import_silently_ignores_value_for_hidden_field(self): + """ + Regression #626: matches the original bug report's payload (a value supplied + for the hidden field). It must not error, and the value must be silently + dropped rather than written, matching core NetBox's own CSV import behavior. + """ + cot = self.create_custom_object_type(name='HiddenFieldImportTest2', slug='hidden-field-import-test-2') + self.create_custom_object_type_field( + cot, name='name', label='Name', type='text', primary=True, + ) + self.create_custom_object_type_field( + cot, + name='identifier', + label='Identifier', + type='text', + required=True, + ui_editable='hidden', + ) + + request = RequestFactory().post('/') + request.user = self.user + + view = views.CustomObjectBulkImportView() + view.setup(request, custom_object_type=cot.slug) + + model = view.queryset.model + form = view.model_form( + data={'name': 'Imported Instance', 'identifier': '12345'}, instance=model(), + ) + self.assertTrue(form.is_valid(), form.errors) + + instance = form.save() + self.assertIsNone(instance.identifier) + + def test_edit_form_omits_hidden_field(self): + """Regression #645: a hidden field must be omitted from the edit form, not just disabled.""" + cot = self.create_custom_object_type(name='HiddenFieldEditTest', slug='hidden-field-edit-test') + self.create_custom_object_type_field( + cot, name='name', label='Name', type='text', primary=True, + ) + self.create_custom_object_type_field( + cot, name='hidden', label='Hidden', type='text', ui_editable='hidden', + ) + self.create_custom_object_type_field( + cot, name='readonly', label='Readonly', type='text', ui_editable='no', + ) + + request = RequestFactory().get('/') + request.user = self.user + + view = views.CustomObjectEditView() + view.setup(request, custom_object_type=cot.slug) + + self.assertNotIn('hidden', view.form.base_fields) + # A read-only (ui_editable=no) field is disabled, not omitted -- distinct from hidden. + self.assertIn('readonly', view.form.base_fields) + + def test_bulk_edit_form_omits_hidden_field(self): + """Regression #645: same as above, for the bulk edit form.""" + cot = self.create_custom_object_type(name='HiddenFieldBulkEditTest', slug='hidden-field-bulk-edit-test') + self.create_custom_object_type_field( + cot, name='name', label='Name', type='text', primary=True, + ) + self.create_custom_object_type_field( + cot, name='hidden', label='Hidden', type='text', ui_editable='hidden', + ) + self.create_custom_object_type_field( + cot, name='readonly', label='Readonly', type='text', ui_editable='no', + ) + + request = RequestFactory().get('/') + request.user = self.user + + view = views.CustomObjectBulkEditView() + view.setup(request, custom_object_type=cot.slug) + + self.assertNotIn('hidden', view.form.base_fields) + self.assertIn('readonly', view.form.base_fields) + + def test_edit_form_omits_hidden_coordinates_field(self): + """Regression #645: a hidden coordinates field must omit both its lat/long sub-fields.""" + cot = self.create_custom_object_type(name='HiddenCoordsEditTest', slug='hidden-coords-edit-test') + self.create_custom_object_type_field( + cot, name='name', label='Name', type='text', primary=True, + ) + self.create_custom_object_type_field( + cot, name='location', label='Location', type='coordinates', ui_editable='hidden', + ) + + request = RequestFactory().get('/') + request.user = self.user + + view = views.CustomObjectEditView() + view.setup(request, custom_object_type=cot.slug) + + self.assertNotIn('location_latitude', view.form.base_fields) + self.assertNotIn('location_longitude', view.form.base_fields) + + def test_bulk_edit_form_omits_hidden_coordinates_field(self): + """Regression #645: same as above, for the bulk edit form.""" + cot = self.create_custom_object_type(name='HiddenCoordsBulkEditTest', slug='hidden-coords-bulk-edit-test') + self.create_custom_object_type_field( + cot, name='name', label='Name', type='text', primary=True, + ) + self.create_custom_object_type_field( + cot, name='location', label='Location', type='coordinates', ui_editable='hidden', + ) + + request = RequestFactory().get('/') + request.user = self.user + + view = views.CustomObjectBulkEditView() + view.setup(request, custom_object_type=cot.slug) + + self.assertNotIn('location_latitude', view.form.base_fields) + self.assertNotIn('location_longitude', view.form.base_fields) + + def test_edit_form_applies_hidden_multiobject_default_on_create(self): + """ + Regression #42/#645: a hidden non-polymorphic MultiObject field is a real M2M + model attribute, so omitting it from the rendered form must not also skip + applying its configured default when creating a new object. + """ + from dcim.models import Site + + site_a = Site.objects.create(name='Site A', slug='site-a-hidden-mo-create') + site_b = Site.objects.create(name='Site B', slug='site-b-hidden-mo-create') + + cot = self.create_custom_object_type(name='HiddenMultiObjCreateTest', slug='hidden-multiobj-create-test') + self.create_custom_object_type_field( + cot, name='name', label='Name', type='text', primary=True, + ) + self.create_custom_object_type_field( + cot, name='sites', label='Sites', type='multiobject', + related_object_type=self.get_site_object_type(), + ui_editable='hidden', + default=[site_a.pk, site_b.pk], + ) + + request = RequestFactory().post('/') + request.user = self.user + + view = views.CustomObjectEditView() + view.setup(request, custom_object_type=cot.slug) + + form = view.form(data={'name': 'New Instance'}, instance=view.object) + self.assertTrue(form.is_valid(), form.errors) + instance = form.save() + + self.assertEqual(set(instance.sites.values_list('pk', flat=True)), {site_a.pk, site_b.pk}) + + def test_edit_form_preserves_hidden_multiobject_relation_on_edit(self): + """ + Regression #645: editing an object must not clear a hidden MultiObject field's + existing relation -- there is no rendered input for it to come from, and a + hidden field is defined as neither displayed nor editable. + """ + from dcim.models import Site + + site_a = Site.objects.create(name='Site A', slug='site-a-hidden-mo-edit') + site_b = Site.objects.create(name='Site B', slug='site-b-hidden-mo-edit') + + cot = self.create_custom_object_type(name='HiddenMultiObjEditTest', slug='hidden-multiobj-edit-test') + self.create_custom_object_type_field( + cot, name='name', label='Name', type='text', primary=True, + ) + self.create_custom_object_type_field( + cot, name='sites', label='Sites', type='multiobject', + related_object_type=self.get_site_object_type(), + ui_editable='hidden', + default=[site_a.pk], + ) + + model = cot.get_model() + obj = model.objects.create(name='Existing Instance') + obj.sites.set([site_a.pk, site_b.pk]) + + request = RequestFactory().post('/') + request.user = self.user + + view = views.CustomObjectEditView() + view.setup(request, custom_object_type=cot.slug, pk=obj.pk) + + form = view.form(data={'name': 'Renamed Instance'}, instance=view.object) + self.assertTrue(form.is_valid(), form.errors) + instance = form.save() + + self.assertEqual(set(instance.sites.values_list('pk', flat=True)), {site_a.pk, site_b.pk}) + def test_bulk_edit_select_all_respects_full_queryset(self): """Regression #380: 'select all matching query' must edit all objects, not just the current page. @@ -458,7 +852,9 @@ def test_add_permission_is_sufficient_to_access_add_url(self): self.assertHttpStatus(self.client.get(edit_url), 200) -class ComplexCustomObjectViewTestCase(CustomObjectsTestCase, ViewTestCases.PrimaryObjectViewTestCase): +class ComplexCustomObjectViewTestCase( + _SkipQueryCountsWhenBranching, CustomObjectsTestCase, ViewTestCases.PrimaryObjectViewTestCase +): """Test cases for complex custom objects with various field types.""" query_count_model_label = 'customobject-complex' @@ -754,7 +1150,9 @@ def test_detail_view_renders_label_for_uncolored_select_field(self): self.assertIn('Yes', response.content.decode()) -class ObjectFieldViewTestCase(CustomObjectsTestCase, ViewTestCases.PrimaryObjectViewTestCase): +class ObjectFieldViewTestCase( + _SkipQueryCountsWhenBranching, CustomObjectsTestCase, ViewTestCases.PrimaryObjectViewTestCase +): """Test cases for custom objects with object and multi-object fields.""" query_count_model_label = 'customobject-objectfields' @@ -1086,6 +1484,35 @@ def test_bulk_edit_half_populated_pair_rejected(self): self.assertEqual(obj.location_latitude, Decimal("40.712800")) self.assertEqual(obj.location_longitude, Decimal("-74.006000")) + def test_bulk_edit_coordinates_not_individually_nullable(self): + """Regression: latitude/longitude must not get independent Set Null controls.""" + request = RequestFactory().get('/') + request.user = self.user + + view = views.CustomObjectBulkEditView() + view.setup(request, custom_object_type=self.cot.slug) + self.assertNotIn('location_latitude', view.form.nullable_fields) + self.assertNotIn('location_longitude', view.form.nullable_fields) + + def test_bulk_edit_set_null_clears_coordinates_atomically(self): + """A single 'Set null' checkbox for the coordinates field clears both halves.""" + from decimal import Decimal + obj = self.model.objects.create( + name="Existing2", + location_latitude=Decimal("40.712800"), + location_longitude=Decimal("-74.006000"), + ) + data = { + "pk": [obj.pk], + "_apply": "Apply", + "_nullify": ["location"], + } + response = self.client.post(self._bulk_edit_url(), data) + self.assertEqual(response.status_code, 302, getattr(response, "content", b"")) + obj.refresh_from_db() + self.assertIsNone(obj.location_latitude) + self.assertIsNone(obj.location_longitude) + class URLFieldLinkTitleViewTest(CustomObjectsTestCase, TestCase): """UI form behaviour for the url field type's link-title support (issue #496).""" diff --git a/netbox_custom_objects/views.py b/netbox_custom_objects/views.py index 8131506b..d469d2fc 100644 --- a/netbox_custom_objects/views.py +++ b/netbox_custom_objects/views.py @@ -16,7 +16,7 @@ from django.utils.translation import gettext_lazy as _ from utilities.exceptions import AbortRequest, PermissionsViolation from django.views.generic import View -from extras.choices import CustomFieldUIVisibleChoices +from extras.choices import CustomFieldUIEditableChoices, CustomFieldUIVisibleChoices from extras.forms import JournalEntryForm from extras.models import ConfigContext, JournalEntry from extras.tables import JournalEntryTable @@ -60,6 +60,27 @@ def _is_in_branch(): return False +def _hidden_field_raw_columns(fields): + """Return backing column name(s) for HIDDEN fields, for a ModelForm's Meta.exclude. + + Polymorphic fields aren't handled here: TYPE_OBJECT's raw columns are + already excluded unconditionally by callers, and TYPE_MULTIOBJECT has + no raw column (backed by a through table). + """ + columns = [] + for f in fields: + if f.ui_editable != CustomFieldUIEditableChoices.HIDDEN or f.is_polymorphic: + continue + if f.type == CustomObjectFieldTypeChoices.TYPE_COORDINATES: + columns += [ + field_types.CoordinatesFieldType.latitude_field_name(f), + field_types.CoordinatesFieldType.longitude_field_name(f), + ] + else: + columns.append(f.name) + return columns + + # --------------------------------------------------------------------------- # Sub-field naming helpers for polymorphic form fields # @@ -737,14 +758,21 @@ def get_object(self, **kwargs): return get_object_or_404(model.objects.all(), **self.kwargs) def get_form(self, model): + cot_fields = list( + self.object.custom_object_type.fields.prefetch_related( + 'related_object_types' + ).order_by("group_name", "weight", "name") + ) + # Collect raw GFK column names to exclude from the auto-generated form fields. # For each polymorphic Object field "foo", Django adds "foo_content_type" and # "foo_object_id" as real model columns; we replace those with per-type selects. poly_obj_raw_exclude = [] - for f in self.object.custom_object_type.fields.filter( - type=CustomFieldTypeChoices.TYPE_OBJECT, is_polymorphic=True - ): - poly_obj_raw_exclude += [f"{f.name}_content_type", f"{f.name}_object_id"] + for f in cot_fields: + if f.type == CustomFieldTypeChoices.TYPE_OBJECT and f.is_polymorphic: + poly_obj_raw_exclude += [f"{f.name}_content_type", f"{f.name}_object_id"] + + hidden_raw_exclude = _hidden_field_raw_columns(cot_fields) meta = type( "Meta", @@ -752,7 +780,7 @@ def get_form(self, model): { "model": model, "fields": "__all__", - "exclude": poly_obj_raw_exclude, + "exclude": list(set(poly_obj_raw_exclude + hidden_raw_exclude)), }, ) @@ -780,9 +808,17 @@ def get_form(self, model): } # Process custom object type fields (with grouping) - for field in self.object.custom_object_type.fields.prefetch_related( - 'related_object_types' - ).order_by("group_name", "weight", "name"): + for field in cot_fields: + # Hidden fields are omitted entirely, not just disabled -- but a + # non-polymorphic MultiObject field is a real M2M model attribute + # whose configured default must still be applied on create (#42). + # Track it in custom_object_type_fields (bookkeeping only, not + # rendered anywhere) so custom_init/custom_save still process it. + if field.ui_editable == CustomFieldUIEditableChoices.HIDDEN: + if field.type == CustomFieldTypeChoices.TYPE_MULTIOBJECT and not field.is_polymorphic: + attrs["custom_object_type_fields"][field.name] = field + continue + field_type = field_types.FIELD_TYPE_CLASS[field.type]() group_name = field.group_name or None @@ -884,6 +920,10 @@ def custom_init(self, *args, **kwargs): self.custom_object_type_poly_obj_ct_names = attrs["custom_object_type_poly_obj_ct_names"] self.custom_object_type_poly_obj_pairs = attrs["custom_object_type_poly_obj_pairs"] self.custom_object_type_coordinates_fields = attrs["custom_object_type_coordinates_fields"] + # A hidden MultiObject field has no rendered form field, so its resolved + # default can't reach cleaned_data via kwargs['initial'] below -- custom_save + # applies these directly on create instead. See the note in the loop above. + self._hidden_multiobject_defaults = {} instance = kwargs.get('instance', None) @@ -914,6 +954,8 @@ def custom_init(self, *args, **kwargs): .values_list('pk', flat=True) ) kwargs['initial'][field_name] = initial_ids + if field_obj.ui_editable == CustomFieldUIEditableChoices.HIDDEN: + self._hidden_multiobject_defaults[field_name] = initial_ids except Exception: logger.debug( "Failed to load default initial values for field %r", @@ -1005,6 +1047,7 @@ def custom_init(self, *args, **kwargs): # Create a custom save method to properly handle M2M fields def custom_save(self, commit=True): instance = forms.NetBoxModelForm.save(self, commit=False) + is_new = instance.pk is None if commit: # Set polymorphic GFK attributes before the first save so the row @@ -1017,7 +1060,16 @@ def custom_save(self, commit=True): # Handle non-polymorphic M2M fields (require PK, so after save) for field_name, field_obj in self.custom_object_type_fields.items(): if field_obj.type == CustomFieldTypeChoices.TYPE_MULTIOBJECT: - current_value = self.cleaned_data.get(field_name, []) + if field_obj.ui_editable == CustomFieldUIEditableChoices.HIDDEN: + # No rendered input to read from cleaned_data: apply the + # resolved default on create (see custom_init), and leave + # existing relations alone on edit -- consistent with a + # hidden field being neither displayed nor editable. + if not is_new: + continue + current_value = self._hidden_multiobject_defaults.get(field_name, []) + else: + current_value = self.cleaned_data.get(field_name, []) instance_field = getattr(instance, field_name) if hasattr(instance_field, 'clear') and hasattr(instance_field, 'set'): instance_field.clear() @@ -1170,7 +1222,7 @@ def setup(self, request, *args, **kwargs): self.table = self.get_table(self.queryset, request).__class__ def get_queryset(self, request): - if self.queryset: + if self.queryset is not None: return self.queryset custom_object_type = self.kwargs.get("custom_object_type", None) self.custom_object_type = CustomObjectType.objects.get( @@ -1180,11 +1232,14 @@ def get_queryset(self, request): return model.objects.all() def get_form(self, queryset): + cot_fields = list(self.custom_object_type.fields.prefetch_related('related_object_types')) + poly_obj_raw_exclude = [] - for f in self.custom_object_type.fields.filter( - type=CustomFieldTypeChoices.TYPE_OBJECT, is_polymorphic=True - ): - poly_obj_raw_exclude += [f"{f.name}_content_type", f"{f.name}_object_id"] + for f in cot_fields: + if f.type == CustomFieldTypeChoices.TYPE_OBJECT and f.is_polymorphic: + poly_obj_raw_exclude += [f"{f.name}_content_type", f"{f.name}_object_id"] + + hidden_raw_exclude = _hidden_field_raw_columns(cot_fields) meta = type( "Meta", @@ -1192,16 +1247,16 @@ def get_form(self, queryset): { "model": queryset.model, "fields": "__all__", - "exclude": poly_obj_raw_exclude, + "exclude": list(set(poly_obj_raw_exclude + hidden_raw_exclude)), }, ) # Pre-build ct_pk → model_class lookup for each poly obj field so the # bulk edit __init__ can wire up the obj picker without a DB query. poly_obj_allowed = {} - for f in self.custom_object_type.fields.filter( - type=CustomFieldTypeChoices.TYPE_OBJECT, is_polymorphic=True - ).prefetch_related('related_object_types'): + for f in cot_fields: + if not (f.type == CustomFieldTypeChoices.TYPE_OBJECT and f.is_polymorphic): + continue poly_obj_allowed[f.name] = { ot.pk: ot.model_class() for ot in f.related_object_types.all() @@ -1220,9 +1275,22 @@ def get_form(self, queryset): "custom_object_type_rendered_names": set(), # field_name → (latitude_field_name, longitude_field_name) "custom_object_type_coordinates_fields": {}, + # latitude_field_name → (sub_names, field_label, field_name); drives a single + # shared "Set null" control so lat/long clear atomically (see post_save_operations). + "custom_object_type_coordinates_groups": {}, } - for field in self.custom_object_type.fields.prefetch_related('related_object_types').all(): + # Names added here get a "Set null" checkbox (nullable_fields). Required fields are + # excluded, matching core's convention of never offering Set Null on a required + # field. Polymorphic fields are also excluded: their sub-field names (e.g. "__ct") + # aren't real model fields, so core's generic nullify lookup can't resolve them. + nullable_field_names = [] + + for field in cot_fields: + # Hidden fields are omitted entirely, not just disabled. + if field.ui_editable == CustomFieldUIEditableChoices.HIDDEN: + continue + field_type = field_types.FIELD_TYPE_CLASS[field.type]() # Coordinates: two optional latitude/longitude inputs in bulk edit @@ -1236,6 +1304,16 @@ def get_form(self, queryset): sub_names.append(sub_name) # (latitude_name, longitude_name) for cross-field validation below. attrs["custom_object_type_coordinates_fields"][field.name] = tuple(sub_names) + # Not added to nullable_field_names: latitude/longitude are one logical + # field, so a shared checkbox (below) clears both atomically instead of + # offering two independent "Set null" controls for a single value. + if not field.required: + field_label = field.label or field.name.replace("_", " ").title() + attrs["custom_object_type_coordinates_groups"][sub_names[0]] = ( + tuple(sub_names), field_label, field.name, + ) + for sub_name in sub_names: + attrs["custom_object_type_rendered_names"].add(sub_name) continue # URL: two optional url/title inputs in bulk edit. No cross-field @@ -1285,11 +1363,15 @@ def get_form(self, queryset): form_field.widget.is_required = False form_field.initial = None attrs[field.name] = form_field + if not field.required: + nullable_field_names.append(field.name) except NotImplementedError: logger.debug( "bulk edit form: {} field is not supported".format(field.name) ) + attrs["nullable_fields"] = tuple(nullable_field_names) + poly_obj_field_map_ref = attrs["_poly_obj_field_map"] poly_grouping_refs = { @@ -1350,6 +1432,20 @@ def bulk_clean(self): def post_save_operations(self, form, obj): super().post_save_operations(form, obj) + # Coordinates: a single "Set null" checkbox clears both lat/long sub-columns + # atomically. They're deliberately absent from form.nullable_fields (see + # get_form()), so core's generic per-field nullify loop never touches them -- + # handled here instead, reading the same raw _nullify POST data core parses. + nullified = self.request.POST.getlist('_nullify') + coords_needs_save = False + for field_name, (lat_name, lon_name) in form.custom_object_type_coordinates_fields.items(): + if field_name in nullified: + setattr(obj, lat_name, None) + setattr(obj, lon_name, None) + coords_needs_save = True + if coords_needs_save: + obj.save() + # Apply polymorphic single-object scope fields: read the obj sub-field needs_save = False for field_name, (ct_sub, obj_sub) in form._poly_obj_field_map.items(): @@ -1406,7 +1502,7 @@ def setup(self, request, *args, **kwargs): self.table = self.get_table(self.queryset, request).__class__ def get_queryset(self, request): - if self.queryset: + if self.queryset is not None: return self.queryset self.custom_object_type = self.kwargs.pop("custom_object_type", None) self.custom_object_type = CustomObjectType.objects.get( @@ -1437,7 +1533,7 @@ def setup(self, request, *args, **kwargs): self.model_form = self.get_model_form(self.queryset) def get_queryset(self, request): - if self.queryset: + if self.queryset is not None: return self.queryset custom_object_type = self.kwargs.get("custom_object_type", None) self.custom_object_type = CustomObjectType.objects.get( @@ -1447,12 +1543,22 @@ def get_queryset(self, request): return model.objects.all() def get_model_form(self, queryset): + # Match core's CSV import (NetBoxModelImportForm._get_custom_fields): a + # non-editable field is omitted from the import form, not disabled. Must + # also go in Meta.exclude, since fields="__all__" auto-generates one otherwise. + fields = list(self.custom_object_type.fields.all()) + non_editable_field_names = tuple( + field.name for field in fields + if field.ui_editable != CustomFieldUIEditableChoices.YES + ) + meta = type( "Meta", (), { "model": queryset.model, "fields": "__all__", + "exclude": non_editable_field_names, }, ) @@ -1461,7 +1567,9 @@ def get_model_form(self, queryset): "__module__": "database.forms", } - for field in self.custom_object_type.fields.all(): + for field in fields: + if field.name in non_editable_field_names: + continue field_type = field_types.FIELD_TYPE_CLASS[field.type]() try: attrs[field.name] = field_type.get_annotated_form_field( diff --git a/pyproject.toml b/pyproject.toml index 33d7b948..f4e45ca9 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -30,7 +30,7 @@ dependencies = [ ] [project.optional-dependencies] -dev = ["check-manifest", "mkdocs", "mkdocs-material", "ruff"] +dev = ["check-manifest", "mkdocs>=1.6,<2", "mkdocs-material>=9.7,<10", "ruff"] test = ["coverage", "pytest", "pytest-cov"] # Install with `pip install "netboxlabs-netbox-custom-objects[branching]"` when # pairing this plugin with netbox-branching. Note: this extra also implies a diff --git a/testing/configuration_branching.py b/testing/configuration_branching.py new file mode 100644 index 00000000..50c1801c --- /dev/null +++ b/testing/configuration_branching.py @@ -0,0 +1,51 @@ +################################################################### +# This file serves as a base configuration for testing purposes # +# only. It is not intended for production use. # +################################################################### + +from netbox_branching.utilities import DynamicSchemaDict + +ALLOWED_HOSTS = ["*"] + +# netbox-branching requires DATABASES (not DATABASE) to be a DynamicSchemaDict. +DATABASES = DynamicSchemaDict({ + 'default': { + 'ENGINE': 'django.db.backends.postgresql', + 'NAME': 'netbox', + 'USER': 'netbox', + 'PASSWORD': 'netbox', + 'HOST': 'localhost', + 'PORT': '', + 'CONN_MAX_AGE': 300, + } +}) + +DATABASE_ROUTERS = ['netbox_branching.database.BranchAwareRouter'] + +PLUGINS = [ + "netbox_custom_objects", + "netbox_branching", +] + +REDIS = { + "tasks": { + "HOST": "localhost", + "PORT": 6379, + "PASSWORD": "", + "DATABASE": 0, + "SSL": False, + }, + "caching": { + "HOST": "localhost", + "PORT": 6379, + "PASSWORD": "", + "DATABASE": 1, + "SSL": False, + }, +} + +SECRET_KEY = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789" + +DEBUG_TOOLBAR_CONFIG = { + "IS_RUNNING_TESTS": False, +}