Skip to content
Snippets Groups Projects
Commit ece49e4a authored by Mohammad Torkashvand's avatar Mohammad Torkashvand
Browse files

add tt_number to input_form_wizard_data

parent 1fccf4e3
No related branches found
No related tags found
1 merge request!71add 2 test cases for create_iptrunk workflow and one for create_site
Pipeline #84090 passed
...@@ -6,6 +6,7 @@ from gso.products import Iptrunk ...@@ -6,6 +6,7 @@ from gso.products import Iptrunk
from gso.products.product_blocks import PhyPortCapacity from gso.products.product_blocks import PhyPortCapacity
from gso.products.product_blocks.iptrunk import IptrunkType from gso.products.product_blocks.iptrunk import IptrunkType
from gso.schemas.enums import ProductType 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.services.subscriptions import get_product_id_by_name
from gso.workflows.utils import customer_selector from gso.workflows.utils import customer_selector
from test.workflows import ( from test.workflows import (
...@@ -24,7 +25,8 @@ def input_form_wizard_data(router_subscription_factory, faker): ...@@ -24,7 +25,8 @@ def input_form_wizard_data(router_subscription_factory, faker):
router_side_b = router_subscription_factory() router_side_b = router_subscription_factory()
create_ip_trunk_step = { 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(), "geant_s_sid": faker.pystr(),
"iptrunk_type": IptrunkType.DARK_FIBER, "iptrunk_type": IptrunkType.DARK_FIBER,
"iptrunk_description": faker.sentence(), "iptrunk_description": faker.sentence(),
...@@ -63,7 +65,7 @@ def _user_accept_and_assert_suspended(process_stat, step_log, extra_data=None): ...@@ -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.provisioning_proxy.provision_ip_trunk")
@patch("gso.workflows.iptrunk.create_iptrunk.infoblox.allocate_v6_network") @patch("gso.workflows.iptrunk.create_iptrunk.infoblox.allocate_v6_network")
@patch("gso.workflows.iptrunk.create_iptrunk.infoblox.allocate_v4_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_result(
mock_allocate_v4_network, mock_allocate_v4_network,
mock_allocate_v6_network, mock_allocate_v6_network,
mock_provision_ip_trunk, mock_provision_ip_trunk,
...@@ -79,7 +81,7 @@ def test_successful_iptrunk_creation_with_standard_resume_data( ...@@ -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) result, process_stat, step_log = run_workflow("create_iptrunk", initial_site_data)
assert_suspended(result) assert_suspended(result)
standard_resume_data = { standard_lso_result = {
"pp_run_results": { "pp_run_results": {
"status": "ok", "status": "ok",
"job_id": "random_job_id", "job_id": "random_job_id",
...@@ -89,10 +91,10 @@ def test_successful_iptrunk_creation_with_standard_resume_data( ...@@ -89,10 +91,10 @@ def test_successful_iptrunk_creation_with_standard_resume_data(
"confirm": "ACCEPTED", "confirm": "ACCEPTED",
} }
for _ in range(5): 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_result)
result, step_log = _user_accept_and_assert_suspended(process_stat, step_log) 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_result)
result, step_log = resume_workflow(process_stat, step_log, {}) result, step_log = resume_workflow(process_stat, step_log, {})
assert_complete(result) assert_complete(result)
...@@ -129,7 +131,7 @@ def test_iptrunk_creation_fails_when_lso_return_code_is_one( ...@@ -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) result, process_stat, step_log = run_workflow("create_iptrunk", initial_site_data)
assert_suspended(result) assert_suspended(result)
standard_resume_data = { standard_lso_result = {
"pp_run_results": { "pp_run_results": {
"status": "ok", "status": "ok",
"job_id": "random_job_id", "job_id": "random_job_id",
...@@ -141,10 +143,10 @@ def test_iptrunk_creation_fails_when_lso_return_code_is_one( ...@@ -141,10 +143,10 @@ def test_iptrunk_creation_fails_when_lso_return_code_is_one(
attempts = 3 attempts = 3
for _ in range(0, attempts - 1): 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_result)
result, step_log = _user_accept_and_assert_suspended(process_stat, step_log) 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_result)
result, step_log = resume_workflow(process_stat, step_log, {}) result, step_log = resume_workflow(process_stat, step_log, {})
assert_aborted(result) assert_aborted(result)
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment