Add create_imported_layer_2_circuit product, WF and unit tests

gso/migrations/versions/2024-10-29_72a4f7aa499d_add_l2circuit_workflows.py

@@ -35,6 +35,12 @@ new_workflows = [
         "target": "TERMINATE",
         "description": "Terminate Layer 2 Circuit Service",
         "product_type": "Layer2Circuit"
+    },
+    {
+        "name": "create_imported_layer_2_circuit",
+        "target": "CREATE",
+        "description": "Create imported Layer 2 Circuit",
+        "product_type": "ImportedLayer2Circuit"
     }
 ]
 

gso/workflows/__init__.py

@@ -129,3 +129,4 @@ LazyWorkflowInstance("gso.workflows.nren_l3_core_service.migrate_nren_l3_core_se
 LazyWorkflowInstance("gso.workflows.l2_circuit.create_layer_2_circuit", "create_layer_2_circuit")
 LazyWorkflowInstance("gso.workflows.l2_circuit.modify_layer_2_circuit", "modify_layer_2_circuit")
 LazyWorkflowInstance("gso.workflows.l2_circuit.terminate_layer_2_circuit", "terminate_layer_2_circuit")
+LazyWorkflowInstance("gso.workflows.l2_circuit.create_imported_layer_2_circuit", "create_imported_layer_2_circuit")

gso/workflows/l2_circuit/create_imported_layer_2_circuit.py

 """A creation workflow that adds an existing Layer 2 Circuit to the database."""
 
-from typing import Self
+from typing import Any, Self
+from uuid import uuid4
 
-from orchestrator import step
+from orchestrator import step, workflow
 from orchestrator.forms import FormPage
-from orchestrator.types import FormGenerator, State
+from orchestrator.targets import Target
+from orchestrator.types import FormGenerator, State, SubscriptionLifecycle
+from orchestrator.workflow import StepList, begin, done
+from orchestrator.workflows.steps import resync, set_status, store_process_subscription
 from pydantic import BaseModel, ConfigDict, model_validator
 from pydantic_forms.types import UUIDstr
 
 from gso.products import ProductName
-from gso.products.product_blocks.layer_2_circuit import Layer2CircuitType
+from gso.products.product_blocks.layer_2_circuit import Layer2CircuitSideBlockInactive, Layer2CircuitType
+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 (
-    ImportedLayer2Circuit,
     ImportedLayer2CircuitInactive,
     Layer2CircuitServiceType,
 )
 from gso.services.partners import get_partner_by_name
 from gso.services.subscriptions import get_product_id_by_name
+from gso.utils.shared_enums import SBPType
 from gso.utils.types.interfaces import BandwidthString
 from gso.utils.types.virtual_identifiers import VC_ID, VLAN_ID
 
@@ -26,36 +32,23 @@ def initial_input_form_generator() -> FormGenerator:
 
     class ServiceBindingPortInput(BaseModel):
         edge_port: UUIDstr
-        geant_sid: str
-        is_tagged: bool
         vlan_id: VLAN_ID
-        custom_firewall_filters: bool
 
     class ImportLayer2CircuitForm(FormPage):
         model_config = ConfigDict(title="Import Layer 2 Circuit")
 
         service_type: Layer2CircuitServiceType
         partner: str
+        geant_sid: str
+        vc_id: VC_ID
         layer_2_circuit_side_a: ServiceBindingPortInput
         layer_2_circuit_side_b: ServiceBindingPortInput
-        virtual_circuit_id: VC_ID
         layer_2_circuit_type: Layer2CircuitType
         vlan_range_lower_bound: VLAN_ID | None = None
         vlan_range_upper_bound: VLAN_ID | None = None
         policer_enabled: bool = False
-        bandwidth: BandwidthString | None = None
-
-        @model_validator(mode="after")
-        def partner_id_matches_edge_port_owner(self) -> Self:
-            """Validate that the entered partner owns both Edge Ports."""
-            partner_id = get_partner_by_name(self.partner)["partner_id"]
-            if partner_id != self.layer_2_circuit_side_a.edge_port:
-                msg = f"Selected Edge Port on side A is not owned by partner {self.partner}."
-                raise ValueError(msg)
-            if partner_id != self.layer_2_circuit_side_b.edge_port:
-                msg = f"Selected Edge Port on side B is not owned by partner {self.partner}."
-                raise ValueError(msg)
-            return self
+        policer_bandwidth: BandwidthString | None = None
+        policer_burst_rate: BandwidthString | None = None
 
         @model_validator(mode="after")
         def tagged_layer_2_circuit_has_vlan_bounds(self) -> Self:
@@ -70,6 +63,17 @@ def initial_input_form_generator() -> FormGenerator:
                 raise ValueError(msg)
             return self
 
+        @model_validator(mode="after")
+        def policer_bandwidth_required_if_policer_enabled(self) -> Self:
+            """If the policer is enabled, the bandwidth and burst rate must be set."""
+            if self.policer_enabled and (self.policer_bandwidth is None or self.policer_burst_rate is None):
+                msg = (
+                    f"If the policer is enabled, the bandwidth and burst rate must be set. Received bandwidth: "
+                    f"{self.policer_bandwidth}, burst rate: {self.policer_burst_rate}."
+                )
+                raise ValueError(msg)
+            return self
+
     user_input = yield ImportLayer2CircuitForm
     return user_input.model_dump()
 
@@ -79,7 +83,7 @@ def create_subscription(partner: str, service_type: Layer2CircuitServiceType) ->
     """Create a new subscription object."""
     partner_id = get_partner_by_name(partner)["partner_id"]
     product_id = get_product_id_by_name(ProductName(service_type))
-    subscription = ImportedLayer2Circuit.from_product_id(product_id, partner_id)
+    subscription = ImportedLayer2CircuitInactive.from_product_id(product_id, partner_id)
 
     return {"subscription": subscription, "subscription_id": subscription.subscription_id}
 
@@ -87,10 +91,57 @@ def create_subscription(partner: str, service_type: Layer2CircuitServiceType) ->
 @step("Initialize subscription")
 def initialize_subscription(
     subscription: ImportedLayer2CircuitInactive,
-    vlan_range_lower_bound: VLAN_ID,
-    vlan_range_upper_bound: VLAN_ID,
+    geant_sid: str,
+    layer_2_circuit_side_a: dict[str, Any],
+    layer_2_circuit_side_b: dict[str, Any],
+    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,
+    policer_burst_rate: BandwidthString | None,
+    vc_id: VC_ID,
 ) -> State:
     """Initialize the subscription object."""
+    layer_2_circuit_sides = []
+    for circuit_side_data in [layer_2_circuit_side_a, layer_2_circuit_side_b]:
+        sbp = ServiceBindingPortInactive.new(
+            uuid4(),
+            edge_port=EdgePort.from_subscription(subscription_id=circuit_side_data["edge_port"]).edge_port,
+            sbp_type=SBPType.L2,
+            vlan_id=circuit_side_data["vlan_id"],
+            geant_sid=geant_sid,
+            is_tagged=layer_2_circuit_type == Layer2CircuitType.TAGGED,
+            custom_firewall_filters=False,
+        )
+        layer2_circuit_side = Layer2CircuitSideBlockInactive.new(uuid4(), sbp=sbp)
+        layer_2_circuit_sides.append(layer2_circuit_side)
+    subscription.layer_2_circuit.layer_2_circuit_sides = layer_2_circuit_sides
+    subscription.layer_2_circuit.virtual_circuit_id = vc_id
+    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
+    subscription.layer_2_circuit.policer_burst_rate = policer_burst_rate
+    subscription.description = f"{subscription.product.name} - {subscription.layer_2_circuit.virtual_circuit_id}"
+
     return {"subscription": subscription}
+
+
+@workflow(
+    "Create imported Layer 2 Circuit",
+    initial_input_form=initial_input_form_generator,
+    target=Target.CREATE,
+)
+def create_imported_layer_2_circuit() -> StepList:
+    """Import a Layer 2 Circuit without provisioning it."""
+    return (
+        begin
+        >> create_subscription
+        >> store_process_subscription(Target.CREATE)
+        >> initialize_subscription
+        >> set_status(SubscriptionLifecycle.ACTIVE)
+        >> resync
+        >> done
+    )

gso/workflows/l2_circuit/import_layer_2_circuit.py

+"""Import a Layer 2 Circuit."""

test/workflows/l2_circuit/test_create_imported_layer_2_circuit.py

+import pytest
+from orchestrator.types import SubscriptionLifecycle
+
+from gso.products.product_blocks.layer_2_circuit import Layer2CircuitType
+from gso.products.product_types.layer_2_circuit import Layer2Circuit, Layer2CircuitServiceType
+from gso.utils.helpers import generate_unique_vc_id
+from test.workflows import assert_complete, extract_state, run_workflow
+
+
+@pytest.mark.parametrize(
+    "layer_2_service_type", [Layer2CircuitServiceType.GEANT_PLUS, Layer2CircuitServiceType.EXPRESSROUTE]
+)
+def test_create_imported_layer_2_circuit_success(
+    faker, partner_factory, edge_port_subscription_factory, layer_2_service_type
+):
+    partner = partner_factory()
+    edge_port_a = edge_port_subscription_factory(partner=partner)
+    edge_port_b = edge_port_subscription_factory(partner=partner)
+    policer_enabled = faker.boolean()
+    creation_form_input_data = [
+        {
+            "service_type": layer_2_service_type,
+            "partner": partner["name"],
+            "layer_2_circuit_type": Layer2CircuitType.TAGGED,
+            "policer_enabled": policer_enabled,
+            "vlan_range_lower_bound": faker.vlan_id(),
+            "vlan_range_upper_bound": faker.vlan_id(),
+            "vc_id": generate_unique_vc_id(),
+            "policer_bandwidth": faker.bandwidth() if policer_enabled else None,
+            "policer_burst_rate": faker.bandwidth() if policer_enabled else None,
+            "geant_sid": faker.geant_sid(),
+            "layer_2_circuit_side_a": {"edge_port": edge_port_a, "vlan_id": faker.vlan_id()},
+            "layer_2_circuit_side_b": {"edge_port": edge_port_b, "vlan_id": faker.vlan_id()},
+        }
+    ]
+
+    result, _, _ = run_workflow("create_imported_layer_2_circuit", creation_form_input_data)
+    state = extract_state(result)
+    assert_complete(result)
+    subscription = Layer2Circuit.from_subscription(state["subscription_id"])
+    assert subscription.status == SubscriptionLifecycle.ACTIVE
+    assert subscription.layer_2_circuit.layer_2_circuit_type == Layer2CircuitType.TAGGED
+    assert subscription.layer_2_circuit.virtual_circuit_id == creation_form_input_data[0]["vc_id"]
+    assert len(subscription.layer_2_circuit.layer_2_circuit_sides) == 2
+    assert subscription.layer_2_circuit.layer_2_circuit_sides[0].sbp.is_tagged is True
+    assert (
+        subscription.layer_2_circuit.layer_2_circuit_sides[0].sbp.geant_sid == creation_form_input_data[0]["geant_sid"]
+    )
+    assert (
+        str(subscription.layer_2_circuit.layer_2_circuit_sides[0].sbp.edge_port.owner_subscription_id)
+        == creation_form_input_data[0]["layer_2_circuit_side_a"]["edge_port"]
+    )
+    assert subscription.layer_2_circuit.policer_enabled == policer_enabled
+    assert subscription.layer_2_circuit.bandwidth == creation_form_input_data[0]["policer_bandwidth"]
+    assert subscription.layer_2_circuit.policer_burst_rate == creation_form_input_data[0]["policer_burst_rate"]