Added support for OCS Recipient API with unit test

owncloud/owncloud.py

@@ -57,6 +57,57 @@ def __init__(self, res):
         ResponseError.__init__(self, res, "HTTP")
 
 
+class Recipient(object):
+    """Recipient information"""
+
+    def __init__(self, recipient_info):
+        self.recipient_info = {}
+        for k, v in recipient_info.items():
+            self.recipient_info[k] = v
+
+    def get_label(self):
+        """Return the label of the recipient.
+        Display name for users.
+
+        :returns: label of recipient
+        """
+        return self.recipient_info['label']
+
+    def get_share_type(self):
+        """Returns the type of the recipient.
+        See OCS_SHARE_TYPE_* constants.
+
+        :returns: share type
+        """
+        return self.recipient_info['value']['shareType']
+
+    def is_user(self):
+        """Returns whether the recipient is an user or not
+
+        :returns: true if recipient is an user
+        """
+        return self.recipient_info['value']['shareType'] == Client.OCS_SHARE_TYPE_USER
+
+    def get_share_with(self):
+        """Returns the share recipient.
+        If share type is OCS_SHARE_TYPE_USER, then the recipient is the name of
+        the user.
+        For OCS_SHARE_TYPE_GROUP it is the name of the group.
+
+        :returns: name of the share recipient
+        """
+        return self.recipient_info['value']['shareWith']
+
+    def __str__(self):
+        info = ''
+        for k, v in self.recipient_info.items():
+            info += '%s=%s,' % (k, v)
+        return 'Recipient(%s)' % info[:-1]
+
+    def __repr__(self):
+        return self.__str__()
+
+
 class ShareInfo(object):
     """Share information"""
 
@@ -324,6 +375,9 @@ class Client(object):
     OCS_SHARE_TYPE_LINK = 3
     OCS_SHARE_TYPE_REMOTE = 6
 
+    OCS_ITEM_TYPE_FOLDER = "folder"
+    OCS_ITEM_TYPE_FILE = "file"
+
     def __init__(self, url, **kwargs):
         """Instantiates a client
 
@@ -1008,6 +1062,31 @@ def get_shares(self, path='', **kwargs):
             return shares
         raise HTTPResponseError(res)
 
+    def get_sharees(self, search, share_type, item_type=OCS_ITEM_TYPE_FOLDER):
+        """Get all share recipients for the provided search term.
+
+        :param search: The search string
+        :param item_type: OCS_ITEM_TYPE_*
+        :param share_type: OCS_SHARE_TYPE_*
+        :returns: list of Recipient class
+        :raises: ResponseError in case an HTTP error status was returned
+        """
+        action_path = 'sharees?itemType={}'.format(item_type)
+        if search is not None:
+            action_path += '&search={}'.format(search)
+        if share_type is not None:
+            action_path += '&shareType={}'.format(share_type)
+        res = self._make_ocs_request(
+                'GET',
+                self.OCS_SERVICE_SHARE,
+                action_path
+                )
+        if res.status_code == 200:
+            tree = ET.fromstring(res.content)
+            self._check_ocs_status(tree)
+            return self._get_recipients(tree.find('data'))
+        raise HTTPResponseError(res)
+
     def create_user(self, user_name, initial_password):
         """Create a new user with an initial password via provisioning API.
         It is not an error, if the user already existed before.
@@ -1924,6 +2003,14 @@ def _get_shareinfo(self, data_el):
             return None
         return ShareInfo(self._xml_to_dict(data_el))
 
+    def _get_recipients(self, data_el):
+        recipients = []
+        if (data_el is None) or not (isinstance(data_el, ET.Element)):
+            return recipients
+        for element in data_el.findall('.//element'):
+            recipients.append(Recipient(self._xml_to_dict(element)))
+        return recipients
+
     def _update_capabilities(self):
         res = self._make_ocs_request(
                 'GET',

owncloud/test/test.py

@@ -1557,5 +1557,47 @@ def test_drop_file(self, file_name):
         self.assertIsNotNone(file_info)
         self.assertEqual(file_info.get_size(), 2 * 1024)
 
+
+class TestRecipientAPI(unittest.TestCase):
+
+    def setUp(self):
+        self.client = owncloud.Client(Config['owncloud_url'])
+        self.client.login(Config['owncloud_login'], Config['owncloud_password'])
+        self.test_group = Config['test_group']
+        self.share2user = Config['owncloud_share2user']
+        self.share2userPwd = 'Avcpwd4l!'
+        try:
+            self.client.create_user(self.share2user, self.share2userPwd)
+        except:
+            self.share2user = None
+        try:
+            self.client.create_group(self.test_group)
+        except:
+            self.test_group = None
+
+    def tearDown(self):
+        try:
+            self.client.delete_user(self.share2user)
+        except:
+            pass
+        try:
+            self.client.delete_group(self.test_group)
+        except:
+            pass
+        self.client.logout()
+
+    def test_get_recipients(self):
+        users = self.client.get_sharees("", share_type=self.client.OCS_SHARE_TYPE_USER)
+        users = [recipient.get_share_with() for recipient in users]
+        self.assertTrue(Config['owncloud_login'] in users)
+        if self.share2user is not None:
+            self.assertTrue(self.share2user in users)
+
+        if self.test_group:
+            groups = self.client.get_sharees("", share_type=self.client.OCS_SHARE_TYPE_GROUP)
+            groups = [recipient.get_share_with() for recipient in groups]
+            self.assertTrue(self.test_group in groups)
+
+
 if __name__ == '__main__':
     unittest.main()