fix: reassign unique imported relations atomically

tests/test_importer_plugin_compat.py

@@ -1,10 +1,14 @@
 from typing import ClassVar
 
-from django.db import models
+import pytest
+from django.db import connection, models
 
 from netbox_export.services.importer import _field_kwargs
+from netbox_export.services.plugin_compat import PluginCompatibility
 from netbox_export.services.references import ReferenceResolver
 
+pytestmark = pytest.mark.django_db(transaction=True)
+
 
 class Tenant(models.Model):
     class Meta:
@@ -36,6 +40,18 @@ class ConstrainedInstallation(models.Model):
         ]
 
 
+class UniqueAddress(models.Model):
+    class Meta:
+        app_label = "compat_tests"
+
+
+class UniqueAddressOwner(models.Model):
+    primary_ip6 = models.OneToOneField(UniqueAddress, on_delete=models.SET_NULL, null=True)
+
+    class Meta:
+        app_label = "compat_tests"
+
+
 def tenant_record():
     return {
         "fields": {},
@@ -93,3 +109,58 @@ def test_unresolved_relation_in_check_constraint_blocks_initial_save():
     assert unresolved == []
     assert unresolved_values == []
     assert missing_required == []
+
+
+def test_resolved_nullable_unique_relation_is_deferred():
+    address = UniqueAddress(pk=50)
+    resolver = ReferenceResolver(
+        {"compat_tests.uniqueaddress:50": address},
+        [],
+        lambda label: None,
+    )
+    record = {
+        "fields": {},
+        "relations": {"primary_ip6": {"ref": "compat_tests.uniqueaddress:50"}},
+    }
+
+    kwargs, unresolved, unresolved_values, missing_required = _field_kwargs(
+        UniqueAddressOwner,
+        record,
+        resolver,
+        tenant_required=False,
+    )
+
+    assert kwargs == {}
+    assert unresolved == [("primary_ip6", {"ref": "compat_tests.uniqueaddress:50"})]
+    assert unresolved_values == []
+    assert missing_required == []
+
+
+def test_existing_unique_relation_is_released_before_reassignment():
+    with connection.schema_editor() as schema_editor:
+        schema_editor.create_model(UniqueAddress)
+        schema_editor.create_model(UniqueAddressOwner)
+    try:
+        address = UniqueAddress.objects.create()
+        previous_owner = UniqueAddressOwner.objects.create(primary_ip6=address)
+        imported_owner = UniqueAddressOwner()
+        warnings = []
+        compatibility = PluginCompatibility(warnings, dry_run=False, tenant_required=False)
+
+        compatibility.release_unique_relation(imported_owner, "primary_ip6", address)
+        imported_owner.primary_ip6 = address
+        imported_owner.save()
+        previous_owner.refresh_from_db()
+
+        assert previous_owner.primary_ip6 is None
+        assert imported_owner.primary_ip6 == address
+        assert warnings == [
+            (
+                "Eindeutige Referenz compat_tests.uniqueaddressowner.primary_ip6 wurde von "
+                f"Zielobjekt(en) {previous_owner.pk} gelöst und dem importierten Objekt neu zugeordnet."
+            )
+        ]
+    finally:
+        with connection.schema_editor() as schema_editor:
+            schema_editor.delete_model(UniqueAddressOwner)
+            schema_editor.delete_model(UniqueAddress)

tests/test_plugin_compat.py

@@ -9,6 +9,7 @@ from netbox_export.services.plugin_compat import (
     check_constraint_field_names,
     is_tenant_relation,
     relation_required_before_save,
+    relation_should_be_deferred,
 )
 
 
@@ -101,6 +102,34 @@ def test_identifies_relations_used_by_check_constraint():
     assert relation_required_before_save(field) is True
 
 
+def test_nullable_unique_relation_is_deferred_unless_required_by_constraint():
+    regular = relation_field("primary_ip6", "ipam.ipaddress")
+    regular.null = True
+    regular.unique = True
+    regular.model = type(
+        "VirtualMachine",
+        (),
+        {"_meta": SimpleNamespace(constraints=[])},
+    )
+
+    assert relation_should_be_deferred(regular) is True
+
+    constrained = relation_field("platform", "dcim.device")
+    constrained.null = True
+    constrained.unique = True
+    constraint = models.CheckConstraint(
+        condition=models.Q(platform__isnull=False),
+        name="platform_required",
+    )
+    constrained.model = type(
+        "Installation",
+        (),
+        {"_meta": SimpleNamespace(constraints=[constraint])},
+    )
+
+    assert relation_should_be_deferred(constrained) is False
+
+
 def test_creates_and_assigns_auto_import_tenant_when_policy_is_active():
     manager = FakeTenantManager()
     tenant_model = type("TenantModel", (), {"_default_manager": manager})