diff --git a/test/workflows/iptrunk/test_modify_trunk_interface.py b/test/workflows/iptrunk/test_modify_trunk_interface.py index 074219de3e11f31825ecb552c2094b6f840b7823..3424a5cc02b273cc6695c4b28b9209af07526c14 100644 --- a/test/workflows/iptrunk/test_modify_trunk_interface.py +++ b/test/workflows/iptrunk/test_modify_trunk_interface.py @@ -4,6 +4,7 @@ import pytest from gso.products import Iptrunk from gso.products.product_blocks.iptrunk import IptrunkType, PhyPortCapacity +from test.conftest import UseJuniperSide from test.workflows import ( assert_complete, assert_pp_interaction_success, @@ -14,9 +15,41 @@ from test.workflows.iptrunk.test_create_iptrunk import MockedNetboxClient @pytest.fixture() -def input_form_iptrunk_data(faker, iptrunk_subscription_factory): - - product_id = iptrunk_subscription_factory() +def input_form_iptrunk_data( + request, + faker, + iptrunk_subscription_factory, + juniper_router_subscription_factory, + nokia_router_subscription_factory, + iptrunk_side_subscription_factory, + ): + use_juniper= getattr(request, "param", UseJuniperSide.NONE) + if use_juniper == UseJuniperSide.SIDE_A: + side_node = juniper_router_subscription_factory() + side_a_node = iptrunk_side_subscription_factory(iptrunk_side_node=side_node) + side_b_node = iptrunk_side_subscription_factory() + new_side_a_ae_members = faker.generate_junniper_members_list() + new_side_b_ae_members = faker.generate_nokia_members_list() + elif use_juniper == UseJuniperSide.SIDE_B: + side_node = juniper_router_subscription_factory() + side_a_node = iptrunk_side_subscription_factory() + side_b_node = iptrunk_side_subscription_factory(iptrunk_side_node=side_node) + new_side_a_ae_members = faker.generate_nokia_members_list() + new_side_b_ae_members = faker.generate_junniper_members_list() + elif use_juniper == UseJuniperSide.SIDE_BOTH: + side_node_1 = juniper_router_subscription_factory() + side_node_2 = juniper_router_subscription_factory() + side_a_node = iptrunk_side_subscription_factory(iptrunk_side_node=side_node_1) + side_b_node = iptrunk_side_subscription_factory(iptrunk_side_node=side_node_2) + new_side_a_ae_members = faker.generate_junniper_members_list() + new_side_b_ae_members = faker.generate_junniper_members_list() + else: + side_a_node = iptrunk_side_subscription_factory() + side_b_node = iptrunk_side_subscription_factory() + new_side_a_ae_members = faker.generate_nokia_members_list() + new_side_b_ae_members = faker.generate_nokia_members_list() + + product_id = iptrunk_subscription_factory(iptrunk_sides=[side_a_node, side_b_node]) new_sid = faker.geant_sid() new_description = faker.sentence() @@ -25,14 +58,8 @@ def input_form_iptrunk_data(faker, iptrunk_subscription_factory): new_link_count = 2 new_side_a_sid = faker.geant_sid() - new_side_a_ae_members = [ - {"interface_name": f"Interface{i}", "interface_description": faker.sentence()} for i in range(5) - ] new_side_b_sid = faker.geant_sid() - new_side_b_ae_members = [ - {"interface_name": f"Interface{i}", "interface_description": faker.sentence()} for i in range(5) - ] return [ {"subscription_id": product_id}, @@ -55,6 +82,10 @@ def input_form_iptrunk_data(faker, iptrunk_subscription_factory): ] +@pytest.mark.parametrize("input_form_iptrunk_data", [UseJuniperSide.NONE, + UseJuniperSide.SIDE_A, + UseJuniperSide.SIDE_B, + UseJuniperSide.SIDE_BOTH], indirect=True) @pytest.mark.workflow() @patch("gso.workflows.iptrunk.modify_trunk_interface.provisioning_proxy.provision_ip_trunk") @patch("gso.services.netbox_client.NetboxClient.get_available_interfaces") @@ -99,17 +130,18 @@ def test_iptrunk_modify_trunk_interface_success( assert subscription.status == "active" assert mock_provision_ip_trunk.call_count == 2 # Assert all Netbox calls have been made - assert mocked_reserve_interface.call_count == 10 # 5 interfaces per side - assert mocked_attach_interface_to_lag.call_count == 10 # 5 interfaces per side - assert mocked_free_interface.call_count == 4 # 2 interfaces per side(The old ones) - assert mocked_detach_interfaces_from_lag.call_count == 2 # 1 time per side - - # Assert all subscription properties have been updated correctly new_sid = input_form_iptrunk_data[1]["geant_s_sid"] new_side_a_sid = input_form_iptrunk_data[2]["side_a_ae_geant_a_sid"] new_side_a_ae_members = input_form_iptrunk_data[2]["side_a_ae_members"] new_side_b_sid = input_form_iptrunk_data[3]["side_b_ae_geant_a_sid"] new_side_b_ae_members = input_form_iptrunk_data[3]["side_b_ae_members"] + num_ifaces = len(new_side_a_sid ) + len(new_side_b_sid ) # The length are randomly generated + assert mocked_reserve_interface.call_count == num_ifaces + assert mocked_attach_interface_to_lag.call_count == 10 # 5 interfaces per side + assert mocked_free_interface.call_count == 4 # 2 interfaces per side(The old ones) + assert mocked_detach_interfaces_from_lag.call_count == 2 # 1 time per side + + # Assert all subscription properties have been updated correctly assert subscription.description == f"IP trunk, geant_s_sid:{new_sid}" assert subscription.iptrunk.geant_s_sid == input_form_iptrunk_data[1]["geant_s_sid"] assert subscription.iptrunk.iptrunk_description == input_form_iptrunk_data[1]["iptrunk_description"] @@ -120,7 +152,7 @@ def test_iptrunk_modify_trunk_interface_success( def _find_interface_by_name(interfaces: list[dict[str, str]], name: str): for interface in interfaces: - if interface["interface_name"] == name: + if interface.interface_name == name: return interface msg = f"Interface {name} not found!" raise IndexError(msg)