PIM-6080

Add unit tests for strict permission handling in `move` and `mkdir` operations

Author: udirctera

cterasdk/cio/core/commands.py

@@ -80,6 +80,7 @@ def _raise_strict_permission_denied(result, path):
     if (rc, msg) in _STRICT_PERMISSION_ERROR_SET:
         raise exceptions.io.core.PrivilegeError(path)
 
+
 def _extract_task_error_tuple(result):
     rc = _normalize_rc(getattr(result, 'rc', None))
     msg = _normalize_msg(getattr(result, 'msg', None))

tests/ut/core/user/test_mkdir.py

@@ -1,6 +1,9 @@
 from unittest import mock
 import munch
 
+from cterasdk.common import Object
+from cterasdk import exceptions
+
 from tests.ut.core.user import base_user
 
 
@@ -21,3 +24,14 @@ def test_mkdir(self):
         })
         actual_param = self._services.api.execute.call_args[0][2]
         self._assert_equal_objects(actual_param, expected_param)
+
+    def test_mkdir_strict_permission_success(self):
+        self._init_services(execute_response=None)
+        ret = self._services.files.mkdir(self._directory, strict_permission=True)
+        self.assertEqual(ret, self._directory)
+
+    def test_mkdir_strict_permission_denied(self):
+        execute_response = Object(rc=0)
+        self._init_services(execute_response=execute_response)
+        with self.assertRaises(exceptions.io.core.PrivilegeError):
+            self._services.files.mkdir(self._directory, strict_permission=True)

tests/ut/core/user/test_move.py

@@ -1,5 +1,8 @@
 from unittest import mock
 
+from cterasdk.common import Object
+from cterasdk import exceptions
+
 from tests.ut.core.user import base_user
 
 
@@ -21,6 +24,14 @@ def test_move_no_wait(self):
         self._assert_equal_objects(actual_param, expected_param)
         self.assertEqual(ret.ref, execute_response)
 
+    def test_move_strict_permission_denied(self):
+        task = Object(msg='permission denied')
+        self._services.tasks.wait = mock.MagicMock(return_value=task)
+        self._init_services(execute_response=self._task_reference)
+        with self.assertRaises(exceptions.io.core.PrivilegeError):
+            self._services.files.move(self._source, destination=self._dest, strict_permission=True)
+        self._services.tasks.wait.assert_called_once_with(self._task_reference)
+
     def _create_move_resource_param(self):
         destinations = [base_user.BaseCoreServicesTest.encode_path(self._dest + '/' + self._filename)]
         return self._create_action_resource_param([base_user.BaseCoreServicesTest.encode_path(self._source)], destinations)