Abstract layer 2 circuit into two product types: GÉANT Plus and Azure ExpressRoute

gso/products/__init__.py

@@ -11,6 +11,7 @@ from pydantic_forms.types import strEnum
 from gso.products.product_types.edge_port import EdgePort, ImportedEdgePort
 from gso.products.product_types.iptrunk import ImportedIptrunk, Iptrunk
 from gso.products.product_types.lan_switch_interconnect import ImportedLanSwitchInterconnect, LanSwitchInterconnect
+from gso.products.product_types.layer_2_circuit import ImportedLayer2Circuit, Layer2Circuit, Layer2CircuitServiceType
 from gso.products.product_types.nren_l3_core_service import ImportedNRENL3CoreService, NRENL3CoreService
 from gso.products.product_types.office_router import ImportedOfficeRouter, OfficeRouter
 from gso.products.product_types.opengear import ImportedOpengear, Opengear
@@ -47,6 +48,10 @@ class ProductName(strEnum):
     IMPORTED_GEANT_IP = "Imported GÉANT IP"
     IAS = "IAS"
     IMPORTED_IAS = "Imported IAS"
+    GEANT_PLUS = Layer2CircuitServiceType.GEANT_PLUS
+    IMPORTED_GEANT_PLUS = Layer2CircuitServiceType.IMPORTED_GEANT_PLUS
+    EXPRESSROUTE = Layer2CircuitServiceType.EXPRESSROUTE
+    IMPORTED_EXPRESSROUTE = Layer2CircuitServiceType.IMPORTED_EXPRESSROUTE
 
 
 class ProductType(strEnum):
@@ -75,6 +80,10 @@ class ProductType(strEnum):
     IMPORTED_GEANT_IP = ImportedNRENL3CoreService.__name__
     IAS = NRENL3CoreService.__name__
     IMPORTED_IAS = ImportedNRENL3CoreService.__name__
+    GEANT_PLUS = Layer2Circuit.__name__
+    IMPORTED_GEANT_PLUS = ImportedLayer2Circuit.__name__
+    EXPRESSROUTE = Layer2Circuit.__name__
+    IMPORTED_EXPRESSROUTE = ImportedLayer2Circuit.__name__
 
 
 SUBSCRIPTION_MODEL_REGISTRY.update(
@@ -102,5 +111,11 @@ SUBSCRIPTION_MODEL_REGISTRY.update(
         ProductName.IMPORTED_GEANT_IP.value: ImportedNRENL3CoreService,
         ProductName.IAS.value: NRENL3CoreService,
         ProductName.IMPORTED_IAS.value: ImportedNRENL3CoreService,
+        ProductName.GEANT_PLUS.value: Layer2Circuit,
+        ProductName.IMPORTED_GEANT_PLUS.value: ImportedLayer2Circuit,
+        ProductName.EXPRESSROUTE.value: Layer2Circuit,
+        ProductName.IMPORTED_EXPRESSROUTE.value: ImportedLayer2Circuit,
     },
 )
+
+__all__ = ["ProductName", "ProductType"]

gso/products/product_types/layer_2_circuit.py

@@ -2,6 +2,7 @@
 
 from orchestrator.domain.base import SubscriptionModel
 from orchestrator.types import SubscriptionLifecycle
+from pydantic_forms.types import strEnum
 
 from gso.products.product_blocks.layer_2_circuit import (
     Layer2CircuitBlock,
@@ -10,27 +11,40 @@ from gso.products.product_blocks.layer_2_circuit import (
 )
 
 
+class Layer2CircuitServiceType(strEnum):
+    """Available types of Layer 2 Circuit services."""
+
+    GEANT_PLUS = "GÉANT Plus"
+    IMPORTED_GEANT_PLUS = "Imported GÉANT Plus"
+    EXPRESSROUTE = "Azure ExpressRoute"
+    IMPORTED_EXPRESSROUTE = "Imported Azure ExpressRoute"
+
+
 class Layer2CircuitInactive(SubscriptionModel, is_base=True):
     """An inactive Layer 2 Circuit."""
 
+    layer_2_circuit_service_type: Layer2CircuitServiceType
     layer_2_circuit: Layer2CircuitBlockInactive
 
 
 class Layer2CircuitProvisioning(Layer2CircuitInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
     """A Layer 2 Circuit that is provisioning."""
 
+    layer_2_circuit_service_type: Layer2CircuitServiceType
     layer_2_circuit: Layer2CircuitBlockProvisioning
 
 
 class Layer2Circuit(Layer2CircuitProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
     """An active Layer 2 Circuit."""
 
+    layer_2_circuit_service_type: Layer2CircuitServiceType
     layer_2_circuit: Layer2CircuitBlock
 
 
 class ImportedLayer2CircuitInactive(SubscriptionModel, is_base=True):
     """An imported, inactive Layer 2 Circuit."""
 
+    layer_2_circuit_service_type: Layer2CircuitServiceType
     layer_2_circuit: Layer2CircuitBlockInactive
 
 
@@ -39,4 +53,5 @@ class ImportedLayer2Circuit(
 ):
     """An imported Layer 2 Circuit."""
 
+    layer_2_circuit_service_type: Layer2CircuitServiceType
     layer_2_circuit: Layer2CircuitBlock

gso/workflows/edge_port/create_imported_edge_port.py

@@ -15,7 +15,7 @@ from pydantic_forms.validators import validate_unique_list
 
 from gso.products import ProductName
 from gso.products.product_blocks.edge_port import EdgePortAEMemberBlockInactive, EdgePortType, EncapsulationType
-from gso.products.product_types.edge_port import EdgePortInactive, ImportedEdgePortInactive
+from gso.products.product_types.edge_port import ImportedEdgePortInactive
 from gso.products.product_types.router import Router
 from gso.services.partners import get_partner_by_name
 from gso.services.subscriptions import get_product_id_by_name
@@ -30,17 +30,14 @@ def create_subscription(partner: str) -> State:
     product_id = get_product_id_by_name(ProductName.IMPORTED_EDGE_PORT)
     subscription = ImportedEdgePortInactive.from_product_id(product_id, partner_id)
 
-    return {
-        "subscription": subscription,
-        "subscription_id": subscription.subscription_id,
-    }
+    return {"subscription": subscription, "subscription_id": subscription.subscription_id}
 
 
 def initial_input_form_generator() -> FormGenerator:
     """Generate a form that is filled in using information passed through the :term:`API` endpoint."""
 
     class ImportEdgePort(FormPage):
-        model_config = ConfigDict(title="Import Router")
+        model_config = ConfigDict(title="Import Edge Port")
 
         node: active_pe_router_selector()  # type: ignore[valid-type]
         partner: str
@@ -63,7 +60,7 @@ def initial_input_form_generator() -> FormGenerator:
 
 @step("Initialize subscription")
 def initialize_subscription(
-    subscription: EdgePortInactive,
+    subscription: ImportedEdgePortInactive,
     node: UUIDstr,
     service_type: EdgePortType,
     speed: PhysicalPortCapacity,

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 orchestrator import step
+from orchestrator.forms import FormPage
+from orchestrator.types import FormGenerator, State
+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_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.types.interfaces import BandwidthString
+from gso.utils.types.virtual_identifiers import VC_ID, VLAN_ID
+
+
+def initial_input_form_generator() -> FormGenerator:
+    """Generate a form that can be pre-filled using an :term:`API` endpoint."""
+
+    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
+        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
+
+        @model_validator(mode="after")
+        def tagged_layer_2_circuit_has_vlan_bounds(self) -> Self:
+            """If a Layer 2 Circuit is tagged, it must have a :term:`VLAN` range set."""
+            if self.layer_2_circuit_type == Layer2CircuitType.TAGGED and (
+                self.vlan_range_lower_bound is None or self.vlan_range_upper_bound is None
+            ):
+                msg = (
+                    f"A tagged Layer 2 Circuit must have a VLAN range set. Received lower: "
+                    f"{self.vlan_range_lower_bound}, upper: {self.vlan_range_upper_bound}."
+                )
+                raise ValueError(msg)
+            return self
+
+    user_input = yield ImportLayer2CircuitForm
+    return user_input.model_dump()
+
+
+@step("Create subscription")
+def create_subscription(partner: str, service_type: Layer2CircuitServiceType) -> State:
+    """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)
+
+    return {"subscription": subscription, "subscription_id": subscription.subscription_id}
+
+
+@step("Initialize subscription")
+def initialize_subscription(
+    subscription: ImportedLayer2CircuitInactive,
+    vlan_range_lower_bound: VLAN_ID,
+    vlan_range_upper_bound: VLAN_ID,
+) -> State:
+    """Initialize the subscription object."""
+    subscription.layer_2_circuit.vlan_range_lower_bound = vlan_range_lower_bound
+    subscription.layer_2_circuit.vlan_range_upper_bound = vlan_range_upper_bound
+    return {"subscription": subscription}

gso/workflows/l2_circuit/create_layer_2_circuit.py

@@ -18,8 +18,8 @@ from pydantic_forms.validators import Divider, Label, ReadOnlyField
 
 from gso.products.product_blocks.layer_2_circuit import Layer2CircuitSideBlockInactive, Layer2CircuitType
 from gso.utils.helpers import active_edge_port_selector, partner_choice
+from gso.utils.shared_enums import SBPType
 from gso.utils.types.interfaces import BandwidthString
-from gso.utils.types.ip_address import IPv4AddressType, IPV4Netmask, IPv6AddressType, IPV6Netmask
 from gso.utils.types.tt_number import TTNumber
 from gso.utils.types.virtual_identifiers import VC_ID, VLAN_ID
 
@@ -46,10 +46,6 @@ def initial_input_generator(product_name: str) -> FormGenerator:
         geant_sid: str
         is_tagged: bool = False
         vlan_id: VLAN_ID
-        ipv4_address: IPv4AddressType
-        ipv4_mask: IPV4Netmask
-        ipv6_address: IPv6AddressType
-        ipv6_mask: IPV6Netmask
         custom_firewall_filters: bool = False
 
     class Layer2CircuitServiceSidesPage(FormPage):
@@ -103,7 +99,9 @@ def initialize_subscription(
 ) -> State:
     """Build a subscription object from all user input."""
     subscription.layer_2_circuit.layer_2_circuit_sides = [
-        Layer2CircuitSideBlockInactive.new(uuid4(), sbp=ServiceBindingPortInactive.new(uuid4(), **circuit_side))
+        Layer2CircuitSideBlockInactive.new(
+            uuid4(), sbp=ServiceBindingPortInactive.new(uuid4(), **circuit_side, sbp_type=SBPType.L2)
+        )
         for circuit_side in [layer_2_circuit_side_a, layer_2_circuit_side_b]
     ]
     subscription.layer_2_circuit.virtual_circuit_id = virtual_circuit_id