diff --git a/gso/translations/en-GB.json b/gso/translations/en-GB.json
index 0834aa693723e35d3bb2268fde753a6896d4df97..67f4bfebe3bd8e2be5ee6bf1be829b95e483671e 100644
--- a/gso/translations/en-GB.json
+++ b/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",
diff --git a/gso/workflows/edge_port/import_edge_port.py b/gso/workflows/edge_port/import_edge_port.py
index abbf49e20e2a083deae98fe008360c9912b82223..3193489a6606eaff8f553e4358f5621352bca613 100644
--- a/gso/workflows/edge_port/import_edge_port.py
+++ b/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
diff --git a/gso/workflows/geant_ip/create_geant_ip.py b/gso/workflows/geant_ip/create_geant_ip.py
index 569fd09ac2c0d35568e1114a137134a6c3c34d78..23153305dcb560658f938a10a3d17bf31b269d1d 100644
--- a/gso/workflows/geant_ip/create_geant_ip.py
+++ b/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}
)