From 20bb21e8563e3a858284e81b535d7dffc55ec170 Mon Sep 17 00:00:00 2001
From: Karel van Klink <karel.vanklink@geant.org>
Date: Tue, 5 Nov 2024 09:32:29 +0100
Subject: [PATCH] Add unit test for LAN Switch interconnect validation workflow
---
...e0c35e20_update_lan_switch_interconnect.py | 4 +-
.../validate_lan_switch_interconnect.py | 4 +-
.../test_validate_lan_switch_interconnect.py | 48 +++++++++++++++++++
3 files changed, 52 insertions(+), 4 deletions(-)
diff --git a/gso/migrations/versions/2024-11-04_e854e0c35e20_update_lan_switch_interconnect.py b/gso/migrations/versions/2024-11-04_e854e0c35e20_update_lan_switch_interconnect.py
index f041528c..dd4c10d9 100644
--- a/gso/migrations/versions/2024-11-04_e854e0c35e20_update_lan_switch_interconnect.py
+++ b/gso/migrations/versions/2024-11-04_e854e0c35e20_update_lan_switch_interconnect.py
@@ -1,7 +1,7 @@
"""Update LAN Switch Interconnect.
Revision ID: e854e0c35e20
-Revises: 0e7e7d749617
+Revises: 543afff041f9
Create Date: 2024-11-04 17:21:14.612740
"""
@@ -10,7 +10,7 @@ from alembic import op
# revision identifiers, used by Alembic.
revision = 'e854e0c35e20'
-down_revision = '0e7e7d749617'
+down_revision = '543afff041f9'
branch_labels = None
depends_on = None
diff --git a/gso/workflows/lan_switch_interconnect/validate_lan_switch_interconnect.py b/gso/workflows/lan_switch_interconnect/validate_lan_switch_interconnect.py
index fbb6886d..c54d53c2 100644
--- a/gso/workflows/lan_switch_interconnect/validate_lan_switch_interconnect.py
+++ b/gso/workflows/lan_switch_interconnect/validate_lan_switch_interconnect.py
@@ -27,7 +27,7 @@ def validate_ipam(subscription: LanSwitchInterconnect) -> None:
host_record = find_host_by_fqdn(fqdn)
if not host_record or str(subscription.subscription_id) not in host_record.comment:
msg = "DNS record is incorrectly configured in IPAM, please investigate this manually!"
- raise ProcessFailureError(msg)
+ raise ProcessFailureError(msg, details=host_record)
lan_interconnect_network = generate_lan_switch_interconnect_subnet(
subscription.lan_switch_interconnect.router_side.node.router_site.site_internal_id
@@ -35,7 +35,7 @@ def validate_ipam(subscription: LanSwitchInterconnect) -> None:
network_record = find_network_by_cidr(lan_interconnect_network)
if not network_record or str(subscription.subscription_id) not in network_record.comment:
msg = "LAN Switch Interconnect network is incorrectly configured in IPAM, please investigate this manually!"
- raise ProcessFailureError(msg)
+ raise ProcessFailureError(msg, details=network_record)
@step("Check config for drift")
diff --git a/test/workflows/lan_switch_interconnect/test_validate_lan_switch_interconnect.py b/test/workflows/lan_switch_interconnect/test_validate_lan_switch_interconnect.py
index e69de29b..2e81ff21 100644
--- a/test/workflows/lan_switch_interconnect/test_validate_lan_switch_interconnect.py
+++ b/test/workflows/lan_switch_interconnect/test_validate_lan_switch_interconnect.py
@@ -0,0 +1,48 @@
+from unittest.mock import patch
+
+import pytest
+from infoblox_client import objects
+
+from gso.products.product_types.lan_switch_interconnect import LanSwitchInterconnect
+from test.workflows import assert_complete, assert_lso_success, extract_state, run_workflow
+
+
+@pytest.mark.workflow()
+@patch("gso.services.infoblox.find_host_by_fqdn")
+@patch("gso.services.infoblox.find_network_by_cidr")
+@patch("gso.services.lso_client._send_request")
+def test_validate_lan_switch_interconnect(
+ mock_lso_interaction, mock_find_network, mock_find_host, lan_switch_interconnect_subscription_factory, faker
+):
+ subscription_id = lan_switch_interconnect_subscription_factory()
+ mocked_netbox_reply = objects.HostRecord(
+ connector=None,
+ aliases=[],
+ comment=subscription_id,
+ ipv4addrs=[
+ objects.IPv4(
+ ipv4addr=str(faker.ipv4()),
+ configure_for_dhcp=False,
+ mac="00:00:00:00:00:00",
+ ip=str(faker.ipv4()),
+ host=faker.domain_name(levels=4),
+ ),
+ ],
+ name=faker.domain_name(levels=4),
+ )
+ mock_find_host.return_value = mocked_netbox_reply
+ mock_find_network.return_value = mocked_netbox_reply
+ initial_lan_switch_interconnect_data = [{"subscription_id": subscription_id}]
+ result, process_stat, step_log = run_workflow(
+ "validate_lan_switch_interconnect", initial_lan_switch_interconnect_data
+ )
+ result, _ = assert_lso_success(result, process_stat, step_log)
+ assert_complete(result)
+
+ state = extract_state(result)
+ subscription_id = state["subscription_id"]
+ subscription = LanSwitchInterconnect.from_subscription(subscription_id)
+ assert subscription.status == "active"
+ assert subscription.insync is True
+ assert mock_find_host.call_count == 2
+ assert mock_find_network.call_count == 1
--
GitLab