add unit test for router validation workflow

gso/workflows/router/validate_router.py

@@ -3,14 +3,15 @@
 import json
 
 from orchestrator.targets import Target
+from orchestrator.utils.errors import ProcessFailureError
 from orchestrator.utils.json import json_dumps
 from orchestrator.workflow import StepList, done, init, step, workflow
 from orchestrator.workflows.steps import resync, store_process_subscription, unsync
 from orchestrator.workflows.utils import wrap_modify_initial_input_form
 
 from gso.products.product_types.router import Router
+from gso.services import infoblox
 from gso.services.provisioning_proxy import execute_playbook, pp_interaction
-from gso.workflows.router.create_router import verify_ipam_loopback
 
 
 @step("Validate router configuration")
@@ -26,6 +27,18 @@ def validate_router_config(subscription: Router, callback_route: str) -> None:
     )
 
 
+@step("Verify IPAM resources for loopback interface")
+def verify_ipam_loopback(subscription: Router) -> None:
+    """Validate the :term:`IPAM` resources for the loopback interface.
+
+    Raises an :class:`orchestrator.utils.errors.ProcessFailureError` if IPAM is configured incorrectly.
+    """
+    host_record = infoblox.find_host_by_fqdn(f"lo0.{subscription.router.router_fqdn}")
+    if not host_record or str(subscription.subscription_id) not in host_record.comment:
+        msg = "Loopback record is incorrectly configured in IPAM, please investigate this manually!"
+        raise ProcessFailureError(msg)
+
+
 @workflow(
     "Validate router configuration",
     target=Target.SYSTEM,

test/fixtures.py

@@ -154,6 +154,7 @@ def nokia_router_subscription_factory(site_subscription_factory, faker, geant_pa
         router_subscription.router.vendor = Vendor.NOKIA
 
         router_subscription = SubscriptionModel.from_other_lifecycle(router_subscription, SubscriptionLifecycle.ACTIVE)
+        router_subscription.insync = True
         router_subscription.description = description
         router_subscription.start_date = start_date
 

test/workflows/router/test_validate_router.py

+from unittest.mock import patch
+
+import pytest
+from infoblox_client import objects
+
+from gso.products.product_types.router import Router
+from test.workflows import (
+    assert_complete,
+    assert_pp_interaction_success,
+    extract_state,
+    run_workflow,
+)
+
+
+@pytest.mark.workflow()
+@pytest.mark.parametrize("product_id", ["nokia_router_subscription_factory", "juniper_router_subscription_factory"])
+@patch("gso.services.infoblox.find_host_by_fqdn")
+@patch("gso.workflows.router.validate_router.execute_playbook")
+def test_validate_router_success(
+    mock_validate_router,
+    mock_find_host_by_fqdn,
+    product_id,
+    faker,
+    data_config_filename,
+    request,
+):
+    #  Run workflow
+    subscription_id = request.getfixturevalue(product_id)()
+    mock_fqdn = Router.from_subscription(subscription_id).router.router_fqdn
+    mock_v4 = faker.ipv4()
+    mock_find_host_by_fqdn.return_value = objects.HostRecord(
+        connector=None,
+        aliases=[mock_fqdn],
+        comment=subscription_id,
+        ipv4addrs=[
+            objects.IPv4(
+                ipv4addr=str(mock_v4),
+                configure_for_dhcp=False,
+                mac="00:00:00:00:00:00",
+                ip=str(mock_v4),
+                host=f"lo0.{mock_fqdn}",
+            ),
+        ],
+        name=mock_fqdn,
+    )
+
+    initial_router_data = [{"subscription_id": subscription_id}]
+    result, process_stat, step_log = run_workflow("validate_router", 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)
+
+    state = extract_state(result)
+    subscription = Router.from_subscription(subscription_id)
+
+    assert subscription.status == "active"
+    assert mock_validate_router.call_count == 1
+    assert mock_find_host_by_fqdn.call_count == 1
+    assert "ipam_warning" not in state