add manual steps to router creation workflow

gso/workflows/router/create_router.py

@@ -2,11 +2,12 @@
 
 from typing import Any
 
+from orchestrator.config.assignee import Assignee
 from orchestrator.forms import FormPage
-from orchestrator.forms.validators import Choice
+from orchestrator.forms.validators import Choice, Label
 from orchestrator.targets import Target
 from orchestrator.types import FormGenerator, State, SubscriptionLifecycle, UUIDstr
-from orchestrator.workflow import StepList, conditional, done, init, step, workflow
+from orchestrator.workflow import StepList, conditional, done, init, inputstep, step, workflow
 from orchestrator.workflows.steps import resync, set_status, store_process_subscription
 from orchestrator.workflows.utils import wrap_create_initial_input_form
 from pydantic import validator
@@ -144,6 +145,53 @@ def verify_ipam_loopback(subscription: RouterProvisioning) -> State:
     return {"subscription": subscription}
 
 
+@inputstep("Prompt to reboot", assignee=Assignee.SYSTEM)
+def prompt_reboot_router() -> FormGenerator:
+    """Wait for confirmation from an operator that the router has been rebooted."""
+
+    class RebootPrompt(FormPage):
+        class Config:
+            title = "Please reboot before continuing"
+
+        info_label: Label = "Reboot the router, and continue the workflow once this has been completed."  # type: ignore[assignment]
+
+    yield RebootPrompt
+
+    return {}
+
+
+@inputstep("Prompt to test the console", assignee=Assignee.SYSTEM)
+def prompt_console_login() -> FormGenerator:
+    """Wait for confirmation from an operator that the router can be logged into."""
+
+    class ConsolePrompt(FormPage):
+        class Config:
+            title = "Please log in before continuing"
+
+        info_label: Label = (
+            "Verify that you are able to log in to the router via the console, and then continue the workflow."  # type: ignore[assignment]
+        )
+
+    yield ConsolePrompt
+
+    return {}
+
+
+@inputstep("Prompt IMS insertion", assignee=Assignee.SYSTEM)
+def prompt_insert_in_ims() -> FormGenerator:
+    """Wait for confirmation from an operator that the router has been inserted in IMS."""
+
+    class IMSPrompt(FormPage):
+        class Config:
+            title = "Please go to IMS before continuing"
+
+        info_label: Label = "Insert the router into IMS, and continue the workflow once this has been completed."  # type: ignore[assignment]
+
+    yield IMSPrompt
+
+    return {}
+
+
 @workflow(
     "Create router",
     initial_input_form=wrap_create_initial_input_form(initial_input_form_generator),
@@ -169,6 +217,9 @@ def create_router() -> StepList:
         >> pp_interaction(deploy_base_config_dry)
         >> pp_interaction(deploy_base_config_real)
         >> verify_ipam_loopback
+        >> prompt_reboot_router
+        >> prompt_console_login
+        >> prompt_insert_in_ims
         >> router_is_nokia(create_netbox_device)
         >> pp_interaction(run_checks_after_base_config)
         >> set_status(SubscriptionLifecycle.ACTIVE)

test/workflows/router/test_create_router.py

@@ -7,11 +7,14 @@ from gso.products import ProductType, Site
 from gso.products.product_blocks.router import RouterRole, RouterVendor
 from gso.products.product_types.router import Router
 from gso.services.subscriptions import get_product_id_by_name
+from test import USER_CONFIRM_EMPTY_FORM
 from test.workflows import (
     assert_complete,
     assert_pp_interaction_failure,
     assert_pp_interaction_success,
+    assert_suspended,
     extract_state,
+    resume_workflow,
     run_workflow,
 )
 
@@ -80,9 +83,16 @@ def test_create_nokia_router_success(
         name=mock_fqdn,
     )
 
-    for _ in range(3):
+    for _ in range(2):
         result, step_log = assert_pp_interaction_success(result, process_stat, step_log)
 
+    # Handle three consecutive user input steps
+    for _ in range(3):
+        assert_suspended(result)
+        result, step_log = resume_workflow(process_stat, step_log, input_data=USER_CONFIRM_EMPTY_FORM)
+
+    result, step_log = assert_pp_interaction_success(result, process_stat, step_log)
+
     assert_complete(result)
 
     state = extract_state(result)