ensure destination is writeable target, update to pass exceptions

Author: saimonation

cterasdk/asynchronous/core/files/io.py

@@ -95,8 +95,7 @@ async def move(core, *paths, destination=None, resolver=None, cursor=None):
 
 async def ensure_directory(core, directory, suppress_error=False):
     present, resource = await metadata(core, directory, suppress_error=True)
-    if (not present or not resource.isFolder) and not suppress_error:
-        raise NotADirectory(directory.absolute)
+    fs.ensure_directory(present, resource, directory, suppress_error)
     return resource.isFolder if present else False, resource
 
 
@@ -146,7 +145,9 @@ async def _validate_destination(core, name, destination):
     is_dir, resource = await ensure_directory(core, destination, suppress_error=True)
     if not is_dir:
         is_dir, resource = await ensure_directory(core, destination.parent)
+        fs.ensure_writeable(resource, destination.parent)
         return resource.cloudFolderInfo.uid, destination.name, destination.parent
+    fs.ensure_writeable(resource, destination)
     return resource.cloudFolderInfo.uid, name, destination
 
 

cterasdk/cio/core.py

@@ -4,10 +4,11 @@
 from ..objects.uri import quote, unquote
 from ..common import Object, DateTimeUtils
 from ..core.enum import ProtectionLevel, CollaboratorType, SearchType, PortalAccountType, FileAccessMode, \
-    UploadError
+    UploadError, ResourceAction, ResourceScope
 from ..core.types import PortalAccount, UserAccount, GroupAccount
 from ..exceptions.io import ResourceExistsError, PathValidationError, NameSyntaxError, \
-    ReservedNameError, RestrictedRoot, InsufficientPermission, FileConflict, RemoteStorageError
+    ReservedNameError, RestrictedRoot, PermissionDenied, FileConflict, RemoteStorageError, NotADirectory, \
+    UnwriteableScope
 from ..exceptions.io import UploadException, OutOfQuota, RejectedByPolicy, NoStorageBucket, WindowsACLError
 from ..lib.iterator import DefaultResponse
 from . import common
@@ -316,12 +317,24 @@ def handle(path):
 
 
 def destination_prerequisite_conditions(destination, name):
-    if not len(destination.reference.parts) > 0:
-        raise RestrictedRoot()
     if any(c in name for c in ['\\', '/', ':', '?', '&', '<', '>', '"', '|']):
         raise NameSyntaxError(destination.join(name).reference.as_posix())
 
 
+def ensure_directory(present, resource, directory, suppress_error):
+    if (not present or not resource.isFolder) and not suppress_error:
+        raise NotADirectory(directory.reference.as_posix())
+
+
+def ensure_writeable(resource, directory):
+    if resource.scope == ResourceScope.Root:
+        raise RestrictedRoot()
+    if resource.scope not in [ResourceScope.Personal, ResourceScope.Project, ResourceScope.InsideCloudFolder]:
+        raise UnwriteableScope(directory.reference.as_posix(), resource.scope)
+    if resource.permission != FileAccessMode.RW:
+        raise PermissionDenied(directory.reference.as_posix(), ResourceAction.Write)
+
+
 @contextmanager
 def upload(core, name, destination, size, fd):
     fd, size = common.encode_stream(fd, size)
@@ -555,7 +568,7 @@ def obtain_current_accounts(param):
 
 file_access_errors = {
     "Conflict": FileConflict,
-    "PermissionDenied": InsufficientPermission,
+    "PermissionDenied": PermissionDenied,
     "DestinationNotExists": PathValidationError,
     "FileWithTheSameNameExist": ResourceExistsError,
     "InvalidName": NameSyntaxError,
@@ -572,7 +585,7 @@ def await_or_future(core, ref, wait):
     """
     if wait:
         task = core.tasks.wait(ref)
-        accept_error(task.error_type, action=task.name.lower(), name=task.cursor.destResource.name, cursor=task.cursor)
+        accept_error(task.error_type, **error_metadata(task))
         return task
     return core.tasks.awaitable_task(ref)
 
@@ -586,11 +599,22 @@ async def a_await_or_future(ctera, ref, wait):
     """
     if wait:
         task = await ctera.tasks.wait(ref)
-        accept_error(task.error_type, action=task.name.lower(), name=task.cursor.destResource.name, cursor=task.cursor)
+        accept_error(task.error_type, **error_metadata(task))
         return task
     return ctera.tasks.awaitable_task(ref)
 
 
+def error_metadata(task):
+    metadata = dict(
+        action=task.name
+    )
+    if task.name in [ResourceAction.Copy, ResourceAction.Move]:
+        metadata.update(dict(cursor=task.cursor))
+        if task.cursor.destResource:
+            metadata.update(dict(name=task.cursor.destResource.name))
+    return metadata
+
+
 def accept_error(error_type, **kwargs):
     """
     Check if response contains an error.
@@ -611,6 +635,9 @@ def accept_error(error_type, **kwargs):
         except ReservedNameError as error:
             logger.error('Reserved name error: the name is reserved and cannot be used.')
             raise error
+        except PermissionDenied as error:
+            logger.error('Permission denied: Inappropriate permissions to access this resource.')
+            raise error
         except FileConflict as error:
             logger.error('Conflict: a file with the same name already exists: %s', error.name)
             raise error

cterasdk/core/enum.py

@@ -676,3 +676,54 @@ class UploadError:
     RejectedByPolicy = "Rejected by Cloud Drive policy rule"
     NoStorageBucket = "No available storage location"
     WindowsACL = "Illegal access to NTACL folder"
+
+
+class ResourceAction:
+    """
+    Resource Action
+
+    :ivar str Delete: Delete.
+    :ivar str Copy: Copy.
+    :ivar str Move: Move.
+    :ivar str Undelete: Undelete.
+    """
+    Delete = 'Delete'
+    Undelete = 'Undelete'
+    Copy = 'Copy'
+    Move = 'Move'
+    Write = 'Write'
+
+
+class ResourceScope:
+    """
+    Resource Scope
+
+    :ivar str Root: Root.
+    :ivar str ProjectsContainer: ProjectsContainer.
+    :ivar str Project: Project.
+    :ivar str SharedContainer: SharedContainer.
+    :ivar str SharedDomain: SharedDomain.
+    :ivar str Shared: Shared.
+    :ivar str BackupsContainer: BackupsContainer.
+    :ivar str Backup: Backup.
+    :ivar str Personal: Personal.
+    :ivar str UsersContainer: UsersContainer.
+    :ivar str UsersFoldersContainer: UsersFoldersContainer.
+    :ivar str CloudDrivesContainer: CloudDrivesContainer.
+    :ivar str OfflineFolder: OfflineFolder.
+    :ivar str InsideCloudFolder: InsideCloudFolder.
+    """
+    Root = "Root"
+    ProjectsContainer = "ProjectsContainer"
+    Project = "Project"
+    SharedContainer = "SharedContainer"
+    SharedDomain = "SharedDomain"
+    Shared = "Shared"
+    BackupsContainer = "BackupsContainer"
+    Backup = "Backup"
+    Personal = "Personal"
+    UsersContainer = "UsersContainer"
+    UsersFoldersContainer = "UsersFoldersContainer"
+    CloudDrivesContainer = "CloudDrivesContainer"
+    OfflineFolder = "OfflineFolder"
+    InsideCloudFolder = "InsideCloudFolder"

cterasdk/core/files/io.py

@@ -1,7 +1,7 @@
 import logging
 from ...cio.common import encode_request_parameter
 from ...cio import core as fs
-from ...exceptions.io import ResourceNotFoundError, ResourceExistsError, NotADirectory
+from ...exceptions.io import ResourceNotFoundError, ResourceExistsError
 from ...core import query
 from ..enum import CollaboratorType
 
@@ -97,8 +97,7 @@ def move(core, *paths, destination=None, resolver=None, cursor=None):
 
 def ensure_directory(core, directory, suppress_error=False):
     present, resource = metadata(core, directory, suppress_error=True)
-    if (not present or not resource.isFolder) and not suppress_error:
-        raise NotADirectory(directory.absolute)
+    fs.ensure_directory(present, resource, directory, suppress_error)
     return resource.isFolder if present else False, resource
 
 
@@ -148,7 +147,9 @@ def _validate_destination(core, name, destination):
     is_dir, resource = ensure_directory(core, destination, suppress_error=True)
     if not is_dir:
         is_dir, resource = ensure_directory(core, destination.parent)
+        fs.ensure_writeable(resource, destination.parent)
         return resource.cloudFolderInfo.uid, destination.name, destination.parent
+    fs.ensure_writeable(resource, destination)
     return resource.cloudFolderInfo.uid, name, destination
 
 

cterasdk/exceptions/io.py

@@ -26,14 +26,14 @@ def __init__(self, path):
 
 class ResourceExistsError(RemoteStorageError):
 
-    def __init__(self):
-        super().__init__('Resource already exists: a file or folder with this name already exists.')
+    def __init__(self, path):
+        super().__init__('Resource already exists: a file or folder with this name already exists.', path)
 
 
 class PathValidationError(RemoteStorageError):
 
-    def __init__(self):
-        super().__init__('Path validation failed: the specified destination path does not exist.')
+    def __init__(self, path=None, **kwargs):
+        super().__init__('Path validation failed: the specified destination path does not exist.', path)
 
 
 class NameSyntaxError(RemoteStorageError):
@@ -44,8 +44,8 @@ def __init__(self, path):
 
 class ReservedNameError(RemoteStorageError):
 
-    def __init__(self):
-        super().__init__('Reserved name error: the name is reserved and cannot be used.')
+    def __init__(self, path):
+        super().__init__('Reserved name error: the name is reserved and cannot be used.', path)
 
 
 class RestrictedPathError(RemoteStorageError):
@@ -60,10 +60,18 @@ def __init__(self):
         super().__init__('Storing files to the root directory is forbidden.', '/')
 
 
-class InsufficientPermission(RemoteStorageError):
+class PermissionDenied(RemoteStorageError):
 
-    def __init__(self):
-        super().__init__('Permission denied: You must have appropriate permissions to access this resource.')
+    def __init__(self, action, path=None):
+        super().__init__('Permission denied: Inappropriate permissions to access this resource.', path)
+        self.action = action
+
+
+class UnwriteableScope(RemoteStorageError):
+
+    def __init__(self, path, scope):
+        super().__init__("Write access denied. This target is protected and cannot be modified.", path)
+        self.scope = scope
 
 
 class FileConflict(RemoteStorageError):