Update input form and subscription model creation for geant ip workflow

gso/utils/helpers.py

@@ -214,7 +214,7 @@ def active_edge_port_selector(*, geant_only: bool | None = None) -> Choice:
             edge_port_subscriptions,
         )
 
-    edge_port_subscriptions = {port["subscription_id"]: port["description"] for port in edge_port_subscriptions}
+    edge_port_subscriptions = {str(port["subscription_id"]): port["description"] for port in edge_port_subscriptions}
 
     return Choice(
         "Select an Edge Port", zip(edge_port_subscriptions.keys(), edge_port_subscriptions.items(), strict=True)

gso/workflows/edge_port/create_edge_port.py

@@ -49,7 +49,7 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
         node: router_enum  # type: ignore[valid-type]
         partner: partner_choice()  # type: ignore[valid-type]
         service_type: EdgePortType
-        enable_lacp: bool
+        enable_lacp: bool = False
         speed: PhysicalPortCapacity
         encapsulation: EncapsulationType = EncapsulationType.DOT1Q
         number_of_members: int

gso/workflows/geant_ip/create_geant_ip.py

@@ -11,7 +11,7 @@ from orchestrator.utils.errors import ProcessFailureError
 from orchestrator.workflow import StepList, begin, done, step, workflow
 from orchestrator.workflows.steps import resync, set_status, store_process_subscription
 from orchestrator.workflows.utils import wrap_create_initial_input_form
-from pydantic import AfterValidator, BaseModel, ConfigDict, Field
+from pydantic import AfterValidator, BaseModel, ConfigDict, PositiveInt
 from pydantic_forms.validators import Divider, validate_unique_list
 
 from gso.products.product_blocks.bgp_session import BGPSession, IPFamily
@@ -20,13 +20,12 @@ from gso.products.product_blocks.service_binding_port import VLAN_ID, ServiceBin
 from gso.products.product_types.edge_port import EdgePort
 from gso.products.product_types.geant_ip import GeantIPInactive
 from gso.services.lso_client import execute_playbook, lso_interaction
-from gso.services.partners import get_partner_by_name
 from gso.utils.helpers import (
     active_edge_port_selector,
     partner_choice,
 )
 from gso.utils.shared_enums import APType, SBPType
-from gso.utils.types.ip_address import IPAddress, IPv4AddressType, IPv6AddressType
+from gso.utils.types.ip_address import IPv4AddressType, IPv6AddressType
 from gso.utils.types.tt_number import TTNumber
 
 
@@ -45,67 +44,87 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
         edge_port: active_edge_port_selector()
         ap_type: APType
 
+        def __hash__(self):
+            return self.edge_port.__hash__()
+
     class EdgePortSelectionForm(FormPage):
         model_config = ConfigDict(title=f"{product_name} - Select Edge Ports")
         info_label: Label = "Please select the Edge Ports where this GÉANT IP service will terminate"
 
-        edge_ports: list[EdgePortSelection] = Field(default_factory=list)
+        edge_ports: Annotated[list[EdgePortSelection], AfterValidator(validate_unique_list)]
 
     selected_edge_ports = yield EdgePortSelectionForm
-    ep_list = selected_edge_ports.edge_port_pair_list
+    ep_list = selected_edge_ports.edge_ports
     total_ep_count = len(ep_list)
     current_ep_index = 0
 
-    class BGPPeer(BaseModel):
-        peer_address: IPAddress
-        bfd_enabled: bool
-        bfd_interval: int | None = None
-        bfd_multiplier: int | None = None
-        families: Annotated[list[IPFamily], AfterValidator(validate_unique_list)]
-        has_custom_policies: bool
-        authentication_key: str
-        multipath_enabled: bool
-        send_default_route: bool
-        is_multi_hop: bool
-
-    class BindingPortsInputForm(FormPage):
-        model_config = ConfigDict(
-            title=f"{product_name} - Configure Service Binding Ports ({current_ep_index + 1}/{total_ep_count})"
-        )
-        info_label: Label = "Please configure the Service Binding Ports for each Edge Port."
-        current_ep_label: Label = f'Currently configuring Edge Port: "{ep_list[current_ep_index]["description"]}"'
-
-        geant_sid: str
-        is_tagged: bool
-        sbp_type: SBPType
-        vlan_id: VLAN_ID | None
-        ipv4_address: IPv4AddressType | None = None
-        ipv6_address: IPv6AddressType | None = None
-        custom_firewall_filters: bool
-        divider: Divider
-        bgp_peers: list[BGPPeer] = Field(default_factory=list)
-
     binding_port_inputs = []
     while current_ep_index < total_ep_count:
+        current_edge_port_description = EdgePort.from_subscription(ep_list[current_ep_index].edge_port).description
+
+        class BindingPortsInputForm(FormPage):
+            model_config = ConfigDict(
+                title=f"{product_name} - Configure Service Binding Ports ({current_ep_index + 1}/{total_ep_count})"
+            )
+            info_label: Label = "Please configure the Service Binding Ports for each Edge Port."
+            current_ep_label: Label = f"Currently configuring on {current_edge_port_description}"
+
+            geant_sid: str
+            is_tagged: bool = False
+            sbp_type: SBPType
+            vlan_id: VLAN_ID | None
+            ipv4_address: IPv4AddressType | None = None
+            ipv6_address: IPv6AddressType | None = None
+            custom_firewall_filters: bool = False
+            divider: Divider
+            bgp_peer_count: PositiveInt = 1
+
         binding_port_input_form = yield BindingPortsInputForm
-        binding_port_inputs.append(
-            binding_port_input_form.model_dump(exclude=set("info_label" "current_ep_label" "divider"))
+        binding_port_input = binding_port_input_form.model_dump(
+            exclude=set("info_label" "current_ep_label" "divider" "bgp_peer_count")
         )
+
+        bgp_peers = []
+        for bgp_peer_index in range(binding_port_input_form.bgp_peer_count):
+
+            class BGPPeerForm(FormPage):
+                model_config = ConfigDict(
+                    title=(
+                        f"Configure BGP peers for {current_edge_port_description} - "
+                        f"({bgp_peer_index + 1}/{binding_port_input_form.bgp_peer_count})"
+                    )
+                )
+                peer_address: IPv4AddressType | IPv6AddressType
+                bfd_enabled: bool = False
+                bfd_interval: int | None = None
+                bfd_multiplier: int | None = None
+                families: Annotated[list[IPFamily], AfterValidator(validate_unique_list)]
+                has_custom_policies: bool = False
+                authentication_key: str
+                multipath_enabled: bool = False
+                send_default_route: bool = False
+                is_multi_hop: bool = False
+
+            bgp_peer_input_form = yield BGPPeerForm
+            bgp_peers.append(bgp_peer_input_form.model_dump())
+
+        binding_port_input["bgp_peers"] = bgp_peers
+        binding_port_inputs.append(binding_port_input)
         current_ep_index += 1
 
     return (
         initial_user_input.model_dump()
         | selected_edge_ports.model_dump(exclude=set("info_label"))
-        | binding_port_inputs
+        | {"binding_port_inputs": binding_port_inputs}
     )
 
 
 @step("Create subscription")
 def create_subscription(product: UUIDstr, partner: str) -> State:
     """Create a new subscription object in the database."""
-    subscription = GeantIPInactive.from_product_id(product, get_partner_by_name(partner)["partner_id"])
+    subscription = GeantIPInactive.from_product_id(product, partner)
 
-    return {"subscription": subscription}
+    return {"subscription": subscription, "subscription_id": subscription.subscription_id}
 
 
 @step("Initialize subscription")
@@ -120,7 +139,7 @@ def initialize_subscription(
             NRENAccessPortInactive.new(
                 subscription_id=uuid4(),
                 nren_ap_type=edge_port_input["ap_type"],
-                geant_ip_ep=edge_port_subscription.subscription_id,
+                geant_ip_ep=edge_port_subscription.edge_port,
             )
         )
         sbp_bgp_session_list = [