NREN L3 creation workflow only allows a single edge port as input

gso/workflows/nren_l3_core_service/create_nren_l3_core_service.py

 """Create a new NREN L3 Core Service subscription including GÉANT IP and IAS."""
 
-from typing import Annotated, Any
+from typing import Any
 from uuid import uuid4
 
 from orchestrator.forms import FormPage
@@ -10,10 +10,10 @@ from orchestrator.types import FormGenerator, State, SubscriptionLifecycle, UUID
 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, computed_field
+from pydantic import BaseModel, ConfigDict, Field, computed_field
 from pydantic_forms.validators import Divider
 
-from gso.products.product_blocks.bgp_session import BFDSettings, BGPSession, IPFamily
+from gso.products.product_blocks.bgp_session import BGPSession, IPFamily
 from gso.products.product_blocks.nren_l3_core_service import NRENAccessPortInactive
 from gso.products.product_blocks.service_binding_port import ServiceBindingPortInactive
 from gso.products.product_types.edge_port import EdgePort
@@ -45,24 +45,15 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
         edge_port: active_edge_port_selector(partner_id=initial_user_input.partner)  # type: ignore[valid-type]
         ap_type: APType
 
-    def validate_edge_ports_are_unique(edge_ports: list[EdgePortSelection]) -> list[EdgePortSelection]:
-        """Verify if interfaces are unique."""
-        port_names = [port.edge_port 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 EdgePortSelectionForm(FormPage):
         model_config = ConfigDict(title=f"{product_name} - Select Edge Ports")
         info_label: Label = Field(
             f"Please select the Edge Ports where this {product_name} service will terminate", exclude=True
         )
 
-        edge_ports: Annotated[list[EdgePortSelection], AfterValidator(validate_edge_ports_are_unique)]
+        edge_port: EdgePortSelection
 
-    selected_edge_ports = yield EdgePortSelectionForm
-    ep_list = selected_edge_ports.edge_ports
+    selected_edge_port = yield EdgePortSelectionForm
 
     class BFDSettingsForm(BaseModel):
         bfd_enabled: bool = False
@@ -70,7 +61,7 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
         bfd_multiplier: int | None = None
 
     class BaseBGPPeer(BaseModel):
-        bfd_settings: BFDSettingsForm = BFDSettingsForm(bfd_enabled=False, bfd_interval=None, bfd_multiplier=None)
+        bfd_settings: BFDSettingsForm
         has_custom_policies: bool = False
         authentication_key: str | None = None
         multipath_enabled: bool = False
@@ -95,47 +86,41 @@ 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]
 
-    binding_port_inputs = []
-    for ep_index, edge_port in enumerate(ep_list):
-
-        class BindingPortsInputForm(FormPage):
-            model_config = ConfigDict(title=f"{product_name} - 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 {EdgePort.from_subscription(edge_port.edge_port).description} "
-                f"(Access Port type: {edge_port.ap_type})",
-                exclude=True,
-            )
-
-            geant_sid: str
-            is_tagged: bool = False
-            v4_bfd_settings: BFDSettingsForm
-            v6_bfd_settings: BFDSettingsForm
-            vlan_id: VLAN_ID
-            ipv4_address: IPv4AddressType
-            ipv4_mask: IPV4Netmask
-            ipv6_address: IPv6AddressType
-            ipv6_mask: IPV6Netmask
-            custom_firewall_filters: bool = False
-            divider: Divider = Field(None, exclude=True)
-            v4_bgp_peer: IPv4BGPPeer
-            v6_bgp_peer: IPv6BGPPeer
-
-        binding_port_input_form = yield BindingPortsInputForm
-        binding_port_inputs.append(
-            binding_port_input_form.model_dump()
-            | {
-                "bgp_peers": [
-                    binding_port_input_form.v4_bgp_peer.model_dump(),
-                    binding_port_input_form.v6_bgp_peer.model_dump(),
-                ]
-            }
+    class BindingPortInputForm(FormPage):
+        model_config = ConfigDict(title=f"{product_name} - Configure Edge Port")
+        info_label: Label = Field("Please configure the Service Binding Ports for the Edge Port.", exclude=True)
+        current_ep_label: Label = Field(
+            f"Currently configuring on {EdgePort.from_subscription(selected_edge_port.edge_port.edge_port).description}"
+            f" (Access Port type: {selected_edge_port.edge_port.ap_type})",
+            exclude=True,
         )
 
+        geant_sid: str
+        is_tagged: bool = False
+        v4_bfd_settings: BFDSettingsForm
+        v6_bfd_settings: BFDSettingsForm
+        vlan_id: VLAN_ID
+        ipv4_address: IPv4AddressType
+        ipv4_mask: IPV4Netmask
+        ipv6_address: IPv6AddressType
+        ipv6_mask: IPV6Netmask
+        custom_firewall_filters: bool = False
+        divider: Divider = Field(None, exclude=True)
+        v4_bgp_peer: IPv4BGPPeer
+        v6_bgp_peer: IPv6BGPPeer
+
+    binding_port_input_form = yield BindingPortInputForm
+    binding_port_input = binding_port_input_form.model_dump() | {
+        "bgp_peers": [
+            binding_port_input_form.v4_bgp_peer.model_dump(),
+            binding_port_input_form.v6_bgp_peer.model_dump(),
+        ]
+    }
+
     return (
         initial_user_input.model_dump()
-        | selected_edge_ports.model_dump()
-        | {"binding_port_inputs": binding_port_inputs, "product_name": product_name}
+        | selected_edge_port.model_dump()
+        | {"binding_port_input": binding_port_input, "product_name": product_name}
     )
 
 
@@ -176,7 +161,7 @@ def initialize_subscription(
 
     subscription.description = f"{product_name} service"
     partner_name = get_partner_by_id(subscription.customer_id).name
- 
+
 
     return {"subscription": subscription, "edge_port_fqdn_list": edge_port_fqdn_list, "partner_name": partner_name}
 

test/workflows/nren_l3_core_service/test_create_nren_l3_core_service.py

@@ -49,7 +49,7 @@ def test_create_nren_l3_core_service_success(
     form_input_data = [
         {"product": product_id},
         {"tt_number": faker.tt_number(), "partner": partner["partner_id"]},
-        {"edge_ports": [{"edge_port": edge_port_a, "ap_type": APType.PRIMARY}]},
+        {"edge_port": {"edge_port": edge_port_a, "ap_type": APType.PRIMARY}},
         {
             "geant_sid": faker.geant_sid(),
             "is_tagged": faker.boolean(),
@@ -88,5 +88,5 @@ def test_create_nren_l3_core_service_success(
     assert len(subscription.nren_l3_core_service.nren_ap_list) == 1
     assert (
         str(subscription.nren_l3_core_service.nren_ap_list[0].sbp.edge_port.owner_subscription_id)
-        == form_input_data[2]["edge_ports"][0]["edge_port"]
+        == form_input_data[2]["edge_port"]["edge_port"]
     )