Support assigning domain users as owners of folders (#22)

* Add named tuple (UserAccount) to core/users.py
Add support for local and domain user lookup
Add support for adding folder groups owned by a domain user
Add support for adding a cloud drive folder owned by a domain user
Add support for deleting a cloud drive folder owned by a domain user
Add support for undeleting a cloud drive folder owned by a domain user
Update the cloudfs tests to support the updated input scheme
Update documentation
* Remove unnecessary spaces and remove unused variables
* Resolve documentation typo (patam to param)
* Use get_multi instead of query in core/users.get
* Use a Class instead of a namedtuple to represent a UserAccount

* Add comment to transition the code to use users.UserAccount once we expose the class to the client

cterasdk/core/cloudfs.py

@@ -26,13 +26,13 @@ def mkfg(self, name, user=None):
         Create a new Folder Group
 
         :param str name: Name of the new folder group
-        :param str user: User name of the new folder group owner (default to None)
+        :param UserAccount user: User account, the user directory and name of the new folder group owner (default to None)
         """
 
         param = Object()
         param.name = name
         param.disabled = True
-        param.owner = self._portal.get('/users/' + user + '/baseObjectRef') if user is not None else None
+        param.owner = self._portal.users.get(user, ['baseObjectRef']).baseObjectRef if user is not None else None
 
         try:
             response = self._portal.execute('', 'createFolderGroup', param)
@@ -59,11 +59,11 @@ def mkdir(self, name, group, owner, winacls=True):
 
         :param str name: Name of the new directory
         :param str group: The Folder Group to which the directory belongs
-        :param str owner: User name of the owner of the new directory
+        :param UserAccount owner: User account, the owner of the new directory
         :param bool,optional winacls: Use Windows ACLs, defaults to True
         """
 
-        owner = self._portal.get('/users/' + owner + '/baseObjectRef')
+        owner = self._portal.users.get(owner, ['baseObjectRef']).baseObjectRef
         group = self._portal.get('/foldersGroups/' + group + '/baseObjectRef')
 
         param = Object()
@@ -91,7 +91,7 @@ def delete(self, name, owner):
         Delete a Cloud Drive Folder
 
         :param str name: Name of the Cloud Drive Folder to delete
-        :param str owner: User name of the owner of the Cloud Drive Folder to delete
+        :param UserAccount owner: User account, the owner of the Cloud Drive Folder to delete
         """
 
         path = self._dirpath(name, owner)
@@ -103,7 +103,7 @@ def undelete(self, name, owner):
         Un-Delete a Cloud Drive Folder
 
         :param str name: Name of the Cloud Drive Folder to un-delete
-        :param str owner: User name of the owner of the Cloud Drive Folder to un-delete
+        :param UserAccount owner: User account, the owner of the Cloud Drive Folder to delete
         """
         path = self._dirpath(name, owner)
         logging.getLogger().info('Restoring cloud drive folder. %s', {'path': path})
@@ -145,6 +145,6 @@ def find(self, name, owner, include):
         raise CTERAException('Could not find cloud folder', None, folder=name, owner=owner)
 
     def _dirpath(self, name, owner):
-        owner = self._portal.get('/users/' + owner + '/displayName')
+        owner = self._portal.users.get(owner, ['displayName']).displayName
         path = owner + '/' + name
         return path

cterasdk/core/users.py

@@ -1,6 +1,7 @@
 import logging
 
 from .base_command import BaseCommand
+from ..exception import CTERAException
 from ..common import Object
 from . import query
 from . import union
@@ -13,6 +14,23 @@ class Users(BaseCommand):
 
     _default_fields = ['name']
 
+    def get(self, user_account, include=None):
+        """
+        Get a user account
+
+        :param UserAccount user_account: User account, including the user directory and user name
+        :param list[str] include: List of fields to retrieve, defaults to ['name']
+        :return: The user account, including the requested fields
+        """
+        directory, user = user_account  # transition this code to use users.UserAccount once exposed to the client
+        baseurl = ('/users/' + user) if directory == 'local' else '/domains/' + directory + '/adUsers/' + user
+        include = union.union(include or [], Users._default_fields)
+        include = ['/' + attr for attr in include]
+        user_object = self._portal.get_multi(baseurl, include)
+        if user_object.name is None:
+            raise CTERAException('Could not find user', None, user_directory=directory, username=user)
+        return user_object
+
     def list_local_users(self, include=None):
         """
         List all local users
@@ -90,3 +108,21 @@ def delete(self, name):
         logging.getLogger().info('User deleted. %s', {'user': name})
 
         return response
+
+
+class UserAccount:
+    """
+    Portal User Account
+
+    :ivar cterasdk.core.users.UserAccount.name name: The user name
+    :ivar cterasdk.core.users.UserAccount.directory directory: The fully-qualified name of the user directory, defaults to None
+    :ivar cterasdk.core.users.UserAccount.is_local is_local: Boolean property to determine if the user account is local
+    """
+    def __init__(self, name, directory=None):
+        """
+        :param str name: The name of the Portal user
+        :param str directory: The the fully qualified domain name, defaults to None
+        """
+        self.name = name
+        self.directory = directory
+        self.is_local = not directory

docs/source/user_guides/Portal/GlobalAdmin.rst

@@ -415,9 +415,13 @@ Create a Folder Group
 
 .. code:: python
 
-   admin.cloudfs.mkfg('FG-001', 'svc_account')
+   """Create a Folder Group, owned by a local user account 'svc_account'"""
+   admin.cloudfs.mkfg('FG-001', ('local', 'svc_account'))
 
-   admin.cloudfs.mkfg('FG-002') # without an owner
+   """Create a Folder Group, owned by the domain user 'ctera.local\wbruce'"""
+   admin.cloudfs.mkfg('FG-002', ('ctera.local', 'wbruce'))
+
+   admin.cloudfs.mkfg('FG-003') # without an owner
 
 Delete a Folder Group
 ^^^^^^^^^^^^^^^^^^^^^
@@ -426,7 +430,7 @@ Delete a Folder Group
 
 .. code:: python
 
-   admin.cloudfs.rmfg('FG-002')
+   admin.cloudfs.rmfg('FG-001')
 
 Create a Cloud Drive Folder
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^
@@ -435,9 +439,13 @@ Create a Cloud Drive Folder
 
 .. code:: python
 
-   admin.cloudfs.mkdir('DIR-001', 'FG-001', 'svc_account')
+   """Create a Cloud Drive folder, owned by a local user account 'svc_account'"""
+   admin.cloudfs.mkdir('DIR-001', 'FG-001', ('local', 'svc_account'))
+
+   """Create a Cloud Drive folder, owned by the domain user 'ctera.local\wbruce'"""
+   admin.cloudfs.mkdir('DIR-002', 'FG-002', ('ctera.local', 'wbruce'))
 
-   admin.cloudfs.mkdir('DIR-002', 'FG-001', 'svc_account', winacls = False) # disable Windows ACL's
+   admin.cloudfs.mkdir('DIR-003', 'FG-003', 'svc_account', winacls = False) # disable Windows ACL's
 
 Delete a Cloud Drive Folder
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^
@@ -446,7 +454,11 @@ Delete a Cloud Drive Folder
 
 .. code:: python
 
-   admin.cloudfs.delete('DIR-001', 'svc_account')
+   """Delete a Cloud Drive folder, owned by the local user account 'svc_account'"""
+   admin.cloudfs.delete('DIR-001', ('local', 'svc_account'))
+
+   """Delete a Cloud Drive folder, owned by the domain user 'ctera.local\wbruce'"""
+   admin.cloudfs.delete('DIR-002', ('ctera.local', 'wbruce'))
 
 Recover a Cloud Drive Folder
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
@@ -455,4 +467,8 @@ Recover a Cloud Drive Folder
 
 .. code:: python
 
-   admin.cloudfs.undelete('DIR-001', 'svc_account')
+   """Recover a deleted Cloud Drive folder, owned by the local user account 'svc_account'"""
+   admin.cloudfs.undelete('DIR-001', ('local', 'svc_account'))
+
+   """Recover a deleted Cloud Drive folder, owned by the domain user 'ctera.local\wbruce'"""
+   admin.cloudfs.undelete('DIR-002', ('ctera.local', 'wbruce'))

tests/ut/test_core_cloudfs.py

@@ -1,6 +1,7 @@
 from unittest import mock
 
 from cterasdk import exception
+from cterasdk.core import users
 from cterasdk.core import cloudfs
 from cterasdk.common import Object
 from tests.ut import base_core
@@ -11,6 +12,7 @@ class TestCoreCloudFS(base_core.BaseCoreTest):
     def setUp(self):
         super().setUp()
         self._owner = 'admin'
+        self._local_user_account = users.UserAccount('local', self._owner)
         self._group = 'admin'
         self._name = 'folderGroup'
 
@@ -27,11 +29,13 @@ def test_mkfg_no_owner(self):
 
         self.assertEqual(ret, 'Success')
 
-    def test_mkfg_with_owner(self):
-        self._init_global_admin(get_response=self._owner, execute_response='Success')
-        ret = cloudfs.CloudFS(self._global_admin).mkfg(self._name, user=self._owner)
+    def test_mkfg_with_local_owner(self):
+        self._init_global_admin(execute_response='Success')
+        self._mock_get_user_base_object_ref()
+
+        ret = cloudfs.CloudFS(self._global_admin).mkfg(self._name, user=self._local_user_account)
 
-        self._global_admin.get.assert_called_once_with('/users/' + self._owner + '/baseObjectRef')
+        self._global_admin.users.get.assert_called_once_with(self._local_user_account, ['baseObjectRef'])
         self._global_admin.execute.assert_called_once_with('', 'createFolderGroup', mock.ANY)
 
         expected_param = self._get_mkfg_object(with_owner=True)
@@ -40,36 +44,27 @@ def test_mkfg_with_owner(self):
 
         self.assertEqual(ret, 'Success')
 
-    def test_mkfg_raise(self):
+    def test_mkfg_no_owner_raise(self):
         error_message = "Expected Failure"
         expected_exception = exception.CTERAException(message=error_message)
         self._global_admin.execute = mock.MagicMock(side_effect=expected_exception)
         with self.assertRaises(exception.CTERAException) as error:
             cloudfs.CloudFS(self._global_admin).mkfg(self._name)
         self.assertEqual(error_message, error.exception.message)
 
-    def _get_mkfg_object(self, with_owner=False):
-        mkfg_param_object = Object()
-        mkfg_param_object.name = self._name
-        mkfg_param_object.disabled = True
-        mkfg_param_object.owner = self._owner if with_owner else None
-        return mkfg_param_object
-
     def test_rmfg(self):
         self._init_global_admin(execute_response='Success')
         cloudfs.CloudFS(self._global_admin).rmfg(self._name)
         self._global_admin.execute.assert_called_once_with('/foldersGroups/' + self._name, 'deleteGroup', True)
 
-    def test_mkdir_no_winacls_param(self):
-        get_response = 'admin'
-        self._init_global_admin(get_response=get_response, execute_response='Success')
-        ret = cloudfs.CloudFS(self._global_admin).mkdir(self._name, self._group, self._owner)
-        self._global_admin.get.assert_has_calls(
-            [
-                mock.call(('/users/' + self._owner + '/baseObjectRef')),
-                mock.call(('/foldersGroups/' + self._group + '/baseObjectRef'))
-            ]
-        )
+    def test_mkdir_with_local_owner_no_winacls_param(self):
+        self._init_global_admin(get_response='admin', execute_response='Success')
+        self._mock_get_user_base_object_ref()
+
+        ret = cloudfs.CloudFS(self._global_admin).mkdir(self._name, self._group, self._local_user_account)
+
+        self._global_admin.users.get.assert_called_once_with(self._local_user_account, ['baseObjectRef'])
+        self._global_admin.get.assert_called_once_with('/foldersGroups/' + self._group + '/baseObjectRef')
         self._global_admin.execute.assert_called_once_with('', 'addCloudDrive', mock.ANY)
 
         expected_param = self._get_mkdir_object()
@@ -78,16 +73,15 @@ def test_mkdir_no_winacls_param(self):
 
         self.assertEqual(ret, 'Success')
 
-    def test_mkdir_winacls_true(self):
+    def test_mkdir_with_local_owner_winacls_true(self):
         get_response = 'admin'
         self._init_global_admin(get_response=get_response, execute_response='Success')
-        ret = cloudfs.CloudFS(self._global_admin).mkdir(self._name, self._group, self._owner, True)
-        self._global_admin.get.assert_has_calls(
-            [
-                mock.call(('/users/' + self._owner + '/baseObjectRef')),
-                mock.call(('/foldersGroups/' + self._group + '/baseObjectRef'))
-            ]
-        )
+        self._mock_get_user_base_object_ref()
+
+        ret = cloudfs.CloudFS(self._global_admin).mkdir(self._name, self._group, self._local_user_account, True)
+
+        self._global_admin.users.get.assert_called_once_with(self._local_user_account, ['baseObjectRef'])
+        self._global_admin.get.assert_called_once_with('/foldersGroups/' + self._group + '/baseObjectRef')
         self._global_admin.execute.assert_called_once_with('', 'addCloudDrive', mock.ANY)
 
         expected_param = self._get_mkdir_object()
@@ -96,16 +90,15 @@ def test_mkdir_winacls_true(self):
 
         self.assertEqual(ret, 'Success')
 
-    def test_mkdir_winacls_false(self):
+    def test_mkdir_with_local_owner_winacls_false(self):
         get_response = 'admin'
         self._init_global_admin(get_response=get_response, execute_response='Success')
-        ret = cloudfs.CloudFS(self._global_admin).mkdir(self._name, self._group, self._owner, False)
-        self._global_admin.get.assert_has_calls(
-            [
-                mock.call(('/users/' + self._owner + '/baseObjectRef')),
-                mock.call(('/foldersGroups/' + self._group + '/baseObjectRef'))
-            ]
-        )
+        self._mock_get_user_base_object_ref()
+
+        ret = cloudfs.CloudFS(self._global_admin).mkdir(self._name, self._group, self._local_user_account, False)
+
+        self._global_admin.users.get.assert_called_once_with(self._local_user_account, ['baseObjectRef'])
+        self._global_admin.get.assert_called_once_with('/foldersGroups/' + self._group + '/baseObjectRef')
         self._global_admin.execute.assert_called_once_with('', 'addCloudDrive', mock.ANY)
 
         expected_param = self._get_mkdir_object(winacls=False)
@@ -114,22 +107,47 @@ def test_mkdir_winacls_false(self):
 
         self.assertEqual(ret, 'Success')
 
-    def test_mkdir_raise(self):
+    def test_mkdir_with_local_owner_raise(self):
         get_response = 'admin'
         self._init_global_admin(get_response=get_response, execute_response='Success')
+        self._mock_get_user_base_object_ref()
+
         error_message = "Expected Failure"
         expected_exception = exception.CTERAException(message=error_message)
         self._global_admin.execute = mock.MagicMock(side_effect=expected_exception)
         with self.assertRaises(exception.CTERAException) as error:
-            cloudfs.CloudFS(self._global_admin).mkdir(self._name, self._group, self._owner)
-        self._global_admin.get.assert_has_calls(
-            [
-                mock.call(('/users/' + self._owner + '/baseObjectRef')),
-                mock.call(('/foldersGroups/' + self._group + '/baseObjectRef'))
-            ]
-        )
+            cloudfs.CloudFS(self._global_admin).mkdir(self._name, self._group, self._local_user_account)
+
+        self._global_admin.users.get.assert_called_once_with(self._local_user_account, ['baseObjectRef'])
+        self._global_admin.get.assert_called_once_with('/foldersGroups/' + self._group + '/baseObjectRef')
+
         self.assertEqual(error_message, error.exception.message)
 
+    def test_delete_with_local_owner(self):
+        self._init_global_admin(get_response=self._owner)
+        self._mock_get_user_display_name()
+
+        self._global_admin.files.delete = mock.MagicMock(return_value='Success')
+        cloudfs.CloudFS(self._global_admin).delete(self._name, self._local_user_account)
+        self._global_admin.users.get.assert_called_once_with(self._local_user_account, ['displayName'])
+        self._global_admin.files.delete.assert_called_once_with(self._owner + '/' + self._name)
+
+    def test_undelete_with_local_owner(self):
+        self._init_global_admin(get_response=self._owner)
+        self._mock_get_user_display_name()
+
+        self._global_admin.files.undelete = mock.MagicMock(return_value='Success')
+        cloudfs.CloudFS(self._global_admin).undelete(self._name, self._local_user_account)
+        self._global_admin.users.get.assert_called_once_with(self._local_user_account, ['displayName'])
+        self._global_admin.files.undelete.assert_called_once_with(self._owner + '/' + self._name)
+
+    def _get_mkfg_object(self, with_owner=False):
+        mkfg_param_object = Object()
+        mkfg_param_object.name = self._name
+        mkfg_param_object.disabled = True
+        mkfg_param_object.owner = self._owner if with_owner else None
+        return mkfg_param_object
+
     def _get_mkdir_object(self, winacls=True):
         mkdir_param_object = Object()
         mkdir_param_object.name = self._name
@@ -138,16 +156,20 @@ def _get_mkdir_object(self, winacls=True):
         mkdir_param_object.enableSyncWinNtExtendedAttributes = winacls
         return mkdir_param_object
 
-    def test_delete(self):
-        self._init_global_admin(get_response=self._owner)
-        self._global_admin.files.delete = mock.MagicMock(return_value='Success')
-        cloudfs.CloudFS(self._global_admin).delete(self._name, self._owner)
-        self._global_admin.get.assert_called_once_with('/users/' + self._owner + '/displayName')
-        self._global_admin.files.delete.assert_called_once_with(self._owner + '/' + self._name)
+    def _get_user_object(self, **kwargs):
+        user_account = Object()
+        user_account.name = self._owner
+        for key, value in kwargs.items():
+            setattr(user_account, key, value)
+        return user_account
 
-    def test_undelete(self):
-        self._init_global_admin(get_response=self._owner)
-        self._global_admin.files.undelete = mock.MagicMock(return_value='Success')
-        cloudfs.CloudFS(self._global_admin).undelete(self._name, self._owner)
-        self._global_admin.get.assert_called_once_with('/users/' + self._owner + '/displayName')
-        self._global_admin.files.undelete.assert_called_once_with(self._owner + '/' + self._name)
+    def _mock_get_user_base_object_ref(self):
+        user_object = self._get_user_object(baseObjectRef=self._owner)
+        self._mock_get_user(user_object)
+
+    def _mock_get_user_display_name(self):
+        user_object = self._get_user_object(displayName=self._owner)
+        self._mock_get_user(user_object)
+
+    def _mock_get_user(self, return_value):
+        self._global_admin.users.get = mock.MagicMock(return_value=return_value)