update iptrunk migration test, split off pp interaction unit tests into reusable method

test/__init__.py

 from uuid import uuid4
 
 LSO_RESULT_SUCCESS = {
-        "callback_result": {
-            "status": "success",
-            "job_id": str(uuid4()),
-            "output": "parsed_output",
-            "return_code": 0,
-        }
+    "callback_result": {
+        "status": "success",
+        "job_id": str(uuid4()),
+        "output": "parsed_output",
+        "return_code": 0,
     }
+}
 
 LSO_RESULT_FAILURE = {
-        "callback_result": {
-            "status": "failure",
-            "job_id": str(uuid4()),
-            "output": "parsed_output",
-            "return_code": 1,
-        }
+    "callback_result": {
+        "status": "failure",
+        "job_id": str(uuid4()),
+        "output": "parsed_output",
+        "return_code": 1,
     }
+}
 
 USER_CONFIRM_EMPTY_FORM = [{}]

test/workflows/__init__.py

@@ -14,6 +14,8 @@ from orchestrator.workflow import Process, ProcessStat, Step, Success, Workflow,
 from orchestrator.workflows import ALL_WORKFLOWS, LazyWorkflowInstance, get_workflow
 from pydantic_forms.core import post_form
 
+from test import LSO_RESULT_FAILURE, LSO_RESULT_SUCCESS, USER_CONFIRM_EMPTY_FORM
+
 logger = structlog.get_logger(__name__)
 
 
@@ -311,3 +313,32 @@ def user_accept_and_assert_suspended(process_stat, step_log, extra_data=None):
     assert_suspended(result)
 
     return result, step_log
+
+
+def assert_pp_interaction_success(result: Process, process_stat: ProcessStat, step_log: list):
+    """Assert a successful pp interaction in a workflow.
+
+    First, the workflow is awaiting callback. It is resumed but a result from LSO, after which the user submits the
+    confirmation input step. Two assertions are made: the workflow is awaiting callback at first, and suspended when
+    waiting for the user to confirm the results received.
+    """
+
+    assert_awaiting_callback(result)
+    result, step_log = resume_workflow(process_stat, step_log, input_data=LSO_RESULT_SUCCESS)
+    assert_suspended(result)
+
+    return resume_workflow(process_stat, step_log, input_data=USER_CONFIRM_EMPTY_FORM)
+
+
+def assert_pp_interaction_failure(result: Process, process_stat: ProcessStat, step_log: list):
+    """Assert a failed pp interaction in a workflow.
+
+    First, the workflow is awaiting callback. It is resumed by a "failure" result from LSO, after which the workflow is
+    in a failed state. This failed state is also returned. Two assertions are made: the workflow is awaiting callback at
+    first, and failed when the result is received from LSO.
+    """
+    assert_awaiting_callback(result)
+    result, step_log = resume_workflow(process_stat, step_log, input_data=LSO_RESULT_FAILURE)
+    assert_failed(result)
+
+    return result, step_log

test/workflows/iptrunk/test_create_iptrunk.py

@@ -8,16 +8,13 @@ from gso.products.product_blocks.iptrunk import IptrunkType, PhyPortCapacity
 from gso.services.crm import customer_selector, get_customer_by_name
 from gso.services.subscriptions import get_product_id_by_name
 from gso.utils.helpers import LAGMember
-from test import LSO_RESULT_SUCCESS, LSO_RESULT_FAILURE, USER_CONFIRM_EMPTY_FORM
 from test.services.conftest import MockedNetboxClient
 from test.workflows import (
-    assert_aborted,
-    assert_awaiting_callback,
     assert_complete,
-    assert_suspended,
+    assert_pp_interaction_failure,
+    assert_pp_interaction_success,
     extract_state,
-    resume_workflow,
-    run_workflow, assert_failed,
+    run_workflow,
 )
 
 
@@ -110,10 +107,7 @@ def test_successful_iptrunk_creation_with_standard_lso_result(
     result, process_stat, step_log = run_workflow("create_iptrunk", initial_site_data)
 
     for _ in range(6):
-        assert_awaiting_callback(result)
-        result, step_log = resume_workflow(process_stat, step_log, input_data=LSO_RESULT_SUCCESS)
-        assert_suspended(result)
-        result, step_log = resume_workflow(process_stat, step_log, input_data=[{}])
+        result, step_log = assert_pp_interaction_success(result, process_stat, step_log)
 
     assert_complete(result)
 
@@ -151,13 +145,8 @@ def test_iptrunk_creation_fails_when_lso_return_code_is_one(
     initial_site_data = [{"product": product_id}, *input_form_wizard_data]
     result, process_stat, step_log = run_workflow("create_iptrunk", initial_site_data)
 
-    assert_awaiting_callback(result)
-    result, step_log = resume_workflow(process_stat, step_log, input_data=LSO_RESULT_SUCCESS)
-    assert_suspended(result)
-    result, step_log = resume_workflow(process_stat, step_log, input_data=USER_CONFIRM_EMPTY_FORM)
+    result, step_log = assert_pp_interaction_success(result, process_stat, step_log)
 
-    assert_awaiting_callback(result)
-    result, step_log = resume_workflow(process_stat, step_log, input_data=LSO_RESULT_FAILURE)
-    assert_failed(result)
+    assert_pp_interaction_failure(result, process_stat, step_log)
 
     assert mock_check_ip_trunk.call_count == 0

test/workflows/iptrunk/test_migrate_iptrunk.py

+from os import PathLike
 from unittest.mock import patch
 
 import pytest
 
 from gso.products import Iptrunk
 from gso.utils.helpers import LAGMember
+from test import USER_CONFIRM_EMPTY_FORM
 from test.workflows import (
     assert_complete,
+    assert_pp_interaction_success,
     assert_suspended,
     extract_state,
     resume_workflow,
     run_workflow,
-    user_accept_and_assert_suspended,
 )
 from test.workflows.iptrunk.test_create_iptrunk import MockedNetboxClient
 
@@ -42,6 +44,7 @@ def test_migrate_iptrunk_success(
     iptrunk_subscription_factory,
     router_subscription_factory,
     faker,
+    data_config_filename: PathLike,
 ):
     #  Set up mock return values
     mocked_netbox = MockedNetboxClient()
@@ -81,30 +84,20 @@ def test_migrate_iptrunk_success(
     ]
 
     result, process_stat, step_log = run_workflow("migrate_iptrunk", migrate_form_input)
-    assert_suspended(result)
 
-    lso_return = {
-        "pp_run_results": {
-            "status": "ok",
-            "job_id": faker.uuid4(),
-            "output": "parsed_output",
-            "return_code": 0,
-        },
-        "confirm": "ACCEPTED",
-    }
-    # Resume steps
     for _ in range(5):
-        result, step_log = user_accept_and_assert_suspended(process_stat, step_log, lso_return)
-        result, step_log = user_accept_and_assert_suspended(process_stat, step_log, [{}, {}])
-    for _ in range(2):
-        result, step_log = user_accept_and_assert_suspended(process_stat, step_log)
-        result, step_log = user_accept_and_assert_suspended(process_stat, step_log, lso_return)
-        result, step_log = user_accept_and_assert_suspended(process_stat, step_log, [{}, {}])
-
-    result, step_log = user_accept_and_assert_suspended(process_stat, step_log, lso_return)
-    result, step_log = user_accept_and_assert_suspended(process_stat, step_log, [{}, {}])
-    result, step_log = user_accept_and_assert_suspended(process_stat, step_log, lso_return)
-    result, step_log = resume_workflow(process_stat, step_log, [{}, {}])
+        result, step_log = assert_pp_interaction_success(result, process_stat, step_log)
+
+    assert_suspended(result)
+    result, step_log = resume_workflow(process_stat, step_log, input_data=USER_CONFIRM_EMPTY_FORM)
+
+    result, step_log = assert_pp_interaction_success(result, process_stat, step_log)
+
+    assert_suspended(result)
+    result, step_log = resume_workflow(process_stat, step_log, input_data=USER_CONFIRM_EMPTY_FORM)
+
+    for _ in range(3):
+        result, step_log = assert_pp_interaction_success(result, process_stat, step_log)
 
     assert_complete(result)