[ENG-7291] Update Contributor filter to filter on permissions

api/base/views.py

@@ -503,6 +503,7 @@ def get_object(self):
 
 
 class BaseContributorList(JSONAPIBaseView, generics.ListAPIView, ListFilterMixin):
+    DEFAULT_OPERATORS = ('eq', 'ne', 'exact')
 
     ordering = ('-user__modified',)
 
@@ -533,23 +534,37 @@ def postprocess_query_param(self, key, field_name, operation):
 
     def build_query_from_field(self, field_name, operation):
         if field_name == 'permission':
-            if operation['op'] != 'eq':
-                raise InvalidFilterOperator(value=operation['op'], valid_operators=['eq'])
+            if operation['op'] not in ['eq', 'exact']:
+                raise InvalidFilterOperator(value=operation['op'], valid_operators=['eq', 'exact'])
+
             # operation['value'] should be 'admin', 'write', or 'read'
             query_val = operation['value'].lower().strip()
             if query_val not in API_CONTRIBUTOR_PERMISSIONS:
                 raise InvalidFilterValue(value=operation['value'])
             # This endpoint should only be returning *contributors* not group members
             resource = self.get_resource()
-            if query_val == READ:
-                # If read, return all contributors
-                return Q(user_id__in=resource.contributors.values_list('id', flat=True))
-            elif query_val == WRITE:
-                # If write, return members of write and admin groups, both groups have write perms
-                return Q(user_id__in=(resource.get_group(WRITE).user_set.values_list('id', flat=True) | resource.get_group(ADMIN).user_set.values_list('id', flat=True)))
-            elif query_val == ADMIN:
-                # If admin, return only members of admin group
-                return Q(user_id__in=resource.get_group(ADMIN).user_set.values_list('id', flat=True))
+            if operation['op'] == 'eq':
+                if query_val == READ:
+                    # If read, return all contributors
+                    return Q(user_id__in=resource.contributors.values_list('id', flat=True))
+                elif query_val == WRITE:
+                    # If write, return members of write and admin groups, both groups have write perms
+                    write_ids = resource.get_group(WRITE).user_set.values_list('id', flat=True)
+                    admin_ids = resource.get_group(ADMIN).user_set.values_list('id', flat=True)
+                    return Q(user_id__in=(write_ids | admin_ids))
+                elif query_val == ADMIN:
+                    # If admin, return only members of admin group
+                    return Q(user_id__in=resource.get_group(ADMIN).user_set.values_list('id', flat=True))
+            elif operation['op'] == 'exact':
+                if query_val == READ:
+                    # If read, return only members of read group
+                    return Q(user_id__in=resource.get_group(READ).user_set.values_list('id', flat=True))
+                elif query_val == WRITE:
+                    # If write, return only members of write group
+                    return Q(user_id__in=resource.get_group(WRITE).user_set.values_list('id', flat=True))
+                elif query_val == ADMIN:
+                    # If admin, return only members of admin group
+                    return Q(user_id__in=resource.get_group(ADMIN).user_set.values_list('id', flat=True))
         return super().build_query_from_field(field_name, operation)
 
 

api_tests/nodes/views/test_node_contributors_list.py

@@ -2880,24 +2880,6 @@ def test_filtering(self, app, user, url, project):
         user_three = AuthUserFactory()
         project.add_contributor(user_two, permissions.WRITE)
         project.add_contributor(user_three, permissions.READ, visible=False)
-        #   test_filtering_permission_field_admin
-        filter_url = f'{url}?filter[permission]=admin'
-        res = app.get(filter_url, auth=user.auth, expect_errors=True)
-        assert res.status_code == 200
-        assert len(res.json['data']) == 1
-        assert res.json['data'][0]['attributes'].get('permission') == permissions.ADMIN
-
-        #   test_filtering_permission_field_write
-        filter_url = f'{url}?filter[permission]=write'
-        res = app.get(filter_url, auth=user.auth, expect_errors=True)
-        assert res.status_code == 200
-        assert len(res.json['data']) == 2
-
-        #   test_filtering_permission_field_read
-        filter_url = f'{url}?filter[permission]=read'
-        res = app.get(filter_url, auth=user.auth, expect_errors=True)
-        assert res.status_code == 200
-        assert len(res.json['data']) == 3
 
         #   test_filtering_node_with_only_bibliographic_contributors
         # no filter
@@ -2950,3 +2932,64 @@ def test_filtering_node_with_non_bibliographic_contributor(
         res = app.get(filter_url, auth=user.auth)
         assert len(res.json['data']) == 1
         assert not res.json['data'][0]['attributes'].get('bibliographic', None)
+
+    def test_filtering_permission_field_admin(self, app, user, project, url):
+
+        user_two = AuthUserFactory()
+        user_three = AuthUserFactory()
+        project.add_contributor(user_two, permissions.WRITE)
+        project.add_contributor(user_three, permissions.READ, visible=False)
+
+        # test_filtering_permission_field_admin
+        filter_url = f'{url}?filter[permission]=admin'
+        res = app.get(filter_url, auth=user.auth, expect_errors=True)
+        assert res.status_code == 200
+        assert len(res.json['data']) == 1
+        assert res.json['data'][0]['attributes'].get('permission') == permissions.ADMIN
+
+        # test filtering permission exact admin
+        filter_url = f'{url}?filter[permission][exact]=admin'
+        res = app.get(filter_url, auth=user.auth, expect_errors=True)
+        assert res.status_code == 200
+        assert len(res.json['data']) == 1
+        assert res.json['data'][0]['attributes'].get('permission') == permissions.ADMIN
+
+    def test_filtering_permission_field_write(self, app, user, project, url):
+
+        user_two = AuthUserFactory()
+        user_three = AuthUserFactory()
+        project.add_contributor(user_two, permissions.WRITE)
+        project.add_contributor(user_three, permissions.READ, visible=False)
+
+        # test_filtering_permission_field_write
+        filter_url = f'{url}?filter[permission]=write'
+        res = app.get(filter_url, auth=user.auth, expect_errors=True)
+        assert res.status_code == 200
+        assert len(res.json['data']) == 2
+
+        # test filtering permission exact write
+        filter_url = f'{url}?filter[permission][exact]=write'
+        res = app.get(filter_url, auth=user.auth, expect_errors=True)
+        assert res.status_code == 200
+        assert len(res.json['data']) == 1
+        assert res.json['data'][0]['attributes'].get('permission') == permissions.WRITE
+
+    def test_filtering_permission_field_read(self, app, user, project, url):
+
+        user_two = AuthUserFactory()
+        user_three = AuthUserFactory()
+        project.add_contributor(user_two, permissions.WRITE)
+        project.add_contributor(user_three, permissions.READ, visible=False)
+
+        # test_filtering_permission_field_read
+        filter_url = f'{url}?filter[permission]=read'
+        res = app.get(filter_url, auth=user.auth, expect_errors=True)
+        assert res.status_code == 200
+        assert len(res.json['data']) == 3
+
+        # test filtering permission exact read
+        filter_url = f'{url}?filter[permission][exact]=read'
+        res = app.get(filter_url, auth=user.auth, expect_errors=True)
+        assert res.status_code == 200
+        assert len(res.json['data']) == 1
+        assert res.json['data'][0]['attributes'].get('permission') == permissions.READ