diff --git a/gso/schedules/__init__.py b/gso/schedules/__init__.py
index 7d55fad176fffa0d16840f95c86233d5e3a516ef..8257a874c6c090c1d284966d2ff4f9064d04e831 100644
--- a/gso/schedules/__init__.py
+++ b/gso/schedules/__init__.py
@@ -1,3 +1 @@
"""Tasks that are scheduled to run periodically in :term:`GSO`."""
-
-TT_NUMBER_ZERO = "TT#0000000000000000"
diff --git a/gso/workflows/iptrunk/validate_iptrunk.py b/gso/workflows/iptrunk/validate_iptrunk.py
index 1b2c1bc14e531c068d2cb1d4d6306f939fcdd8c0..4f3327de425cc7c2055996cfc301632f495679df 100644
--- a/gso/workflows/iptrunk/validate_iptrunk.py
+++ b/gso/workflows/iptrunk/validate_iptrunk.py
@@ -15,7 +15,6 @@ from gso.services.lso_client import anonymous_lso_interaction, execute_playbook
from gso.services.netbox_client import NetboxClient
from gso.utils.helpers import get_router_vendor
from gso.utils.shared_enums import Vendor
-from gso.workflows.iptrunk.migrate_iptrunk import check_ip_trunk_isis
@step("Validate IP trunk configuration")
@@ -70,7 +69,7 @@ def verify_ipam_records(subscription: Iptrunk) -> None:
]
# Allocated host record needs to be set correctly
- if record.comment != subscription.subscription_id:
+ if record.comment != str(subscription.subscription_id):
ipam_errors += [
(
f"Incorrect host record found for '{lag_fqdn}' at '{side_v4}'. Comment should have been equal "
@@ -93,7 +92,7 @@ def verify_ipam_records(subscription: Iptrunk) -> None:
]
# Allocated host record needs to be set correctly
- if record.comment != subscription.subscription_id:
+ if record.comment != str(subscription.subscription_id):
ipam_errors += [
(
f"Incorrect host record found for '{lag_fqdn}' at '{side_v6}'. Comment should have been equal "
@@ -157,6 +156,20 @@ def verify_iptrunk_config(subscription: Iptrunk, callback_route: str) -> None:
)
+@step("Check ISIS adjacency")
+def check_ip_trunk_isis(subscription: Iptrunk, callback_route: str) -> None:
+ """Run an Ansible playbook to confirm :term:`ISIS` adjacency."""
+ extra_vars = {"wfo_ip_trunk_json": json.loads(json_dumps(subscription)), "check": "isis"}
+
+ execute_playbook(
+ playbook_name="iptrunks_checks.yaml",
+ callback_route=callback_route,
+ inventory=f"{subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.router_fqdn}\n"
+ f"{subscription.iptrunk.iptrunk_sides[1].iptrunk_side_node.router_fqdn}\n",
+ extra_vars=extra_vars,
+ )
+
+
@step("Verify TWAMP configuration")
def verify_twamp_config(subscription: Iptrunk, callback_route: str) -> None:
"""Check for configuration drift of TWAMP."""
@@ -172,9 +185,7 @@ def verify_twamp_config(subscription: Iptrunk, callback_route: str) -> None:
@workflow(
- "Validate IP trunk configuration",
- target=Target.SYSTEM,
- initial_input_form=wrap_modify_initial_input_form(None),
+ "Validate IP trunk configuration", target=Target.SYSTEM, initial_input_form=wrap_modify_initial_input_form(None)
)
def validate_iptrunk() -> StepList:
"""Validate an existing, active IP Trunk subscription.
diff --git a/gso/workflows/router/validate_router.py b/gso/workflows/router/validate_router.py
index 5ed8189798ad1f838ec053be0aeb2e2aef175fed..94d346283a2c2efe446cb390606066afef8bf37d 100644
--- a/gso/workflows/router/validate_router.py
+++ b/gso/workflows/router/validate_router.py
@@ -8,7 +8,6 @@ from orchestrator.workflows.utils import wrap_modify_initial_input_form
from pydantic_forms.types import State, UUIDstr
from gso.products.product_types.router import Router
-from gso.schedules import TT_NUMBER_ZERO
from gso.services import infoblox
from gso.services.librenms_client import LibreNMSClient
from gso.services.lso_client import anonymous_lso_interaction, execute_playbook
@@ -18,13 +17,10 @@ from gso.utils.shared_enums import Vendor
@step("Prepare required keys in state")
def prepare_state(subscription_id: UUIDstr) -> State:
- """Add required keys to the state for the workflow to run successfully.
-
- This includes the router product itself and a fake TT number for running playbooks.
- """
+ """Add required keys to the state for the workflow to run successfully."""
router = Router.from_subscription(subscription_id)
- return {"subscription": router, "tt_number": TT_NUMBER_ZERO}
+ return {"subscription": router}
@step("Verify IPAM resources for loopback interface")
diff --git a/test/workflows/iptrunk/test_validate_iptrunk.py b/test/workflows/iptrunk/test_validate_iptrunk.py
index cad4b52457fc192a2900d764d0924e40899083de..48de3d1bc069f879ae307273544abbeef57ebc5b 100644
--- a/test/workflows/iptrunk/test_validate_iptrunk.py
+++ b/test/workflows/iptrunk/test_validate_iptrunk.py
@@ -4,20 +4,46 @@ import pytest
from infoblox_client import objects
from gso.products.product_types.iptrunk import Iptrunk
+from test.services.conftest import MockedNetboxClient
from test.workflows import (
assert_complete,
- assert_lso_interaction_success,
+ assert_lso_success,
extract_state,
run_workflow,
)
+@pytest.fixture()
+def _mocked_netbox_client():
+ with (
+ patch("gso.services.netbox_client.NetboxClient.get_device_by_name") as mock_get_device_by_name,
+ patch("gso.services.netbox_client.NetboxClient.get_available_interfaces") as mock_get_available_interfaces,
+ patch("gso.services.netbox_client.NetboxClient.get_available_lags") as mock_get_available_lags,
+ patch("gso.services.netbox_client.NetboxClient.create_interface") as mock_create_interface,
+ patch("gso.services.netbox_client.NetboxClient.attach_interface_to_lag") as mock_attach_interface_to_lag,
+ patch("gso.services.netbox_client.NetboxClient.reserve_interface") as mock_reserve_interface,
+ patch("gso.services.netbox_client.NetboxClient.allocate_interface") as mock_allocate_interface,
+ ):
+ mock_get_device_by_name.return_value = MockedNetboxClient().get_device_by_name()
+ mock_get_available_interfaces.return_value = MockedNetboxClient().get_available_interfaces()
+ mock_get_available_lags.return_value = MockedNetboxClient().get_available_lags()
+ mock_create_interface.return_value = MockedNetboxClient().create_interface()
+ mock_attach_interface_to_lag.return_value = MockedNetboxClient().attach_interface_to_lag()
+ mock_reserve_interface.return_value = MockedNetboxClient().reserve_interface()
+ mock_allocate_interface.return_value = MockedNetboxClient().allocate_interface()
+
+ yield
+
+
@pytest.mark.workflow()
+@pytest.mark.usefixtures("_mocked_netbox_client")
@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(
+@patch("gso.services.netbox_client.NetboxClient.get_interface_by_name_and_device")
+def test_validate_iptrunk_success(
+ mock_get_interface_by_name,
mock_validate_iptrunk,
mock_find_host_by_fqdn,
mock_find_v6_host_by_fqdn,
@@ -71,12 +97,12 @@ def test_validate_router_success(
connector=None,
aliases=[trunk.iptrunk_sides[0].iptrunk_side_node.router_fqdn],
comment=subscription_id,
- ipv4addrs=[
+ ipv6addrs=[
objects.IPv6(
- ipv6addr=str(trunk.iptrunk_ipv6_network[0]),
+ ipv6addr=str(trunk.iptrunk_ipv6_network[1]),
configure_for_dhcp=False,
mac="00:00:00:00:00:00",
- ip=str(trunk.iptrunk_ipv6_network[0]),
+ ip=str(trunk.iptrunk_ipv6_network[1]),
host=f"{trunk.iptrunk_sides[0].iptrunk_side_ae_iface}.{trunk.iptrunk_sides[0].iptrunk_side_node.router_fqdn}",
),
],
@@ -86,12 +112,12 @@ def test_validate_router_success(
connector=None,
aliases=[trunk.iptrunk_sides[1].iptrunk_side_node.router_fqdn],
comment=subscription_id,
- ipv4addrs=[
+ ipv6addrs=[
objects.IPv6(
- ipv4addr=str(trunk.iptrunk_ipv6_network[1]),
+ ipv6addr=str(trunk.iptrunk_ipv6_network[2]),
configure_for_dhcp=False,
mac="00:00:00:00:00:00",
- ip=str(trunk.iptrunk_ipv6_network[1]),
+ ip=str(trunk.iptrunk_ipv6_network[2]),
host=f"{trunk.iptrunk_sides[1].iptrunk_side_ae_iface}.{trunk.iptrunk_sides[1].iptrunk_side_node.router_fqdn}",
),
],
@@ -99,21 +125,61 @@ def test_validate_router_success(
),
]
+ mock_get_interface_by_name.side_effect = [
+ MockedNetboxClient.BaseMockObject(
+ name="Boeiee",
+ module=MockedNetboxClient.BaseMockObject(display="display1"),
+ description=subscription_id,
+ enabled=True,
+ ),
+ MockedNetboxClient.BaseMockObject(
+ name="2222222",
+ module=MockedNetboxClient.BaseMockObject(display="display2"),
+ description=subscription_id,
+ enabled=True,
+ ),
+ MockedNetboxClient.BaseMockObject(
+ name="3333",
+ module=MockedNetboxClient.BaseMockObject(display="display 333"),
+ description=subscription_id,
+ enabled=True,
+ ),
+ MockedNetboxClient.BaseMockObject(
+ name="44",
+ module=MockedNetboxClient.BaseMockObject(display="display 444"),
+ description=subscription_id,
+ enabled=True,
+ ),
+ MockedNetboxClient.BaseMockObject(
+ name="55",
+ module=MockedNetboxClient.BaseMockObject(display="display25555"),
+ description=subscription_id,
+ enabled=True,
+ ),
+ MockedNetboxClient.BaseMockObject(
+ name="666",
+ module=MockedNetboxClient.BaseMockObject(display="display666"),
+ description=subscription_id,
+ enabled=True,
+ ),
+ ]
+
# 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_lso_interaction_success(result, process_stat, step_log)
+ for _ in range(3):
+ result, step_log = assert_lso_success(result, process_stat, step_log)
assert_complete(result)
+ subscription_id = state["subscription_id"]
subscription = Iptrunk.from_subscription(subscription_id)
assert subscription.status == "active"
- assert mock_validate_iptrunk.call_count == 1
+ assert mock_validate_iptrunk.call_count == 3
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