Handle Invalid Characters in Path

src/kitools/data_adapters/synapse/exceptions.py

@@ -0,0 +1,3 @@
+
+class InvalidNameError(ValueError):
+    pass

src/kitools/data_adapters/synapse/synapse_adapter.py

@@ -13,9 +13,11 @@
 # limitations under the License.
 
 import os
+import re
 import synapseclient
 from ..base_adapter import BaseAdapter
 from .synapse_remote_entity import SynapseRemoteEntity
+from .exceptions import InvalidNameError
 from ...data_uri import DataUri
 from ...sys_path import SysPath
 from ...env import Env
@@ -30,6 +32,7 @@ class SynapseAdapter(BaseAdapter):
     """
 
     DATA_URI_SCHEME = 'syn'
+    VALID_NAME_REGEX = re.compile(r'^[A-Za-z0-9_\-\.\+\(\) ]*$')
     _client = None
 
     @classmethod
@@ -226,6 +229,17 @@ def _data_push(self, ki_project_resource, syn_parent):
         sys_path = SysPath(ki_project_resource.abs_path, rel_start=kiproject.local_path)
         syn_entity = None
 
+        # check for valid characters in string
+        for rel_part in sys_path.rel_parts:
+            if not re.match(self.VALID_NAME_REGEX, rel_part):
+                # TODO: Add logging statement
+                message = (
+                    "Invalid character(s) found in path part '{part}' (from full path '{full}'). "
+                    "Paths may only contain letters, numbers, spaces, underscores, hypens, periods, plus signs, and parentheses. "
+                    "Please correct the filename and update any instances in kiproject.json."
+                    ).format(part=rel_part, full=sys_path.abs_path)
+                raise InvalidNameError(message)
+
         if sys_path.is_dir:
             # Find or create the folder in Synapse.
             syn_entity = self._find_or_create_syn_folder(syn_parent, sys_path.basename)
@@ -234,8 +248,8 @@ def _data_push(self, ki_project_resource, syn_parent):
             self._push_children(ki_project_resource.root_resource or ki_project_resource, syn_entity, sys_path.abs_path)
         else:
             # Upload the file
-            syn_entity = SynapseAdapter.client().store(synapseclient.File(path=sys_path.abs_path, parent=syn_parent),
-                                                       forceVersion=False)
+            syn_file =  synapseclient.File(path=sys_path.abs_path, parent=syn_parent)
+            syn_entity = SynapseAdapter.client().store(syn_file,forceVersion=False)
 
         has_changes = False
 

tests/kitools/data_adapters/synapse/test_synapse_adapter.py

@@ -16,8 +16,12 @@
 import os
 import responses
 from src.kitools.data_adapters import SynapseAdapter
+from src.kitools.data_adapters.synapse.exceptions import InvalidNameError
+from src.kitools.ki_project_resource import KiProjectResource
 import synapseclient
 
+from ...test_ki_project import kiproject
+
 
 def test_name():
     assert SynapseAdapter().name() == 'Synapse'
@@ -47,4 +51,12 @@ def test_connected():
         rsps.replace(responses.GET, 'https://repo-prod.prod.sagebase.org/repo/v1/userProfile', status=418)
         assert SynapseAdapter().connected() is False
 
+
+@pytest.mark.xfail(raises=InvalidNameError)
+def test_it_fails_with_invalid_filename(tmp_path, kiproject):
+    f = tmp_path / 'il^egal.dat'
+    f.write_text(u'\x00')
+    kiproject_resource = KiProjectResource(kiproject, local_path=str(f))
+    SynapseAdapter().data_push(ki_project_resource=kiproject_resource)
+
 # TODO: add remaining tests.