From c9bd7a057da90678125dc24257f78f5cd635b3d5 Mon Sep 17 00:00:00 2001 From: udir Date: Wed, 11 Feb 2026 11:09:12 +0200 Subject: [PATCH 1/2] 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 --- cterasdk/cio/core/commands.py | 21 +++------------------ tests/ut/core/user/test_mkdir.py | 7 +++++++ tests/ut/core/user/test_move.py | 15 +++++++++++++++ 3 files changed, 25 insertions(+), 18 deletions(-) diff --git a/cterasdk/cio/core/commands.py b/cterasdk/cio/core/commands.py index 6fb55acc..cc83a442 100644 --- a/cterasdk/cio/core/commands.py +++ b/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) diff --git a/tests/ut/core/user/test_mkdir.py b/tests/ut/core/user/test_mkdir.py index 4af36857..f89e19dc 100644 --- a/tests/ut/core/user/test_mkdir.py +++ b/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) diff --git a/tests/ut/core/user/test_move.py b/tests/ut/core/user/test_move.py index d35a03f9..fe220dbd 100644 --- a/tests/ut/core/user/test_move.py +++ b/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) From 32e76a03cdc70d68ed95d69f5504ed125a65c483 Mon Sep 17 00:00:00 2001 From: udir Date: Wed, 11 Feb 2026 14:53:22 +0200 Subject: [PATCH 2/2] SDK-348 Add tests for handling strict permission errors during file operations and simplify permission error handling logic --- cterasdk/cio/core/commands.py | 66 ++++++++++++++++---------------- tests/ut/core/user/test_mkdir.py | 6 +++ tests/ut/core/user/test_move.py | 8 ++++ 3 files changed, 48 insertions(+), 32 deletions(-) diff --git a/cterasdk/cio/core/commands.py b/cterasdk/cio/core/commands.py index cc83a442..4bc0ee78 100644 --- a/cterasdk/cio/core/commands.py +++ b/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 diff --git a/tests/ut/core/user/test_mkdir.py b/tests/ut/core/user/test_mkdir.py index f89e19dc..901f6aca 100644 --- a/tests/ut/core/user/test_mkdir.py +++ b/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) diff --git a/tests/ut/core/user/test_move.py b/tests/ut/core/user/test_move.py index fe220dbd..6e5e2985 100644 --- a/tests/ut/core/user/test_move.py +++ b/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