fix: resolve constrained relations before insert
This commit is contained in:
@@ -116,6 +116,9 @@ erzwungene Mandantenpflicht wird beim Import berücksichtigt: Das Zielobjekt wir
|
|||||||
erst gespeichert, nachdem sein Mandant importiert und zugeordnet wurde. Ist kein
|
erst gespeichert, nachdem sein Mandant importiert und zugeordnet wurde. Ist kein
|
||||||
Mandant auflösbar, wird automatisch ein vorhandener Mandant `Auto-Import`
|
Mandant auflösbar, wird automatisch ein vorhandener Mandant `Auto-Import`
|
||||||
verwendet oder neu angelegt. Der Importbericht weist darauf hin.
|
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.
|
||||||
|
|
||||||
Auf Quelle und Ziel müssen jeweils dieselben Plugin-Versionen und Migrationen
|
Auf Quelle und Ziel müssen jeweils dieselben Plugin-Versionen und Migrationen
|
||||||
installiert sein. Verschlüsselte Zugangsdaten von NetBox-VM-Import sind nur bei
|
installiert sein. Verschlüsselte Zugangsdaten von NetBox-VM-Import sind nur bei
|
||||||
|
|||||||
@@ -7,7 +7,7 @@ class NetBoxExportConfig(PluginConfig):
|
|||||||
name = "netbox_export"
|
name = "netbox_export"
|
||||||
verbose_name = "NetBox-Export"
|
verbose_name = "NetBox-Export"
|
||||||
description = "Portable ZIP export and import for tenants and locations"
|
description = "Portable ZIP export and import for tenants and locations"
|
||||||
version = "0.3.4"
|
version = "0.3.5"
|
||||||
author = "NetBox Export contributors"
|
author = "NetBox Export contributors"
|
||||||
base_url = "netbox-export"
|
base_url = "netbox-export"
|
||||||
min_version = "4.6.0"
|
min_version = "4.6.0"
|
||||||
|
|||||||
@@ -39,7 +39,7 @@ def export_scope(
|
|||||||
"created_at": datetime.now(UTC).isoformat(),
|
"created_at": datetime.now(UTC).isoformat(),
|
||||||
"source_instance": str(InstanceIdentity.local_id()),
|
"source_instance": str(InstanceIdentity.local_id()),
|
||||||
"source_netbox_version": getattr(getattr(settings, "RELEASE", None), "version", "4.6"),
|
"source_netbox_version": getattr(getattr(settings, "RELEASE", None), "version", "4.6"),
|
||||||
"plugin_version": "0.3.4",
|
"plugin_version": "0.3.5",
|
||||||
"scope": {
|
"scope": {
|
||||||
"type": scope_type,
|
"type": scope_type,
|
||||||
"source_pk": str(scope_id),
|
"source_pk": str(scope_id),
|
||||||
|
|||||||
@@ -19,6 +19,7 @@ from .exceptions import ArchiveValidationError, ExportImportError, ImportConflic
|
|||||||
from .plugin_compat import (
|
from .plugin_compat import (
|
||||||
PluginCompatibility,
|
PluginCompatibility,
|
||||||
is_tenant_relation,
|
is_tenant_relation,
|
||||||
|
relation_required_before_save,
|
||||||
)
|
)
|
||||||
from .references import MISSING_REFERENCE, ReferenceResolver
|
from .references import MISSING_REFERENCE, ReferenceResolver
|
||||||
|
|
||||||
@@ -245,16 +246,19 @@ def _field_kwargs(model, record, resolver, *, tenant_required: bool):
|
|||||||
if not isinstance(field, (models.ForeignKey, models.OneToOneField)):
|
if not isinstance(field, (models.ForeignKey, models.OneToOneField)):
|
||||||
continue
|
continue
|
||||||
tenant_relation = is_tenant_relation(field)
|
tenant_relation = is_tenant_relation(field)
|
||||||
|
required_before_save = relation_required_before_save(field)
|
||||||
value, available = resolver.resolve(spec)
|
value, available = resolver.resolve(spec)
|
||||||
if available:
|
if available:
|
||||||
if value is MISSING_REFERENCE:
|
if value is MISSING_REFERENCE:
|
||||||
if not field.null and not field.has_default():
|
if (required_before_save and not tenant_relation) or (
|
||||||
|
not field.null and not field.has_default() and not tenant_relation
|
||||||
|
):
|
||||||
missing_required.append(name)
|
missing_required.append(name)
|
||||||
elif value is None and tenant_relation and tenant_required:
|
elif value is None and tenant_relation and tenant_required:
|
||||||
continue
|
continue
|
||||||
else:
|
else:
|
||||||
kwargs[name] = value
|
kwargs[name] = value
|
||||||
elif field.null and not tenant_relation:
|
elif field.null and not required_before_save:
|
||||||
unresolved.append((name, spec))
|
unresolved.append((name, spec))
|
||||||
else:
|
else:
|
||||||
return None, [], [], []
|
return None, [], [], []
|
||||||
|
|||||||
@@ -1,7 +1,10 @@
|
|||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from functools import cache
|
||||||
|
|
||||||
from django.apps import apps
|
from django.apps import apps
|
||||||
from django.core.exceptions import FieldDoesNotExist, ValidationError
|
from django.core.exceptions import FieldDoesNotExist, ValidationError
|
||||||
|
from django.db import models
|
||||||
|
|
||||||
AUTO_IMPORT_TENANT_NAME = "Auto-Import"
|
AUTO_IMPORT_TENANT_NAME = "Auto-Import"
|
||||||
AUTO_IMPORT_TENANT_SLUG = "auto-import"
|
AUTO_IMPORT_TENANT_SLUG = "auto-import"
|
||||||
@@ -24,6 +27,31 @@ def is_tenant_relation(field) -> bool:
|
|||||||
return getattr(getattr(related_model, "_meta", None), "label_lower", None) == "tenancy.tenant"
|
return getattr(getattr(related_model, "_meta", None), "label_lower", None) == "tenancy.tenant"
|
||||||
|
|
||||||
|
|
||||||
|
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 field.name in check_constraint_field_names(
|
||||||
|
getattr(field, "model", None)
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
class PluginCompatibility:
|
class PluginCompatibility:
|
||||||
def __init__(
|
def __init__(
|
||||||
self,
|
self,
|
||||||
|
|||||||
+1
-1
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
|
|||||||
|
|
||||||
[project]
|
[project]
|
||||||
name = "netbox-export"
|
name = "netbox-export"
|
||||||
version = "0.3.4"
|
version = "0.3.5"
|
||||||
description = "Portable ZIP export and import for scoped NetBox data"
|
description = "Portable ZIP export and import for scoped NetBox data"
|
||||||
readme = "README.md"
|
readme = "README.md"
|
||||||
requires-python = ">=3.12"
|
requires-python = ">=3.12"
|
||||||
|
|||||||
@@ -1,3 +1,5 @@
|
|||||||
|
from typing import ClassVar
|
||||||
|
|
||||||
from django.db import models
|
from django.db import models
|
||||||
|
|
||||||
from netbox_export.services.importer import _field_kwargs
|
from netbox_export.services.importer import _field_kwargs
|
||||||
@@ -16,6 +18,24 @@ class TenantManagedObject(models.Model):
|
|||||||
app_label = "compat_tests"
|
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",
|
||||||
|
)
|
||||||
|
]
|
||||||
|
|
||||||
|
|
||||||
def tenant_record():
|
def tenant_record():
|
||||||
return {
|
return {
|
||||||
"fields": {},
|
"fields": {},
|
||||||
@@ -53,3 +73,23 @@ def test_unresolved_tenant_also_blocks_when_policy_detection_is_unavailable():
|
|||||||
assert unresolved == []
|
assert unresolved == []
|
||||||
assert unresolved_values == []
|
assert unresolved_values == []
|
||||||
assert missing_required == []
|
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 == []
|
||||||
|
|||||||
@@ -1,11 +1,14 @@
|
|||||||
from types import SimpleNamespace
|
from types import SimpleNamespace
|
||||||
|
|
||||||
from django.core.exceptions import FieldDoesNotExist, ValidationError
|
from django.core.exceptions import FieldDoesNotExist, ValidationError
|
||||||
|
from django.db import models
|
||||||
|
|
||||||
from netbox_export.services.plugin_compat import (
|
from netbox_export.services.plugin_compat import (
|
||||||
MISSING_TENANT_MESSAGE,
|
MISSING_TENANT_MESSAGE,
|
||||||
PluginCompatibility,
|
PluginCompatibility,
|
||||||
|
check_constraint_field_names,
|
||||||
is_tenant_relation,
|
is_tenant_relation,
|
||||||
|
relation_required_before_save,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
@@ -82,6 +85,22 @@ def test_identifies_tenant_relation():
|
|||||||
assert is_tenant_relation(relation_field("site", "dcim.site")) is False
|
assert is_tenant_relation(relation_field("site", "dcim.site")) is False
|
||||||
|
|
||||||
|
|
||||||
|
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_creates_and_assigns_auto_import_tenant_when_policy_is_active():
|
def test_creates_and_assigns_auto_import_tenant_when_policy_is_active():
|
||||||
manager = FakeTenantManager()
|
manager = FakeTenantManager()
|
||||||
tenant_model = type("TenantModel", (), {"_default_manager": manager})
|
tenant_model = type("TenantModel", (), {"_default_manager": manager})
|
||||||
|
|||||||
Reference in New Issue
Block a user