diff --git a/.gitignore b/.gitignore index 6b48311..4af4f3c 100644 --- a/.gitignore +++ b/.gitignore @@ -12,4 +12,4 @@ htmlcov .pytest_cache deps venv -.vscode/settings.json +.vscode/ diff --git a/mergin/client.py b/mergin/client.py index 101c62e..d5c5f55 100644 --- a/mergin/client.py +++ b/mergin/client.py @@ -16,6 +16,8 @@ import re import typing import warnings +from enum import Enum +from typing import Optional, Type, Union from .common import ( ClientError, @@ -39,7 +41,13 @@ ) from .client_pull import pull_project_async, pull_project_wait, pull_project_finalize from .client_push import push_project_async, push_project_wait, push_project_finalize -from .utils import DateTimeEncoder, get_versions_with_file_changes, int_version, is_version_acceptable +from .utils import ( + DateTimeEncoder, + get_versions_with_file_changes, + int_version, + is_version_acceptable, + normalize_role, +) from .version import __version__ this_dir = os.path.dirname(os.path.realpath(__file__)) @@ -1313,7 +1321,7 @@ def create_user( email: str, password: str, workspace_id: int, - workspace_role: WorkspaceRole, + workspace_role: Union[str, WorkspaceRole], username: str = None, notify_user: bool = False, ) -> dict: @@ -1328,11 +1336,15 @@ def create_user( param notify_user: flag for email notifications - confirmation email will be sent """ self.check_collaborators_members_support() + role_enum = normalize_role(workspace_role, WorkspaceRole) + if role_enum is None: + raise ValueError("bad role") + params = { "email": email, "password": password, "workspace_id": workspace_id, - "role": workspace_role.value, + "role": role_enum.value, "notify_user": notify_user, } if username: @@ -1357,7 +1369,11 @@ def list_workspace_members(self, workspace_id: int) -> typing.List[dict]: return json.load(resp) def update_workspace_member( - self, workspace_id: int, user_id: int, workspace_role: WorkspaceRole, reset_projects_roles: bool = False + self, + workspace_id: int, + user_id: int, + workspace_role: Union[str, WorkspaceRole], + reset_projects_roles: bool = False, ) -> dict: """ Update workspace role of a workspace member, optionally resets the projects role @@ -1365,9 +1381,14 @@ def update_workspace_member( param reset_projects_roles: all project specific roles will be removed """ self.check_collaborators_members_support() + + role_enum = normalize_role(workspace_role, WorkspaceRole) + if role_enum is None: + raise ValueError("bad role") + params = { "reset_projects_roles": reset_projects_roles, - "workspace_role": workspace_role.value, + "workspace_role": role_enum.value, } workspace_member = self.patch(f"v2/workspaces/{workspace_id}/members/{user_id}", params, json_headers) return json.load(workspace_member) @@ -1387,7 +1408,7 @@ def list_project_collaborators(self, project_id: str) -> typing.List[dict]: project_collaborators = self.get(f"v2/projects/{project_id}/collaborators") return json.load(project_collaborators) - def add_project_collaborator(self, project_id: str, user: str, project_role: ProjectRole) -> dict: + def add_project_collaborator(self, project_id: str, user: str, project_role: Union[str, ProjectRole]) -> dict: """ Add a user to project collaborators and grant them a project role. Fails if user is already a member of the project. @@ -1395,17 +1416,27 @@ def add_project_collaborator(self, project_id: str, user: str, project_role: Pro param user: login (username or email) of the user """ self.check_collaborators_members_support() + + role_enum = normalize_role(project_role, ProjectRole) + if role_enum is None: + raise ValueError("bad role") + params = {"role": project_role.value, "user": user} project_collaborator = self.post(f"v2/projects/{project_id}/collaborators", params, json_headers) return json.load(project_collaborator) - def update_project_collaborator(self, project_id: str, user_id: int, project_role: ProjectRole) -> dict: + def update_project_collaborator(self, project_id: str, user_id: int, project_role: Union[str, ProjectRole]) -> dict: """ Update project role of the existing project collaborator. Fails if user is not a member of the project yet. """ self.check_collaborators_members_support() + + role_enum = normalize_role(project_role, ProjectRole) + if role_enum is None: + raise ValueError("bad role") params = {"role": project_role.value} + project_collaborator = self.patch(f"v2/projects/{project_id}/collaborators/{user_id}", params, json_headers) return json.load(project_collaborator) @@ -1481,13 +1512,18 @@ def send_logs( request = urllib.request.Request(url, data=payload, headers=header) return self._do_request(request) - def create_invitation(self, workspace_id: int, email: str, workspace_role: WorkspaceRole): + def create_invitation(self, workspace_id: int, email: str, workspace_role: Union[str, WorkspaceRole]): """ Create invitation to workspace for specific role """ min_version = "2025.6.1" if not is_version_acceptable(self.server_version(), min_version): raise NotImplementedError(f"This needs server at version {min_version} or later") - params = {"email": email, "role": workspace_role.value} + + role_enum = normalize_role(workspace_role, WorkspaceRole) + if role_enum is None: + raise ValueError("bad role") + + params = {"email": email, "role": role_enum.value} ws_inv = self.post(f"v2/workspaces/{workspace_id}/invitations", params, json_headers) return json.load(ws_inv) diff --git a/mergin/test/test_client.py b/mergin/test/test_client.py index 07cdeec..d0ab1de 100644 --- a/mergin/test/test_client.py +++ b/mergin/test/test_client.py @@ -38,6 +38,7 @@ unique_path_name, conflicted_copy_file_name, edit_conflict_file_name, + normalize_role, ) from ..merginproject import pygeodiff from ..report import create_report @@ -3026,3 +3027,16 @@ def test_server_type(mc): mock_client_get.side_effect = ClientError(detail="Service unavailable", http_error=503) with pytest.raises(ClientError, match="Service unavailable"): mc.server_type() + + +def test_string_roles(): + assert normalize_role("guest", WorkspaceRole) == WorkspaceRole.GUEST + assert normalize_role(" GuEsT ", WorkspaceRole) == WorkspaceRole.GUEST + assert normalize_role("writer", ProjectRole) == ProjectRole.WRITER + assert normalize_role(" WRITER ", ProjectRole) == ProjectRole.WRITER + + assert normalize_role("guuuest", WorkspaceRole) is None + assert normalize_role("ownerr", ProjectRole) is None + assert normalize_role("", WorkspaceRole) is None + assert normalize_role(None, WorkspaceRole) is None + assert normalize_role(123, WorkspaceRole) is None diff --git a/mergin/utils.py b/mergin/utils.py index c9b480a..43fa5a5 100644 --- a/mergin/utils.py +++ b/mergin/utils.py @@ -7,7 +7,9 @@ from datetime import datetime from pathlib import Path import tempfile -from .common import ClientError +from enum import Enum +from typing import Optional, Type, Union +from .common import ClientError, WorkspaceRole def generate_checksum(file, chunk_size=4096): @@ -309,3 +311,16 @@ def cleanup_tmp_dir(mp, tmp_dir: tempfile.TemporaryDirectory): mp.log.warning(f"Permission error during tmp dir cleanup: {tmp_dir.name}") except Exception as e: mp.log.error(f"Error during tmp dir cleanup: {tmp_dir.name}: {e}") + + +def normalize_role(role: Union[str, Enum], enum_cls: Type[Enum]) -> Optional[Enum]: + if isinstance(role, enum_cls): + return role + + if isinstance(role, str): + try: + return enum_cls(role.strip().lower()) + except ValueError: + return None + + return None