SDK-348

- Add unit test for handling `completed_with_warnings` in file move operations
- Simplify `_parents_generator` logic, removing strict permission argument from directory creation commands
- Add unit test for strict permission directory creation at root paths

Author: udirctera

cterasdk/cio/core/commands.py

@@ -685,17 +685,7 @@ def _before_command(self):
     def _parents_generator(self):
         if self.parents:
             parts = self.path.parts
-            start_index = 1
-            known_roots = ('My Files', 'Shared With Me', 'Shared', 'Team Portal')
-            if parts:
-                if parts[0] in known_roots:
-                    start_index = 2
-                elif parts[0] in ('Users', 'Groups') and len(parts) > 1:
-                    if len(parts) > 2 and parts[2] in known_roots:
-                        start_index = 4
-                    else:
-                        start_index = 3
-            for i in range(start_index, len(parts)):
+            for i in range(1, len(parts)):
                 yield automatic_resolution('/'.join(parts[:i]), self._receiver.context)
         else:
             yield self.path
@@ -707,8 +697,7 @@ def _execute(self):
                     CreateDirectory(
                         self._function,
                         self._receiver,
-                        path,
-                        strict_permission=self._strict_permission
+                        path
                     ).execute()
                 except exceptions.io.core.CreateDirectoryError as e:
                     CreateDirectory._suppress_file_conflict_error(e)
@@ -722,8 +711,7 @@ async def _a_execute(self):
                     await CreateDirectory(
                         self._function,
                         self._receiver,
-                        path,
-                        strict_permission=self._strict_permission
+                        path
                     ).a_execute()
                 except exceptions.io.core.CreateDirectoryError as e:
                     CreateDirectory._suppress_file_conflict_error(e)
@@ -1043,9 +1031,6 @@ def _handle_response(self, r):
         if r.completed:
             return self._task_complete(r)
 
-        if r.completed_with_warnings:
-            return r
-
         if r.failed or r.completed_with_warnings:
             return self._task_error(r)
 

tests/ut/core/user/test_mkdir.py

@@ -35,3 +35,10 @@ def test_mkdir_strict_permission_denied(self):
         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)
+        ret = self._services.files.makedirs(rooted_directories, strict_permission=True)
+        self.assertEqual(self._services.api.execute.call_count, len(rooted_directories.split('/')))
+        self.assertEqual(ret, rooted_directories)

tests/ut/core/user/test_move.py

@@ -32,6 +32,21 @@ 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_completed_with_warnings_raises_move_error(self):
+        task = mock.MagicMock()
+        task.completed = False
+        task.completed_with_warnings = True
+        task.failed = False
+        task.cursor = None
+        task.error_type = None
+        task.unknown_object.return_value = True
+        task.progress_str = None
+        self._services.tasks.wait = mock.MagicMock(return_value=task)
+        self._init_services(execute_response=self._task_reference)
+        with self.assertRaises(exceptions.io.core.MoveError):
+            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)