catch permission errors on delete and undelete operations

add graceful errors when attempting to access edge modules remotely

Author: saimonation

cterasdk/cio/common.py

@@ -110,6 +110,9 @@ def __getitem__(self, key):
     def __eq__(self, p):
         return self.absolute == p.absolute
 
+    def __repr__(self):
+        return str(f"'{self}'")
+
     def __str__(self):
         return self.relative
 

cterasdk/cio/core/commands.py

@@ -1027,25 +1027,40 @@ def _before_command(self):
     def _task_complete(self, task):
         return [str(path) for path in self.paths]
 
+    @property
+    @abstractmethod
+    def _error_object(self):
+        raise NotImplementedError('Subclass must implement the "_error_object" property.')
+
+    def _task_error(self, task):
+        cursor = task.cursor
+        error = self._error_object(self.paths, cursor)
+
+        if task.error_type == ResourceError.PermissionDenied:
+            resource = automatic_resolution(cursor).relative
+            raise error from exceptions.io.core.PrivilegeError(resource)
+
+        raise error
+
 
 class Delete(MultiResourceCommand):
 
     def _progress_str(self):
         return 'Deleting'
 
-    def _task_error(self, task):
-        cursor = task.cursor
-        raise exceptions.io.core.DeleteError(self.paths, cursor)
+    @property
+    def _error_object(self):
+        return exceptions.io.core.DeleteError
 
 
 class Recover(MultiResourceCommand):
 
     def _progress_str(self):
         return 'Recovering'
 
-    def _task_error(self, task):
-        cursor = task.cursor
-        raise exceptions.io.core.RecoverError(self.paths, cursor)
+    @property
+    def _error_object(self):
+        return exceptions.io.core.RecoverError
 
 
 class ResolverCommand(TaskCommand):
@@ -1132,7 +1147,7 @@ def _task_error(self, task):
             )
             raise error from exceptions.io.core.FolderNotFoundError(directory.relative)
 
-        raise self._error_object(self.paths, cursor)
+        raise error
 
 
 class Copy(ResolverCommand):

cterasdk/clients/clients.py

@@ -436,3 +436,24 @@ def new():
         async def new_response(response):
             return SyncResponse(response)
         return new_response
+
+
+class RestrictedAPI:
+
+    def __init__(self, name):
+        self._name = name
+
+    def _raise_error(self):
+        raise RuntimeError(f'{self._name} is unavailable in the current context.')
+
+    def __getattr__(self, _):
+        self._raise_error()
+
+    def __call__(self, *_args, **_kwargs):
+        self._raise_error()
+
+    def __bool__(self):
+        self._raise_error()
+
+    def __repr__(self):
+        return f'<{self._name} is unavailable in the current context>'

cterasdk/objects/synchronous/edge.py

@@ -23,7 +23,10 @@ def __init__(self, edge, Portal):
             edge._Portal = Portal
             edge.default.close()
             edge._ctera_session.start_remote_session(Portal.session())
+            self.migrate = clients.RestrictedAPI('migrate')
             self.api = Portal.default.clone(clients.API, EndpointBuilder.new(edge.base), authenticator=lambda *_: True)
+            self.stats = clients.RestrictedAPI('stats')
+            self.io = clients.RestrictedAPI('io')
         else:
             self.migrate = edge.default.clone(clients.Migrate, EndpointBuilder.new(edge.base, '/migration/rest/v1'))
             self.api = edge.default.clone(clients.API, EndpointBuilder.new(edge.base, '/admingui/api'))