Add unit test for IP Trunk validation workflow

gso/translations/en-GB.json

@@ -43,9 +43,6 @@
         "create_router": "Create Router",
         "create_site": "Create Site",
         "deploy_twamp": "Deploy TWAMP",
-        "import_iptrunk": "Import IP Trunk",
-        "import_router": "Import Router",
-        "import_site": "Import Site",
         "migrate_iptrunk": "Migrate IP Trunk",
         "modify_isis_metric": "Modify the ISIS metric",
         "modify_site": "Modify Site",
@@ -65,6 +62,7 @@
         "import_iptrunk": "NOT FOR HUMANS -- Finalize import into an IP trunk product",
         "import_office_router": "NOT FOR HUMANS -- Finalize import into an Office router product",
         "import_super_pop_switch": "NOT FOR HUMANS -- Finalize import into a Super PoP switch",
+        "validate_iptrunk": "Validate IP Trunk configuration",
         "validate_router": "Validate router configuration"
     }
 }

gso/workflows/iptrunk/validate_iptrunk.py

@@ -39,11 +39,13 @@ def verify_ipam_loopback(subscription: Iptrunk) -> None:
     ipam_v6_network = infoblox.find_network_by_cidr(subscription.iptrunk.iptrunk_ipv6_network)
 
     if not ipam_v4_network or not ipam_v6_network:
-        ipam_errors += (
-            "Missing IP trunk IPAM records, found the following instead.\n"
-            f"IPv4 expected {subscription.iptrunk.iptrunk_ipv4_network}, actual: {ipam_v4_network}\n"
-            f"IPv6 expected {subscription.iptrunk.iptrunk_ipv6_network}, actual: {ipam_v6_network}"
-        )
+        ipam_errors += [
+            (
+                "Missing IP trunk IPAM records, found the following instead.\n"
+                f"IPv4 expected {subscription.iptrunk.iptrunk_ipv4_network}, actual: {ipam_v4_network}\n"
+                f"IPv6 expected {subscription.iptrunk.iptrunk_ipv6_network}, actual: {ipam_v6_network}"
+            )
+        ]
 
     #  Validate both sides of the trunk.
     for trunk_side in subscription.iptrunk.iptrunk_sides:
@@ -51,9 +53,9 @@ def verify_ipam_loopback(subscription: Iptrunk) -> None:
         #  Validate both IPv4 and IPv6 records.
         for record in [infoblox.find_host_by_fqdn(lag_fqdn), infoblox.find_v6_host_by_fqdn(lag_fqdn)]:
             if not record:
-                ipam_errors += f"Missing IPAM record for LAG interface {lag_fqdn}."
-            elif subscription.subscription_id not in record.comment:
-                ipam_errors += f"Found a misconfigured IPAM entry for {lag_fqdn}. Received: {record}"
+                ipam_errors += [f"Missing IPAM record for LAG interface {lag_fqdn}."]
+            elif str(subscription.subscription_id) not in record.comment:
+                ipam_errors += [f"Found a misconfigured IPAM entry for {lag_fqdn}. Received: {record}"]
 
     if ipam_errors:
         raise ProcessFailureError(str(ipam_errors))

test/workflows/iptrunk/test_validate_iptrunk.py

+from unittest.mock import patch
+
+import pytest
+from infoblox_client import objects
+
+from gso.products.product_types.iptrunk import Iptrunk
+from test.workflows import (
+    assert_complete,
+    assert_pp_interaction_success,
+    extract_state,
+    run_workflow,
+)
+
+
+@pytest.mark.workflow()
+@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")
+def test_validate_router_success(
+    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,
+):
+    #  Mock value setup
+    subscription_id = iptrunk_subscription_factory()
+    trunk = Iptrunk.from_subscription(subscription_id).iptrunk
+    mock_find_network_by_cidr.side_effects = [
+        objects.Network(connector=None, ipv4addrs=[trunk.iptrunk_ipv4_network]),
+        objects.NetworkV6(connector=None, ipv6addrs=[trunk.iptrunk_ipv6_network]),
+    ]
+    mock_find_host_by_fqdn.side_effect = [
+        objects.HostRecord(
+            connector=None,
+            aliases=[trunk.iptrunk_sides[0].iptrunk_side_node.router_fqdn],
+            comment=subscription_id,
+            ipv4addrs=[
+                objects.IPv4(
+                    ipv4addr=str(trunk.iptrunk_ipv4_network[0]),
+                    configure_for_dhcp=False,
+                    mac="00:00:00:00:00:00",
+                    ip=str(trunk.iptrunk_ipv4_network[0]),
+                    host=f"{trunk.iptrunk_sides[0].iptrunk_side_ae_iface}.{trunk.iptrunk_sides[0].iptrunk_side_node.router_fqdn}",
+                ),
+            ],
+            name=trunk.iptrunk_sides[0].iptrunk_side_node.router_fqdn,
+        ),
+        objects.HostRecord(
+            connector=None,
+            aliases=[trunk.iptrunk_sides[1].iptrunk_side_node.router_fqdn],
+            comment=subscription_id,
+            ipv4addrs=[
+                objects.IPv4(
+                    ipv4addr=str(trunk.iptrunk_ipv4_network[1]),
+                    configure_for_dhcp=False,
+                    mac="00:00:00:00:00:00",
+                    ip=str(trunk.iptrunk_ipv4_network[1]),
+                    host=f"{trunk.iptrunk_sides[1].iptrunk_side_ae_iface}.{trunk.iptrunk_sides[1].iptrunk_side_node.router_fqdn}",
+                ),
+            ],
+            name=trunk.iptrunk_sides[1].iptrunk_side_node.router_fqdn,
+        ),
+    ]
+
+    mock_find_v6_host_by_fqdn.side_effect = [
+        objects.HostRecordV6(
+            connector=None,
+            aliases=[trunk.iptrunk_sides[0].iptrunk_side_node.router_fqdn],
+            comment=subscription_id,
+            ipv4addrs=[
+                objects.IPv6(
+                    ipv6addr=str(trunk.iptrunk_ipv6_network[0]),
+                    configure_for_dhcp=False,
+                    mac="00:00:00:00:00:00",
+                    ip=str(trunk.iptrunk_ipv6_network[0]),
+                    host=f"{trunk.iptrunk_sides[0].iptrunk_side_ae_iface}.{trunk.iptrunk_sides[0].iptrunk_side_node.router_fqdn}",
+                ),
+            ],
+            name=trunk.iptrunk_sides[0].iptrunk_side_node.router_fqdn,
+        ),
+        objects.HostRecordV6(
+            connector=None,
+            aliases=[trunk.iptrunk_sides[1].iptrunk_side_node.router_fqdn],
+            comment=subscription_id,
+            ipv4addrs=[
+                objects.IPv6(
+                    ipv4addr=str(trunk.iptrunk_ipv6_network[1]),
+                    configure_for_dhcp=False,
+                    mac="00:00:00:00:00:00",
+                    ip=str(trunk.iptrunk_ipv6_network[1]),
+                    host=f"{trunk.iptrunk_sides[1].iptrunk_side_ae_iface}.{trunk.iptrunk_sides[1].iptrunk_side_node.router_fqdn}",
+                ),
+            ],
+            name=trunk.iptrunk_sides[1].iptrunk_side_node.router_fqdn,
+        ),
+    ]
+
+    #  Run workflow
+    initial_router_data = [{"subscription_id": subscription_id}]
+    result, process_stat, step_log = run_workflow("validate_iptrunk", initial_router_data)
+
+    state = extract_state(result)
+    subscription_id = state["subscription_id"]
+
+    result, step_log = assert_pp_interaction_success(result, process_stat, step_log)
+
+    assert_complete(result)
+
+    subscription = Iptrunk.from_subscription(subscription_id)
+
+    assert subscription.status == "active"
+    assert mock_validate_iptrunk.call_count == 1
+    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