more authz functionality impl

test: add more tests / fix test issues

chord_metadata_service/authz/permissions.py

@@ -79,13 +79,9 @@ async def view_request_has_data_type_permission(
     )
 
 
-async def _has_data_type_permission_obj(request: DrfRequest, view, data_type: str, obj: BaseScopeableModel) -> bool:
+async def _has_data_type_permission_obj(request: DrfRequest, view, obj: BaseScopeableModel) -> bool:
     scope = await _get_scope_for_request_and_api_view(request, view)
-
-    if not await obj.scope_contains_object_async(scope):
-        return False
-
-    return await view_request_has_data_type_permission(request, view, data_type, scope)
+    return await obj.scope_contains_object_async(scope)
 
 
 class BentoPhenopacketDataPermission(BasePermission):
@@ -95,7 +91,9 @@ async def has_permission(self, request: DrfRequest, view):
 
     @async_to_sync
     async def has_object_permission(self, request, view, obj: BaseScopeableModel):
-        return await _has_data_type_permission_obj(request, view, DATA_TYPE_PHENOPACKET, obj)
+        # if this is called, has_data_type_permission has already been called and handled the overall action type
+        # TODO: eliminate duplicate scope check somehow without enabling permissions on objects outside of scope
+        return await _has_data_type_permission_obj(request, view, obj)
 
 
 class BentoExperimentDataPermission(BasePermission):
@@ -105,7 +103,9 @@ async def has_permission(self, request: DrfRequest, view):
 
     @async_to_sync
     async def has_object_permission(self, request, view, obj: BaseScopeableModel):
-        return await _has_data_type_permission_obj(request, view, DATA_TYPE_PHENOPACKET, obj)
+        # if this is called, has_data_type_permission has already been called and handled the overall action type
+        # TODO: eliminate duplicate scope check somehow without enabling permissions on objects outside of scope
+        return await _has_data_type_permission_obj(request, view, obj)
 
 
 class ReadOnly(BasePermission):

chord_metadata_service/authz/tests/helpers.py

@@ -2,7 +2,7 @@
 
 from aioresponses import aioresponses
 from bento_lib.auth.types import EvaluationResultMatrix
-from rest_framework.test import APITestCase
+from rest_framework.test import APITransactionTestCase
 from typing import Literal
 
 from ..types import DataPermissionsDict
@@ -28,7 +28,7 @@ def mock_authz_eval_result(m: aioresponses, result: EvaluationResultMatrix | lis
 DTAccessLevel = Literal["none", "bool", "counts", "full"]
 
 
-class AuthzAPITestCase(APITestCase):
+class AuthzAPITestCase(APITransactionTestCase):
     # data type permissions: bool, counts, data
     dt_none_eval_res = [[False, False, False]]
     dt_bool_eval_res = [[True, False, False]]

chord_metadata_service/chord/tests/helpers.py

@@ -1,5 +1,3 @@
-import json
-
 from django.db.models import Model
 from django.test import TestCase
 from django.urls import reverse

chord_metadata_service/chord/tests/test_api.py

@@ -222,7 +222,7 @@ def test_resources(self):
             "iri_prefix": "http://purl.obolibrary.org/obo/NCBITaxon_",
         }
 
-        r = self.client.post("/api/resources", data=json.dumps(resource), content_type="application/json")
+        r = self.one_authz_post("/api/resources", json=resource)
         self.assertEqual(r.status_code, status.HTTP_201_CREATED)
 
         r = self.one_authz_post(

chord_metadata_service/patients/api_views.py

@@ -85,6 +85,10 @@ class IndividualViewSet(viewsets.ModelViewSet):
     search_fields = ["sex"]
     lookup_value_regex = MODEL_ID_PATTERN
 
+    # We scope the queryset according to requested discovery scope below, which lets us have more fine-grained
+    # permissions.
+    scope_enabled = True
+
     @async_to_sync
     async def get_queryset(self):
         scope = await get_request_discovery_scope(self.request)

chord_metadata_service/patients/tests/test_api.py

@@ -34,7 +34,7 @@
 CONFIG_PUBLIC_TEST_NO_THRESHOLD["rules"]["count_threshold"] = 0
 
 
-class CreateIndividualTest(APITestCase):
+class CreateIndividualTest(AuthzAPITestCase):
     """ Test module for creating an Individual. """
 
     def setUp(self):
@@ -45,19 +45,19 @@ def setUp(self):
     def test_create_individual(self):
         """ POST a new individual. """
 
-        response = self.client.post(
-            reverse('individuals-list'),
-            data=json.dumps(self.valid_payload),
-            content_type='application/json'
-        )
+        response = self.one_authz_post(reverse('individuals-list'), json=self.valid_payload)
         self.assertEqual(response.status_code, status.HTTP_201_CREATED)
         self.assertEqual(Individual.objects.count(), 1)
         self.assertEqual(Individual.objects.get().id, 'patient:1')
 
+    def test_create_individual_forbidden(self):
+        response = self.one_no_authz_post(reverse('individuals-list'), json=self.valid_payload)
+        self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
+
     def test_create_invalid_individual(self):
         """ POST a new individual with invalid data. """
 
-        invalid_response = self.client.post(
+        invalid_response = self.one_authz_post(
             reverse('individuals-list'),
             data=json.dumps(self.invalid_payload),
             content_type='application/json'
@@ -66,7 +66,7 @@ def test_create_invalid_individual(self):
         self.assertEqual(Individual.objects.count(), 0)
 
 
-class UpdateIndividualTest(APITestCase):
+class UpdateIndividualTest(AuthzAPITestCase):
     """ Test module for updating an existing Individual record. """
 
     def setUp(self):
@@ -96,31 +96,30 @@ def setUp(self):
     def test_update_individual(self):
         """ PUT new data in an existing Individual record. """
 
-        response = self.client.put(
-            reverse(
-                'individuals-detail',
-                kwargs={'pk': self.individual_one.id}
-            ),
-            data=json.dumps(self.put_valid_payload),
-            content_type='application/json'
+        response = self.one_authz_put(
+            reverse('individuals-detail', kwargs={'pk': self.individual_one.id}),
+            json=self.put_valid_payload
         )
         self.assertEqual(response.status_code, status.HTTP_200_OK)
 
+    def test_update_individual_forbidden(self):
+        response = self.one_no_authz_put(
+            reverse('individuals-detail', kwargs={'pk': self.individual_one.id}),
+            json=self.put_valid_payload
+        )
+        self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
+
     def test_update_invalid_individual(self):
         """ PUT new invalid data in an existing Individual record. """
 
-        response = self.client.put(
-            reverse(
-                'individuals-detail',
-                kwargs={'pk': self.individual_one.id}
-            ),
-            data=json.dumps(self.invalid_payload),
-            content_type='application/json'
+        response = self.one_authz_put(
+            reverse('individuals-detail', kwargs={'pk': self.individual_one.id}),
+            json=self.invalid_payload,
         )
         self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
 
 
-class DeleteIndividualTest(APITestCase):
+class DeleteIndividualTest(AuthzAPITestCase):
     """ Test module for deleting an existing Individual record. """
 
     def setUp(self):
@@ -129,34 +128,34 @@ def setUp(self):
     def test_delete_individual(self):
         """ DELETE an existing Individual record. """
 
-        response = self.client.delete(
-            reverse(
-                'individuals-detail',
-                kwargs={'pk': self.individual_one.id}
-            )
+        response = self.one_authz_delete(
+            reverse('individuals-detail', kwargs={'pk': self.individual_one.id})
         )
         self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
 
+    def test_delete_individual_forbidden(self):
+        response = self.one_no_authz_delete(
+            reverse('individuals-detail', kwargs={'pk': self.individual_one.id})
+        )
+        self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
+
     def test_delete_non_existing_individual(self):
         """ DELETE a non-existing Individual record. """
 
-        response = self.client.delete(
-            reverse(
-                'individuals-detail',
-                kwargs={'pk': 'patient:what'}
-            )
+        response = self.one_authz_delete(
+            reverse('individuals-detail', kwargs={'pk': 'patient:what'})
         )
         self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
 
 
-class IndividualCSVRendererTest(APITestCase):
+class IndividualCSVRendererTest(AuthzAPITestCase):
     """ Test csv export for Individuals. """
 
     def setUp(self):
         self.individual_one = Individual.objects.create(**c.VALID_INDIVIDUAL)
 
     def test_csv_export(self):
-        get_resp = self.client.get('/api/individuals?format=csv')
+        get_resp = self.one_authz_get('/api/individuals?format=csv')
         self.assertEqual(get_resp.status_code, status.HTTP_200_OK)
         content = get_resp.content.decode('utf-8')
         cvs_reader = csv.reader(io.StringIO(content))
@@ -167,10 +166,21 @@ def test_csv_export(self):
                        'age', 'diseases', 'created', 'updated']:
             self.assertIn(column, [column_name.lower() for column_name in headers])
 
+    def test_csv_export_forbidden(self):
+        get_resp = self.one_no_authz_get('/api/individuals?format=csv')
+        self.assertEqual(get_resp.status_code, status.HTTP_403_FORBIDDEN)
 
-class IndividualWithPhenopacketSearchTest(APITestCase):
+
+class IndividualWithPhenopacketSearchTest(AuthzAPITestCase):
     """ Test for api/individuals?search= """
 
+    search_test_params = (
+        ("search=P49Y", 1, None),
+        ("search=NCBITaxon:9606", 2, None),
+        # 5 fields in the bento search response:
+        ("search=P49Y&format=bento_search_result", 1, 5),
+    )
+
     def setUp(self):
         self.individual_one = Individual.objects.create(**c.VALID_INDIVIDUAL)
         self.individual_two = Individual.objects.create(**c.VALID_INDIVIDUAL_2)
@@ -179,31 +189,30 @@ def setUp(self):
             **ph_c.valid_phenopacket(subject=self.individual_one, meta_data=self.metadata_1)
         )
 
-    def test_search(self):  # test full-text search
-        get_resp_1 = self.client.get('/api/individuals?search=P49Y')
-        self.assertEqual(get_resp_1.status_code, status.HTTP_200_OK)
-        response_obj_1 = get_resp_1.json()
-        self.assertEqual(len(response_obj_1['results']), 1)
-
-        get_resp_2 = self.client.get('/api/individuals?search=NCBITaxon:9606')
-        self.assertEqual(get_resp_2.status_code, status.HTTP_200_OK)
-        response_obj_2 = get_resp_2.json()
-        self.assertEqual(len(response_obj_2['results']), 2)
-
-    def test_search_bento_search_format(self):  # test full-text search - bento search result format
-        get_resp_1 = self.client.get('/api/individuals?search=P49Y&format=bento_search_result')
-        self.assertEqual(get_resp_1.status_code, status.HTTP_200_OK)
-        response_obj_1 = get_resp_1.json()
-        self.assertEqual(len(response_obj_1['results']), 1)
-        self.assertEqual(len(response_obj_1['results'][0]), 5)  # 5 fields in the bento search response
+    def test_search(self):  # test full-text search (standard + bento search format)
+        for params in self.search_test_params:
+            with self.subTest(params=params):
+                res = self.one_authz_get(f"/api/individuals?{params[0]}")
+                self.assertEqual(res.status_code, status.HTTP_200_OK)
+                res_data = res.json()
+                self.assertEqual(len(res_data["results"]), params[1])
+                if (n_keys := params[2]) is not None:
+                    self.assertEqual(len(res_data["results"][0]), n_keys)
+
+    def test_search_forbidden(self):
+        for params in self.search_test_params:
+            with self.subTest(params=params):
+                res = self.one_no_authz_get(f"/api/individuals?{params[0]}")
+                self.assertEqual(res.status_code, status.HTTP_403_FORBIDDEN)
 
     def test_individual_phenopackets(self):
-        get_resp = self.client.get(f"/api/individuals/{self.individual_one.id}/phenopackets")
+        get_resp = self.one_authz_get(f"/api/individuals/{self.individual_one.id}/phenopackets")
         self.assertEqual(get_resp.status_code, status.HTTP_200_OK)
         response_obj_1 = get_resp.json()
         self.assertEqual(len(response_obj_1), 1)  # 1 phenopacket for individual
 
-        post_resp = self.client.post(f"/api/individuals/{self.individual_one.id}/phenopackets?attachment=1")
+    def test_individual_phenopackets_attachment(self):
+        post_resp = self.one_authz_post(f"/api/individuals/{self.individual_one.id}/phenopackets?attachment=1")
         self.assertEqual(post_resp.status_code, status.HTTP_200_OK)
         self.assertIn("attachment; filename=", post_resp.headers.get("Content-Disposition", ""))
         response_obj_2 = post_resp.json()

chord_metadata_service/phenopackets/api_views.py

@@ -3,13 +3,13 @@
 from bento_lib.responses import errors
 from django_filters.rest_framework import DjangoFilterBackend
 from drf_spectacular.utils import extend_schema, inline_serializer
-from rest_framework import serializers, status, viewsets
+from rest_framework import mixins, serializers, status, viewsets
 from rest_framework.settings import api_settings
 from rest_framework.decorators import api_view, permission_classes
 from rest_framework.response import Response
 
 from chord_metadata_service.authz.middleware import authz_middleware
-from chord_metadata_service.authz.permissions import BentoPhenopacketDataPermission, BentoAllowAny
+from chord_metadata_service.authz.permissions import BentoPhenopacketDataPermission, BentoAllowAny, BentoDeferToHandler
 from chord_metadata_service.chord.data_types import DATA_TYPE_PHENOPACKET
 from chord_metadata_service.discovery.scope import get_request_discovery_scope
 from chord_metadata_service.restapi.api_renderers import (
@@ -139,12 +139,7 @@ class BiosampleBatchViewSet(viewsets.ModelViewSet):
         IndividualBentoSearchRenderer,
     )
     content_negotiation_class = FormatInPostContentNegotiation
-
-    # We scope the queryset according to requested discovery scope below, which lets us have more fine-grained
-    # permissions.
-    scope_enabled = True
-
-    # TODO: this shouldn't be its own separate viewset maybe...
+    permission_classes = (BentoDeferToHandler,)
 
     @async_to_sync
     async def _get_filtered_queryset(self, ids_list: list[str] | None = None):
@@ -167,6 +162,11 @@ async def check_batch_permissions(self, request):
             request, scope.as_authz_resource(data_type=DATA_TYPE_PHENOPACKET), P_QUERY_DATA, mark_authz_done=True
         )
 
+    def list(self, request, *args, **kwargs):
+        if not self.check_batch_permissions(request):
+            return Response(errors.forbidden_error(), status=status.HTTP_403_FORBIDDEN)
+        return super().list(request, *args, **kwargs)
+
     def create(self, request, *args, **kwargs):
         """
         Despite the name, this is a POST request for returning a list of biosamples. Since query parameters have a