Курнат Андрей
154 lines
4.6 KiB
Python
154 lines
4.6 KiB
Python
import json
|
|
from urllib import parse
|
|
|
|
from helpers.ConfigHelper import get_config
|
|
from helpers.api.utils import url_join
|
|
import helpers.api.http_helper as request
|
|
|
|
created_spaces = {}
|
|
user_spaces = {}
|
|
space_role = ['manager', 'editor', 'viewer']
|
|
|
|
|
|
def get_space_endpint():
|
|
return url_join(get_config('localBackendUrl'), 'graph', 'v1.0', 'drives')
|
|
|
|
|
|
def get_dav_endpint():
|
|
return url_join(get_config('localBackendUrl'), 'dav', 'spaces')
|
|
|
|
|
|
def get_share_endpint():
|
|
return url_join(
|
|
get_config('localBackendUrl'),
|
|
'ocs/v2.php/apps',
|
|
'files_sharing/api/v1/shares',
|
|
)
|
|
|
|
|
|
def create_space(space_name):
|
|
body = json.dumps({'name': space_name})
|
|
response = request.post(get_space_endpint(), body)
|
|
request.assert_http_status(response, 201, f'Failed to create space {space_name}')
|
|
# save created space
|
|
resp_object = response.json()
|
|
created_spaces[space_name] = resp_object['id']
|
|
|
|
|
|
def fetch_spaces(user=None, query=''):
|
|
if query:
|
|
query = '?' + query
|
|
url = get_space_endpint() + query
|
|
response = request.get(url=url, user=user)
|
|
request.assert_http_status(response, 200, 'Failed to get spaces')
|
|
return response.json()['value']
|
|
|
|
|
|
def get_project_spaces(user=None):
|
|
search_query = '$filter=driveType eq \'project\''
|
|
return fetch_spaces(query=search_query, user=user)
|
|
|
|
def get_personal_space_id(user):
|
|
search_query = '$filter=driveType eq \'personal\''
|
|
space = fetch_spaces(query=search_query, user=user)
|
|
return space[0]['id']
|
|
|
|
def get_space_id(space_name, user=None):
|
|
spaces = {**created_spaces, **user_spaces}
|
|
if not space_name in spaces.keys():
|
|
return fetch_space_id(space_name, user)
|
|
return spaces.get(space_name)
|
|
|
|
|
|
def fetch_space_id(space_name, user=None):
|
|
spaces = fetch_spaces(user=user)
|
|
space_id = None
|
|
for space in spaces:
|
|
if space['name'] == space_name:
|
|
user_spaces[space_name] = space['id']
|
|
space_id = space['id']
|
|
break
|
|
return space_id
|
|
|
|
|
|
def delete_project_spaces():
|
|
global created_spaces, user_spaces
|
|
for _, space_id in created_spaces.items():
|
|
disable_project_space(space_id)
|
|
delete_project_space(space_id)
|
|
created_spaces = {}
|
|
user_spaces = {}
|
|
|
|
|
|
def disable_project_space(space_id):
|
|
url = url_join(get_space_endpint(), space_id)
|
|
response = request.delete(url)
|
|
request.assert_http_status(response, 204, f'Failed to disable space {space_id}')
|
|
|
|
|
|
def delete_project_space(space_id):
|
|
url = url_join(get_space_endpint(), space_id)
|
|
response = request.delete(url, {'Purge': 'T'})
|
|
request.assert_http_status(response, 204, f'Failed to delete space {space_id}')
|
|
|
|
|
|
def create_space_folder(space_name, folder_name):
|
|
space_id = get_space_id(space_name)
|
|
url = url_join(get_dav_endpint(), space_id, folder_name)
|
|
response = request.mkcol(url)
|
|
request.assert_http_status(
|
|
response,
|
|
201,
|
|
f'Failed to create folder "{folder_name}" in space "{space_name}"',
|
|
)
|
|
|
|
|
|
def create_space_file(space_name, file_name, content):
|
|
space_id = get_space_id(space_name)
|
|
url = url_join(get_dav_endpint(), space_id, file_name)
|
|
response = request.put(url, content)
|
|
if response.status_code not in (201, 204):
|
|
raise AssertionError(
|
|
f"Creating file '{file_name}' in space '{space_name}' failed with {response.status_code}\n"
|
|
+ response.text
|
|
)
|
|
|
|
|
|
def add_user_to_space(user, space_name, role):
|
|
role = role.lower()
|
|
if not role in space_role:
|
|
raise ValueError(f"Cannot set the role '{role}' to a space")
|
|
|
|
space_id = get_space_id(space_name)
|
|
url = get_share_endpint()
|
|
body = parse.urlencode(
|
|
{
|
|
'space_ref': space_id,
|
|
'shareType': 7,
|
|
'shareWith': user,
|
|
'role': role,
|
|
}
|
|
)
|
|
headers = {'Content-Type': 'application/x-www-form-urlencoded'}
|
|
response = request.post(url, body, headers)
|
|
request.assert_http_status(
|
|
response, 200, f'Failed to add user "{user}" to space "{space_name}"'
|
|
)
|
|
|
|
|
|
def get_file_content(space_name, file_name, user=None):
|
|
space_id = get_space_id(space_name, user)
|
|
url = url_join(get_dav_endpint(), space_id, file_name)
|
|
response = request.get(url=url, user=user)
|
|
request.assert_http_status(response, 200, f'Failed to get file "{file_name}"')
|
|
return response.text
|
|
|
|
|
|
def resource_exists(space_name, resource, user=None):
|
|
space_id = get_space_id(space_name, user)
|
|
url = url_join(get_dav_endpint(), space_id, resource)
|
|
response = request.get(url=url, user=user)
|
|
if response.status_code == 200:
|
|
return True
|
|
return False
|