diff --git a/gso/workflows/iptrunk/validate_iptrunk.py b/gso/workflows/iptrunk/validate_iptrunk.py index 5b2e3f50cf4615962096af32f4838579f418a4cc..4ffe3deb8a00a6d9efa8e1df944a2e7cc497fb76 100644 --- a/gso/workflows/iptrunk/validate_iptrunk.py +++ b/gso/workflows/iptrunk/validate_iptrunk.py @@ -5,7 +5,7 @@ import json from orchestrator.targets import Target from orchestrator.utils.errors import ProcessFailureError from orchestrator.utils.json import json_dumps -from orchestrator.workflow import StepList, begin, done, step, workflow +from orchestrator.workflow import StepList, begin, conditional, done, step, workflow from orchestrator.workflows.steps import resync, store_process_subscription, unsync from orchestrator.workflows.utils import wrap_modify_initial_input_form @@ -192,10 +192,20 @@ def validate_iptrunk() -> StepList: * Verify the configuration on both sides of the trunk is intact. * Check the ISIS metric of the trunk. * Verify that TWAMP configuration is correct. + + If a trunk has a Juniper router on both sides, it is considered legacy and does not require validation. """ + is_legacy_trunk = conditional( + lambda state: all( + side.iptrunk_side_node.vendor == Vendor.JUNIPER + for side in Iptrunk.from_subscription(state["subscription_id"]).iptrunk.iptrunk_sides + ) + ) + return ( begin >> store_process_subscription(Target.SYSTEM) + >> is_legacy_trunk(done) >> unsync >> verify_ipam_records >> verify_netbox_entries diff --git a/test/workflows/iptrunk/test_validate_iptrunk.py b/test/workflows/iptrunk/test_validate_iptrunk.py index 48de3d1bc069f879ae307273544abbeef57ebc5b..13bcaed9b4be4284a6f1dc5fa4e41410d2f8dab6 100644 --- a/test/workflows/iptrunk/test_validate_iptrunk.py +++ b/test/workflows/iptrunk/test_validate_iptrunk.py @@ -4,6 +4,7 @@ import pytest from infoblox_client import objects from gso.products.product_types.iptrunk import Iptrunk +from test.conftest import UseJuniperSide from test.services.conftest import MockedNetboxClient from test.workflows import ( assert_complete, @@ -13,6 +14,31 @@ from test.workflows import ( ) +@pytest.fixture() +def trunk_validation_setup( + request, + iptrunk_subscription_factory, + juniper_router_subscription_factory, + nokia_router_subscription_factory, + iptrunk_side_subscription_factory, +): + if request.param == UseJuniperSide.SIDE_A: + # Nokia -> Juniper + side_a = iptrunk_side_subscription_factory(iptrunk_side_node=nokia_router_subscription_factory()) + side_b = iptrunk_side_subscription_factory(iptrunk_side_node=juniper_router_subscription_factory()) + elif request.param == UseJuniperSide.SIDE_B: + # Juniper -> Nokia + side_a = iptrunk_side_subscription_factory(iptrunk_side_node=juniper_router_subscription_factory()) + side_b = iptrunk_side_subscription_factory(iptrunk_side_node=nokia_router_subscription_factory()) + else: + # Nokia -> Nokia + side_a = iptrunk_side_subscription_factory(iptrunk_side_node=nokia_router_subscription_factory()) + side_b = iptrunk_side_subscription_factory(iptrunk_side_node=nokia_router_subscription_factory()) + + subscription_id = iptrunk_subscription_factory(iptrunk_sides=[side_a, side_b]) + return [{"subscription_id": subscription_id}] + + @pytest.fixture() def _mocked_netbox_client(): with ( @@ -35,6 +61,9 @@ def _mocked_netbox_client(): yield +@pytest.mark.parametrize( + "trunk_validation_setup", [UseJuniperSide.NONE, UseJuniperSide.SIDE_A, UseJuniperSide.SIDE_B], indirect=True +) @pytest.mark.workflow() @pytest.mark.usefixtures("_mocked_netbox_client") @patch("gso.services.infoblox.find_network_by_cidr") @@ -50,10 +79,10 @@ def test_validate_iptrunk_success( mock_find_network_by_cidr, faker, data_config_filename, - iptrunk_subscription_factory, + trunk_validation_setup, ): # Mock value setup - subscription_id = iptrunk_subscription_factory() + subscription_id = trunk_validation_setup[0]["subscription_id"] trunk = Iptrunk.from_subscription(subscription_id).iptrunk mock_find_network_by_cidr.side_effects = [ objects.Network(connector=None, ipv4addrs=[trunk.iptrunk_ipv4_network]), @@ -183,3 +212,44 @@ def test_validate_iptrunk_success( assert mock_find_host_by_fqdn.call_count == 2 assert mock_find_v6_host_by_fqdn.call_count == 2 assert mock_find_network_by_cidr.call_count == 2 + + +@pytest.mark.workflow() +@pytest.mark.usefixtures("_mocked_netbox_client") +@patch("gso.services.infoblox.find_network_by_cidr") +@patch("gso.services.infoblox.find_v6_host_by_fqdn") +@patch("gso.services.infoblox.find_host_by_fqdn") +@patch("gso.workflows.iptrunk.validate_iptrunk.execute_playbook") +@patch("gso.services.netbox_client.NetboxClient.get_interface_by_name_and_device") +def test_validate_iptrunk_skip_legacy_trunks( + mock_get_interface_by_name, + mock_validate_iptrunk, + mock_find_host_by_fqdn, + mock_find_v6_host_by_fqdn, + mock_find_network_by_cidr, + faker, + data_config_filename, + iptrunk_subscription_factory, + iptrunk_side_subscription_factory, + juniper_router_subscription_factory, +): + # Mock value setup + side_a = iptrunk_side_subscription_factory(iptrunk_side_node=juniper_router_subscription_factory()) + side_b = iptrunk_side_subscription_factory(iptrunk_side_node=juniper_router_subscription_factory()) + subscription_id = iptrunk_subscription_factory(iptrunk_sides=[side_a, side_b]) + + # Run workflow + initial_router_data = [{"subscription_id": subscription_id}] + result, _, _ = run_workflow("validate_iptrunk", initial_router_data) + state = extract_state(result) + assert_complete(result) + + subscription_id = state["subscription_id"] + subscription = Iptrunk.from_subscription(subscription_id) + + assert subscription.status == "active" + assert mock_get_interface_by_name.call_count == 0 + assert mock_validate_iptrunk.call_count == 0 + assert mock_find_host_by_fqdn.call_count == 0 + assert mock_find_v6_host_by_fqdn.call_count == 0 + assert mock_find_network_by_cidr.call_count == 0