Add IP Trunk activation workflow, including unit test. IP Trunk workflows can...

Add IP Trunk activation workflow, including unit test. IP Trunk workflows can execute on PROVISIONING state

gso/migrations/versions/2024-02-27_5bea5647f61d_add_ip_trunk_activation_workflow.py

+"""Add IP Trunk activation workflow.
+
+Revision ID: 5bea5647f61d
+Revises: 113a81d2a40a
+Create Date: 2024-02-27 17:01:57.300326
+
+"""
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = '5bea5647f61d'
+down_revision = '113a81d2a40a'
+branch_labels = None
+depends_on = None
+
+
+from orchestrator.migrations.helpers import create_workflow, delete_workflow
+
+new_workflows = [
+    {
+        "name": "activate_iptrunk",
+        "target": "MODIFY",
+        "description": "Activate an IP Trunk",
+        "product_type": "Iptrunk"
+    }
+]
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+    for workflow in new_workflows:
+        create_workflow(conn, workflow)
+
+
+def downgrade() -> None:
+    conn = op.get_bind()
+    for workflow in new_workflows:
+        delete_workflow(conn, workflow["name"])

gso/workflows/__init__.py

@@ -8,9 +8,13 @@ WF_USABLE_MAP.update(
         "redeploy_base_config": ["provisioning", "active"],
         "update_ibgp_mesh": ["provisioning", "active"],
         "activate_router": ["provisioning"],
+        "deploy_twamp": ["provisioning", "active"],
+        "modify_trunk_interface": ["provisioning", "active"],
+        "activate_iptrunk": ["provisioning"],
     }
 )
 
+LazyWorkflowInstance("gso.workflows.iptrunk.activate_iptrunk", "activate_iptrunk")
 LazyWorkflowInstance("gso.workflows.iptrunk.create_iptrunk", "create_iptrunk")
 LazyWorkflowInstance("gso.workflows.iptrunk.deploy_twamp", "deploy_twamp")
 LazyWorkflowInstance("gso.workflows.iptrunk.modify_isis_metric", "modify_isis_metric")

gso/workflows/iptrunk/activate_iptrunk.py

+"""Activate IP trunk takes a provisioning trunk to the active lifecycle state."""
+
+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, SubscriptionLifecycle, UUIDstr
+from orchestrator.workflow import StepList, done, init, inputstep, workflow
+from orchestrator.workflows.steps import resync, set_status, store_process_subscription, unsync
+from orchestrator.workflows.utils import wrap_modify_initial_input_form
+
+from gso.products.product_types.iptrunk import Iptrunk
+
+
+def _initial_input_form(subscription_id: UUIDstr) -> FormGenerator:
+    trunk = Iptrunk.from_subscription(subscription_id)
+
+    class ActivateTrunkForm(FormPage):
+        info_label: Label = "Start approval process for IP trunk activation."  # type:ignore[assignment]
+
+    user_input = yield ActivateTrunkForm
+
+    return user_input.dict() | {"subscription": trunk}
+
+
+@inputstep("Verify checklist completion", assignee=Assignee.SYSTEM)
+def verify_complete_checklist() -> FormGenerator:
+    """Show a form for the operator to input a link to the completed checklist."""
+
+    class VerifyCompleteForm(FormPage):
+        info_label: Label = "Verify that the checklist has been completed. Then continue this workflow."  # type: ignore[assignment]
+        checklist_url: str = ""
+
+    user_input = yield VerifyCompleteForm
+
+    return {"checklist_url": user_input.dict()["checklist_url"]}
+
+
+@workflow(
+    "Activate an IP Trunk",
+    initial_input_form=wrap_modify_initial_input_form(_initial_input_form),
+    target=Target.MODIFY,
+)
+def activate_iptrunk() -> StepList:
+    """Move an IP Trunk from a ``PROVISIONING`` state to an ``ACTIVE`` state.
+
+    * Send email notifications to different teams.
+    * Wait for approval to be given.
+    * Update the subscription lifecycle state to ``ACTIVE``.
+    """
+    return (
+        init
+        >> store_process_subscription(Target.MODIFY)
+        >> unsync
+        >> verify_complete_checklist
+        >> set_status(SubscriptionLifecycle.ACTIVE)
+        >> resync
+        >> done
+    )

test/fixtures.py

@@ -226,6 +226,7 @@ def iptrunk_subscription_factory(iptrunk_side_subscription_factory, faker):
         iptrunk_ipv4_network=None,
         iptrunk_ipv6_network=None,
         iptrunk_sides=None,
+        status: SubscriptionLifecycle | None = None,
     ) -> UUIDstr:
         product_id = subscriptions.get_product_id_by_name(ProductType.IP_TRUNK)
         description = description or faker.sentence()
@@ -255,6 +256,10 @@ def iptrunk_subscription_factory(iptrunk_side_subscription_factory, faker):
             iptrunk_subscription,
             SubscriptionLifecycle.ACTIVE,
         )
+
+        if status:
+            iptrunk_subscription.status = status
+
         iptrunk_subscription.description = description
         iptrunk_subscription.start_date = start_date
         iptrunk_subscription.save()

test/workflows/iptrunk/test_activate_iptrunk.py

+import pytest
+
+from gso.products import Iptrunk
+from test.workflows import (
+    assert_complete,
+    assert_suspended,
+    extract_state,
+    resume_workflow,
+    run_workflow,
+)
+
+
+@pytest.mark.workflow()
+def test_activate_router_success(
+    iptrunk_subscription_factory,
+    faker,
+):
+    #  Set up mock return values
+    product_id = iptrunk_subscription_factory(status="provisioning")
+    #  Sanity check
+    assert Iptrunk.from_subscription(product_id).status == "provisioning"
+
+    #  Run workflow
+    initial_input_data = [{"subscription_id": product_id}, {}]
+    result, process_stat, step_log = run_workflow("activate_iptrunk", initial_input_data)
+
+    assert_suspended(result)
+    result, step_log = resume_workflow(process_stat, step_log, input_data=[{"checklist_url": "http://localhost"}])
+
+    assert_complete(result)
+
+    state = extract_state(result)
+    subscription_id = state["subscription_id"]
+    subscription = Iptrunk.from_subscription(subscription_id)
+
+    assert subscription.status == "active"