added pause/resume user_activity

src/superannotate/lib/app/interface/sdk_interface.py

@@ -71,6 +71,7 @@
 from lib.infrastructure.utils import extract_project_folder
 from lib.infrastructure.validators import wrap_error
 from lib.app.serializers import WMProjectSerializer
+from lib.core.entities.work_managament import WMUserTypeEnum
 
 logger = logging.getLogger("sa")
 
@@ -453,6 +454,32 @@ def list_users(self, *, include: List[Literal["custom_fields"]] = None, **filter
             self.controller.work_management.list_users(include=include, **filters)
         )
 
+    def pause_user_activity(
+        self, pk: Union[int, str], projects: Union[List[int], List[str], Literal["*"]]
+    ):
+        user = self.controller.work_management.get_user_metadata(pk=pk)
+        if user.role is not WMUserTypeEnum.Contributor:
+            raise AppException("User must have a contributor role to pause activity.")
+        self.controller.work_management.update_user_activity(
+            user_email=user.email, provided_projects=projects, action="pause"
+        )
+        logger.info(
+            f"User with email {user.email} has been successfully paused from the specified projects: {projects}."
+        )
+
+    def resume_user_activity(
+        self, pk: Union[int, str], projects: Union[List[int], List[str], Literal["*"]]
+    ):
+        user = self.controller.work_management.get_user_metadata(pk=pk)
+        if user.role is not WMUserTypeEnum.Contributor:
+            raise AppException("User must have a contributor role to resume activity.")
+        self.controller.work_management.update_user_activity(
+            user_email=user.email, provided_projects=projects, action="resume"
+        )
+        logger.info(
+            f"User with email {user.email} has been successfully unblocked from the specified projects: {projects}."
+        )
+
     def get_component_config(self, project: Union[NotEmptyStr, int], component_id: str):
         """
         Retrieves the configuration for a given project and component ID.

src/superannotate/lib/core/serviceproviders.py

@@ -186,6 +186,12 @@ def set_custom_field_value(
     ):
         raise NotImplementedError
 
+    @abstractmethod
+    def update_user_activity(
+        self, body_query: Query, action=Literal["resume", "pause"]
+    ) -> ServiceResponse:
+        raise NotImplementedError
+
 
 class BaseProjectService(SuperannotateServiceProvider):
     @abstractmethod

src/superannotate/lib/infrastructure/controller.py

@@ -196,6 +196,44 @@ def list_users(self, include: List[Literal["custom_fields"]] = None, **filters):
             return response.data
         return self.service_provider.work_management.list_users(query).data
 
+    def update_user_activity(
+        self,
+        user_email: str,
+        provided_projects: Union[List[int], List[str], Literal["*"]],
+        action: Literal["resume", "pause"],
+    ):
+        if isinstance(provided_projects, list):
+            if not provided_projects:
+                raise AppException("Provided projects list cannot be empty.")
+            body_query = EmptyQuery()
+            if isinstance(provided_projects[0], int):
+                body_query &= Filter("id", provided_projects, OperatorEnum.IN)
+            else:
+                body_query &= Filter("name", provided_projects, OperatorEnum.IN)
+            exist_projects = self.service_provider.work_management.search_projects(
+                body_query
+            ).res_data
+
+            # project validation
+            if len(set(provided_projects)) > len(exist_projects):
+                raise AppException("Invalid project(s) provided.")
+        else:
+            exist_projects = self.service_provider.work_management.search_projects(
+                EmptyQuery()
+            ).res_data
+
+        chunked_projects_ids = divide_to_chunks([i.id for i in exist_projects], 50)
+        for chunk in chunked_projects_ids:
+            body_query = EmptyQuery()
+            body_query &= Filter("projects.id", chunk, OperatorEnum.IN)
+            body_query &= Filter(
+                "projects.contributors.email", user_email, OperatorEnum.EQ
+            )
+            res = self.service_provider.work_management.update_user_activity(
+                body_query=body_query, action=action
+            )
+            res.raise_for_status()
+
 
 class ProjectManager(BaseManager):
     def __init__(self, service_provider: ServiceProvider, team: TeamEntity):

src/superannotate/lib/infrastructure/services/work_management.py

@@ -12,6 +12,7 @@
 from lib.core.jsx_conditions import Filter
 from lib.core.jsx_conditions import OperatorEnum
 from lib.core.jsx_conditions import Query
+from lib.core.pydantic_v1 import Literal
 from lib.core.service_types import ListCategoryResponse
 from lib.core.service_types import ServiceResponse
 from lib.core.service_types import WMCustomFieldResponse
@@ -60,6 +61,7 @@ class WorkManagementService(BaseWorkManagementService):
     URL_SEARCH_CUSTOM_ENTITIES = "customentities/search"
     URL_SEARCH_TEAM_USERS = "teamusers/search"
     URL_SEARCH_PROJECTS = "projects/search"
+    URL_RESUME_PAUSE_USER = "teams/editprojectsusers"
 
     @staticmethod
     def _generate_context(**kwargs):
@@ -362,3 +364,22 @@ def set_custom_field_value(
                 "parentEntity": parent_entity.value,
             },
         )
+
+    def update_user_activity(
+        self, body_query: Query, action=Literal["resume", "pause"]
+    ) -> ServiceResponse:
+        """resume or pause user by projects"""
+        body = body_query.body_builder()
+        body["body"] = {
+            "projectUsers": {"permissions": {"paused": 1 if action == "pause" else 0}}
+        }
+        return self.client.request(
+            url=self.URL_RESUME_PAUSE_USER,
+            method="post",
+            data=body,
+            headers={
+                "x-sa-entity-context": self._generate_context(
+                    team_id=self.client.team_id
+                ),
+            },
+        )

tests/integration/work_management/test_pause_resume_user_activity.py

@@ -0,0 +1,70 @@
+from lib.core.exceptions import AppException
+from src.superannotate import SAClient
+from tests.integration.base import BaseTestCase
+
+sa = SAClient()
+
+
+class TestPauseUserActivity(BaseTestCase):
+    PROJECT_NAME = "TestPauseUserActivity"
+    PROJECT_TYPE = "Vector"
+    PROJECT_DESCRIPTION = "DESCRIPTION"
+    ATTACHMENT_LIST = [
+        {
+            "url": "https://drive.google.com/uc?export=download&id=1vwfCpTzcjxoEA4hhDxqapPOVvLVeS7ZS",
+            "name": "6022a74d5384c50017c366b3",
+        },
+        {
+            "url": "https://drive.google.com/uc?export=download&id=1geS2YtQiTYuiduEirKVYxBujHJaIWA3V",
+            "name": "6022a74b5384c50017c366ad",
+        },
+        {
+            "url": "https://drive.google.com/uc?export=download&id=1geS2YtQiTYuiduEirKVYxBujHJaIWA3V",
+            "name": "6022a74b5384c50017c366ad",
+        },
+    ]
+
+    def setUp(self, *args, **kwargs):
+        super().setUp(*args, **kwargs)
+        uploaded, _, _ = sa.attach_items(self.PROJECT_NAME, self.ATTACHMENT_LIST)
+        users = sa.list_users()
+        self.scapegoat = [u for u in users if u["role"] == "Contributor"][0]
+        sa.add_contributors_to_project(
+            self.PROJECT_NAME, [self.scapegoat["email"]], "QA"
+        )
+
+    def test_pause_and_resume_user_activity(self):
+        with self.assertLogs("sa", level="INFO") as cm:
+            sa.pause_user_activity(
+                pk=self.scapegoat["email"], projects=[self.PROJECT_NAME]
+            )
+            assert (
+                cm.output[0]
+                == f"INFO:sa:User with email {self.scapegoat['email']} has been successfully paused"
+                f" from the specified projects: {[self.PROJECT_NAME]}."
+            )
+        with self.assertRaisesRegexp(
+            AppException,
+            "The user does not have the required permissions for this assignment.",
+        ):
+            sa.assign_items(
+                self.PROJECT_NAME,
+                [i["name"] for i in self.ATTACHMENT_LIST],
+                self.scapegoat["email"],
+            )
+
+        with self.assertLogs("sa", level="INFO") as cm:
+            sa.resume_user_activity(
+                pk=self.scapegoat["email"], projects=[self.PROJECT_NAME]
+            )
+            assert (
+                cm.output[0]
+                == f"INFO:sa:User with email {self.scapegoat['email']} has been successfully unblocked"
+                f" from the specified projects: {[self.PROJECT_NAME]}."
+            )
+
+        sa.assign_items(
+            self.PROJECT_NAME,
+            [i["name"] for i in self.ATTACHMENT_LIST],
+            self.scapegoat["email"],
+        )