From 19b26d17b9429e6948a50a8c7c196ee4015f6c1c Mon Sep 17 00:00:00 2001
From: Karel van Klink <karel.vanklink@geant.org>
Date: Thu, 10 Oct 2024 09:39:51 +0200
Subject: [PATCH] =?UTF-8?q?Add=20G=C3=89ANT=20IP=20migration=20workflow?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
---
...dge_port_and_g\303\251ant_ip_workflows.py" | 6 ++
gso/translations/en-GB.json | 1 +
gso/workflows/__init__.py | 1 +
gso/workflows/geant_ip/migrate_geant_ip.py | 99 +++++++++++++++++++
4 files changed, 107 insertions(+)
create mode 100644 gso/workflows/geant_ip/migrate_geant_ip.py
diff --git "a/gso/migrations/versions/2024-10-08_bf05800fe9fc_add_edge_port_and_g\303\251ant_ip_workflows.py" "b/gso/migrations/versions/2024-10-08_bf05800fe9fc_add_edge_port_and_g\303\251ant_ip_workflows.py"
index b07a22c3..c4cf724b 100644
--- "a/gso/migrations/versions/2024-10-08_bf05800fe9fc_add_edge_port_and_g\303\251ant_ip_workflows.py"
+++ "b/gso/migrations/versions/2024-10-08_bf05800fe9fc_add_edge_port_and_g\303\251ant_ip_workflows.py"
@@ -66,6 +66,12 @@ new_workflows = [
"description": "Modify G\u00c9ANT IP",
"product_type": "GeantIP"
},
+ {
+ "name": "migrate_geant_ip",
+ "target": "MODIFY",
+ "description": "Migrate G\u00c9ANT IP",
+ "product_type": "GeantIP"
+ },
{
"name": "create_imported_geant_ip",
"target": "CREATE",
diff --git a/gso/translations/en-GB.json b/gso/translations/en-GB.json
index 2ace1604..d136f881 100644
--- a/gso/translations/en-GB.json
+++ b/gso/translations/en-GB.json
@@ -48,6 +48,7 @@
"create_geant_ip": "Create GÉANT IP",
"deploy_twamp": "Deploy TWAMP",
"migrate_iptrunk": "Migrate IP Trunk",
+ "migrate_geant_ip": "Migrate GÉANT IP",
"modify_isis_metric": "Modify the ISIS metric",
"modify_site": "Modify Site",
"modify_trunk_interface": "Modify IP Trunk interface",
diff --git a/gso/workflows/__init__.py b/gso/workflows/__init__.py
index e778e142..377edf17 100644
--- a/gso/workflows/__init__.py
+++ b/gso/workflows/__init__.py
@@ -92,3 +92,4 @@ LazyWorkflowInstance("gso.workflows.geant_ip.create_geant_ip", "create_geant_ip"
LazyWorkflowInstance("gso.workflows.geant_ip.modify_geant_ip", "modify_geant_ip")
LazyWorkflowInstance("gso.workflows.geant_ip.create_imported_geant_ip", "create_imported_geant_ip")
LazyWorkflowInstance("gso.workflows.geant_ip.import_geant_ip", "import_geant_ip")
+LazyWorkflowInstance("gso.workflows.geant_ip.migrate_geant_ip", "migrate_geant_ip")
diff --git a/gso/workflows/geant_ip/migrate_geant_ip.py b/gso/workflows/geant_ip/migrate_geant_ip.py
new file mode 100644
index 00000000..b207635f
--- /dev/null
+++ b/gso/workflows/geant_ip/migrate_geant_ip.py
@@ -0,0 +1,99 @@
+"""A modification workflow that migrates a GÉANT IP subscription to a different set of Edge Ports."""
+
+from typing import Annotated
+
+from annotated_types import Len
+from orchestrator import workflow
+from orchestrator.targets import Target
+from orchestrator.workflow import StepList, begin, done, step
+from orchestrator.workflows.steps import resync, store_process_subscription, unsync
+from orchestrator.workflows.utils import wrap_modify_initial_input_form
+from pydantic import AfterValidator, BaseModel, ConfigDict, Field
+from pydantic_forms.core import FormPage
+from pydantic_forms.types import FormGenerator, State, UUIDstr
+from pydantic_forms.validators import Choice, Divider
+
+from gso.products.product_types.edge_port import EdgePort
+from gso.products.product_types.geant_ip import GeantIP
+from gso.services.subscriptions import get_active_edge_port_subscriptions
+from gso.utils.types.tt_number import TTNumber
+
+
+def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
+ """Gather input from the operator on what new Edge Ports this GÉANT IP should migrate to."""
+ subscription = GeantIP.from_subscription(subscription_id)
+ partner_id = subscription.customer_id
+ edge_port_count = len(subscription.geant_ip.geant_ip_ap_list)
+
+ def _new_edge_port_selector(pid: UUIDstr) -> Choice:
+ existing_ep_name_list = [
+ ap.geant_ip_sbp.edge_port.owner_subscription_id for ap in subscription.geant_ip.geant_ip_ap_list
+ ]
+ edge_port_subscriptions = list(
+ filter(
+ lambda ep: bool(ep["customer_id"] == pid) and ep["subscription_id"] not in existing_ep_name_list,
+ get_active_edge_port_subscriptions(includes=["subscription_id", "description", "customer_id"]),
+ )
+ )
+
+ edge_ports = {str(port["subscription_id"]): port["description"] for port in edge_port_subscriptions}
+
+ return Choice(
+ "Select an Edge Port",
+ zip(edge_ports.keys(), edge_ports.items(), strict=True), # type: ignore[arg-type]
+ )
+
+ class NewEdgePortSelection(BaseModel):
+ old_edge_port: str
+ new_edge_port: _new_edge_port_selector(partner_id) | str # type: ignore[valid-type]
+
+ def _validate_new_edge_ports_are_unique(edge_ports: list[NewEdgePortSelection]) -> list[NewEdgePortSelection]:
+ new_edge_ports = [str(port.new_edge_port) for port in edge_ports]
+ if len(new_edge_ports) != len(set(new_edge_ports)):
+ msg = "New Edge Ports must be unique"
+ raise ValueError(msg)
+ return edge_ports
+
+ class GeantIPEdgePortSelectionForm(FormPage):
+ model_config = ConfigDict(title="Migrating GÉANT IP to a new set of Edge Ports")
+
+ tt_number: TTNumber
+ divider: Divider = Field(None, exclude=True)
+ edge_port_selection: Annotated[
+ list[NewEdgePortSelection],
+ AfterValidator(_validate_new_edge_ports_are_unique),
+ Len(min_length=edge_port_count, max_length=edge_port_count),
+ ] = [ # noqa: RUF012
+ NewEdgePortSelection(
+ old_edge_port=f"{
+ EdgePort.from_subscription(ap.geant_ip_sbp.edge_port.owner_subscription_id).description
+ } ({ap.nren_ap_type})",
+ new_edge_port="",
+ )
+ for ap in subscription.geant_ip.geant_ip_ap_list
+ ]
+
+ ep_user_input = yield GeantIPEdgePortSelectionForm
+
+ return {"subscription_id": subscription_id, "subscription": subscription} | ep_user_input.model_dump()
+
+
+@step("Update subscription model")
+def update_subscription_model(subscription: GeantIP, edge_port_selection: list[dict]) -> State:
+ """Update the subscription model with the new list of Access Ports."""
+ for index, selected_port in enumerate(edge_port_selection):
+ subscription.geant_ip.geant_ip_ap_list[index].geant_ip_sbp.edge_port = EdgePort.from_subscription(
+ selected_port["new_edge_port"]
+ ).edge_port
+
+ return {"subscription": subscription}
+
+
+@workflow(
+ "Migrate GÉANT IP",
+ initial_input_form=wrap_modify_initial_input_form(initial_input_form_generator),
+ target=Target.MODIFY,
+)
+def migrate_geant_ip() -> StepList:
+ """Migrate a GÉANT IP to a new set of Edge Ports."""
+ return begin >> store_process_subscription(Target.MODIFY) >> unsync >> update_subscription_model >> resync >> done
--
GitLab