Add tests for the retrieve_asset_permissions & assign_change_asset_pe…

…rmission functions

onadata/libs/tests/utils/test_project_utils.py

@@ -5,12 +5,15 @@
 from django.test.utils import override_settings
 
 from kombu.exceptions import OperationalError
-from mock import patch
+from mock import MagicMock, patch
+from requests import Response, session
 
 from onadata.apps.logger.models import Project
 from onadata.apps.main.tests.test_base import TestBase
 from onadata.libs.permissions import DataEntryRole
-from onadata.libs.utils.project_utils import (set_project_perms_to_xform,
+from onadata.libs.utils.project_utils import (assign_change_asset_permission,
+                                              retrieve_asset_permissions,
+                                              set_project_perms_to_xform,
                                               set_project_perms_to_xform_async)
 
 
@@ -65,6 +68,58 @@ def test_set_project_perms_to_xform_async(self, mock):
         args, _kwargs = mock.call_args_list[0]
         self.assertEqual(args[0], self.xform)
         self.assertEqual(args[1], self.project)
+
+    def test_assign_change_asset_permission(self):
+        """
+        Test that the `assign_change_asset_permission` function calls
+        the External service with the correct payload
+        """
+        session_mock = MagicMock()
+        asset_id = "some_id"
+        usernames = ["bob", "john", "doe"]
+        resp = Response()
+        resp.status_code = 200
+        session_mock.post.return_value = resp
+
+        ret = assign_change_asset_permission("http://example.com", asset_id, usernames, session_mock)
+        session_mock.post.assert_called()
+        session_mock.post.assert_called_with(
+            f"http://example.com/api/v2/assets/{asset_id}/permission-assignments/bulk/",
+            json=[
+                {
+                "user": "http://example.com/api/v2/users/bob/",
+                "permission": "http://example.com/api/v2/permissions/change_asset/"
+                },
+                {
+                "user": "http://example.com/api/v2/users/john/",
+                "permission": "http://example.com/api/v2/permissions/change_asset/"
+                },
+                {
+                "user": "http://example.com/api/v2/users/doe/",
+                "permission": "http://example.com/api/v2/permissions/change_asset/"
+                },
+            ]
+        )
+
+    def test_retrieve_asset_permission(self):
+        """
+        Test that the `retrieve_asset_permissions` function calls
+        the External service with the correct payload
+        """
+        session_mock = MagicMock()
+        asset_id = "some_id"
+        service_url = "http://example.com"
+        resp = Response()
+        resp.status_code = 200
+        resp._content = b'[{"user": "http://example.com/api/v2/users/bob", "url": "http://example.com/api/v2/permission-assignments/some_uuid"}]'
+        session_mock.get.return_value = resp
+
+        ret = retrieve_asset_permissions(service_url, asset_id, session_mock)
+        session_mock.get.assert_called()
+        session_mock.get.assert_called_with(
+            f"{service_url}/api/v2/assets/{asset_id}/permission-assignments/"
+        )
+        self.assertEqual(ret, {"bob": ["http://example.com/api/v2/permission-assignments/some_uuid"]})
 
     @override_settings(CELERY_TASK_ALWAYS_EAGER=True)
     @patch(