From dd033080fc5e8aa454baa8aab63392da400b5903 Mon Sep 17 00:00:00 2001
From: Karel van Klink <karel.vanklink@geant.org>
Date: Tue, 5 Nov 2024 11:30:45 +0100
Subject: [PATCH] NREN L3 creation workflow only allows a single edge port as
input
---
.../create_nren_l3_core_service.py | 91 ++++++++-----------
.../test_create_nren_l3_core_service.py | 4 +-
2 files changed, 40 insertions(+), 55 deletions(-)
diff --git a/gso/workflows/nren_l3_core_service/create_nren_l3_core_service.py b/gso/workflows/nren_l3_core_service/create_nren_l3_core_service.py
index b07c06ea..c07ad501 100644
--- a/gso/workflows/nren_l3_core_service/create_nren_l3_core_service.py
+++ b/gso/workflows/nren_l3_core_service/create_nren_l3_core_service.py
@@ -1,6 +1,6 @@
"""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}
diff --git a/test/workflows/nren_l3_core_service/test_create_nren_l3_core_service.py b/test/workflows/nren_l3_core_service/test_create_nren_l3_core_service.py
index b1797f83..c2b582f4 100644
--- a/test/workflows/nren_l3_core_service/test_create_nren_l3_core_service.py
+++ b/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"]
)
--
GitLab