introduce helper to normalize workspace/project roles from strings

.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