Initial QSfera import

Desktop/test/gui/helpers/api/http_helper.py

@@ -0,0 +1,60 @@
+import requests
+import urllib3
+
+from helpers.UserHelper import basic_auth_header
+from helpers.ConfigHelper import get_config
+
+urllib3.disable_warnings()
+
+
+def send_request(url, method, body=None, headers=None, user=None, password=None):
+    auth_header = basic_auth_header(user, password)
+    if not headers:
+        headers = {}
+    headers.update(auth_header)
+    return requests.request(
+        method,
+        url,
+        data=body,
+        headers=headers,
+        verify=False,
+        # in seconds
+        # e.g.: 60
+        timeout=get_config("maxSyncTimeout"),
+    )
+
+
+def get(url, headers=None, user=None, password=None):
+    return send_request(
+        url=url, method="GET", headers=headers, user=user, password=password
+    )
+
+
+def post(url, body=None, headers=None, user=None):
+    return send_request(url, "POST", body, headers, user)
+
+
+def put(url, body=None, headers=None, user=None):
+    return send_request(url, "PUT", body, headers, user)
+
+
+def delete(url, headers=None, user=None):
+    return send_request(url=url, method="DELETE", headers=headers, user=user)
+
+
+def mkcol(url, headers=None, user=None):
+    return send_request(url=url, method="MKCOL", headers=headers, user=user)
+
+
+def propfind(url, body=None, headers=None, user=None, password=None):
+    return send_request(url, "PROPFIND", body, headers, user, password)
+
+
+def assert_http_status(response, expected_code, message=""):
+    response_body = ""
+    if response.text:
+        response_body = response.text
+
+    assert (
+        response.status_code == expected_code
+    ), f"{message}\nRequest failed with status code '{response.status_code}'\n{response_body}"

Desktop/test/gui/helpers/api/provisioning.py

@@ -0,0 +1,70 @@
+import json
+from PySide6.QtCore import QJsonDocument
+
+import helpers.api.http_helper as request
+from helpers.ConfigHelper import get_config
+from helpers import UserHelper
+from helpers.api.utils import url_join
+
+
+created_groups = {}
+created_users = {}
+
+
+def get_graph_url():
+    return url_join(get_config("localBackendUrl"), "graph", "v1.0")
+
+
+def create_user(username):
+    if username in UserHelper.test_users:
+        user = UserHelper.test_users[username]
+    else:
+        user = UserHelper.User(
+            username=username,
+            displayname=username,
+            email=f'{username}@mail.com',
+            password=UserHelper.get_default_password(),
+        )
+
+    url = url_join(get_graph_url(), "users")
+    body = json.dumps(
+        {
+            "onPremisesSamAccountName": user.username,
+            "passwordProfile": {"password": user.password},
+            "displayName": user.displayname,
+            "mail": user.email,
+        }
+    )
+    response = request.post(url, body)
+    request.assert_http_status(response, 201, f"Failed to create user '{username}'")
+    resp_object = response.json()
+
+    # Check if the user already exists
+    user_info = {
+        "id": resp_object["id"],
+        "username": user.username,
+        "password": user.password,
+        "displayname": resp_object["displayName"],
+        "email": resp_object["mail"],
+    }
+    created_users[username] = user_info
+
+
+def delete_created_users():
+    for username, user_info in list(created_users.items()):
+        user_id = user_info['id']
+        url = url_join(get_graph_url(), "users", user_id)
+        response = request.delete(url)
+        request.assert_http_status(response, 204, "Failed to delete user")
+        del created_users[username]
+
+
+def get_capabilities():
+    server_url = get_config('localBackendUrl')
+    url = url_join(server_url, '/ocs/v1.php/cloud/capabilities?format=json')
+    response = request.get(url)
+    response_str = response.text
+    response_doc = QJsonDocument.fromJson(response_str.encode("utf-8"))
+    response_obj = response_doc.object()
+    capabilities = response_obj.get('ocs').get('data').get('capabilities')
+    return capabilities

Desktop/test/gui/helpers/api/utils.py

@@ -0,0 +1,8 @@
+from posixpath import join
+
+
+def url_join(*args):
+    paths = []
+    for path in list(args):
+        paths.append(path.strip("/"))
+    return join("", *paths)

Desktop/test/gui/helpers/api/webdav_helper.py

@@ -0,0 +1,147 @@
+from urllib.parse import quote
+import xml.etree.ElementTree as ET
+import json
+
+import helpers.api.http_helper as request
+from helpers.api.utils import url_join
+from helpers.ConfigHelper import get_config, PERMISSION_ROLES
+from helpers.FilesHelper import get_file_for_upload
+from helpers.SpaceHelper import get_personal_space_id
+from helpers.api import provisioning
+
+
+def get_webdav_url():
+    return url_join(get_config('localBackendUrl'), 'remote.php/dav/files')
+
+
+def get_beta_graph_url():
+    return url_join(get_config("localBackendUrl"), "graph", "v1beta1")
+
+
+def get_resource_path(user, resource):
+    resource = resource.strip('/').replace('\\', '/')
+    encoded_resource_path = [quote(path, safe='') for path in resource.split('/')]
+    encoded_resource_path = '/'.join(encoded_resource_path)
+    url = url_join(get_webdav_url(), user, encoded_resource_path)
+    return url
+
+
+def resource_exists(user, resource):
+    response = request.propfind(get_resource_path(user, resource), user=user)
+    if response.status_code == 207:
+        return True
+    if response.status_code == 404:
+        return False
+    raise AssertionError(f'Server returned status code: {response.status_code}')
+
+
+def get_file_content(user, resource):
+    response = request.get(get_resource_path(user, resource), user=user)
+    if resource.lower().endswith('.txt'):
+        return response.text
+    return response.content
+
+
+def get_folder_items(user, resource):
+    """Get the root XML element from a PROPFIND request"""
+    path = get_resource_path(user, resource)
+    xml_response = request.propfind(path, user=user)
+
+    if xml_response.status_code != 207:
+        raise AssertionError(f'Failed to get resource properties: {xml_response.status_code}')
+
+    return ET.fromstring(xml_response.content)
+
+
+def get_folder_items_count(user, folder_name):
+    folder_name = folder_name.strip('/')
+    root_element = get_folder_items(user, folder_name)
+    total_items = 0
+    for response_element in root_element:
+        for href_element in response_element:
+            # The first item is folder itself so excluding it
+            if href_element.tag == '{DAV:}href' and not href_element.text.endswith(
+                f'{user}/{folder_name}/'
+            ):
+                total_items += 1
+    return str(total_items)
+
+
+def create_folder(user, folder_name):
+    url = get_resource_path(user, folder_name)
+    response = request.mkcol(url, user=user)
+    assert (
+        response.status_code == 201
+    ), f'Could not create the folder: {folder_name} for user {user}'
+
+
+def create_file(user, file_name, contents):
+    url = get_resource_path(user, file_name)
+    response = request.put(url, body=contents, user=user)
+    assert response.status_code in [
+        201,
+        204,
+    ], f"Could not create file '{file_name}' for user {user}"
+
+
+def upload_file(user, file_name, destination):
+    file_path = get_file_for_upload(file_name)
+    with open(file_path, 'rb') as file:
+        contents = file.read()
+    create_file(user, destination, contents)
+
+
+def delete_resource(user, resource):
+    url = get_resource_path(user, resource)
+    response = request.delete(url, user=user)
+    assert response.status_code == 204, f"Could not delete folder '{resource}'"
+
+
+def get_resource_id(user, resource):
+    root_element = get_folder_items(user, resource)
+
+    # Finding the fileid elements using XPath with namespace notation
+    fileid_elements = root_element.findall(".//{http://owncloud.org/ns}fileid")
+
+    # The First element is the desired resource's file id
+    if fileid_elements:
+        return fileid_elements[0].text
+
+    raise AssertionError(f'Could not find resource ID for {resource}')
+
+
+def get_permission_role_id(role_name):
+    """Get the permission role ID for a given role name"""
+    if role_name not in PERMISSION_ROLES:
+        raise ValueError(f"Unknown permission role: {role_name}")
+    return PERMISSION_ROLES[role_name]
+
+
+def get_user_id(username):
+    """Get the user ID for a given username from created users"""
+    if username in provisioning.created_users:
+        return provisioning.created_users[username]['id']
+
+    raise AssertionError(f'User {username} not found in created users. Make sure the user is created first.')
+
+
+def send_resource_share_invitation(user, resource, sharee, permission_role):
+    space_id = get_personal_space_id(user)
+    resource_id = get_resource_id(user, resource)
+    recipient_user_id = get_user_id(sharee)
+    role_id = get_permission_role_id(permission_role)
+
+    url = url_join(get_beta_graph_url(), "drives", space_id, "items", resource_id, "invite")
+
+    body = {
+        "roles": [role_id],
+        "recipients": [
+            {
+                "objectId": recipient_user_id,
+                "@libre.graph.recipient.type": "user"
+            }
+        ]
+    }
+
+    response = request.post(url, body=json.dumps(body), user=user)
+    assert response.status_code in [200, 201], f"Failed to send share invitation: {response.status_code}"