fix: preserve cable paths across export and import

README.md

@@ -92,8 +92,10 @@ Import werden vorhandene Objekte über ihre eindeutigen Fachschlüssel erkannt.
 ## Verhalten und Grenzen
 
 - Quelle und Ziel müssen NetBox 4.6.x und dieselben Plugins/Modelle verwenden.
-- Benutzerkonten und Berechtigungen werden nicht exportiert. Referenzen auf
+- Benutzerkonten und Berechtigungen werden nicht exportiert. Fehlende oder
-  Benutzer oder Gruppen müssen auf dem Ziel bereits eindeutig vorhanden sein.
+  nicht eindeutige Referenzen auf Benutzer und Gruppen werden ausgelassen und
+  nach dem Import als Warnung angezeigt. Benötigt ein neuer Datensatz zwingend
+  eine solche Referenz, wird nur dieser Datensatz übersprungen.
 - Der Import erstellt und aktualisiert Objekte. Zielobjekte, die im Archiv nicht
   vorkommen, werden bewusst nicht gelöscht.
 - Fehlende Bilddateien werden im Archiv vermerkt, können aber nicht rekonstruiert
@@ -105,6 +107,53 @@ Import werden vorhandene Objekte über ihre eindeutigen Fachschlüssel erkannt.
   Werte zwischen `250` und `1000` erlauben eine Anpassung an Arbeitsspeicher und
   Datenbankleistung.
 
+## Plugin-Kompatibilität
+
+Der Exportgraph berücksichtigt installierte, mandantenbezogene Modelle und
+Dateien aus NetBox-SLM, Netbox-DokiWiki und NetBox-VM-Import. Private oder
+temporäre Plugin-Modelle werden nicht exportiert. Die von Netbox-Utilities
+erzwungene Mandantenpflicht wird beim Import berücksichtigt: Das Zielobjekt wird
+erst gespeichert, nachdem sein Mandant importiert und zugeordnet wurde. Ist kein
+Mandant auflösbar, wird automatisch ein vorhandener Mandant `Auto-Import`
+verwendet oder neu angelegt. Der Importbericht weist darauf hin.
+Beziehungen, die Teil einer Plugin-Datenbankprüfung sind, werden vollständig
+aufgelöst, bevor das Objekt erstmals gespeichert wird. Dies betrifft unter
+anderem die Plattformzuordnung von NetBox-SLM-Softwareinstallationen.
+Eindeutige optionale Beziehungen wie die primären IP-Adressen von Geräten und
+virtuellen Maschinen werden in einer zweiten Phase zugewiesen. Eine veraltete
+Zielzuordnung wird dabei atomar gelöst und als Warnung protokolliert.
+Gespeicherte Importzuordnungen werden bei Wiederholungsimporten gegen den
+aktuellen Fachschlüssel geprüft. Existiert das Objekt bereits unter diesem
+Schlüssel, wird die Zuordnung korrigiert, statt ein Duplikat anzulegen.
+Bei neuen NetBox-Modulen wird die automatische Komponentenreplikation
+deaktiviert. Ports, Interfaces und Bays werden stattdessen ausschließlich aus
+den Archivdatensätzen angelegt beziehungsweise vorhandenen Komponenten
+zugeordnet.
+Geräte werden in einer separaten Abschlussphase im Rack platziert, damit auch
+Positionswechsel ohne temporäre Doppelbelegung funktionieren. Bei fremden
+Belegungen löst **Aktualisieren** das Zielgerät mit Warnung von seiner Position,
+**Überspringen** lässt das importierte Gerät positionslos und **Import abbrechen**
+meldet den Rackplatzkonflikt vor dem Datenbankfehler. Mehr-U- und
+Full-Depth-Belegungen werden dabei berücksichtigt.
+Front-/Rear-Port-Zuordnungen von Patchpanels werden als eigene Datensätze
+exportiert und bei **Aktualisieren** auf den Stand der Quelle gebracht. Nach dem
+Import stößt das Plugin für alle enthaltenen Kabel die NetBox-eigene
+Neuberechnung der Kabelpfade an. Für diese Korrektur muss mit Plugin-Version
+`0.3.12` oder neuer ein neues Archiv auf der Quellinstanz erzeugt werden, da
+ältere Archive keine Portzuordnungen enthalten.
+Bei Bildanhängen werden Breite und Höhe direkt aus der Bilddatei im Archiv
+ermittelt. Dadurch sind die Pflichtfelder von NetBox auch im Prüflauf und bei
+Dateispeichern ohne unmittelbaren Modell-Save gesetzt. Bilder oberhalb des in
+NetBox 4.6.x verwendeten Limits von 25 Millionen Pixeln werden proportional auf
+höchstens 20 Millionen Pixel verkleinert und im Importbericht als Warnung
+ausgewiesen. Zum Schutz des Importprozesses bleibt eine harte Quellgrenze von
+100 Millionen Pixeln bestehen.
+
+Auf Quelle und Ziel müssen jeweils dieselben Plugin-Versionen und Migrationen
+installiert sein. Verschlüsselte Zugangsdaten von NetBox-VM-Import sind nur bei
+identischem Django-`SECRET_KEY` direkt nutzbar; andernfalls muss das Kennwort am
+Ziel neu gesetzt werden.
+
 ## Tests
 
 ```bash

netbox_export/__init__.py

@@ -7,7 +7,7 @@ class NetBoxExportConfig(PluginConfig):
     name = "netbox_export"
     verbose_name = "NetBox-Export"
     description = "Portable ZIP export and import for tenants and locations"
-    version = "0.3.0"
+    version = "0.3.12"
     author = "NetBox Export contributors"
     base_url = "netbox-export"
     min_version = "4.6.0"

netbox_export/services/exporter.py

@@ -39,7 +39,7 @@ def export_scope(
         "created_at": datetime.now(UTC).isoformat(),
         "source_instance": str(InstanceIdentity.local_id()),
         "source_netbox_version": getattr(getattr(settings, "RELEASE", None), "version", "4.6"),
-        "plugin_version": "0.3.0",
+        "plugin_version": "0.3.12",
         "scope": {
             "type": scope_type,
             "source_pk": str(scope_id),

netbox_export/services/graph.py

@@ -31,6 +31,11 @@ EXCLUDED_MODELS = {
 }
 SCOPE_LINK_FIELDS = {"tenant", "site", "location", "region"}
 PEER_CONTAINER_MODELS = {"dcim.cable", "circuits.circuit", "circuits.virtualcircuit"}
+PRIVATE_EXPORTABLE_MODELS = {
+    # NetBox derives cable paths from these mappings, but marks the model private
+    # because it has no public API of its own.
+    "dcim.portmapping",
+}
 
 
 def is_exportable_model(model) -> bool:
@@ -40,6 +45,10 @@ def is_exportable_model(model) -> bool:
         and not opts.abstract
         and not opts.proxy
         and not opts.auto_created
+        and (
+            not getattr(model, "_netbox_private", False)
+            or opts.label_lower in PRIVATE_EXPORTABLE_MODELS
+        )
         and opts.app_label not in EXCLUDED_APP_LABELS
         and opts.label_lower not in EXCLUDED_MODELS
     )

netbox_export/services/importer.py

@@ -1,21 +1,34 @@
 from __future__ import annotations
 
+import io
 import logging
+import math
+import threading
 import uuid
+import warnings
 from dataclasses import dataclass
 from dataclasses import field as dataclass_field
+from decimal import Decimal
 
 from django.apps import apps
 from django.conf import settings
 from django.contrib.contenttypes.models import ContentType
 from django.core.files.base import ContentFile
 from django.db import IntegrityError, models, transaction
+from PIL import Image as PillowImage
 
 from netbox_export.models import ImportedObjectMapping
 
 from .archive import ParsedArchive
 from .codec import SKIP_FIELD_NAMES, decode_scalar, generic_foreign_keys
 from .exceptions import ArchiveValidationError, ExportImportError, ImportConflictError
+from .plugin_compat import (
+    PluginCompatibility,
+    is_tenant_relation,
+    relation_required_before_save,
+    relation_should_be_deferred,
+)
+from .references import MISSING_REFERENCE, ReferenceResolver
 
 EXPLICIT_IDENTITIES = {
     "tenancy.tenantgroup": ("slug",),
@@ -31,8 +44,11 @@ EXPLICIT_IDENTITIES = {
     "dcim.consoleserverport": ("device", "name"),
     "dcim.powerport": ("device", "name"),
     "dcim.poweroutlet": ("device", "name"),
+    "dcim.frontport": ("device", "name"),
+    "dcim.rearport": ("device", "name"),
+    "dcim.portmapping": ("front_port", "front_port_position"),
     "dcim.devicebay": ("device", "name"),
-    "dcim.modulebay": ("device", "name"),
+    "dcim.modulebay": ("device", "module", "name"),
     "dcim.inventoryitem": ("device", "parent", "name"),
     "ipam.prefix": ("vrf", "prefix"),
     "ipam.ipaddress": ("vrf", "address"),
@@ -42,6 +58,11 @@ EXPLICIT_IDENTITIES = {
     "virtualization.vminterface": ("virtual_machine", "name"),
 }
 
+NETBOX_IMAGE_MAX_PIXELS = 25_000_000
+IMPORTED_IMAGE_TARGET_PIXELS = 20_000_000
+IMPORTED_IMAGE_SOURCE_MAX_PIXELS = 100_000_000
+_IMAGE_LIMIT_LOCK = threading.Lock()
+
 
 class IdentityNotReady(Exception):
     pass
@@ -66,6 +87,13 @@ class ImportReport:
         counters[action] += 1
 
 
+@dataclass(frozen=True)
+class DeferredDevicePlacement:
+    rack_spec: dict | None
+    position: object
+    face: object
+
+
 def _model_for(label: str):
     try:
         model = apps.get_model(label)
@@ -76,48 +104,27 @@ def _model_for(label: str):
     return model
 
 
-def _external_object(spec: dict):
+def _decode_archived_value(encoded, resolver):
-    model = _model_for(spec["model"])
-    lookup = {key: decode_scalar(value) for key, value in spec.get("lookup", {}).items()}
-    try:
-        return model._default_manager.get(**lookup)
-    except model.DoesNotExist as exc:
-        raise ArchiveValidationError(
-            f"Externe Referenz fehlt: {spec['model']} mit {lookup}."
-        ) from exc
-    except model.MultipleObjectsReturned as exc:
-        raise ArchiveValidationError(
-            f"Externe Referenz ist nicht eindeutig: {spec['model']} mit {lookup}."
-        ) from exc
-
-
-def _resolve_reference(spec, resolved: dict[str, models.Model]):
-    if spec is None:
-        return None, True
-    if "ref" in spec:
-        return resolved.get(spec["ref"]), spec["ref"] in resolved
-    if "external" in spec:
-        return _external_object(spec["external"]), True
-    raise ArchiveValidationError("Eine Objektreferenz im Archiv ist ungültig.")
-
-
-def _decode_archived_value(encoded, resolved):
     if isinstance(encoded, dict) and encoded.get("$type") == "object_ref":
-        target, available = _resolve_reference(encoded.get("value"), resolved)
+        target, available = resolver.resolve(encoded.get("value"))
+        if target is MISSING_REFERENCE:
+            return None, True
         return (target.pk if available and target is not None else None), available
     if isinstance(encoded, dict) and encoded.get("$type") == "multiobject_ref":
         values = []
         for spec in encoded.get("value", []):
-            target, available = _resolve_reference(spec, resolved)
+            target, available = resolver.resolve(spec)
             if not available:
                 return None, False
+            if target is MISSING_REFERENCE:
+                continue
             values.append(target.pk)
         return values, True
     if isinstance(encoded, dict) and "$type" not in encoded:
         value = {}
         all_available = True
         for key, item in encoded.items():
-            decoded, available = _decode_archived_value(item, resolved)
+            decoded, available = _decode_archived_value(item, resolver)
             value[key] = decoded
             all_available &= available
         return value, all_available
@@ -138,7 +145,7 @@ def _identity_candidates(model):
             yield tuple(constraint.fields)
 
 
-def _identity_lookup(model, record, resolved):
+def _identity_lookup(model, record, resolver):
     scalar_values = record.get("fields", {})
     relation_values = record.get("relations", {})
     generic_values = record.get("generic_relations", {})
@@ -156,14 +163,20 @@ def _identity_lookup(model, record, resolved):
             if name in scalar_values:
                 value = decode_scalar(scalar_values[name])
             elif name in relation_values:
-                value, available = _resolve_reference(relation_values[name], resolved)
+                value, available = resolver.resolve(relation_values[name])
                 if not available:
                     raise IdentityNotReady
+                if value is MISSING_REFERENCE:
+                    usable = False
+                    break
             elif name in generic_storage:
                 value_type, spec = generic_storage[name]
-                target, available = _resolve_reference(spec, resolved)
+                target, available = resolver.resolve(spec)
                 if not available:
                     raise IdentityNotReady
+                if target is MISSING_REFERENCE:
+                    usable = False
+                    break
                 if target is None:
                     value = None
                 elif value_type == "content_type":
@@ -197,21 +210,175 @@ def _mapped_object(source_instance: uuid.UUID, record: dict, model):
         return None
 
 
-def _find_existing(source_instance, model, record, resolved):
+def _find_existing(source_instance, model, record, resolver):
     mapped = _mapped_object(source_instance, record, model)
-    if mapped is not None:
+    lookup = _identity_lookup(model, record, resolver)
-        return mapped
-    lookup = _identity_lookup(model, record, resolved)
     if not lookup:
-        return None
+        return mapped
     try:
-        return model._default_manager.get(**lookup)
+        natural = model._default_manager.get(**lookup)
     except model.DoesNotExist:
-        return None
+        return mapped
     except model.MultipleObjectsReturned as exc:
         raise ImportConflictError(
             f"Mehrere Zielobjekte passen auf {record['model']} mit {lookup}."
         ) from exc
+    if mapped is not None and mapped.pk != natural.pk:
+        resolver.warn(
+            ("mapping-rebound", record["id"]),
+            f"Gespeicherte Zuordnung für {record['id']} wurde von Ziel-ID {mapped.pk} auf "
+            f"Ziel-ID {natural.pk} korrigiert, da der Fachschlüssel {lookup} bereits existiert.",
+        )
+    return natural
+
+
+def _defer_device_placement(model, record):
+    if model._meta.label_lower != "dcim.device":
+        return record, None
+
+    fields = record.get("fields", {})
+    relations = record.get("relations", {})
+    if "rack" not in relations or "position" not in fields or "face" not in fields:
+        return record, None
+
+    prepared = dict(record)
+    prepared["fields"] = {
+        name: value for name, value in fields.items() if name not in ("position", "face")
+    }
+    prepared["relations"] = {name: value for name, value in relations.items() if name != "rack"}
+    placement = DeferredDevicePlacement(
+        rack_spec=relations["rack"],
+        position=decode_scalar(fields["position"]),
+        face=decode_scalar(fields["face"]),
+    )
+    return prepared, placement
+
+
+def _placement_can_be_applied(placement, resolver):
+    rack, available = resolver.resolve(placement.rack_spec)
+    return not (available and rack is MISSING_REFERENCE)
+
+
+def _stage_device_placement(device):
+    device.position = None
+    device.face = None
+
+
+def _device_footprint(device):
+    device_type = getattr(device, "device_type", None)
+    height = Decimal(str(getattr(device_type, "u_height", 1) or 0))
+    if height <= 0:
+        height = Decimal("0.5")
+    return height, bool(getattr(device_type, "is_full_depth", False))
+
+
+def _device_placement_conflicts(device, rack, position, face):
+    if rack is None or position is None:
+        return []
+
+    position = Decimal(str(position))
+    height, full_depth = _device_footprint(device)
+    end = position + height
+    candidates = (
+        type(device)
+        ._default_manager.select_for_update()
+        .select_related("device_type")
+        .filter(rack=rack, position__isnull=False)
+        .exclude(pk=device.pk)
+    )
+    conflicts = []
+    for candidate in candidates:
+        candidate_position = Decimal(str(candidate.position))
+        candidate_height, candidate_full_depth = _device_footprint(candidate)
+        faces_overlap = full_depth or candidate_full_depth or candidate.face == face
+        positions_overlap = position < candidate_position + candidate_height and candidate_position < end
+        if faces_overlap and positions_overlap:
+            conflicts.append(candidate)
+    return conflicts
+
+
+def _placement_target(rack, position, face):
+    rack_name = getattr(rack, "name", None) or str(rack.pk)
+    return f"Rack {rack_name}, Position {position}, Seite {face or '-'}"
+
+
+def _conflicting_devices(conflicts):
+    return ", ".join(
+        f"{device.pk} ({getattr(device, 'name', None) or 'ohne Namen'})" for device in conflicts
+    )
+
+
+def _save_device_placement(device, rack, position, face, compatibility):
+    device.rack = rack
+    device.position = position
+    device.face = face
+    update_fields = ["rack", "position", "face"]
+    rack_location = getattr(rack, "location", None)
+    if rack_location is not None:
+        device.location = rack_location
+        update_fields.append("location")
+    compatibility.save(device, update_fields=update_fields)
+
+
+def _apply_device_placements(
+    placements,
+    resolved,
+    resolver,
+    compatibility,
+    *,
+    conflict_strategy,
+):
+    for record_id, placement in placements:
+        rack, available = resolver.resolve(placement.rack_spec)
+        if not available:
+            raise ArchiveValidationError(f"Rack-Referenz für {record_id} konnte nicht aufgelöst werden.")
+        if rack is MISSING_REFERENCE:
+            continue
+
+        device = resolved[record_id]
+        position = placement.position
+        face = placement.face
+        if rack is None and (position is not None or face is not None):
+            resolver.warn(
+                ("device-placement-without-rack", record_id),
+                f"Rackplatz für {record_id} wurde ausgelassen, da kein Rack zugeordnet ist.",
+            )
+            position = None
+            face = None
+
+        if rack is not None and position is not None:
+            rack = type(rack)._default_manager.select_for_update().get(pk=rack.pk)
+
+        conflicts = _device_placement_conflicts(device, rack, position, face)
+        if not conflicts:
+            _save_device_placement(device, rack, position, face, compatibility)
+            continue
+
+        target = _placement_target(rack, position, face)
+        occupants = _conflicting_devices(conflicts)
+        if conflict_strategy == "fail":
+            raise ImportConflictError(
+                f"Rackplatzkonflikt für {record_id}: {target} ist durch Zielgerät(e) {occupants} belegt."
+            )
+        if conflict_strategy == "skip":
+            _save_device_placement(device, rack, None, None, compatibility)
+            resolver.warn(
+                ("device-placement-skipped", record_id),
+                f"Rackplatz für {record_id} wurde übersprungen: {target} bleibt durch "
+                f"Zielgerät(e) {occupants} belegt. Das importierte Gerät wurde ohne Position im Rack gespeichert.",
+            )
+            continue
+
+        for conflict in conflicts:
+            conflict.position = None
+            conflict.face = None
+            compatibility.save(conflict, update_fields=["position", "face"])
+        _save_device_placement(device, rack, position, face, compatibility)
+        resolver.warn(
+            ("device-placement-released", record_id),
+            f"Rackplatzkonflikt für {record_id} gelöst: Zielgerät(e) {occupants} wurden aus {target} "
+            "gelöst und ohne Position im Rack belassen.",
+        )
 
 
 def _write_mapping(source_instance, record, obj):
@@ -224,11 +391,66 @@ def _write_mapping(source_instance, record, obj):
     )
 
 
-def _field_kwargs(model, record, resolved):
+def _release_port_mapping_conflicts(obj, resolver, record_id):
+    if obj._meta.label_lower != "dcim.portmapping":
+        return
+
+    conflicts = type(obj)._default_manager.select_for_update().filter(
+        models.Q(
+            front_port_id=obj.front_port_id,
+            front_port_position=obj.front_port_position,
+        )
+        | models.Q(
+            rear_port_id=obj.rear_port_id,
+            rear_port_position=obj.rear_port_position,
+        )
+    )
+    if obj.pk is not None:
+        conflicts = conflicts.exclude(pk=obj.pk)
+    conflicts = list(conflicts)
+    if not conflicts:
+        return
+
+    conflict_ids = ", ".join(str(conflict.pk) for conflict in conflicts)
+    for conflict in conflicts:
+        conflict.delete()
+    resolver.warn(
+        ("port-mapping-replaced", record_id),
+        f"Portzuordnung(en) {conflict_ids} wurden durch {record_id} ersetzt, damit die "
+        "Front-/Rear-Port-Verkabelung der Quelle entspricht.",
+    )
+
+
+def _trace_paths_signal():
+    from dcim.models.cables import trace_paths
+
+    return trace_paths
+
+
+def _rebuild_imported_cable_paths(records, resolved):
+    cable_ids = {
+        resolved[record_id].cable_id
+        for record_id, record in records.items()
+        if record["model"] == "dcim.cabletermination" and record_id in resolved
+    }
+    if not cable_ids:
+        return
+
+    cable_model = _model_for("dcim.cable")
+    trace_paths = _trace_paths_signal()
+    for cable in cable_model._default_manager.filter(pk__in=cable_ids).order_by("pk"):
+        # CableTermination records are imported directly, so Cable.save() never
+        # emits NetBox's normal path-rebuild signal for their new endpoints.
+        cable._terminations_modified = True
+        trace_paths.send(cable_model, instance=cable, created=False)
+
+
+def _field_kwargs(model, record, resolver, *, tenant_required: bool):
     valid_fields = {field.name: field for field in model._meta.concrete_fields}
     kwargs = {}
     unresolved = []
     unresolved_values = []
+    missing_required = []
     for name, encoded in record.get("fields", {}).items():
         field = valid_fields.get(name)
         if (
@@ -241,63 +463,181 @@ def _field_kwargs(model, record, resolved):
             continue
         if isinstance(field, (models.ForeignKey, models.OneToOneField)):
             continue
-        value, available = _decode_archived_value(encoded, resolved)
+        value, available = _decode_archived_value(encoded, resolver)
         if available:
             kwargs[name] = value
         elif name in ("custom_field_data", "default"):
             kwargs[name] = value if value is not None else ({} if name == "custom_field_data" else None)
             unresolved_values.append((name, encoded))
         else:
-            return None, [], []
+            return None, [], [], []
     for name, spec in record.get("relations", {}).items():
         field = valid_fields.get(name)
         if not isinstance(field, (models.ForeignKey, models.OneToOneField)):
             continue
-        value, available = _resolve_reference(spec, resolved)
+        tenant_relation = is_tenant_relation(field)
+        required_before_save = relation_required_before_save(field)
+        value, available = resolver.resolve(spec)
         if available:
-            kwargs[name] = value
+            if value is MISSING_REFERENCE:
-        elif field.null:
+                if (required_before_save and not tenant_relation) or (
-            kwargs[name] = None
+                    not field.null and not field.has_default() and not tenant_relation
+                ):
+                    missing_required.append(name)
+            elif value is None and tenant_relation and tenant_required:
+                continue
+            elif value is not None and relation_should_be_deferred(field):
+                unresolved.append((name, spec))
+            else:
+                kwargs[name] = value
+        elif field.null and not required_before_save:
             unresolved.append((name, spec))
         else:
-            return None, [], []
+            return None, [], [], []
-    return kwargs, unresolved, unresolved_values
+    return kwargs, unresolved, unresolved_values, missing_required
 
 
-def _set_generic_relations(obj, record, resolved, *, allow_deferred: bool):
+def _set_generic_relations(obj, record, resolver, *, allow_deferred: bool):
     fields = {field.name: field for field in generic_foreign_keys(type(obj))}
     unresolved = []
+    missing_required = []
     for name, spec in record.get("generic_relations", {}).items():
         field = fields.get(name)
         if not field:
             continue
-        value, available = _resolve_reference(spec, resolved)
+        value, available = resolver.resolve(spec)
         if available:
-            setattr(obj, name, value)
+            if value is MISSING_REFERENCE:
+                ct_field = obj._meta.get_field(field.ct_field)
+                id_field = obj._meta.get_field(field.fk_field)
+                if not ct_field.null or not id_field.null:
+                    missing_required.append(name)
+            else:
+                setattr(obj, name, value)
         elif allow_deferred:
             ct_field = obj._meta.get_field(field.ct_field)
             id_field = obj._meta.get_field(field.fk_field)
             if ct_field.null and id_field.null:
-                setattr(obj, field.ct_field, None)
-                setattr(obj, field.fk_field, None)
                 unresolved.append((name, spec))
             else:
                 return None
         else:
             return None
-    return unresolved
+    return unresolved, missing_required
 
 
-def _set_files(obj, record, assets, saved_files, *, dry_run: bool):
+def _open_image_with_bounded_override(data):
-    if dry_run:
+    stream = io.BytesIO(data)
-        return
+    try:
+        with warnings.catch_warnings():
+            warnings.simplefilter("error", PillowImage.DecompressionBombWarning)
+            image = PillowImage.open(stream)
+        return stream, image
+    except (PillowImage.DecompressionBombError, PillowImage.DecompressionBombWarning):
+        stream.close()
+    except Exception:
+        stream.close()
+        raise
+
+    stream = io.BytesIO(data)
+    with _IMAGE_LIMIT_LOCK:
+        original_limit = PillowImage.MAX_IMAGE_PIXELS
+        PillowImage.MAX_IMAGE_PIXELS = IMPORTED_IMAGE_SOURCE_MAX_PIXELS
+        try:
+            with warnings.catch_warnings():
+                warnings.simplefilter("ignore", PillowImage.DecompressionBombWarning)
+                image = PillowImage.open(stream)
+        except PillowImage.DecompressionBombError as exc:
+            stream.close()
+            raise ArchiveValidationError(
+                "Das Bild überschreitet die sichere Importgrenze von "
+                f"{IMPORTED_IMAGE_SOURCE_MAX_PIXELS:,} Pixeln."
+            ) from exc
+        except Exception:
+            stream.close()
+            raise
+        finally:
+            PillowImage.MAX_IMAGE_PIXELS = original_limit
+    return stream, image
+
+
+def _resized_image_save_options(image_format, image):
+    options = {}
+    if icc_profile := image.info.get("icc_profile"):
+        options["icc_profile"] = icc_profile
+    if image_format in ("JPEG", "MPO"):
+        options.update(quality=85, optimize=True, progressive=True)
+    elif image_format == "WEBP":
+        options.update(quality=85, method=4)
+    elif image_format in ("PNG", "GIF"):
+        options["optimize"] = True
+    elif image_format == "TIFF":
+        options["compression"] = "tiff_deflate"
+    return options
+
+
+def _prepare_image_asset(data):
+    stream, image = _open_image_with_bounded_override(data)
+    try:
+        width, height = image.size
+        source_pixels = width * height
+        if source_pixels > IMPORTED_IMAGE_SOURCE_MAX_PIXELS:
+            raise ArchiveValidationError(
+                f"Das Bild mit {source_pixels:,} Pixeln überschreitet die sichere Importgrenze von "
+                f"{IMPORTED_IMAGE_SOURCE_MAX_PIXELS:,} Pixeln."
+            )
+        if source_pixels <= NETBOX_IMAGE_MAX_PIXELS:
+            return data, width, height, None
+
+        scale = math.sqrt(IMPORTED_IMAGE_TARGET_PIXELS / source_pixels)
+        target_size = (max(1, math.floor(width * scale)), max(1, math.floor(height * scale)))
+        image.thumbnail(target_size, PillowImage.Resampling.LANCZOS, reducing_gap=3.0)
+        image_format = image.format or "PNG"
+        if image_format == "MPO":
+            image_format = "JPEG"
+        output = io.BytesIO()
+        image.save(output, format=image_format, **_resized_image_save_options(image_format, image))
+        resized_width, resized_height = image.size
+        resize_info = (width, height, resized_width, resized_height)
+        return output.getvalue(), resized_width, resized_height, resize_info
+    finally:
+        image.close()
+        stream.close()
+
+
+def _set_files(obj, record, assets, saved_files, *, dry_run: bool, import_warnings=None):
     for name, spec in record.get("files", {}).items():
         if not spec or "path" not in spec or spec["path"] not in assets:
             continue
         filename = spec.get("name", spec["path"]).replace("\\", "/").rsplit("/", 1)[-1]
+        field = obj._meta.get_field(name)
+        content_data = assets[spec["path"]]
+        dimensions = (None, None)
+        if isinstance(field, models.ImageField):
+            content_data, width, height, resize_info = _prepare_image_asset(content_data)
+            dimensions = (width, height)
+            if field.width_field and width is not None:
+                setattr(obj, field.width_field, width)
+            if field.height_field and height is not None:
+                setattr(obj, field.height_field, height)
+            if resize_info is not None and import_warnings is not None:
+                old_width, old_height, new_width, new_height = resize_info
+                record_id = record.get("id", obj._meta.label_lower)
+                import_warnings.append(
+                    f"Bild für {record_id} wurde von {old_width}×{old_height} auf "
+                    f"{new_width}×{new_height} Pixel verkleinert."
+                )
+        if dry_run:
+            continue
+        content = ContentFile(content_data)
         file_value = getattr(obj, name)
-        file_value.save(filename, ContentFile(assets[spec["path"]]), save=False)
+        file_value.save(filename, content, save=False)
         saved_files.append((file_value.storage, file_value.name))
+        width, height = dimensions
+        if isinstance(field, models.ImageField) and field.width_field and width is not None:
+            setattr(obj, field.width_field, width)
+        if isinstance(field, models.ImageField) and field.height_field and height is not None:
+            setattr(obj, field.height_field, height)
 
 
 def _cleanup_files(saved_files):
@@ -308,7 +648,7 @@ def _cleanup_files(saved_files):
             logger.warning("Could not remove rolled-back import file %s", name, exc_info=True)
 
 
-def _apply_m2m(obj, record, resolved):
+def _apply_m2m(obj, record, resolver):
     for name, specs in record.get("many_to_many", {}).items():
         try:
             manager = getattr(obj, name)
@@ -316,9 +656,11 @@ def _apply_m2m(obj, record, resolved):
             continue
         values = []
         for spec in specs:
-            value, available = _resolve_reference(spec, resolved)
+            value, available = resolver.resolve(spec)
             if not available:
                 raise ArchiveValidationError(f"M2M-Referenz für {record['id']} konnte nicht aufgelöst werden.")
+            if value is MISSING_REFERENCE:
+                continue
             values.append(value)
         manager.set(values)
 
@@ -347,9 +689,12 @@ def import_archive(parsed: ParsedArchive, *, conflict_strategy: str, dry_run: bo
 
     report = ImportReport(dry_run=dry_run, warnings=list(parsed.warnings))
     resolved = {}
+    resolver = ReferenceResolver(resolved, report.warnings, _model_for)
+    compatibility = PluginCompatibility(report.warnings, dry_run=dry_run)
     deferred_relations = []
     deferred_generic = []
     deferred_values = []
+    deferred_device_placements = []
     writable = set()
     saved_files = []
 
@@ -360,11 +705,21 @@ def import_archive(parsed: ParsedArchive, *, conflict_strategy: str, dry_run: bo
                 progressed = False
                 for record_id, record in list(pending.items()):
                     model = _model_for(record["model"])
-                    kwargs, unresolved, unresolved_value_fields = _field_kwargs(model, record, resolved)
+                    prepared_record, device_placement = _defer_device_placement(model, record)
+                    if device_placement is not None and not _placement_can_be_applied(
+                        device_placement, resolver
+                    ):
+                        device_placement = None
+                    kwargs, unresolved, unresolved_value_fields, missing_required = _field_kwargs(
+                        model,
+                        prepared_record,
+                        resolver,
+                        tenant_required=compatibility.tenant_required,
+                    )
                     if kwargs is None:
                         continue
                     try:
-                        existing = _find_existing(source_instance, model, record, resolved)
+                        existing = _find_existing(source_instance, model, record, resolver)
                     except IdentityNotReady:
                         continue
                     if existing is not None and conflict_strategy == "fail":
@@ -377,11 +732,30 @@ def import_archive(parsed: ParsedArchive, *, conflict_strategy: str, dry_run: bo
                         obj = existing or model()
                         for name, value in kwargs.items():
                             setattr(obj, name, value)
-                        generic_unresolved = _set_generic_relations(obj, record, resolved, allow_deferred=True)
+                        generic_result = _set_generic_relations(obj, record, resolver, allow_deferred=True)
-                        if generic_unresolved is None:
+                        if generic_result is None:
                             continue
-                        _set_files(obj, record, parsed.assets, saved_files, dry_run=dry_run)
+                        generic_unresolved, missing_generic = generic_result
-                        obj.save()
+                        missing_required.extend(missing_generic)
+                        if missing_required and existing is None:
+                            resolver.skip(record_id, record["model"], missing_required)
+                            report.add(record["model"], "skipped")
+                            pending.pop(record_id)
+                            progressed = True
+                            continue
+                        _set_files(
+                            obj,
+                            record,
+                            parsed.assets,
+                            saved_files,
+                            dry_run=dry_run,
+                            import_warnings=report.warnings,
+                        )
+                        if device_placement is not None:
+                            _stage_device_placement(obj)
+                        compatibility.prepare_initial_save(obj, is_new=existing is None)
+                        _release_port_mapping_conflicts(obj, resolver, record_id)
+                        compatibility.save(obj)
                         action = "updated" if existing is not None else "created"
                         writable.add(record_id)
                         deferred_relations.extend((record_id, name, spec) for name, spec in unresolved)
@@ -389,6 +763,8 @@ def import_archive(parsed: ParsedArchive, *, conflict_strategy: str, dry_run: bo
                         deferred_values.extend(
                             (record_id, name, encoded) for name, encoded in unresolved_value_fields
                         )
+                        if device_placement is not None:
+                            deferred_device_placements.append((record_id, device_placement))
 
                     resolved[record_id] = obj
                     _write_mapping(source_instance, record, obj)
@@ -405,37 +781,52 @@ def import_archive(parsed: ParsedArchive, *, conflict_strategy: str, dry_run: bo
             for record_id, name, spec in deferred_relations:
                 if record_id not in writable:
                     continue
-                value, available = _resolve_reference(spec, resolved)
+                value, available = resolver.resolve(spec)
                 if not available:
                     raise ArchiveValidationError(f"Referenz {name} für {record_id} konnte nicht aufgelöst werden.")
+                if value is MISSING_REFERENCE:
+                    continue
                 obj = resolved[record_id]
+                compatibility.release_unique_relation(obj, name, value)
                 setattr(obj, name, value)
-                obj.save(update_fields=[name])
+                compatibility.save(obj, update_fields=[name])
 
             for record_id, name, spec in deferred_generic:
                 if record_id not in writable:
                     continue
-                value, available = _resolve_reference(spec, resolved)
+                value, available = resolver.resolve(spec)
                 if not available:
                     raise ArchiveValidationError(f"Generische Referenz {name} für {record_id} fehlt.")
+                if value is MISSING_REFERENCE:
+                    continue
                 obj = resolved[record_id]
                 setattr(obj, name, value)
                 field = next(field for field in generic_foreign_keys(type(obj)) if field.name == name)
-                obj.save(update_fields=[field.ct_field, field.fk_field])
+                compatibility.save(obj, update_fields=[field.ct_field, field.fk_field])
 
             for record_id, name, encoded in deferred_values:
                 if record_id not in writable:
                     continue
-                value, available = _decode_archived_value(encoded, resolved)
+                value, available = _decode_archived_value(encoded, resolver)
                 if not available:
                     raise ArchiveValidationError(f"Custom-Field-Referenz {name} für {record_id} fehlt.")
                 obj = resolved[record_id]
                 setattr(obj, name, value)
-                obj.save(update_fields=[name])
+                compatibility.save(obj, update_fields=[name])
+
+            _apply_device_placements(
+                deferred_device_placements,
+                resolved,
+                resolver,
+                compatibility,
+                conflict_strategy=conflict_strategy,
+            )
 
             for record_id, record in records.items():
                 if record_id in writable:
-                    _apply_m2m(resolved[record_id], record, resolved)
+                    _apply_m2m(resolved[record_id], record, resolver)
+
+            _rebuild_imported_cable_paths(records, resolved)
 
             if dry_run:
                 transaction.set_rollback(True)

netbox_export/services/plugin_compat.py

@@ -0,0 +1,172 @@
+from __future__ import annotations
+
+from functools import cache
+
+from django.apps import apps
+from django.core.exceptions import FieldDoesNotExist, ValidationError
+from django.db import models
+
+AUTO_IMPORT_TENANT_NAME = "Auto-Import"
+AUTO_IMPORT_TENANT_SLUG = "auto-import"
+MISSING_TENANT_MESSAGE = "Für dieses Objekt muss ein Mandant angegeben werden."
+
+
+def netbox_utilities_tenant_required() -> bool:
+    """Return the effective tenant policy without requiring the optional plugin."""
+    try:
+        from netbox_utilities.runtime import tenant_required
+    except ImportError:
+        return False
+    return bool(tenant_required())
+
+
+def is_tenant_relation(field) -> bool:
+    if field.name != "tenant":
+        return False
+    related_model = getattr(field.remote_field, "model", None)
+    return getattr(getattr(related_model, "_meta", None), "label_lower", None) == "tenancy.tenant"
+
+
+def is_module_relation(field) -> bool:
+    if field.name != "module":
+        return False
+    related_model = getattr(field.remote_field, "model", None)
+    return getattr(getattr(related_model, "_meta", None), "label_lower", None) == "dcim.module"
+
+
+def _condition_field_names(condition):
+    for child in getattr(condition, "children", ()):
+        if isinstance(child, tuple):
+            yield child[0].split("__", 1)[0]
+        else:
+            yield from _condition_field_names(child)
+
+
+@cache
+def check_constraint_field_names(model) -> frozenset[str]:
+    if model is None:
+        return frozenset()
+    names = set()
+    for constraint in model._meta.constraints:
+        if isinstance(constraint, models.CheckConstraint):
+            names.update(_condition_field_names(constraint.condition))
+    return frozenset(names)
+
+
+def relation_required_before_save(field) -> bool:
+    return (
+        is_tenant_relation(field)
+        or is_module_relation(field)
+        or field.name in check_constraint_field_names(getattr(field, "model", None))
+    )
+
+
+def relation_should_be_deferred(field) -> bool:
+    return bool(field.null and field.unique and not relation_required_before_save(field))
+
+
+class PluginCompatibility:
+    def __init__(
+        self,
+        warnings,
+        *,
+        dry_run: bool,
+        tenant_required: bool | None = None,
+        tenant_model_loader=None,
+    ):
+        self.warnings = warnings
+        self.dry_run = dry_run
+        self.tenant_required = (
+            netbox_utilities_tenant_required() if tenant_required is None else tenant_required
+        )
+        self.tenant_model_loader = tenant_model_loader or (lambda: apps.get_model("tenancy.tenant"))
+        self._fallback_tenant = None
+
+    @staticmethod
+    def _supports_tenant(obj) -> bool:
+        try:
+            field = obj._meta.get_field("tenant")
+        except FieldDoesNotExist:
+            return False
+        return is_tenant_relation(field)
+
+    @staticmethod
+    def _is_missing_tenant_error(exc: ValidationError) -> bool:
+        return MISSING_TENANT_MESSAGE in exc.messages
+
+    def _auto_import_tenant(self):
+        if self._fallback_tenant is not None:
+            return self._fallback_tenant
+
+        model = self.tenant_model_loader()
+        tenant = model._default_manager.filter(name=AUTO_IMPORT_TENANT_NAME).order_by("pk").first()
+        created = tenant is None
+        if tenant is None:
+            slug = AUTO_IMPORT_TENANT_SLUG
+            suffix = 2
+            while model._default_manager.filter(slug=slug).exists():
+                slug = f"{AUTO_IMPORT_TENANT_SLUG}-{suffix}"
+                suffix += 1
+            tenant = model._default_manager.create(name=AUTO_IMPORT_TENANT_NAME, slug=slug)
+
+        if self.dry_run:
+            message = (
+                'Für den Prüflauf wurde der Mandant "Auto-Import" automatisch als Ersatzmandant '
+                "verwendet; er wird nicht dauerhaft angelegt oder geändert."
+            )
+        elif created:
+            message = (
+                'Der Mandant "Auto-Import" wurde automatisch erstellt und Objekten ohne '
+                "auflösbaren Mandanten zugewiesen."
+            )
+        else:
+            message = (
+                'Der vorhandene Mandant "Auto-Import" wurde automatisch für Objekte ohne '
+                "auflösbaren Mandanten ausgewählt."
+            )
+        self.warnings.append(message)
+        self._fallback_tenant = tenant
+        return tenant
+
+    def _assign_fallback_tenant(self, obj, *, force: bool = False) -> bool:
+        if getattr(obj, "tenant_id", None) is not None or not self._supports_tenant(obj):
+            return False
+        if not force and not self.tenant_required:
+            return False
+        obj.tenant = self._auto_import_tenant()
+        return True
+
+    def save(self, obj, **kwargs):
+        self._assign_fallback_tenant(obj)
+        try:
+            return obj.save(**kwargs)
+        except ValidationError as exc:
+            if not self._is_missing_tenant_error(exc) or not self._assign_fallback_tenant(obj, force=True):
+                raise
+            return obj.save(**kwargs)
+
+    @staticmethod
+    def prepare_initial_save(obj, *, is_new: bool):
+        if is_new and obj._meta.label_lower == "dcim.module":
+            # Module.save() otherwise replicates ModuleType components which are
+            # imported explicitly from the archive and may already exist.
+            obj._disable_replication = True
+
+    def release_unique_relation(self, obj, field_name: str, value):
+        field = obj._meta.get_field(field_name)
+        if value is None or not relation_should_be_deferred(field):
+            return
+
+        conflicts = type(obj)._default_manager.filter(**{field.attname: value.pk})
+        if obj.pk is not None:
+            conflicts = conflicts.exclude(pk=obj.pk)
+        conflict_ids = list(conflicts.values_list("pk", flat=True))
+        if not conflict_ids:
+            return
+
+        type(obj)._default_manager.filter(pk__in=conflict_ids).update(**{field.attname: None})
+        identifiers = ", ".join(str(pk) for pk in conflict_ids)
+        self.warnings.append(
+            f"Eindeutige Referenz {obj._meta.label_lower}.{field_name} wurde von Zielobjekt(en) "
+            f"{identifiers} gelöst und dem importierten Objekt neu zugeordnet."
+        )

netbox_export/services/references.py

@@ -0,0 +1,68 @@
+from __future__ import annotations
+
+from .codec import decode_scalar
+from .exceptions import ArchiveValidationError
+
+MISSING_REFERENCE = object()
+
+
+class ReferenceResolver:
+    def __init__(self, resolved, warnings, model_loader):
+        self.resolved = resolved
+        self.warnings = warnings
+        self.model_loader = model_loader
+        self.skipped = set()
+        self._external_cache = {}
+        self._warning_keys = set()
+
+    def warn(self, key, message):
+        if key not in self._warning_keys:
+            self._warning_keys.add(key)
+            self.warnings.append(message)
+
+    def skip(self, record_id, model_label, field_names):
+        self.skipped.add(record_id)
+        fields = ", ".join(field_names)
+        self.warn(
+            ("skipped", record_id),
+            f"{record_id} ({model_label}) wurde wegen fehlender Pflichtreferenz übersprungen: {fields}.",
+        )
+
+    def _external_object(self, spec):
+        cache_key = repr(spec)
+        if cache_key in self._external_cache:
+            return self._external_cache[cache_key]
+        model = self.model_loader(spec["model"])
+        lookup = {key: decode_scalar(value) for key, value in spec.get("lookup", {}).items()}
+        reason = None
+        try:
+            value = model._default_manager.get(**lookup)
+        except model.DoesNotExist:
+            value = MISSING_REFERENCE
+            reason = "nicht gefunden"
+        except model.MultipleObjectsReturned:
+            value = MISSING_REFERENCE
+            reason = "nicht eindeutig"
+        if value is MISSING_REFERENCE:
+            self.warn(
+                ("external", cache_key),
+                f"Externe Referenz ausgelassen: {spec['model']} mit {lookup} ({reason}).",
+            )
+        self._external_cache[cache_key] = value
+        return value
+
+    def resolve(self, spec):
+        if spec is None:
+            return None, True
+        if "ref" in spec:
+            record_id = spec["ref"]
+            if record_id in self.skipped:
+                self.warn(
+                    ("internal", record_id),
+                    f"Referenzen auf das übersprungene Objekt {record_id} wurden ausgelassen.",
+                )
+                return MISSING_REFERENCE, True
+            return self.resolved.get(record_id), record_id in self.resolved
+        if "external" in spec:
+            return self._external_object(spec["external"]), True
+        raise ArchiveValidationError("Eine Objektreferenz im Archiv ist ungültig.")

netbox_export/templates/netbox_export/dashboard.html

@@ -149,6 +149,8 @@
         form.setAttribute('aria-busy', busy ? 'true' : 'false');
       };
 
+      const formTarget = (form) => form.getAttribute('action') || window.location.href;
+
       const updateProgress = (container, label, percent = null, state = 'active') => {
         const bar = container.querySelector('.progress-bar');
         const progress = container.querySelector('[role="progressbar"]');
@@ -179,7 +181,7 @@
         updateProgress(exportProgress, 'Export wird vorbereitet ...');
 
         const request = new XMLHttpRequest();
-        request.open('POST', exportForm.action || window.location.href);
+        request.open('POST', formTarget(exportForm));
         request.responseType = 'blob';
         request.onprogress = (progressEvent) => {
           if (!progressEvent.lengthComputable) return;
@@ -220,7 +222,7 @@
         updateProgress(importProgress, 'Archiv wird hochgeladen ...', 0);
 
         const request = new XMLHttpRequest();
-        request.open('POST', importForm.action || window.location.href);
+        request.open('POST', formTarget(importForm));
         request.upload.onprogress = (progressEvent) => {
           if (!progressEvent.lengthComputable) return;
           const percent = Math.min(100, Math.round((progressEvent.loaded / progressEvent.total) * 100));

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "netbox-export"
-version = "0.3.0"
+version = "0.3.12"
 description = "Portable ZIP export and import for scoped NetBox data"
 readme = "README.md"
 requires-python = ">=3.12"

tests/conftest.py

@@ -6,7 +6,7 @@ from django.conf import settings
 if not settings.configured:
     settings.configure(
         DATABASES={"default": {"ENGINE": "django.db.backends.sqlite3", "NAME": ":memory:"}},
-        INSTALLED_APPS=["django.contrib.contenttypes"],
+        INSTALLED_APPS=["django.contrib.contenttypes", "netbox_export"],
         SECRET_KEY="tests",
     )
 

tests/test_cable_import.py

@@ -0,0 +1,80 @@
+from types import SimpleNamespace
+from unittest.mock import MagicMock
+
+from netbox_export.services import importer
+from netbox_export.services.importer import (
+    _rebuild_imported_cable_paths,
+    _release_port_mapping_conflicts,
+)
+
+
+class FakeQuerySet(list):
+    def exclude(self, **kwargs):
+        return FakeQuerySet(item for item in self if item.pk != kwargs["pk"])
+
+    def order_by(self, *args):
+        return self
+
+
+class FakeManager:
+    def __init__(self, objects):
+        self.objects = FakeQuerySet(objects)
+        self.filters = []
+
+    def select_for_update(self):
+        return self
+
+    def filter(self, *args, **kwargs):
+        self.filters.append((args, kwargs))
+        return FakeQuerySet(self.objects)
+
+
+def test_port_mapping_conflicts_are_replaced_before_save():
+    deleted = []
+    conflict = SimpleNamespace(pk=12, delete=lambda: deleted.append(12))
+
+    class PortMapping:
+        _meta = SimpleNamespace(label_lower="dcim.portmapping")
+        _default_manager = FakeManager([conflict])
+
+    mapping = PortMapping()
+    mapping.pk = 11
+    mapping.front_port_id = 21
+    mapping.front_port_position = 1
+    mapping.rear_port_id = 31
+    mapping.rear_port_position = 1
+    resolver = SimpleNamespace(warn=MagicMock())
+
+    _release_port_mapping_conflicts(mapping, resolver, "dcim.portmapping:99")
+
+    assert deleted == [12]
+    resolver.warn.assert_called_once()
+    assert "Front-/Rear-Port-Verkabelung" in resolver.warn.call_args.args[1]
+
+
+def test_imported_cable_terminations_trigger_netbox_path_rebuild(monkeypatch):
+    cables = [SimpleNamespace(pk=7), SimpleNamespace(pk=9)]
+
+    class Cable:
+        _default_manager = FakeManager(cables)
+
+    trace_paths = SimpleNamespace(send=MagicMock())
+    monkeypatch.setattr(importer, "_model_for", lambda label: Cable)
+    monkeypatch.setattr(importer, "_trace_paths_signal", lambda: trace_paths)
+    records = {
+        "dcim.cabletermination:1": {"model": "dcim.cabletermination"},
+        "dcim.cabletermination:2": {"model": "dcim.cabletermination"},
+        "dcim.device:3": {"model": "dcim.device"},
+    }
+    resolved = {
+        "dcim.cabletermination:1": SimpleNamespace(cable_id=7),
+        "dcim.cabletermination:2": SimpleNamespace(cable_id=9),
+        "dcim.device:3": object(),
+    }
+
+    _rebuild_imported_cable_paths(records, resolved)
+
+    assert all(cable._terminations_modified is True for cable in cables)
+    assert trace_paths.send.call_count == 2
+    trace_paths.send.assert_any_call(Cable, instance=cables[0], created=False)
+    trace_paths.send.assert_any_call(Cable, instance=cables[1], created=False)

tests/test_dashboard_template.py

@@ -0,0 +1,15 @@
+from pathlib import Path
+
+
+def test_form_action_field_cannot_shadow_request_target():
+    template = (
+        Path(__file__).parents[1]
+        / "netbox_export"
+        / "templates"
+        / "netbox_export"
+        / "dashboard.html"
+    ).read_text(encoding="utf-8")
+
+    assert "form.getAttribute('action')" in template
+    assert "exportForm.action" not in template
+    assert "importForm.action" not in template

tests/test_device_placement.py

@@ -0,0 +1,291 @@
+from decimal import Decimal
+from types import SimpleNamespace
+from typing import ClassVar
+
+import pytest
+from django.db import connection, models
+
+from netbox_export.services.exceptions import ImportConflictError
+from netbox_export.services.importer import (
+    DeferredDevicePlacement,
+    _apply_device_placements,
+    _defer_device_placement,
+    _save_device_placement,
+    _stage_device_placement,
+)
+from netbox_export.services.plugin_compat import PluginCompatibility
+from netbox_export.services.references import ReferenceResolver
+
+pytestmark = pytest.mark.django_db(transaction=True)
+
+
+class PlacementRack(models.Model):
+    name = models.CharField(max_length=64)
+
+    class Meta:
+        app_label = "placement_tests"
+
+
+class PlacementDeviceType(models.Model):
+    u_height = models.DecimalField(max_digits=4, decimal_places=1, default=1)
+    is_full_depth = models.BooleanField(default=False)
+
+    class Meta:
+        app_label = "placement_tests"
+
+
+class PlacementDevice(models.Model):
+    name = models.CharField(max_length=64)
+    device_type = models.ForeignKey(PlacementDeviceType, on_delete=models.PROTECT)
+    rack = models.ForeignKey(PlacementRack, on_delete=models.PROTECT, null=True)
+    position = models.DecimalField(max_digits=4, decimal_places=1, null=True)
+    face = models.CharField(max_length=16, null=True)
+
+    class Meta:
+        app_label = "placement_tests"
+        constraints: ClassVar[list] = [
+            models.UniqueConstraint(
+                fields=("rack", "position", "face"),
+                name="placement_tests_unique_rack_position_face",
+            )
+        ]
+
+
+@pytest.fixture
+def placement_schema():
+    with connection.schema_editor() as schema_editor:
+        schema_editor.create_model(PlacementRack)
+        schema_editor.create_model(PlacementDeviceType)
+        schema_editor.create_model(PlacementDevice)
+    try:
+        yield
+    finally:
+        with connection.schema_editor() as schema_editor:
+            schema_editor.delete_model(PlacementDevice)
+            schema_editor.delete_model(PlacementDeviceType)
+            schema_editor.delete_model(PlacementRack)
+
+
+def placement_context(device, rack, *, position="31.0", face="front"):
+    record_id = "dcim.device:77"
+    resolved = {record_id: device, "dcim.rack:2": rack}
+    warnings = []
+    resolver = ReferenceResolver(resolved, warnings, lambda label: None)
+    compatibility = PluginCompatibility(warnings, dry_run=False, tenant_required=False)
+    placement = DeferredDevicePlacement(
+        rack_spec={"ref": "dcim.rack:2"},
+        position=Decimal(position) if position is not None else None,
+        face=face,
+    )
+    return record_id, resolved, warnings, resolver, compatibility, placement
+
+
+def test_device_placement_is_removed_from_initial_record():
+    model = SimpleNamespace(_meta=SimpleNamespace(label_lower="dcim.device"))
+    record = {
+        "fields": {"name": "Router 1", "position": {"$type": "decimal", "value": "31.0"}, "face": "front"},
+        "relations": {"site": {"ref": "dcim.site:1"}, "rack": {"ref": "dcim.rack:2"}},
+    }
+
+    prepared, placement = _defer_device_placement(model, record)
+
+    assert prepared["fields"] == {"name": "Router 1"}
+    assert prepared["relations"] == {"site": {"ref": "dcim.site:1"}}
+    assert placement == DeferredDevicePlacement(
+        rack_spec={"ref": "dcim.rack:2"},
+        position=Decimal("31.0"),
+        face="front",
+    )
+    assert "position" in record["fields"]
+    assert "rack" in record["relations"]
+
+
+def test_saving_placement_persists_location_inherited_from_rack():
+    location = object()
+    rack = SimpleNamespace(location=location)
+    device = SimpleNamespace(rack=None, position=None, face=None, location=None)
+    saved = []
+    compatibility = SimpleNamespace(save=lambda obj, **kwargs: saved.append((obj, kwargs)))
+
+    _save_device_placement(device, rack, Decimal("31.0"), "front", compatibility)
+
+    assert device.location is location
+    assert saved == [
+        (
+            device,
+            {"update_fields": ["rack", "position", "face", "location"]},
+        )
+    ]
+
+
+def test_update_releases_exact_rack_occupant_and_places_imported_device(placement_schema):
+    rack = PlacementRack.objects.create(name="R01")
+    device_type = PlacementDeviceType.objects.create()
+    occupant = PlacementDevice.objects.create(
+        name="Existing", device_type=device_type, rack=rack, position=Decimal("31.0"), face="front"
+    )
+    imported = PlacementDevice.objects.create(name="Imported", device_type=device_type)
+    record_id, resolved, warnings, resolver, compatibility, placement = placement_context(imported, rack)
+
+    _apply_device_placements(
+        [(record_id, placement)],
+        resolved,
+        resolver,
+        compatibility,
+        conflict_strategy="update",
+    )
+
+    occupant.refresh_from_db()
+    imported.refresh_from_db()
+    assert (occupant.rack, occupant.position, occupant.face) == (rack, None, None)
+    assert (imported.rack, imported.position, imported.face) == (rack, Decimal("31.0"), "front")
+    assert len(warnings) == 1
+    assert f"Zielgerät(e) {occupant.pk} (Existing)" in warnings[0]
+
+
+def test_skip_keeps_occupant_and_leaves_imported_device_unpositioned(placement_schema):
+    rack = PlacementRack.objects.create(name="R01")
+    device_type = PlacementDeviceType.objects.create()
+    occupant = PlacementDevice.objects.create(
+        name="Existing", device_type=device_type, rack=rack, position=Decimal("31.0"), face="front"
+    )
+    imported = PlacementDevice.objects.create(name="Imported", device_type=device_type)
+    record_id, resolved, warnings, resolver, compatibility, placement = placement_context(imported, rack)
+
+    _apply_device_placements(
+        [(record_id, placement)],
+        resolved,
+        resolver,
+        compatibility,
+        conflict_strategy="skip",
+    )
+
+    occupant.refresh_from_db()
+    imported.refresh_from_db()
+    assert (occupant.rack, occupant.position, occupant.face) == (rack, Decimal("31.0"), "front")
+    assert (imported.rack, imported.position, imported.face) == (rack, None, None)
+    assert len(warnings) == 1
+    assert "wurde ohne Position im Rack gespeichert" in warnings[0]
+
+
+def test_fail_reports_rack_conflict_before_database_constraint(placement_schema):
+    rack = PlacementRack.objects.create(name="R01")
+    device_type = PlacementDeviceType.objects.create()
+    PlacementDevice.objects.create(
+        name="Existing", device_type=device_type, rack=rack, position=Decimal("31.0"), face="front"
+    )
+    imported = PlacementDevice.objects.create(name="Imported", device_type=device_type)
+    record_id, resolved, _, resolver, compatibility, placement = placement_context(imported, rack)
+
+    with pytest.raises(ImportConflictError, match="Rackplatzkonflikt.*Rack R01"):
+        _apply_device_placements(
+            [(record_id, placement)],
+            resolved,
+            resolver,
+            compatibility,
+            conflict_strategy="fail",
+        )
+
+
+def test_half_depth_devices_can_share_position_on_opposite_faces(placement_schema):
+    rack = PlacementRack.objects.create(name="R01")
+    device_type = PlacementDeviceType.objects.create(is_full_depth=False)
+    occupant = PlacementDevice.objects.create(
+        name="Existing", device_type=device_type, rack=rack, position=Decimal("31.0"), face="front"
+    )
+    imported = PlacementDevice.objects.create(name="Imported", device_type=device_type)
+    record_id, resolved, warnings, resolver, compatibility, placement = placement_context(
+        imported, rack, face="rear"
+    )
+
+    _apply_device_placements(
+        [(record_id, placement)],
+        resolved,
+        resolver,
+        compatibility,
+        conflict_strategy="update",
+    )
+
+    occupant.refresh_from_db()
+    imported.refresh_from_db()
+    assert occupant.position == Decimal("31.0")
+    assert (imported.position, imported.face) == (Decimal("31.0"), "rear")
+    assert warnings == []
+
+
+def test_full_depth_multi_u_overlap_is_released(placement_schema):
+    rack = PlacementRack.objects.create(name="R01")
+    full_depth = PlacementDeviceType.objects.create(u_height=Decimal("2.0"), is_full_depth=True)
+    half_depth = PlacementDeviceType.objects.create(u_height=Decimal("1.0"), is_full_depth=False)
+    occupant = PlacementDevice.objects.create(
+        name="Existing",
+        device_type=full_depth,
+        rack=rack,
+        position=Decimal("30.5"),
+        face="front",
+    )
+    imported = PlacementDevice.objects.create(name="Imported", device_type=half_depth)
+    record_id, resolved, warnings, resolver, compatibility, placement = placement_context(
+        imported, rack, position="31.0", face="rear"
+    )
+
+    _apply_device_placements(
+        [(record_id, placement)],
+        resolved,
+        resolver,
+        compatibility,
+        conflict_strategy="update",
+    )
+
+    occupant.refresh_from_db()
+    imported.refresh_from_db()
+    assert occupant.position is None
+    assert (imported.position, imported.face) == (Decimal("31.0"), "rear")
+    assert len(warnings) == 1
+
+
+def test_imported_devices_can_swap_rack_positions(placement_schema):
+    rack = PlacementRack.objects.create(name="R01")
+    device_type = PlacementDeviceType.objects.create()
+    first = PlacementDevice.objects.create(
+        name="First", device_type=device_type, rack=rack, position=Decimal("10.0"), face="front"
+    )
+    second = PlacementDevice.objects.create(
+        name="Second", device_type=device_type, rack=rack, position=Decimal("20.0"), face="front"
+    )
+    for device in (first, second):
+        _stage_device_placement(device)
+        device.save(update_fields=["position", "face"])
+
+    resolved = {
+        "dcim.device:1": first,
+        "dcim.device:2": second,
+        "dcim.rack:2": rack,
+    }
+    warnings = []
+    resolver = ReferenceResolver(resolved, warnings, lambda label: None)
+    compatibility = PluginCompatibility(warnings, dry_run=False, tenant_required=False)
+    placements = [
+        (
+            "dcim.device:1",
+            DeferredDevicePlacement({"ref": "dcim.rack:2"}, Decimal("20.0"), "front"),
+        ),
+        (
+            "dcim.device:2",
+            DeferredDevicePlacement({"ref": "dcim.rack:2"}, Decimal("10.0"), "front"),
+        ),
+    ]
+
+    _apply_device_placements(
+        placements,
+        resolved,
+        resolver,
+        compatibility,
+        conflict_strategy="update",
+    )
+
+    first.refresh_from_db()
+    second.refresh_from_db()
+    assert first.position == Decimal("20.0")
+    assert second.position == Decimal("10.0")
+    assert warnings == []

tests/test_file_import.py

@@ -0,0 +1,107 @@
+import io
+
+from django.core.files.storage import InMemoryStorage
+from django.db import models
+from PIL import Image
+
+from netbox_export.services import importer as importer_module
+from netbox_export.services.importer import _cleanup_files, _set_files
+
+
+class ImageAsset(models.Model):
+    image = models.ImageField(
+        upload_to="test-images",
+        height_field="image_height",
+        width_field="image_width",
+    )
+    image_height = models.PositiveSmallIntegerField()
+    image_width = models.PositiveSmallIntegerField()
+
+    class Meta:
+        app_label = "file_import_tests"
+
+
+def image_bytes(width=3, height=2):
+    output = io.BytesIO()
+    Image.new("RGB", (width, height), color="white").save(output, format="PNG")
+    return output.getvalue()
+
+
+def image_record():
+    return {
+        "files": {
+            "image": {
+                "path": "assets/extras.imageattachment_17/image/server-room.png",
+                "name": "image-attachments/location_2_Serverraum.jpg",
+            }
+        }
+    }
+
+
+def test_dry_run_derives_required_image_dimensions_without_storing_file():
+    obj = ImageAsset()
+    assets = {"assets/extras.imageattachment_17/image/server-room.png": image_bytes()}
+    saved_files = []
+
+    _set_files(obj, image_record(), assets, saved_files, dry_run=True)
+
+    assert (obj.image_width, obj.image_height) == (3, 2)
+    assert not obj.image
+    assert saved_files == []
+
+
+def test_file_import_keeps_derived_dimensions_after_storage_save(monkeypatch):
+    field = ImageAsset._meta.get_field("image")
+    storage = InMemoryStorage()
+    monkeypatch.setattr(field, "storage", storage)
+    obj = ImageAsset()
+    assets = {"assets/extras.imageattachment_17/image/server-room.png": image_bytes(5, 4)}
+    saved_files = []
+
+    _set_files(obj, image_record(), assets, saved_files, dry_run=False)
+
+    assert (obj.image_width, obj.image_height) == (5, 4)
+    assert storage.exists(obj.image.name)
+    assert saved_files == [(storage, obj.image.name)]
+
+    _cleanup_files(saved_files)
+    assert not storage.exists(obj.image.name)
+
+
+def test_oversized_image_is_resized_after_pillow_bomb_error(monkeypatch):
+    monkeypatch.setattr(importer_module, "NETBOX_IMAGE_MAX_PIXELS", 50)
+    monkeypatch.setattr(importer_module, "IMPORTED_IMAGE_TARGET_PIXELS", 40)
+    monkeypatch.setattr(importer_module, "IMPORTED_IMAGE_SOURCE_MAX_PIXELS", 200)
+    monkeypatch.setattr(Image, "MAX_IMAGE_PIXELS", 50)
+    field = ImageAsset._meta.get_field("image")
+    storage = InMemoryStorage()
+    monkeypatch.setattr(field, "storage", storage)
+    obj = ImageAsset()
+    record = image_record()
+    record["id"] = "extras.imageattachment:17"
+    assets = {"assets/extras.imageattachment_17/image/server-room.png": image_bytes(11, 10)}
+    import_warnings = []
+    saved_files = []
+
+    _set_files(
+        obj,
+        record,
+        assets,
+        saved_files,
+        dry_run=False,
+        import_warnings=import_warnings,
+    )
+
+    assert obj.image_width * obj.image_height <= 40
+    with storage.open(obj.image.name, "rb") as stored_file, Image.open(stored_file) as stored_image:
+        assert stored_image.size == (obj.image_width, obj.image_height)
+        assert stored_image.format == "PNG"
+    assert import_warnings == [
+        (
+            f"Bild für extras.imageattachment:17 wurde von 11×10 auf "
+            f"{obj.image_width}×{obj.image_height} Pixel verkleinert."
+        )
+    ]
+    assert Image.MAX_IMAGE_PIXELS == 50
+
+    _cleanup_files(saved_files)

tests/test_graph.py

@@ -36,6 +36,36 @@ class GraphDetail(models.Model):
         app_label = "graph_tests"
 
 
+class PrivateGraphModel(models.Model):
+    _netbox_private = True
+
+    class Meta:
+        app_label = "graph_tests"
+
+
+class PortMapping(models.Model):
+    _netbox_private = True
+
+    class Meta:
+        app_label = "dcim"
+
+
+class CablePath(models.Model):
+    _netbox_private = True
+
+    class Meta:
+        app_label = "dcim"
+
+
+def test_private_plugin_models_are_not_exportable():
+    assert graph_module.is_exportable_model(PrivateGraphModel) is False
+
+
+def test_private_port_mappings_are_exported_but_cable_paths_are_not():
+    assert graph_module.is_exportable_model(PortMapping) is True
+    assert graph_module.is_exportable_model(CablePath) is False
+
+
 def test_batched_graph_collects_members_and_dependencies(monkeypatch):
     graph_models = (GraphParent, GraphReference, GraphChild, GraphDetail)
     with connection.schema_editor() as schema_editor:

tests/test_importer_plugin_compat.py

@@ -0,0 +1,299 @@
+import uuid
+from typing import ClassVar
+
+import pytest
+from django.db import connection, models
+
+from netbox_export.services import importer as importer_module
+from netbox_export.services.importer import (
+    _field_kwargs,
+    _find_existing,
+    _identity_lookup,
+)
+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:
+        app_label = "tenancy"
+
+
+class TenantManagedObject(models.Model):
+    tenant = models.ForeignKey(Tenant, on_delete=models.PROTECT, null=True)
+
+    class Meta:
+        app_label = "compat_tests"
+
+
+class Platform(models.Model):
+    class Meta:
+        app_label = "compat_tests"
+
+
+class ConstrainedInstallation(models.Model):
+    platform = models.ForeignKey(Platform, on_delete=models.PROTECT, null=True)
+
+    class Meta:
+        app_label = "compat_tests"
+        constraints: ClassVar[list] = [
+            models.CheckConstraint(
+                condition=models.Q(platform__isnull=False),
+                name="compat_tests_installation_platform",
+            )
+        ]
+
+
+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"
+
+
+class Device(models.Model):
+    class Meta:
+        app_label = "dcim"
+
+
+class RearPort(models.Model):
+    device = models.ForeignKey(Device, on_delete=models.CASCADE)
+    name = models.CharField(max_length=64)
+
+    class Meta:
+        app_label = "dcim"
+        constraints: ClassVar[list] = [
+            models.UniqueConstraint(
+                fields=("device", "name"),
+                name="dcim_rearport_unique_device_name",
+            )
+        ]
+
+
+class Module(models.Model):
+    class Meta:
+        app_label = "dcim"
+
+
+class ModuleBay(models.Model):
+    device = models.ForeignKey(Device, on_delete=models.CASCADE)
+    module = models.ForeignKey(Module, on_delete=models.CASCADE)
+    name = models.CharField(max_length=64)
+
+    class Meta:
+        app_label = "dcim"
+
+
+class ModularRearPort(models.Model):
+    module = models.ForeignKey(Module, on_delete=models.SET_NULL, null=True)
+
+    class Meta:
+        app_label = "compat_tests"
+
+
+def tenant_record():
+    return {
+        "fields": {},
+        "relations": {"tenant": {"ref": "tenancy.tenant:23"}},
+    }
+
+
+def test_unresolved_tenant_always_blocks_initial_save():
+    resolver = ReferenceResolver({}, [], lambda label: None)
+
+    kwargs, unresolved, unresolved_values, missing_required = _field_kwargs(
+        TenantManagedObject,
+        tenant_record(),
+        resolver,
+        tenant_required=True,
+    )
+
+    assert kwargs is None
+    assert unresolved == []
+    assert unresolved_values == []
+    assert missing_required == []
+
+
+def test_unresolved_tenant_also_blocks_when_policy_detection_is_unavailable():
+    resolver = ReferenceResolver({}, [], lambda label: None)
+
+    kwargs, unresolved, unresolved_values, missing_required = _field_kwargs(
+        TenantManagedObject,
+        tenant_record(),
+        resolver,
+        tenant_required=False,
+    )
+
+    assert kwargs is None
+    assert unresolved == []
+    assert unresolved_values == []
+    assert missing_required == []
+
+
+def test_unresolved_relation_in_check_constraint_blocks_initial_save():
+    resolver = ReferenceResolver({}, [], lambda label: None)
+    record = {
+        "fields": {},
+        "relations": {"platform": {"ref": "compat_tests.platform:5"}},
+    }
+
+    kwargs, unresolved, unresolved_values, missing_required = _field_kwargs(
+        ConstrainedInstallation,
+        record,
+        resolver,
+        tenant_required=False,
+    )
+
+    assert kwargs is None
+    assert unresolved == []
+    assert unresolved_values == []
+    assert missing_required == []
+
+
+def test_modular_component_waits_for_module_before_initial_save():
+    resolver = ReferenceResolver({}, [], lambda label: None)
+    record = {
+        "fields": {},
+        "relations": {"module": {"ref": "dcim.module:12"}},
+    }
+
+    kwargs, unresolved, unresolved_values, missing_required = _field_kwargs(
+        ModularRearPort,
+        record,
+        resolver,
+        tenant_required=False,
+    )
+
+    assert kwargs is None
+    assert unresolved == []
+    assert unresolved_values == []
+    assert missing_required == []
+
+
+def test_module_bay_identity_includes_owning_module():
+    device = Device(pk=210)
+    module = Module(pk=12)
+    resolver = ReferenceResolver(
+        {
+            "dcim.device:210": device,
+            "dcim.module:12": module,
+        },
+        [],
+        lambda label: None,
+    )
+    record = {
+        "fields": {"name": "3-LC"},
+        "relations": {
+            "device": {"ref": "dcim.device:210"},
+            "module": {"ref": "dcim.module:12"},
+        },
+    }
+
+    lookup = _identity_lookup(ModuleBay, record, resolver)
+
+    assert lookup == {"device": device, "module": module, "name": "3-LC"}
+
+
+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)
+
+
+def test_natural_key_rebinds_stale_mapping_instead_of_creating_duplicate(monkeypatch):
+    with connection.schema_editor() as schema_editor:
+        schema_editor.create_model(Device)
+        schema_editor.create_model(RearPort)
+    try:
+        device = Device.objects.create()
+        natural = RearPort.objects.create(device=device, name="3-LC")
+        stale_mapping_target = RearPort(pk=999, device=device, name="Alt")
+        monkeypatch.setattr(
+            importer_module,
+            "_mapped_object",
+            lambda source_instance, record, model: stale_mapping_target,
+        )
+        warnings = []
+        resolver = ReferenceResolver(
+            {"dcim.device:160": device},
+            warnings,
+            lambda label: None,
+        )
+        record = {
+            "id": "dcim.rearport:77",
+            "model": "dcim.rearport",
+            "source_pk": "77",
+            "fields": {"name": "3-LC"},
+            "relations": {"device": {"ref": "dcim.device:160"}},
+        }
+
+        existing = _find_existing(uuid.uuid4(), RearPort, record, resolver)
+
+        assert existing == natural
+        assert warnings == [
+            (
+                "Gespeicherte Zuordnung für dcim.rearport:77 wurde von Ziel-ID 999 auf "
+                f"Ziel-ID {natural.pk} korrigiert, da der Fachschlüssel "
+                f"{{'device': <Device: Device object ({device.pk})>, 'name': '3-LC'}} bereits existiert."
+            )
+        ]
+    finally:
+        with connection.schema_editor() as schema_editor:
+            schema_editor.delete_model(RearPort)
+            schema_editor.delete_model(Device)

tests/test_plugin_compat.py

@@ -0,0 +1,203 @@
+from types import SimpleNamespace
+
+from django.core.exceptions import FieldDoesNotExist, ValidationError
+from django.db import models
+
+from netbox_export.services.plugin_compat import (
+    MISSING_TENANT_MESSAGE,
+    PluginCompatibility,
+    check_constraint_field_names,
+    is_module_relation,
+    is_tenant_relation,
+    relation_required_before_save,
+    relation_should_be_deferred,
+)
+
+
+def relation_field(name="tenant", related_label="tenancy.tenant"):
+    related_model = SimpleNamespace(_meta=SimpleNamespace(label_lower=related_label))
+    return SimpleNamespace(name=name, remote_field=SimpleNamespace(model=related_model))
+
+
+class FakeQuerySet:
+    def __init__(self, result=None, exists=False):
+        self.result = result
+        self._exists = exists
+
+    def order_by(self, *fields):
+        return self
+
+    def first(self):
+        return self.result
+
+    def exists(self):
+        return self._exists
+
+
+class FakeTenantManager:
+    def __init__(self, existing=None):
+        self.existing = existing
+        self.created = []
+
+    def filter(self, **lookup):
+        if "name" in lookup:
+            return FakeQuerySet(self.existing)
+        return FakeQuerySet(exists=False)
+
+    def create(self, **values):
+        tenant = SimpleNamespace(pk=42, **values)
+        self.created.append(tenant)
+        return tenant
+
+
+class FakeTenantModel:
+    _default_manager = FakeTenantManager()
+
+
+class FakeMeta:
+    def get_field(self, name):
+        if name != "tenant":
+            raise FieldDoesNotExist(name)
+        return relation_field()
+
+
+class TenantRequiredObject:
+    _meta = FakeMeta()
+
+    def __init__(self):
+        self.tenant_id = None
+        self.save_calls = 0
+
+    @property
+    def tenant(self):
+        return None
+
+    @tenant.setter
+    def tenant(self, value):
+        self.tenant_id = value.pk
+
+    def save(self, **kwargs):
+        self.save_calls += 1
+        if self.tenant_id is None:
+            raise ValidationError(MISSING_TENANT_MESSAGE)
+
+
+def test_identifies_tenant_relation():
+    assert is_tenant_relation(relation_field()) is True
+    assert is_tenant_relation(relation_field("site", "dcim.site")) is False
+
+
+def test_identifies_module_component_relation():
+    assert is_module_relation(relation_field("module", "dcim.module")) is True
+    assert is_module_relation(relation_field("device", "dcim.device")) is False
+
+
+def test_new_module_disables_automatic_component_replication():
+    module = SimpleNamespace(_meta=SimpleNamespace(label_lower="dcim.module"))
+
+    PluginCompatibility.prepare_initial_save(module, is_new=True)
+
+    assert module._disable_replication is True
+
+
+def test_existing_module_keeps_normal_save_behavior():
+    module = SimpleNamespace(_meta=SimpleNamespace(label_lower="dcim.module"))
+
+    PluginCompatibility.prepare_initial_save(module, is_new=False)
+
+    assert not hasattr(module, "_disable_replication")
+
+
+def test_identifies_relations_used_by_check_constraint():
+    constraint = models.CheckConstraint(
+        condition=(
+            models.Q(device__isnull=False, virtualmachine__isnull=True)
+            | models.Q(device__isnull=True, virtualmachine__isnull=False)
+        ),
+        name="platform",
+    )
+    model = type("ConstrainedModel", (), {"_meta": SimpleNamespace(constraints=[constraint])})
+    field = relation_field("device", "dcim.device")
+    field.model = model
+
+    assert check_constraint_field_names(model) == frozenset({"device", "virtualmachine"})
+    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})
+    warnings = []
+    compatibility = PluginCompatibility(
+        warnings,
+        dry_run=False,
+        tenant_required=True,
+        tenant_model_loader=lambda: tenant_model,
+    )
+    obj = TenantRequiredObject()
+
+    compatibility.save(obj)
+
+    assert obj.tenant_id == 42
+    assert obj.save_calls == 1
+    assert len(manager.created) == 1
+    assert warnings == [
+        (
+            'Der Mandant "Auto-Import" wurde automatisch erstellt und Objekten ohne '
+            'auflösbaren Mandanten zugewiesen.'
+        )
+    ]
+
+
+def test_validation_error_forces_fallback_for_older_plugin_versions():
+    existing = SimpleNamespace(pk=7, name="Auto-Import", slug="auto-import")
+    manager = FakeTenantManager(existing=existing)
+    tenant_model = type("TenantModel", (), {"_default_manager": manager})
+    warnings = []
+    compatibility = PluginCompatibility(
+        warnings,
+        dry_run=False,
+        tenant_required=False,
+        tenant_model_loader=lambda: tenant_model,
+    )
+    obj = TenantRequiredObject()
+
+    compatibility.save(obj)
+
+    assert obj.tenant_id == 7
+    assert obj.save_calls == 2
+    assert manager.created == []
+    assert warnings == [
+        (
+            'Der vorhandene Mandant "Auto-Import" wurde automatisch für Objekte ohne '
+            'auflösbaren Mandanten ausgewählt.'
+        )
+    ]

tests/test_references.py

@@ -0,0 +1,91 @@
+from netbox_export.services.references import MISSING_REFERENCE, ReferenceResolver
+
+
+class DoesNotExist(Exception):
+    pass
+
+
+class MultipleObjectsReturned(Exception):
+    pass
+
+
+class FakeManager:
+    def __init__(self, result):
+        self.result = result
+        self.calls = 0
+
+    def get(self, **lookup):
+        self.calls += 1
+        if isinstance(self.result, Exception):
+            raise self.result
+        return self.result
+
+
+def fake_model(manager):
+    return type(
+        "FakeModel",
+        (),
+        {
+            "DoesNotExist": DoesNotExist,
+            "MultipleObjectsReturned": MultipleObjectsReturned,
+            "_default_manager": manager,
+        },
+    )
+
+
+def test_missing_external_reference_is_omitted_and_warned_once():
+    manager = FakeManager(DoesNotExist())
+    model = fake_model(manager)
+    warnings = []
+    resolver = ReferenceResolver({}, warnings, lambda label: model)
+    spec = {"external": {"model": "users.user", "lookup": {"username": "lke"}}}
+
+    first, first_available = resolver.resolve(spec)
+    second, second_available = resolver.resolve(spec)
+
+    assert first is MISSING_REFERENCE
+    assert second is MISSING_REFERENCE
+    assert first_available is True
+    assert second_available is True
+    assert manager.calls == 1
+    assert warnings == [
+        "Externe Referenz ausgelassen: users.user mit {'username': 'lke'} (nicht gefunden)."
+    ]
+
+
+def test_ambiguous_external_reference_is_omitted():
+    manager = FakeManager(MultipleObjectsReturned())
+    model = fake_model(manager)
+    warnings = []
+    resolver = ReferenceResolver({}, warnings, lambda label: model)
+
+    value, available = resolver.resolve(
+        {"external": {"model": "users.group", "lookup": {"name": "NetBox"}}}
+    )
+
+    assert value is MISSING_REFERENCE
+    assert available is True
+    assert warnings == [
+        "Externe Referenz ausgelassen: users.group mit {'name': 'NetBox'} (nicht eindeutig)."
+    ]
+
+
+def test_reference_to_skipped_record_is_omitted_and_warned_once():
+    warnings = []
+    resolver = ReferenceResolver({}, warnings, lambda label: None)
+    resolver.skip("dcim.rackreservation:17", "dcim.rackreservation", ["user"])
+
+    first, first_available = resolver.resolve({"ref": "dcim.rackreservation:17"})
+    second, second_available = resolver.resolve({"ref": "dcim.rackreservation:17"})
+
+    assert first is MISSING_REFERENCE
+    assert second is MISSING_REFERENCE
+    assert first_available is True
+    assert second_available is True
+    assert warnings == [
+        (
+            "dcim.rackreservation:17 (dcim.rackreservation) wurde wegen fehlender "
+            "Pflichtreferenz übersprungen: user."
+        ),
+        "Referenzen auf das übersprungene Objekt dcim.rackreservation:17 wurden ausgelassen.",
+    ]