Saimon/pim 6659

cterasdk/cio/core/commands.py

@@ -360,7 +360,7 @@ def __init__(self, function, receiver, path):
         self.path = automatic_resolution(path, receiver.context)
 
     def get_parameter(self):
-        return self.path.reference
+        return self.path.relative_encode
 
     def _before_command(self):
         raise_or_suppress_access_error(self._receiver, self.path)
@@ -422,7 +422,7 @@ def _before_command(self):
 
     def get_parameter(self):
         param = Object()
-        param.paths = ['/'.join([self.directory.absolute, filename]) for filename in self.objects]
+        param.paths = [self.directory.join(filename).absolute_encode for filename in self.objects]
         param.snapshot = None
         param.password = None
         param.portalName = None
@@ -629,7 +629,7 @@ def _before_command(self):
         logger.info('Listing versions: %s', self.path)
 
     def get_parameter(self):
-        return self.path.absolute
+        return self.path.absolute_encode
 
     def _execute(self):
         with self.trace_execution():

tests/ut/core/user/test_special_characters.py

@@ -0,0 +1,249 @@
+from unittest import mock
+from urllib.parse import quote
+from datetime import datetime
+
+import munch
+
+from cterasdk.common import Object
+from tests.ut.core.user import base_user
+
+
+class TestSpecialCharacterPaths(base_user.BaseCoreServicesTest):
+    """
+    PIM-6659: Verify that all file operations properly URL-encode paths
+    containing special characters (%, #, spaces, etc.) when constructing
+    WebDAV or API request parameters.
+    """
+
+    SPECIAL_FILENAMES = [
+        'file_100%_done.txt',
+        'file_%25_encoded.txt',
+        'report #1.txt',
+        'file with spaces.txt',
+        'file_!@#$%^([{.txt',
+        'résumé.txt',
+    ]
+
+    SPECIAL_DIRECTORIES = [
+        'My Files/Documents 2026',
+        'My Files/100% Complete',
+        'My Files/Report #1',
+    ]
+
+    def setUp(self):
+        super().setUp()
+        self._special_dir = 'My Files/100% Complete'
+        self._special_filename = 'file_100%_done.txt'
+        self._special_path = f'{self._special_dir}/{self._special_filename}'
+
+    def _expected_encoded_absolute(self, path):
+        return f'{self._base}/{quote(path)}'
+
+    # --- Open / handle ---
+
+    def test_handle_encodes_special_characters(self):
+        for filename in self.SPECIAL_FILENAMES:
+            path = f'My Files/{filename}'
+            self._init_services()
+            mock_download = mock.MagicMock()
+            self._services.io._webdav.download = mock_download  # pylint: disable=protected-access
+            self._services.files.handle(path)
+            call_args = mock_download.call_args[0]
+            actual_path = str(call_args[0])
+            expected_path = quote(f'My Files/{filename}')
+            self.assertEqual(actual_path, expected_path,
+                             f'handle() did not encode path for filename: {filename}')
+
+    def test_handle_percent_in_directory(self):
+        for directory in self.SPECIAL_DIRECTORIES:
+            path = f'{directory}/document.txt'
+            self._init_services()
+            mock_download = mock.MagicMock()
+            self._services.io._webdav.download = mock_download  # pylint: disable=protected-access
+            self._services.files.handle(path)
+            call_args = mock_download.call_args[0]
+            actual_path = str(call_args[0])
+            expected_path = quote(path)
+            self.assertEqual(actual_path, expected_path,
+                             f'handle() did not encode directory: {directory}')
+
+    # --- Delete ---
+
+    def test_delete_encodes_special_characters(self):
+        for filename in self.SPECIAL_FILENAMES:
+            path = f'My Files/{filename}'
+            self._init_services(execute_response=self._task_reference)
+            self._services.files.delete(path, wait=False)
+            self._services.api.execute.assert_called_once_with('', 'deleteResources', mock.ANY)
+            actual_param = self._services.api.execute.call_args[0][2]
+            self.assertEqual(actual_param.urls[0].src, self._expected_encoded_absolute(path),
+                             f'delete() did not encode path for filename: {filename}')
+
+    # --- Undelete / Recover ---
+
+    def test_undelete_encodes_special_characters(self):
+        for filename in self.SPECIAL_FILENAMES:
+            path = f'My Files/{filename}'
+            self._init_services(execute_response=self._task_reference)
+            self._services.files.undelete(path, wait=False)
+            self._services.api.execute.assert_called_once_with('', 'restoreResources', mock.ANY)
+            actual_param = self._services.api.execute.call_args[0][2]
+            self.assertEqual(actual_param.urls[0].src, self._expected_encoded_absolute(path),
+                             f'undelete() did not encode path for filename: {filename}')
+
+    # --- Copy ---
+
+    def test_copy_encodes_special_characters(self):
+        for filename in self.SPECIAL_FILENAMES:
+            source = f'My Files/{filename}'
+            dest_dir = 'My Files/Archive'
+            self._init_services(execute_response=self._task_reference)
+            self._services.files.copy(source, destination=dest_dir, wait=False)
+            self._services.api.execute.assert_called_once_with('', 'copyResources', mock.ANY)
+            actual_param = self._services.api.execute.call_args[0][2]
+            self.assertEqual(actual_param.urls[0].src, self._expected_encoded_absolute(source),
+                             f'copy() did not encode source for filename: {filename}')
+            expected_dest = self._expected_encoded_absolute(f'{dest_dir}/{filename}')
+            self.assertEqual(actual_param.urls[0].dest, expected_dest,
+                             f'copy() did not encode dest for filename: {filename}')
+
+    def test_copy_to_special_char_destination(self):
+        source = 'My Files/document.txt'
+        dest_dir = 'My Files/100% Complete'
+        self._init_services(execute_response=self._task_reference)
+        self._services.files.copy(source, destination=dest_dir, wait=False)
+        actual_param = self._services.api.execute.call_args[0][2]
+        expected_dest = self._expected_encoded_absolute(f'{dest_dir}/document.txt')
+        self.assertEqual(actual_param.urls[0].dest, expected_dest)
+
+    # --- Move ---
+
+    def test_move_encodes_special_characters(self):
+        for filename in self.SPECIAL_FILENAMES:
+            source = f'My Files/{filename}'
+            dest_dir = 'My Files/Archive'
+            self._init_services(execute_response=self._task_reference)
+            self._services.files.move(source, destination=dest_dir, wait=False)
+            self._services.api.execute.assert_called_once_with('', 'moveResources', mock.ANY)
+            actual_param = self._services.api.execute.call_args[0][2]
+            self.assertEqual(actual_param.urls[0].src, self._expected_encoded_absolute(source),
+                             f'move() did not encode source for filename: {filename}')
+            expected_dest = self._expected_encoded_absolute(f'{dest_dir}/{filename}')
+            self.assertEqual(actual_param.urls[0].dest, expected_dest,
+                             f'move() did not encode dest for filename: {filename}')
+
+    def test_move_from_special_char_directory(self):
+        source = 'My Files/100% Complete/document.txt'
+        dest_dir = 'My Files/Archive'
+        self._init_services(execute_response=self._task_reference)
+        self._services.files.move(source, destination=dest_dir, wait=False)
+        actual_param = self._services.api.execute.call_args[0][2]
+        self.assertEqual(actual_param.urls[0].src, self._expected_encoded_absolute(source))
+
+    # --- Rename ---
+
+    def test_rename_encodes_special_characters(self):
+        for filename in self.SPECIAL_FILENAMES:
+            parent = 'My Files/Documents'
+            path = f'{parent}/{filename}'
+            new_name = 'renamed.txt'
+            self._init_services(execute_response=self._task_reference)
+            self._services.files.rename(path, new_name, wait=False)
+            self._services.api.execute.assert_called_once_with('', 'moveResources', mock.ANY)
+            actual_param = self._services.api.execute.call_args[0][2]
+            self.assertEqual(actual_param.urls[0].src, self._expected_encoded_absolute(path),
+                             f'rename() did not encode source for filename: {filename}')
+            expected_dest = self._expected_encoded_absolute(f'{parent}/{new_name}')
+            self.assertEqual(actual_param.urls[0].dest, expected_dest,
+                             f'rename() did not encode dest for filename: {filename}')
+
+    def test_rename_to_special_char_name(self):
+        parent = 'My Files/Documents'
+        path = f'{parent}/document.txt'
+        for new_name in self.SPECIAL_FILENAMES:
+            self._init_services(execute_response=self._task_reference)
+            self._services.files.rename(path, new_name, wait=False)
+            actual_param = self._services.api.execute.call_args[0][2]
+            expected_dest = self._expected_encoded_absolute(f'{parent}/{new_name}')
+            self.assertEqual(actual_param.urls[0].dest, expected_dest,
+                             f'rename() did not encode dest for new name: {new_name}')
+
+    def test_rename_special_char_in_directory(self):
+        parent = 'My Files/100% Complete'
+        old_name = 'document.txt'
+        new_name = 'renamed.txt'
+        path = f'{parent}/{old_name}'
+        self._init_services(execute_response=self._task_reference)
+        self._services.files.rename(path, new_name, wait=False)
+        actual_param = self._services.api.execute.call_args[0][2]
+        self.assertEqual(actual_param.urls[0].src, self._expected_encoded_absolute(path))
+        self.assertEqual(actual_param.urls[0].dest, self._expected_encoded_absolute(f'{parent}/{new_name}'))
+
+    def test_rename_percent_to_percent(self):
+        parent = 'My Files'
+        old_name = 'file_100%_done.txt'
+        new_name = 'file_50%_done.txt'
+        path = f'{parent}/{old_name}'
+        self._init_services(execute_response=self._task_reference)
+        self._services.files.rename(path, new_name, wait=False)
+        actual_param = self._services.api.execute.call_args[0][2]
+        self.assertEqual(actual_param.urls[0].src, self._expected_encoded_absolute(path))
+        self.assertEqual(actual_param.urls[0].dest, self._expected_encoded_absolute(f'{parent}/{new_name}'))
+
+    # --- ListVersions ---
+
+    def test_versions_encodes_special_characters(self):
+        for filename in self.SPECIAL_FILENAMES:
+            path = f'My Files/{filename}'
+            self._init_services(execute_response=[self._create_snapshot_response(path)])
+            self._services.files.versions(path)
+            self._services.api.execute.assert_called_once_with(
+                '', 'listSnapshots', self._expected_encoded_absolute(path)
+            )
+
+    def test_versions_percent_in_directory(self):
+        path = 'My Files/100% Complete/document.txt'
+        self._init_services(execute_response=[self._create_snapshot_response(path)])
+        self._services.files.versions(path)
+        self._services.api.execute.assert_called_once_with(
+            '', 'listSnapshots', self._expected_encoded_absolute(path)
+        )
+
+    # --- ListDir ---
+
+    def test_listdir_encodes_special_characters(self):
+        self.patch_call("cterasdk.cio.core.commands.EnsureDirectory.execute")
+        for directory in self.SPECIAL_DIRECTORIES:
+            self._init_services(execute_response=Object(**{
+                'errorType': None,
+                'hasMore': False,
+                'items': []
+            }))
+            list(self._services.files.listdir(directory))
+            actual_param = self._services.api.execute.call_args[0][2]
+            self.assertEqual(actual_param.root, self._expected_encoded_absolute(directory),
+                             f'listdir() did not encode path for directory: {directory}')
+
+    # --- Mkdir ---
+
+    def test_mkdir_encodes_special_characters(self):
+        for directory in self.SPECIAL_DIRECTORIES:
+            self._init_services()
+            self._services.files.mkdir(directory)
+            actual_param = self._services.api.execute.call_args[0][2]
+            parts = directory.split('/')
+            parent = '/'.join(parts[:-1])
+            expected_parent = f'{self._base}/{quote(parent)}'
+            self.assertEqual(actual_param.parentPath, expected_parent,
+                             f'mkdir() did not encode parentPath for directory: {directory}')
+
+    # --- Helpers ---
+
+    def _create_snapshot_response(self, path):
+        return munch.Munch({
+            'url': self._base,
+            'path': path,
+            'current': True,
+            'startTimestamp': datetime.now().isoformat(),
+            'calculatedTimestamp': datetime.now().isoformat()
+        })

tests/ut/core/user/test_versions.py

@@ -1,4 +1,5 @@
 from datetime import datetime
+from urllib.parse import quote
 import munch
 from tests.ut.core.user import base_user
 
@@ -9,7 +10,7 @@ def test_list_versions(self):
         directory = 'My Files'
         self._init_services(execute_response=[self._create_snapshot_response(directory, True)])
         ret = self._services.files.versions(directory)
-        self._services.api.execute.assert_called_once_with('', 'listSnapshots', f'{self._base}/{directory}')
+        self._services.api.execute.assert_called_once_with('', 'listSnapshots', f'{self._base}/{quote(directory)}')
         self.assertEqual(str(ret[0].path), directory)
 
     def _create_snapshot_response(self, path, current):