From 0c6ac0a6c304899bd3d7756438b8e71d64e3946a Mon Sep 17 00:00:00 2001
From: Neda Moeini <neda.moeini@geant.org>
Date: Thu, 31 Oct 2024 16:24:09 +0100
Subject: [PATCH] Add modify workflow for Layer2Circuit. WIP
---
.../l2_circuit/modify_layer_2_circuit.py | 117 ++++++++++++++++++
1 file changed, 117 insertions(+)
create mode 100644 gso/workflows/l2_circuit/modify_layer_2_circuit.py
diff --git a/gso/workflows/l2_circuit/modify_layer_2_circuit.py b/gso/workflows/l2_circuit/modify_layer_2_circuit.py
new file mode 100644
index 00000000..899035a8
--- /dev/null
+++ b/gso/workflows/l2_circuit/modify_layer_2_circuit.py
@@ -0,0 +1,117 @@
+"""A modification workflow for a Layer 2 Circuit subscription."""
+
+from orchestrator import begin, done, workflow
+from orchestrator.forms import FormPage
+from orchestrator.targets import Target
+from orchestrator.types import FormGenerator, UUIDstr
+from orchestrator.workflow import StepList, step
+from orchestrator.workflows.steps import resync, store_process_subscription, unsync
+from orchestrator.workflows.utils import wrap_modify_initial_input_form
+from pydantic import BaseModel, ConfigDict, Field
+from pydantic_forms.validators import Divider, Label, ReadOnlyField
+
+from gso.products import Layer2Circuit
+from gso.products.product_blocks.layer_2_circuit import Layer2CircuitType
+from gso.services.partners import get_partner_by_id
+from gso.utils.types.interfaces import BandwidthString
+from gso.utils.types.tt_number import TTNumber
+from gso.utils.types.virtual_identifiers import VLAN_ID
+
+
+def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
+ """Get input from the operator about the modifications to be made to a Layer 2 Circuit subscription."""
+ subscription = Layer2Circuit.from_subscription(subscription_id)
+ product_name = subscription.product.name
+
+ class ModifyL2CircuitForm(FormPage):
+ model_config = ConfigDict(title=f"Modify {product_name}")
+ tt_number: TTNumber
+ partner: ReadOnlyField(get_partner_by_id(subscription.customer_id).name, default_type=str)
+ divider: Divider = Field(None, exclude=True)
+
+ layer_2_circuit_type: Layer2CircuitType = subscription.layer_2_circuit.layer_2_circuit_type
+ policer_enabled: bool = subscription.layer_2_circuit.policer_enabled
+
+ layer_2_circuit_input = yield ModifyL2CircuitForm
+
+ class ModifyLayer2CircuitSideA(BaseModel):
+ edge_port: ReadOnlyField(subscription.layer_2_circuit.layer_2_circuit_sides[0].sbp.edge_port.edge_port_name,
+ default_type=str)
+ geant_sid: ReadOnlyField(subscription.layer_2_circuit.layer_2_circuit_sides[0].sbp.geant_sid, default_type=str)
+ is_tagged: ReadOnlyField(subscription.layer_2_circuit.layer_2_circuit_sides[0].sbp.is_tagged, default_type=bool)
+ vlan_id: ReadOnlyField(subscription.layer_2_circuit.layer_2_circuit_sides[0].sbp.vlan_id, default_type=int)
+ custom_firewall_filters: ReadOnlyField(
+ subscription.layer_2_circuit.layer_2_circuit_sides[0].sbp.custom_firewall_filters, default_type=bool)
+
+ class ModifyLayer2CircuitSideB(BaseModel):
+ edge_port: ReadOnlyField(subscription.layer_2_circuit.layer_2_circuit_sides[1].sbp.edge_port.edge_port_name,
+ default_type=str)
+ geant_sid: ReadOnlyField(subscription.layer_2_circuit.layer_2_circuit_sides[1].sbp.geant_sid, default_type=str)
+ is_tagged: ReadOnlyField(subscription.layer_2_circuit.layer_2_circuit_sides[1].sbp.is_tagged, default_type=bool)
+ vlan_id: ReadOnlyField(subscription.layer_2_circuit.layer_2_circuit_sides[1].sbp.vlan_id, default_type=int)
+ custom_firewall_filters: ReadOnlyField(
+ subscription.layer_2_circuit.layer_2_circuit_sides[1].sbp.custom_firewall_filters, default_type=bool)
+
+ class ModifyLayer2CircuitServiceSidesPage(FormPage):
+ model_config = ConfigDict(title=f"{product_name} - Configure Edge Ports")
+
+ if layer_2_circuit_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_divider: Divider = Field(None, exclude=True)
+
+ if layer_2_circuit_input.policer_enabled:
+ policer_bandwidth: BandwidthString
+ else:
+ policer_bandwidth: ReadOnlyField(None, default_type=str)
+
+ policer_divider: Divider = Field(None, exclude=True)
+
+ layer_2_circuit_side_a: ModifyLayer2CircuitSideA
+ side_divider: Divider = Field(None, exclude=True)
+ layer_2_circuit_side_b: ModifyLayer2CircuitSideB
+
+ layer_2_circuit_sides = yield ModifyLayer2CircuitServiceSidesPage
+
+ return {"product_name": product_name} | layer_2_circuit_input.model_dump() | layer_2_circuit_sides.model_dump()
+
+
+@step("Update Layer 2 Circuit subscription")
+def modify_layer_2_circuit_subscription(
+ subscription: Layer2Circuit,
+ layer_2_circuit_type: Layer2CircuitType,
+ vlan_range_lower_bound: VLAN_ID | None,
+ vlan_range_upper_bound: VLAN_ID | None,
+ policer_enabled: bool, # noqa: FBT001
+ policer_bandwidth: BandwidthString | None,
+):
+ """Update the Layer 2 Circuit subscription with the new values."""
+ subscription.layer_2_circuit.layer_2_circuit_type = layer_2_circuit_type
+ subscription.layer_2_circuit.vlan_range_lower_bound = vlan_range_lower_bound
+ 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
+
+ return {"subscription": subscription}
+
+
+@workflow(
+ "Modify Layer 2 Circuit Service",
+ initial_input_form=wrap_modify_initial_input_form(initial_input_form_generator),
+ target=Target.MODIFY,
+)
+def modify_layer_2_circuit() -> StepList:
+ """Modify an existing Layer 2 Circuit service subscription."""
+ return (
+ begin
+ >> store_process_subscription(Target.MODIFY)
+ >> unsync
+ >> modify_layer_2_circuit_subscription
+ >> resync
+ >> done
+ )
--
GitLab