Skip to content
Snippets Groups Projects
Verified Commit 20bb21e8 authored by Karel van Klink's avatar Karel van Klink :smiley_cat:
Browse files

Add unit test for LAN Switch interconnect validation workflow

parent 2ae3cef7
No related branches found
No related tags found
1 merge request!302Feature/update lan interconnect
This commit is part of merge request !302. Comments created here will be created in the context of that merge request.
"""Update LAN Switch Interconnect. """Update LAN Switch Interconnect.
Revision ID: e854e0c35e20 Revision ID: e854e0c35e20
Revises: 0e7e7d749617 Revises: 543afff041f9
Create Date: 2024-11-04 17:21:14.612740 Create Date: 2024-11-04 17:21:14.612740
""" """
...@@ -10,7 +10,7 @@ from alembic import op ...@@ -10,7 +10,7 @@ from alembic import op
# revision identifiers, used by Alembic. # revision identifiers, used by Alembic.
revision = 'e854e0c35e20' revision = 'e854e0c35e20'
down_revision = '0e7e7d749617' down_revision = '543afff041f9'
branch_labels = None branch_labels = None
depends_on = None depends_on = None
......
...@@ -27,7 +27,7 @@ def validate_ipam(subscription: LanSwitchInterconnect) -> None: ...@@ -27,7 +27,7 @@ def validate_ipam(subscription: LanSwitchInterconnect) -> None:
host_record = find_host_by_fqdn(fqdn) host_record = find_host_by_fqdn(fqdn)
if not host_record or str(subscription.subscription_id) not in host_record.comment: 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!" 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( lan_interconnect_network = generate_lan_switch_interconnect_subnet(
subscription.lan_switch_interconnect.router_side.node.router_site.site_internal_id subscription.lan_switch_interconnect.router_side.node.router_site.site_internal_id
...@@ -35,7 +35,7 @@ def validate_ipam(subscription: LanSwitchInterconnect) -> None: ...@@ -35,7 +35,7 @@ def validate_ipam(subscription: LanSwitchInterconnect) -> None:
network_record = find_network_by_cidr(lan_interconnect_network) network_record = find_network_by_cidr(lan_interconnect_network)
if not network_record or str(subscription.subscription_id) not in network_record.comment: 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!" 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") @step("Check config for drift")
......
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
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment