SDK-348

Add tests for handling strict permission errors during file operations and simplify permission error handling logic

Author: udirctera

cterasdk/cio/core/commands.py

@@ -40,54 +40,56 @@ def _extract_rc_msg(result):
     return getattr(result, 'rc', None), getattr(result, 'msg', None)
 
 
-_STRICT_PERMISSION_ERROR_SET = {
-    (None, None),
-    (None, ''),
-    (0, None),
-    ('0', None),
-    (None, 'permission denied'),
-    (None, 'access denied'),
-    (None, 'read only'),
-    (None, 'action is not allowed'),
-    ('permissiondenied', None),
-    (None, 'permissiondenied'),
-    ('permissiondenied', 'permissiondenied'),
+_STRICT_PERMISSION_DENIED_MESSAGES = {
+    'permission denied',
+    'access denied',
+    'read only',
+    'action is not allowed',
 }
 
-_STRICT_PERMISSION_TASK_ERROR_SET = {
-    (None, None, 'permissiondenied'),
-    (None, None, 'permission denied'),
-    (None, None, 'access denied'),
-    (None, None, 'read only'),
-    (None, None, 'action is not allowed'),
-    (None, 'permission denied', None),
-    (None, 'access denied', None),
-    (None, 'read only', None),
-    (None, 'action is not allowed', None),
-    (0, None, 'permissiondenied'),
-    ('0', None, 'permissiondenied'),
-}
+# Some APIs return condensed "permissiondenied" in rc/msg/error_type.
+_STRICT_PERMISSION_DENIED_TOKENS = {'permissiondenied'}
+_STRICT_PERMISSION_AMBIGUOUS_RC = {None, 0, '0'}
+_STRICT_PERMISSION_EMPTY_MESSAGES = {None, ''}
 
 
 def _raise_strict_permission_denied(result, path):
-    rc, msg = _extract_rc_msg(result)
-    rc = _normalize_rc(rc)
-    msg = _normalize_msg(msg)
+    rc, msg = _extract_normalized_rc_msg(result)
     logger.info(
         'strict_permission response for %s: rc=%r msg=%r raw=%s',
         path, rc, msg, type(result).__name__
     )
-    if (rc, msg) in _STRICT_PERMISSION_ERROR_SET:
+    if _is_strict_permission_denied_response(rc, msg):
         raise exceptions.io.core.PrivilegeError(path)
 
 
-def _extract_task_error_tuple(result):
+def _extract_normalized_rc_msg(result):
+    rc, msg = _extract_rc_msg(result)
+    return _normalize_rc(rc), _normalize_msg(msg)
+
+
+def _extract_task_error_fields(result):
     rc = _normalize_rc(getattr(result, 'rc', None))
     msg = _normalize_msg(getattr(result, 'msg', None))
     error_type = _normalize_msg(getattr(result, 'error_type', None))
     return rc, msg, error_type
 
 
+def _is_strict_permission_denied_response(rc, msg):
+    if msg in _STRICT_PERMISSION_DENIED_MESSAGES:
+        return True
+    if rc in _STRICT_PERMISSION_DENIED_TOKENS or msg in _STRICT_PERMISSION_DENIED_TOKENS:
+        return True
+    return rc in _STRICT_PERMISSION_AMBIGUOUS_RC and msg in _STRICT_PERMISSION_EMPTY_MESSAGES
+
+
+def _is_strict_permission_denied_task(rc, msg, error_type):
+    _ = rc
+    if error_type in _STRICT_PERMISSION_DENIED_MESSAGES or error_type in _STRICT_PERMISSION_DENIED_TOKENS:
+        return True
+    return msg in _STRICT_PERMISSION_DENIED_MESSAGES
+
+
 def split_file_directory(listdir, receiver, destination):
     """
     Split a path into its parent directory and final component.
@@ -1022,8 +1024,8 @@ async def _a_execute(self):
 
     def _handle_response(self, r):
         if self._strict_permission:
-            rc, msg, error_type = _extract_task_error_tuple(r)
-            if (rc, msg, error_type) in _STRICT_PERMISSION_TASK_ERROR_SET:
+            rc, msg, error_type = _extract_task_error_fields(r)
+            if _is_strict_permission_denied_task(rc, msg, error_type):
                 raise exceptions.io.core.PrivilegeError('')
         if not self.block:
             return r

tests/ut/core/user/test_mkdir.py

@@ -36,6 +36,12 @@ def test_mkdir_strict_permission_denied(self):
         with self.assertRaises(exceptions.io.core.PrivilegeError):
             self._services.files.mkdir(self._directory, strict_permission=True)
 
+    def test_mkdir_strict_permission_denied_message(self):
+        execute_response = Object(msg='access denied')
+        self._init_services(execute_response=execute_response)
+        with self.assertRaises(exceptions.io.core.PrivilegeError):
+            self._services.files.mkdir(self._directory, strict_permission=True)
+
     def test_makedirs_strict_permission_root_path(self):
         rooted_directories = 'Team Portal/Engineering/Documents'
         self._init_services(execute_response=None)

tests/ut/core/user/test_move.py

@@ -32,6 +32,14 @@ def test_move_strict_permission_denied(self):
             self._services.files.move(self._source, destination=self._dest, strict_permission=True)
         self._services.tasks.wait.assert_called_once_with(self._task_reference)
 
+    def test_move_strict_permission_denied_error_type(self):
+        task = Object(error_type='permissiondenied')
+        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 test_move_completed_with_warnings_raises_move_error(self):
         task = mock.MagicMock()
         task.completed = False