From b4fedb4f76eab4e90675c3ff58c75026fad5ce8c Mon Sep 17 00:00:00 2001
From: Karel van Klink <karel.vanklink@geant.org>
Date: Fri, 20 Sep 2024 15:49:14 +0200
Subject: [PATCH] Update input form and subscription model creation for geant
ip workflow
---
gso/utils/helpers.py | 2 +-
gso/workflows/edge_port/create_edge_port.py | 2 +-
gso/workflows/geant_ip/create_geant_ip.py | 99 ++++++++++++---------
3 files changed, 61 insertions(+), 42 deletions(-)
diff --git a/gso/utils/helpers.py b/gso/utils/helpers.py
index 18d7674e..e3dda41d 100644
--- a/gso/utils/helpers.py
+++ b/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)
diff --git a/gso/workflows/edge_port/create_edge_port.py b/gso/workflows/edge_port/create_edge_port.py
index 312461a0..fdf1e760 100644
--- a/gso/workflows/edge_port/create_edge_port.py
+++ b/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
diff --git a/gso/workflows/geant_ip/create_geant_ip.py b/gso/workflows/geant_ip/create_geant_ip.py
index cc01eee1..fdaaf38e 100644
--- a/gso/workflows/geant_ip/create_geant_ip.py
+++ b/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 = [
--
GitLab