FWF-3718: [Feature] Add testcase

AOT-Technologies · Nov 5, 2024 · 7e48501 · 7e48501
1 parent c957dd3
commit 7e48501
Show file tree

Hide file tree

Showing 3 changed files with 115 additions and 1 deletion.
diff --git a/forms-flow-api/src/formsflow_api/constants/__init__.py b/forms-flow-api/src/formsflow_api/constants/__init__.py
@@ -110,6 +110,7 @@ class BusinessErrorCode(ErrorCodeMixin, Enum):
         HTTPStatus.BAD_REQUEST,
     )
     FORM_VALIDATION_FAILED = "FORM_VALIDATION_FAILED.", HTTPStatus.BAD_REQUEST
+    INVALID_PROCESS = "Invalid process.", HTTPStatus.BAD_REQUEST
 
     def __new__(cls, message, status_code):
         """Constructor."""

diff --git a/forms-flow-api/src/formsflow_api/services/process.py b/forms-flow-api/src/formsflow_api/services/process.py
@@ -682,6 +682,10 @@ def migrate(cls, request):
         data = MigrateRequestSchema().load(request.get_json())
         process_key = data.get("process_key")
         mapper_id = data.get("mapper_id")
+        mapper = FormProcessMapper.find_form_by_id(mapper_id)
+        # If the process_key in the mapper is different from the process_key in the payload
+        if mapper.process_key != process_key:
+            raise BusinessException(BusinessErrorCode.INVALID_PROCESS)
         mappers = FormProcessMapper.get_mappers_by_process_key(process_key, mapper_id)
         current_app.logger.debug(f"Mappers found..{mappers}")
         if mappers:
@@ -717,6 +721,5 @@ def migrate(cls, request):
                     }
                 )
             # Update is_migrated to main mapper by id.
-            mapper = FormProcessMapper.find_form_by_id(mapper_id)
             mapper.update({"is_migrated": True})
         return {}
diff --git a/forms-flow-api/tests/unit/api/test_process.py b/forms-flow-api/tests/unit/api/test_process.py
@@ -568,3 +568,113 @@ def test_get_process_by_key_invalid_key(self, app, client, session, jwt):
             headers=headers,
         )
         assert response.status_code == 400
+
+
+class MigrateProcess:
+    """Test suite for the migrate process."""
+
+    def migrate_process_success(self, app, client, session, jwt, create_mapper_custom):
+        """Migrate process with success."""
+        payload = {
+            "formId": "1234",
+            "formName": "Sample form1",
+            "processKey": "onestepapproval",
+            "processName": "One Step Approval",
+            "status": "inactive",
+            "formType": "form",
+            "parentFormId": "1234",
+            "is_migrated": False,
+        }
+        rv = create_mapper_custom(payload)
+        mapper_id = rv["id"]
+        payload = {
+            "formId": "12345",
+            "formName": "Sample form2",
+            "processKey": "onestepapproval",
+            "processName": "One Step Approval",
+            "status": "inactive",
+            "formType": "form",
+            "parentFormId": "12345",
+            "is_migrated": False,
+        }
+        rv = create_mapper_custom(payload)
+        token = get_token(jwt, role=CREATE_DESIGNS, username="designer")
+        headers = {
+            "Authorization": f"Bearer {token}",
+            "content-type": "application/json",
+        }
+        rv = client.post(
+            "/process/migrate",
+            headers=headers,
+            json={"mapperId": mapper_id, "processKey": "onestepapproval"},
+        )
+        assert rv.status_code == 200
+
+    def migrate_process_unauthorized(self, app, client, session, jwt, create_mapper_custom):
+        """Migrate process without proper authorization."""
+        payload = {
+            "formId": "1234",
+            "formName": "Sample form1",
+            "processKey": "onestepapproval",
+            "processName": "One Step Approval",
+            "status": "inactive",
+            "formType": "form",
+            "parentFormId": "1234",
+            "is_migrated": False,
+        }
+        rv = create_mapper_custom(payload)
+        mapper_id = rv["id"]
+        payload = {
+            "formId": "12345",
+            "formName": "Sample form2",
+            "processKey": "onestepapproval",
+            "processName": "One Step Approval",
+            "status": "inactive",
+            "formType": "form",
+            "parentFormId": "12345",
+            "is_migrated": False,
+        }
+        rv = create_mapper_custom(payload)
+        rv = client.post(
+            "/process/migrate",
+            json={"mapperId": mapper_id, "processKey": "onestepapproval"},
+        )
+        assert rv.status_code == 401
+
+    def migrate_process_invalid(self, app, client, session, jwt, create_mapper_custom):
+        """Migrate process with invalid data."""
+        payload = {
+            "formId": "1234",
+            "formName": "Sample form1",
+            "processKey": "onestepapproval",
+            "processName": "One Step Approval",
+            "status": "inactive",
+            "formType": "form",
+            "parentFormId": "1234",
+            "is_migrated": False,
+        }
+        rv = create_mapper_custom(payload)
+        mapper_id = rv["id"]
+        payload = {
+            "formId": "12345",
+            "formName": "Sample form2",
+            "processKey": "onestepapproval",
+            "processName": "One Step Approval",
+            "status": "inactive",
+            "formType": "form",
+            "parentFormId": "12345",
+            "is_migrated": False,
+        }
+        rv = create_mapper_custom(payload)
+        token = get_token(jwt, role=CREATE_DESIGNS, username="designer")
+        headers = {
+            "Authorization": f"Bearer {token}",
+            "content-type": "application/json",
+        }
+        # Test with different process_key other than mapper process_key
+        rv = client.post(
+            "/process/migrate",
+            headers=headers,
+            json={"mapperId": mapper_id, "processKey": "twostepapproval"},
+        )
+        assert rv.status_code == 400