Add modification workflow for GÉANT IP

gso/migrations/versions/2024-09-26_289e5334848f_add_géant_ip_modification_workflow.py

+"""Add GÉANT IP modification workflow.
+
+Revision ID: 289e5334848f
+Revises: f4239c9361b4
+Create Date: 2024-09-26 15:27:51.471877
+
+"""
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = '289e5334848f'
+down_revision = 'f4239c9361b4'
+branch_labels = None
+depends_on = None
+
+
+from orchestrator.migrations.helpers import create_workflow, delete_workflow
+
+new_workflows = [
+    {
+        "name": "modify_geant_ip",
+        "target": "MODIFY",
+        "description": "Modify G\u00c9ANT IP",
+        "product_type": "GeantIP"
+    }
+]
+
+
+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/translations/en-GB.json

@@ -54,6 +54,7 @@
         "modify_connection_strategy": "Modify connection strategy",
         "modify_router_kentik_license": "Modify device license in Kentik",
         "modify_edge_port": "Modify Edge Port",
+        "modify_geant_ip": "Modify GÉANT IP",
         "terminate_iptrunk": "Terminate IP Trunk",
         "terminate_router": "Terminate Router",
         "terminate_site": "Terminate Site",

gso/workflows/__init__.py

@@ -89,3 +89,4 @@ LazyWorkflowInstance("gso.workflows.edge_port.import_edge_port", "import_edge_po
 
 #  GÉANT IP workflows
 LazyWorkflowInstance("gso.workflows.geant_ip.create_geant_ip", "create_geant_ip")
+LazyWorkflowInstance("gso.workflows.geant_ip.modify_geant_ip", "modify_geant_ip")

gso/workflows/geant_ip/create_geant_ip.py

@@ -33,7 +33,7 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
     """Gather input from the operator to build a new subscription object."""
 
     class CreateGeantIPForm(FormPage):
-        model_config = ConfigDict(title=f"{product_name} - Select partner")
+        model_config = ConfigDict(title="GÉANT IP - Select partner")
 
         tt_number: TTNumber
         partner: partner_choice()  # type: ignore[valid-type]
@@ -53,7 +53,7 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
         return edge_ports
 
     class EdgePortSelectionForm(FormPage):
-        model_config = ConfigDict(title=f"{product_name} - Select Edge Ports")
+        model_config = ConfigDict(title="GÉANT IP - Select Edge Ports")
         info_label: Label = Field(
             "Please select the Edge Ports where this GÉANT IP service will terminate", exclude=True
         )
@@ -62,19 +62,15 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
 
     selected_edge_ports = yield EdgePortSelectionForm
     ep_list = selected_edge_ports.edge_ports
-    total_ep_count = len(ep_list)
-    current_ep_index = 0
 
     class BaseBGPPeer(BaseModel):
         bfd_enabled: bool = False
         bfd_interval: int | None = None
         bfd_multiplier: int | None = None
-        rtbh_enabled: bool = False
         has_custom_policies: bool = False
         authentication_key: str
         multipath_enabled: bool = False
         send_default_route: bool = False
-        is_multi_hop: bool = False
         is_passive: bool = False
 
     class IPv4BGPPeer(BaseBGPPeer):
@@ -95,18 +91,16 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
         def families(self) -> list[IPFamily]:
             return [IPFamily.V6UNICAST, IPFamily.V6MULTICAST] if self.add_v6_multicast else [IPFamily.V6UNICAST]
 
+    bgp_peer_defaults = {"rtbh_enabled": True, "is_multi_hop": True}
     binding_port_inputs = []
-    while current_ep_index < total_ep_count:
-        current_edge_port = EdgePort.from_subscription(ep_list[current_ep_index].edge_port)
+    for ep_index, edge_port in enumerate(ep_list):
 
         class BindingPortsInputForm(FormPage):
-            model_config = ConfigDict(
-                title=f"{product_name} - Configure Service Binding Ports ({current_ep_index + 1}/{total_ep_count})"
-            )
+            model_config = ConfigDict(title=f"GÉANT IP - Configure Edge Ports ({ep_index + 1}/{len(ep_list)})")
             info_label: Label = Field("Please configure the Service Binding Ports for each Edge Port.", exclude=True)
             current_ep_label: Label = Field(
-                f"Currently configuring on {current_edge_port.description} "
-                f"(Access Port type: {ep_list[current_ep_index].ap_type})",
+                f"Currently configuring on {EdgePort.from_subscription(edge_port.edge_port).description} "
+                f"(Access Port type: {edge_port.ap_type})",
                 exclude=True,
             )
 
@@ -125,11 +119,10 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
 
         binding_port_input["sbp_type"] = SBPType.L3
         binding_port_input["bgp_peers"] = [
-            binding_port_input_form.v4_bgp_peer.model_dump(),
-            binding_port_input_form.v6_bgp_peer.model_dump(),
+            binding_port_input_form.v4_bgp_peer.model_dump() | bgp_peer_defaults,
+            binding_port_input_form.v6_bgp_peer.model_dump() | bgp_peer_defaults,
         ]
         binding_port_inputs.append(binding_port_input)
-        current_ep_index += 1
 
     return (
         initial_user_input.model_dump()

gso/workflows/geant_ip/modify_geant_ip.py

+"""A modification workflow for a GÉANT IP subscription."""
+
+from orchestrator import begin, done, step, workflow
+from orchestrator.forms import FormPage
+from orchestrator.targets import Target
+from orchestrator.types import FormGenerator, UUIDstr
+from orchestrator.workflows.steps import State, resync, store_process_subscription, unsync
+from orchestrator.workflows.utils import wrap_modify_initial_input_form
+from pydantic import BaseModel, ConfigDict
+
+from gso.products.product_types.geant_ip import GeantIP
+from gso.utils.helpers import active_edge_port_selector
+from gso.utils.shared_enums import APType
+
+
+def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
+    subscription = GeantIP.from_subscription(subscription_id)
+
+    class AccessPortSelection(BaseModel):
+        geant_ip_ep: active_edge_port_selector(partner_id=subscription.customer_id)  # type: ignore[valid-type]
+        nren_ap_type: APType
+
+    def validate_edge_ports_are_unique(edge_ports: list[AccessPortSelection]) -> list[AccessPortSelection]:
+        """Verify if interfaces are unique."""
+        port_names = [port.geant_ip_ep for port in edge_ports]
+        if len(port_names) != len(set(port_names)):
+            msg = "Edge Ports must be unique."
+            raise ValueError(msg)
+        return edge_ports
+
+    class ModifyGeantIPAccessPortsForm(FormPage):
+        model_config = ConfigDict(title="Modify GÉANT IP")
+        access_ports: list[AccessPortSelection] = [
+            AccessPortSelection(
+                geant_ip_ep=str(access_port.geant_ip_ep.owner_subscription_id), nren_ap_type=access_port.nren_ap_type
+            )
+            for access_port in subscription.geant_ip.geant_ip_ap_list
+        ]
+
+    access_port_input = yield ModifyGeantIPAccessPortsForm
+    ap_list = access_port_input.access_ports
+    total_ap_count = len(ap_list)
+
+    access_port_inputs = []
+    for access_port_index, access_port in enumerate(ap_list):
+        access_port_input.append(access_port)
+
+    return access_port_input.model_dump()
+
+
+@step("Update subscription model")
+def modify_geant_ip_subscription(subscription: GeantIP) -> State:
+    """Update the subscription model."""
+    return {"subscription": subscription}
+
+
+@workflow(
+    "Modify GÉANT IP",
+    initial_input_form=wrap_modify_initial_input_form(initial_input_form_generator),
+    target=Target.MODIFY,
+)
+def modify_geant_ip():
+    """Modify a GÉANT IP subscription."""
+    return (
+        begin >> store_process_subscription(Target.MODIFY) >> unsync >> modify_geant_ip_subscription >> resync >> done
+    )
+
+
+# we can change the list of edge ports, and reflect this in new SBPs
+# we can change BFD
+# we can change firewall filters and policies
+# for BGP peers:
+# is_passive chan change