Skip to content

Adapted PR 120 to new structure #143

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 5 commits into
base: master
Choose a base branch
from
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
69 changes: 56 additions & 13 deletions owncloud/owncloud.py
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Expand Down Expand Up @@ -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
Expand All @@ -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',
Expand Down Expand Up @@ -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)
Expand All @@ -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',
Expand All @@ -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)
Expand Down Expand Up @@ -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)
Expand All @@ -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)

Expand Down Expand Up @@ -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

Expand Down Expand Up @@ -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
Expand Down
43 changes: 42 additions & 1 deletion owncloud/test/test.py
Original file line number Diff line number Diff line change
Expand Up @@ -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");
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You mean "This test should not run against < 8.2 servers" ?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

No, that string is correct, which is not correct is the logic, it should check the major before, I'll change it.


path = self.test_root + file_name
self.assertTrue(self.client.put_file_contents(path, 'hello world!'))

Expand All @@ -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:
Expand Down Expand Up @@ -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'
Expand Down