fix: honor tenant validation during plugin imports
This commit is contained in:
@@ -107,6 +107,19 @@ 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.
|
||||
|
||||
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
|
||||
|
||||
@@ -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.2"
|
||||
version = "0.3.3"
|
||||
author = "NetBox Export contributors"
|
||||
base_url = "netbox-export"
|
||||
min_version = "4.6.0"
|
||||
|
||||
@@ -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.2",
|
||||
"plugin_version": "0.3.3",
|
||||
"scope": {
|
||||
"type": scope_type,
|
||||
"source_pk": str(scope_id),
|
||||
|
||||
@@ -40,6 +40,7 @@ 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)
|
||||
and opts.app_label not in EXCLUDED_APP_LABELS
|
||||
and opts.label_lower not in EXCLUDED_MODELS
|
||||
)
|
||||
|
||||
@@ -16,6 +16,10 @@ 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 (
|
||||
netbox_utilities_tenant_required,
|
||||
relation_required_before_save,
|
||||
)
|
||||
from .references import MISSING_REFERENCE, ReferenceResolver
|
||||
|
||||
EXPLICIT_IDENTITIES = {
|
||||
@@ -210,7 +214,7 @@ def _write_mapping(source_instance, record, obj):
|
||||
)
|
||||
|
||||
|
||||
def _field_kwargs(model, record, resolver):
|
||||
def _field_kwargs(model, record, resolver, *, tenant_required: bool):
|
||||
valid_fields = {field.name: field for field in model._meta.concrete_fields}
|
||||
kwargs = {}
|
||||
unresolved = []
|
||||
@@ -240,14 +244,17 @@ def _field_kwargs(model, record, resolver):
|
||||
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)
|
||||
value, available = resolver.resolve(spec)
|
||||
if available:
|
||||
if value is MISSING_REFERENCE:
|
||||
if not field.null and not field.has_default():
|
||||
if required_before_save or (not field.null and not field.has_default()):
|
||||
missing_required.append(name)
|
||||
elif value is None and required_before_save:
|
||||
missing_required.append(name)
|
||||
else:
|
||||
kwargs[name] = value
|
||||
elif field.null:
|
||||
elif field.null and not required_before_save:
|
||||
unresolved.append((name, spec))
|
||||
else:
|
||||
return None, [], [], []
|
||||
@@ -345,6 +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()
|
||||
deferred_relations = []
|
||||
deferred_generic = []
|
||||
deferred_values = []
|
||||
@@ -359,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
|
||||
model, record, resolver, tenant_required=tenant_required
|
||||
)
|
||||
if kwargs is None:
|
||||
continue
|
||||
|
||||
@@ -0,0 +1,17 @@
|
||||
from __future__ import annotations
|
||||
|
||||
|
||||
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 relation_required_before_save(field, *, tenant_required: bool) -> bool:
|
||||
if not tenant_required or 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"
|
||||
+1
-1
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
|
||||
|
||||
[project]
|
||||
name = "netbox-export"
|
||||
version = "0.3.2"
|
||||
version = "0.3.3"
|
||||
description = "Portable ZIP export and import for scoped NetBox data"
|
||||
readme = "README.md"
|
||||
requires-python = ">=3.12"
|
||||
|
||||
+1
-1
@@ -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",
|
||||
)
|
||||
|
||||
|
||||
@@ -36,6 +36,17 @@ class GraphDetail(models.Model):
|
||||
app_label = "graph_tests"
|
||||
|
||||
|
||||
class PrivateGraphModel(models.Model):
|
||||
_netbox_private = True
|
||||
|
||||
class Meta:
|
||||
app_label = "graph_tests"
|
||||
|
||||
|
||||
def test_private_plugin_models_are_not_exportable():
|
||||
assert graph_module.is_exportable_model(PrivateGraphModel) is False
|
||||
|
||||
|
||||
def test_batched_graph_collects_members_and_dependencies(monkeypatch):
|
||||
graph_models = (GraphParent, GraphReference, GraphChild, GraphDetail)
|
||||
with connection.schema_editor() as schema_editor:
|
||||
|
||||
@@ -0,0 +1,55 @@
|
||||
from django.db import models
|
||||
|
||||
from netbox_export.services.importer import _field_kwargs
|
||||
from netbox_export.services.references import ReferenceResolver
|
||||
|
||||
|
||||
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"
|
||||
|
||||
|
||||
def tenant_record():
|
||||
return {
|
||||
"fields": {},
|
||||
"relations": {"tenant": {"ref": "tenancy.tenant:23"}},
|
||||
}
|
||||
|
||||
|
||||
def test_unresolved_tenant_blocks_initial_save_when_netbox_utilities_requires_it():
|
||||
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_optional_tenant_remains_deferred_without_requirement():
|
||||
resolver = ReferenceResolver({}, [], lambda label: None)
|
||||
|
||||
kwargs, unresolved, unresolved_values, missing_required = _field_kwargs(
|
||||
TenantManagedObject,
|
||||
tenant_record(),
|
||||
resolver,
|
||||
tenant_required=False,
|
||||
)
|
||||
|
||||
assert kwargs == {}
|
||||
assert unresolved == [("tenant", {"ref": "tenancy.tenant:23"})]
|
||||
assert unresolved_values == []
|
||||
assert missing_required == []
|
||||
@@ -0,0 +1,26 @@
|
||||
from types import SimpleNamespace
|
||||
|
||||
from netbox_export.services.plugin_compat import relation_required_before_save
|
||||
|
||||
|
||||
def relation_field(name, related_label):
|
||||
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")
|
||||
|
||||
assert relation_required_before_save(field, tenant_required=True) is True
|
||||
|
||||
|
||||
def test_tenant_relation_can_still_be_deferred_when_policy_is_disabled():
|
||||
field = relation_field("tenant", "tenancy.tenant")
|
||||
|
||||
assert relation_required_before_save(field, tenant_required=False) is False
|
||||
|
||||
|
||||
def test_unrelated_relation_is_not_affected_by_tenant_policy():
|
||||
field = relation_field("site", "dcim.site")
|
||||
|
||||
assert relation_required_before_save(field, tenant_required=True) is False
|
||||
Reference in New Issue
Block a user