Skip IP trunk validation for trunks that are Juniper on both sides

Also updated unit tests

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

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