[change] Temporarily import logic from openwisp_users

openwisp_ipam/api/serializers.py

@@ -1,9 +1,8 @@
+from openwisp_users.api.mixins import FilterSerializerByOrgManaged
 from openwisp_utils.api.serializers import ValidatedModelSerializer
 from rest_framework import serializers
 from swapper import load_model
 
-from .utils import FilterSerializerByOrgManaged
-
 IpAddress = load_model('openwisp_ipam', 'IpAddress')
 Subnet = load_model('openwisp_ipam', 'Subnet')
 

openwisp_ipam/api/utils.py

@@ -1,60 +1,10 @@
 import swapper
 from django.contrib.auth.models import Permission
-from django.core.exceptions import ValidationError
-from rest_framework.exceptions import NotFound, PermissionDenied
+from rest_framework.exceptions import PermissionDenied
 
 Organization = swapper.load_model('openwisp_users', 'Organization')
 
 
-class FilterByOrganization:
-    def get_queryset(self):
-        qs = super().get_queryset()
-        # superuser has access to every organization
-        if self.request.user.is_superuser:
-            return qs
-        # non superuser has access only to some organizations
-        return self.get_organization_queryset(qs)
-
-    def get_organization_queryset(self):
-        raise NotImplementedError()
-
-
-class FilterByOrganizationManaged(FilterByOrganization):
-    """
-    Allows to filter only organizations which the current user manages
-    """
-
-    def get_organization_queryset(self, qs):
-        return qs.filter(organization__in=self.request.user.organizations_managed)
-
-
-class FilterByParent:
-    def get_queryset(self):
-        qs = super().get_queryset()
-        self.assert_parent_exists()
-        return qs
-
-    def assert_parent_exists(self):
-        parent_queryset = self.get_parent_queryset()
-        if not self.request.user.is_superuser:
-            parent_queryset = self.get_organization_queryset(parent_queryset)
-        try:
-            assert parent_queryset.exists()
-        except (AssertionError, ValidationError):
-            raise NotFound(detail='No relevant data found.')
-
-    def get_parent_queryset(self):
-        raise NotImplementedError()
-
-    def get_organization_queryset(self):
-        raise NotImplementedError()
-
-
-class FilterByParentManaged(FilterByParent):
-    def get_organization_queryset(self, qs):
-        return qs.filter(organization__in=self.request.user.organizations_managed)
-
-
 class AuthorizeCSVImport:
     def assert_organization_permissions(self, request):
         if request.user.is_superuser:
@@ -83,32 +33,3 @@ def get_user_organizations(self):
 class AuthorizeCSVOrgManaged(AuthorizeCSVImport):
     def get_user_organizations(self):
         return self.request.user.organizations_managed
-
-
-class FilterSerializerByOrganization:
-    def __init__(self, *args, **kwargs):
-        super().__init__(*args, **kwargs)
-        if self.context['request'].user.is_superuser:
-            return
-        self.filter_fields()
-
-    def filter_fields(self):
-        raise NotImplementedError()
-
-
-class FilterSerializerByOrgManaged(FilterSerializerByOrganization):
-    def filter_fields(self):
-        user = self.context['request'].user
-        organization_filter = user.organizations_managed
-        for field in self.fields:
-            if field == 'organization':
-                self.fields[field].queryset = self.fields[field].queryset.filter(
-                    pk__in=organization_filter
-                )
-                continue
-            try:
-                self.fields[field].queryset = self.fields[field].queryset.filter(
-                    organization__in=organization_filter
-                )
-            except AttributeError:
-                pass

openwisp_ipam/api/views.py

@@ -6,6 +6,7 @@
 from django.http import HttpResponse
 from django.utils.translation import gettext_lazy as _
 from openwisp_users.api.authentication import BearerAuthentication
+from openwisp_users.api.mixins import FilterByOrganizationManaged, FilterByParentManaged
 from openwisp_users.api.permissions import IsOrganizationManager
 from rest_framework import pagination, serializers, status
 from rest_framework.authentication import SessionAuthentication
@@ -30,11 +31,7 @@
     IpRequestSerializer,
     SubnetSerializer,
 )
-from .utils import (
-    AuthorizeCSVOrgManaged,
-    FilterByOrganizationManaged,
-    FilterByParentManaged,
-)
+from .utils import AuthorizeCSVOrgManaged
 
 IpAddress = swapper.load_model('openwisp_ipam', 'IpAddress')
 Subnet = swapper.load_model('openwisp_ipam', 'Subnet')
@@ -178,7 +175,9 @@ def get_queryset(self):
         return subnet.ipaddress_set.all().order_by('ip_address')
 
 
-class SubnetListCreateView(FilterByOrganizationManaged, ProtectedAPIMixin, ListCreateAPIView):
+class SubnetListCreateView(
+    FilterByOrganizationManaged, ProtectedAPIMixin, ListCreateAPIView
+):
     serializer_class = SubnetSerializer
     pagination_class = ListViewPagination
     queryset = Subnet.objects.all()

openwisp_ipam/tests/test_multitenant.py

@@ -6,12 +6,7 @@
 from openwisp_users.tests.utils import TestMultitenantAdminMixin
 from swapper import load_model
 
-from openwisp_ipam.api.utils import (
-    AuthorizeCSVImport,
-    FilterByOrganization,
-    FilterByParent,
-    FilterSerializerByOrganization,
-)
+from openwisp_ipam.api.utils import AuthorizeCSVImport
 
 from . import CreateModelsMixin, PostDataMixin
 
@@ -430,21 +425,9 @@ def test_browsable_api_ipaddress_list(self):
             self.assertContains(response, '10.0.0.0/24</option>')
             self.assertContains(response, '10.10.0.0/24</option>')
 
-    def test_not_implemented_error(self):
-        with self.assertRaises(NotImplementedError):
-            FilterByOrganization.get_organization_queryset(self)
-
-        with self.assertRaises(NotImplementedError):
-            FilterByParent.get_parent_queryset(self)
-
-        with self.assertRaises(NotImplementedError):
-            FilterByParent.get_organization_queryset(self)
-
+    def test_authorize_csv_import_implementation_error(self):
         with self.assertRaises(NotImplementedError):
             AuthorizeCSVImport.get_csv_organization(self)
 
         with self.assertRaises(NotImplementedError):
             AuthorizeCSVImport.get_user_organizations(self)
-
-        with self.assertRaises(NotImplementedError):
-            FilterSerializerByOrganization.filter_fields(self)

requirements.txt

@@ -1,4 +1,5 @@
-openwisp-users~=0.5.1
+# TODO: change this when next version of openwisp_users is released
+openwisp-users @ https://github.com/purhan/openwisp-users/tarball/issues/210-add-filter-mixins
 openwisp-utils[rest]~=0.7.2
 django>=2.2,<3.2
 swapper~=1.1