diff --git a/gso/workflows/router/promote_p_to_pe.py b/gso/workflows/router/promote_p_to_pe.py
index 6496267a73c7e4eb7f4772422e425536e2974a8c..5b3a4bdbc2cfa0a8d7458298f1464a56993278d7 100644
--- a/gso/workflows/router/promote_p_to_pe.py
+++ b/gso/workflows/router/promote_p_to_pe.py
@@ -14,7 +14,7 @@ from orchestrator.workflows.utils import wrap_modify_initial_input_form
from pydantic import ConfigDict, field_validator
from gso.products.product_blocks.router import RouterRole
-from gso.products.product_types.router import Router, RouterInactive
+from gso.products.product_types.router import Router
from gso.services import lso_client
from gso.services.kentik_client import KentikClient, NewKentikDevice
from gso.services.lso_client import lso_interaction
@@ -40,6 +40,14 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
return user_input.model_dump()
+@step("Prepare required keys in state")
+def prepare_state(subscription_id: UUIDstr) -> State:
+ """Add required keys to the state for the workflow to run successfully."""
+ router = Router.from_subscription(subscription_id)
+
+ return {"subscription": router}
+
+
@step("Evacuate the router by setting isis_overload")
def set_isis_overload(subscription: dict[str, Any], callback_route: str, tt_number: str, process_id: UUIDstr) -> None:
"""Evacuate the router by setting isis overload."""
@@ -115,7 +123,7 @@ def prompt_insert_in_earl(subscription: dict[str, Any]) -> FormGenerator:
@step("Create Kentik device")
-def create_kentik_device(subscription: RouterInactive) -> State:
+def create_kentik_device(subscription: Router) -> State:
"""Create a new device in Kentik."""
if not (
subscription.router.router_site
@@ -565,6 +573,7 @@ def promote_p_to_pe() -> StepList:
return (
begin
>> store_process_subscription(Target.MODIFY)
+ >> prepare_state
>> router_is_juniper(done)
>> router_is_pe(done)
>> unsync
diff --git a/test/workflows/router/test_promote_p_to_pe.py b/test/workflows/router/test_promote_p_to_pe.py
new file mode 100644
index 0000000000000000000000000000000000000000..67f789e4bd57f75299f7014c9c391ee165de56f1
--- /dev/null
+++ b/test/workflows/router/test_promote_p_to_pe.py
@@ -0,0 +1,97 @@
+from unittest.mock import patch
+
+import pytest
+from orchestrator.types import SubscriptionLifecycle
+from pydantic_forms.exceptions import FormValidationError
+
+from gso.products.product_blocks.router import RouterRole
+from test import USER_CONFIRM_EMPTY_FORM
+from test.workflows import (
+ assert_complete,
+ assert_lso_interaction_success,
+ extract_state,
+ resume_workflow,
+ run_workflow,
+)
+
+
+@pytest.mark.workflow()
+@patch("gso.workflows.router.promote_p_to_pe.lso_client.execute_playbook")
+@patch("gso.workflows.router.promote_p_to_pe.KentikClient.create_device")
+@patch("gso.workflows.router.promote_p_to_pe.KentikClient.get_site_by_name")
+def test_promote_p_to_pe_success(
+ mock_kentik_create_device,
+ mock_kentik_get_site_by_name,
+ mock_execute_playbook,
+ nokia_router_subscription_factory,
+ data_config_filename,
+ faker,
+):
+ """Test the successful promotion of a Nokia P router to a PE router."""
+ mock_kentik_create_device.return_value = {"id": faker.pyint()}
+ mock_kentik_get_site_by_name.return_value = {"id": faker.pyint()}
+ router_id = nokia_router_subscription_factory(router_role=RouterRole.P, status=SubscriptionLifecycle.ACTIVE)
+ input_data = [{"subscription_id": router_id}, {"tt_number": faker.tt_number()}]
+ result, process_stat, step_log = run_workflow("promote_p_to_pe", input_data)
+ for _ in range(3):
+ result, step_log = assert_lso_interaction_success(result, process_stat, step_log)
+ result, step_log = resume_workflow(process_stat, step_log, input_data=USER_CONFIRM_EMPTY_FORM)
+ for _ in range(19):
+ result, step_log = assert_lso_interaction_success(result, process_stat, step_log)
+ state = extract_state(result)
+ assert_complete(result)
+ assert mock_execute_playbook.call_count == 22
+ assert mock_kentik_create_device.call_count == 1
+ assert state["subscription"]["router"]["router_role"] == RouterRole.PE
+
+
+@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
+):
+ """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
+
+
+@pytest.mark.workflow()
+@patch("gso.workflows.router.promote_p_to_pe.lso_client.execute_playbook")
+def test_promote_p_to_pe_nokia_p_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
+
+
+def test_promote_p_to_pe_missing_tt_number(nokia_router_subscription_factory):
+ """Test that a missing TT number results in a validation error."""
+ router_id = nokia_router_subscription_factory(router_role=RouterRole.P, status=SubscriptionLifecycle.ACTIVE)
+ with pytest.raises(FormValidationError) as error:
+ run_workflow("promote_p_to_pe", [{"subscription_id": router_id}, {}])
+ error = error.value.errors[0]
+ assert error["msg"] == "Field required"
+ assert error["loc"][0] == "tt_number"
+
+
+def test_promote_p_to_pe_with_invalid_router_life_cycle(nokia_router_subscription_factory):
+ """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"}])
+ 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"
+ )
+ assert error["loc"][0] == "subscription_id"