MrBlake/Netbox-Utilities
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add tenant groups and required tenancy

Browse Source
This commit is contained in:
MrBlake
2026-07-28 16:57:52 +02:00
parent 54f6a3f6c9
commit 35e06ee004
17 changed files with 427 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files
netbox_utilities/__init__.py
+4 -1
View File
@@ -5,7 +5,7 @@ class NetBoxUtilitiesConfig(PluginConfig):
name = "netbox_utilities"
verbose_name = "NetBox Utilities"
description = "Personal navigation and a global tenant filter"
version = "0.1.0"
version = "0.2.0"
author = "LKE"
base_url = "utilities"
min_version = "4.6.5"
@@ -13,6 +13,7 @@ class NetBoxUtilitiesConfig(PluginConfig):
default_settings = {
"navigation_customization_enabled": True,
"tenant_filter_enabled": True,
"tenant_required": True,
}
middleware = [
"netbox_utilities.middleware.GlobalTenantFilterMiddleware",
@@ -21,8 +22,10 @@ class NetBoxUtilitiesConfig(PluginConfig):
def ready(self):
super().ready()
from .tenant_scope import install_search_filter
from .tenant_validation import install_tenant_validation
install_search_filter()
install_tenant_validation()
config = NetBoxUtilitiesConfig
netbox_utilities/forms.py
+12 -1
View File
@@ -4,18 +4,29 @@ from .models import UtilitiesSettings
class UtilitiesSettingsForm(forms.ModelForm):
def __init__(self, *args, tenant_filter_locked=False, tenant_required_locked=False, **kwargs):
super().__init__(*args, **kwargs)
self.fields["tenant_filter_enabled"].disabled = tenant_filter_locked
self.fields["tenant_required"].disabled = tenant_required_locked
class Meta:
model = UtilitiesSettings
fields = ("tenant_filter_enabled",)
fields = ("tenant_filter_enabled", "tenant_required")
labels = {
"tenant_filter_enabled": "Globalen Mandantenfilter aktivieren",
"tenant_required": "Mandant für mandantenfähige Objekte verpflichtend machen",
}
help_texts = {
"tenant_filter_enabled": (
"Blendet das Mandanten-Dropdown ein und ergänzt unterstützte NetBox-Listen "
"automatisch um den gewählten Mandantenfilter."
),
"tenant_required": (
"Verhindert das Speichern mandantenfähiger Objekte ohne Mandant. "
"Dies gilt auch für Importe, API-Aufrufe und Skripte."
),
}
widgets = {
"tenant_filter_enabled": forms.CheckboxInput(attrs={"class": "form-check-input"}),
"tenant_required": forms.CheckboxInput(attrs={"class": "form-check-input"}),
}
netbox_utilities/middleware.py
+50 -14
View File
@@ -1,10 +1,11 @@
from django.urls import Resolver404, resolve
from tenancy.models import Tenant
from tenancy.models import Tenant, TenantGroup
from .runtime import tenant_filter_enabled
from .tenant_scope import active_tenant_id
from .tenant_scope import ActiveTenantScope, active_tenant_scope
SESSION_TENANT_KEY = "netbox_utilities.selected_tenant_id"
SESSION_TENANT_GROUP_KEY = "netbox_utilities.selected_tenant_group_id"
class GlobalTenantFilterMiddleware:
@@ -15,35 +16,53 @@ class GlobalTenantFilterMiddleware:
def __call__(self, request):
request.netbox_utilities_tenant = None
tenant_id = self._get_selected_tenant_id(request)
token = active_tenant_id.set(tenant_id)
request.netbox_utilities_tenant_group = None
scope = self._get_selected_scope(request)
token = active_tenant_scope.set(scope)
try:
if tenant_id is not None and request.method in {"GET", "HEAD"}:
self._inject_filter_parameter(request, tenant_id)
if scope is not None and request.method in {"GET", "HEAD"}:
self._inject_filter_parameter(request, scope)
return self.get_response(request)
finally:
active_tenant_id.reset(token)
active_tenant_scope.reset(token)
@staticmethod
def _get_selected_tenant_id(request):
def _get_selected_scope(request):
if not tenant_filter_enabled() or not request.user.is_authenticated:
request.session.pop(SESSION_TENANT_KEY, None)
GlobalTenantFilterMiddleware.clear_scope(request)
return None
tenant_group_id = request.session.get(SESSION_TENANT_GROUP_KEY)
if tenant_group_id is not None:
group = TenantGroup.objects.restrict(request.user, "view").filter(pk=tenant_group_id).first()
if group is None:
GlobalTenantFilterMiddleware.clear_scope(request)
return None
descendant_groups = group.get_descendants(include_self=True)
group_ids = frozenset(descendant_groups.values_list("pk", flat=True))
tenant_ids = Tenant.objects.filter(group_id__in=group_ids).values_list("pk", flat=True)
request.netbox_utilities_tenant_group = group
return ActiveTenantScope("group", group.pk, frozenset(tenant_ids), group_ids)
tenant_id = request.session.get(SESSION_TENANT_KEY)
if tenant_id is None:
return None
tenant = Tenant.objects.restrict(request.user, "view").filter(pk=tenant_id).first()
if tenant is None:
request.session.pop(SESSION_TENANT_KEY, None)
GlobalTenantFilterMiddleware.clear_scope(request)
return None
request.netbox_utilities_tenant = tenant
return tenant.pk
return ActiveTenantScope("tenant", tenant.pk, frozenset({tenant.pk}))
@staticmethod
def _inject_filter_parameter(request, tenant_id):
def clear_scope(request):
request.session.pop(SESSION_TENANT_KEY, None)
request.session.pop(SESSION_TENANT_GROUP_KEY, None)
@staticmethod
def _inject_filter_parameter(request, scope):
try:
match = resolve(request.path_info)
except Resolver404:
@@ -54,15 +73,32 @@ class GlobalTenantFilterMiddleware:
return
parameter = None
values = []
if match.view_name == "tenancy:tenant_list":
if scope.kind == "tenant":
parameter = "id"
values = [scope.object_id]
else:
parameter = "group_id"
values = [scope.object_id]
elif match.view_name == "tenancy:tenantgroup_list" and scope.kind == "group":
parameter = "id"
values = list(scope.group_ids) or [scope.object_id]
else:
view_class = getattr(match.func, "view_class", None)
filterset = getattr(view_class, "filterset", None)
if filterset and "tenant_id" in getattr(filterset, "base_filters", {}):
base_filters = getattr(filterset, "base_filters", {}) if filterset else {}
if scope.kind == "tenant" and "tenant_id" in base_filters:
parameter = "tenant_id"
values = [scope.object_id]
elif scope.kind == "group" and "tenant_group_id" in base_filters:
parameter = "tenant_group_id"
values = [scope.object_id]
elif scope.kind == "group" and "tenant_id" in base_filters:
parameter = "tenant_id"
values = list(scope.tenant_ids) or [0]
if parameter:
query = request.GET.copy()
query.setlist(parameter, [str(tenant_id)])
query.setlist(parameter, [str(value) for value in values])
request.GET = query
netbox_utilities/migrations/0002_utilitiessettings_tenant_required.py
+15
View File
@@ -0,0 +1,15 @@
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("netbox_utilities", "0001_initial"),
]
operations = [
migrations.AddField(
model_name="utilitiessettings",
name="tenant_required",
field=models.BooleanField(default=True),
),
]
netbox_utilities/models.py
+1
View File
@@ -28,6 +28,7 @@ class UtilitiesSettings(models.Model):
validators=[MinValueValidator(1)],
)
tenant_filter_enabled = models.BooleanField(default=True)
tenant_required = models.BooleanField(default=True)
updated = models.DateTimeField(auto_now=True)
class Meta:
netbox_utilities/runtime.py
+25 -9
View File
@@ -4,7 +4,7 @@ from netbox.plugins import get_plugin_config
from .models import UtilitiesSettings
SETTINGS_CACHE_KEY = "netbox_utilities.runtime_settings.v1"
SETTINGS_CACHE_KEY = "netbox_utilities.runtime_settings.v2"
def navigation_customization_enabled():
@@ -15,18 +15,34 @@ def tenant_filter_enabled():
if not get_plugin_config("netbox_utilities", "tenant_filter_enabled"):
return False
enabled = cache.get(SETTINGS_CACHE_KEY)
if enabled is None:
return bool(_get_database_settings()["tenant_filter_enabled"])
def tenant_required():
if not get_plugin_config("netbox_utilities", "tenant_required"):
return False
return bool(_get_database_settings()["tenant_required"])
def _get_database_settings():
runtime_settings = cache.get(SETTINGS_CACHE_KEY)
if runtime_settings is None:
try:
enabled = (
UtilitiesSettings.objects.filter(singleton_id=1).values_list("tenant_filter_enabled", flat=True).first()
runtime_settings = (
UtilitiesSettings.objects.filter(singleton_id=1)
.values("tenant_filter_enabled", "tenant_required")
.first()
)
except (OperationalError, ProgrammingError):
# The middleware can be imported before the first plugin migration.
enabled = True
enabled = True if enabled is None else enabled
cache.set(SETTINGS_CACHE_KEY, enabled, timeout=30)
return bool(enabled)
runtime_settings = None
runtime_settings = runtime_settings or {
"tenant_filter_enabled": True,
"tenant_required": True,
}
cache.set(SETTINGS_CACHE_KEY, runtime_settings, timeout=30)
return runtime_settings
def clear_runtime_settings_cache():
netbox_utilities/template_content.py
+4 -1
View File
@@ -1,5 +1,5 @@
from netbox.plugins import PluginTemplateExtension
from tenancy.models import Tenant
from tenancy.models import Tenant, TenantGroup
from .models import NavigationPreference
from .navigation_helpers import get_visible_menus, normalize_preferences
@@ -37,11 +37,14 @@ class UtilitiesGlobalContent(PluginTemplateExtension):
return ""
tenants = Tenant.objects.restrict(request.user, "view").order_by("name")
tenant_groups = TenantGroup.objects.restrict(request.user, "view").order_by("name")
return self.render(
"netbox_utilities/tenant_dropdown.html",
{
"tenants": tenants,
"tenant_groups": tenant_groups,
"selected_tenant": getattr(request, "netbox_utilities_tenant", None),
"selected_tenant_group": getattr(request, "netbox_utilities_tenant_group", None),
"return_url": request.get_full_path(),
},
)
netbox_utilities/templates/netbox_utilities/settings.html
+19 -5
View File
@@ -10,13 +10,18 @@
<div class="row justify-content-center">
<div class="col col-md-9 col-xl-7">
<div class="card">
<div class="card-header"><h2 class="card-title">Mandantenfilter</h2></div>
<div class="card-header"><h2 class="card-title">Mandanten</h2></div>
<div class="card-body">
{% if config_forces_disabled %}
{% if tenant_filter_locked %}
<div class="alert alert-warning">
Der Mandantenfilter ist in <code>PLUGINS_CONFIG</code> fest deaktiviert und kann hier nicht aktiviert werden.
</div>
{% endif %}
{% if tenant_required_locked %}
<div class="alert alert-warning">
Die verpflichtende Mandantenzuordnung ist in <code>PLUGINS_CONFIG</code> fest deaktiviert.
</div>
{% endif %}
<form method="post">
{% csrf_token %}
{% if form.non_field_errors %}<div class="alert alert-danger">{{ form.non_field_errors }}</div>{% endif %}
@@ -27,9 +32,18 @@
</label>
</div>
<div class="form-text mb-3">{{ form.tenant_filter_enabled.help_text }}</div>
<button class="btn btn-primary" type="submit"{% if config_forces_disabled %} disabled{% endif %}>
Speichern
</button>
<div class="form-check form-switch mt-4">
{{ form.tenant_required }}
<label class="form-check-label fw-bold" for="{{ form.tenant_required.id_for_label }}">
{{ form.tenant_required.label }}
</label>
</div>
<div class="form-text mb-3">{{ form.tenant_required.help_text }}</div>
<div class="alert alert-info">
Die Pflicht gilt nur für Objekttypen, die in NetBox ein Mandantenfeld besitzen.
Bestehende Objekte ohne Mandant müssen beim nächsten Speichern ergänzt werden.
</div>
<button class="btn btn-primary" type="submit">Speichern</button>
</form>
</div>
</div>
netbox_utilities/templates/netbox_utilities/tenant_dropdown.html
+28 -9
View File
@@ -9,7 +9,13 @@
>
<i class="mdi mdi-domain" aria-hidden="true"></i>
<span class="d-none d-xl-inline ms-1 tenant-filter-label">
{% if selected_tenant %}{{ selected_tenant }}{% else %}Alle Mandanten{% endif %}
{% if selected_tenant %}
{{ selected_tenant }}
{% elif selected_tenant_group %}
{{ selected_tenant_group }}
{% else %}
Alle Mandanten
{% endif %}
</span>
<i class="mdi mdi-chevron-down" aria-hidden="true"></i>
</button>
@@ -18,26 +24,39 @@
{% csrf_token %}
<input type="hidden" name="next" value="{{ return_url }}">
<label class="form-label fw-bold">
Mandant
Mandant oder Mandantengruppe
</label>
<select
class="form-select"
name="tenant"
name="scope"
onchange="this.form.submit()"
aria-label="Mandant global auswählen"
>
<option value="">Alle Mandanten</option>
{% for tenant in tenants %}
<option value="{{ tenant.pk }}"{% if selected_tenant.pk == tenant.pk %} selected{% endif %}>
{{ tenant.name }}
</option>
{% endfor %}
{% if tenant_groups %}
<optgroup label="Mandantengruppen">
{% for group in tenant_groups %}
<option value="group:{{ group.pk }}"{% if selected_tenant_group.pk == group.pk %} selected{% endif %}>
{{ group.name }}
</option>
{% endfor %}
</optgroup>
{% endif %}
{% if tenants %}
<optgroup label="Mandanten">
{% for tenant in tenants %}
<option value="tenant:{{ tenant.pk }}"{% if selected_tenant.pk == tenant.pk %} selected{% endif %}>
{{ tenant.name }}
</option>
{% endfor %}
</optgroup>
{% endif %}
</select>
<noscript>
<button class="btn btn-primary w-100 mt-2" type="submit">Anwenden</button>
</noscript>
<p class="text-secondary small mb-0 mt-2">
Wirkt auf mandantenfähige Listen und die globale Suche.
Gruppen schließen alle direkt und untergeordnet zugeordneten Mandanten ein.
</p>
</form>
</div>
netbox_utilities/tenant_scope.py
+21 -7
View File
@@ -1,9 +1,19 @@
from contextvars import ContextVar
from dataclasses import dataclass
from functools import wraps
from tenancy.models import Tenant
active_tenant_id = ContextVar("netbox_utilities_active_tenant_id", default=None)
@dataclass(frozen=True)
class ActiveTenantScope:
kind: str
object_id: int
tenant_ids: frozenset[int]
group_ids: frozenset[int] = frozenset()
active_tenant_scope = ContextVar("netbox_utilities_active_tenant_scope", default=None)
_PARENT_RELATIONS = (
"device",
@@ -22,6 +32,10 @@ _PARENT_RELATIONS = (
def object_matches_tenant(obj, tenant_id, depth=0):
return object_matches_tenant_ids(obj, frozenset({tenant_id}), depth=depth)
def object_matches_tenant_ids(obj, tenant_ids, depth=0):
"""
Return whether a search result belongs to the selected tenant.
@@ -31,11 +45,11 @@ def object_matches_tenant(obj, tenant_id, depth=0):
if obj is None:
return False
if isinstance(obj, Tenant):
return obj.pk == tenant_id
return obj.pk in tenant_ids
field_names = {field.name for field in obj._meta.get_fields()}
if "tenant" in field_names:
return getattr(obj, "tenant_id", None) == tenant_id
return getattr(obj, "tenant_id", None) in tenant_ids
if depth < 2:
for relation in _PARENT_RELATIONS:
@@ -44,7 +58,7 @@ def object_matches_tenant(obj, tenant_id, depth=0):
parent = getattr(obj, relation, None)
if parent is None:
return False
return object_matches_tenant(parent, tenant_id, depth + 1)
return object_matches_tenant_ids(parent, tenant_ids, depth + 1)
# Global reference objects (roles, manufacturers, statuses, etc.) are not
# tenant-owned and must remain usable while a filter is active.
@@ -63,10 +77,10 @@ def install_search_filter():
@wraps(original_search)
def tenant_aware_search(*args, **kwargs):
results = original_search(*args, **kwargs)
tenant_id = active_tenant_id.get()
if tenant_id is None:
scope = active_tenant_scope.get()
if scope is None:
return results
return [result for result in results if object_matches_tenant(result.object, tenant_id)]
return [result for result in results if object_matches_tenant_ids(result.object, scope.tenant_ids)]
search_backend.search = tenant_aware_search
search_backend._netbox_utilities_wrapped = True
netbox_utilities/tenant_validation.py
+76
View File
@@ -0,0 +1,76 @@
from functools import wraps
from django.apps import apps
from django.core.exceptions import FieldDoesNotExist, ValidationError
from django.db.models.signals import pre_save
from django.forms.models import BaseModelForm
from tenancy.models import Tenant
from .runtime import tenant_required
VALIDATION_MESSAGE = "Für dieses Objekt muss ein Mandant angegeben werden."
def model_supports_tenant(model):
try:
field = model._meta.get_field("tenant")
except FieldDoesNotExist:
return False
return field.concrete and field.is_relation and field.related_model is Tenant
def validate_tenant_assignment(instance):
if tenant_required() and model_supports_tenant(type(instance)) and getattr(instance, "tenant_id", None) is None:
# A non-field error is intentional: a small number of specialized
# ModelForms omit the tenant field and Django cannot attach errors to
# fields which are absent from a form.
raise ValidationError(VALIDATION_MESSAGE)
def _install_model_clean_validation(model):
if not model_supports_tenant(model) or getattr(model, "_netbox_utilities_tenant_validation", False):
return
original_clean = model.clean
@wraps(original_clean)
def tenant_aware_clean(instance, *args, **kwargs):
result = original_clean(instance, *args, **kwargs)
validate_tenant_assignment(instance)
return result
model.clean = tenant_aware_clean
model._netbox_utilities_tenant_validation = True
def _install_form_validation():
if getattr(BaseModelForm, "_netbox_utilities_tenant_validation", False):
return
original_init = BaseModelForm.__init__
@wraps(original_init)
def tenant_aware_init(form, *args, **kwargs):
original_init(form, *args, **kwargs)
model = getattr(form._meta, "model", None)
if tenant_required() and model and model_supports_tenant(model) and "tenant" in form.fields:
form.fields["tenant"].required = True
BaseModelForm.__init__ = tenant_aware_init
BaseModelForm._netbox_utilities_tenant_validation = True
def _enforce_before_save(sender, instance, raw=False, **kwargs):
if not raw:
validate_tenant_assignment(instance)
def install_tenant_validation():
for model in apps.get_models():
_install_model_clean_validation(model)
_install_form_validation()
pre_save.connect(
_enforce_before_save,
dispatch_uid="netbox_utilities.enforce_tenant_before_save",
weak=False,
)
netbox_utilities/tests/test_middleware.py
+40 -3
View File
@@ -4,6 +4,7 @@ from unittest.mock import patch
from django.test import RequestFactory, SimpleTestCase
from netbox_utilities.middleware import GlobalTenantFilterMiddleware
from netbox_utilities.tenant_scope import ActiveTenantScope
class TenantFilterSet:
@@ -32,7 +33,8 @@ class GlobalTenantFilterMiddlewareTest(SimpleTestCase):
mocked_resolve.return_value = resolver_match(TenantFilterSet)
request = self.factory.get("/dcim/devices/?tenant_id=7&status=active")
GlobalTenantFilterMiddleware._inject_filter_parameter(request, 42)
scope = ActiveTenantScope("tenant", 42, frozenset({42}))
GlobalTenantFilterMiddleware._inject_filter_parameter(request, scope)
self.assertEqual(request.GET.getlist("tenant_id"), ["42"])
self.assertEqual(request.GET["status"], "active")
@@ -42,7 +44,8 @@ class GlobalTenantFilterMiddlewareTest(SimpleTestCase):
mocked_resolve.return_value = resolver_match(SharedFilterSet)
request = self.factory.get("/dcim/manufacturers/?status=active")
GlobalTenantFilterMiddleware._inject_filter_parameter(request, 42)
scope = ActiveTenantScope("tenant", 42, frozenset({42}))
GlobalTenantFilterMiddleware._inject_filter_parameter(request, scope)
self.assertNotIn("tenant_id", request.GET)
self.assertEqual(request.GET["status"], "active")
@@ -52,6 +55,40 @@ class GlobalTenantFilterMiddlewareTest(SimpleTestCase):
mocked_resolve.return_value = resolver_match(None, "tenancy:tenant_list")
request = self.factory.get("/tenancy/tenants/")
GlobalTenantFilterMiddleware._inject_filter_parameter(request, 42)
scope = ActiveTenantScope("tenant", 42, frozenset({42}))
GlobalTenantFilterMiddleware._inject_filter_parameter(request, scope)
self.assertEqual(request.GET.getlist("id"), ["42"])
@patch("netbox_utilities.middleware.resolve")
def test_injects_tenant_group(self, mocked_resolve):
class TenantGroupFilterSet:
base_filters = {"tenant_id": object(), "tenant_group_id": object()}
mocked_resolve.return_value = resolver_match(TenantGroupFilterSet)
request = self.factory.get("/dcim/devices/")
scope = ActiveTenantScope("group", 8, frozenset({42, 43}))
GlobalTenantFilterMiddleware._inject_filter_parameter(request, scope)
self.assertEqual(request.GET.getlist("tenant_group_id"), ["8"])
@patch("netbox_utilities.middleware.resolve")
def test_group_falls_back_to_tenant_ids(self, mocked_resolve):
mocked_resolve.return_value = resolver_match(TenantFilterSet)
request = self.factory.get("/dcim/virtual-chassis/")
scope = ActiveTenantScope("group", 8, frozenset({42, 43}))
GlobalTenantFilterMiddleware._inject_filter_parameter(request, scope)
self.assertCountEqual(request.GET.getlist("tenant_id"), ["42", "43"])
@patch("netbox_utilities.middleware.resolve")
def test_tenant_group_list_includes_descendant_groups(self, mocked_resolve):
mocked_resolve.return_value = resolver_match(None, "tenancy:tenantgroup_list")
request = self.factory.get("/tenancy/tenant-groups/")
scope = ActiveTenantScope("group", 8, frozenset({42}), frozenset({8, 9}))
GlobalTenantFilterMiddleware._inject_filter_parameter(request, scope)
self.assertCountEqual(request.GET.getlist("id"), ["8", "9"])
netbox_utilities/tests/test_tenant_scope.py
+6 -1
View File
@@ -2,7 +2,7 @@ from types import SimpleNamespace
from django.test import SimpleTestCase
from netbox_utilities.tenant_scope import object_matches_tenant
from netbox_utilities.tenant_scope import object_matches_tenant, object_matches_tenant_ids
class FakeMeta:
@@ -32,3 +32,8 @@ class TenantScopeTest(SimpleTestCase):
def test_unassigned_tenant_object_is_hidden(self):
obj = SimpleNamespace(_meta=FakeMeta("tenant"), tenant_id=None)
self.assertFalse(object_matches_tenant(obj, 42))
def test_object_matches_any_tenant_in_group_scope(self):
obj = SimpleNamespace(_meta=FakeMeta("tenant"), tenant_id=43)
self.assertTrue(object_matches_tenant_ids(obj, frozenset({42, 43})))
self.assertFalse(object_matches_tenant_ids(obj, frozenset({7, 8})))
netbox_utilities/tests/test_tenant_validation.py
+50
View File
@@ -0,0 +1,50 @@
from types import SimpleNamespace
from unittest.mock import patch
from django.core.exceptions import FieldDoesNotExist, ValidationError
from django.test import SimpleTestCase
from tenancy.models import Tenant
from netbox_utilities.tenant_validation import model_supports_tenant, validate_tenant_assignment
class FakeMeta:
def __init__(self, supports_tenant=True):
self.supports_tenant = supports_tenant
def get_field(self, name):
if name != "tenant" or not self.supports_tenant:
raise FieldDoesNotExist(name)
return SimpleNamespace(concrete=True, is_relation=True, related_model=Tenant)
class TenantModel:
_meta = FakeMeta()
def __init__(self, tenant_id=None):
self.tenant_id = tenant_id
class SharedModel:
_meta = FakeMeta(supports_tenant=False)
class TenantValidationTest(SimpleTestCase):
def test_detects_tenant_capable_model(self):
self.assertTrue(model_supports_tenant(TenantModel))
self.assertFalse(model_supports_tenant(SharedModel))
@patch("netbox_utilities.tenant_validation.tenant_required", return_value=True)
def test_rejects_missing_tenant_when_required(self, _tenant_required):
with self.assertRaises(ValidationError) as context:
validate_tenant_assignment(TenantModel())
self.assertEqual(context.exception.messages, ["Für dieses Objekt muss ein Mandant angegeben werden."])
@patch("netbox_utilities.tenant_validation.tenant_required", return_value=True)
def test_accepts_assigned_tenant(self, _tenant_required):
validate_tenant_assignment(TenantModel(tenant_id=42))
@patch("netbox_utilities.tenant_validation.tenant_required", return_value=False)
def test_allows_missing_tenant_when_disabled(self, _tenant_required):
validate_tenant_assignment(TenantModel())
netbox_utilities/views.py
+45 -14
View File
@@ -6,10 +6,14 @@ from django.urls import reverse
from django.utils.http import url_has_allowed_host_and_scheme
from django.views import View
from netbox.plugins import get_plugin_config
from tenancy.models import Tenant
from tenancy.models import Tenant, TenantGroup
from .forms import UtilitiesSettingsForm
from .middleware import SESSION_TENANT_KEY
from .middleware import (
SESSION_TENANT_GROUP_KEY,
SESSION_TENANT_KEY,
GlobalTenantFilterMiddleware,
)
from .models import NavigationPreference, UtilitiesSettings
from .navigation_helpers import get_visible_menus, normalize_preferences
from .runtime import (
@@ -87,25 +91,35 @@ class UtilitiesSettingsView(LoginRequiredMixin, UserPassesTestMixin, View):
def get(self, request):
instance, _ = UtilitiesSettings.objects.get_or_create(singleton_id=1)
tenant_filter_locked = not get_plugin_config("netbox_utilities", "tenant_filter_enabled")
tenant_required_locked = not get_plugin_config("netbox_utilities", "tenant_required")
return render(
request,
self.template_name,
{
"form": UtilitiesSettingsForm(instance=instance),
"config_forces_disabled": not get_plugin_config("netbox_utilities", "tenant_filter_enabled"),
"form": UtilitiesSettingsForm(
instance=instance,
tenant_filter_locked=tenant_filter_locked,
tenant_required_locked=tenant_required_locked,
),
"tenant_filter_locked": tenant_filter_locked,
"tenant_required_locked": tenant_required_locked,
},
)
def post(self, request):
if not get_plugin_config("netbox_utilities", "tenant_filter_enabled"):
return HttpResponseBadRequest("Der Mandantenfilter ist in PLUGINS_CONFIG deaktiviert.")
instance, _ = UtilitiesSettings.objects.get_or_create(singleton_id=1)
form = UtilitiesSettingsForm(request.POST, instance=instance)
form = UtilitiesSettingsForm(
request.POST,
instance=instance,
tenant_filter_locked=not get_plugin_config("netbox_utilities", "tenant_filter_enabled"),
tenant_required_locked=not get_plugin_config("netbox_utilities", "tenant_required"),
)
if form.is_valid():
form.save()
clear_runtime_settings_cache()
if not form.cleaned_data["tenant_filter_enabled"]:
request.session.pop(SESSION_TENANT_KEY, None)
if not tenant_filter_enabled():
GlobalTenantFilterMiddleware.clear_scope(request)
messages.success(request, "Die Plugin-Einstellungen wurden gespeichert.")
return redirect("plugins:netbox_utilities:settings")
return render(request, self.template_name, {"form": form})
@@ -118,15 +132,32 @@ class SelectTenantView(LoginRequiredMixin, View):
if not tenant_filter_enabled():
return HttpResponseBadRequest("Der globale Mandantenfilter ist deaktiviert.")
tenant_id = request.POST.get("tenant")
if tenant_id:
tenant = Tenant.objects.restrict(request.user, "view").filter(pk=tenant_id).first()
scope_value = request.POST.get("scope") or request.POST.get("tenant", "")
if not scope_value:
GlobalTenantFilterMiddleware.clear_scope(request)
messages.success(request, "Der globale Mandantenfilter wurde aufgehoben.")
return redirect(_safe_return_url(request))
if ":" in scope_value:
scope_kind, object_id = scope_value.split(":", 1)
else:
scope_kind, object_id = "tenant", scope_value
if scope_kind == "tenant":
tenant = Tenant.objects.restrict(request.user, "view").filter(pk=object_id).first()
if tenant is None:
return HttpResponseBadRequest("Unbekannter oder nicht erlaubter Mandant.")
request.session.pop(SESSION_TENANT_GROUP_KEY, None)
request.session[SESSION_TENANT_KEY] = tenant.pk
messages.success(request, f"Globaler Mandantenfilter: {tenant}")
else:
elif scope_kind == "group":
group = TenantGroup.objects.restrict(request.user, "view").filter(pk=object_id).first()
if group is None:
return HttpResponseBadRequest("Unbekannte oder nicht erlaubte Mandantengruppe.")
request.session.pop(SESSION_TENANT_KEY, None)
messages.success(request, "Der globale Mandantenfilter wurde aufgehoben.")
request.session[SESSION_TENANT_GROUP_KEY] = group.pk
messages.success(request, f"Globale Mandantengruppe: {group}")
else:
return HttpResponseBadRequest("Ungültiger Mandantenfilter.")
return redirect(_safe_return_url(request))