From a5c2315b05c240c4002fecff0609e82c33810a65 Mon Sep 17 00:00:00 2001 From: Danil Date: Sat, 22 Dec 2018 04:04:48 +0200 Subject: [PATCH 1/4] share api fixes, docstrings, tests (except federated cloudhsares) --- NextCloud.py | 167 +++++++++++++++++++++++++++++-------------- tests/base.py | 34 ++++++++- tests/test_shares.py | 167 +++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 312 insertions(+), 56 deletions(-) create mode 100644 tests/test_shares.py diff --git a/NextCloud.py b/NextCloud.py index 1b02ed4..2054ea4 100644 --- a/NextCloud.py +++ b/NextCloud.py @@ -1,8 +1,6 @@ import enum - import requests - PUBLIC_API_NAME_CLASS_MAP = dict() @@ -63,7 +61,7 @@ def get_full_url(self, additional_url=""): if self.to_json: self.query_components.append("format=json") - ret = "{base_url}/{api_url}{additional_url}".format( + ret = "{base_url}{api_url}{additional_url}".format( base_url=self.base_url, api_url=self.API_URL, additional_url=additional_url) if self.to_json: @@ -159,103 +157,163 @@ class Share(WithRequester): def get_local_url(self, additional_url=""): if additional_url: - return "/".join(self.LOCAL, additional_url) + return "/".join([self.LOCAL, additional_url]) return self.LOCAL def get_federated_url(self, additional_url=""): if additional_url: - return "/".join(self.FEDERATED, additional_url) - return self.LOCAL + return "/".join([self.FEDERATED, additional_url]) + return self.FEDERATED @nextcloud_method def get_shares(self): - self.requester.get(self.requester.get_local_url()) + """ Get all shares from the user """ + return self.requester.get(self.get_local_url()) @nextcloud_method def get_shares_from_path(self, path, reshares=None, subfiles=None): - url = self.requester.get_local_url(path) + """ + Get all shares from a given file/folder - if reshares is not None: - self.query_components.append("reshares=true") + Args: + path (str): path to file/folder + reshares (bool): (optional) return not only the shares from the current user but all shares from the given file + subfiles (bool): (optional) return all shares within a folder, given that path defines a folder - if subfiles is not None: - self.query_components.append("subfiles=true") + Returns: - return self.requester.get(url) + """ + url = self.get_local_url() + params = { + "path": path, + "reshares": None if reshares is None else str(bool(reshares)).lower(), # TODO: test reshares, subfiles + "subfiles": None if subfiles is None else str(bool(subfiles)).lower(), + } + return self.requester.get(url, params=params) @nextcloud_method def get_share_info(self, sid): - self.requester.get(self.requester.get_local_url(sid)) + """ + Get information about a given share + + Args: + sid (int): share id + + Returns: + """ + return self.requester.get(self.get_local_url(sid)) @nextcloud_method def create_share( - self, path, shareType, shareWith=None, publicUpload=None, + self, path, share_type, share_with=None, public_upload=None, password=None, permissions=None): - url = self.requester.get_local_url() - if publicUpload: - publicUpload = "true" - if (path is None or not isinstance(shareType, int)) or (shareType in [0, 1] and shareWith is None): + """ + Share a file/folder with a user/group or as public link + + Mandatory fields: share_type, path and share_with for share_type USER (0) or GROUP (1). + + Args: + path (str): path to the file/folder which should be shared + share_type (int): ShareType attribute + share_with (str): user/group id with which the file should be shared + public_upload (bool): bool, allow public upload to a public shared folder (true/false) + password (str): password to protect public link Share with + permissions (int): sum of selected Permission attributes + + Returns: + + """ + url = self.get_local_url() + if public_upload: + public_upload = "true" + if (path is None or not isinstance(share_type, int)) or (share_type in [0, 1] and share_with is None): return False - msg = {"path": path, "shareType": shareType} - if shareType in [0, 1]: - msg["shareWith"] = shareWith - if publicUpload: - msg["publicUpload"] = publicUpload - if shareType == 3 and password is not None: - msg["password"] = str(password) + + data = {"path": path, "shareType": share_type} + if share_type in [ShareType.GROUP, ShareType.USER, ShareType.FEDERATED_CLOUD_SHARE]: + data["shareWith"] = share_with + if public_upload: + data["publicUpload"] = public_upload + if share_type == 3 and password is not None: + data["password"] = str(password) if permissions is not None: - msg["permissions"] = permissions - return self.requester.post(url, msg) + data["permissions"] = permissions + return self.requester.post(url, data) @nextcloud_method def delete_share(self, sid): - return self.requester.delete(self.requester.get_local_url(sid)) - - @nextcloud_method - def update_share(self, sid, permissions=None, password=None, publicUpload=None, expireDate=""): - msg = {} - if permissions: - msg["permissions"] = permissions - if password is not None: - msg["password"] = str(password) - if publicUpload: - msg["publicUpload"] = "true" - if publicUpload is False: - msg["publicUpload"] = "false" - if expireDate: - msg["expireDate"] = expireDate - url = self.requester.get_local_url(sid) - return self.requester.put(url, msg) + """ + Remove the given share + + Args: + sid (str): share id + + Returns: + + """ + return self.requester.delete(self.get_local_url(sid)) + + @nextcloud_method + def update_share(self, sid, permissions=None, password=None, public_upload=None, expire_date=""): + """ + Update a given share, only one value can be updated per request + + Args: + sid (str): share id + permissions (int): sum of selected Permission attributes + password (str): password to protect public link Share with + public_upload (bool): bool, allow public upload to a public shared folder (true/false) + expire_date (str): set an expire date for public link shares. Format: ‘YYYY-MM-DD’ + + Returns: + + """ + params = dict( + permissions=permissions, + password=password, + expireDate=expire_date + ) + if public_upload: + params["publicUpload"] = "true" + if public_upload is False: + params["publicUpload"] = "false" + + # check if only one param specified + specified_params_count = sum([int(bool(each)) for each in params.values()]) + if specified_params_count > 1: + raise ValueError("Only one parameter for update can be specified per request") + + url = self.get_local_url(sid) + return self.requester.put(url, data=params) @nextcloud_method def list_accepted_federated_cloudshares(self): - # FIXME: doesn't work - url = self.requester.get_federated_url() + url = self.get_federated_url() return self.requester.get(url) @nextcloud_method def get_known_federated_cloudshare(self, sid): - url = self.requester.get_federated_url(sid) + url = self.get_federated_url(sid) return self.requester.get(url) @nextcloud_method def delete_accepted_federated_cloudshare(self, sid): - url = self.requester.get_federated_url(sid) + url = self.get_federated_url(sid) return self.requester.delete(url) @nextcloud_method def list_pending_federated_cloudshares(self, sid): - url = self.requester.get_federated_url("pending") + url = self.get_federated_url("pending") return self.requester.get(url) @nextcloud_method def accept_pending_federated_cloudshare(self, sid): - url = self.requester.get_federated_url("pending/{sid}".format(sid=sid)) + url = self.get_federated_url("pending/{sid}".format(sid=sid)) return self.requester.post(url) @nextcloud_method def decline_pending_federated_cloudshare(self, sid): - url = self.requester.get_federated_url("pending/{sid}".format(sid=sid)) + url = self.get_federated_url("pending/{sid}".format(sid=sid)) return self.requester.delete(url) @@ -554,11 +612,12 @@ class OCSCode(enum.IntEnum): class ShareType(enum.IntEnum): USER = 0 GROUP = 1 - PUBLIClINK = 3 + PUBLIC_LINK = 3 FEDERATED_CLOUD_SHARE = 6 class Permission(enum.IntEnum): + """ Permission for Share have to be sum of selected permissions """ READ = 1 UPDATE = 2 CREATE = 4 @@ -570,5 +629,5 @@ class Permission(enum.IntEnum): QUOTE_UNLIMITED = -3 -def datttetime_to_expireDate(date): +def datetime_to_expire_date(date): return date.strftime("%Y-%m-%d") diff --git a/tests/base.py b/tests/base.py index 3bc4d13..75fa9c4 100644 --- a/tests/base.py +++ b/tests/base.py @@ -18,14 +18,17 @@ class BaseTestCase(TestCase): UNKNOWN_ERROR_CODE = 103 NOT_FOUND_CODE = 404 + SHARE_API_SUCCESS_CODE = 200 # share api has different code + def setUp(self): self.username = NEXTCLOUD_USERNAME self.nxc = NextCloud(NEXTCLOUD_URL, NEXTCLOUD_USERNAME, NEXTCLOUD_PASSWORD, js=True) - def create_new_user(self, username_prefix): + def create_new_user(self, username_prefix, password=None): """ Helper method to create new user """ new_user_username = username_prefix + self.get_random_string(length=4) - res = self.nxc.add_user(new_user_username, self.get_random_string(length=8)) + user_password = password or self.get_random_string(length=8) + res = self.nxc.add_user(new_user_username, user_password) assert res['ocs']['meta']['statuscode'] == self.SUCCESS_CODE return new_user_username @@ -34,6 +37,33 @@ def delete_user(self, username): res = self.nxc.delete_user(username) assert res['ocs']['meta']['statuscode'] == self.SUCCESS_CODE + def clear(self, nxc=None, user_ids=None, group_ids=None, share_ids=None): + """ + Delete created objects during tests + + Args: + nxc (NextCloud object): (optional) Nextcloud instance, if not given - self.nxc is used + user_ids (list): list of user_ids + group_ids (list): list of group_ids + share_ids (list): list of group_ids + + Returns: + + """ + nxc = nxc or self.nxc + if share_ids: + for share_id in share_ids: + res = nxc.delete_share(share_id) + assert res['ocs']['meta']['statuscode'] == self.SHARE_API_SUCCESS_CODE + if group_ids: + for group_id in group_ids: + res = nxc.delete_group(group_id) + assert res['ocs']['meta']['statuscode'] == self.SUCCESS_CODE + if user_ids: + for user_id in user_ids: + res = nxc.delete_user(user_id) + assert res['ocs']['meta']['statuscode'] == self.SUCCESS_CODE + def get_random_string(self, length=6): """ Helper method to get random string with set length """ return ''.join(random.choice(string.ascii_letters + string.digits) for _ in range(length)) diff --git a/tests/test_shares.py b/tests/test_shares.py new file mode 100644 index 0000000..b04e481 --- /dev/null +++ b/tests/test_shares.py @@ -0,0 +1,167 @@ +import requests +from datetime import datetime, timedelta + +from NextCloud import ShareType, Permission, datetime_to_expire_date + +from .base import BaseTestCase, NextCloud, NEXTCLOUD_URL + + +class TestShares(BaseTestCase): + + def setUp(self): + super(TestShares, self).setUp() + user_password = self.get_random_string(length=8) + self.user_username = self.create_new_user('shares_user_', password=user_password) + self.nxc_local = self.nxc_local = NextCloud(NEXTCLOUD_URL, self.user_username, user_password, js=True) + # make user admin + self.nxc.add_to_group(self.user_username, 'admin') + + def tearDown(self): + self.nxc.delete_user(self.user_username) + + def test_share_create_retrieve_delete(self): + """ shallow test for base retrieving single, list, creating and deleting share """ + # check no shares exists + res = self.nxc_local.get_shares() + assert res['ocs']['meta']['statuscode'] == self.SHARE_API_SUCCESS_CODE + assert len(res['ocs']['data']) == 0 + + # create public share + res = self.nxc_local.create_share('Documents', share_type=ShareType.PUBLIC_LINK.value) + assert res['ocs']['meta']['statuscode'] == self.SHARE_API_SUCCESS_CODE + share_id = res['ocs']['data']['id'] + + # get all shares + all_shares = self.nxc_local.get_shares()['ocs']['data'] + assert len(all_shares) == 1 + assert all_shares[0]['id'] == share_id and all_shares[0]['share_type'] == ShareType.PUBLIC_LINK.value + + # get single share info + created_share = self.nxc_local.get_share_info(share_id) + assert res['ocs']['meta']['statuscode'] == self.SHARE_API_SUCCESS_CODE + created_share_data = created_share['ocs']['data'][0] + assert (created_share_data['id'] == share_id + and created_share_data['share_type'] == ShareType.PUBLIC_LINK.value + and created_share_data['uid_owner'] == self.user_username) + + # delete share + res = self.nxc_local.delete_share(share_id) + assert res['ocs']['meta']['statuscode'] == self.SHARE_API_SUCCESS_CODE + all_shares = self.nxc_local.get_shares()['ocs']['data'] + assert len(all_shares) == 0 + + def test_create(self): + """ creating share with different params """ + share_path = "Documents" + user_to_share_with = self.create_new_user("test_shares_") + group_to_share_with = 'group_to_share_with' + self.nxc.add_group(group_to_share_with) + + # create for user, group + for (share_type, share_with, permissions) in [(ShareType.USER.value, user_to_share_with, Permission.READ.value), + (ShareType.GROUP.value, group_to_share_with, Permission.READ.value + Permission.CREATE.value)]: + # create share with user + res = self.nxc_local.create_share(share_path, + share_type=share_type, + share_with=share_with, + permissions=permissions) + assert res['ocs']['meta']['statuscode'] == self.SHARE_API_SUCCESS_CODE + share_id = res['ocs']['data']['id'] + + # check if shared with right user/group, permission + created_share = self.nxc_local.get_share_info(share_id) + assert res['ocs']['meta']['statuscode'] == self.SHARE_API_SUCCESS_CODE + created_share_data = created_share['ocs']['data'][0] + assert (created_share_data['id'] == share_id + and created_share_data['share_type'] == share_type + and created_share_data['share_with'] == share_with + and created_share_data['permissions'] == permissions) + + # delete share, user + self.nxc_local.delete_share(share_id) + self.nxc.delete_user(user_to_share_with) + + def test_create_with_password(self): + share_path = "Documents" + res = self.nxc_local.create_share(path=share_path, + share_type=ShareType.PUBLIC_LINK.value, + password=self.get_random_string(length=8)) + assert res['ocs']['meta']['statuscode'] == self.SHARE_API_SUCCESS_CODE + share_url = res['ocs']['data']['url'] + share_resp = requests.get(share_url) + assert "This share is password-protected" in share_resp.text + self.nxc_local.delete_share(res['ocs']['data']['id']) + + def test_get_path_shares(self): + share_path = "Documents" + group_to_share_with_name = self.get_random_string(length=4) + "_test_add" + self.nxc.add_group(group_to_share_with_name) + + # check that path has no shares yet + res = self.nxc_local.get_shares_from_path(share_path) + assert res['ocs']['meta']['statuscode'] == self.SHARE_API_SUCCESS_CODE + assert len(res['ocs']['data']) == 0 + + # first path share + first_share = self.nxc_local.create_share(path=share_path, + share_type=ShareType.PUBLIC_LINK.value) + + # create second path share + second_share = self.nxc_local.create_share(path=share_path, + share_type=ShareType.GROUP.value, + share_with=group_to_share_with_name, + permissions=Permission.READ.value) + + all_shares_ids = [first_share['ocs']['data']['id'], second_share['ocs']['data']['id']] + + # check that path has two shares + res = self.nxc_local.get_shares_from_path(share_path) + assert res['ocs']['meta']['statuscode'] == self.SHARE_API_SUCCESS_CODE + assert len(res['ocs']['data']) == 2 + assert all([each['id'] in all_shares_ids for each in res['ocs']['data']]) + + # clean shares, groups + self.clear(self.nxc_local, share_ids=all_shares_ids, group_ids=[group_to_share_with_name]) + + def test_update_share(self): + share_path = "Documents" + user_to_share_with = self.create_new_user("test_shares_") + + share_with = user_to_share_with + share_type = ShareType.USER.value + # create share with user + res = self.nxc_local.create_share(share_path, + share_type=ShareType.USER.value, + share_with=user_to_share_with, + permissions=Permission.READ.value) + assert res['ocs']['meta']['statuscode'] == self.SHARE_API_SUCCESS_CODE + share_id = res['ocs']['data']['id'] + + # update share permissions + new_permissions = Permission.READ.value + Permission.CREATE.value + res = self.nxc_local.update_share(share_id, permissions=new_permissions) + assert res['ocs']['meta']['statuscode'] == self.SHARE_API_SUCCESS_CODE + + updated_share_data = res['ocs']['data'] + assert (updated_share_data['id'] == share_id + and updated_share_data['share_type'] == share_type + and updated_share_data['share_with'] == share_with + and updated_share_data['permissions'] == new_permissions + and updated_share_data['expiration'] is None) + + # update share expire date + expire_date = datetime_to_expire_date(datetime.now() + timedelta(days=5)) + res = self.nxc_local.update_share(share_id, expire_date=expire_date) + assert res['ocs']['meta']['statuscode'] == self.SHARE_API_SUCCESS_CODE + + updated_share_data = res['ocs']['data'] + assert (updated_share_data['id'] == share_id + and updated_share_data['share_type'] == share_type + and updated_share_data['share_with'] == share_with + and updated_share_data['permissions'] == new_permissions + and updated_share_data['expiration'] == "{} 00:00:00".format(expire_date)) + + self.clear(self.nxc_local, share_ids=[share_id], user_ids=[user_to_share_with]) + + + From c3720713ee854967b90482074852ba013a3a72c2 Mon Sep 17 00:00:00 2001 From: Danil Date: Sat, 22 Dec 2018 23:35:37 +0200 Subject: [PATCH 2/4] Move federated cloud share api methods to separate class --- NextCloud.py | 17 +++++++++++------ 1 file changed, 11 insertions(+), 6 deletions(-) diff --git a/NextCloud.py b/NextCloud.py index 2054ea4..817a1c4 100644 --- a/NextCloud.py +++ b/NextCloud.py @@ -88,6 +88,7 @@ def __init__(self, endpoint, user, passwd, js=False): "GroupFolders": GroupFolders(requester), "Share": Share(requester), "User": User(requester), + "FederatedCloudShare": FederatedCloudShare(requester) } for name, location in PUBLIC_API_NAME_CLASS_MAP.items(): setattr(self, name, getattr(self.functionality[location], name)) @@ -153,18 +154,12 @@ def rename_group_folder(self, fid, mountpoint): class Share(WithRequester): API_URL = "/ocs/v2.php/apps/files_sharing/api/v1" LOCAL = "shares" - FEDERATED = "remote_shares" def get_local_url(self, additional_url=""): if additional_url: return "/".join([self.LOCAL, additional_url]) return self.LOCAL - def get_federated_url(self, additional_url=""): - if additional_url: - return "/".join([self.FEDERATED, additional_url]) - return self.FEDERATED - @nextcloud_method def get_shares(self): """ Get all shares from the user """ @@ -286,6 +281,16 @@ def update_share(self, sid, permissions=None, password=None, public_upload=None, url = self.get_local_url(sid) return self.requester.put(url, data=params) + +class FederatedCloudShare(WithRequester): + API_URL = "/ocs/v2.php/apps/files_sharing/api/v1" + FEDERATED = "remote_shares" + + def get_federated_url(self, additional_url=""): + if additional_url: + return "/".join([self.FEDERATED, additional_url]) + return self.FEDERATED + @nextcloud_method def list_accepted_federated_cloudshares(self): url = self.get_federated_url() From 01d84be36c74c5bc625769df28cbe0672ea84bb5 Mon Sep 17 00:00:00 2001 From: Danil Date: Sat, 22 Dec 2018 23:52:45 +0200 Subject: [PATCH 3/4] change hardcoded share_type ids to constants --- NextCloud.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/NextCloud.py b/NextCloud.py index 817a1c4..7e77a89 100644 --- a/NextCloud.py +++ b/NextCloud.py @@ -221,7 +221,9 @@ def create_share( url = self.get_local_url() if public_upload: public_upload = "true" - if (path is None or not isinstance(share_type, int)) or (share_type in [0, 1] and share_with is None): + if (path is None or not isinstance(share_type, int)) \ + or (share_type in [ShareType.GROUP, ShareType.USER, ShareType.FEDERATED_CLOUD_SHARE] + and share_with is None): return False data = {"path": path, "shareType": share_type} @@ -229,7 +231,7 @@ def create_share( data["shareWith"] = share_with if public_upload: data["publicUpload"] = public_upload - if share_type == 3 and password is not None: + if share_type == ShareType.PUBLIC_LINK and password is not None: data["password"] = str(password) if permissions is not None: data["permissions"] = permissions From 7f3d9be04b0c283c3fe69e7d4c9535c5e4b94fa4 Mon Sep 17 00:00:00 2001 From: Danil Date: Sun, 23 Dec 2018 00:14:20 +0200 Subject: [PATCH 4/4] Federated cloudshares tests: new docker-compose nextcloud service to make cloudshare with, first test (not working yet) --- tests/docker-compose.yml | 28 ++++++++++++++++++++++++++++ tests/test_federated_cloudshares.py | 25 +++++++++++++++++++++++++ 2 files changed, 53 insertions(+) create mode 100644 tests/test_federated_cloudshares.py diff --git a/tests/docker-compose.yml b/tests/docker-compose.yml index 649a01f..c32801d 100644 --- a/tests/docker-compose.yml +++ b/tests/docker-compose.yml @@ -35,6 +35,29 @@ services: depends_on: - db + app2: + image: nextcloud:apache + restart: always + networks: + backend: + aliases: + - app2 + ports: + - 8081:80 + volumes: + - nextcloud:/var/www/html + environment: + POSTGRES_DB: 'nextcloud2' + POSTGRES_USER: 'postgres' + POSTGRES_PASSWORD: 'secret' + POSTGRES_HOST: 'db' + + NEXTCLOUD_TRUSTED_DOMAINS: "app2" + NEXTCLOUD_ADMIN_USER: "admin" + NEXTCLOUD_ADMIN_PASSWORD: "admin" + depends_on: + - db + python-api: build: context: ../ @@ -43,6 +66,11 @@ services: NEXTCLOUD_HOST: "app" NEXTCLOUD_USERNAME: "admin" NEXTCLOUD_PASSWORD: "admin" + + NEXTCLOUD2_HOST: "app2" + NEXTCLOUD2_USERNAME: "admin" + NEXTCLOUD2_PASSWORD: "admin" + networks: backend: aliases: diff --git a/tests/test_federated_cloudshares.py b/tests/test_federated_cloudshares.py new file mode 100644 index 0000000..39743e4 --- /dev/null +++ b/tests/test_federated_cloudshares.py @@ -0,0 +1,25 @@ +from NextCloud import ShareType + +from .base import BaseTestCase, NextCloud, NEXTCLOUD_URL + + +class TestFederatedCloudShares(BaseTestCase): + + def setUp(self): + super(TestFederatedCloudShares, self).setUp() + user_password = self.get_random_string(length=8) + self.user_username = self.create_new_user('shares_user_', password=user_password) + self.nxc_local = self.nxc_local = NextCloud(NEXTCLOUD_URL, self.user_username, user_password, js=True) + # make user admin + self.nxc.add_to_group(self.user_username, 'admin') + + def tearDown(self): + self.nxc.delete_user(self.user_username) + + def test_create_federated_cloudhsare(self): + share_path = "Documents" + + res = self.nxc_local.create_share(share_path, + share_type=ShareType.FEDERATED_CLOUD_SHARE.value, + share_with="admin@app:80") + assert res['ocs']['meta']['statuscode'] == self.SHARE_API_SUCCESS_CODE