diff --git a/owncloud/owncloud.py b/owncloud/owncloud.py index 3f90def..2a8b583 100644 --- a/owncloud/owncloud.py +++ b/owncloud/owncloud.py @@ -129,18 +129,27 @@ def get_share_time(self): return datetime.datetime.fromtimestamp( self.__get_int('stime') ) - - def get_expiration(self): + + def get_expiration_datetime(self): """Returns the expiration date. :returns: expiration date :rtype: datetime object """ - exp = self.__get_int('expiration') + exp = self.share_info['expiration'] if exp is not None: - return datetime.datetime.fromtimestamp( - exp - ) + return datetime.datetime.strptime(exp, '%Y-%m-%d %H:%M:%S') + return None + + def get_expiration(self): + """Returns the expiration date. + + :returns: expiration date in YYYY-MM-DD hh:mm:ss format + :rtype: string + """ + exp = self.share_info['expiration'] + if exp is not None: + return exp return None def get_token(self): @@ -727,9 +736,10 @@ def update_share(self, share_id, **kwargs): perms = kwargs.get('perms', None) password = kwargs.get('password', None) public_upload = kwargs.get('public_upload', None) + expiration = kwargs.get('expiration', None) if (isinstance(perms, int)) and (perms > self.OCS_PERMISSION_ALL): perms = None - if not (perms or password or (public_upload is not None)): + if not (perms or password or (public_upload is not None) or (expiration is not None)): return False if not isinstance(share_id, int): return False @@ -741,6 +751,8 @@ def update_share(self, share_id, **kwargs): data['password'] = password if (public_upload is not None) and (isinstance(public_upload, bool)): data['publicUpload'] = str(public_upload).lower() + if expiration is not None: + data['expireDate'] = self.__parse_expiration_date(expiration) res = self.__make_ocs_request( 'PUT', @@ -791,7 +803,8 @@ def share_file_with_link(self, path, **kwargs): """ perms = kwargs.get('perms', None) public_upload = kwargs.get('public_upload', 'false') - password = kwargs.get('password', None) + password = kwargs.get('password', None) + expiration = kwargs.get('expiration', None) path = self.__normalize_path(path) @@ -805,6 +818,8 @@ def share_file_with_link(self, path, **kwargs): post_data['password'] = password if perms: post_data['permissions'] = perms + if expiration is not None: + post_data['expireDate'] = self.__parse_expiration_date(expiration) res = self.__make_ocs_request( 'POST', @@ -816,12 +831,25 @@ def share_file_with_link(self, path, **kwargs): tree = ET.fromstring(res.content) self.__check_ocs_status(tree) data_el = tree.find('data') + + expiration = None + exp_el = data_el.find('expiration') + if exp_el is not None and exp_el.text is not None and len(exp_el.text) > 0: + expiration = exp_el.text + + permissions = None + perms_el = data_el.find('permissions') + if perms_el is not None and perms_el.text is not None and len(perms_el.text) > 0: + permissions = int(perms_el.text) + return ShareInfo( { 'id': data_el.find('id').text, 'path':path, 'link': data_el.find('url').text, - 'token': data_el.find('token').text + 'token': data_el.find('token').text, + 'expiration': expiration, + 'permissions':permissions } ) raise HTTPResponseError(res) @@ -965,8 +993,8 @@ def user_exists(self, user_name): """Checks a user via provisioning API. If you get back an error 999, then the provisioning API is not enabled. - :param user_name: name of user to be checked - :returns: True if user found + :param user_name: name of user to be checked + :returns: True if user found """ users=self.search_users(user_name) @@ -992,7 +1020,7 @@ def search_users(self, user_name): tree = ET.fromstring(res.text) users = [x.text for x in tree.findall('data/users/element')] - return users + return users raise HTTPResponseError(res) @@ -1701,7 +1729,7 @@ def __strip_dav_path(self, path): if path.startswith(self.__davpath): return path[len(self.__davpath):] return path - + def __webdav_move_copy(self,remote_path_source,remote_path_target,operation): """Copies or moves a remote file or directory @@ -1749,6 +1777,21 @@ def __xml_to_dict(self, element): else: return_dict[el.tag] = el.text return return_dict + + @staticmethod + def __parse_expiration_date(date): + """Converts the given datetime object into the format required + by the share API + :param date: datetime object + :returns: string encoded to use as expireDate parameter in the share API + """ + if date is None: + return None + + if isinstance(date, datetime.datetime): + return date.strftime('YYYY-MM-DD') + + return date def __get_shareinfo(self, data_el): """Simple helper which returns instance of ShareInfo class diff --git a/owncloud/test/test.py b/owncloud/test/test.py index 6ae89f9..07ddc3a 100644 --- a/owncloud/test/test.py +++ b/owncloud/test/test.py @@ -567,9 +567,15 @@ def test_copy_to_non_existing_dir(self): ) @data_provider(files) - def test_share_with_link(self, file_name): + def test_share_with_link_lesser_8_2(self, file_name): """Test sharing a file with link""" + current_version = self.client.get_version().split('.') + current_version_major = int(current_version[0]) + current_version_minor = int(current_version[1]) + if ((current_version_major > 8) or ((current_version_major == 8) and (current_version_minor >= 2))): + raise unittest.SkipTest("This test should not run against >=8.2 servers"); + path = self.test_root + file_name self.assertTrue(self.client.put_file_contents(path, 'hello world!')) @@ -582,6 +588,30 @@ def test_share_with_link(self, file_name): self.assertTrue(type(share_info.get_link()) is str) self.assertTrue(type(share_info.get_token()) is str) + @data_provider(files) + def test_share_with_link_greater_8_2(self, file_name): + """Test sharing a file with link""" + + current_version = self.client.get_version().split('.') + current_version_major = int(current_version[0]) + current_version_minor = int(current_version[1]) + if ((current_version_major < 8) or ((current_version_major == 8) and (current_version_minor < 2))): + raise unittest.SkipTest("This test should not run against <8.2 servers"); + + path = self.test_root + file_name + self.assertTrue(self.client.put_file_contents(path, 'hello world!')) + + share_info = self.client.share_file_with_link(path, public_upload=True, password='1234', expiration='2150-02-04') + + self.assertTrue(self.client.is_shared(path)) + self.assertTrue(isinstance(share_info, owncloud.ShareInfo)) + self.assertTrue(type(share_info.get_id()) is int) + self.assertEquals(share_info.get_path(), path) + self.assertTrue(type(share_info.get_link()) is str) + self.assertTrue(type(share_info.get_token()) is str) + self.assertEquals(share_info.get_permissions(), 7) + self.assertEquals(share_info.get_expiration(), '2150-02-04 00:00:00') + def test_share_with_link_non_existing_file(self): """Test sharing a file with link""" with self.assertRaises(owncloud.ResponseError) as e: @@ -781,6 +811,17 @@ def test_update_share_user(self): self.assertEqual(int(perms), self.client.OCS_PERMISSION_ALL) self.assertTrue(self.client.delete_share(share_id)) + def test_update_share_expiration(self): + """Test updating a share parameters - expiration date""" + path = self.test_root + 'update_share_expiration' + self.client.mkdir(path) + share_info = self.client.share_file_with_link(path) + share_id = share_info.get_id() + self.assertTrue(self.client.update_share(share_id, expiration='2150-02-04')) + share_info = self.client.get_shares(path)[0] + self.assertEquals(share_info.get_expiration(), '2150-02-04 00:00:00') + self.assertTrue(self.client.delete_share(share_id)) + def test_update_share_public(self): """Test updating a share parameters - public share""" path = self.test_root + 'update_share_public.txt'