diff --git a/gso/migrations/versions/2024-02-27_5bea5647f61d_add_ip_trunk_activation_workflow.py b/gso/migrations/versions/2024-02-27_5bea5647f61d_add_ip_trunk_activation_workflow.py
new file mode 100644
index 0000000000000000000000000000000000000000..c7b03f9f4f1c2ccc9b953d8afa4b38819d0ea7e5
--- /dev/null
+++ b/gso/migrations/versions/2024-02-27_5bea5647f61d_add_ip_trunk_activation_workflow.py
@@ -0,0 +1,39 @@
+"""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"])
diff --git a/gso/workflows/__init__.py b/gso/workflows/__init__.py
index bbc40e87c7a13d3548eec7f730db0b03a2a92b91..3566d7fc4fcf997d286223edf0dc097fa2aa789e 100644
--- a/gso/workflows/__init__.py
+++ b/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")
diff --git a/gso/workflows/iptrunk/activate_iptrunk.py b/gso/workflows/iptrunk/activate_iptrunk.py
new file mode 100644
index 0000000000000000000000000000000000000000..f686a8cb7e3c825dceffeb876c644a37342ce3d8
--- /dev/null
+++ b/gso/workflows/iptrunk/activate_iptrunk.py
@@ -0,0 +1,59 @@
+"""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
+ )
diff --git a/test/fixtures.py b/test/fixtures.py
index bc96381bfa605c3934610eac8490fa275e6051de..32bbfdca03b7218df8e6ce36b7f184f1390668c5 100644
--- a/test/fixtures.py
+++ b/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()
diff --git a/test/workflows/iptrunk/test_activate_iptrunk.py b/test/workflows/iptrunk/test_activate_iptrunk.py
new file mode 100644
index 0000000000000000000000000000000000000000..837f340c6adc77e1aa148760e425429288e4d01f
--- /dev/null
+++ b/test/workflows/iptrunk/test_activate_iptrunk.py
@@ -0,0 +1,36 @@
+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"