From 7235cee5f36badf7204de85fd50f2cf2e86439e4 Mon Sep 17 00:00:00 2001
From: Neda Moeini <neda.moeini@geant.org>
Date: Mon, 4 Nov 2024 11:19:52 +0100
Subject: [PATCH] Improve L2Circuits creation WF
---
.../l2_circuit/create_layer_2_circuit.py | 50 +++++++++++--------
1 file changed, 30 insertions(+), 20 deletions(-)
diff --git a/gso/workflows/l2_circuit/create_layer_2_circuit.py b/gso/workflows/l2_circuit/create_layer_2_circuit.py
index c3fdae2a..73e6d168 100644
--- a/gso/workflows/l2_circuit/create_layer_2_circuit.py
+++ b/gso/workflows/l2_circuit/create_layer_2_circuit.py
@@ -1,6 +1,6 @@
"""Workflow for creating a new Layer 2 Circuit."""
-from typing import Any
+from typing import Any, Self
from uuid import uuid4
from orchestrator import step, workflow
@@ -10,7 +10,7 @@ from orchestrator.types import SubscriptionLifecycle
from orchestrator.workflow import StepList, begin, done
from orchestrator.workflows.steps import resync, set_status, store_process_subscription
from orchestrator.workflows.utils import wrap_create_initial_input_form
-from pydantic import BaseModel, ConfigDict, Field
+from pydantic import BaseModel, ConfigDict, Field, model_validator
from pydantic_forms.types import FormGenerator, State, UUIDstr
from pydantic_forms.validators import Divider, Label, ReadOnlyField
@@ -18,6 +18,7 @@ from gso.products.product_blocks.layer_2_circuit import Layer2CircuitSideBlockIn
from gso.products.product_blocks.service_binding_port import ServiceBindingPortInactive
from gso.products.product_types.edge_port import EdgePort
from gso.products.product_types.layer_2_circuit import Layer2Circuit, Layer2CircuitInactive
+from gso.services.partners import get_partner_by_name
from gso.utils.helpers import active_edge_port_selector, generate_unique_vc_id, partner_choice
from gso.utils.shared_enums import SBPType
from gso.utils.types.interfaces import BandwidthString
@@ -27,12 +28,13 @@ from gso.utils.types.virtual_identifiers import VLAN_ID
def initial_input_generator(product_name: str) -> FormGenerator:
"""Gather input from the operator about a new Layer 2 Circuit subscription."""
+ geant_partner_id = get_partner_by_name("GEANT")["partner_id"]
class CreateLayer2CircuitServicePage(FormPage):
model_config = ConfigDict(title=f"{product_name}")
tt_number: TTNumber
- partner: partner_choice() # type: ignore[valid-type]
+ partner: partner_choice() = geant_partner_id # type: ignore[valid-type]
divider: Divider = Field(None, exclude=True)
layer_2_circuit_type: Layer2CircuitType
@@ -41,35 +43,42 @@ def initial_input_generator(product_name: str) -> FormGenerator:
initial_user_input = yield CreateLayer2CircuitServicePage
class Layer2CircuitSideSelection(BaseModel):
- edge_port: active_edge_port_selector(partner_id=initial_user_input.partner) # type: ignore[valid-type]
+ edge_port: active_edge_port_selector(
+ partner_id=initial_user_input.partner if initial_user_input.partner != geant_partner_id else None
+ )
vlan_id: VLAN_ID
+ def vlan_range_field(*, is_tagged: bool) -> Any:
+ """Return the appropriate field type based on whether the circuit is tagged."""
+ return VLAN_ID if is_tagged else ReadOnlyField(None, default_type=int)
+
class Layer2CircuitServiceSidesPage(FormPage):
model_config = ConfigDict(title=f"{product_name} - Configure Edge Ports")
- if initial_user_input.layer_2_circuit_type == Layer2CircuitType.TAGGED:
- vlan_range_label: Label = Field("Please set a VLAN range, bounds including.", exclude=True)
- vlan_range_lower_bound: VLAN_ID
- vlan_range_upper_bound: VLAN_ID
- else:
- vlan_range_lower_bound: ReadOnlyField(None, default_type=int)
- vlan_range_upper_bound: ReadOnlyField(None, default_type=int)
-
+ vlan_range_label: Label = Field("Please set a VLAN range, bounds including.", exclude=True)
+ vlan_range_lower_bound: vlan_range_field(
+ is_tagged=initial_user_input.layer_2_circuit_type == Layer2CircuitType.TAGGED
+ )
+ vlan_range_upper_bound: vlan_range_field(
+ is_tagged=initial_user_input.layer_2_circuit_type == Layer2CircuitType.TAGGED
+ )
vlan_divider: Divider = Field(None, exclude=True)
-
- if initial_user_input.policer_enabled:
- policer_bandwidth: BandwidthString
- else:
- policer_bandwidth: ReadOnlyField(None, default_type=str)
-
- policer_divider: Divider = Field(None, exclude=True)
-
+ policer_bandwidth: (
+ BandwidthString if initial_user_input.policer_enabled else ReadOnlyField(None, default_type=str)
+ )
geant_sid: str
layer_2_circuit_side_a: Layer2CircuitSideSelection
side_divider: Divider = Field(None, exclude=True)
layer_2_circuit_side_b: Layer2CircuitSideSelection
+ @model_validator(mode="after")
+ def check_unique_sides(self) -> Self:
+ if self.layer_2_circuit_side_a.edge_port == self.layer_2_circuit_side_b.edge_port:
+ msg = "Both sides of the circuit cannot be connected to the same edge port"
+ raise ValueError(msg)
+ return self
+
layer_2_circuit_input = yield Layer2CircuitServiceSidesPage
return {"product_name": product_name} | initial_user_input.model_dump() | layer_2_circuit_input.model_dump()
@@ -116,6 +125,7 @@ def initialize_subscription(
subscription.layer_2_circuit.vlan_range_upper_bound = vlan_range_upper_bound
subscription.layer_2_circuit.policer_enabled = policer_enabled
subscription.layer_2_circuit.bandwidth = policer_bandwidth
+ subscription.description = (f"GÉANT PLUS {subscription.layer_2_circuit.virtual_circuit_id}")
subscription = Layer2Circuit.from_other_lifecycle(subscription, SubscriptionLifecycle.PROVISIONING)
--
GitLab