From 6bed0a95fcd7f9e70e13337b34dd310541e7c229 Mon Sep 17 00:00:00 2001 From: Brian Tiemann Date: Mon, 6 Apr 2026 16:22:23 -0400 Subject: [PATCH 01/19] Exportable schema support --- netbox_custom_objects/api/serializers.py | 7 + ...stomobjecttype_schema_document_and_more.py | 49 +++ netbox_custom_objects/models.py | 58 +++- netbox_custom_objects/schema_format.py | 107 ++++++ .../schemas/cot_schema_v1.json | 309 ++++++++++++++++++ 5 files changed, 529 insertions(+), 1 deletion(-) create mode 100644 netbox_custom_objects/migrations/0005_customobjecttype_schema_document_and_more.py create mode 100644 netbox_custom_objects/schema_format.py create mode 100644 netbox_custom_objects/schemas/cot_schema_v1.json diff --git a/netbox_custom_objects/api/serializers.py b/netbox_custom_objects/api/serializers.py index 414ca140..a3703332 100644 --- a/netbox_custom_objects/api/serializers.py +++ b/netbox_custom_objects/api/serializers.py @@ -50,6 +50,7 @@ class Meta: model = CustomObjectTypeField fields = ( "id", + "url", "name", "label", "custom_object_type", @@ -75,6 +76,10 @@ class Meta: "weight", "is_cloneable", "comments", + "schema_id", + "deprecated", + "deprecated_since", + "scheduled_removal", ) def validate(self, attrs): @@ -152,12 +157,14 @@ class Meta: "verbose_name", "verbose_name_plural", "slug", + "version", "group_name", "description", "tags", "created", "last_updated", "fields", + "schema_document", "table_model_name", "object_type_name", ] diff --git a/netbox_custom_objects/migrations/0005_customobjecttype_schema_document_and_more.py b/netbox_custom_objects/migrations/0005_customobjecttype_schema_document_and_more.py new file mode 100644 index 00000000..b38b91dd --- /dev/null +++ b/netbox_custom_objects/migrations/0005_customobjecttype_schema_document_and_more.py @@ -0,0 +1,49 @@ +# Generated by Django 5.2.12 on 2026-04-06 20:15 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('core', '0021_job_queue_name'), + ('extras', '0134_owner'), + ('netbox_custom_objects', '0004_customobjecttype_group_name'), + ] + + operations = [ + migrations.AddField( + model_name='customobjecttype', + name='schema_document', + field=models.JSONField(blank=True, null=True), + ), + migrations.AddField( + model_name='customobjecttypefield', + name='deprecated', + field=models.BooleanField(default=False), + ), + migrations.AddField( + model_name='customobjecttypefield', + name='deprecated_since', + field=models.CharField(blank=True, max_length=50), + ), + migrations.AddField( + model_name='customobjecttypefield', + name='scheduled_removal', + field=models.CharField(blank=True, max_length=50), + ), + migrations.AddField( + model_name='customobjecttypefield', + name='schema_id', + field=models.PositiveSmallIntegerField(blank=True, null=True), + ), + migrations.AlterField( + model_name='customobjecttype', + name='version', + field=models.CharField(blank=True, max_length=50), + ), + migrations.AddConstraint( + model_name='customobjecttypefield', + constraint=models.UniqueConstraint(condition=models.Q(('schema_id__isnull', False)), fields=('schema_id', 'custom_object_type'), name='netbox_custom_objects_customobjecttypefield_unique_schema_id'), + ), + ] diff --git a/netbox_custom_objects/models.py b/netbox_custom_objects/models.py index 58f85d6d..91c65614 100644 --- a/netbox_custom_objects/models.py +++ b/netbox_custom_objects/models.py @@ -197,7 +197,7 @@ class CustomObjectType(NetBoxModel): verbose_name=_('comments'), blank=True ) - version = models.CharField(max_length=10, blank=True) + version = models.CharField(max_length=50, blank=True) verbose_name = models.CharField(max_length=100, blank=True) verbose_name_plural = models.CharField(max_length=100, blank=True) slug = models.SlugField(max_length=100, unique=True, db_index=True, blank=False) @@ -207,6 +207,14 @@ class CustomObjectType(NetBoxModel): blank=True, help_text=_("Used to group similar custom object types in the navigation menu") ) + schema_document = models.JSONField( + blank=True, + null=True, + help_text=_( + "The last applied or exported schema document for this Custom Object Type. " + "Serves as the source of truth for schema history, including tombstoned fields." + ), + ) cache_timestamp = models.DateTimeField( auto_now=True, help_text=_("Timestamp used for cache invalidation") @@ -924,6 +932,35 @@ class CustomObjectTypeField(CloningMixin, ExportTemplatesMixin, ChangeLoggedMode help_text=_("Replicate this value when cloning objects"), ) comments = models.TextField(verbose_name=_("comments"), blank=True) + schema_id = models.PositiveSmallIntegerField( + blank=True, + null=True, + verbose_name=_("schema ID"), + help_text=_( + "Stable numeric identifier for this field used during schema diffing. " + "Auto-assigned on creation; never changes and never reused within this Custom Object Type." + ), + ) + deprecated = models.BooleanField( + default=False, + verbose_name=_("deprecated"), + help_text=_( + "Mark this field as deprecated. Deprecated fields remain in the database but " + "are read-only in the UI and should not be used in new objects." + ), + ) + deprecated_since = models.CharField( + max_length=50, + blank=True, + verbose_name=_("deprecated since"), + help_text=_("Schema version in which this field was marked deprecated (e.g. '2.0.0')."), + ) + scheduled_removal = models.CharField( + max_length=50, + blank=True, + verbose_name=_("scheduled removal"), + help_text=_("Schema version in which this field is planned to be removed (e.g. '3.0.0')."), + ) clone_fields = ("custom_object_type",) @@ -943,6 +980,11 @@ class Meta: fields=("name", "custom_object_type"), name="%(app_label)s_%(class)s_unique_name", ), + models.UniqueConstraint( + fields=("schema_id", "custom_object_type"), + name="%(app_label)s_%(class)s_unique_schema_id", + condition=models.Q(schema_id__isnull=False), + ), ) def __init__(self, *args, **kwargs): @@ -1529,6 +1571,20 @@ def through_model_name(self): def save(self, *args, **kwargs): is_new = self._state.adding + + # Auto-assign schema_id for new fields that don't have one yet. + # Locks existing rows for this COT to prevent concurrent assignment of the same ID. + if self._state.adding and self.schema_id is None: + with transaction.atomic(): + locked_ids = list( + CustomObjectTypeField.objects + .filter(custom_object_type=self.custom_object_type) + .select_for_update() + .values_list('schema_id', flat=True) + ) + used_ids = [i for i in locked_ids if i is not None] + self.schema_id = max(used_ids, default=0) + 1 + field_type = FIELD_TYPE_CLASS[self.type]() model_field = field_type.get_model_field(self) model = self.custom_object_type.get_model() diff --git a/netbox_custom_objects/schema_format.py b/netbox_custom_objects/schema_format.py new file mode 100644 index 00000000..2f7de0f9 --- /dev/null +++ b/netbox_custom_objects/schema_format.py @@ -0,0 +1,107 @@ +""" +Constants and helpers for the COT portable schema format. + +The schema format is a YAML (or JSON) document describing one or more Custom +Object Type definitions in a portable, versionable way. A multi-type export +always uses a top-level ``types:`` list; the importer also accepts a bare +single-type document for convenience. + +Format version history +---------------------- +"1" Initial version (introduced alongside schema_id / deprecated field support). +""" + +from extras.choices import CustomFieldTypeChoices + +# ── Format version ────────────────────────────────────────────────────────── +# Bump this only when the format itself changes in a breaking way. +SCHEMA_FORMAT_VERSION = "1" + +# ── Field type names (value → schema string) ──────────────────────────────── +# These are the canonical type names used in schema documents. +# They happen to match CustomFieldTypeChoices values, but are redefined here +# explicitly so the schema format is not silently broken by upstream changes. +FIELD_TYPE_TEXT = "text" +FIELD_TYPE_LONGTEXT = "longtext" +FIELD_TYPE_INTEGER = "integer" +FIELD_TYPE_DECIMAL = "decimal" +FIELD_TYPE_BOOLEAN = "boolean" +FIELD_TYPE_DATE = "date" +FIELD_TYPE_DATETIME = "datetime" +FIELD_TYPE_URL = "url" +FIELD_TYPE_JSON = "json" +FIELD_TYPE_SELECT = "select" +FIELD_TYPE_MULTISELECT = "multiselect" +FIELD_TYPE_OBJECT = "object" +FIELD_TYPE_MULTIOBJECT = "multiobject" + +# Mapping from CustomFieldTypeChoices values to schema type names. +# Used by the exporter; the importer uses the inverse. +CHOICES_TO_SCHEMA_TYPE = { + CustomFieldTypeChoices.TYPE_TEXT: FIELD_TYPE_TEXT, + CustomFieldTypeChoices.TYPE_LONGTEXT: FIELD_TYPE_LONGTEXT, + CustomFieldTypeChoices.TYPE_INTEGER: FIELD_TYPE_INTEGER, + CustomFieldTypeChoices.TYPE_DECIMAL: FIELD_TYPE_DECIMAL, + CustomFieldTypeChoices.TYPE_BOOLEAN: FIELD_TYPE_BOOLEAN, + CustomFieldTypeChoices.TYPE_DATE: FIELD_TYPE_DATE, + CustomFieldTypeChoices.TYPE_DATETIME: FIELD_TYPE_DATETIME, + CustomFieldTypeChoices.TYPE_URL: FIELD_TYPE_URL, + CustomFieldTypeChoices.TYPE_JSON: FIELD_TYPE_JSON, + CustomFieldTypeChoices.TYPE_SELECT: FIELD_TYPE_SELECT, + CustomFieldTypeChoices.TYPE_MULTISELECT: FIELD_TYPE_MULTISELECT, + CustomFieldTypeChoices.TYPE_OBJECT: FIELD_TYPE_OBJECT, + CustomFieldTypeChoices.TYPE_MULTIOBJECT: FIELD_TYPE_MULTIOBJECT, +} + +SCHEMA_TYPE_TO_CHOICES = {v: k for k, v in CHOICES_TO_SCHEMA_TYPE.items()} + +# ── related_object_type encoding ───────────────────────────────────────────── +# Built-in NetBox objects: "dcim/device" (app_label/model) +# Custom Object Types: "custom-objects/circuit" (using the COT slug) +CUSTOM_OBJECTS_APP_LABEL_SLUG = "custom-objects" + +# ── Field attribute defaults ───────────────────────────────────────────────── +# Attributes that match these defaults MAY be omitted from the schema document. +# The importer applies them when a key is absent. +FIELD_DEFAULTS = { + "label": "", # resolves to name.replace("_", " ").capitalize() at runtime + "description": "", + "group_name": "", + "primary": False, + "required": False, + "unique": False, + "default": None, + "weight": 100, + "search_weight": 500, + "filter_logic": "loose", + "ui_visible": "always", + "ui_editable": "yes", + "is_cloneable": False, + "deprecated": False, + "deprecated_since": "", + "scheduled_removal": "", + # type-specific defaults + "validation_regex": "", + "validation_minimum": None, + "validation_maximum": None, + "related_object_filter": None, +} + +# ── Field groups by type ───────────────────────────────────────────────────── +# Which type-specific attributes are valid for each field type. +# Used by the exporter to omit irrelevant keys and by the JSON Schema. +FIELD_TYPE_ATTRS = { + FIELD_TYPE_TEXT: {"validation_regex"}, + FIELD_TYPE_LONGTEXT: {"validation_regex"}, + FIELD_TYPE_INTEGER: {"validation_minimum", "validation_maximum"}, + FIELD_TYPE_DECIMAL: {"validation_minimum", "validation_maximum"}, + FIELD_TYPE_BOOLEAN: set(), + FIELD_TYPE_DATE: set(), + FIELD_TYPE_DATETIME: set(), + FIELD_TYPE_URL: set(), + FIELD_TYPE_JSON: set(), + FIELD_TYPE_SELECT: {"choice_set"}, + FIELD_TYPE_MULTISELECT: {"choice_set"}, + FIELD_TYPE_OBJECT: {"related_object_type", "related_object_filter"}, + FIELD_TYPE_MULTIOBJECT: {"related_object_type", "related_object_filter"}, +} \ No newline at end of file diff --git a/netbox_custom_objects/schemas/cot_schema_v1.json b/netbox_custom_objects/schemas/cot_schema_v1.json new file mode 100644 index 00000000..a0e3fbdc --- /dev/null +++ b/netbox_custom_objects/schemas/cot_schema_v1.json @@ -0,0 +1,309 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://netboxlabs.com/schemas/netbox-custom-objects/cot-schema-v1.json", + "title": "NetBox Custom Object Type Schema (v1)", + "description": "Portable schema format for NetBox Custom Object Type definitions.", + + "$defs": { + "semver": { + "type": "string", + "description": "Semantic version string (PEP 440 / semver compatible).", + "examples": ["1.0.0", "2.3.1", "1.0.0-alpha.1"] + }, + + "identifier": { + "type": "string", + "description": "Lowercase alphanumeric characters and single underscores only.", + "pattern": "^[a-z0-9]+(_[a-z0-9]+)*$" + }, + + "slug": { + "type": "string", + "description": "URL-friendly identifier (lowercase alphanumeric and hyphens).", + "pattern": "^[a-z0-9]+(-[a-z0-9]+)*$" + }, + + "related_object_type": { + "type": "string", + "description": "Reference to a NetBox object type. Built-in: 'app_label/model' (e.g. 'dcim/device'). Custom Object Type: 'custom-objects/'.", + "pattern": "^[a-z0-9_-]+/[a-z0-9_-]+$", + "examples": ["dcim/device", "ipam/prefix", "custom-objects/circuit"] + }, + + "field_type": { + "type": "string", + "enum": [ + "text", + "longtext", + "integer", + "decimal", + "boolean", + "date", + "datetime", + "url", + "json", + "select", + "multiselect", + "object", + "multiobject" + ] + }, + + "field_base": { + "type": "object", + "required": ["id", "name", "type"], + "properties": { + "id": { + "type": "integer", + "minimum": 1, + "description": "Stable numeric identifier. Never changes and never reused within this COT." + }, + "name": { "$ref": "#/$defs/identifier" }, + "type": { "$ref": "#/$defs/field_type" }, + "label": { + "type": "string", + "maxLength": 50, + "description": "Display label. Defaults to name with underscores replaced by spaces." + }, + "description": { "type": "string", "maxLength": 200 }, + "group_name": { "type": "string", "maxLength": 50 }, + "primary": { "type": "boolean", "default": false }, + "required": { "type": "boolean", "default": false }, + "unique": { "type": "boolean", "default": false }, + "default": { "description": "JSON-serialisable default value." }, + "weight": { "type": "integer", "minimum": 0, "default": 100 }, + "search_weight": { "type": "integer", "minimum": 0, "default": 500 }, + "filter_logic": { + "type": "string", + "enum": ["disabled", "loose", "exact"], + "default": "loose" + }, + "ui_visible": { + "type": "string", + "enum": ["always", "if-set", "hidden"], + "default": "always" + }, + "ui_editable": { + "type": "string", + "enum": ["yes", "no", "hidden"], + "default": "yes" + }, + "is_cloneable": { "type": "boolean", "default": false }, + "deprecated": { "type": "boolean", "default": false }, + "deprecated_since": { + "type": "string", + "description": "Schema version in which this field was marked deprecated." + }, + "scheduled_removal": { + "type": "string", + "description": "Schema version in which this field is planned to be removed." + } + } + }, + + "text_field": { + "allOf": [ + { "$ref": "#/$defs/field_base" }, + { + "properties": { + "type": { "const": "text" }, + "validation_regex": { "type": "string", "maxLength": 500 } + } + } + ] + }, + + "longtext_field": { + "allOf": [ + { "$ref": "#/$defs/field_base" }, + { + "properties": { + "type": { "const": "longtext" }, + "validation_regex": { "type": "string", "maxLength": 500 } + } + } + ] + }, + + "integer_field": { + "allOf": [ + { "$ref": "#/$defs/field_base" }, + { + "properties": { + "type": { "const": "integer" }, + "validation_minimum": { "type": "integer" }, + "validation_maximum": { "type": "integer" } + } + } + ] + }, + + "decimal_field": { + "allOf": [ + { "$ref": "#/$defs/field_base" }, + { + "properties": { + "type": { "const": "decimal" }, + "validation_minimum": { "type": "number" }, + "validation_maximum": { "type": "number" } + } + } + ] + }, + + "select_field": { + "allOf": [ + { "$ref": "#/$defs/field_base" }, + { + "required": ["choice_set"], + "properties": { + "type": { "const": "select" }, + "choice_set": { + "type": "string", + "description": "Name of the CustomFieldChoiceSet to use." + } + } + } + ] + }, + + "multiselect_field": { + "allOf": [ + { "$ref": "#/$defs/field_base" }, + { + "required": ["choice_set"], + "properties": { + "type": { "const": "multiselect" }, + "choice_set": { + "type": "string", + "description": "Name of the CustomFieldChoiceSet to use." + } + } + } + ] + }, + + "object_field": { + "allOf": [ + { "$ref": "#/$defs/field_base" }, + { + "required": ["related_object_type"], + "properties": { + "type": { "const": "object" }, + "related_object_type": { "$ref": "#/$defs/related_object_type" }, + "related_object_filter": { + "type": ["object", "null"], + "description": "Query params dict to filter the related object selection." + } + } + } + ] + }, + + "multiobject_field": { + "allOf": [ + { "$ref": "#/$defs/field_base" }, + { + "required": ["related_object_type"], + "properties": { + "type": { "const": "multiobject" }, + "related_object_type": { "$ref": "#/$defs/related_object_type" }, + "related_object_filter": { + "type": ["object", "null"], + "description": "Query params dict to filter the related object selection." + } + } + } + ] + }, + + "simple_field": { + "allOf": [ + { "$ref": "#/$defs/field_base" }, + { + "properties": { + "type": { + "enum": ["boolean", "date", "datetime", "url", "json"] + } + } + } + ] + }, + + "field": { + "oneOf": [ + { "$ref": "#/$defs/text_field" }, + { "$ref": "#/$defs/longtext_field" }, + { "$ref": "#/$defs/integer_field" }, + { "$ref": "#/$defs/decimal_field" }, + { "$ref": "#/$defs/select_field" }, + { "$ref": "#/$defs/multiselect_field" }, + { "$ref": "#/$defs/object_field" }, + { "$ref": "#/$defs/multiobject_field" }, + { "$ref": "#/$defs/simple_field" } + ], + "discriminator": { "propertyName": "type" } + }, + + "removed_field": { + "type": "object", + "required": ["id", "name", "type"], + "additionalProperties": false, + "properties": { + "id": { + "type": "integer", + "minimum": 1, + "description": "The stable ID that was permanently retired when this field was removed." + }, + "name": { "$ref": "#/$defs/identifier" }, + "type": { "$ref": "#/$defs/field_type" }, + "removed_in": { + "type": "string", + "description": "Schema version in which this field was removed." + } + } + }, + + "cot_definition": { + "type": "object", + "required": ["name", "slug"], + "additionalProperties": false, + "properties": { + "name": { "$ref": "#/$defs/identifier" }, + "slug": { "$ref": "#/$defs/slug" }, + "version": { "$ref": "#/$defs/semver" }, + "verbose_name": { "type": "string", "maxLength": 100 }, + "verbose_name_plural": { "type": "string", "maxLength": 100 }, + "description": { "type": "string", "maxLength": 200 }, + "group_name": { "type": "string", "maxLength": 100 }, + "fields": { + "type": "array", + "items": { "$ref": "#/$defs/field" }, + "description": "Active and deprecated fields. Each id must be unique within this COT." + }, + "removed_fields": { + "type": "array", + "items": { "$ref": "#/$defs/removed_field" }, + "description": "Tombstone records for permanently removed fields. IDs here must not appear in 'fields'." + } + } + } + }, + + "type": "object", + "required": ["schema_version", "types"], + "additionalProperties": false, + "properties": { + "schema_version": { + "type": "string", + "const": "1", + "description": "Format spec version. Identifies this document as COT Schema Format v1." + }, + "types": { + "type": "array", + "minItems": 1, + "items": { "$ref": "#/$defs/cot_definition" }, + "description": "One or more Custom Object Type definitions." + } + } +} \ No newline at end of file From 9c536ea181098e4e7e694efd0b247bef96383a5a Mon Sep 17 00:00:00 2001 From: Brian Tiemann Date: Mon, 6 Apr 2026 16:44:01 -0400 Subject: [PATCH 02/19] Add comments and validation improvements --- netbox_custom_objects/api/serializers.py | 1 + netbox_custom_objects/schema_format.py | 7 +++++-- netbox_custom_objects/schemas/cot_schema_v1.json | 12 ++++++++---- 3 files changed, 14 insertions(+), 6 deletions(-) diff --git a/netbox_custom_objects/api/serializers.py b/netbox_custom_objects/api/serializers.py index a3703332..1694e910 100644 --- a/netbox_custom_objects/api/serializers.py +++ b/netbox_custom_objects/api/serializers.py @@ -168,6 +168,7 @@ class Meta: "table_model_name", "object_type_name", ] + read_only_fields = ("schema_document",) brief_fields = ("id", "url", "name", "slug", "description") def get_table_model_name(self, obj): diff --git a/netbox_custom_objects/schema_format.py b/netbox_custom_objects/schema_format.py index 2f7de0f9..c3e38e4e 100644 --- a/netbox_custom_objects/schema_format.py +++ b/netbox_custom_objects/schema_format.py @@ -64,7 +64,9 @@ # Attributes that match these defaults MAY be omitted from the schema document. # The importer applies them when a key is absent. FIELD_DEFAULTS = { - "label": "", # resolves to name.replace("_", " ").capitalize() at runtime + # label resolves to name.replace("_", " ").capitalize() at runtime. Importer must implement this same logic. + # An empty or absent label means "derive from name". + "label": "", "description": "", "group_name": "", "primary": False, @@ -90,6 +92,7 @@ # ── Field groups by type ───────────────────────────────────────────────────── # Which type-specific attributes are valid for each field type. # Used by the exporter to omit irrelevant keys and by the JSON Schema. +# Note: The exporter should use FIELD_TYPE_ATTRS to drop irrelevant keys inherited from field_base. FIELD_TYPE_ATTRS = { FIELD_TYPE_TEXT: {"validation_regex"}, FIELD_TYPE_LONGTEXT: {"validation_regex"}, @@ -104,4 +107,4 @@ FIELD_TYPE_MULTISELECT: {"choice_set"}, FIELD_TYPE_OBJECT: {"related_object_type", "related_object_filter"}, FIELD_TYPE_MULTIOBJECT: {"related_object_type", "related_object_filter"}, -} \ No newline at end of file +} diff --git a/netbox_custom_objects/schemas/cot_schema_v1.json b/netbox_custom_objects/schemas/cot_schema_v1.json index a0e3fbdc..20448454 100644 --- a/netbox_custom_objects/schemas/cot_schema_v1.json +++ b/netbox_custom_objects/schemas/cot_schema_v1.json @@ -13,7 +13,7 @@ "identifier": { "type": "string", - "description": "Lowercase alphanumeric characters and single underscores only.", + "description": "Lowercase alphanumeric characters and single underscores only. Note: this pattern is intentionally more restrictive than the model validator (^[a-z0-9_]+$) — it disallows leading/trailing underscores and double underscores. A valid DB field name that violates this pattern will fail schema export.", "pattern": "^[a-z0-9]+(_[a-z0-9]+)*$" }, @@ -242,7 +242,10 @@ { "$ref": "#/$defs/multiobject_field" }, { "$ref": "#/$defs/simple_field" } ], - "discriminator": { "propertyName": "type" } + "discriminator": { + "propertyName": "type", + "$comment": "discriminator is an OpenAPI 3.x extension, not part of JSON Schema 2020-12. Standard validators (including the jsonschema library) will silently ignore it. The oneOf above is what actually enforces field type constraints; the discriminator is present only as documentation for tooling that understands it." + } }, "removed_field": { @@ -285,7 +288,8 @@ "type": "array", "items": { "$ref": "#/$defs/removed_field" }, "description": "Tombstone records for permanently removed fields. IDs here must not appear in 'fields'." - } + }, + "$comment": "The 'comments' field present on both CustomObjectType and CustomObjectTypeField is intentionally excluded from this schema format. It is editorial annotation, not structural schema, and would create noise in diffs and cross-installation sharing." } } }, @@ -306,4 +310,4 @@ "description": "One or more Custom Object Type definitions." } } -} \ No newline at end of file +} From 56ddc08431991cfcd0dc49e4217ee792dce5a601 Mon Sep 17 00:00:00 2001 From: Brian Tiemann Date: Mon, 6 Apr 2026 16:45:46 -0400 Subject: [PATCH 03/19] Fix ruff errors --- ...stomobjecttype_schema_document_and_more.py | 6 +- netbox_custom_objects/models.py | 3 + .../tests/test_schema_format.py | 475 ++++++++++++++++++ 3 files changed, 483 insertions(+), 1 deletion(-) create mode 100644 netbox_custom_objects/tests/test_schema_format.py diff --git a/netbox_custom_objects/migrations/0005_customobjecttype_schema_document_and_more.py b/netbox_custom_objects/migrations/0005_customobjecttype_schema_document_and_more.py index b38b91dd..a0a6b606 100644 --- a/netbox_custom_objects/migrations/0005_customobjecttype_schema_document_and_more.py +++ b/netbox_custom_objects/migrations/0005_customobjecttype_schema_document_and_more.py @@ -44,6 +44,10 @@ class Migration(migrations.Migration): ), migrations.AddConstraint( model_name='customobjecttypefield', - constraint=models.UniqueConstraint(condition=models.Q(('schema_id__isnull', False)), fields=('schema_id', 'custom_object_type'), name='netbox_custom_objects_customobjecttypefield_unique_schema_id'), + constraint=models.UniqueConstraint( + condition=models.Q(('schema_id__isnull', False)), + fields=('schema_id', 'custom_object_type'), + name='netbox_custom_objects_customobjecttypefield_unique_schema_id', + ), ), ] diff --git a/netbox_custom_objects/models.py b/netbox_custom_objects/models.py index 91c65614..5ea02d51 100644 --- a/netbox_custom_objects/models.py +++ b/netbox_custom_objects/models.py @@ -1574,6 +1574,9 @@ def save(self, *args, **kwargs): # Auto-assign schema_id for new fields that don't have one yet. # Locks existing rows for this COT to prevent concurrent assignment of the same ID. + # Note: bulk_create() bypasses save() entirely, so auto-assignment will NOT fire for + # fields created via CustomObjectTypeField.objects.bulk_create(...). Always set + # schema_id explicitly when using bulk_create. if self._state.adding and self.schema_id is None: with transaction.atomic(): locked_ids = list( diff --git a/netbox_custom_objects/tests/test_schema_format.py b/netbox_custom_objects/tests/test_schema_format.py new file mode 100644 index 00000000..dabaf936 --- /dev/null +++ b/netbox_custom_objects/tests/test_schema_format.py @@ -0,0 +1,475 @@ +""" +Tests for the COT portable schema format (issue #386). + +Covers: +- schema_id auto-assignment and uniqueness on CustomObjectTypeField +- deprecated / deprecated_since / scheduled_removal field behaviour +- schema_document and version fields on CustomObjectType +- JSON Schema document validation via cot_schema_v1.json +""" +import json +import unittest +from pathlib import Path + +from django.db import IntegrityError +from django.test import TestCase, TransactionTestCase + +from netbox_custom_objects.schema_format import ( + CHOICES_TO_SCHEMA_TYPE, + SCHEMA_FORMAT_VERSION, + SCHEMA_TYPE_TO_CHOICES, +) + +from .base import CustomObjectsTestCase, TransactionCleanupMixin + +# --------------------------------------------------------------------------- +# Optional jsonschema dependency — skip structural-validation tests if absent +# --------------------------------------------------------------------------- +try: + import jsonschema + HAS_JSONSCHEMA = True +except ImportError: + HAS_JSONSCHEMA = False + +_SCHEMA_PATH = ( + Path(__file__).resolve().parent.parent / "schemas" / "cot_schema_v1.json" +) + + +# =========================================================================== +# schema_id / deprecated model field tests (require TransactionTestCase) +# =========================================================================== + +class SchemaIdAutoAssignmentTestCase( + TransactionCleanupMixin, CustomObjectsTestCase, TransactionTestCase +): + """schema_id is auto-assigned on field creation and never reused.""" + + def test_schema_id_assigned_on_create(self): + """A new field with no explicit schema_id gets one automatically.""" + cot = self.create_custom_object_type(name='schidtest', slug='sch-id-test') + field = self.create_custom_object_type_field(cot, name='label', type='text') + + self.assertIsNotNone(field.schema_id, "schema_id must be set after save.") + self.assertGreaterEqual(field.schema_id, 1) + + def test_schema_ids_are_sequential(self): + """Each new field in the same COT gets the next available integer.""" + cot = self.create_custom_object_type(name='schseq', slug='sch-seq') + f1 = self.create_custom_object_type_field(cot, name='first', type='text') + f2 = self.create_custom_object_type_field(cot, name='second', type='text') + f3 = self.create_custom_object_type_field(cot, name='third', type='text') + + ids = sorted([f1.schema_id, f2.schema_id, f3.schema_id]) + self.assertEqual(ids, list(range(ids[0], ids[0] + 3)), + "schema_ids must form a contiguous sequence within the COT.") + + def test_schema_id_scoped_to_cot(self): + """Two different COTs can have fields with the same schema_id.""" + cot_a = self.create_custom_object_type(name='schcota', slug='sch-cot-a') + cot_b = self.create_custom_object_type(name='schcotb', slug='sch-cot-b') + + fa = self.create_custom_object_type_field(cot_a, name='label', type='text') + fb = self.create_custom_object_type_field(cot_b, name='label', type='text') + + # Both are assigned ID 1 — that is fine because uniqueness is per-COT. + self.assertEqual(fa.schema_id, fb.schema_id, + "First field in each COT should both receive schema_id=1.") + + def test_explicit_schema_id_is_respected(self): + """A field created with an explicit schema_id keeps that value.""" + cot = self.create_custom_object_type(name='schexpl', slug='sch-expl') + field = self.create_custom_object_type_field( + cot, name='label', type='text', schema_id=42 + ) + self.assertEqual(field.schema_id, 42) + + def test_duplicate_schema_id_within_cot_raises(self): + """Assigning the same schema_id to two fields in the same COT is rejected.""" + cot = self.create_custom_object_type(name='schdup', slug='sch-dup') + self.create_custom_object_type_field( + cot, name='first', type='text', schema_id=7 + ) + with self.assertRaises(IntegrityError): + self.create_custom_object_type_field( + cot, name='second', type='text', schema_id=7 + ) + + def test_schema_id_gap_after_deletion(self): + """ + After a field is deleted its schema_id is not reused; the next + auto-assigned ID continues from the highest ever used. + """ + cot = self.create_custom_object_type(name='schgap', slug='sch-gap') + self.create_custom_object_type_field(cot, name='first', type='text') + f2 = self.create_custom_object_type_field(cot, name='second', type='text') + id_before_delete = f2.schema_id + + f2.delete() + + f3 = self.create_custom_object_type_field(cot, name='third', type='text') + self.assertGreater( + f3.schema_id, id_before_delete, + "Auto-assigned schema_id must not reuse a previously deleted field's ID.", + ) + + +class DeprecationFieldsTestCase( + TransactionCleanupMixin, CustomObjectsTestCase, TransactionTestCase +): + """deprecated, deprecated_since, and scheduled_removal field behaviour.""" + + def test_deprecated_defaults_to_false(self): + cot = self.create_custom_object_type(name='depdefault', slug='dep-default') + field = self.create_custom_object_type_field(cot, name='label', type='text') + self.assertFalse(field.deprecated) + + def test_can_mark_field_deprecated(self): + cot = self.create_custom_object_type(name='depmark', slug='dep-mark') + field = self.create_custom_object_type_field(cot, name='label', type='text') + + field.deprecated = True + field.deprecated_since = '2.0.0' + field.scheduled_removal = '3.0.0' + field.save() + + field.refresh_from_db() + self.assertTrue(field.deprecated) + self.assertEqual(field.deprecated_since, '2.0.0') + self.assertEqual(field.scheduled_removal, '3.0.0') + + def test_deprecation_version_strings_accept_semver(self): + """Long semver strings (pre-release, build metadata) fit in max_length=50.""" + cot = self.create_custom_object_type(name='depsemver', slug='dep-semver') + field = self.create_custom_object_type_field(cot, name='label', type='text') + + field.deprecated = True + field.deprecated_since = '1.0.0-alpha.1+build.42' + field.scheduled_removal = '2.0.0-beta.3' + field.save() + + field.refresh_from_db() + self.assertEqual(field.deprecated_since, '1.0.0-alpha.1+build.42') + + +# =========================================================================== +# schema_document and version field tests (plain TestCase — no DDL needed) +# =========================================================================== + +class SchemaDocumentFieldTestCase(CustomObjectsTestCase, TestCase): + """schema_document and version fields on CustomObjectType.""" + + def test_schema_document_defaults_to_null(self): + cot = self.create_custom_object_type(name='schemadoc', slug='schema-doc') + self.assertIsNone(cot.schema_document) + + def test_schema_document_can_store_json(self): + cot = self.create_custom_object_type(name='schemadoc2', slug='schema-doc-2') + document = { + "schema_version": "1", + "name": "schemadoc2", + "slug": "schema-doc-2", + "version": "1.0.0", + "fields": [], + "removed_fields": [], + } + cot.schema_document = document + cot.save() + + cot.refresh_from_db() + self.assertEqual(cot.schema_document, document) + + def test_version_field_accepts_long_semver(self): + """version field max_length=50 accommodates pre-release semver strings.""" + cot = self.create_custom_object_type( + name='semvertest', slug='semver-test', version='10.200.300-alpha.1' + ) + cot.refresh_from_db() + self.assertEqual(cot.version, '10.200.300-alpha.1') + + def test_version_field_optional(self): + """version is not required; blank is accepted.""" + cot = self.create_custom_object_type(name='nover', slug='no-ver') + self.assertEqual(cot.version, '') + + +# =========================================================================== +# schema_format.py constants tests +# =========================================================================== + +class SchemaFormatConstantsTestCase(TestCase): + """Sanity checks on schema_format module constants.""" + + def test_format_version_is_string(self): + self.assertIsInstance(SCHEMA_FORMAT_VERSION, str) + self.assertEqual(SCHEMA_FORMAT_VERSION, "1") + + def test_type_mapping_is_bijective(self): + """CHOICES_TO_SCHEMA_TYPE and SCHEMA_TYPE_TO_CHOICES must be inverses.""" + for choices_val, schema_name in CHOICES_TO_SCHEMA_TYPE.items(): + self.assertEqual( + SCHEMA_TYPE_TO_CHOICES[schema_name], choices_val, + f"Round-trip failed for {choices_val!r} ↔ {schema_name!r}", + ) + + def test_all_field_types_are_mapped(self): + """Every CustomFieldTypeChoices value must have a schema type entry.""" + from extras.choices import CustomFieldTypeChoices + for attr in dir(CustomFieldTypeChoices): + if attr.startswith('TYPE_'): + value = getattr(CustomFieldTypeChoices, attr) + self.assertIn( + value, CHOICES_TO_SCHEMA_TYPE, + f"CustomFieldTypeChoices.{attr} ({value!r}) is missing from CHOICES_TO_SCHEMA_TYPE", + ) + + +# =========================================================================== +# JSON Schema structural validation tests +# =========================================================================== + +@unittest.skipUnless(HAS_JSONSCHEMA, "jsonschema library not installed") +class COTJsonSchemaTestCase(TestCase): + """Validate that cot_schema_v1.json correctly accepts/rejects documents.""" + + @classmethod + def setUpClass(cls): + super().setUpClass() + with open(_SCHEMA_PATH) as fh: + cls.schema = json.load(fh) + cls.validator_cls = jsonschema.Draft202012Validator + cls.validator_cls.check_schema(cls.schema) + + def _validate(self, document): + """Return a list of validation errors (empty means valid).""" + v = self.validator_cls(self.schema) + return list(v.iter_errors(document)) + + def _assert_valid(self, document): + errors = self._validate(document) + self.assertEqual(errors, [], f"Expected valid document but got errors: {errors}") + + def _assert_invalid(self, document): + errors = self._validate(document) + self.assertGreater(len(errors), 0, "Expected invalid document but it passed validation.") + + # ------------------------------------------------------------------ + # Valid documents + # ------------------------------------------------------------------ + + def test_minimal_valid_document(self): + """A document with only the required fields is valid.""" + self._assert_valid({ + "schema_version": "1", + "types": [ + {"name": "widget", "slug": "widget"} + ], + }) + + def test_full_valid_document(self): + """A document exercising all field types passes validation.""" + self._assert_valid({ + "schema_version": "1", + "types": [ + { + "name": "circuit_endpoint", + "slug": "circuit-endpoint", + "version": "1.2.0", + "verbose_name": "Circuit Endpoint", + "verbose_name_plural": "Circuit Endpoints", + "description": "One end of a circuit", + "group_name": "Circuits", + "fields": [ + {"id": 1, "name": "label", "type": "text", "primary": True, "required": True}, + {"id": 2, "name": "notes", "type": "longtext"}, + {"id": 3, "name": "speed", "type": "integer", "validation_minimum": 0}, + {"id": 4, "name": "ratio", "type": "decimal"}, + {"id": 5, "name": "active", "type": "boolean"}, + {"id": 6, "name": "install_date", "type": "date"}, + {"id": 7, "name": "last_seen", "type": "datetime"}, + {"id": 8, "name": "docs_url", "type": "url"}, + {"id": 9, "name": "metadata", "type": "json"}, + {"id": 10, "name": "status", "type": "select", "choice_set": "endpoint_statuses"}, + {"id": 11, "name": "tags", "type": "multiselect", "choice_set": "endpoint_tags"}, + {"id": 12, "name": "device", "type": "object", "related_object_type": "dcim/device"}, + {"id": 13, "name": "sites", "type": "multiobject", "related_object_type": "dcim/site"}, + ], + "removed_fields": [ + {"id": 14, "name": "old_code", "type": "text", "removed_in": "1.1.0"} + ], + } + ], + }) + + def test_cot_reference_to_another_cot(self): + """related_object_type using 'custom-objects/' format is valid.""" + self._assert_valid({ + "schema_version": "1", + "types": [ + { + "name": "endpoint", + "slug": "endpoint", + "fields": [ + { + "id": 1, + "name": "circuit", + "type": "object", + "related_object_type": "custom-objects/circuit", + } + ], + } + ], + }) + + def test_deprecated_field_is_valid(self): + self._assert_valid({ + "schema_version": "1", + "types": [ + { + "name": "widget", + "slug": "widget", + "fields": [ + { + "id": 1, + "name": "legacy_code", + "type": "text", + "deprecated": True, + "deprecated_since": "2.0.0", + "scheduled_removal": "3.0.0", + } + ], + } + ], + }) + + def test_multi_type_export(self): + """Multiple COT definitions in a single document are valid.""" + self._assert_valid({ + "schema_version": "1", + "types": [ + {"name": "circuit", "slug": "circuit"}, + {"name": "endpoint", "slug": "endpoint"}, + ], + }) + + # ------------------------------------------------------------------ + # Invalid documents + # ------------------------------------------------------------------ + + def test_missing_schema_version_is_invalid(self): + self._assert_invalid({ + "types": [{"name": "widget", "slug": "widget"}], + }) + + def test_wrong_schema_version_is_invalid(self): + self._assert_invalid({ + "schema_version": "99", + "types": [{"name": "widget", "slug": "widget"}], + }) + + def test_missing_types_is_invalid(self): + self._assert_invalid({"schema_version": "1"}) + + def test_empty_types_list_is_invalid(self): + self._assert_invalid({"schema_version": "1", "types": []}) + + def test_field_missing_id_is_invalid(self): + self._assert_invalid({ + "schema_version": "1", + "types": [ + { + "name": "widget", + "slug": "widget", + "fields": [{"name": "label", "type": "text"}], + } + ], + }) + + def test_field_missing_name_is_invalid(self): + self._assert_invalid({ + "schema_version": "1", + "types": [ + { + "name": "widget", + "slug": "widget", + "fields": [{"id": 1, "type": "text"}], + } + ], + }) + + def test_field_missing_type_is_invalid(self): + self._assert_invalid({ + "schema_version": "1", + "types": [ + { + "name": "widget", + "slug": "widget", + "fields": [{"id": 1, "name": "label"}], + } + ], + }) + + def test_select_field_missing_choice_set_is_invalid(self): + self._assert_invalid({ + "schema_version": "1", + "types": [ + { + "name": "widget", + "slug": "widget", + "fields": [{"id": 1, "name": "status", "type": "select"}], + } + ], + }) + + def test_object_field_missing_related_object_type_is_invalid(self): + self._assert_invalid({ + "schema_version": "1", + "types": [ + { + "name": "widget", + "slug": "widget", + "fields": [{"id": 1, "name": "device", "type": "object"}], + } + ], + }) + + def test_unknown_field_type_is_invalid(self): + self._assert_invalid({ + "schema_version": "1", + "types": [ + { + "name": "widget", + "slug": "widget", + "fields": [{"id": 1, "name": "thing", "type": "frobnitz"}], + } + ], + }) + + def test_invalid_filter_logic_value_is_invalid(self): + self._assert_invalid({ + "schema_version": "1", + "types": [ + { + "name": "widget", + "slug": "widget", + "fields": [ + {"id": 1, "name": "label", "type": "text", "filter_logic": "fuzzy"} + ], + } + ], + }) + + def test_removed_field_with_extra_properties_is_invalid(self): + """removed_field uses additionalProperties: false, so unknown keys are rejected.""" + self._assert_invalid({ + "schema_version": "1", + "types": [ + { + "name": "widget", + "slug": "widget", + "removed_fields": [ + {"id": 5, "name": "old", "type": "text", "unexpected_key": True} + ], + } + ], + }) From 2ed64d868583632fd735080f65d574a54052c7fd Mon Sep 17 00:00:00 2001 From: Brian Tiemann Date: Mon, 6 Apr 2026 21:41:45 -0400 Subject: [PATCH 04/19] Fix ruff errors --- ...stomobjecttype_next_schema_id_and_more.py} | 7 +++++- netbox_custom_objects/models.py | 25 +++++++++++++------ .../schemas/cot_schema_v1.json | 4 +-- .../tests/test_schema_format.py | 23 ++++++++++++----- 4 files changed, 42 insertions(+), 17 deletions(-) rename netbox_custom_objects/migrations/{0005_customobjecttype_schema_document_and_more.py => 0005_customobjecttype_next_schema_id_and_more.py} (87%) diff --git a/netbox_custom_objects/migrations/0005_customobjecttype_schema_document_and_more.py b/netbox_custom_objects/migrations/0005_customobjecttype_next_schema_id_and_more.py similarity index 87% rename from netbox_custom_objects/migrations/0005_customobjecttype_schema_document_and_more.py rename to netbox_custom_objects/migrations/0005_customobjecttype_next_schema_id_and_more.py index a0a6b606..0dedcd93 100644 --- a/netbox_custom_objects/migrations/0005_customobjecttype_schema_document_and_more.py +++ b/netbox_custom_objects/migrations/0005_customobjecttype_next_schema_id_and_more.py @@ -1,4 +1,4 @@ -# Generated by Django 5.2.12 on 2026-04-06 20:15 +# Generated by Django 5.2.12 on 2026-04-07 01:35 from django.db import migrations, models @@ -12,6 +12,11 @@ class Migration(migrations.Migration): ] operations = [ + migrations.AddField( + model_name='customobjecttype', + name='next_schema_id', + field=models.PositiveSmallIntegerField(default=0, editable=False), + ), migrations.AddField( model_name='customobjecttype', name='schema_document', diff --git a/netbox_custom_objects/models.py b/netbox_custom_objects/models.py index 5ea02d51..2b1456d6 100644 --- a/netbox_custom_objects/models.py +++ b/netbox_custom_objects/models.py @@ -215,6 +215,14 @@ class CustomObjectType(NetBoxModel): "Serves as the source of truth for schema history, including tombstoned fields." ), ) + next_schema_id = models.PositiveSmallIntegerField( + default=0, + editable=False, + help_text=_( + "Monotonically increasing counter tracking the highest schema_id ever assigned " + "to a field on this Custom Object Type. Never decreases, even after field deletion." + ), + ) cache_timestamp = models.DateTimeField( auto_now=True, help_text=_("Timestamp used for cache invalidation") @@ -1573,20 +1581,21 @@ def save(self, *args, **kwargs): is_new = self._state.adding # Auto-assign schema_id for new fields that don't have one yet. - # Locks existing rows for this COT to prevent concurrent assignment of the same ID. + # Increments the monotonic counter on the parent CustomObjectType so that IDs are + # never reused, even after a field is deleted. The UniqueConstraint on + # (schema_id, custom_object_type) is the safety net against races; a concurrent + # writer would get an IntegrityError and must retry. # Note: bulk_create() bypasses save() entirely, so auto-assignment will NOT fire for # fields created via CustomObjectTypeField.objects.bulk_create(...). Always set # schema_id explicitly when using bulk_create. if self._state.adding and self.schema_id is None: with transaction.atomic(): - locked_ids = list( - CustomObjectTypeField.objects - .filter(custom_object_type=self.custom_object_type) - .select_for_update() - .values_list('schema_id', flat=True) + cot = CustomObjectType.objects.select_for_update().get( + pk=self.custom_object_type_id ) - used_ids = [i for i in locked_ids if i is not None] - self.schema_id = max(used_ids, default=0) + 1 + cot.next_schema_id = cot.next_schema_id + 1 + cot.save(update_fields=['next_schema_id']) + self.schema_id = cot.next_schema_id field_type = FIELD_TYPE_CLASS[self.type]() model_field = field_type.get_model_field(self) diff --git a/netbox_custom_objects/schemas/cot_schema_v1.json b/netbox_custom_objects/schemas/cot_schema_v1.json index 20448454..4de0abe1 100644 --- a/netbox_custom_objects/schemas/cot_schema_v1.json +++ b/netbox_custom_objects/schemas/cot_schema_v1.json @@ -268,6 +268,7 @@ }, "cot_definition": { + "$comment": "The 'comments' field present on both CustomObjectType and CustomObjectTypeField is intentionally excluded from this schema format. It is editorial annotation, not structural schema, and would create noise in diffs and cross-installation sharing.", "type": "object", "required": ["name", "slug"], "additionalProperties": false, @@ -288,8 +289,7 @@ "type": "array", "items": { "$ref": "#/$defs/removed_field" }, "description": "Tombstone records for permanently removed fields. IDs here must not appear in 'fields'." - }, - "$comment": "The 'comments' field present on both CustomObjectType and CustomObjectTypeField is intentionally excluded from this schema format. It is editorial annotation, not structural schema, and would create noise in diffs and cross-installation sharing." + } } } }, diff --git a/netbox_custom_objects/tests/test_schema_format.py b/netbox_custom_objects/tests/test_schema_format.py index dabaf936..9fa215db 100644 --- a/netbox_custom_objects/tests/test_schema_format.py +++ b/netbox_custom_objects/tests/test_schema_format.py @@ -11,7 +11,7 @@ import unittest from pathlib import Path -from django.db import IntegrityError +from django.db import IntegrityError, transaction from django.test import TestCase, TransactionTestCase from netbox_custom_objects.schema_format import ( @@ -90,10 +90,15 @@ def test_duplicate_schema_id_within_cot_raises(self): self.create_custom_object_type_field( cot, name='first', type='text', schema_id=7 ) + # Wrap in transaction.atomic() so that when IntegrityError is raised and caught + # by assertRaises, the savepoint is rolled back cleanly. Without this, PostgreSQL + # leaves the connection in an aborted-transaction state and all subsequent SQL + # calls (including tearDown) fail with "connection is closed". with self.assertRaises(IntegrityError): - self.create_custom_object_type_field( - cot, name='second', type='text', schema_id=7 - ) + with transaction.atomic(): + self.create_custom_object_type_field( + cot, name='second', type='text', schema_id=7 + ) def test_schema_id_gap_after_deletion(self): """ @@ -126,8 +131,12 @@ def test_deprecated_defaults_to_false(self): def test_can_mark_field_deprecated(self): cot = self.create_custom_object_type(name='depmark', slug='dep-mark') - field = self.create_custom_object_type_field(cot, name='label', type='text') + self.create_custom_object_type_field(cot, name='label', type='text') + # Re-fetch from DB so that from_db() is called and self.original is set, + # which the save() method requires when updating an existing field. + from netbox_custom_objects.models import CustomObjectTypeField + field = CustomObjectTypeField.objects.get(custom_object_type=cot, name='label') field.deprecated = True field.deprecated_since = '2.0.0' field.scheduled_removal = '3.0.0' @@ -141,8 +150,10 @@ def test_can_mark_field_deprecated(self): def test_deprecation_version_strings_accept_semver(self): """Long semver strings (pre-release, build metadata) fit in max_length=50.""" cot = self.create_custom_object_type(name='depsemver', slug='dep-semver') - field = self.create_custom_object_type_field(cot, name='label', type='text') + self.create_custom_object_type_field(cot, name='label', type='text') + from netbox_custom_objects.models import CustomObjectTypeField + field = CustomObjectTypeField.objects.get(custom_object_type=cot, name='label') field.deprecated = True field.deprecated_since = '1.0.0-alpha.1+build.42' field.scheduled_removal = '2.0.0-beta.3' From a0f6ce64c0889e332d2b8022c6f94706584472c0 Mon Sep 17 00:00:00 2001 From: Brian Tiemann Date: Tue, 7 Apr 2026 05:14:00 -0400 Subject: [PATCH 05/19] Fix monotonic schema id increment --- netbox_custom_objects/models.py | 12 +++++++++--- 1 file changed, 9 insertions(+), 3 deletions(-) diff --git a/netbox_custom_objects/models.py b/netbox_custom_objects/models.py index 2b1456d6..2b40c118 100644 --- a/netbox_custom_objects/models.py +++ b/netbox_custom_objects/models.py @@ -1593,9 +1593,15 @@ def save(self, *args, **kwargs): cot = CustomObjectType.objects.select_for_update().get( pk=self.custom_object_type_id ) - cot.next_schema_id = cot.next_schema_id + 1 - cot.save(update_fields=['next_schema_id']) - self.schema_id = cot.next_schema_id + new_schema_id = cot.next_schema_id + 1 + # Use update() rather than save() to avoid dispatching post_save on + # CustomObjectType, which would clear the model cache prematurely. + # The model cache must remain valid until this field's own save() calls + # get_model() below (to contribute the new field and alter the DB table). + CustomObjectType.objects.filter(pk=self.custom_object_type_id).update( + next_schema_id=new_schema_id + ) + self.schema_id = new_schema_id field_type = FIELD_TYPE_CLASS[self.type]() model_field = field_type.get_model_field(self) From 9ecc5eb4333ab149ae9c0fb0e443fd0b2723d085 Mon Sep 17 00:00:00 2001 From: Brian Tiemann Date: Tue, 7 Apr 2026 05:41:52 -0400 Subject: [PATCH 06/19] Use PositiveIntegerField for schema_id fields --- .../0005_customobjecttype_next_schema_id_and_more.py | 4 ++-- netbox_custom_objects/models.py | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/netbox_custom_objects/migrations/0005_customobjecttype_next_schema_id_and_more.py b/netbox_custom_objects/migrations/0005_customobjecttype_next_schema_id_and_more.py index 0dedcd93..6484e85b 100644 --- a/netbox_custom_objects/migrations/0005_customobjecttype_next_schema_id_and_more.py +++ b/netbox_custom_objects/migrations/0005_customobjecttype_next_schema_id_and_more.py @@ -15,7 +15,7 @@ class Migration(migrations.Migration): migrations.AddField( model_name='customobjecttype', name='next_schema_id', - field=models.PositiveSmallIntegerField(default=0, editable=False), + field=models.PositiveIntegerField(default=0, editable=False), ), migrations.AddField( model_name='customobjecttype', @@ -40,7 +40,7 @@ class Migration(migrations.Migration): migrations.AddField( model_name='customobjecttypefield', name='schema_id', - field=models.PositiveSmallIntegerField(blank=True, null=True), + field=models.PositiveIntegerField(blank=True, null=True), ), migrations.AlterField( model_name='customobjecttype', diff --git a/netbox_custom_objects/models.py b/netbox_custom_objects/models.py index 2b40c118..ba395d06 100644 --- a/netbox_custom_objects/models.py +++ b/netbox_custom_objects/models.py @@ -215,7 +215,7 @@ class CustomObjectType(NetBoxModel): "Serves as the source of truth for schema history, including tombstoned fields." ), ) - next_schema_id = models.PositiveSmallIntegerField( + next_schema_id = models.PositiveIntegerField( default=0, editable=False, help_text=_( From db9e3f93d5db790fbb4ac66e10acd8b2613369e8 Mon Sep 17 00:00:00 2001 From: Brian Tiemann Date: Tue, 7 Apr 2026 05:55:04 -0400 Subject: [PATCH 07/19] Tighten jsonschema validation and add tests --- .../schemas/cot_schema_v1.json | 9 ++ .../tests/test_schema_format.py | 106 +++++++++++++++++- 2 files changed, 112 insertions(+), 3 deletions(-) diff --git a/netbox_custom_objects/schemas/cot_schema_v1.json b/netbox_custom_objects/schemas/cot_schema_v1.json index 4de0abe1..01fac763 100644 --- a/netbox_custom_objects/schemas/cot_schema_v1.json +++ b/netbox_custom_objects/schemas/cot_schema_v1.json @@ -102,6 +102,7 @@ }, "text_field": { + "unevaluatedProperties": false, "allOf": [ { "$ref": "#/$defs/field_base" }, { @@ -114,6 +115,7 @@ }, "longtext_field": { + "unevaluatedProperties": false, "allOf": [ { "$ref": "#/$defs/field_base" }, { @@ -126,6 +128,7 @@ }, "integer_field": { + "unevaluatedProperties": false, "allOf": [ { "$ref": "#/$defs/field_base" }, { @@ -139,6 +142,7 @@ }, "decimal_field": { + "unevaluatedProperties": false, "allOf": [ { "$ref": "#/$defs/field_base" }, { @@ -152,6 +156,7 @@ }, "select_field": { + "unevaluatedProperties": false, "allOf": [ { "$ref": "#/$defs/field_base" }, { @@ -168,6 +173,7 @@ }, "multiselect_field": { + "unevaluatedProperties": false, "allOf": [ { "$ref": "#/$defs/field_base" }, { @@ -184,6 +190,7 @@ }, "object_field": { + "unevaluatedProperties": false, "allOf": [ { "$ref": "#/$defs/field_base" }, { @@ -201,6 +208,7 @@ }, "multiobject_field": { + "unevaluatedProperties": false, "allOf": [ { "$ref": "#/$defs/field_base" }, { @@ -218,6 +226,7 @@ }, "simple_field": { + "unevaluatedProperties": false, "allOf": [ { "$ref": "#/$defs/field_base" }, { diff --git a/netbox_custom_objects/tests/test_schema_format.py b/netbox_custom_objects/tests/test_schema_format.py index 9fa215db..bfaa5366 100644 --- a/netbox_custom_objects/tests/test_schema_format.py +++ b/netbox_custom_objects/tests/test_schema_format.py @@ -12,13 +12,17 @@ from pathlib import Path from django.db import IntegrityError, transaction +from django.db.models.fields import NOT_PROVIDED from django.test import TestCase, TransactionTestCase +from netbox_custom_objects.models import CustomObjectTypeField from netbox_custom_objects.schema_format import ( CHOICES_TO_SCHEMA_TYPE, + FIELD_DEFAULTS, SCHEMA_FORMAT_VERSION, SCHEMA_TYPE_TO_CHOICES, ) +from extras.choices import CustomFieldTypeChoices from .base import CustomObjectsTestCase, TransactionCleanupMixin @@ -135,7 +139,6 @@ def test_can_mark_field_deprecated(self): # Re-fetch from DB so that from_db() is called and self.original is set, # which the save() method requires when updating an existing field. - from netbox_custom_objects.models import CustomObjectTypeField field = CustomObjectTypeField.objects.get(custom_object_type=cot, name='label') field.deprecated = True field.deprecated_since = '2.0.0' @@ -152,7 +155,6 @@ def test_deprecation_version_strings_accept_semver(self): cot = self.create_custom_object_type(name='depsemver', slug='dep-semver') self.create_custom_object_type_field(cot, name='label', type='text') - from netbox_custom_objects.models import CustomObjectTypeField field = CustomObjectTypeField.objects.get(custom_object_type=cot, name='label') field.deprecated = True field.deprecated_since = '1.0.0-alpha.1+build.42' @@ -208,6 +210,90 @@ def test_version_field_optional(self): # schema_format.py constants tests # =========================================================================== +class FieldDefaultsConsistencyTestCase(TestCase): + """ + FIELD_DEFAULTS in schema_format.py must stay in sync with the corresponding + Django model field defaults on CustomObjectTypeField. + + When a model field's default changes, the test will fail and remind the + developer to update FIELD_DEFAULTS to match. + """ + + # Maps each FIELD_DEFAULTS key to the field name on CustomObjectTypeField. + # Only includes fields that carry an explicit model-level default= argument. + # Fields that are blank=True / null=True without an explicit default are + # tracked in _NO_MODEL_DEFAULT_SENTINELS below. + _MODEL_DEFAULT_FIELDS = { + "primary": "primary", + "required": "required", + "unique": "unique", + "weight": "weight", + "search_weight": "search_weight", + "filter_logic": "filter_logic", + "ui_visible": "ui_visible", + "ui_editable": "ui_editable", + "is_cloneable": "is_cloneable", + "deprecated": "deprecated", + } + + # Fields where the model has no explicit default (blank/null only). + # FIELD_DEFAULTS should use "" or None as the schema-level sentinel. + _NO_MODEL_DEFAULT_SENTINELS = { + "label": "", + "description": "", + "group_name": "", + "deprecated_since": "", + "scheduled_removal": "", + "validation_regex": "", + "validation_minimum": None, + "validation_maximum": None, + "related_object_filter": None, + "default": None, + } + + def test_field_defaults_match_model_defaults(self): + """Every FIELD_DEFAULTS entry with a model-level default must match it.""" + for schema_key, model_field_name in self._MODEL_DEFAULT_FIELDS.items(): + with self.subTest(field=schema_key): + model_field = CustomObjectTypeField._meta.get_field(model_field_name) + self.assertIsNot( + model_field.default, + NOT_PROVIDED, + f"{model_field_name} no longer has a model default — " + f"move {schema_key!r} to _NO_MODEL_DEFAULT_SENTINELS.", + ) + self.assertEqual( + model_field.default, + FIELD_DEFAULTS[schema_key], + f"FIELD_DEFAULTS[{schema_key!r}] is {FIELD_DEFAULTS[schema_key]!r} " + f"but {model_field_name}.default is {model_field.default!r}. " + f"Update schema_format.FIELD_DEFAULTS to match the model.", + ) + + def test_no_model_default_sentinels_are_correct(self): + """Fields without model defaults should use '' or None in FIELD_DEFAULTS.""" + for schema_key, expected_sentinel in self._NO_MODEL_DEFAULT_SENTINELS.items(): + with self.subTest(field=schema_key): + self.assertEqual( + FIELD_DEFAULTS[schema_key], + expected_sentinel, + f"FIELD_DEFAULTS[{schema_key!r}] should be {expected_sentinel!r} " + f"(no model default exists for this field).", + ) + # Also verify the model field genuinely has no explicit default. + try: + model_field = CustomObjectTypeField._meta.get_field(schema_key) + self.assertIs( + model_field.default, + NOT_PROVIDED, + f"{schema_key} now has a model default — " + f"move it to _MODEL_DEFAULT_FIELDS.", + ) + except Exception: + # Field may not exist on the model (type-specific virtual attrs). + pass + + class SchemaFormatConstantsTestCase(TestCase): """Sanity checks on schema_format module constants.""" @@ -225,7 +311,6 @@ def test_type_mapping_is_bijective(self): def test_all_field_types_are_mapped(self): """Every CustomFieldTypeChoices value must have a schema type entry.""" - from extras.choices import CustomFieldTypeChoices for attr in dir(CustomFieldTypeChoices): if attr.startswith('TYPE_'): value = getattr(CustomFieldTypeChoices, attr) @@ -484,3 +569,18 @@ def test_removed_field_with_extra_properties_is_invalid(self): } ], }) + + def test_active_field_with_unknown_key_is_invalid(self): + """Active fields use unevaluatedProperties: false, so unknown keys are rejected.""" + self._assert_invalid({ + "schema_version": "1", + "types": [ + { + "name": "widget", + "slug": "widget", + "fields": [ + {"id": 1, "name": "label", "type": "text", "unexpected_key": True} + ], + } + ], + }) From 7e5767aee86aa0e4c35b400389a14430d712a12e Mon Sep 17 00:00:00 2001 From: Brian Tiemann Date: Tue, 7 Apr 2026 21:32:34 -0400 Subject: [PATCH 08/19] Fix migration tree --- ...py => 0006_customobjecttypefield_related_name_and_more.py} | 2 +- ...re.py => 0007_customobjecttype_next_schema_id_and_more.py} | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) rename netbox_custom_objects/migrations/{0005_customobjecttypefield_related_name_and_more.py => 0006_customobjecttypefield_related_name_and_more.py} (96%) rename netbox_custom_objects/migrations/{0005_customobjecttype_next_schema_id_and_more.py => 0007_customobjecttype_next_schema_id_and_more.py} (95%) diff --git a/netbox_custom_objects/migrations/0005_customobjecttypefield_related_name_and_more.py b/netbox_custom_objects/migrations/0006_customobjecttypefield_related_name_and_more.py similarity index 96% rename from netbox_custom_objects/migrations/0005_customobjecttypefield_related_name_and_more.py rename to netbox_custom_objects/migrations/0006_customobjecttypefield_related_name_and_more.py index 52512800..bd1658c5 100644 --- a/netbox_custom_objects/migrations/0005_customobjecttypefield_related_name_and_more.py +++ b/netbox_custom_objects/migrations/0006_customobjecttypefield_related_name_and_more.py @@ -10,7 +10,7 @@ class Migration(migrations.Migration): dependencies = [ ('core', '0021_job_queue_name'), ('extras', '0134_owner'), - ('netbox_custom_objects', '0004_customobjecttype_group_name'), + ('netbox_custom_objects', '0005_customobjecttype_next_schema_id_and_more'), ] operations = [ diff --git a/netbox_custom_objects/migrations/0005_customobjecttype_next_schema_id_and_more.py b/netbox_custom_objects/migrations/0007_customobjecttype_next_schema_id_and_more.py similarity index 95% rename from netbox_custom_objects/migrations/0005_customobjecttype_next_schema_id_and_more.py rename to netbox_custom_objects/migrations/0007_customobjecttype_next_schema_id_and_more.py index 6484e85b..73df57b3 100644 --- a/netbox_custom_objects/migrations/0005_customobjecttype_next_schema_id_and_more.py +++ b/netbox_custom_objects/migrations/0007_customobjecttype_next_schema_id_and_more.py @@ -8,7 +8,7 @@ class Migration(migrations.Migration): dependencies = [ ('core', '0021_job_queue_name'), ('extras', '0134_owner'), - ('netbox_custom_objects', '0004_customobjecttype_group_name'), + ('netbox_custom_objects', '0006_customobjecttypefield_related_name_and_more'), ] operations = [ @@ -55,4 +55,4 @@ class Migration(migrations.Migration): name='netbox_custom_objects_customobjecttypefield_unique_schema_id', ), ), - ] + ] \ No newline at end of file From 8b527fdcd378a09b9536b5e74d56a0ff129a775d Mon Sep 17 00:00:00 2001 From: Brian Tiemann Date: Tue, 7 Apr 2026 21:35:31 -0400 Subject: [PATCH 09/19] Fix migration tree --- .../0006_customobjecttypefield_related_name_and_more.py | 2 +- netbox_custom_objects/models.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/netbox_custom_objects/migrations/0006_customobjecttypefield_related_name_and_more.py b/netbox_custom_objects/migrations/0006_customobjecttypefield_related_name_and_more.py index bd1658c5..8eb63d6c 100644 --- a/netbox_custom_objects/migrations/0006_customobjecttypefield_related_name_and_more.py +++ b/netbox_custom_objects/migrations/0006_customobjecttypefield_related_name_and_more.py @@ -10,7 +10,7 @@ class Migration(migrations.Migration): dependencies = [ ('core', '0021_job_queue_name'), ('extras', '0134_owner'), - ('netbox_custom_objects', '0005_customobjecttype_next_schema_id_and_more'), + ('netbox_custom_objects', '0005_customobjecttypefield_context'), ] operations = [ diff --git a/netbox_custom_objects/models.py b/netbox_custom_objects/models.py index 29db2a60..a0438e89 100644 --- a/netbox_custom_objects/models.py +++ b/netbox_custom_objects/models.py @@ -975,7 +975,7 @@ class CustomObjectTypeField(CloningMixin, ExportTemplatesMixin, ChangeLoggedMode help_text=_("Replicate this value when cloning objects"), ) comments = models.TextField(verbose_name=_("comments"), blank=True) - schema_id = models.PositiveSmallIntegerField( + schema_id = models.PositiveIntegerField( blank=True, null=True, verbose_name=_("schema ID"), From 74a8c31518452982b45dc5588a78cf20e727364f Mon Sep 17 00:00:00 2001 From: Brian Tiemann Date: Tue, 7 Apr 2026 21:36:33 -0400 Subject: [PATCH 10/19] Ruff fix --- .../migrations/0007_customobjecttype_next_schema_id_and_more.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/netbox_custom_objects/migrations/0007_customobjecttype_next_schema_id_and_more.py b/netbox_custom_objects/migrations/0007_customobjecttype_next_schema_id_and_more.py index 73df57b3..6af13db9 100644 --- a/netbox_custom_objects/migrations/0007_customobjecttype_next_schema_id_and_more.py +++ b/netbox_custom_objects/migrations/0007_customobjecttype_next_schema_id_and_more.py @@ -55,4 +55,4 @@ class Migration(migrations.Migration): name='netbox_custom_objects_customobjecttypefield_unique_schema_id', ), ), - ] \ No newline at end of file + ] From 921ccd46d178d7c75b5dd06b728c5ecaa243e7fb Mon Sep 17 00:00:00 2001 From: bctiemann Date: Wed, 8 Apr 2026 11:54:45 -0400 Subject: [PATCH 11/19] Closes #388: CustomObjectType schema exporter (#450) * CustomObjectType schema exporter * Ruff fixes * Fix MRO and add validation/notes * Release v0.4.7 (#448) * Fixes: #441 - Patch ObjectSelectorView to support targeting custom objects from core custom fields (#445) * Release v0.4.8 (#451) --- netbox_custom_objects/exporter.py | 224 +++++++++ .../migrations/0006_backfill_schema_ids.py | 63 +++ netbox_custom_objects/tests/test_exporter.py | 434 ++++++++++++++++++ .../tests/test_schema_format.py | 191 ++++++++ 4 files changed, 912 insertions(+) create mode 100644 netbox_custom_objects/exporter.py create mode 100644 netbox_custom_objects/migrations/0006_backfill_schema_ids.py create mode 100644 netbox_custom_objects/tests/test_exporter.py diff --git a/netbox_custom_objects/exporter.py b/netbox_custom_objects/exporter.py new file mode 100644 index 00000000..e9ec4d71 --- /dev/null +++ b/netbox_custom_objects/exporter.py @@ -0,0 +1,224 @@ +""" +Exporter for the COT portable schema format (issue #388). + +Converts live CustomObjectType DB state into a schema document dict that +conforms to cot_schema_v1.json. The returned dict can be serialised to YAML +or JSON by the caller. + +Public API +---------- + export_cot(cot) → dict # single COT definition (no top-level wrapper) + export_cots(cots) → dict # full schema document { schema_version, types } + +Notes +----- +- Fields without a schema_id (created before the schema-format feature) are + skipped with a WARNING log entry. They cannot be tracked across installs. +- Attribute values that equal FIELD_DEFAULTS are omitted to keep the output + minimal (round-trip safe: the importer re-applies the same defaults). +- Tombstones (removed_fields) are read from the COT's schema_document. Until + the apply endpoint (#390) is implemented this will always be empty; once + apply is wired up, deletions will be persisted there automatically. +""" + +import logging +import re + +from netbox_custom_objects import constants +from netbox_custom_objects.schema_format import ( + CHOICES_TO_SCHEMA_TYPE, + CUSTOM_OBJECTS_APP_LABEL_SLUG, + FIELD_DEFAULTS, + FIELD_TYPE_ATTRS, + SCHEMA_FORMAT_VERSION, +) + +logger = logging.getLogger(__name__) + +# Matches the generated model name produced by CustomObjectType.get_table_model_name(). +# Capturing group 1 is the numeric COT id. +_TABLE_MODEL_RE = re.compile(r'^table(\d+)model$', re.IGNORECASE) + +# Ordered list of field_base attributes to check for non-default values. +# Type-specific attributes (validation_*, choice_set, related_*) are handled +# separately via FIELD_TYPE_ATTRS. +_BASE_ATTRS = ( + "label", + "description", + "group_name", + "primary", + "required", + "unique", + "default", + "weight", + "search_weight", + "filter_logic", + "ui_visible", + "ui_editable", + "is_cloneable", + "deprecated", + "deprecated_since", + "scheduled_removal", +) + + +# --------------------------------------------------------------------------- +# Internal helpers +# --------------------------------------------------------------------------- + +def _encode_related_object_type(rot) -> str: + """ + Encode an ObjectType FK as a schema ``related_object_type`` string. + + Built-in NetBox objects → ``"/"`` (e.g. ``"dcim/device"``) + Custom Object Types → ``"custom-objects/"`` + """ + if rot.app_label == constants.APP_LABEL: + m = _TABLE_MODEL_RE.match(rot.model) + if m: + # Avoid a circular import — import here so the module can be loaded + # independently of the full Django app stack in unit tests. + from netbox_custom_objects.models import CustomObjectType # noqa: PLC0415 + cot_id = int(m.group(1)) + slug = CustomObjectType.objects.values_list('slug', flat=True).get(pk=cot_id) + return f"{CUSTOM_OBJECTS_APP_LABEL_SLUG}/{slug}" + return f"{rot.app_label}/{rot.model}" + + +def _export_field(field) -> dict: + """ + Serialise a single ``CustomObjectTypeField`` instance to a schema field dict. + + Raises ``ValueError`` if ``field.schema_id`` is ``None``; callers should + pre-filter or handle this case before calling this function. + """ + if field.schema_id is None: + raise ValueError( + f"Field {field.name!r} on COT {field.custom_object_type_id!r} " + "has no schema_id and cannot be exported." + ) + + schema_type = CHOICES_TO_SCHEMA_TYPE[field.type] + + result = { + "id": field.schema_id, + "name": field.name, + "type": schema_type, + } + + # ── Base attributes (omit when equal to documented defaults) ──────────── + for attr in _BASE_ATTRS: + value = getattr(field, attr) + if value != FIELD_DEFAULTS.get(attr): + result[attr] = value + + # ── Type-specific attributes ───────────────────────────────────────────── + for attr in sorted(FIELD_TYPE_ATTRS[schema_type]): + if attr == "choice_set": + # Required for select/multiselect; validate. + if field.choice_set is None: + raise ValueError( + f"Field {field.name!r} is type {schema_type!r} but has no choice_set assigned." + ) + result["choice_set"] = field.choice_set.name + elif attr == "related_object_type": + # Required for object/multiobject; always present. + result["related_object_type"] = _encode_related_object_type( + field.related_object_type + ) + elif attr == "related_object_filter": + value = field.related_object_filter + if value != FIELD_DEFAULTS.get("related_object_filter"): + result["related_object_filter"] = value + elif attr in ("validation_regex", "validation_minimum", "validation_maximum"): + value = getattr(field, attr) + if value != FIELD_DEFAULTS.get(attr): + result[attr] = value + + return result + + +def _removed_fields_from_document(cot) -> list: + """ + Extract the ``removed_fields`` tombstone list for *cot* from its stored + ``schema_document``. Returns an empty list if the document is absent or + does not reference this COT. + """ + if not cot.schema_document: + return [] + # NOTE: matches by COT name. If the COT is renamed after tombstones + # are persisted, they will not be found. This will be addressed when + # #390 (apply) is implemented and tombstones are managed more explicitly. + for type_def in cot.schema_document.get("types", []): + if type_def.get("name") == cot.name: + return list(type_def.get("removed_fields", [])) + return [] + + +# --------------------------------------------------------------------------- +# Public API +# --------------------------------------------------------------------------- + +def export_cot(cot) -> dict: + """ + Serialise a single ``CustomObjectType`` to its schema definition dict + (the inner object that goes inside the ``types`` list). + + Fields without a ``schema_id`` are skipped; a WARNING is logged for each. + """ + result: dict = { + "name": cot.name, + "slug": cot.slug, + } + + # Optional COT-level attributes — omit when blank/unset. + if cot.version: + result["version"] = cot.version + if cot.verbose_name: + result["verbose_name"] = cot.verbose_name + if cot.verbose_name_plural: + result["verbose_name_plural"] = cot.verbose_name_plural + if cot.description: + result["description"] = cot.description + if cot.group_name: + result["group_name"] = cot.group_name + + # Active + deprecated fields, ordered by schema_id for stable output. + exported_fields = [] + for field in cot.fields.order_by("schema_id"): + if field.schema_id is None: + logger.warning( + "Skipping field %r on COT %r during export: no schema_id assigned. " + "This field was likely created before the schema-format feature was " + "introduced and cannot be tracked portably.", + field.name, + cot.name, + ) + continue + exported_fields.append(_export_field(field)) + + if exported_fields: + result["fields"] = exported_fields + + # Tombstones from previous apply operations. + removed = _removed_fields_from_document(cot) + if removed: + result["removed_fields"] = removed + + return result + + +def export_cots(cots) -> dict: + """ + Serialise one or more ``CustomObjectType`` instances to a complete schema + document dict (``{ schema_version, types }``) that validates against + ``cot_schema_v1.json``. + + *cots* may be any iterable of ``CustomObjectType`` instances. + """ + if not cots: + raise ValueError("Minimum 1 Custom Object Type required.") + return { + "schema_version": SCHEMA_FORMAT_VERSION, + "types": [export_cot(cot) for cot in cots], + } diff --git a/netbox_custom_objects/migrations/0006_backfill_schema_ids.py b/netbox_custom_objects/migrations/0006_backfill_schema_ids.py new file mode 100644 index 00000000..8be00e2f --- /dev/null +++ b/netbox_custom_objects/migrations/0006_backfill_schema_ids.py @@ -0,0 +1,63 @@ +""" +Data migration: assign schema_id to existing CustomObjectTypeField rows that +predate the schema-format feature and never received one. + +Strategy +-------- +For each CustomObjectType: + 1. Find the current maximum schema_id already in use (may be 0 if none). + 2. Assign the next available integer to every field with schema_id=NULL, + ordered by the field's primary-key (creation order) for determinism. + 3. Update next_schema_id on the parent CustomObjectType to the highest ID + now assigned, so that future field additions continue from the right value. + +The reverse operation is intentionally a no-op: rolling back would leave the +schema_id column in an indeterminate state, and re-running the forward +migration is safe (it only touches NULL rows). +""" + +from django.db import migrations +from django.db.models import Max + + +# Exposed as a module-level name so tests can import and call it directly +# without going through the migration runner. +def assign_schema_ids(apps, schema_editor): + CustomObjectType = apps.get_model('netbox_custom_objects', 'CustomObjectType') + CustomObjectTypeField = apps.get_model('netbox_custom_objects', 'CustomObjectTypeField') + + for cot in CustomObjectType.objects.all(): + # Highest schema_id already in use for this COT (0 if none). + current_max = ( + CustomObjectTypeField.objects + .filter(custom_object_type=cot, schema_id__isnull=False) + .aggregate(max_id=Max('schema_id'))['max_id'] or 0 + ) + + # Assign the next integers to all unassigned fields, ordered by pk. + next_id = current_max + 1 + for field in ( + CustomObjectTypeField.objects + .filter(custom_object_type=cot, schema_id__isnull=True) + .order_by('id') + ): + CustomObjectTypeField.objects.filter(pk=field.pk).update(schema_id=next_id) + next_id += 1 + + # Sync next_schema_id upward. Never decrease it. + highest_assigned = next_id - 1 # equals current_max when no NULLs existed + if highest_assigned > cot.next_schema_id: + CustomObjectType.objects.filter(pk=cot.pk).update( + next_schema_id=highest_assigned + ) + + +class Migration(migrations.Migration): + + dependencies = [ + ('netbox_custom_objects', '0005_customobjecttype_next_schema_id_and_more'), + ] + + operations = [ + migrations.RunPython(assign_schema_ids, migrations.RunPython.noop), + ] diff --git a/netbox_custom_objects/tests/test_exporter.py b/netbox_custom_objects/tests/test_exporter.py new file mode 100644 index 00000000..aa71f6ad --- /dev/null +++ b/netbox_custom_objects/tests/test_exporter.py @@ -0,0 +1,434 @@ +""" +Tests for the COT schema exporter (issue #388). + +Covers: +- Minimal and full COT serialisation +- Default-value elision +- Encoding of built-in and custom related_object_type values +- choice_set serialisation +- Deprecated fields included in 'fields' list +- Fields without schema_id are skipped (warning emitted) +- Tombstones read from schema_document +- Multi-type document structure +- Output validates against cot_schema_v1.json +""" +import unittest +from pathlib import Path + +from django.test import TestCase, TransactionTestCase + +from netbox_custom_objects.exporter import export_cot, export_cots +from netbox_custom_objects.models import CustomObjectTypeField +from netbox_custom_objects.schema_format import SCHEMA_FORMAT_VERSION + +from .base import CustomObjectsTestCase, TransactionCleanupMixin + +try: + import jsonschema + HAS_JSONSCHEMA = True +except ImportError: + HAS_JSONSCHEMA = False + +_SCHEMA_PATH = ( + Path(__file__).resolve().parent.parent / "schemas" / "cot_schema_v1.json" +) + + +# =========================================================================== +# Helpers +# =========================================================================== + +def _field_by_name(exported_cot: dict, name: str) -> dict: + """Return the exported field dict with the given name, or raise.""" + for f in exported_cot.get("fields", []): + if f["name"] == name: + return f + raise KeyError(f"No exported field named {name!r}") + + +# =========================================================================== +# Tests +# =========================================================================== + +class ExporterBasicTestCase(CustomObjectsTestCase, TestCase): + """Basic structure and minimal-output tests.""" + + @classmethod + def setUpTestData(cls): + cls.cot = cls.create_custom_object_type( + name='widget', + slug='widget', + description='A widget', + verbose_name='Widget', + verbose_name_plural='Widgets', + version='1.0.0', + group_name='inventory', + ) + cls.field = cls.create_custom_object_type_field( + cls.cot, + name='label', + type='text', + label='', # will be omitted (default) + required=False, # will be omitted (default) + ) + + def test_top_level_structure(self): + doc = export_cots([self.cot]) + self.assertEqual(doc["schema_version"], SCHEMA_FORMAT_VERSION) + self.assertIsInstance(doc["types"], list) + self.assertEqual(len(doc["types"]), 1) + + def test_cot_required_keys(self): + cot_def = export_cot(self.cot) + self.assertEqual(cot_def["name"], "widget") + self.assertEqual(cot_def["slug"], "widget") + + def test_optional_cot_attrs_included_when_set(self): + cot_def = export_cot(self.cot) + self.assertEqual(cot_def["description"], "A widget") + self.assertEqual(cot_def["verbose_name"], "Widget") + self.assertEqual(cot_def["verbose_name_plural"], "Widgets") + self.assertEqual(cot_def["version"], "1.0.0") + self.assertEqual(cot_def["group_name"], "inventory") + + def test_optional_cot_attrs_omitted_when_blank(self): + bare = self.create_custom_object_type( + name='bare', slug='bare', + description='', verbose_name='', verbose_name_plural='', + version='', group_name='', + ) + cot_def = export_cot(bare) + for key in ("description", "verbose_name", "verbose_name_plural", + "version", "group_name"): + self.assertNotIn(key, cot_def) + + def test_field_required_keys_present(self): + f = _field_by_name(export_cot(self.cot), "label") + self.assertIn("id", f) + self.assertEqual(f["name"], "label") + self.assertEqual(f["type"], "text") + + def test_field_schema_id_matches_model(self): + f = _field_by_name(export_cot(self.cot), "label") + self.assertEqual(f["id"], self.field.schema_id) + + def test_fields_ordered_by_schema_id(self): + cot = self.create_custom_object_type(name='ordered', slug='ordered') + self.create_custom_object_type_field(cot, name='first', type='text') + self.create_custom_object_type_field(cot, name='second', type='text') + self.create_custom_object_type_field(cot, name='third', type='text') + exported = export_cot(cot) + ids = [f["id"] for f in exported["fields"]] + self.assertEqual(ids, sorted(ids)) + + def test_no_fields_key_when_no_exportable_fields(self): + bare = self.create_custom_object_type(name='nofields', slug='no-fields') + cot_def = export_cot(bare) + self.assertNotIn("fields", cot_def) + + def test_multi_type_document(self): + cot2 = self.create_custom_object_type(name='gadget', slug='gadget') + doc = export_cots([self.cot, cot2]) + names = [t["name"] for t in doc["types"]] + self.assertIn("widget", names) + self.assertIn("gadget", names) + + +class ExporterDefaultElisionTestCase(CustomObjectsTestCase, TestCase): + """Default values must be omitted; non-defaults must be present.""" + + @classmethod + def setUpTestData(cls): + cls.cot = cls.create_custom_object_type(name='elide', slug='elide') + cls.default_field = cls.create_custom_object_type_field( + cls.cot, name='plain', type='text', + label='', # default — omit + description='', # default — omit + required=False, # default — omit + primary=False, # default — omit + weight=100, # default — omit + search_weight=500, # default — omit + ) + cls.non_default_field = cls.create_custom_object_type_field( + cls.cot, name='special', type='text', + label='Special Label', + description='A description', + required=True, + primary=True, + weight=200, + search_weight=1000, + filter_logic='exact', + ui_visible='if-set', + ui_editable='no', + is_cloneable=True, + ) + + def test_default_attrs_omitted(self): + f = _field_by_name(export_cot(self.cot), "plain") + for key in ("label", "description", "required", "primary", + "weight", "search_weight"): + self.assertNotIn(key, f, f"{key!r} should be omitted when equal to default") + + def test_non_default_attrs_included(self): + f = _field_by_name(export_cot(self.cot), "special") + self.assertEqual(f["label"], "Special Label") + self.assertEqual(f["description"], "A description") + self.assertTrue(f["required"]) + self.assertTrue(f["primary"]) + self.assertEqual(f["weight"], 200) + self.assertEqual(f["search_weight"], 1000) + self.assertEqual(f["filter_logic"], "exact") + self.assertEqual(f["ui_visible"], "if-set") + self.assertEqual(f["ui_editable"], "no") + self.assertTrue(f["is_cloneable"]) + + def test_default_value_omitted_when_null(self): + """field.default=None (the default) must not appear in export.""" + f = _field_by_name(export_cot(self.cot), "plain") + self.assertNotIn("default", f) + + def test_non_null_default_value_included(self): + cot = self.create_custom_object_type(name='withdefault', slug='with-default') + self.create_custom_object_type_field( + cot, name='active', type='boolean', default=True + ) + f = _field_by_name(export_cot(cot), "active") + self.assertIn("default", f) + self.assertTrue(f["default"]) + + +class ExporterFieldTypesTestCase(CustomObjectsTestCase, TestCase): + """Type-specific attribute serialisation.""" + + @classmethod + def setUpTestData(cls): + cls.cot = cls.create_custom_object_type(name='typed', slug='typed') + cls.choice_set = cls.create_choice_set(name='Status Choices') + cls.device_ot = cls.get_device_object_type() + + def test_text_field_validation_regex_included(self): + self.create_custom_object_type_field( + self.cot, name='code', type='text', validation_regex=r'^[A-Z]{3}$' + ) + f = _field_by_name(export_cot(self.cot), "code") + self.assertEqual(f["validation_regex"], r'^[A-Z]{3}$') + + def test_text_field_no_regex_omitted(self): + self.create_custom_object_type_field( + self.cot, name='note', type='text' + ) + f = _field_by_name(export_cot(self.cot), "note") + self.assertNotIn("validation_regex", f) + + def test_integer_field_min_max_included(self): + self.create_custom_object_type_field( + self.cot, name='count', type='integer', + validation_minimum=0, validation_maximum=999 + ) + f = _field_by_name(export_cot(self.cot), "count") + self.assertEqual(f["validation_minimum"], 0) + self.assertEqual(f["validation_maximum"], 999) + + def test_integer_field_no_min_max_omitted(self): + self.create_custom_object_type_field( + self.cot, name='qty', type='integer' + ) + f = _field_by_name(export_cot(self.cot), "qty") + self.assertNotIn("validation_minimum", f) + self.assertNotIn("validation_maximum", f) + + def test_select_field_choice_set_name(self): + self.create_custom_object_type_field( + self.cot, name='status', type='select', choice_set=self.choice_set + ) + f = _field_by_name(export_cot(self.cot), "status") + self.assertEqual(f["choice_set"], "Status Choices") + + def test_object_field_builtin_encoding(self): + self.create_custom_object_type_field( + self.cot, name='device', type='object', + related_object_type=self.device_ot + ) + f = _field_by_name(export_cot(self.cot), "device") + self.assertEqual(f["related_object_type"], "dcim/device") + + def test_object_field_custom_cot_encoding(self): + other = self.create_custom_object_type(name='rack', slug='rack') + rack_ot = other.object_type + self.create_custom_object_type_field( + self.cot, name='rack', type='object', + related_object_type=rack_ot + ) + f = _field_by_name(export_cot(self.cot), "rack") + self.assertEqual(f["related_object_type"], "custom-objects/rack") + + def test_object_field_filter_included_when_set(self): + self.create_custom_object_type_field( + self.cot, name='filtered_device', type='object', + related_object_type=self.device_ot, + related_object_filter={"site_id": [1, 2]} + ) + f = _field_by_name(export_cot(self.cot), "filtered_device") + self.assertEqual(f["related_object_filter"], {"site_id": [1, 2]}) + + def test_object_field_filter_omitted_when_null(self): + self.create_custom_object_type_field( + self.cot, name='unfiltered', type='object', + related_object_type=self.device_ot + ) + f = _field_by_name(export_cot(self.cot), "unfiltered") + self.assertNotIn("related_object_filter", f) + + def test_type_specific_attrs_not_leaked_across_types(self): + """A boolean field must not carry validation_regex or choice_set.""" + self.create_custom_object_type_field( + self.cot, name='flag', type='boolean' + ) + f = _field_by_name(export_cot(self.cot), "flag") + for spurious in ("validation_regex", "validation_minimum", + "validation_maximum", "choice_set", + "related_object_type", "related_object_filter"): + self.assertNotIn(spurious, f) + + +class ExporterDeprecationTestCase(CustomObjectsTestCase, TestCase): + """Deprecated fields are exported in 'fields', not removed_fields.""" + + @classmethod + def setUpTestData(cls): + cls.cot = cls.create_custom_object_type(name='lifecycle', slug='lifecycle') + cls.create_custom_object_type_field(cls.cot, name='active', type='text') + # Reload from DB and mark deprecated + dep = CustomObjectTypeField.objects.get( + custom_object_type=cls.cot, name='active' + ) + dep.deprecated = True + dep.deprecated_since = '1.1.0' + dep.scheduled_removal = '2.0.0' + dep.save() + + def test_deprecated_field_in_fields_list(self): + cot_def = export_cot(self.cot) + names = [f["name"] for f in cot_def.get("fields", [])] + self.assertIn("active", names) + + def test_deprecated_attrs_exported(self): + f = _field_by_name(export_cot(self.cot), "active") + self.assertTrue(f["deprecated"]) + self.assertEqual(f["deprecated_since"], "1.1.0") + self.assertEqual(f["scheduled_removal"], "2.0.0") + + def test_deprecated_field_not_in_removed_fields(self): + cot_def = export_cot(self.cot) + self.assertNotIn("removed_fields", cot_def) + + +class ExporterSchemaIdTestCase( + TransactionCleanupMixin, CustomObjectsTestCase, TransactionTestCase +): + """Fields without schema_id are skipped with a warning.""" + + def test_field_without_schema_id_is_skipped(self): + cot = self.create_custom_object_type(name='skiptest', slug='skip-test') + field = self.create_custom_object_type_field(cot, name='noid', type='text') + # Force schema_id to None after creation (simulating a pre-feature field) + CustomObjectTypeField.objects.filter(pk=field.pk).update(schema_id=None) + + with self.assertLogs('netbox_custom_objects.exporter', level='WARNING') as cm: + cot_def = export_cot(cot) + + self.assertNotIn("fields", cot_def) + self.assertTrue(any("noid" in line for line in cm.output)) + + +class ExporterTombstoneTestCase(CustomObjectsTestCase, TestCase): + """removed_fields are read from schema_document.""" + + @classmethod + def setUpTestData(cls): + cls.cot = cls.create_custom_object_type(name='tombstone', slug='tombstone') + cls.cot.schema_document = { + "schema_version": "1", + "types": [ + { + "name": "tombstone", + "slug": "tombstone", + "fields": [], + "removed_fields": [ + {"id": 7, "name": "old_field", "type": "text", + "removed_in": "2.0.0"} + ], + } + ], + } + cls.cot.save(update_fields=["schema_document"]) + + def test_removed_fields_from_schema_document(self): + cot_def = export_cot(self.cot) + self.assertIn("removed_fields", cot_def) + self.assertEqual(len(cot_def["removed_fields"]), 1) + self.assertEqual(cot_def["removed_fields"][0]["id"], 7) + self.assertEqual(cot_def["removed_fields"][0]["name"], "old_field") + + def test_no_removed_fields_key_when_document_empty(self): + bare = self.create_custom_object_type(name='noremoved', slug='no-removed') + cot_def = export_cot(bare) + self.assertNotIn("removed_fields", cot_def) + + +@unittest.skipUnless(HAS_JSONSCHEMA, "jsonschema not installed") +class ExporterSchemaValidationTestCase(CustomObjectsTestCase, TestCase): + """Exported documents must validate against cot_schema_v1.json.""" + + _validator = None + + @classmethod + def setUpClass(cls): + super().setUpClass() + import json + raw = json.loads(_SCHEMA_PATH.read_text()) + cls._validator = jsonschema.Draft202012Validator(raw) + + def _assert_valid(self, doc): + errors = list(self._validator.iter_errors(doc)) + if errors: + self.fail( + "Schema validation failed:\n" + + "\n".join(f" {e.json_path}: {e.message}" for e in errors) + ) + + @classmethod + def setUpTestData(cls): + cls.cot = cls.create_custom_object_type( + name='schvalid', slug='sch-valid', version='1.0.0' + ) + cls.choice_set = cls.create_choice_set(name='Sch Valid Choices') + cls.device_ot = cls.get_device_object_type() + + cls.create_custom_object_type_field( + cls.cot, name='label', type='text', + required=True, primary=True + ) + cls.create_custom_object_type_field( + cls.cot, name='count', type='integer', + validation_minimum=0 + ) + cls.create_custom_object_type_field( + cls.cot, name='status', type='select', + choice_set=cls.choice_set + ) + cls.create_custom_object_type_field( + cls.cot, name='device', type='object', + related_object_type=cls.device_ot + ) + + def test_single_cot_validates(self): + self._assert_valid(export_cots([self.cot])) + + def test_minimal_cot_validates(self): + bare = self.create_custom_object_type(name='barevalid', slug='bare-valid') + self._assert_valid(export_cots([bare])) + + def test_multi_type_document_validates(self): + cot2 = self.create_custom_object_type(name='second', slug='second') + self._assert_valid(export_cots([self.cot, cot2])) diff --git a/netbox_custom_objects/tests/test_schema_format.py b/netbox_custom_objects/tests/test_schema_format.py index bfaa5366..bd296b36 100644 --- a/netbox_custom_objects/tests/test_schema_format.py +++ b/netbox_custom_objects/tests/test_schema_format.py @@ -165,6 +165,197 @@ def test_deprecation_version_strings_accept_semver(self): self.assertEqual(field.deprecated_since, '1.0.0-alpha.1+build.42') +# =========================================================================== +# Backfill migration logic tests (TransactionTestCase — uses DDL) +# =========================================================================== + +class SchemaIdBackfillTestCase( + TransactionCleanupMixin, CustomObjectsTestCase, TransactionTestCase +): + """ + Tests for the 0006_backfill_schema_ids migration logic. + + Rather than exercising the migration runner itself (which would require + replaying the full migration history), these tests call the same backfill + function directly and verify its behaviour against real model instances. + """ + + @staticmethod + def _run_backfill(): + """Execute the backfill function directly against the live DB.""" + import importlib + mod = importlib.import_module( + 'netbox_custom_objects.migrations.0006_backfill_schema_ids' + ) + # The function accepts (apps, schema_editor) but only uses apps.get_model(). + # We pass a lightweight shim that delegates to the real models. + + class _AppsShim: + @staticmethod + def get_model(app_label, model_name): + from django.apps import apps + return apps.get_model(app_label, model_name) + + mod.assign_schema_ids(_AppsShim(), None) + + # ------------------------------------------------------------------ + # Core backfill behaviour + # ------------------------------------------------------------------ + + def test_fields_without_schema_id_are_assigned(self): + cot = self.create_custom_object_type(name='bf1', slug='bf-1') + f1 = self.create_custom_object_type_field(cot, name='alpha', type='text') + f2 = self.create_custom_object_type_field(cot, name='beta', type='text') + # Force both to NULL (simulate pre-feature state) + CustomObjectTypeField.objects.filter( + custom_object_type=cot + ).update(schema_id=None) + + self._run_backfill() + + f1.refresh_from_db() + f2.refresh_from_db() + self.assertIsNotNone(f1.schema_id) + self.assertIsNotNone(f2.schema_id) + + def test_assigned_ids_are_sequential_from_one(self): + cot = self.create_custom_object_type(name='bf2', slug='bf-2') + [ + self.create_custom_object_type_field(cot, name=f'f{i}', type='text') + for i in range(1, 4) + ] + CustomObjectTypeField.objects.filter( + custom_object_type=cot + ).update(schema_id=None) + + self._run_backfill() + + ids = sorted( + CustomObjectTypeField.objects.filter(custom_object_type=cot) + .values_list('schema_id', flat=True) + ) + self.assertEqual(ids, [1, 2, 3]) + + def test_existing_schema_ids_are_not_changed(self): + cot = self.create_custom_object_type(name='bf3', slug='bf-3') + f_existing = self.create_custom_object_type_field(cot, name='kept', type='text') + existing_id = f_existing.schema_id # set by auto-assign + self.assertIsNotNone(existing_id) + + # Add a second field and force it to NULL + f_new = self.create_custom_object_type_field(cot, name='new_one', type='text') + CustomObjectTypeField.objects.filter(pk=f_new.pk).update(schema_id=None) + + self._run_backfill() + + f_existing.refresh_from_db() + f_new.refresh_from_db() + self.assertEqual(f_existing.schema_id, existing_id, "Pre-existing ID must not change") + self.assertIsNotNone(f_new.schema_id) + self.assertGreater(f_new.schema_id, existing_id, "New ID must follow existing max") + + def test_backfill_continues_from_existing_max(self): + """New IDs start after the highest already-assigned ID.""" + cot = self.create_custom_object_type(name='bf4', slug='bf-4') + # Manually assign schema_id=5 to simulate a gap + f1 = self.create_custom_object_type_field(cot, name='five', type='text') + CustomObjectTypeField.objects.filter(pk=f1.pk).update(schema_id=5) + + f2 = self.create_custom_object_type_field(cot, name='null_one', type='text') + CustomObjectTypeField.objects.filter(pk=f2.pk).update(schema_id=None) + + self._run_backfill() + + f2.refresh_from_db() + self.assertEqual(f2.schema_id, 6) + + def test_next_schema_id_updated_on_cot(self): + from netbox_custom_objects.models import CustomObjectType + cot = self.create_custom_object_type(name='bf5', slug='bf-5') + for i in range(1, 4): + self.create_custom_object_type_field(cot, name=f'g{i}', type='text') + # Force all to NULL and reset counter + CustomObjectTypeField.objects.filter( + custom_object_type=cot + ).update(schema_id=None) + CustomObjectType.objects.filter(pk=cot.pk).update(next_schema_id=0) + + self._run_backfill() + + cot.refresh_from_db() + self.assertEqual(cot.next_schema_id, 3) + + def test_next_schema_id_never_decreases(self): + """If next_schema_id is already high, the backfill must not lower it.""" + from netbox_custom_objects.models import CustomObjectType + cot = self.create_custom_object_type(name='bf6', slug='bf-6') + f = self.create_custom_object_type_field(cot, name='only', type='text') + CustomObjectTypeField.objects.filter(pk=f.pk).update(schema_id=None) + # Artificially set next_schema_id to a large value + CustomObjectType.objects.filter(pk=cot.pk).update(next_schema_id=99) + + self._run_backfill() + + cot.refresh_from_db() + self.assertGreaterEqual(cot.next_schema_id, 99) + + def test_ids_unique_within_cot(self): + cot = self.create_custom_object_type(name='bf7', slug='bf-7') + for i in range(1, 6): + self.create_custom_object_type_field(cot, name=f'h{i}', type='text') + CustomObjectTypeField.objects.filter( + custom_object_type=cot + ).update(schema_id=None) + + self._run_backfill() + + ids = list( + CustomObjectTypeField.objects.filter(custom_object_type=cot) + .values_list('schema_id', flat=True) + ) + self.assertEqual(len(ids), len(set(ids)), "All schema_ids must be unique within COT") + + def test_ids_scoped_per_cot(self): + """Two different COTs both start their IDs from 1.""" + cotA = self.create_custom_object_type(name='bfA', slug='bf-a') + cotB = self.create_custom_object_type(name='bfB', slug='bf-b') + fa = self.create_custom_object_type_field(cotA, name='x', type='text') + fb = self.create_custom_object_type_field(cotB, name='x', type='text') + CustomObjectTypeField.objects.filter(pk__in=[fa.pk, fb.pk]).update(schema_id=None) + from netbox_custom_objects.models import CustomObjectType + CustomObjectType.objects.filter(pk__in=[cotA.pk, cotB.pk]).update(next_schema_id=0) + + self._run_backfill() + + fa.refresh_from_db() + fb.refresh_from_db() + self.assertEqual(fa.schema_id, 1) + self.assertEqual(fb.schema_id, 1) + + def test_idempotent(self): + """Running the backfill twice must not change any already-assigned IDs.""" + cot = self.create_custom_object_type(name='bfIdem', slug='bf-idem') + for i in range(1, 4): + self.create_custom_object_type_field(cot, name=f'i{i}', type='text') + CustomObjectTypeField.objects.filter( + custom_object_type=cot + ).update(schema_id=None) + + self._run_backfill() + ids_after_first = list( + CustomObjectTypeField.objects.filter(custom_object_type=cot) + .order_by('id').values_list('schema_id', flat=True) + ) + + self._run_backfill() + ids_after_second = list( + CustomObjectTypeField.objects.filter(custom_object_type=cot) + .order_by('id').values_list('schema_id', flat=True) + ) + + self.assertEqual(ids_after_first, ids_after_second) + + # =========================================================================== # schema_document and version field tests (plain TestCase — no DDL needed) # =========================================================================== From d9a25ef8a1b38b71d74af3ea9c5fb8e02144e248 Mon Sep 17 00:00:00 2001 From: Brian Tiemann Date: Wed, 8 Apr 2026 19:11:05 -0400 Subject: [PATCH 12/19] Fix migrations --- ...{0006_backfill_schema_ids.py => 0008_backfill_schema_ids.py} | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) rename netbox_custom_objects/migrations/{0006_backfill_schema_ids.py => 0008_backfill_schema_ids.py} (97%) diff --git a/netbox_custom_objects/migrations/0006_backfill_schema_ids.py b/netbox_custom_objects/migrations/0008_backfill_schema_ids.py similarity index 97% rename from netbox_custom_objects/migrations/0006_backfill_schema_ids.py rename to netbox_custom_objects/migrations/0008_backfill_schema_ids.py index 8be00e2f..e4d522f8 100644 --- a/netbox_custom_objects/migrations/0006_backfill_schema_ids.py +++ b/netbox_custom_objects/migrations/0008_backfill_schema_ids.py @@ -55,7 +55,7 @@ def assign_schema_ids(apps, schema_editor): class Migration(migrations.Migration): dependencies = [ - ('netbox_custom_objects', '0005_customobjecttype_next_schema_id_and_more'), + ('netbox_custom_objects', '0007_customobjecttype_next_schema_id_and_more'), ] operations = [ From 9ae9ac728be962aaace04fc3d6a5df4c0834fdeb Mon Sep 17 00:00:00 2001 From: Brian Tiemann Date: Wed, 8 Apr 2026 19:24:39 -0400 Subject: [PATCH 13/19] Fix reference to 0006 migration --- netbox_custom_objects/tests/test_schema_format.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/netbox_custom_objects/tests/test_schema_format.py b/netbox_custom_objects/tests/test_schema_format.py index bd296b36..a6866d89 100644 --- a/netbox_custom_objects/tests/test_schema_format.py +++ b/netbox_custom_objects/tests/test_schema_format.py @@ -173,7 +173,7 @@ class SchemaIdBackfillTestCase( TransactionCleanupMixin, CustomObjectsTestCase, TransactionTestCase ): """ - Tests for the 0006_backfill_schema_ids migration logic. + Tests for the 0008_backfill_schema_ids migration logic. Rather than exercising the migration runner itself (which would require replaying the full migration history), these tests call the same backfill @@ -185,7 +185,7 @@ def _run_backfill(): """Execute the backfill function directly against the live DB.""" import importlib mod = importlib.import_module( - 'netbox_custom_objects.migrations.0006_backfill_schema_ids' + 'netbox_custom_objects.migrations.0008_backfill_schema_ids' ) # The function accepts (apps, schema_editor) but only uses apps.get_model(). # We pass a lightweight shim that delegates to the real models. From 3497ad2a179321ac3434f19ed8a3a5fe28055b29 Mon Sep 17 00:00:00 2001 From: Brian Tiemann Date: Tue, 21 Apr 2026 16:09:57 -0400 Subject: [PATCH 14/19] Fix COT name validation --- ...omobjecttypefield_related_name_and_more.py | 63 ---------- .../migrations/0006_portable_schema.py | 119 ++++++++++++++++++ ...ema_ids.py => 0007_backfill_schema_ids.py} | 2 +- ...ustomobjecttype_next_schema_id_and_more.py | 58 --------- netbox_custom_objects/models.py | 22 +--- .../schemas/cot_schema_v1.json | 2 +- netbox_custom_objects/tests/test_models.py | 45 ++++--- 7 files changed, 155 insertions(+), 156 deletions(-) delete mode 100644 netbox_custom_objects/migrations/0006_customobjecttypefield_related_name_and_more.py create mode 100644 netbox_custom_objects/migrations/0006_portable_schema.py rename netbox_custom_objects/migrations/{0008_backfill_schema_ids.py => 0007_backfill_schema_ids.py} (96%) delete mode 100644 netbox_custom_objects/migrations/0007_customobjecttype_next_schema_id_and_more.py diff --git a/netbox_custom_objects/migrations/0006_customobjecttypefield_related_name_and_more.py b/netbox_custom_objects/migrations/0006_customobjecttypefield_related_name_and_more.py deleted file mode 100644 index 8eb63d6c..00000000 --- a/netbox_custom_objects/migrations/0006_customobjecttypefield_related_name_and_more.py +++ /dev/null @@ -1,63 +0,0 @@ -# Generated by Django 5.2.12 on 2026-04-05 02:32 - -import django.core.validators -import re -from django.db import migrations, models - - -class Migration(migrations.Migration): - - dependencies = [ - ('core', '0021_job_queue_name'), - ('extras', '0134_owner'), - ('netbox_custom_objects', '0005_customobjecttypefield_context'), - ] - - operations = [ - migrations.AddField( - model_name='customobjecttypefield', - name='related_name', - field=models.CharField( - blank=True, - max_length=100, - validators=[ - django.core.validators.RegexValidator( - message='Only lowercase alphanumeric characters and underscores are allowed.', - regex='^[a-z0-9_]+$'), - django.core.validators.RegexValidator( - flags=re.RegexFlag['IGNORECASE'], - inverse_match=True, - message='Double underscores are not permitted in the reverse relation name.', - regex='__', - ) - ] - ), - ), - migrations.AlterField( - model_name='customobjecttypefield', - name='name', - field=models.CharField( - max_length=50, - validators=[ - django.core.validators.RegexValidator( - message='Only lowercase alphanumeric characters and underscores are allowed.', - regex='^[a-z0-9_]+$', - ), - django.core.validators.RegexValidator( - flags=re.RegexFlag['IGNORECASE'], - inverse_match=True, - message='Double underscores are not permitted in custom object field names.', - regex='__', - ) - ] - ), - ), - migrations.AddConstraint( - model_name='customobjecttypefield', - constraint=models.UniqueConstraint( - condition=models.Q(('related_name__gt', '')), - fields=('related_object_type', 'related_name'), - name='netbox_custom_objects_customobjecttypefield_unique_related_name', - ), - ), - ] diff --git a/netbox_custom_objects/migrations/0006_portable_schema.py b/netbox_custom_objects/migrations/0006_portable_schema.py new file mode 100644 index 00000000..89d2d6fd --- /dev/null +++ b/netbox_custom_objects/migrations/0006_portable_schema.py @@ -0,0 +1,119 @@ +import django.core.validators +import re +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('core', '0021_job_queue_name'), + ('extras', '0134_owner'), + ('netbox_custom_objects', '0005_customobjecttypefield_context'), + ] + + operations = [ + migrations.AddField( + model_name='customobjecttype', + name='next_schema_id', + field=models.PositiveIntegerField(default=0, editable=False), + ), + migrations.AddField( + model_name='customobjecttype', + name='schema_document', + field=models.JSONField(blank=True, null=True), + ), + migrations.AddField( + model_name='customobjecttypefield', + name='deprecated', + field=models.BooleanField(default=False), + ), + migrations.AddField( + model_name='customobjecttypefield', + name='deprecated_since', + field=models.CharField(blank=True, max_length=50), + ), + migrations.AddField( + model_name='customobjecttypefield', + name='related_name', + field=models.CharField( + blank=True, + max_length=100, + validators=[ + django.core.validators.RegexValidator( + message='Only lowercase alphanumeric characters and underscores are allowed.', + regex='^[a-z0-9_]+$', + ), + django.core.validators.RegexValidator( + flags=re.RegexFlag['IGNORECASE'], + inverse_match=True, + message='Double underscores are not permitted in the reverse relation name.', + regex='__', + ), + ] + ), + ), + migrations.AddField( + model_name='customobjecttypefield', + name='scheduled_removal', + field=models.CharField(blank=True, max_length=50), + ), + migrations.AddField( + model_name='customobjecttypefield', + name='schema_id', + field=models.PositiveIntegerField(blank=True, null=True), + ), + migrations.AlterField( + model_name='customobjecttype', + name='name', + field=models.CharField( + max_length=100, + unique=True, + validators=[ + django.core.validators.RegexValidator( + message=( + 'Only lowercase alphanumeric characters and underscores are allowed. ' + 'Names may not start or end with an underscore, and double underscores are not permitted.' + ), + regex='^[a-z0-9]+(_[a-z0-9]+)*$', + ), + ] + ), + ), + migrations.AlterField( + model_name='customobjecttype', + name='version', + field=models.CharField(blank=True, max_length=50), + ), + migrations.AlterField( + model_name='customobjecttypefield', + name='name', + field=models.CharField( + max_length=50, + validators=[ + django.core.validators.RegexValidator( + message=( + 'Only lowercase alphanumeric characters and underscores are allowed. ' + 'Names may not start or end with an underscore, and double underscores are not permitted.' + ), + regex='^[a-z0-9]+(_[a-z0-9]+)*$', + ), + ] + ), + ), + migrations.AddConstraint( + model_name='customobjecttypefield', + constraint=models.UniqueConstraint( + condition=models.Q(('related_name__gt', '')), + fields=('related_object_type', 'related_name'), + name='netbox_custom_objects_customobjecttypefield_unique_related_name', + ), + ), + migrations.AddConstraint( + model_name='customobjecttypefield', + constraint=models.UniqueConstraint( + condition=models.Q(('schema_id__isnull', False)), + fields=('schema_id', 'custom_object_type'), + name='netbox_custom_objects_customobjecttypefield_unique_schema_id', + ), + ), + ] diff --git a/netbox_custom_objects/migrations/0008_backfill_schema_ids.py b/netbox_custom_objects/migrations/0007_backfill_schema_ids.py similarity index 96% rename from netbox_custom_objects/migrations/0008_backfill_schema_ids.py rename to netbox_custom_objects/migrations/0007_backfill_schema_ids.py index e4d522f8..c201b068 100644 --- a/netbox_custom_objects/migrations/0008_backfill_schema_ids.py +++ b/netbox_custom_objects/migrations/0007_backfill_schema_ids.py @@ -55,7 +55,7 @@ def assign_schema_ids(apps, schema_editor): class Migration(migrations.Migration): dependencies = [ - ('netbox_custom_objects', '0007_customobjecttype_next_schema_id_and_more'), + ('netbox_custom_objects', '0006_portable_schema'), ] operations = [ diff --git a/netbox_custom_objects/migrations/0007_customobjecttype_next_schema_id_and_more.py b/netbox_custom_objects/migrations/0007_customobjecttype_next_schema_id_and_more.py deleted file mode 100644 index 6af13db9..00000000 --- a/netbox_custom_objects/migrations/0007_customobjecttype_next_schema_id_and_more.py +++ /dev/null @@ -1,58 +0,0 @@ -# Generated by Django 5.2.12 on 2026-04-07 01:35 - -from django.db import migrations, models - - -class Migration(migrations.Migration): - - dependencies = [ - ('core', '0021_job_queue_name'), - ('extras', '0134_owner'), - ('netbox_custom_objects', '0006_customobjecttypefield_related_name_and_more'), - ] - - operations = [ - migrations.AddField( - model_name='customobjecttype', - name='next_schema_id', - field=models.PositiveIntegerField(default=0, editable=False), - ), - migrations.AddField( - model_name='customobjecttype', - name='schema_document', - field=models.JSONField(blank=True, null=True), - ), - migrations.AddField( - model_name='customobjecttypefield', - name='deprecated', - field=models.BooleanField(default=False), - ), - migrations.AddField( - model_name='customobjecttypefield', - name='deprecated_since', - field=models.CharField(blank=True, max_length=50), - ), - migrations.AddField( - model_name='customobjecttypefield', - name='scheduled_removal', - field=models.CharField(blank=True, max_length=50), - ), - migrations.AddField( - model_name='customobjecttypefield', - name='schema_id', - field=models.PositiveIntegerField(blank=True, null=True), - ), - migrations.AlterField( - model_name='customobjecttype', - name='version', - field=models.CharField(blank=True, max_length=50), - ), - migrations.AddConstraint( - model_name='customobjecttypefield', - constraint=models.UniqueConstraint( - condition=models.Q(('schema_id__isnull', False)), - fields=('schema_id', 'custom_object_type'), - name='netbox_custom_objects_customobjecttypefield_unique_schema_id', - ), - ), - ] diff --git a/netbox_custom_objects/models.py b/netbox_custom_objects/models.py index a0438e89..c128defc 100644 --- a/netbox_custom_objects/models.py +++ b/netbox_custom_objects/models.py @@ -175,16 +175,11 @@ class CustomObjectType(NetBoxModel): unique=True, validators=( RegexValidator( - regex=r"^[a-z0-9_]+$", - message=_("Only lowercase alphanumeric characters and underscores are allowed."), - ), - RegexValidator( - regex=r"__", + regex=r"^[a-z0-9]+(_[a-z0-9]+)*$", message=_( - "Double underscores are not permitted in custom object object type names." + "Only lowercase alphanumeric characters and underscores are allowed. " + "Names may not start or end with an underscore, and double underscores are not permitted." ), - flags=re.IGNORECASE, - inverse_match=True, ), ), ) @@ -819,16 +814,11 @@ class CustomObjectTypeField(CloningMixin, ExportTemplatesMixin, ChangeLoggedMode help_text=_("Internal field name, e.g. \"vendor_label\""), validators=( RegexValidator( - regex=r"^[a-z0-9_]+$", - message=_("Only lowercase alphanumeric characters and underscores are allowed."), - ), - RegexValidator( - regex=r"__", + regex=r"^[a-z0-9]+(_[a-z0-9]+)*$", message=_( - "Double underscores are not permitted in custom object field names." + "Only lowercase alphanumeric characters and underscores are allowed. " + "Names may not start or end with an underscore, and double underscores are not permitted." ), - flags=re.IGNORECASE, - inverse_match=True, ), ), ) diff --git a/netbox_custom_objects/schemas/cot_schema_v1.json b/netbox_custom_objects/schemas/cot_schema_v1.json index 01fac763..ee6c2787 100644 --- a/netbox_custom_objects/schemas/cot_schema_v1.json +++ b/netbox_custom_objects/schemas/cot_schema_v1.json @@ -13,7 +13,7 @@ "identifier": { "type": "string", - "description": "Lowercase alphanumeric characters and single underscores only. Note: this pattern is intentionally more restrictive than the model validator (^[a-z0-9_]+$) — it disallows leading/trailing underscores and double underscores. A valid DB field name that violates this pattern will fail schema export.", + "description": "Lowercase alphanumeric characters and single underscores only. No leading/trailing underscores and no double underscores. This pattern matches the model validator for CustomObjectType.name and CustomObjectTypeField.name.", "pattern": "^[a-z0-9]+(_[a-z0-9]+)*$" }, diff --git a/netbox_custom_objects/tests/test_models.py b/netbox_custom_objects/tests/test_models.py index 324c9c06..3663bc51 100644 --- a/netbox_custom_objects/tests/test_models.py +++ b/netbox_custom_objects/tests/test_models.py @@ -39,6 +39,20 @@ def test_custom_object_type_creation(self): self.assertEqual(custom_object_type.slug, "test-objects") self.assertEqual(str(custom_object_type), "TestObject") + def test_custom_object_type_name_validation(self): + """COT name must match the schema identifier pattern (no leading/trailing/double underscores).""" + from netbox_custom_objects.models import CustomObjectType + invalid_names = [ + "test-type", # hyphen not allowed + "test__type", # double underscore not allowed + "_test_type", # leading underscore not allowed + "test_type_", # trailing underscore not allowed + ] + for invalid_name in invalid_names: + with self.assertRaises(ValidationError, msg=f"Expected ValidationError for name={invalid_name!r}"): + cot = CustomObjectType(name=invalid_name, slug=f"slug-{invalid_name}") + cot.full_clean() + def test_custom_object_type_unique_name_constraint(self): """Test that custom object type names must be unique (case-insensitive).""" self.create_custom_object_type(name="TestObject") @@ -303,23 +317,20 @@ def test_custom_object_type_field_creation(self): def test_custom_object_type_field_name_validation(self): """Test field name validation.""" - # Test invalid characters - with self.assertRaises(ValidationError): - field = CustomObjectTypeField( - custom_object_type=self.custom_object_type, - name="test-field", # Invalid: contains hyphen - type="text" - ) - field.full_clean() - - # Test double underscores - with self.assertRaises(ValidationError): - field = CustomObjectTypeField( - custom_object_type=self.custom_object_type, - name="test__field", # Invalid: contains double underscore - type="text" - ) - field.full_clean() + invalid_names = [ + "test-field", # hyphen not allowed + "test__field", # double underscore not allowed + "_test_field", # leading underscore not allowed + "test_field_", # trailing underscore not allowed + ] + for invalid_name in invalid_names: + with self.assertRaises(ValidationError, msg=f"Expected ValidationError for name={invalid_name!r}"): + field = CustomObjectTypeField( + custom_object_type=self.custom_object_type, + name=invalid_name, + type="text", + ) + field.full_clean() def test_custom_object_type_field_unique_name_per_type(self): """Test that field names must be unique within a custom object type.""" From 52fd1c5f0029dfd07d353e8080da1dc3850a6ba4 Mon Sep 17 00:00:00 2001 From: Brian Tiemann Date: Tue, 21 Apr 2026 16:23:07 -0400 Subject: [PATCH 15/19] Add documentation file --- docs/portable-schema.md | 497 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 497 insertions(+) create mode 100644 docs/portable-schema.md diff --git a/docs/portable-schema.md b/docs/portable-schema.md new file mode 100644 index 00000000..9522941c --- /dev/null +++ b/docs/portable-schema.md @@ -0,0 +1,497 @@ +# Portable Schema + +The portable schema feature allows Custom Object Type (COT) definitions to be exported as +structured JSON documents, versioned in source control, and applied to other NetBox instances. +This makes COT schemas shareable, auditable, and deployable across environments in a consistent +and repeatable way. + +## Concepts + +### Schema documents + +A schema document is a JSON object that fully describes one or more Custom Object Types — +their names, metadata, and all field definitions. The document is self-contained: a reader +does not need access to the originating NetBox instance to understand or validate it. + +```json +{ + "schema_version": "1", + "types": [ + { + "name": "circuit", + "slug": "circuit", + "verbose_name": "Circuit", + "verbose_name_plural": "Circuits", + "description": "WAN circuit inventory", + "fields": [ + { "id": 1, "name": "carrier", "type": "text", "required": true }, + { "id": 2, "name": "bandwidth_mbps", "type": "integer", "validation_minimum": 0 } + ], + "removed_fields": [] + } + ] +} +``` + +### Schema IDs + +Every `CustomObjectTypeField` carries a `schema_id` — a stable, monotonically increasing integer +scoped to its parent COT. Schema IDs are the primary key used by the comparator and executor to +match fields across export and apply cycles: + +- Assigned automatically on first save if not set explicitly. +- Scoped per COT — field 3 in COT "circuit" is unrelated to field 3 in COT "device-profile". +- **Never reused.** When a field is deleted its `schema_id` is retired. The parent COT's + `next_schema_id` counter only ever advances, even if fields are removed. +- Stable across renames — a field retains its `schema_id` when its `name` attribute changes, + so the comparator correctly identifies a rename rather than a deletion plus addition. +- Assigned atomically using `SELECT ... FOR UPDATE` on the parent COT row to prevent + race conditions under concurrent field creation. + +> **Note for bulk operations:** `bulk_create()` bypasses the model `save()` method and will +> not auto-assign `schema_id`. Callers that use `bulk_create()` must set `schema_id` +> explicitly on each field instance. + +### Tombstones + +When a field is removed from a COT, the comparator needs to distinguish "this field was +intentionally deleted" from "this field is not in the schema yet." Tombstone entries in +`removed_fields` provide that signal: + +```json +"removed_fields": [ + { "id": 4, "name": "legacy_carrier_code", "type": "text", "removed_in": "2.0.0" } +] +``` + +A tombstone records the field's last-known `id`, `name`, `type`, and the version string when +it was removed. The executor uses tombstones to drop the corresponding DB column; without a +tombstone the comparator treats the absent field as ambiguous and emits a warning rather than +a `REMOVE`. + +Tombstones are persisted in the `CustomObjectType.schema_document` field and read back into +every subsequent export, so the full removal history accumulates over time. + +### Schema document storage + +`CustomObjectType.schema_document` stores the most recently applied or exported schema +snapshot as a JSON blob. It is written by the executor after a successful apply, and read +by the exporter when building tombstone lists. Until a COT has been exported or a schema +applied, the field is `null`. + +--- + +## Design decisions + +### Integer `schema_id` instead of UUID + +Several alternatives were considered for the stable field identity value: + +| Approach | Pros | Cons | +|----------|------|------| +| **Monotonic integer** (chosen) | Human-readable diffs; encodes creation order; no collision risk in single-source model | False-matches if two instances independently assign the same ID to different fields | +| UUID | Globally unique; safe for peer-merge scenarios | Opaque in JSON diffs; no ordering information | + +Integers were chosen because the intended workflow is **single-source-of-truth**: one canonical +environment exports schemas, and downstream instances receive and apply them. In this hub-and-spoke +model there is no opportunity for two instances to independently create fields and generate +conflicting IDs, so UUID collision avoidance is unnecessary. + +This follows the same convention as Protocol Buffers, which also uses integer field IDs and +places the responsibility for incrementing them on the schema author. + +> **Important constraint:** if the product ever requires bidirectional sync or multi-master +> schema evolution — where two environments independently evolve the same COT and their schemas +> need to be reconciled — integer IDs would need to be replaced with UUIDs. The current design +> does not support that workflow. + +### Single-source-of-truth distribution model + +The feature is designed around a hub-and-spoke topology: + +1. A **canonical environment** (e.g., a development instance or a dedicated schema registry) + defines and maintains COT schemas. +2. Schema documents are exported and committed to version control. +3. **Downstream instances** (staging, production) receive schema documents and apply them via + the API. + +The comparator and executor handle the downstream side: they diff an incoming schema against +the live DB state and apply the delta atomically. There is no merge logic for reconciling +independent changes from multiple peers. + +### Identifier pattern alignment + +COT names (`CustomObjectType.name`) and field names (`CustomObjectTypeField.name`) must satisfy +the pattern `^[a-z0-9]+(_[a-z0-9]+)*$`. This is the same pattern used by the JSON Schema +`identifier` definition, ensuring that any name accepted by the database will also pass schema +validation and round-trip cleanly through export and apply cycles. + +The pattern permits: +- `circuit`, `bandwidth_mbps`, `carrier_code_v2` + +The pattern rejects: +- `_private` (leading underscore) +- `foo_` (trailing underscore) +- `test__field` (double underscore) +- `my-field` (hyphen) + +--- + +## Schema document format + +The JSON Schema validator for schema documents lives at +`netbox_custom_objects/schemas/cot_schema_v1.json` and is used by the API endpoints to +validate incoming documents before any DB access. + +### Top-level structure + +| Key | Type | Description | +|-----|------|-------------| +| `schema_version` | `"1"` | Format version. Currently only `"1"` is supported. | +| `types` | array of COT definitions | One entry per Custom Object Type. | + +### COT definition + +| Key | Required | Description | +|-----|----------|-------------| +| `name` | yes | Internal name, must match identifier pattern. | +| `slug` | yes | URL-safe slug. Used as the stable lookup key when applying. | +| `verbose_name` | no | Singular display name. | +| `verbose_name_plural` | no | Plural display name. | +| `version` | no | Free-form version string for the COT schema (e.g. `"2.1.0"`). | +| `description` | no | Short description. | +| `fields` | yes | Array of active field definitions. | +| `removed_fields` | no | Array of tombstone records for previously removed fields. | + +### Field definition + +All fields share these base attributes: + +| Key | Required | Description | +|-----|----------|-------------| +| `id` | yes | Stable integer schema ID (>= 1). | +| `name` | yes | Internal field name, must match identifier pattern. | +| `type` | yes | Field type (see below). | +| `label` | no | Display label. Defaults to `name` with underscores replaced by spaces. | +| `description` | no | Help text shown in the UI. | +| `group_name` | no | UI grouping. | +| `primary` | no | Whether this is the primary display field. | +| `required` | no | Whether the field is required. Default: `false`. | +| `unique` | no | Whether values must be unique. Default: `false`. | +| `default` | no | JSON default value. | +| `weight` | no | Display order weight. Default: `100`. | +| `search_weight` | no | Search relevance weight. Default: `500`. | +| `filter_logic` | no | `"loose"`, `"exact"`, or `"disabled"`. Default: `"loose"`. | +| `ui_visible` | no | `"always"`, `"if-set"`, or `"hidden"`. Default: `"always"`. | +| `ui_editable` | no | `"yes"`, `"no"`, or `"hidden"`. Default: `"yes"`. | +| `is_cloneable` | no | Whether the field is copied when cloning objects. Default: `false`. | +| `deprecated` | no | Marks the field as deprecated. Default: `false`. | +| `deprecated_since` | no | Version string when the field was deprecated (e.g. `"2.0.0"`). | +| `scheduled_removal` | no | Version string when the field is planned for removal (e.g. `"3.0.0"`). | + +Attributes that match their defaults are omitted from exported documents to keep output minimal. + +### Field types and type-specific attributes + +| Type | Additional attributes | +|------|-----------------------| +| `text`, `longtext` | `validation_regex` | +| `integer`, `decimal` | `validation_minimum`, `validation_maximum` | +| `select`, `multiselect` | `choice_set` (required — name of a `CustomFieldChoiceSet`) | +| `object`, `multiobject` | `related_object_type` (required), `related_object_filter` | +| `boolean`, `date`, `datetime`, `url`, `json` | (none) | + +### Related object type encoding + +`related_object_type` is encoded as a `/`-separated string: + +- **Built-in NetBox model:** `"dcim/device"`, `"ipam/prefix"` +- **Custom Object Type:** `"custom-objects/"`, e.g. `"custom-objects/circuit"` + +### Tombstone record + +```json +{ + "id": 4, + "name": "legacy_carrier_code", + "type": "text", + "removed_in": "2.0.0" +} +``` + +`removed_in` is optional but recommended. The `id` value must match the original field's +`schema_id` and must not appear in the active `fields` list. + +--- + +## Usage + +### Exporting a schema + +Use the Python API from the `exporter` module. This is typically called from a management +command or script: + +```python +from netbox_custom_objects.exporter import export_cots +from netbox_custom_objects.models import CustomObjectType + +cots = CustomObjectType.objects.filter(slug__in=["circuit", "device-profile"]) +document = export_cots(cots) + +import json +print(json.dumps(document, indent=2)) +``` + +`export_cots` returns a dict with `schema_version` and `types`. For a single COT without the +document wrapper, use `export_cot(cot)`. + +> **Fields without a `schema_id`** (created before the portable schema feature was introduced) +> are skipped with a `WARNING` log entry. Run the backfill migration (see below) to assign IDs +> to pre-existing fields. + +### Previewing a schema (API) + +`POST /api/plugins/custom-objects/schema/preview/` + +Submit a schema document and receive a structured diff showing what would change, **without +modifying the database**: + +```http +POST /api/plugins/custom-objects/schema/preview/ +Content-Type: application/json +Authorization: Token + +{ + "schema_version": "1", + "types": [ + { + "name": "circuit", + "slug": "circuit", + "verbose_name_plural": "Circuits", + "fields": [ + { "id": 1, "name": "carrier", "type": "text", "required": true }, + { "id": 3, "name": "contract_ref", "type": "text" } + ], + "removed_fields": [ + { "id": 2, "name": "bandwidth_mbps", "type": "integer", "removed_in": "2.0.0" } + ] + } + ] +} +``` + +Response `200`: + +```json +{ + "diffs": [ + { + "slug": "circuit", + "name": "circuit", + "is_new": false, + "has_changes": true, + "has_destructive_changes": true, + "cot_changes": {}, + "field_changes": [ + { + "op": "add", + "schema_id": 3, + "db_name": null, + "schema_def": { "id": 3, "name": "contract_ref", "type": "text" } + }, + { + "op": "remove", + "schema_id": 2, + "db_name": "bandwidth_mbps", + "schema_def": { "id": 2, "name": "bandwidth_mbps", "type": "integer", "removed_in": "2.0.0" } + } + ], + "warnings": [] + } + ] +} +``` + +`has_destructive_changes: true` indicates that applying this schema would drop at least one +column. The preview endpoint never returns `409` — it is safe to call at any time. + +### Applying a schema (API) + +`POST /api/plugins/custom-objects/schema/apply/` + +```http +POST /api/plugins/custom-objects/schema/apply/ +Content-Type: application/json +Authorization: Token + +{ + "allow_destructive": false, + "schema": { ... } +} +``` + +- **`allow_destructive`** (default `false`): must be `true` for the apply to proceed when the + diff contains `REMOVE` operations. If `false` and removals are present, the endpoint returns + `409 Conflict`. +- The apply is **fully atomic** — a failure at any point rolls back all changes including newly + created COT tables (PostgreSQL supports transactional DDL). +- On success, `schema_document` is persisted on each affected COT so tombstones are available + for future export/diff cycles. + +Response `200`: + +```json +{ + "applied": true, + "diffs": [ ... ] +} +``` + +Response `409 Conflict`: + +```json +{ + "error": "destructive_changes", + "detail": "Schema contains destructive field removals for COT(s): circuit.", + "destructive_slugs": ["circuit"] +} +``` + +Response `400 Bad Request` (invalid schema, unresolvable reference, or circular COT dependency): + +```json +{ + "error": "unresolvable_reference", + "detail": "..." +} +``` + +### Typical end-to-end workflow + +1. **Define and iterate** on COT schemas in a development environment using the NetBox UI or + API. +2. **Export** the schemas to a JSON file and commit to version control. +3. **Review** the diff in the PR — because IDs are stable integers and defaults are elided, + the diff is human-readable. +4. **Preview** the schema on a staging instance using the preview endpoint to confirm the diff + matches expectations. +5. **Apply** the schema on staging (and then production), using `allow_destructive: true` only + when column drops have been explicitly reviewed. + +--- + +## Field deprecation lifecycle + +Fields can be marked deprecated without being removed, allowing a grace period before deletion: + +```json +{ + "id": 5, + "name": "old_carrier_name", + "type": "text", + "deprecated": true, + "deprecated_since": "2.1.0", + "scheduled_removal": "3.0.0" +} +``` + +- `deprecated: true` marks the field as read-only in the UI; no new values can be entered. +- `deprecated_since` is an informational version string (no format enforced). +- `scheduled_removal` signals to consumers when the field will be tombstoned. + +Deprecation is non-destructive. The field remains in `fields` (not `removed_fields`) until it +is actually deleted, at which point a tombstone entry should be added. + +--- + +## Comparator (developer reference) + +`netbox_custom_objects/comparator.py` — pure-read, no DB writes. + +```python +from netbox_custom_objects.comparator import diff_document, diff_cot + +diffs = diff_document(schema_doc) # list[COTDiff] +diff = diff_cot(type_def) # COTDiff +``` + +### `COTDiff` + +| Attribute | Type | Description | +|-----------|------|-------------| +| `name` | `str` | COT name from the schema document | +| `slug` | `str` | Lookup key (COT is matched by slug) | +| `is_new` | `bool` | `True` if no COT with this slug exists in the DB | +| `cot_changes` | `dict[str, tuple]` | `{attr: (db_value, schema_value)}` for changed top-level attributes | +| `field_changes` | `list[FieldChange]` | Per-field operations | +| `warnings` | `list[str]` | Non-fatal issues (untracked fields, ambiguous absences) | + +Convenience properties: `has_changes`, `has_destructive_changes`, `adds`, `removes`, `alters`. + +### `FieldChange` + +| Attribute | Type | Description | +|-----------|------|-------------| +| `op` | `FieldOp` | `ADD`, `REMOVE`, or `ALTER` | +| `schema_id` | `int` | Stable field identifier | +| `db_name` | `str \| None` | Current DB field name (`None` for `ADD`) | +| `schema_def` | `dict` | Raw field dict from the schema document | +| `changed_attrs` | `dict[str, tuple]` | `{attr: (db_value, schema_value)}` for `ALTER` operations | + +Properties: `is_rename`, `is_type_change`. + +### Matching rules + +- Fields are matched exclusively by `schema_id`. DB fields with no `schema_id` generate a + **warning**, not a `REMOVE`. +- `REMOVE` is emitted only when a `schema_id` appears in the document's `removed_fields`. + A field absent from both `fields` and `removed_fields` generates a **warning**. + +--- + +## Executor (developer reference) + +`netbox_custom_objects/executor.py` — writes to the DB. + +```python +from netbox_custom_objects.executor import apply_document, apply_diffs + +diffs = apply_document(schema_doc, allow_destructive=False) # list[COTDiff] +apply_diffs(diffs, type_defs_by_slug, allow_destructive=False) # lower-level +``` + +`apply_document` is the primary entry point. `apply_diffs` is available when diffs have been +pre-computed by the comparator (e.g. for preview-then-apply flows). + +All DB writes are wrapped in a single `transaction.atomic()` block. Any exception causes a +full rollback. + +### Exceptions + +| Exception | Raised when | +|-----------|-------------| +| `DestructiveChangesError` | `REMOVE` operations are present and `allow_destructive=False` | +| `CircularDependencyError` | Cross-COT `related_object_type` references form a cycle among new COTs | +| `UnknownChoiceSetError` | A `choice_set` name cannot be resolved | +| `UnknownObjectTypeError` | A `related_object_type` string cannot be resolved | + +`DestructiveChangesError` is raised **before** the transaction opens, so the DB is never +touched. The other exceptions may be raised mid-transaction, triggering a full rollback. + +### Dependency ordering + +When a schema document contains multiple new COTs that reference each other via +`related_object_type: "custom-objects/"`, the executor performs a topological sort to +ensure referenced COT tables exist before any referencing field is added. Cycles among new +COTs raise `CircularDependencyError`. + +--- + +## Backfilling pre-existing fields + +Fields created before the portable schema feature was introduced have `schema_id = null`. +Migration `0007_backfill_schema_ids` assigns IDs to all such fields in PK order and updates +each COT's `next_schema_id` counter accordingly. This migration runs automatically with +`manage.py migrate`. + +After the backfill, all existing fields participate in export and diff cycles normally. From 95b8f6a5c19d729fe923abdd57de878fc2b7d9a2 Mon Sep 17 00:00:00 2001 From: Brian Tiemann Date: Wed, 22 Apr 2026 19:31:45 -0400 Subject: [PATCH 16/19] Ruff fix --- netbox_custom_objects/migrations/0007_portable_schema.py | 1 - 1 file changed, 1 deletion(-) diff --git a/netbox_custom_objects/migrations/0007_portable_schema.py b/netbox_custom_objects/migrations/0007_portable_schema.py index 6abbb305..a21788f8 100644 --- a/netbox_custom_objects/migrations/0007_portable_schema.py +++ b/netbox_custom_objects/migrations/0007_portable_schema.py @@ -1,5 +1,4 @@ import django.core.validators -import re from django.db import migrations, models From ed47029d8431ce50fc48ce81a5d046259884cbb7 Mon Sep 17 00:00:00 2001 From: Brian Tiemann Date: Thu, 23 Apr 2026 21:27:00 -0400 Subject: [PATCH 17/19] Make schema_id read-only in CustomObjectTypeFieldSerializer schema_id is system-assigned to ensure stable rename detection across COT schema versions. Adding read_only_fields = ('schema_id',) to the serializer Meta prevents POST/PATCH from overwriting it. Adds SchemaIdReadOnlyTest to verify: schema_id appears in responses, supplied values on POST are ignored, and PATCH attempts are silently discarded. Co-Authored-By: Claude Sonnet 4.6 --- netbox_custom_objects/api/serializers.py | 1 + netbox_custom_objects/tests/test_api.py | 75 +++++++++++++++++++++++- 2 files changed, 75 insertions(+), 1 deletion(-) diff --git a/netbox_custom_objects/api/serializers.py b/netbox_custom_objects/api/serializers.py index 52ce5847..46131d4d 100644 --- a/netbox_custom_objects/api/serializers.py +++ b/netbox_custom_objects/api/serializers.py @@ -83,6 +83,7 @@ class Meta: "deprecated_since", "scheduled_removal", ) + read_only_fields = ("schema_id",) def validate(self, attrs): app_label = attrs.pop("app_label", None) diff --git a/netbox_custom_objects/tests/test_api.py b/netbox_custom_objects/tests/test_api.py index 32ec7167..03076a79 100644 --- a/netbox_custom_objects/tests/test_api.py +++ b/netbox_custom_objects/tests/test_api.py @@ -7,7 +7,7 @@ from utilities.testing import APIViewTestCases, create_test_user from rest_framework import status -from netbox_custom_objects.models import CustomObjectType +from netbox_custom_objects.models import CustomObjectType, CustomObjectTypeField from .base import CustomObjectsTestCase from core.models import ObjectType from dcim.models import Device, DeviceRole, DeviceType, Manufacturer, Rack, Site @@ -959,3 +959,76 @@ def test_multiple_context_fields_omits_empty_values(self): self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertIsNotNone(response.data['_context']) self.assertEqual(response.data['_context']['display'], 'Dave') + + +class SchemaIdReadOnlyTest(CustomObjectsTestCase, TestCase): + """ + schema_id on CustomObjectTypeField is read-only via the API. + POSTing a value for it must be silently ignored; PATCHing an existing + field with a new schema_id must also leave the stored value unchanged. + """ + + def setUp(self): + self.user = create_test_user('schemauser') + token_key = create_token(self.user) + self.header = {'HTTP_AUTHORIZATION': f'Token {token_key}'} + + # Add add + change + view permissions on CustomObjectTypeField + perm = ObjectPermission(name='Schema perm', actions=['add', 'change', 'view']) + perm.save() + perm.users.add(self.user) + perm.object_types.add(ObjectType.objects.get_for_model(CustomObjectTypeField)) + # Also need add on CustomObjectType (for creating the parent) + cot_perm = ObjectPermission(name='COT perm', actions=['add', 'view']) + cot_perm.save() + cot_perm.users.add(self.user) + cot_perm.object_types.add(ObjectType.objects.get_for_model(CustomObjectType)) + + self.cot = self.create_custom_object_type(name='ro_schema', slug='ro-schema') + + def _field_list_url(self): + return reverse('plugins-api:netbox_custom_objects-api:customobjecttypefield-list') + + def _field_detail_url(self, pk): + return reverse( + 'plugins-api:netbox_custom_objects-api:customobjecttypefield-detail', + kwargs={'pk': pk}, + ) + + def test_schema_id_in_response(self): + """schema_id must be present and non-null in the API response.""" + field = self.create_custom_object_type_field(self.cot, name='alpha', type='text') + response = self.client.get(self._field_detail_url(field.pk), **self.header) + self.assertEqual(response.status_code, status.HTTP_200_OK) + self.assertIn('schema_id', response.data) + self.assertIsNotNone(response.data['schema_id']) + + def test_schema_id_ignored_on_create(self): + """Supplying schema_id on POST must be silently ignored; auto-assignment wins.""" + data = { + 'custom_object_type': self.cot.pk, + 'name': 'beta', + 'type': 'text', + 'schema_id': 999, + } + response = self.client.post( + self._field_list_url(), data, format='json', **self.header + ) + self.assertEqual(response.status_code, status.HTTP_201_CREATED) + self.assertNotEqual(response.data['schema_id'], 999) + + def test_schema_id_ignored_on_patch(self): + """PATCHing schema_id must not change the stored value.""" + field = self.create_custom_object_type_field(self.cot, name='gamma', type='text') + original_id = field.schema_id + + response = self.client.patch( + self._field_detail_url(field.pk), + {'schema_id': original_id + 100}, + format='json', + **self.header, + ) + self.assertEqual(response.status_code, status.HTTP_200_OK) + + field.refresh_from_db() + self.assertEqual(field.schema_id, original_id) From 0f08ab73dc4e0e5251a73cc67a2c080c192964db Mon Sep 17 00:00:00 2001 From: Brian Tiemann Date: Thu, 23 Apr 2026 21:38:47 -0400 Subject: [PATCH 18/19] Fix PATCH content type in SchemaIdReadOnlyTest Django's built-in test Client ignores format='json' (that's a DRF APIClient feature); PATCH with form-encoded body gets rejected with 415. Use json.dumps + content_type='application/json' instead. Co-Authored-By: Claude Sonnet 4.6 --- netbox_custom_objects/tests/test_api.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/netbox_custom_objects/tests/test_api.py b/netbox_custom_objects/tests/test_api.py index 03076a79..bfd824c1 100644 --- a/netbox_custom_objects/tests/test_api.py +++ b/netbox_custom_objects/tests/test_api.py @@ -1019,13 +1019,14 @@ 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 response = self.client.patch( self._field_detail_url(field.pk), - {'schema_id': original_id + 100}, - format='json', + json.dumps({'schema_id': original_id + 100}), + content_type='application/json', **self.header, ) self.assertEqual(response.status_code, status.HTTP_200_OK) From 3bd69c41f781d089e3638b55e18812a0e819d308 Mon Sep 17 00:00:00 2001 From: Brian Tiemann Date: Thu, 23 Apr 2026 21:49:59 -0400 Subject: [PATCH 19/19] Fix KeyError in CustomObjectTypeFieldSerializer.validate for partial PATCH attrs['type'] raises KeyError on any PATCH that omits the type field. Use attrs.get('type') so the object/select-type checks are simply skipped when type is not being updated, which is the correct behaviour for a partial update. This was exposed by the new SchemaIdReadOnlyTest.test_schema_id_ignored_on_patch test, which sends a PATCH body containing only schema_id. Co-Authored-By: Claude Sonnet 4.6 --- netbox_custom_objects/api/serializers.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/netbox_custom_objects/api/serializers.py b/netbox_custom_objects/api/serializers.py index 46131d4d..132fde4a 100644 --- a/netbox_custom_objects/api/serializers.py +++ b/netbox_custom_objects/api/serializers.py @@ -88,7 +88,8 @@ class Meta: def validate(self, attrs): app_label = attrs.pop("app_label", None) model = attrs.pop("model", None) - if attrs["type"] in [ + field_type = attrs.get("type") + if field_type in [ CustomFieldTypeChoices.TYPE_OBJECT, CustomFieldTypeChoices.TYPE_MULTIOBJECT, ]: @@ -114,7 +115,7 @@ def validate(self, attrs): raise ValidationError( "Must provide valid app_label and model for object field type." ) - if attrs["type"] in [ + if field_type in [ CustomFieldTypeChoices.TYPE_SELECT, CustomFieldTypeChoices.TYPE_MULTISELECT, ]: