diff --git a/owncloud/owncloud.py b/owncloud/owncloud.py index 7e3e804..911b949 100644 --- a/owncloud/owncloud.py +++ b/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', diff --git a/owncloud/test/test.py b/owncloud/test/test.py index 7c427c8..745c10d 100644 --- a/owncloud/test/test.py +++ b/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()