Make updating iBGP and SBP meshes optional in mesh update workflow

gso/workflows/router/update_ibgp_mesh.py

@@ -69,6 +69,8 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
         model_config = ConfigDict(title=f"Add {subscription.router.router_fqdn} to the iBGP mesh?")
 
         tt_number: TTNumber
+        update_ibgp_mesh: bool = True
+        update_sbp_mesh: bool = True
 
         @model_validator(mode="before")
         def router_has_a_trunk(cls, data: Any) -> Any:
@@ -232,28 +234,31 @@ def update_ibgp_mesh() -> StepList:
     router_is_pe = conditional(lambda state: state["subscription"]["router"]["router_role"] == RouterRole.PE)
     router_is_p = conditional(lambda state: state["subscription"]["router"]["router_role"] == RouterRole.P)
 
+    update_bgp = conditional(lambda state: bool(state["update_ibgp_mesh"]))
+    update_sbp = conditional(lambda state: bool(state["update_sbp_mesh"]))
+
     return (
         begin
         >> store_process_subscription(Target.MODIFY)
         >> unsync
-        >> router_is_p(lso_interaction(add_p_to_mesh_dry))
-        >> router_is_p(lso_interaction(add_p_to_mesh_real))
-        >> router_is_p(lso_interaction(add_all_pe_to_p_dry))
-        >> router_is_p(lso_interaction(add_all_pe_to_p_real))
-        >> router_is_p(lso_interaction(check_ibgp_session))
-        >> router_is_pe(lso_interaction(add_pe_mesh_to_pe_dry))
-        >> router_is_pe(lso_interaction(add_pe_mesh_to_pe_real))
-        >> router_is_pe(lso_interaction(add_pe_to_pe_mesh_dry))
-        >> router_is_pe(lso_interaction(add_pe_to_pe_mesh_real))
-        >> router_is_pe(lso_interaction(add_all_p_to_pe_dry))
-        >> router_is_pe(lso_interaction(add_all_p_to_pe_real))
-        >> router_is_pe(lso_interaction(add_pe_to_all_p_dry))
-        >> router_is_pe(lso_interaction(add_pe_to_all_p_real))
-        >> router_is_pe(lso_interaction(update_sdp_single_pe_dry))
-        >> router_is_pe(lso_interaction(update_sdp_single_pe_real))
-        >> router_is_pe(lso_interaction(add_pe_to_sdp_mesh_dry))
-        >> router_is_pe(lso_interaction(add_pe_to_sdp_mesh_real))
-        >> router_is_pe(lso_interaction(check_pe_ibgp))
+        >> update_bgp(router_is_p(lso_interaction(add_p_to_mesh_dry)))
+        >> update_bgp(router_is_p(lso_interaction(add_p_to_mesh_real)))
+        >> update_bgp(router_is_p(lso_interaction(add_all_pe_to_p_dry)))
+        >> update_bgp(router_is_p(lso_interaction(add_all_pe_to_p_real)))
+        >> update_bgp(router_is_p(lso_interaction(check_ibgp_session)))
+        >> update_bgp(router_is_pe(lso_interaction(add_pe_mesh_to_pe_dry)))
+        >> update_bgp(router_is_pe(lso_interaction(add_pe_mesh_to_pe_real)))
+        >> update_bgp(router_is_pe(lso_interaction(add_pe_to_pe_mesh_dry)))
+        >> update_bgp(router_is_pe(lso_interaction(add_pe_to_pe_mesh_real)))
+        >> update_bgp(router_is_pe(lso_interaction(add_all_p_to_pe_dry)))
+        >> update_bgp(router_is_pe(lso_interaction(add_all_p_to_pe_real)))
+        >> update_bgp(router_is_pe(lso_interaction(add_pe_to_all_p_dry)))
+        >> update_bgp(router_is_pe(lso_interaction(add_pe_to_all_p_real)))
+        >> update_sbp(router_is_pe(lso_interaction(update_sdp_single_pe_dry)))
+        >> update_sbp(router_is_pe(lso_interaction(update_sdp_single_pe_real)))
+        >> update_sbp(router_is_pe(lso_interaction(add_pe_to_sdp_mesh_dry)))
+        >> update_sbp(router_is_pe(lso_interaction(add_pe_to_sdp_mesh_real)))
+        >> update_bgp(router_is_pe(lso_interaction(check_pe_ibgp)))
         >> router_is_pe(lso_interaction(check_l3_services))
         >> add_device_to_librenms
         >> prompt_insert_in_radius

test/workflows/router/test_update_ibgp_mesh.py

@@ -18,6 +18,8 @@ from test.workflows import (
 
 @pytest.mark.parametrize("trunk_status", [SubscriptionLifecycle.PROVISIONING, SubscriptionLifecycle.ACTIVE])
 @pytest.mark.parametrize("router_role", [RouterRole.P, RouterRole.PE])
+@pytest.mark.parametrize("update_ibgp_mesh", [True, False])
+@pytest.mark.parametrize("update_sbp_mesh", [True, False])
 @pytest.mark.workflow()
 @patch("gso.services.lso_client._send_request")
 @patch("gso.workflows.router.update_ibgp_mesh.librenms_client.LibreNMSClient.add_device")
@@ -26,6 +28,8 @@ def test_update_ibgp_mesh_success(
     mock_librenms_device_exists,
     mock_librenms_add_device,
     mock_execute_playbook,
+    update_sbp_mesh,
+    update_ibgp_mesh,
     router_role,
     trunk_status,
     iptrunk_subscription_factory,
@@ -41,12 +45,29 @@ def test_update_ibgp_mesh_success(
     router_subscription_factory()
 
     ip_trunk = iptrunk_subscription_factory(status=trunk_status, iptrunk_sides=[side_a, side_b])
-    callback_step_count = 5 if router_role == RouterRole.P else 14
+    callback_step_count = 0
+    if router_role == RouterRole.PE:
+        callback_step_count += 1
+        if update_ibgp_mesh:
+            callback_step_count += 9
+        if update_sbp_mesh:
+            callback_step_count += 4
+    elif router_role == RouterRole.P and update_ibgp_mesh:
+        callback_step_count += 5
+
     ibgp_mesh_input_form_data = {
         "subscription_id": ip_trunk.iptrunk.iptrunk_sides[0].iptrunk_side_node.owner_subscription_id
     }
     result, process_stat, step_log = run_workflow(
-        "update_ibgp_mesh", [ibgp_mesh_input_form_data, {"tt_number": faker.tt_number()}]
+        "update_ibgp_mesh",
+        [
+            ibgp_mesh_input_form_data,
+            {
+                "tt_number": faker.tt_number(),
+                "update_ibgp_mesh": update_ibgp_mesh,
+                "update_sbp_mesh": update_sbp_mesh,
+            },
+        ],
     )
 
     for _ in range(callback_step_count):