add unit test for iBGP mesh workflow

gso/services/librenms_client.py

@@ -75,6 +75,7 @@ class LibreNMSClient:
         device_data.update(getattr(self.snmp_config, snmp_version))
 
         device = requests.post(f"{self.base_url}/devices", headers=self.headers, json=device_data, timeout=(0.5, 75))
+        device.raise_for_status()
 
         return device.json()
 
@@ -86,8 +87,8 @@ class LibreNMSClient:
         :raises HTTPError: Raises an exception if the request did not succeed.
         """
         device = requests.delete(f"{self.base_url}/devices/{fqdn}", headers=self.headers, timeout=(0.5, 75))
-
         device.raise_for_status()
+
         return device.json()
 
     def validate_device(self, fqdn: str) -> list[str]:

gso/workflows/router/update_ibgp_mesh.py

@@ -30,7 +30,7 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
         class Config:
             title = f"Add {subscription.router.router_fqdn} to the iBGP mesh?"
 
-        @root_validator
+        @root_validator(allow_reuse=True)
         def router_has_a_trunk(cls, values: dict[str, Any]) -> dict[str, Any]:
             if len(get_active_trunks_that_terminate_on_router(subscription_id)) == 0:
                 msg = "Selected router does not terminate any active IP trunks."
@@ -196,9 +196,9 @@ def check_ibgp_session(subscription: Router, callback_route: str) -> State:
 def add_device_to_librenms(subscription: Router) -> State:
     """Add the router as a device to LibreNMS."""
     client = librenms_client.LibreNMSClient()
-    client.add_device(subscription.router.router_fqdn, SNMPVersion.V2C)
+    librenms_result = client.add_device(subscription.router.router_fqdn, SNMPVersion.V2C)
 
-    return {"subscription": subscription}
+    return {"librenms_device": librenms_result}
 
 
 @step("Update subscription model")

test/workflows/router/test_update_ibgp_mesh.py

+from unittest.mock import patch
+
+import pytest
+from orchestrator.workflow import StepStatus
+from pydantic_forms.exceptions import FormValidationError
+
+from gso.products import Iptrunk
+from gso.products.product_blocks.router import RouterRole
+from test.workflows import assert_pp_interaction_success, extract_state, run_workflow
+
+
+@pytest.fixture()
+def ibgp_mesh_input_form_data(iptrunk_subscription_factory, faker):
+    ip_trunk = Iptrunk.from_subscription(iptrunk_subscription_factory())
+
+    return {"subscription_id": ip_trunk.iptrunk.iptrunk_sides[0].iptrunk_side_node.owner_subscription_id}
+
+
+@pytest.mark.workflow()
+@patch("gso.workflows.router.update_ibgp_mesh.provisioning_proxy.execute_playbook")
+@patch("gso.workflows.router.update_ibgp_mesh.librenms_client.LibreNMSClient.add_device")
+def test_update_ibgp_mesh_success(
+    mock_librenms_add_device,
+    mock_execute_playbook,
+    ibgp_mesh_input_form_data,
+    data_config_filename,
+):
+    result, process_stat, step_log = run_workflow("update_ibgp_mesh", [ibgp_mesh_input_form_data, {}])
+
+    for _ in range(5):
+        result, step_log = assert_pp_interaction_success(result, process_stat, step_log)
+
+    state = extract_state(result)
+
+    assert mock_execute_playbook.call_count == 5
+    assert mock_librenms_add_device.call_count == 1
+    assert result.status == StepStatus.COMPLETE
+    assert state["subscription"]["router"]["router_access_via_ts"] is False
+
+
+@pytest.mark.workflow()
+def test_update_ibgp_mesh_isolated_router(nokia_router_subscription_factory, data_config_filename):
+    router_id = nokia_router_subscription_factory(router_role=RouterRole.P)
+
+    exception_message = "Selected router does not terminate any active IP trunks."
+    with pytest.raises(FormValidationError, match=exception_message):
+        run_workflow("update_ibgp_mesh", [{"subscription_id": router_id}, {}])