diff --git a/test/__init__.py b/test/__init__.py
index b419ed13c4f512880f1b1b16e6f7f4152678c066..17c55014877312ca91ee846b5639eda4a36597eb 100644
--- a/test/__init__.py
+++ b/test/__init__.py
@@ -1,21 +1,21 @@
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 = [{}]
diff --git a/test/workflows/__init__.py b/test/workflows/__init__.py
index 596393002c6ce5c2f273c329be374861d9e0c5db..3843a79b6d7e1fa7c990600e664e3cc0134971b4 100644
--- a/test/workflows/__init__.py
+++ b/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
diff --git a/test/workflows/iptrunk/test_create_iptrunk.py b/test/workflows/iptrunk/test_create_iptrunk.py
index efe9aa79fe099070777c5d8cfa0f4ef775c84075..10a87d1253033f7964bb7482db18e6498586d059 100644
--- a/test/workflows/iptrunk/test_create_iptrunk.py
+++ b/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
diff --git a/test/workflows/iptrunk/test_migrate_iptrunk.py b/test/workflows/iptrunk/test_migrate_iptrunk.py
index 89eb44f55cb9ea5776487e760d8335101b3980f4..02b4f268e9837e4172ceb3350e4ed1c1792eea62 100644
--- a/test/workflows/iptrunk/test_migrate_iptrunk.py
+++ b/test/workflows/iptrunk/test_migrate_iptrunk.py
@@ -1,16 +1,18 @@
+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)