From be7d176dda8117443ad3e7e4b918b54774fe683c Mon Sep 17 00:00:00 2001
From: Karel van Klink <karel.vanklink@geant.org>
Date: Thu, 11 Jan 2024 14:28:09 +0100
Subject: [PATCH] add unit test for router validation workflow
---
gso/workflows/router/validate_router.py | 15 ++++-
test/fixtures.py | 1 +
test/workflows/router/test_validate_router.py | 63 +++++++++++++++++++
3 files changed, 78 insertions(+), 1 deletion(-)
create mode 100644 test/workflows/router/test_validate_router.py
diff --git a/gso/workflows/router/validate_router.py b/gso/workflows/router/validate_router.py
index e89cabc9..f63c9e8a 100644
--- a/gso/workflows/router/validate_router.py
+++ b/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,
diff --git a/test/fixtures.py b/test/fixtures.py
index 4a996e7d..14ce92ae 100644
--- a/test/fixtures.py
+++ b/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
diff --git a/test/workflows/router/test_validate_router.py b/test/workflows/router/test_validate_router.py
new file mode 100644
index 00000000..596176c2
--- /dev/null
+++ b/test/workflows/router/test_validate_router.py
@@ -0,0 +1,63 @@
+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
--
GitLab