diff --git a/test/workflows/iptrunk/test_modify_trunk_interface.py b/test/workflows/iptrunk/test_modify_trunk_interface.py index 8601b8ad675bc0c287b98ad277f58bf3cc52c81e..074219de3e11f31825ecb552c2094b6f840b7823 100644 --- a/test/workflows/iptrunk/test_modify_trunk_interface.py +++ b/test/workflows/iptrunk/test_modify_trunk_interface.py @@ -13,36 +13,11 @@ from test.workflows import ( from test.workflows.iptrunk.test_create_iptrunk import MockedNetboxClient -@pytest.mark.workflow() -@patch("gso.workflows.iptrunk.modify_trunk_interface.provisioning_proxy.provision_ip_trunk") -@patch("gso.services.netbox_client.NetboxClient.get_available_interfaces") -@patch("gso.services.netbox_client.NetboxClient.attach_interface_to_lag") -@patch("gso.services.netbox_client.NetboxClient.reserve_interface") -@patch("gso.services.netbox_client.NetboxClient.allocate_interface") -@patch("gso.services.netbox_client.NetboxClient.free_interface") -@patch("gso.services.netbox_client.NetboxClient.detach_interfaces_from_lag") -def test_iptrunk_modify_trunk_interface_success( - mocked_detach_interfaces_from_lag, - mocked_free_interface, - mocked_allocate_interface, - mocked_reserve_interface, - mocked_attach_interface_to_lag, - mocked_get_available_interfaces, - mock_provision_ip_trunk, - iptrunk_subscription_factory, - faker, - data_config_filename, -): - # Set up mock return values - mocked_netbox = MockedNetboxClient() - mocked_get_available_interfaces.return_value = mocked_netbox.get_available_interfaces() - mocked_attach_interface_to_lag.return_value = mocked_netbox.attach_interface_to_lag() - mocked_reserve_interface.return_value = mocked_netbox.reserve_interface() - mocked_allocate_interface.return_value = mocked_netbox.allocate_interface() - mocked_free_interface.return_value = mocked_netbox.free_interface() - mocked_detach_interfaces_from_lag.return_value = mocked_netbox.detach_interfaces_from_lag() +@pytest.fixture() +def input_form_iptrunk_data(faker, iptrunk_subscription_factory): product_id = iptrunk_subscription_factory() + new_sid = faker.geant_sid() new_description = faker.sentence() new_type = IptrunkType.LEASED @@ -59,8 +34,7 @@ def test_iptrunk_modify_trunk_interface_success( {"interface_name": f"Interface{i}", "interface_description": faker.sentence()} for i in range(5) ] - # Run workflow - initial_iptrunk_data = [ + return [ {"subscription_id": product_id}, { "tt_number": faker.tt_number(), @@ -80,7 +54,38 @@ def test_iptrunk_modify_trunk_interface_success( }, ] - result, process_stat, step_log = run_workflow("modify_trunk_interface", initial_iptrunk_data) + +@pytest.mark.workflow() +@patch("gso.workflows.iptrunk.modify_trunk_interface.provisioning_proxy.provision_ip_trunk") +@patch("gso.services.netbox_client.NetboxClient.get_available_interfaces") +@patch("gso.services.netbox_client.NetboxClient.attach_interface_to_lag") +@patch("gso.services.netbox_client.NetboxClient.reserve_interface") +@patch("gso.services.netbox_client.NetboxClient.allocate_interface") +@patch("gso.services.netbox_client.NetboxClient.free_interface") +@patch("gso.services.netbox_client.NetboxClient.detach_interfaces_from_lag") +def test_iptrunk_modify_trunk_interface_success( + mocked_detach_interfaces_from_lag, + mocked_free_interface, + mocked_allocate_interface, + mocked_reserve_interface, + mocked_attach_interface_to_lag, + mocked_get_available_interfaces, + mock_provision_ip_trunk, + input_form_iptrunk_data, + faker, + data_config_filename, +): + # Set up mock return values + mocked_netbox = MockedNetboxClient() + mocked_get_available_interfaces.return_value = mocked_netbox.get_available_interfaces() + mocked_attach_interface_to_lag.return_value = mocked_netbox.attach_interface_to_lag() + mocked_reserve_interface.return_value = mocked_netbox.reserve_interface() + mocked_allocate_interface.return_value = mocked_netbox.allocate_interface() + mocked_free_interface.return_value = mocked_netbox.free_interface() + mocked_detach_interfaces_from_lag.return_value = mocked_netbox.detach_interfaces_from_lag() + + # Run workflow + result, process_stat, step_log = run_workflow("modify_trunk_interface", input_form_iptrunk_data) for _ in range(2): result, step_log = assert_pp_interaction_success(result, process_stat, step_log) @@ -100,12 +105,17 @@ def test_iptrunk_modify_trunk_interface_success( 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"] assert subscription.description == f"IP trunk, geant_s_sid:{new_sid}" - assert subscription.iptrunk.geant_s_sid == new_sid - assert subscription.iptrunk.iptrunk_description == new_description - assert subscription.iptrunk.iptrunk_type == new_type - assert subscription.iptrunk.iptrunk_speed == new_speed - assert subscription.iptrunk.iptrunk_minimum_links == new_link_count + 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"] + assert subscription.iptrunk.iptrunk_type == input_form_iptrunk_data[1]["iptrunk_type"] + assert subscription.iptrunk.iptrunk_speed == input_form_iptrunk_data[1]["iptrunk_speed"] + assert subscription.iptrunk.iptrunk_minimum_links == input_form_iptrunk_data[1]["iptrunk_minimum_links"] assert subscription.iptrunk.iptrunk_sides[0].iptrunk_side_ae_geant_a_sid == new_side_a_sid def _find_interface_by_name(interfaces: list[dict[str, str]], name: str): @@ -126,5 +136,5 @@ def test_iptrunk_modify_trunk_interface_success( for member in subscription.iptrunk.iptrunk_sides[1].iptrunk_side_ae_members: assert ( member.interface_description - == _find_interface_by_name(new_side_b_ae_members, member.interface_name)["interface_description"] + == _find_interface_by_name(new_side_b_ae_members , member.interface_name)["interface_description"] )