feat: add scoped NetBox ZIP export and import plugin

This commit is contained in:
2026-08-05 11:22:47 +02:00
commit d0db98b690
23 changed files with 1811 additions and 0 deletions
+27
View File
@@ -0,0 +1,27 @@
import sys
import types
from django.conf import settings
if not settings.configured:
settings.configure(
INSTALLED_APPS=["django.contrib.contenttypes"],
SECRET_KEY="tests",
)
netbox_module = types.ModuleType("netbox")
plugins_module = types.ModuleType("netbox.plugins")
class PluginConfig:
pass
plugins_module.PluginConfig = PluginConfig
netbox_module.plugins = plugins_module
sys.modules.setdefault("netbox", netbox_module)
sys.modules.setdefault("netbox.plugins", plugins_module)
import django
django.setup()
+99
View File
@@ -0,0 +1,99 @@
import io
import json
import zipfile
import pytest
from netbox_export.services.archive import build_archive, read_archive
from netbox_export.services.exceptions import ArchiveValidationError
def manifest():
return {
"source_instance": "b89196f8-3d87-466a-9278-f68e22d6d2cc",
"scope": {"type": "site", "source_pk": "1", "label": "Berlin"},
"object_count": 1,
}
def records():
return [
{
"id": "dcim.site:1",
"model": "dcim.site",
"source_pk": "1",
"fields": {"name": "Berlin"},
}
]
def test_archive_round_trip():
payload = build_archive(manifest(), records(), {"assets/a.txt": b"data"})
parsed = read_archive(payload, max_size=1024 * 1024, max_objects=10)
assert parsed.records == records()
assert parsed.assets == {"assets/a.txt": b"data"}
assert parsed.warnings == ["Das Archiv ist nicht signiert."]
def test_signed_archive_requires_matching_key():
payload = build_archive(manifest(), records(), {}, signing_key="secret-a")
with pytest.raises(ArchiveValidationError, match="Archivsignatur"):
read_archive(payload, max_size=1024 * 1024, max_objects=10, signing_key="secret-b")
def test_signed_archive_without_local_key_is_not_claimed_as_verified():
payload = build_archive(manifest(), records(), {}, signing_key="secret-a")
parsed = read_archive(payload, max_size=1024 * 1024, max_objects=10)
assert parsed.warnings == ["Das Archiv ist signiert, aber ohne konfigurierten Schlüssel nicht verifiziert."]
def test_modified_object_stream_is_rejected():
payload = build_archive(manifest(), records(), {})
source = zipfile.ZipFile(io.BytesIO(payload))
output = io.BytesIO()
with source, zipfile.ZipFile(output, "w") as target:
for info in source.infolist():
content = source.read(info.filename)
if info.filename == "objects.ndjson":
content = content.replace(b"Berlin", b"Hamburg")
target.writestr(info, content)
with pytest.raises(ArchiveValidationError, match="Prüfsumme"):
read_archive(output.getvalue(), max_size=1024 * 1024, max_objects=10)
def test_modified_asset_is_rejected():
payload = build_archive(manifest(), records(), {"assets/a.txt": b"original"})
source = zipfile.ZipFile(io.BytesIO(payload))
output = io.BytesIO()
with source, zipfile.ZipFile(output, "w") as target:
for info in source.infolist():
content = source.read(info.filename)
if info.filename == "assets/a.txt":
content = b"modified"
target.writestr(info, content)
with pytest.raises(ArchiveValidationError, match="Prüfsumme der Datei"):
read_archive(output.getvalue(), max_size=1024 * 1024, max_objects=10)
def test_path_traversal_is_rejected():
output = io.BytesIO()
with zipfile.ZipFile(output, "w") as archive:
archive.writestr("../manifest.json", json.dumps(manifest()))
with pytest.raises(ArchiveValidationError, match="Dateipfad"):
read_archive(output.getvalue(), max_size=1024 * 1024, max_objects=10)
def test_object_limit_is_enforced():
payload = build_archive(manifest(), records() * 2, {})
with pytest.raises(ArchiveValidationError, match="zu viele Objekte"):
read_archive(payload, max_size=1024 * 1024, max_objects=1)
+19
View File
@@ -0,0 +1,19 @@
import datetime
import decimal
import uuid
from netbox_export.services.codec import decode_scalar, encode_scalar
def test_scalar_round_trip():
value = {
"decimal": decimal.Decimal("12.340"),
"uuid": uuid.UUID("00112233-4455-6677-8899-aabbccddeeff"),
"date": datetime.date(2026, 8, 5),
"datetime": datetime.datetime(2026, 8, 5, 10, 30, tzinfo=datetime.UTC),
"duration": datetime.timedelta(seconds=42),
"bytes": b"binary",
}
assert decode_scalar(encode_scalar(value)) == value