OBS-1202 fix permissions (#115)

Author: arthanson

docker/requirements-diode-netbox-plugin.txt

@@ -4,4 +4,4 @@ coverage==7.6.0
 grpcio==1.62.1
 protobuf==5.28.1
 pytest==8.0.2
-netboxlabs-netbox-branching
+netboxlabs-netbox-branching==0.5.7

netbox_diode_plugin/api/applier.py

@@ -101,7 +101,7 @@ def _pre_apply(model_class: models.Model, change: Change, created: dict):
     # resolve foreign key references to new objects
     for ref_field in change.new_refs:
         v = _get_path(data, ref_field)
-        if isinstance(v, (list, tuple)):
+        if isinstance(v, list | tuple):
             ref_list = []
             for ref in v:
                 if isinstance(ref, str):

netbox_diode_plugin/api/common.py

@@ -9,6 +9,7 @@
 from collections import defaultdict
 from dataclasses import dataclass, field
 from enum import Enum
+from zoneinfo import ZoneInfo
 
 import netaddr
 from django.apps import apps
@@ -20,7 +21,6 @@
 from extras.models import CustomField
 from netaddr.eui import EUI
 from rest_framework import status
-from zoneinfo import ZoneInfo
 
 logger = logging.getLogger("netbox.diode_data")
 
@@ -166,7 +166,7 @@ def _validate_relations(self, change_data: dict, model: models.Model) -> tuple[l
         excluded_relation_fields = []
         rel_errors = defaultdict(list)
         for f in model._meta.get_fields():
-            if isinstance(f, (GenericRelation, GenericForeignKey)):
+            if isinstance(f, GenericRelation | GenericForeignKey):
                 excluded_relation_fields.append(f.name)
                 continue
             if not f.is_relation:
@@ -251,7 +251,7 @@ def error_from_validation_error(e, object_name):
     if e.detail:
         if isinstance(e.detail, dict):
             errors[object_name] = e.detail
-        elif isinstance(e.detail, (list, tuple)):
+        elif isinstance(e.detail, list | tuple):
             errors[object_name] = {
                 NON_FIELD_ERRORS: e.detail
             }

netbox_diode_plugin/api/differ.py

@@ -84,7 +84,7 @@ def prechange_data_from_instance(instance) -> dict: # noqa: C901
         custom_field_values = instance.get_custom_fields()
         cfmap = {}
         for cf, value in custom_field_values.items():
-            if isinstance(value, (datetime.datetime, datetime.date)):
+            if isinstance(value, datetime.datetime | datetime.date):
                 cfmap[cf.name] = value
             else:
                 cfmap[cf.name] = cf.serialize(value)

netbox_diode_plugin/api/matcher.py

@@ -5,7 +5,6 @@
 import logging
 from dataclasses import dataclass
 from functools import cache, lru_cache
-from typing import Type
 
 import netaddr
 from django.contrib.contenttypes.fields import ContentType
@@ -221,7 +220,7 @@ class ObjectMatchCriteria:
     fields: tuple[str] | None = None
     expressions: tuple | None = None
     condition: Q | None = None
-    model_class: Type[models.Model] | None = None
+    model_class: type[models.Model] | None = None
     name: str | None = None
 
     def __hash__(self):
@@ -365,7 +364,7 @@ def _build_expressions_queryset(self, data) -> models.QuerySet:
         """Builds a queryset for the constraint with the given data."""
         data = self._prepare_data(data)
         replacements = {
-            F(field): Value(value) if isinstance(value, (str, int, float, bool)) else value
+            F(field): Value(value) if isinstance(value, str | int | float | bool) else value
             for field, value in data.items()
         }
 
@@ -413,7 +412,7 @@ class CustomFieldMatcher:
 
     name: str
     custom_field: str
-    model_class: Type[models.Model]
+    model_class: type[models.Model]
 
     def fingerprint(self, data: dict) -> str|None:
         """Fingerprint the custom field value."""
@@ -448,7 +447,7 @@ class GlobalIPNetworkIPMatcher:
 
     ip_fields: tuple[str]
     vrf_field: str
-    model_class: Type[models.Model]
+    model_class: type[models.Model]
     name: str
 
     def _check_condition(self, data: dict) -> bool:
@@ -508,7 +507,7 @@ class VRFIPNetworkIPMatcher:
 
     ip_fields: tuple[str]
     vrf_field: str
-    model_class: Type[models.Model]
+    model_class: type[models.Model]
     name: str
 
     def _check_condition(self, data: dict) -> bool:
@@ -583,7 +582,7 @@ class AutoSlugMatcher:
 
     name: str
     slug_field: str
-    model_class: Type[models.Model]
+    model_class: type[models.Model]
 
     def fingerprint(self, data: dict) -> str|None:
         """Fingerprint the custom field value."""
@@ -750,7 +749,7 @@ def _fingerprint_all(data: dict, object_type: str|None = None) -> str:
         if k.startswith("_"):
             continue
         values.append(k)
-        if isinstance(v, (list, tuple)):
+        if isinstance(v, list | tuple):
             values.extend(sorted(v))
         elif isinstance(v, dict):
             values.append(_fingerprint_all(v))

netbox_diode_plugin/api/supported_models.py

@@ -6,7 +6,6 @@
 import logging
 import time
 from functools import lru_cache
-from typing import List, Type
 
 from django.apps import apps
 from django.db import models
@@ -82,9 +81,9 @@ def extract_supported_models() -> dict[str, dict]:
     return extracted_models
 
 
-def get_prerequisites(model_class, fields) -> List[dict[str, str]]:
+def get_prerequisites(model_class, fields) -> list[dict[str, str]]:
     """Get the prerequisite models for the model."""
-    prerequisites: List[dict[str, str]] = []
+    prerequisites: list[dict[str, str]] = []
     prerequisite_models = getattr(model_class, "prerequisite_models", [])
 
     for prereq in prerequisite_models:
@@ -252,7 +251,7 @@ def get_serializer_for_model(model, prefix=""):
     return netbox_get_serializer_for_model(model, prefix)
 
 
-def discover_models(root_packages: List[str]) -> list[Type[models.Model]]:
+def discover_models(root_packages: list[str]) -> list[type[models.Model]]:
     """Discovers all model classes in specified root packages."""
     discovered_models = []
 

netbox_diode_plugin/api/transformer.py

@@ -4,14 +4,14 @@
 
 import copy
 import datetime
+import graphlib
 import json
 import logging
 import re
 from collections import defaultdict
 from functools import lru_cache
 from uuid import uuid4
 
-import graphlib
 from django.utils.text import slugify
 from extras.models.customfields import CustomField
 from rest_framework import serializers
@@ -474,7 +474,7 @@ def _update_dict_refs(data, new_refs):
     for k, v in data.items():
         if isinstance(v, UnresolvedReference) and v.uuid in new_refs:
             v.uuid = new_refs[v.uuid]
-        elif isinstance(v, (list, tuple)):
+        elif isinstance(v, list | tuple):
             for item in v:
                 if isinstance(item, UnresolvedReference) and item.uuid in new_refs:
                     item.uuid = new_refs[item.uuid]
@@ -517,7 +517,7 @@ def _update_resolved_refs(data, new_refs):
     for k, v in list(data.items()):
         if isinstance(v, UnresolvedReference) and v.uuid in new_refs:
             data[k] = new_refs[v.uuid]
-        elif isinstance(v, (list, tuple)):
+        elif isinstance(v, list | tuple):
             new_items = []
             has_refs = False
             for item in v:
@@ -539,7 +539,7 @@ def cleanup_unresolved_references(data: dict) -> list[str]:
             if k != 'id':
                 unresolved.add(k)
             data[k] = str(v)
-        elif isinstance(v, (list, tuple)):
+        elif isinstance(v, list | tuple):
             items = []
             for item in v:
                 if isinstance(item, UnresolvedReference):

netbox_diode_plugin/navigation.py

@@ -9,12 +9,12 @@
     PluginMenuItem(
         link="plugins:netbox_diode_plugin:settings",
         link_text=_("Settings"),
-        staff_only= True,
+        permissions=("netbox_diode_plugin.view_setting",),
     ),
     PluginMenuItem(
         link="plugins:netbox_diode_plugin:client_credential_list",
         link_text=_("Client Credentials"),
-        staff_only= True,
+        permissions=("netbox_diode_plugin.view_clientcredentials",),
     ),
 )
 

netbox_diode_plugin/tests/test_updates.py

@@ -26,7 +26,7 @@ def _harmonize_formats(data):
     return _tuples_to_lists(data)
 
 def _tuples_to_lists(data):
-    if isinstance(data, (tuple, list)):
+    if isinstance(data, tuple | list):
         return [_tuples_to_lists(d) for d in data]
     if isinstance(data, dict):
         return {k: _tuples_to_lists(v) for k, v in data.items()}
@@ -165,7 +165,7 @@ def _check_set_by(self, obj, path, value):
         path = path[:-1]
         cur = self._follow_path(obj, path)
 
-        if isinstance(value, (list, tuple)):
+        if isinstance(value, list | tuple):
             vals = set(value)
         else:
             vals = {value}

netbox_diode_plugin/tests/test_views.py

@@ -8,16 +8,32 @@
 from django.contrib.messages.middleware import MessageMiddleware
 from django.contrib.messages.storage.fallback import FallbackStorage
 from django.contrib.sessions.middleware import SessionMiddleware
-from django.test import RequestFactory, TestCase
+from django.test import RequestFactory
+from django.test import TestCase as _TestCase
 from django.urls import reverse
 from rest_framework import status
+from users.models import ObjectPermission
+from utilities.permissions import resolve_permission_type
 
 from netbox_diode_plugin.models import Setting
 from netbox_diode_plugin.views import SettingsEditView, SettingsView
 
 User = get_user_model()
 
 
+class TestCase(_TestCase):
+    """Base test case class for NetBox Diode plugin tests."""
+
+    def add_permissions(self, user, *names):
+        """Assign a set of permissions to the test user. Accepts permission names in the form <app>.<action>_<model>."""
+        for name in names:
+            object_type, action = resolve_permission_type(name)
+            obj_perm = ObjectPermission(name=name, actions=[action])
+            obj_perm.save()
+            obj_perm.users.add(user)
+            obj_perm.object_types.add(object_type)
+
+
 class SettingsViewTestCase(TestCase):
     """Test case for the SettingsView."""
 
@@ -31,7 +47,7 @@ def setUp(self):
     def test_returns_200_for_authenticated(self):
         """Test that the view returns 200 for an authenticated user."""
         self.request.user = User.objects.create_user("foo", password="pass")
-        self.request.user.is_staff = True
+        self.add_permissions(self.request.user, "netbox_diode_plugin.view_setting")
 
         response = self.view.get(self.request)
         self.assertEqual(response.status_code, status.HTTP_200_OK)
@@ -49,7 +65,7 @@ def test_redirects_to_login_page_for_unauthenticated_user(self):
     def test_settings_created_if_not_found(self):
         """Test that the settings are created with placeholder data if not found."""
         self.request.user = User.objects.create_user("foo", password="pass")
-        self.request.user.is_staff = True
+        self.add_permissions(self.request.user, "netbox_diode_plugin.view_setting")
 
         with mock.patch("netbox_diode_plugin.models.Setting.objects.get") as mock_get:
             mock_get.side_effect = Setting.DoesNotExist
@@ -71,8 +87,8 @@ def setUp(self):
     def test_returns_200_for_authenticated(self):
         """Test that the view returns 200 for an authenticated user."""
         request = self.request_factory.get(self.path)
-        request.user = User.objects.create_user("foo", password="pass")
-        request.user.is_staff = True
+        request.user = User.objects.create_user("foo", password="pass", is_staff=True)
+        self.add_permissions(request.user, "netbox_diode_plugin.view_setting", "netbox_diode_plugin.change_setting")
         request.htmx = None
         self.view.setup(request)
 
@@ -91,8 +107,8 @@ def test_redirects_to_login_page_for_unauthenticated_user(self):
 
     def test_settings_updated(self):
         """Test that the settings are updated."""
-        user = User.objects.create_user("foo", password="pass")
-        user.is_staff = True
+        user = User.objects.create_user("foo", password="pass", is_staff=True)
+        self.add_permissions(user, "netbox_diode_plugin.view_setting", "netbox_diode_plugin.change_setting")
 
         request = self.request_factory.get(self.path)
         request.user = user
@@ -149,8 +165,13 @@ def test_settings_update_disallowed_on_get_method(self):
         ) as mock_get_plugin_config:
             mock_get_plugin_config.return_value = "grpc://localhost:8080/diode"
 
-            user = User.objects.create_user("foo", password="pass")
-            user.is_staff = True
+            user = User.objects.create_user("foo", password="pass", is_staff=True)
+            self.add_permissions(
+                user,
+                "netbox_diode_plugin.view_setting",
+                "netbox_diode_plugin.add_setting",
+                "netbox_diode_plugin.change_setting",
+            )
 
             request = self.request_factory.post(self.path)
             request.user = user
@@ -188,8 +209,13 @@ def test_settings_update_disallowed_on_post_method(self):
         ) as mock_get_plugin_config:
             mock_get_plugin_config.return_value = "grpc://localhost:8080/diode"
 
-            user = User.objects.create_user("foo", password="pass")
-            user.is_staff = True
+            user = User.objects.create_user("foo", password="pass", is_staff=True)
+            self.add_permissions(
+                user,
+                "netbox_diode_plugin.view_setting",
+                "netbox_diode_plugin.add_setting",
+                "netbox_diode_plugin.change_setting",
+            )
 
             request = self.request_factory.post(self.path)
             request.user = user