Add some unit tests for testing promote P to PE promotion workflow.

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

test/workflows/router/test_promote_p_to_pe.py

+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"