Raise tastypie.NotFound in couch mixin helper

Ensures that a 404 is returned when a couch resource that uses this
helper is not found

Author: gherceg

corehq/apps/api/resources/__init__.py

@@ -1,6 +1,6 @@
 import json
 
-from django.core.exceptions import ValidationError
+from django.core.exceptions import ObjectDoesNotExist, ValidationError
 from django.http import HttpResponse
 from django.urls import NoReverseMatch, include, re_path
 
@@ -16,7 +16,7 @@
 from corehq.apps.accounting.utils import domain_has_privilege
 from corehq.apps.analytics.tasks import track_workflow_noop
 from corehq.apps.api.cors import add_cors_headers_to_response
-from corehq.apps.api.util import get_obj
+from corehq.apps.api.util import get_obj, get_object_or_not_exist
 from corehq.apps.users.util import is_dimagi_email
 
 
@@ -356,3 +356,12 @@ def detail_uri_kwargs(self, bundle_or_obj):
         return {
             'pk': get_obj(bundle_or_obj)._id
         }
+
+    def get_document_for_update(self, doc_id, domain):
+        """
+        Converts ``ObjectDoesNotExist`` into ``tastypie.NotFound``
+        """
+        try:
+            return get_object_or_not_exist(self._meta.object_class, doc_id, domain)
+        except ObjectDoesNotExist as err:
+            raise NotFound(str(err))

corehq/apps/api/resources/v0_5.py

@@ -349,8 +349,7 @@ def validate_new_user_input(require_account_confirmation, send_confirmation_emai
             raise BadRequest(_("You must require account confirmation to send a confirmation email."))
 
     def obj_update(self, bundle, **kwargs):
-        bundle.obj = CommCareUser.get(kwargs['pk'])
-        assert bundle.obj.domain == kwargs['domain']
+        bundle.obj = self.get_document_for_update(kwargs['pk'], kwargs['domain'])
         send_confirmation_email = string_to_boolean(bundle.data.pop('send_confirmation_email_now', False))
         user_change_logger = self._get_user_change_logger(bundle)
         errors = self._update(bundle, user_change_logger)
@@ -760,8 +759,7 @@ def obj_create(self, bundle, request=None, **kwargs):
         return bundle
 
     def obj_update(self, bundle, **kwargs):
-        bundle.obj = Group.get(kwargs['pk'])
-        assert bundle.obj.domain == kwargs['domain']
+        bundle.obj = self.get_document_for_update(kwargs['pk'], kwargs['domain'])
         if self._update(bundle):
             assert bundle.obj.domain == kwargs['domain']
             bundle.obj.save()

corehq/apps/api/tests/group_resources.py

@@ -1,4 +1,5 @@
 import json
+import uuid
 
 from corehq.apps.api.resources import v0_5
 from corehq.apps.es.groups import group_adapter
@@ -140,6 +141,25 @@ def test_update(self):
         self.assertTrue(modified.case_sharing)
         self.assertEqual(modified.metadata["localization"], "Ghana")
 
+    def test_cant_update_missing_group(self):
+        response = self._assert_auth_post_resource(self.single_endpoint(uuid.uuid4().hex),
+                                                   json.dumps({"name": "test group"}),
+                                                   content_type='application/json',
+                                                   method='PUT')
+
+        assert response.status_code == 404, response.content
+
+    def test_cant_update_group_in_another_domain(self):
+        not_my_group = self._add_group(Group({"name": "theirs", "domain": "not-my-project"}))
+
+        response = self._assert_auth_post_resource(self.single_endpoint(not_my_group._id),
+                                                   json.dumps({"name": "mine now"}),
+                                                   content_type='application/json',
+                                                   method='PUT')
+
+        assert response.status_code == 404, response.content
+        assert Group.get(not_my_group._id).name == "theirs"
+
     def test_delete_group(self):
 
         group = self._add_group(Group({"name": "test", "domain": self.domain.name}))

corehq/apps/api/tests/test_user_resources.py

@@ -1,4 +1,5 @@
 import json
+import uuid
 from unittest.mock import Mock, patch, call
 
 from django.test import TestCase
@@ -458,6 +459,28 @@ def test_update(self):
         self.assertEqual(user_history.change_messages['password'], UserChangeMessage.password_reset()['password'])
         self.assertEqual(user_history.changed_via, USER_CHANGE_VIA_API)
 
+    def test_cant_update_missing_user(self):
+        response = self._assert_auth_post_resource(self.single_endpoint(uuid.uuid4().hex),
+                                                   json.dumps({"first_name": "test"}),
+                                                   content_type='application/json',
+                                                   method='PUT')
+
+        assert response.status_code == 404, response.content
+
+    def test_cant_update_user_in_another_domain(self):
+        not_my_user = CommCareUser.create(domain='not-my-project', username="theirs",
+                                          password="qwer1234", created_by=None, created_via=None)
+        self.addCleanup(not_my_user.delete, 'not-my-project', deleted_by=None)
+        first_name_before = not_my_user.first_name
+
+        response = self._assert_auth_post_resource(self.single_endpoint(not_my_user._id),
+                                                   json.dumps({"first_name": "mine now"}),
+                                                   content_type='application/json',
+                                                   method='PUT')
+
+        assert response.status_code == 404, response.content
+        assert CommCareUser.get(not_my_user._id).first_name == first_name_before
+
     def test_update_fails(self):
         user = CommCareUser.create(domain=self.domain.name, username="test", password="qwer1234",
                                    created_by=None, created_via=None, phone_number="50253311398")