diff --git a/README.md b/README.md index 348d23c..768ef80 100644 --- a/README.md +++ b/README.md @@ -113,7 +113,9 @@ 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. +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. Auf Quelle und Ziel müssen jeweils dieselben Plugin-Versionen und Migrationen installiert sein. Verschlüsselte Zugangsdaten von NetBox-VM-Import sind nur bei diff --git a/netbox_export/__init__.py b/netbox_export/__init__.py index 4c36610..23fc6b2 100644 --- a/netbox_export/__init__.py +++ b/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.3" + version = "0.3.4" author = "NetBox Export contributors" base_url = "netbox-export" min_version = "4.6.0" diff --git a/netbox_export/services/exporter.py b/netbox_export/services/exporter.py index 7735cb1..8b6b37c 100644 --- a/netbox_export/services/exporter.py +++ b/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.3", + "plugin_version": "0.3.4", "scope": { "type": scope_type, "source_pk": str(scope_id), diff --git a/netbox_export/services/importer.py b/netbox_export/services/importer.py index 3f11a26..69b9d3d 100644 --- a/netbox_export/services/importer.py +++ b/netbox_export/services/importer.py @@ -17,8 +17,8 @@ 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 ( - netbox_utilities_tenant_required, - relation_required_before_save, + PluginCompatibility, + is_tenant_relation, ) from .references import MISSING_REFERENCE, ReferenceResolver @@ -244,17 +244,17 @@ def _field_kwargs(model, record, resolver, *, tenant_required: bool): field = valid_fields.get(name) if not isinstance(field, (models.ForeignKey, models.OneToOneField)): continue - required_before_save = relation_required_before_save(field, tenant_required=tenant_required) + tenant_relation = is_tenant_relation(field) value, available = resolver.resolve(spec) if available: if value is MISSING_REFERENCE: - if required_before_save or (not field.null and not field.has_default()): + if not field.null and not field.has_default(): missing_required.append(name) - elif value is None and required_before_save: - missing_required.append(name) + elif value is None and tenant_relation and tenant_required: + continue else: kwargs[name] = value - elif field.null and not required_before_save: + elif field.null and not tenant_relation: unresolved.append((name, spec)) else: return None, [], [], [] @@ -352,7 +352,7 @@ 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) - tenant_required = netbox_utilities_tenant_required() + compatibility = PluginCompatibility(report.warnings, dry_run=dry_run) deferred_relations = [] deferred_generic = [] deferred_values = [] @@ -367,7 +367,7 @@ def import_archive(parsed: ParsedArchive, *, conflict_strategy: str, dry_run: bo for record_id, record in list(pending.items()): model = _model_for(record["model"]) kwargs, unresolved, unresolved_value_fields, missing_required = _field_kwargs( - model, record, resolver, tenant_required=tenant_required + model, record, resolver, tenant_required=compatibility.tenant_required ) if kwargs is None: continue @@ -397,7 +397,7 @@ def import_archive(parsed: ParsedArchive, *, conflict_strategy: str, dry_run: bo progressed = True continue _set_files(obj, record, parsed.assets, saved_files, dry_run=dry_run) - obj.save() + 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) @@ -428,7 +428,7 @@ def import_archive(parsed: ParsedArchive, *, conflict_strategy: str, dry_run: bo continue obj = resolved[record_id] 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: @@ -441,7 +441,7 @@ def import_archive(parsed: ParsedArchive, *, conflict_strategy: str, dry_run: bo 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: @@ -451,7 +451,7 @@ def import_archive(parsed: ParsedArchive, *, conflict_strategy: str, dry_run: bo 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]) for record_id, record in records.items(): if record_id in writable: diff --git a/netbox_export/services/plugin_compat.py b/netbox_export/services/plugin_compat.py index 583cdb0..9a7d667 100644 --- a/netbox_export/services/plugin_compat.py +++ b/netbox_export/services/plugin_compat.py @@ -1,5 +1,12 @@ from __future__ import annotations +from django.apps import apps +from django.core.exceptions import FieldDoesNotExist, ValidationError + +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.""" @@ -10,8 +17,89 @@ def netbox_utilities_tenant_required() -> bool: return bool(tenant_required()) -def relation_required_before_save(field, *, tenant_required: bool) -> bool: - if not tenant_required or field.name != "tenant": +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" + + +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) diff --git a/pyproject.toml b/pyproject.toml index 4bc5a6a..a282721 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "netbox-export" -version = "0.3.3" +version = "0.3.4" description = "Portable ZIP export and import for scoped NetBox data" readme = "README.md" requires-python = ">=3.12" diff --git a/tests/test_importer_plugin_compat.py b/tests/test_importer_plugin_compat.py index 5c51a41..6a28b7f 100644 --- a/tests/test_importer_plugin_compat.py +++ b/tests/test_importer_plugin_compat.py @@ -23,7 +23,7 @@ def tenant_record(): } -def test_unresolved_tenant_blocks_initial_save_when_netbox_utilities_requires_it(): +def test_unresolved_tenant_always_blocks_initial_save(): resolver = ReferenceResolver({}, [], lambda label: None) kwargs, unresolved, unresolved_values, missing_required = _field_kwargs( @@ -39,7 +39,7 @@ def test_unresolved_tenant_blocks_initial_save_when_netbox_utilities_requires_it assert missing_required == [] -def test_unresolved_optional_tenant_remains_deferred_without_requirement(): +def test_unresolved_tenant_also_blocks_when_policy_detection_is_unavailable(): resolver = ReferenceResolver({}, [], lambda label: None) kwargs, unresolved, unresolved_values, missing_required = _field_kwargs( @@ -49,7 +49,7 @@ def test_unresolved_optional_tenant_remains_deferred_without_requirement(): tenant_required=False, ) - assert kwargs == {} - assert unresolved == [("tenant", {"ref": "tenancy.tenant:23"})] + assert kwargs is None + assert unresolved == [] assert unresolved_values == [] assert missing_required == [] diff --git a/tests/test_plugin_compat.py b/tests/test_plugin_compat.py index d9fa36b..ba75f3d 100644 --- a/tests/test_plugin_compat.py +++ b/tests/test_plugin_compat.py @@ -1,26 +1,133 @@ from types import SimpleNamespace -from netbox_export.services.plugin_compat import relation_required_before_save +from django.core.exceptions import FieldDoesNotExist, ValidationError + +from netbox_export.services.plugin_compat import ( + MISSING_TENANT_MESSAGE, + PluginCompatibility, + is_tenant_relation, +) -def relation_field(name, related_label): +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)) -def test_tenant_relation_is_resolved_before_initial_save_when_required(): - field = relation_field("tenant", "tenancy.tenant") +class FakeQuerySet: + def __init__(self, result=None, exists=False): + self.result = result + self._exists = exists - assert relation_required_before_save(field, tenant_required=True) is True + def order_by(self, *fields): + return self + + def first(self): + return self.result + + def exists(self): + return self._exists -def test_tenant_relation_can_still_be_deferred_when_policy_is_disabled(): - field = relation_field("tenant", "tenancy.tenant") +class FakeTenantManager: + def __init__(self, existing=None): + self.existing = existing + self.created = [] - assert relation_required_before_save(field, tenant_required=False) is False + 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 -def test_unrelated_relation_is_not_affected_by_tenant_policy(): - field = relation_field("site", "dcim.site") +class FakeTenantModel: + _default_manager = FakeTenantManager() - assert relation_required_before_save(field, tenant_required=True) is False + +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_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.' + ) + ]