add dummy ip trunk migration workflow

gso/migrations/versions/2023-08-16_e68720f2ec32_add_ip_trunk_migration_workflow.py

+"""Add IP Trunk migration workflow.
+
+Revision ID: e68720f2ec32
+Revises: a6eefd32c4f7
+Create Date: 2023-08-16 14:48:00.227803
+
+"""
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = 'e68720f2ec32'
+down_revision = 'a6eefd32c4f7'
+branch_labels = None
+depends_on = None
+
+
+from orchestrator.migrations.helpers import create_workflow, delete_workflow
+
+new_workflows = [
+    {
+        "name": "migrate_iptrunk",
+        "target": "MODIFY",
+        "description": "Migrate 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

 """Initialisation class that imports all workflows into {term}`GSO`."""
 from orchestrator.workflows import LazyWorkflowInstance
 
-LazyWorkflowInstance("gso.workflows.router.create_router", "create_router")
-LazyWorkflowInstance("gso.workflows.router.terminate_router", "terminate_router")
 LazyWorkflowInstance("gso.workflows.iptrunk.create_iptrunk", "create_iptrunk")
+LazyWorkflowInstance("gso.workflows.iptrunk.modify_isis_metric", "modify_isis_metric")
 LazyWorkflowInstance("gso.workflows.iptrunk.modify_trunk_interface", "modify_trunk_interface")
+LazyWorkflowInstance("gso.workflows.iptrunk.migrate_iptrunk", "migrate_iptrunk")
 LazyWorkflowInstance("gso.workflows.iptrunk.terminate_iptrunk", "terminate_iptrunk")
-LazyWorkflowInstance("gso.workflows.iptrunk.modify_isis_metric", "modify_isis_metric")
+LazyWorkflowInstance("gso.workflows.router.create_router", "create_router")
+LazyWorkflowInstance("gso.workflows.router.terminate_router", "terminate_router")
 LazyWorkflowInstance("gso.workflows.site.create_site", "create_site")
 LazyWorkflowInstance("gso.workflows.tasks.import_site", "import_site")
 LazyWorkflowInstance("gso.workflows.tasks.import_router", "import_router")

gso/workflows/iptrunk/migrate_iptrunk.py

+import re
+from typing import NoReturn
+
+from orchestrator import step, workflow
+from orchestrator.db import SubscriptionTable, ProductTable
+from orchestrator.forms import FormPage
+from orchestrator.forms.validators import Choice, UniqueConstrainedList
+from orchestrator.targets import Target
+from orchestrator.types import UUIDstr, FormGenerator
+from orchestrator.workflow import StepList, init, done
+from orchestrator.workflows.steps import store_process_subscription, unsync, resync
+from orchestrator.workflows.utils import wrap_modify_initial_input_form
+from pydantic import validator
+
+from products import Iptrunk
+
+
+def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
+    subscription = Iptrunk.from_subscription(subscription_id)
+
+    routers = {}
+    for router_id, router_description in (
+            SubscriptionTable.query.join(ProductTable)
+                    .filter(
+                ProductTable.product_type == "Router",
+                SubscriptionTable.status == "active",
+            )
+                    .with_entities(SubscriptionTable.subscription_id, SubscriptionTable.description)
+                    .all()
+    ):
+        if router_id not in [subscription.iptrunk.iptrunk_sideA_node, subscription.iptrunk.iptrunk_sideB_node]:
+            routers[str(router_id)] = router_description
+
+    NewRouterEnum = Choice("Select a new router", zip(routers.keys(), routers.items()))  # type: ignore
+
+    class LagMemberList(UniqueConstrainedList[str]):
+        min_items = len(subscription.iptrunk.iptrunk_sideA_ae_members)
+        max_items = len(subscription.iptrunk.iptrunk_sideA_ae_members)
+
+    class ModifyIptrunkForm(FormPage):
+        class Config:
+            title = f"Subscription {subscription.iptrunk.geant_s_sid} from " \
+                    f"{subscription.iptrunk.iptrunk_sideA_node.router_fqdn} to " \
+                    f"{subscription.iptrunk.iptrunk_sideB_node.router_fqdn}"
+
+        replace_side = Choice("Select the side of the IP trunk to be replaced",
+                              [subscription.iptrunk.iptrunk_sideA_node, subscription.iptrunk.iptrunk_sideB_node])
+        new_node: NewRouterEnum  # type: ignore
+        new_lag_interface: str
+        new_lag_member_interfaces = LagMemberList
+
+        @validator("new_lag_interface", pre=True, always=True)
+        def lag_interface_proper_name(cls, new_lag_name: str) -> str | NoReturn:
+            nokia_lag_re = re.compile("^lag-\\d+$")
+            juniper_lag_re = re.compile("^ae\\d{1,2}$")
+
+            if nokia_lag_re.match(new_lag_name) or juniper_lag_re.match(new_lag_name):
+                return new_lag_name
+
+            raise ValueError("Invalid LAG name, please try again.")
+
+    yield ModifyIptrunkForm
+
+
+@step("Hilversum Stinks")
+def temp_test_step(subscription: Iptrunk) -> Iptrunk:
+    return subscription
+
+
+@workflow(
+    "Migrate an IP Trunk",
+    initial_input_form=wrap_modify_initial_input_form(initial_input_form_generator),
+    target=Target.MODIFY,
+)
+def migrate_iptrunk() -> StepList:
+    return (
+        init
+        >> store_process_subscription(Target.MODIFY)
+        >> unsync
+        >> temp_test_step
+        >> resync
+        >> done
+    )