Adjust GÉANT IP creation form

gso/translations/en-GB.json

@@ -32,7 +32,13 @@
             "migrate_to_different_site": "Migrating to a different Site",
             "remove_configuration": "Remove configuration from the router",
             "clean_up_ipam": "Clean up related entries in IPAM",
-            "restore_isis_metric": "Restore ISIS metric to original value"
+            "restore_isis_metric": "Restore ISIS metric to original value",
+
+            "bgp_peers": {
+                "bfd_enabled": "BFD enabled",
+                "bfd_interval": "BFD interval",
+                "bfd_multiplier": "BFD multiplier"
+            }
         }
     },
     "workflow": {
@@ -66,12 +72,14 @@
         "create_imported_super_pop_switch": "NOT FOR HUMANS -- Import existing super PoP switch",
         "create_imported_office_router": "NOT FOR HUMANS -- Import existing office router",
         "create_imported_opengear": "NOT FOR HUMANS -- Import existing OpenGear",
+        "create_imported_edge_port": "NOT FOR HUMANS -- Import existing Edge Port",
         "import_site": "NOT FOR HUMANS -- Finalize import into a Site product",
         "import_router": "NOT FOR HUMANS -- Finalize import into a Router product",
         "import_iptrunk": "NOT FOR HUMANS -- Finalize import into an IP trunk product",
         "import_office_router": "NOT FOR HUMANS -- Finalize import into an Office router product",
         "import_super_pop_switch": "NOT FOR HUMANS -- Finalize import into a Super PoP switch",
         "import_opengear": "NOT FOR HUMANS -- Finalize import into an OpenGear",
+        "import_edge_port": "NOT FOR HUMANS -- Finalize import into an Edge Port",
         "validate_iptrunk": "Validate IP Trunk configuration",
         "validate_router": "Validate Router configuration",
         "validate_switch": "Validate Switch configuration",

gso/workflows/edge_port/import_edge_port.py

@@ -6,7 +6,8 @@ from orchestrator.workflow import StepList, done, init, step, workflow
 from orchestrator.workflows.steps import resync, store_process_subscription, unsync
 from orchestrator.workflows.utils import wrap_modify_initial_input_form
 
-from gso.products import EdgePort, ImportedEdgePort, ProductName
+from gso.products import ProductName
+from gso.products.product_types.edge_port import EdgePort, ImportedEdgePort
 from gso.services.subscriptions import get_product_id_by_name
 
 

gso/workflows/geant_ip/create_geant_ip.py

@@ -11,8 +11,9 @@ 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, PositiveInt
-from pydantic_forms.validators import Divider, validate_unique_list
+from pydantic import AfterValidator, BaseModel, ConfigDict, Field, computed_field
+from pydantic_forms.types import strEnum
+from pydantic_forms.validators import Divider
 
 from gso.products.product_blocks.bgp_session import BGPSession, IPFamily
 from gso.products.product_blocks.geant_ip import NRENAccessPortInactive
@@ -40,9 +41,15 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
 
     initial_user_input = yield CreateGeantIPForm
 
+    class BGPPeerFamily(strEnum):
+        V4 = "IPv4 only"
+        V6 = "IPv6 only"
+        DUAL_STACK = "Dual stack"
+
     class EdgePortSelection(BaseModel):
         edge_port: active_edge_port_selector(partner_id=initial_user_input.partner)  # type: ignore[valid-type]
         ap_type: APType
+        ip_family: BGPPeerFamily
 
     def validate_edge_ports_are_unique(edge_ports: list[EdgePortSelection]) -> list[EdgePortSelection]:
         """Verify if interfaces are unique."""
@@ -54,7 +61,9 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
 
     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"
+        info_label: Label = Field(
+            "Please select the Edge Ports where this GÉANT IP service will terminate", exclude=True
+        )
 
         edge_ports: Annotated[list[EdgePortSelection], AfterValidator(validate_edge_ports_are_unique)]
 
@@ -63,18 +72,71 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
     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
+        has_custom_policies: bool = False
+        authentication_key: str
+        multipath_enabled: bool = False
+        send_default_route: bool = False
+        is_multi_hop: bool = False
+
+    class V4OnlyBGPPeer(BaseBGPPeer):
+        peer_address: IPv4AddressType
+        add_v4_multicast: bool = Field(default=False, exclude=True)
+
+        @computed_field  # type: ignore[misc]
+        @property
+        def families(self) -> list[IPFamily]:
+            return [IPFamily.V4UNICAST, IPFamily.V4MULTICAST] if self.add_v4_multicast else [IPFamily.V4UNICAST]
+
+    class V6OnlyBGPPeer(BaseBGPPeer):
+        peer_address: IPv6AddressType
+        add_v6_multicast: bool = Field(default=False, exclude=True)
+
+        @computed_field  # type: ignore[misc]
+        @property
+        def families(self) -> list[IPFamily]:
+            return [IPFamily.V6UNICAST, IPFamily.V6MULTICAST] if self.add_v6_multicast else [IPFamily.V6UNICAST]
+
+    class DualStackBGPPeer(V4OnlyBGPPeer, V6OnlyBGPPeer):
+        peer_address: IPAddress  # type: ignore[assignment]
+
+        @computed_field  # type: ignore[misc]
+        @property
+        def families(self) -> list[IPFamily]:
+            result = [IPFamily.V4UNICAST, IPFamily.V6UNICAST]
+            if self.add_v4_multicast:
+                result.append(IPFamily.V4MULTICAST)
+            if self.add_v6_multicast:
+                result.append(IPFamily.V6MULTICAST)
+            return result
+
     binding_port_inputs = []
     while current_ep_index < total_ep_count:
         current_edge_port = EdgePort.from_subscription(ep_list[current_ep_index].edge_port)
+        bgp_peer_type: type[object]
+        match ep_list[current_ep_index].ip_family:
+            case BGPPeerFamily.V4:
+                bgp_peer_type = V4OnlyBGPPeer
+            case BGPPeerFamily.V6:
+                bgp_peer_type = V6OnlyBGPPeer
+            case BGPPeerFamily.DUAL_STACK:
+                bgp_peer_type = DualStackBGPPeer
+            case _:
+                msg = f"Invalid IP family selected: {ep_list[current_ep_index].ip_family}"
+                raise ValueError(msg)
 
         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 = (
+            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"(Access Port type: {ep_list[current_ep_index].ap_type})",
+                exclude=True,
             )
 
             geant_sid: str
@@ -83,46 +145,19 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
             ipv4_address: IPv4AddressType
             ipv6_address: IPv6AddressType
             custom_firewall_filters: bool = False
-            divider: Divider
-            bgp_peer_count: PositiveInt = 1
+            divider: Divider = Field(None, exclude=True)
+            bgp_peers: list[bgp_peer_type]  # type: ignore[valid-type]
 
         binding_port_input_form = yield BindingPortsInputForm
-        binding_port_input = binding_port_input_form.model_dump(
-            exclude=set("info_label" "current_ep_label" "divider" "bgp_peer_count")
-        )
+        binding_port_input = binding_port_input_form.model_dump()
 
-        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: IPAddress
-                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_input["sbp_type"] = SBPType.L3
         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"))
+        | selected_edge_ports.model_dump()
         | {"binding_port_inputs": binding_port_inputs}
     )