From 2fc4682a9d0cd4def42cdab60dd30c57e97433f0 Mon Sep 17 00:00:00 2001
From: Karel van Klink <karel.vanklink@geant.org>
Date: Thu, 11 Jan 2024 16:55:07 +0100
Subject: [PATCH] Add unit test for IP Trunk validation workflow
---
gso/translations/en-GB.json | 4 +-
gso/workflows/iptrunk/validate_iptrunk.py | 18 +--
.../iptrunk/test_validate_iptrunk.py | 119 ++++++++++++++++++
3 files changed, 130 insertions(+), 11 deletions(-)
create mode 100644 test/workflows/iptrunk/test_validate_iptrunk.py
diff --git a/gso/translations/en-GB.json b/gso/translations/en-GB.json
index c8046f41..3131efde 100644
--- a/gso/translations/en-GB.json
+++ b/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"
}
}
diff --git a/gso/workflows/iptrunk/validate_iptrunk.py b/gso/workflows/iptrunk/validate_iptrunk.py
index 8d4f7be2..47d30dc5 100644
--- a/gso/workflows/iptrunk/validate_iptrunk.py
+++ b/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))
diff --git a/test/workflows/iptrunk/test_validate_iptrunk.py b/test/workflows/iptrunk/test_validate_iptrunk.py
new file mode 100644
index 00000000..8e8990c1
--- /dev/null
+++ b/test/workflows/iptrunk/test_validate_iptrunk.py
@@ -0,0 +1,119 @@
+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
--
GitLab