Merge pull request #284 from MerginMaps/roles_as_string

introduce helper to normalize workspace/project roles from strings

Author: MarcelGeo

.gitignore

@@ -12,4 +12,4 @@ htmlcov
 .pytest_cache
 deps
 venv
-.vscode/settings.json
+.vscode/

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(f"Invalid role: {workspace_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,17 +1369,26 @@ 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
 
         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(f"Invalid role: {workspace_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,25 +1408,35 @@ 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.
 
         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(f"Invalid role: {project_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(f"Invalid role: {project_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(f"Invalid role: {workspace_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)

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,25 @@ 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()
+
+
+@pytest.mark.parametrize(
+    "value, role_enum, expected",
+    [
+        ("guest", WorkspaceRole, WorkspaceRole.GUEST),
+        ("  GuEsT  ", WorkspaceRole, WorkspaceRole.GUEST),
+        ("writer", ProjectRole, ProjectRole.WRITER),
+        ("  WRITER ", ProjectRole, ProjectRole.WRITER),
+        ("guuuest", WorkspaceRole, None),
+        ("ownerr", ProjectRole, None),
+        ("", WorkspaceRole, None),
+        (None, WorkspaceRole, None),
+        (123, WorkspaceRole, None),
+    ],
+)
+def test_normalize_role_parametrized(value, role_enum, expected):
+    result = normalize_role(value, role_enum)
+    if expected is None:
+        assert result is None
+    else:
+        assert result == expected

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,20 @@ 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]:
+    """
+    Takes a role as a string or an Enum member and returns the corresponding Enum member
+    from the given enum class. Returns None if the input is invalid or no match is found.
+    """
+    if isinstance(role, enum_cls):
+        return role
+
+    if isinstance(role, str):
+        try:
+            return enum_cls(role.strip().lower())
+        except ValueError:
+            return None
+
+    return None