updated custom_workflow test setup

Author: Narek Mkhitaryan

src/superannotate/lib/core/entities/project.py

@@ -163,6 +163,7 @@ class TeamEntity(BaseModel):
     users: Optional[List[UserEntity]]
     pending_invitations: Optional[List[Any]]
     creator_id: Optional[str]
+    owner_id: Optional[str]
 
     class Config:
         extra = Extra.ignore

src/superannotate/lib/infrastructure/controller.py

@@ -909,6 +909,10 @@ def __init__(self, config: ConfigEntity):
         self.subsets = SubsetManager(self.service_provider)
         self.integrations = IntegrationManager(self.service_provider)
 
+    @property
+    def org_id(self):
+        return self._team.owner_id
+
     @property
     def current_user(self):
         return self._user

src/superannotate/lib/infrastructure/serviceprovider.py

@@ -1,3 +1,4 @@
+import base64
 import datetime
 from typing import List
 
@@ -36,6 +37,7 @@ class ServiceProvider(BaseServiceProvider):
     URL_FOLDERS_IMAGES = "images-folders"
     URL_INVITE_CONTRIBUTORS = "api/v1/team/{}/inviteUsers"
     URL_ANNOTATION_UPLOAD_PATH_TOKEN = "images/getAnnotationsPathsAndTokens"
+    URL_CREATE_WORKFLOW = "api/v1/workflows/submit"
 
     def __init__(self, client: HttpClient):
         self.enum_mapping = {"approval_status": ApprovalStatus.get_mapping()}
@@ -254,3 +256,17 @@ def invite_contributors(self, team_id: int, team_role: int, emails: List[str]):
             "post",
             data=dict(emails=emails, team_role=team_role),
         )
+
+    def create_custom_workflow(self, org_id: str, data: dict):
+        return self.client.request(
+            url=self.URL_CREATE_WORKFLOW,
+            method="post",
+            headers={
+                "x-sa-entity-context": base64.b64encode(
+                    f'{{"team_id":{self.client.team_id},"organization_id":"{org_id}"}}'.encode(
+                        "utf-8"
+                    )
+                ).decode()
+            },
+            data=data,
+        )

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

@@ -13,6 +13,8 @@ class WorkManagementService(BaseWorkManagementService):
     URL_LIST = "workflows"
     URL_LIST_STATUSES = "workflows/{workflow_id}/workflowstatuses"
     URL_LIST_ROLES = "workflows/{workflow_id}/workflowroles"
+    URL_CREATE_ROLE = "roles"
+    URL_CREATE_STATUS = "statuses"
 
     def get_workflow(self, pk: int) -> WorkflowEntity:
         response = self.list_workflows(Filter("id", pk, OperatorEnum.EQ))
@@ -65,3 +67,31 @@ def list_workflow_roles(self, project_id: int, workflow_id: int):
                 "join": "role",
             },
         )
+
+    def create_custom_role(self, org_id: str, data: dict):
+        return self.client.request(
+            url=self.URL_CREATE_ROLE,
+            method="post",
+            headers={
+                "x-sa-entity-context": base64.b64encode(
+                    f'{{"team_id":{self.client.team_id},"organization_id":"{org_id}"}}'.encode(
+                        "utf-8"
+                    )
+                ).decode()
+            },
+            data=data,
+        )
+
+    def create_custom_status(self, org_id: str, data: dict):
+        return self.client.request(
+            url=self.URL_CREATE_STATUS,
+            method="post",
+            headers={
+                "x-sa-entity-context": base64.b64encode(
+                    f'{{"team_id":{self.client.team_id},"organization_id":"{org_id}"}}'.encode(
+                        "utf-8"
+                    )
+                ).decode()
+            },
+            data=data,
+        )

tests/applicatoin/custom_workflow.py

@@ -1,4 +1,5 @@
 import contextvars
+import json
 import os
 import re
 import time
@@ -24,9 +25,39 @@ class TestWorkflow(TestCase):
     CLASSES_PATH = "sample_project_vector/classes/classes.json"
     ANNOTATIONS_PATH = "sample_project_vector"
     PROJECT_TYPE = "Vector"
+    CUSTOM_WORKFLOW = "application/custom_workflow_payload.json"
 
     @classmethod
     def setUpClass(cls, *args, **kwargs):
+
+        # setup custom role
+        sa.controller.service_provider.work_management.create_custom_role(
+            org_id=sa.controller.org_id,
+            data={
+                "name": "CustomRole",
+                "description": "Test custom role",
+                "rolePermissions": [{"permission_id": 11}, {"permission_id": 12}],
+            },
+        )
+
+        # setup custom status
+        sa.controller.service_provider.work_management.create_custom_status(
+            org_id=sa.controller.org_id,
+            data={
+                "name": "CustomStatus",
+                "icon_id": 7,
+                "shortcut_id": 7,
+                "description": "test status",
+            },
+        )
+
+        # setup custom workflow
+        with open(os.path.join(DATA_SET_PATH, cls.CUSTOM_WORKFLOW)) as f:
+            sa.controller.service_provider.create_custom_workflow(
+                org_id=sa.controller.org_id,
+                data=json.load(f),
+            )
+
         cls.tearDownClass()
         cls._project = sa.create_project(
             cls.PROJECT_NAME, cls.PROJECT_DESCRIPTION, cls.PROJECT_TYPE, workflow="ttp"
@@ -37,12 +68,9 @@ def tearDownClass(cls) -> None:
         try:
             projects = sa.search_projects(cls.PROJECT_NAME, return_metadata=True)
             for project in projects:
-                try:
-                    sa.delete_project(project)
-                except Exception as e:
-                    print(str(e))
-        except Exception as e:
-            print(str(e))
+                sa.delete_project(project)
+        except Exception:
+            pass
 
     @property
     def classes_path(self):

tests/data_set/application/custom_workflow_payload.json

@@ -0,0 +1,80 @@
+{
+    "name": "ttp",
+    "description": "mock workflow for test",
+    "raw_config": {
+        "roles": [
+            "Annotator",
+            "QA",
+            "CustomRole"
+        ],
+        "statuses": [
+            "NotStarted",
+            "QualityCheck",
+            "Returned",
+            "Skipped",
+            "Completed",
+            "CustomStatus"
+        ],
+        "transitions": [
+            {
+                "to": "QualityCheck",
+                "from": "NotStarted",
+                "name": "move to in progress",
+                "allowed_roles": [
+                    "Annotator"
+                ]
+            },
+            {
+                "to": "QualityCheck",
+                "from": "Returned",
+                "name": "move to quality check",
+                "allowed_roles": [
+                    "Annotator",
+                    "QA"
+                ]
+            },
+            {
+                "to": "Returned",
+                "from": "QualityCheck",
+                "name": "move back to in progress",
+                "allowed_roles": [
+                    "Annotator",
+                    "QA"
+                ]
+            },
+            {
+                "to": "Completed",
+                "from": "QualityCheck",
+                "name": "move to completed",
+                "allowed_roles": [
+                    "QA"
+                ]
+            }
+        ],
+        "start_status": "Returned",
+        "allowed_resources_to_roles": {
+            "QA": {
+                "view_items": {
+                    "statuses": [
+                        "QualityCheck"
+                    ]
+                }
+            },
+            "Annotator": {
+                "view_items": {
+                    "statuses": [
+                        "NotStarted",
+                        "Returned"
+                    ]
+                }
+            },
+            "CustomRole": {
+                "view_items": {
+                    "statuses": [
+                        "CustomStatus"
+                    ]
+                }
+            }
+        }
+    }
+}