diff --git a/netbox_custom_objects/comparator.py b/netbox_custom_objects/comparator.py new file mode 100644 index 0000000..1c36cf3 --- /dev/null +++ b/netbox_custom_objects/comparator.py @@ -0,0 +1,421 @@ +""" +COT state comparator / diff engine (issue #387). + +Compares an incoming schema document (produced by the exporter or hand-authored) +against the live DB state of each referenced CustomObjectType and returns a +structured diff that the upgrade executor (#389) can consume. + +Public API +---------- + diff_document(schema_doc) → list[COTDiff] + diff_cot(type_def) → COTDiff + +Data model +---------- + COTDiff — top-level result for one COT + .cot_changes — {attr: (db_val, schema_val)} for COT-level attribute changes + .field_changes — list[FieldChange] + .warnings — non-fatal issues (e.g. untracked DB fields) + + FieldChange + .op — FieldOp.ADD | REMOVE | ALTER + .schema_id — the stable numeric field identifier + .db_name — current DB field name (None for ADD) + .schema_def — raw schema field dict (for ADD; also available for ALTER) + .changed_attrs — {attr: (db_val, schema_val)} — populated for ALTER + +Notes +----- +- Fields are matched by schema_id. Fields in the DB without a schema_id cannot + be tracked and are reported as warnings, not as removals. +- A REMOVE operation is only emitted when the field's schema_id appears in the + schema's removed_fields tombstone list. A field absent from both schema.fields + and schema.removed_fields is ambiguous (possibly added outside the workflow) + and generates a warning instead. +- Type changes are included in changed_attrs but are not validated here; the + executor decides whether to allow or reject them. When the type changes, + type-specific attributes from the *old* DB type (e.g. validation_regex on a + text field being converted to integer) are not included in the diff — only + attributes relevant to the incoming schema type are compared. +- related_object_type values are compared in their encoded schema form + ("app_label/model" or "custom-objects/") so the diff output is + round-trip compatible with the schema format. +""" + +import re +from dataclasses import dataclass, field +from enum import Enum +from typing import TYPE_CHECKING + +from netbox_custom_objects import constants + +if TYPE_CHECKING: + from django.contrib.contenttypes.models import ContentType +from netbox_custom_objects.schema_format import ( + CUSTOM_OBJECTS_APP_LABEL_SLUG, + FIELD_DEFAULTS, + FIELD_TYPE_ATTRS, + SCHEMA_TYPE_TO_CHOICES, +) + +# Matches TableModel (generated model names for custom object types). +_TABLE_MODEL_RE = re.compile(r'^table(\d+)model$') + +# Ordered base attributes compared between DB and schema for each field. +# Does NOT include 'name' or 'type' — those are handled separately. +_FIELD_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", +) + +# COT-level attributes that may change between schema versions. +# Each maps to its schema-absent default (empty string). +_COT_ATTRS = ( + "name", + "version", + "verbose_name", + "verbose_name_plural", + "description", + "group_name", +) + + +# --------------------------------------------------------------------------- +# Data structures +# --------------------------------------------------------------------------- + +class FieldOp(Enum): + ADD = "add" # field exists in schema but not in DB + REMOVE = "remove" # field tombstoned in schema; still exists in DB + ALTER = "alter" # field in both; has differences (may include rename/type change) + + +@dataclass +class FieldChange: + """A single field-level operation within a COTDiff.""" + op: FieldOp + schema_id: int + db_name: str | None # current DB name; None for ADD + schema_def: dict # the schema field dict + changed_attrs: dict[str, tuple] = field(default_factory=dict) + # {attr: (db_value, schema_value)} — populated for ALTER; + # includes "name" if renamed, "type" if type differs. + + @property + def is_rename(self) -> bool: + return "name" in self.changed_attrs + + @property + def is_type_change(self) -> bool: + return "type" in self.changed_attrs + + +@dataclass +class COTDiff: + """All changes needed to bring one COT in sync with a schema definition.""" + name: str # from schema + slug: str # from schema (used as lookup key) + is_new: bool # True → COT does not yet exist in DB + cot_changes: dict[str, tuple] = field(default_factory=dict) + # {attr: (db_val, schema_val)} for COT-level attribute differences + field_changes: list[FieldChange] = field(default_factory=list) + warnings: list[str] = field(default_factory=list) + + @property + def has_changes(self) -> bool: + """True if there are attribute-level or field-level changes to apply. + + Note: a brand-new COT with no fields yields ``is_new=True`` but + ``has_changes=False`` — there are no individual changes to apply, but + the COT itself must still be created. Callers should check ``is_new`` + independently when deciding whether to run a create operation. + """ + return bool(self.cot_changes or self.field_changes) + + @property + def has_destructive_changes(self) -> bool: + """True if any field will be dropped from the DB.""" + return any(fc.op is FieldOp.REMOVE for fc in self.field_changes) + + @property + def adds(self) -> list[FieldChange]: + return [fc for fc in self.field_changes if fc.op is FieldOp.ADD] + + @property + def removes(self) -> list[FieldChange]: + return [fc for fc in self.field_changes if fc.op is FieldOp.REMOVE] + + @property + def alters(self) -> list[FieldChange]: + return [fc for fc in self.field_changes if fc.op is FieldOp.ALTER] + + +# --------------------------------------------------------------------------- +# Internal helpers +# --------------------------------------------------------------------------- + +def _encode_related_object_type(rot: "ContentType", cot_slug_cache: dict, warnings: list) -> str: + """ + Encode a related ObjectType FK as a schema ``related_object_type`` string. + Shares the same encoding logic as ``exporter._encode_related_object_type`` + but uses a pre-fetched slug cache and a warnings list rather than a live + DB query, making it safe to call in a tight loop. + + *cot_slug_cache* is a ``{pk: slug}`` dict pre-fetched in :func:`diff_cot` + to avoid one DB query per object field. If a custom COT is referenced but + not present in the cache (i.e. its DB row was deleted), a warning is + appended to *warnings* and a stable fallback string is returned so the + diff can still proceed. + """ + if rot.app_label == constants.APP_LABEL: + m = _TABLE_MODEL_RE.match(rot.model) + if m: + cot_id = int(m.group(1)) + slug = cot_slug_cache.get(cot_id) + if slug is None: + warnings.append( + f"CustomObjectType pk={cot_id} is referenced by a field but " + "no longer exists in the DB. The related_object_type comparison " + "for this field may be inaccurate." + ) + return f"{CUSTOM_OBJECTS_APP_LABEL_SLUG}/" + return f"{CUSTOM_OBJECTS_APP_LABEL_SLUG}/{slug}" + return f"{rot.app_label}/{rot.model}" + + +def _compare_field_attrs(db_field, schema_field: dict, cot_slug_cache: dict, warnings: list) -> dict[str, tuple]: + """ + Return ``{attr: (db_value, schema_value)}`` for every attribute that + differs between *db_field* (a ``CustomObjectTypeField`` instance) and + *schema_field* (the schema dict for that field). + """ + changes: dict[str, tuple] = {} + schema_type = schema_field["type"] # schema string, e.g. "text" + + # ── name ──────────────────────────────────────────────────────────────── + schema_name = schema_field["name"] + if db_field.name != schema_name: + changes["name"] = (db_field.name, schema_name) + + # ── type ───────────────────────────────────────────────────────────────── + expected_choice = SCHEMA_TYPE_TO_CHOICES[schema_type] + if db_field.type != expected_choice: + changes["type"] = (db_field.type, expected_choice) + + # ── base scalar attributes ─────────────────────────────────────────────── + for attr in _FIELD_BASE_ATTRS: + db_val = getattr(db_field, attr) + schema_val = schema_field.get(attr, FIELD_DEFAULTS.get(attr)) + if db_val != schema_val: + changes[attr] = (db_val, schema_val) + + # ── type-specific attributes ───────────────────────────────────────────── + type_specific = FIELD_TYPE_ATTRS.get(schema_type, set()) + + if "validation_regex" in type_specific: + dv = db_field.validation_regex or "" + sv = schema_field.get("validation_regex", "") + if dv != sv: + changes["validation_regex"] = (dv, sv) + + if "validation_minimum" in type_specific: + dv = db_field.validation_minimum + sv = schema_field.get("validation_minimum", None) + if dv != sv: + changes["validation_minimum"] = (dv, sv) + + if "validation_maximum" in type_specific: + dv = db_field.validation_maximum + sv = schema_field.get("validation_maximum", None) + if dv != sv: + changes["validation_maximum"] = (dv, sv) + + if "choice_set" in type_specific: + dv = db_field.choice_set.name if db_field.choice_set_id else None + sv = schema_field.get("choice_set") + if dv != sv: + changes["choice_set"] = (dv, sv) + + if "related_object_type" in type_specific: + dv = ( + _encode_related_object_type(db_field.related_object_type, cot_slug_cache, warnings) + ) if db_field.related_object_type_id else None + sv = schema_field.get("related_object_type") + if dv != sv: + changes["related_object_type"] = (dv, sv) + + if "related_object_filter" in type_specific: + dv = db_field.related_object_filter + sv = schema_field.get("related_object_filter", None) + if dv != sv: + changes["related_object_filter"] = (dv, sv) + + return changes + + +def _compare_cot_attrs(cot, type_def: dict) -> dict[str, tuple]: + """ + Return ``{attr: (db_value, schema_value)}`` for COT-level attributes + that differ. Absent schema keys are treated as empty string (same + convention as the exporter). + """ + changes: dict[str, tuple] = {} + for attr in _COT_ATTRS: + # All _COT_ATTRS are string fields; None and "" both mean "absent" — same + # convention as the exporter. _COT_ATTRS must never include numeric or + # boolean fields, as `or ""` would swallow falsy values like 0 or False. + db_val = getattr(cot, attr) or "" + schema_val = type_def.get(attr) or "" + if db_val != schema_val: + changes[attr] = (db_val, schema_val) + return changes + + +# --------------------------------------------------------------------------- +# Public API +# --------------------------------------------------------------------------- + +def diff_cot(type_def: dict) -> COTDiff: + """ + Compare one COT schema definition against the live DB state. + + *type_def* is a single entry from the ``types`` list of a schema document + (as produced by :func:`export_cot` or hand-authored). + + Returns a :class:`COTDiff` describing what would need to change. + """ + missing = [k for k in ("slug", "name") if k not in type_def] + if missing: + raise ValueError( + f"type_def is missing required key(s) {missing}; got keys: {list(type_def)}" + ) + + from netbox_custom_objects.models import CustomObjectType # noqa: PLC0415 + + slug = type_def["slug"] + name = type_def["name"] + + # ── COT existence check ────────────────────────────────────────────────── + try: + cot = CustomObjectType.objects.get(slug=slug) + except CustomObjectType.DoesNotExist: + # Brand-new COT — every schema field is an ADD. + field_changes = [ + FieldChange( + op=FieldOp.ADD, + schema_id=sf["id"], + db_name=None, + schema_def=sf, + ) + for sf in type_def.get("fields", []) + ] + return COTDiff( + name=name, + slug=slug, + is_new=True, + field_changes=field_changes, + ) + + diff = COTDiff(name=name, slug=slug, is_new=False) + + # ── COT-level attribute diff ───────────────────────────────────────────── + diff.cot_changes = _compare_cot_attrs(cot, type_def) + + # ── Build lookup indexes ───────────────────────────────────────────────── + schema_fields: dict[int, dict] = { + sf["id"]: sf for sf in type_def.get("fields", []) + } + # Map schema_id → full tombstone dict so REMOVE FieldChanges carry the + # original field definition (name, type, etc.) rather than an empty dict. + tombstoned: dict[int, dict] = { + rf["id"]: rf for rf in type_def.get("removed_fields", []) + } + # Single query for all fields; partition into tracked/untracked in Python. + db_fields: dict[int, object] = {} + for f in cot.fields.select_related("choice_set", "related_object_type"): + if f.schema_id is None: + diff.warnings.append( + f"Field {f.name!r} (pk={f.pk}) has no schema_id and cannot be " + "tracked by the schema diff. It will not be affected by apply operations." + ) + else: + db_fields[f.schema_id] = f + + # Pre-fetch slugs for all custom-COT related_object_type references in a + # single query to avoid one DB round-trip per object/multiobject field. + cot_ids: set[int] = set() + for f in db_fields.values(): + if f.related_object_type_id and f.related_object_type.app_label == constants.APP_LABEL: + m = _TABLE_MODEL_RE.match(f.related_object_type.model) + if m: + cot_ids.add(int(m.group(1))) + cot_slug_cache: dict[int, str] = ( + dict(CustomObjectType.objects.filter(pk__in=cot_ids).values_list("pk", "slug")) + if cot_ids else {} + ) + + # ── Schema fields → ADD or ALTER ───────────────────────────────────────── + for schema_id, schema_field in schema_fields.items(): + if schema_id in db_fields: + db_field = db_fields[schema_id] + changed = _compare_field_attrs(db_field, schema_field, cot_slug_cache, diff.warnings) + if changed: + diff.field_changes.append(FieldChange( + op=FieldOp.ALTER, + schema_id=schema_id, + db_name=db_field.name, + schema_def=schema_field, + changed_attrs=changed, + )) + else: + diff.field_changes.append(FieldChange( + op=FieldOp.ADD, + schema_id=schema_id, + db_name=None, + schema_def=schema_field, + )) + + # ── DB fields absent from schema → REMOVE or warn ──────────────────────── + for schema_id, db_field in db_fields.items(): + if schema_id in schema_fields: + continue # already handled above + if schema_id in tombstoned: + diff.field_changes.append(FieldChange( + op=FieldOp.REMOVE, + schema_id=schema_id, + db_name=db_field.name, + schema_def=tombstoned[schema_id], + )) + else: + diff.warnings.append( + f"Field {db_field.name!r} (schema_id={schema_id}) exists in the DB " + "but is absent from both schema.fields and schema.removed_fields. " + "It was likely added outside the schema workflow and will not be " + "affected by apply operations." + ) + + return diff + + +def diff_document(schema_doc: dict) -> list[COTDiff]: + """ + Diff all COTs in a schema document against current DB state. + + Returns a list of :class:`COTDiff` objects, one per entry in + ``schema_doc["types"]``. + """ + return [diff_cot(type_def) for type_def in schema_doc.get("types", [])] diff --git a/netbox_custom_objects/tests/test_comparator.py b/netbox_custom_objects/tests/test_comparator.py new file mode 100644 index 0000000..fcb3de0 --- /dev/null +++ b/netbox_custom_objects/tests/test_comparator.py @@ -0,0 +1,531 @@ +""" +Tests for the COT state comparator / diff engine (issue #387). + +Covers: +- No-change (clean) diff +- New COT (not in DB) +- COT-level attribute changes +- Field ADD (in schema, not in DB) +- Field REMOVE (tombstoned; still in DB) +- Field ALTER: rename, type change, scalar attribute changes +- Field ALTER: choice_set change +- Field ALTER: related_object_type change (built-in and custom) +- Field ALTER: related_object_filter change +- Untracked fields (no schema_id) → warning, not REMOVE +- DB field absent from schema AND not tombstoned → warning, not REMOVE +- _encode_related_object_type deleted-COT path → warning + stable fallback string +- Multi-COT document (including empty/missing types key) +- has_changes / has_destructive_changes / adds / removes / alters helpers +""" + +from django.test import TestCase + +from netbox_custom_objects.comparator import ( + FieldOp, + _encode_related_object_type, + diff_cot, + diff_document, +) +from netbox_custom_objects.exporter import export_cot, export_cots +from netbox_custom_objects.models import CustomObjectTypeField + +from .base import CustomObjectsTestCase + + +# --------------------------------------------------------------------------- +# Test cases +# --------------------------------------------------------------------------- + +class ComparatorCleanDiffTestCase(CustomObjectsTestCase, TestCase): + """Round-trip: export → diff should produce no changes.""" + + @classmethod + def setUpTestData(cls): + cls.cot = cls.create_custom_object_type( + name='clean', slug='clean', version='1.0.0', + ) + cls.choice_set = cls.create_choice_set(name='Clean Choices') + cls.device_ot = cls.get_device_object_type() + + cls.create_custom_object_type_field(cls.cot, name='label', type='text') + cls.create_custom_object_type_field( + cls.cot, name='count', type='integer', + validation_minimum=0, validation_maximum=100, + ) + 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_clean_diff_has_no_changes(self): + type_def = export_cot(self.cot) + result = diff_cot(type_def) + self.assertFalse(result.has_changes) + self.assertEqual(result.field_changes, []) + self.assertEqual(result.cot_changes, {}) + + def test_clean_diff_document_has_no_changes(self): + doc = export_cots([self.cot]) + diffs = diff_document(doc) + self.assertEqual(len(diffs), 1) + self.assertFalse(diffs[0].has_changes) + + def test_no_warnings_for_fully_tracked_cot(self): + type_def = export_cot(self.cot) + result = diff_cot(type_def) + self.assertEqual(result.warnings, []) + + +class ComparatorNewCOTTestCase(CustomObjectsTestCase, TestCase): + """COT not yet in DB → is_new=True, all fields are ADD.""" + + def test_new_cot_is_new(self): + result = diff_cot({ + "name": "newtype", + "slug": "newtype", + "fields": [ + {"id": 1, "name": "label", "type": "text"}, + {"id": 2, "name": "count", "type": "integer"}, + ], + }) + self.assertTrue(result.is_new) + + def test_new_cot_all_fields_are_add(self): + result = diff_cot({ + "name": "newtype2", + "slug": "newtype2", + "fields": [ + {"id": 1, "name": "label", "type": "text"}, + {"id": 2, "name": "count", "type": "integer"}, + ], + }) + self.assertEqual(len(result.field_changes), 2) + for fc in result.field_changes: + self.assertIs(fc.op, FieldOp.ADD) + + def test_new_cot_no_fields_no_field_changes(self): + result = diff_cot({"name": "empty", "slug": "empty"}) + self.assertTrue(result.is_new) + self.assertEqual(result.field_changes, []) + + def test_missing_required_keys_raises_value_error(self): + with self.assertRaises(ValueError): + diff_cot({"name": "no-slug"}) + with self.assertRaises(ValueError): + diff_cot({"slug": "no-name"}) + with self.assertRaises(ValueError): + diff_cot({}) + + +class ComparatorCOTAttrsTestCase(CustomObjectsTestCase, TestCase): + """COT-level attribute changes are captured in cot_changes.""" + + @classmethod + def setUpTestData(cls): + cls.cot = cls.create_custom_object_type( + name='attrtest', slug='attr-test', + version='1.0.0', description='old desc', + verbose_name='', verbose_name_plural='', + ) + + def test_version_change_detected(self): + type_def = export_cot(self.cot) + type_def["version"] = "2.0.0" + result = diff_cot(type_def) + self.assertIn("version", result.cot_changes) + self.assertEqual(result.cot_changes["version"], ("1.0.0", "2.0.0")) + + def test_description_change_detected(self): + type_def = export_cot(self.cot) + type_def["description"] = "new desc" + result = diff_cot(type_def) + self.assertIn("description", result.cot_changes) + + def test_unchanged_cot_attrs_not_in_cot_changes(self): + type_def = export_cot(self.cot) + result = diff_cot(type_def) + self.assertNotIn("version", result.cot_changes) + self.assertNotIn("description", result.cot_changes) + + def test_cot_changes_do_not_produce_field_changes(self): + type_def = export_cot(self.cot) + type_def["version"] = "9.9.9" + result = diff_cot(type_def) + self.assertEqual(result.field_changes, []) + + +class ComparatorFieldAddTestCase(CustomObjectsTestCase, TestCase): + """Fields present in schema but absent from DB → ADD.""" + + @classmethod + def setUpTestData(cls): + cls.cot = cls.create_custom_object_type(name='addtest', slug='add-test') + cls.create_custom_object_type_field(cls.cot, name='existing', type='text') + + def test_new_field_in_schema_is_add(self): + type_def = export_cot(self.cot) + # Append a field with a new schema_id that doesn't exist in DB + type_def.setdefault("fields", []).append( + {"id": 999, "name": "brand_new", "type": "boolean"} + ) + result = diff_cot(type_def) + adds = result.adds + self.assertEqual(len(adds), 1) + self.assertEqual(adds[0].schema_id, 999) + self.assertIsNone(adds[0].db_name) + self.assertEqual(adds[0].schema_def["name"], "brand_new") + + def test_add_does_not_affect_existing_fields(self): + type_def = export_cot(self.cot) + type_def.setdefault("fields", []).append( + {"id": 999, "name": "extra", "type": "text"} + ) + result = diff_cot(type_def) + # existing field should not appear in changes + self.assertEqual(len(result.alters), 0) + + +class ComparatorFieldRemoveTestCase(CustomObjectsTestCase, TestCase): + """Fields tombstoned in schema and still in DB → REMOVE.""" + + @classmethod + def setUpTestData(cls): + cls.cot = cls.create_custom_object_type(name='removetest', slug='remove-test') + cls.field = cls.create_custom_object_type_field( + cls.cot, name='to_remove', type='text' + ) + + def test_tombstoned_db_field_is_remove(self): + field_schema_id = self.field.schema_id + type_def = { + "name": self.cot.name, + "slug": self.cot.slug, + "fields": [], + "removed_fields": [ + {"id": field_schema_id, "name": "to_remove", "type": "text"} + ], + } + result = diff_cot(type_def) + removes = result.removes + self.assertEqual(len(removes), 1) + self.assertEqual(removes[0].schema_id, field_schema_id) + self.assertEqual(removes[0].db_name, "to_remove") + + def test_remove_is_flagged_as_destructive(self): + field_schema_id = self.field.schema_id + type_def = { + "name": self.cot.name, + "slug": self.cot.slug, + "fields": [], + "removed_fields": [ + {"id": field_schema_id, "name": "to_remove", "type": "text"} + ], + } + result = diff_cot(type_def) + self.assertTrue(result.has_destructive_changes) + + def test_already_removed_from_db_is_noop(self): + """If tombstoned field is already gone from DB, no REMOVE emitted.""" + type_def = { + "name": self.cot.name, + "slug": self.cot.slug, + "fields": [], + "removed_fields": [ + {"id": 9999, "name": "ghost", "type": "text"} # not in DB + ], + } + result = diff_cot(type_def) + self.assertEqual(result.removes, []) + self.assertFalse(result.has_destructive_changes) + + +class ComparatorFieldAlterTestCase(CustomObjectsTestCase, TestCase): + """Attribute changes on existing fields → ALTER.""" + + @classmethod + def setUpTestData(cls): + cls.cot = cls.create_custom_object_type(name='altertest', slug='alter-test') + cls.choice_set_a = cls.create_choice_set(name='Set A') + cls.choice_set_b = cls.create_choice_set(name='Set B') + cls.device_ot = cls.get_device_object_type() + cls.site_ot = cls.get_site_object_type() + + def _alter_field(self, field_name, **overrides): + """Export the COT, override one field's schema dict, return diff.""" + type_def = export_cot(self.cot) + for sf in type_def.get("fields", []): + if sf["name"] == field_name: + sf.update(overrides) + break + else: + raise AssertionError(f"field {field_name!r} not found in exported schema") + return diff_cot(type_def) + + def test_rename_detected(self): + self.create_custom_object_type_field(self.cot, name='old_name', type='text') + result = self._alter_field("old_name", name="new_name") + fc = next(fc for fc in result.alters if fc.db_name == "old_name") + self.assertTrue(fc.is_rename) + self.assertEqual(fc.changed_attrs["name"], ("old_name", "new_name")) + + def test_type_change_detected(self): + self.create_custom_object_type_field(self.cot, name='typed', type='text') + result = self._alter_field("typed", type="longtext") + fc = next(fc for fc in result.alters if fc.db_name == "typed") + self.assertTrue(fc.is_type_change) + self.assertIn("type", fc.changed_attrs) + + def test_required_change_detected(self): + self.create_custom_object_type_field( + self.cot, name='req_field', type='text', required=False + ) + result = self._alter_field("req_field", required=True) + fc = next(fc for fc in result.alters if fc.db_name == "req_field") + self.assertIn("required", fc.changed_attrs) + self.assertEqual(fc.changed_attrs["required"], (False, True)) + + def test_weight_change_detected(self): + self.create_custom_object_type_field( + self.cot, name='weighted', type='text', weight=100 + ) + result = self._alter_field("weighted", weight=200) + fc = next(fc for fc in result.alters if fc.db_name == "weighted") + self.assertIn("weight", fc.changed_attrs) + + def test_validation_regex_change_detected(self): + self.create_custom_object_type_field( + self.cot, name='regex_field', type='text', + validation_regex=r'^[A-Z]+$' + ) + result = self._alter_field("regex_field", validation_regex=r'^\d+$') + fc = next(fc for fc in result.alters if fc.db_name == "regex_field") + self.assertIn("validation_regex", fc.changed_attrs) + + def test_validation_min_max_change_detected(self): + self.create_custom_object_type_field( + self.cot, name='numeric', type='integer', + validation_minimum=0, validation_maximum=100 + ) + result = self._alter_field("numeric", validation_minimum=10, validation_maximum=200) + fc = next(fc for fc in result.alters if fc.db_name == "numeric") + self.assertIn("validation_minimum", fc.changed_attrs) + self.assertIn("validation_maximum", fc.changed_attrs) + + def test_choice_set_change_detected(self): + self.create_custom_object_type_field( + self.cot, name='select_field', type='select', + choice_set=self.choice_set_a, + ) + result = self._alter_field("select_field", choice_set="Set B") + fc = next(fc for fc in result.alters if fc.db_name == "select_field") + self.assertIn("choice_set", fc.changed_attrs) + self.assertEqual(fc.changed_attrs["choice_set"], ("Set A", "Set B")) + + def test_related_object_type_change_detected(self): + self.create_custom_object_type_field( + self.cot, name='obj_field', type='object', + related_object_type=self.device_ot, + ) + result = self._alter_field("obj_field", related_object_type="dcim/site") + fc = next(fc for fc in result.alters if fc.db_name == "obj_field") + self.assertIn("related_object_type", fc.changed_attrs) + self.assertEqual(fc.changed_attrs["related_object_type"][0], "dcim/device") + self.assertEqual(fc.changed_attrs["related_object_type"][1], "dcim/site") + + def test_related_object_type_custom_cot_encoding(self): + other_cot = self.create_custom_object_type(name='rack', slug='rack') + rack_ot = other_cot.object_type + self.create_custom_object_type_field( + self.cot, name='rack_field', type='object', + related_object_type=rack_ot, + ) + # Export and re-diff — should be clean (no change) + type_def = export_cot(self.cot) + result = diff_cot(type_def) + rack_alters = [fc for fc in result.alters if fc.db_name == "rack_field"] + self.assertEqual(rack_alters, [], "Custom COT related_object_type round-trips cleanly") + + def test_related_object_filter_change_detected(self): + self.create_custom_object_type_field( + self.cot, name='filtered', type='object', + related_object_type=self.device_ot, + related_object_filter={"site_id": [1]}, + ) + result = self._alter_field("filtered", related_object_filter={"site_id": [2]}) + fc = next(fc for fc in result.alters if fc.db_name == "filtered") + self.assertIn("related_object_filter", fc.changed_attrs) + + def test_no_alter_when_nothing_changed(self): + self.create_custom_object_type_field( + self.cot, name='unchanged', type='text' + ) + type_def = export_cot(self.cot) + result = diff_cot(type_def) + unchanged_alters = [fc for fc in result.alters if fc.db_name == "unchanged"] + self.assertEqual(unchanged_alters, []) + + def test_deprecated_change_detected(self): + self.create_custom_object_type_field( + self.cot, name='dep_field', type='text' + ) + result = self._alter_field( + "dep_field", + deprecated=True, + deprecated_since="1.1.0", + scheduled_removal="2.0.0", + ) + fc = next(fc for fc in result.alters if fc.db_name == "dep_field") + self.assertIn("deprecated", fc.changed_attrs) + self.assertIn("deprecated_since", fc.changed_attrs) + self.assertIn("scheduled_removal", fc.changed_attrs) + + +class ComparatorWarningsTestCase(CustomObjectsTestCase, TestCase): + """Warning conditions: untracked fields, ambiguous absences.""" + + @classmethod + def setUpTestData(cls): + cls.cot = cls.create_custom_object_type(name='warntest', slug='warn-test') + + def test_untracked_field_emits_warning(self): + f = self.create_custom_object_type_field( + self.cot, name='untracked', type='text' + ) + CustomObjectTypeField.objects.filter(pk=f.pk).update(schema_id=None) + type_def = export_cot(self.cot) # won't include the untracked field + result = diff_cot(type_def) + self.assertTrue( + any("untracked" in w for w in result.warnings), + "Expected warning about field with no schema_id", + ) + + def test_untracked_field_not_in_field_changes(self): + f = self.create_custom_object_type_field( + self.cot, name='untracked2', type='text' + ) + CustomObjectTypeField.objects.filter(pk=f.pk).update(schema_id=None) + type_def = export_cot(self.cot) + result = diff_cot(type_def) + self.assertEqual(result.field_changes, []) + + def test_encode_related_object_type_deleted_cot_emits_warning(self): + """_encode_related_object_type emits a warning and returns a stable fallback string when + the referenced CustomObjectType no longer exists (slug not in cache). + This is the only warning path in comparator.py that can't be triggered + via diff_cot because CustomObjectType.delete() removes referencing fields + before deleting the ObjectType, making the state unreachable via normal + model deletion. + """ + target = self.create_custom_object_type(name='rot-target', slug='rot-target') + rot = target.object_type # ObjectType with app_label matching constants.APP_LABEL + warnings = [] + result = _encode_related_object_type(rot, cot_slug_cache={}, warnings=warnings) + self.assertEqual(len(warnings), 1) + self.assertIn("no longer exists", warnings[0]) + self.assertIn("