Skip to content
Snippets Groups Projects

add 2 test cases for create_iptrunk workflow and one for create_site

Closed Mohammad Torkashvand requested to merge feature/NAT-262-add-tests-for-workflows into develop
1 file
+ 10
8
Compare changes
  • Side-by-side
  • Inline
@@ -6,6 +6,7 @@ from gso.products import Iptrunk
from gso.products.product_blocks import PhyPortCapacity
from gso.products.product_blocks.iptrunk import IptrunkType
from gso.schemas.enums import ProductType
from gso.services.crm import get_customer_by_name
from gso.services.subscriptions import get_product_id_by_name
from gso.workflows.utils import customer_selector
from test.workflows import (
@@ -24,7 +25,8 @@ def input_form_wizard_data(router_subscription_factory, faker):
router_side_b = router_subscription_factory()
create_ip_trunk_step = {
"customer": getattr(customer_selector(), "8f0df561-ce9d-4d9c-89a8-7953d3ffc961"),
"tt_number": faker.pystr(),
"customer": getattr(customer_selector(), get_customer_by_name("GÉANT")["id"]),
"geant_s_sid": faker.pystr(),
"iptrunk_type": IptrunkType.DARK_FIBER,
"iptrunk_description": faker.sentence(),
@@ -63,7 +65,7 @@ def _user_accept_and_assert_suspended(process_stat, step_log, extra_data=None):
@patch("gso.workflows.iptrunk.create_iptrunk.provisioning_proxy.provision_ip_trunk")
@patch("gso.workflows.iptrunk.create_iptrunk.infoblox.allocate_v6_network")
@patch("gso.workflows.iptrunk.create_iptrunk.infoblox.allocate_v4_network")
def test_successful_iptrunk_creation_with_standard_resume_data(
def test_successful_iptrunk_creation_with_standard_lso_data(
mock_allocate_v4_network,
mock_allocate_v6_network,
mock_provision_ip_trunk,
@@ -79,7 +81,7 @@ def test_successful_iptrunk_creation_with_standard_resume_data(
result, process_stat, step_log = run_workflow("create_iptrunk", initial_site_data)
assert_suspended(result)
standard_resume_data = {
standard_lso_data = {
"pp_run_results": {
"status": "ok",
"job_id": "random_job_id",
@@ -89,10 +91,10 @@ def test_successful_iptrunk_creation_with_standard_resume_data(
"confirm": "ACCEPTED",
}
for _ in range(5):
result, step_log = _user_accept_and_assert_suspended(process_stat, step_log, standard_resume_data)
result, step_log = _user_accept_and_assert_suspended(process_stat, step_log, standard_lso_data)
result, step_log = _user_accept_and_assert_suspended(process_stat, step_log)
result, step_log = _user_accept_and_assert_suspended(process_stat, step_log, standard_resume_data)
result, step_log = _user_accept_and_assert_suspended(process_stat, step_log, standard_lso_data)
result, step_log = resume_workflow(process_stat, step_log, {})
assert_complete(result)
@@ -129,7 +131,7 @@ def test_iptrunk_creation_fails_when_lso_return_code_is_one(
result, process_stat, step_log = run_workflow("create_iptrunk", initial_site_data)
assert_suspended(result)
standard_resume_data = {
standard_lso_data = {
"pp_run_results": {
"status": "ok",
"job_id": "random_job_id",
@@ -141,10 +143,10 @@ def test_iptrunk_creation_fails_when_lso_return_code_is_one(
attempts = 3
for _ in range(0, attempts - 1):
result, step_log = _user_accept_and_assert_suspended(process_stat, step_log, standard_resume_data)
result, step_log = _user_accept_and_assert_suspended(process_stat, step_log, standard_lso_data)
result, step_log = _user_accept_and_assert_suspended(process_stat, step_log)
result, step_log = _user_accept_and_assert_suspended(process_stat, step_log, standard_resume_data)
result, step_log = _user_accept_and_assert_suspended(process_stat, step_log, standard_lso_data)
result, step_log = resume_workflow(process_stat, step_log, {})
assert_aborted(result)
Loading