Skip to content
Snippets Groups Projects

Make use of new callback step for external provisioning

Merged Karel van Klink requested to merge feature/use-async-steps into develop
All threads resolved!
4 files
+ 65
52
Compare changes
  • Side-by-side
  • Inline
Files
4
@@ -8,16 +8,13 @@ from gso.products.product_blocks.iptrunk import IptrunkType, PhyPortCapacity
@@ -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.crm import customer_selector, get_customer_by_name
from gso.services.subscriptions import get_product_id_by_name
from gso.services.subscriptions import get_product_id_by_name
from gso.utils.helpers import LAGMember
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.services.conftest import MockedNetboxClient
from test.workflows import (
from test.workflows import (
assert_aborted,
assert_awaiting_callback,
assert_complete,
assert_complete,
assert_suspended,
assert_pp_interaction_failure,
 
assert_pp_interaction_success,
extract_state,
extract_state,
resume_workflow,
run_workflow,
run_workflow, assert_failed,
)
)
@@ -110,10 +107,7 @@ def test_successful_iptrunk_creation_with_standard_lso_result(
@@ -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)
result, process_stat, step_log = run_workflow("create_iptrunk", initial_site_data)
for _ in range(6):
for _ in range(6):
assert_awaiting_callback(result)
result, step_log = assert_pp_interaction_success(result, process_stat, step_log)
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=[{}])
assert_complete(result)
assert_complete(result)
@@ -151,13 +145,8 @@ def test_iptrunk_creation_fails_when_lso_return_code_is_one(
@@ -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]
initial_site_data = [{"product": product_id}, *input_form_wizard_data]
result, process_stat, step_log = run_workflow("create_iptrunk", initial_site_data)
result, process_stat, step_log = run_workflow("create_iptrunk", initial_site_data)
assert_awaiting_callback(result)
result, step_log = assert_pp_interaction_success(result, process_stat, step_log)
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)
assert_awaiting_callback(result)
assert_pp_interaction_failure(result, process_stat, step_log)
result, step_log = resume_workflow(process_stat, step_log, input_data=LSO_RESULT_FAILURE)
assert_failed(result)
assert mock_check_ip_trunk.call_count == 0
assert mock_check_ip_trunk.call_count == 0
Loading