Add modify workflow for Layer2Circuit. WIP

gso/workflows/l2_circuit/modify_layer_2_circuit.py

+"""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
+    )