Restrict promotion workflow to Nokia P-routers

gso/workflows/router/promote_p_to_pe.py

@@ -10,10 +10,10 @@ from orchestrator.targets import Target
 from orchestrator.types import FormGenerator, State, UUIDstr
 from orchestrator.utils.errors import ProcessFailureError
 from orchestrator.utils.json import json_dumps
-from orchestrator.workflow import StepList, begin, conditional, done, inputstep, step, workflow
+from orchestrator.workflow import StepList, begin, done, 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 ConfigDict
+from pydantic import ConfigDict, model_validator
 
 from gso.products.product_blocks.router import RouterRole
 from gso.products.product_types.router import Router
@@ -35,6 +35,14 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
 
         tt_number: TTNumber
 
+        @model_validator(mode="before")
+        def router_must_be_nokia_p(cls, data: Any) -> Any:
+            if not (subscription.router.router_role == RouterRole.P and subscription.router.vendor == Vendor.NOKIA):
+                msg = "PE promotion workflow can only be run on Nokia P routers."
+                raise ValueError(msg)
+
+            return data
+
     user_input = yield PromotePToPEForm
 
     return user_input.model_dump() | {"subscription": subscription}
@@ -147,10 +155,6 @@ def create_kentik_device(subscription: Router) -> State:
     )
     kentik_device = kentik_client.create_device(new_device)
 
-    if "error" in kentik_device or "kentik_error" in kentik_device:
-        raise ProcessFailureError(str(kentik_device))
-
-    kentik_device.pop("custom_column_data", None)
     return {"kentik_device": kentik_device}
 
 
@@ -559,14 +563,9 @@ def delete_default_routes_real(
 )
 def promote_p_to_pe() -> StepList:
     """Promote a P router to a PE router."""
-    router_is_juniper = conditional(lambda state: state["subscription"]["router"]["vendor"] == Vendor.JUNIPER)
-    router_is_pe = conditional(lambda state: state["subscription"]["router"]["router_role"] == RouterRole.PE)
-
     return (
         begin
         >> store_process_subscription(Target.MODIFY)
-        >> router_is_juniper(done)
-        >> router_is_pe(done)
         >> unsync
         >> lso_interaction(set_isis_overload)
         >> lso_interaction(deploy_pe_base_config_dry)

test/workflows/router/test_promote_p_to_pe.py

@@ -46,33 +46,30 @@ def test_promote_p_to_pe_success(
 
 
 @pytest.mark.workflow()
-@patch("gso.workflows.router.promote_p_to_pe.lso_client.execute_playbook")
-def test_promote_p_to_pe_juniper_router(
-    mock_execute_playbook, juniper_router_subscription_factory, data_config_filename, faker
-):
+def test_promote_p_to_pe_juniper_router(juniper_router_subscription_factory, data_config_filename, faker):
     """Test that the workflow does not run for a Juniper P router since this workflow is only for Nokia routers."""
     router_id = juniper_router_subscription_factory(router_role=RouterRole.P, status=SubscriptionLifecycle.ACTIVE)
     input_data = [{"subscription_id": router_id}, {"tt_number": faker.tt_number()}]
-    result, _, _ = run_workflow("promote_p_to_pe", input_data)
-    assert_complete(result)
-    state = extract_state(result)
-    assert mock_execute_playbook.call_count == 0
-    assert state["subscription"]["router"]["router_role"] == RouterRole.P
+    with pytest.raises(FormValidationError) as error:
+        run_workflow("promote_p_to_pe", input_data)
+    error = error.value.errors[0]
+    assert error["msg"] == "PE promotion workflow can only be run on Nokia P routers."
+    assert error["loc"][0] == "__root__"
 
 
 @pytest.mark.workflow()
 @patch("gso.workflows.router.promote_p_to_pe.lso_client.execute_playbook")
-def test_promote_p_to_pe_nokia_p_router(
+def test_promote_p_to_pe_nokia_pe_router(
     mock_execute_playbook, nokia_router_subscription_factory, data_config_filename, faker
 ):
     """Test that the workflow does not run for a Nokia PE router since it is already a PE router."""
     router_id = nokia_router_subscription_factory(router_role=RouterRole.PE, status=SubscriptionLifecycle.ACTIVE)
     input_data = [{"subscription_id": router_id}, {"tt_number": faker.tt_number()}]
-    result, _, _ = run_workflow("promote_p_to_pe", input_data)
-    assert_complete(result)
-    state = extract_state(result)
-    assert mock_execute_playbook.call_count == 0
-    assert state["subscription"]["router"]["router_role"] == RouterRole.PE
+    with pytest.raises(FormValidationError) as error:
+        run_workflow("promote_p_to_pe", input_data)
+    error = error.value.errors[0]
+    assert error["msg"] == "PE promotion workflow can only be run on Nokia P routers."
+    assert error["loc"][0] == "__root__"
 
 
 def test_promote_p_to_pe_missing_tt_number(nokia_router_subscription_factory):
@@ -85,11 +82,11 @@ def test_promote_p_to_pe_missing_tt_number(nokia_router_subscription_factory):
     assert error["loc"][0] == "tt_number"
 
 
-def test_promote_p_to_pe_with_invalid_router_life_cycle(nokia_router_subscription_factory):
+def test_promote_p_to_pe_with_invalid_router_life_cycle(nokia_router_subscription_factory, faker):
     """Test that the router life cycle must be ACTIVE to run this workflow."""
     router_id = nokia_router_subscription_factory(router_role=RouterRole.P, status=SubscriptionLifecycle.PROVISIONING)
     with pytest.raises(FormValidationError) as error:
-        run_workflow("promote_p_to_pe", [{"subscription_id": router_id}, {"tt_number": "TT123"}])
+        run_workflow("promote_p_to_pe", [{"subscription_id": router_id}, {"tt_number": faker.tt_number()}])
     error = error.value.errors[0]
     assert error["msg"] == (
         "This workflow cannot be started: This subscription can not be modified because of the status it has"