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 f041528c680db07517f94e05c7c6b7f767d0919a..dd4c10d9f3078d2b02b584d33540dcc208387960 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 fbb6886d66eb36971d046a41421b20bf62382b8a..c54d53c273f0f0a7eb949e0c8b9ffac0b8ca19d7 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 e69de29bb2d1d6434b8b29ae775ad8c2e48c5391..2e81ff21b1cc0bc9829bd5f28fb02223b6905beb 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