OpenConceptLab/ocl_online#477 | Repository create takes its owner only from the URL parent

core/collections/tests/tests.py

@@ -5534,6 +5534,45 @@ def test_verify_scope_no_kwargs_non_get_raises_404(self):
         response = self.client.post('/collections/', {}, format='json')
         self.assertEqual(response.status_code, 404)
 
+    def test_create_rejects_conflicting_body_owner_collection(self):
+        user = UserProfileFactory()
+        org = OrganizationFactory()
+        org.members.add(user)
+        other_org = OrganizationFactory()
+        other_user = UserProfileFactory()
+        token = f"Token {user.get_token()}"
+        cases = (
+            (f'/users/{user.username}/collections/', {'organization_id': other_org.id}),
+            (f'/users/{user.username}/collections/', {'organization': org.id}),
+            (f'/users/{user.username}/collections/', {'user_id': other_user.id}),
+            (f'/users/{user.username}/collections/', {'user': other_user.id}),
+            (f'/orgs/{org.mnemonic}/collections/', {'organization_id': other_org.id}),
+            (f'/orgs/{org.mnemonic}/collections/', {'user_id': user.id}),
+        )
+        for index, (url, owner_data) in enumerate(cases):
+            response = self.client.post(
+                url, {'id': f'collection-{index}', 'name': 'n', **owner_data}, HTTP_AUTHORIZATION=token, format='json'
+            )
+            self.assertEqual(response.status_code, 400)
+            self.assertFalse({'Source': Source, 'Collection': Collection}['Collection'].objects.filter(
+                mnemonic=f'collection-{index}').exists())
+
+    def test_create_accepts_omitted_or_matching_body_owner_collection(self):
+        user = UserProfileFactory()
+        org = OrganizationFactory()
+        org.members.add(user)
+        token = f"Token {user.get_token()}"
+        for index, (url, owner_data) in enumerate((
+                (f'/users/{user.username}/collections/', {}),
+                (f'/users/{user.username}/collections/', {'user_id': user.id}),
+                (f'/orgs/{org.mnemonic}/collections/', {'organization': org.id}),
+        )):
+            response = self.client.post(
+                url, {'id': f'collection-ok-{index}', 'name': 'n', **owner_data},
+                HTTP_AUTHORIZATION=token, format='json'
+            )
+            self.assertEqual(response.status_code, 201)
+
     def test_verify_scope_no_owner_scope_raises_404(self):
         response = self.client.get('/collections/some-collection/')
         self.assertEqual(response.status_code, 404)

core/common/mixins.py

@@ -18,6 +18,7 @@
 from psycopg2.errorcodes import QUERY_CANCELED
 from pydash import compact, get
 from rest_framework import status
+from rest_framework.exceptions import ValidationError as DRFValidationError
 from rest_framework.mixins import ListModelMixin, CreateModelMixin
 from rest_framework.permissions import IsAuthenticated
 from rest_framework.response import Response
@@ -483,6 +484,18 @@ def set_parent_resource(self):
 
         self.kwargs['parent_resource'] = self.parent_resource = parent_resource
 
+    def set_owner_from_parent(self, data, key_suffix='', reject_conflict=True):
+        """The URL parent is the only owner. A body owner naming anyone else raises a 400 (or is dropped
+        if reject_conflict is False); an omitted or matching one is accepted."""
+        parent_type = self.parent_resource.resource_type.lower()
+        for key in ('organization', 'user', 'organization_id', 'user_id'):
+            value = data.pop(key, None)
+            is_parent = key.split('_', maxsplit=1)[0] == parent_type and str(value) == str(self.parent_resource.id)
+            if reject_conflict and value not in (None, '') and not is_parent:
+                raise DRFValidationError({key: ['Owner must be the one in the URL.']})
+        data[parent_type + key_suffix] = self.parent_resource.id
+        return data
+
     def create(self, request, **kwargs):  # pylint: disable=unused-argument
         if not self.parent_resource:
             return Response(status=status.HTTP_405_METHOD_NOT_ALLOWED)
@@ -497,11 +510,9 @@ def create(self, request, **kwargs):  # pylint: disable=unused-argument
         if isinstance(supported_locales, str):
             supported_locales = compact(supported_locales.split(','))
 
-        data = {
-            'mnemonic': data.get('id'),
-            'supported_locales': supported_locales,
-            'version': HEAD, **data, **{self.parent_resource.resource_type.lower(): self.parent_resource.id}
-        }
+        data = self.set_owner_from_parent({
+            'mnemonic': data.get('id'), 'supported_locales': supported_locales, 'version': HEAD, **data
+        })
 
         serializer = self.get_serializer(data=data)
         if serializer.is_valid():

core/map_projects/views.py

@@ -90,13 +90,9 @@ def create(self, request, **kwargs):  # pylint: disable=unused-argument
         if not self.parent_resource:
             return Response(status=status.HTTP_405_METHOD_NOT_ALLOWED)
 
-        data = MapProject.format_request_data(request.data, self.parent_resource)
-        # Ownership comes only from the URL parent: a body organization_id posted to a user route would re-parent
-        # the project past HasMapProjectParentOwnership and CanCreateOrgMapProjects (OpenConceptLab/ocl_issues#2804).
-        parent_key = self.parent_resource.resource_type.lower() + '_id'
-        for key in ('organization_id', 'user_id'):
-            if key != parent_key:
-                data.pop(key, None)
+        data = self.set_owner_from_parent(
+            MapProject.format_request_data(request.data), '_id', reject_conflict=False
+        )
         serializer = self.get_serializer(data=data)
         if not serializer.is_valid():
             return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)

core/sources/tests/tests.py

@@ -54,6 +54,45 @@ def test_verify_scope_no_kwargs_non_get_raises_404(self):
         response = self.client.post('/sources/', {}, format='json')
         self.assertEqual(response.status_code, 404)
 
+    def test_create_rejects_conflicting_body_owner_source(self):
+        user = UserProfileFactory()
+        org = OrganizationFactory()
+        org.members.add(user)
+        other_org = OrganizationFactory()
+        other_user = UserProfileFactory()
+        token = f"Token {user.get_token()}"
+        cases = (
+            (f'/users/{user.username}/sources/', {'organization_id': other_org.id}),
+            (f'/users/{user.username}/sources/', {'organization': org.id}),
+            (f'/users/{user.username}/sources/', {'user_id': other_user.id}),
+            (f'/users/{user.username}/sources/', {'user': other_user.id}),
+            (f'/orgs/{org.mnemonic}/sources/', {'organization_id': other_org.id}),
+            (f'/orgs/{org.mnemonic}/sources/', {'user_id': user.id}),
+        )
+        for index, (url, owner_data) in enumerate(cases):
+            response = self.client.post(
+                url, {'id': f'source-{index}', 'name': 'n', **owner_data}, HTTP_AUTHORIZATION=token, format='json'
+            )
+            self.assertEqual(response.status_code, 400)
+            self.assertFalse({'Source': Source, 'Collection': Collection}['Source'].objects.filter(
+                mnemonic=f'source-{index}').exists())
+
+    def test_create_accepts_omitted_or_matching_body_owner_source(self):
+        user = UserProfileFactory()
+        org = OrganizationFactory()
+        org.members.add(user)
+        token = f"Token {user.get_token()}"
+        for index, (url, owner_data) in enumerate((
+                (f'/users/{user.username}/sources/', {}),
+                (f'/users/{user.username}/sources/', {'user_id': user.id}),
+                (f'/orgs/{org.mnemonic}/sources/', {'organization': org.id}),
+        )):
+            response = self.client.post(
+                url, {'id': f'source-ok-{index}', 'name': 'n', **owner_data},
+                HTTP_AUTHORIZATION=token, format='json'
+            )
+            self.assertEqual(response.status_code, 201)
+
     def test_verify_scope_no_owner_scope_raises_404(self):
         response = self.client.get('/sources/some-source/')
         self.assertEqual(response.status_code, 404)