add manual steps to router insertion workflow

gso/workflows/router/update_ibgp_mesh.py

@@ -2,10 +2,12 @@
 
 from typing import Any
 
+from orchestrator.config.assignee import Assignee
 from orchestrator.forms import FormPage
+from orchestrator.forms.validators import Label
 from orchestrator.targets import Target
 from orchestrator.types import FormGenerator, State, UUIDstr
-from orchestrator.workflow import StepList, done, init, step, workflow
+from orchestrator.workflow import StepList, done, init, inputstep, step, workflow
 from orchestrator.workflows.steps import resync, store_process_subscription, unsync
 from orchestrator.workflows.utils import wrap_modify_initial_input_form
 from pydantic import root_validator
@@ -188,6 +190,36 @@ def add_device_to_librenms(subscription: Router) -> State:
     return {"librenms_device": librenms_result}
 
 
+@inputstep("Prompt RADIUS insertion", assignee=Assignee.SYSTEM)
+def prompt_insert_in_radius() -> FormGenerator:
+    """Wait for confirmation from an operator that the router has been inserted in RADIUS."""
+
+    class RADIUSPrompt(FormPage):
+        class Config:
+            title = "Please update RADIUS before continuing"
+
+        info_label: Label = "Insert the router into RADIUS, and continue the workflow once this has been completed."  # type: ignore[assignment]
+
+    yield RADIUSPrompt
+
+    return {}
+
+
+@inputstep("Prompt RADIUS login", assignee=Assignee.SYSTEM)
+def prompt_radius_login() -> FormGenerator:
+    """Wait for confirmation from an operator that the router can be logged into using RADIUS."""
+
+    class RADIUSPrompt(FormPage):
+        class Config:
+            title = "Please check RADIUS before continuing"
+
+        info_label: Label = "Log in to the router using RADIUS, and continue the workflow when this was successful."  # type: ignore[assignment]
+
+    yield RADIUSPrompt
+
+    return {}
+
+
 @step("Update subscription model")
 def update_subscription_model(subscription: Router) -> State:
     """Update the database model, such that it should not be reached via :term:`OOB` access anymore."""
@@ -221,6 +253,8 @@ def update_ibgp_mesh() -> StepList:
         >> pp_interaction(add_all_pe_to_p_real)
         >> pp_interaction(check_ibgp_session)
         >> add_device_to_librenms
+        >> prompt_insert_in_radius
+        >> prompt_radius_login
         >> update_subscription_model
         >> resync
         >> done

test/workflows/router/test_update_ibgp_mesh.py

@@ -6,7 +6,8 @@ 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
+from test import USER_CONFIRM_EMPTY_FORM
+from test.workflows import assert_pp_interaction_success, assert_suspended, extract_state, resume_workflow, run_workflow
 
 
 @pytest.fixture()
@@ -33,6 +34,11 @@ def test_update_ibgp_mesh_success(
     for _ in range(5):
         result, step_log = assert_pp_interaction_success(result, process_stat, step_log)
 
+    # Handle two consecutive user input steps
+    for _ in range(2):
+        assert_suspended(result)
+        result, step_log = resume_workflow(process_stat, step_log, input_data=USER_CONFIRM_EMPTY_FORM)
+
     state = extract_state(result)
 
     assert mock_execute_playbook.call_count == 5