Add Layer 2 Circuit product blocks and creation workflow

gso/products/product_blocks/layer_2_circuit.py

+"""Layer 2 Circuit product block."""
+
+from typing import Annotated, TypeVar
+
+from annotated_types import Len
+from orchestrator.domain.base import ProductBlockModel
+from orchestrator.types import SubscriptionLifecycle
+from pydantic import AfterValidator
+from pydantic_forms.types import strEnum
+from pydantic_forms.validators import validate_unique_list
+from typing_extensions import Doc
+
+from gso.products.product_blocks.service_binding_port import (
+    ServiceBindingPort,
+    ServiceBindingPortInactive,
+    ServiceBindingPortProvisioning,
+)
+from gso.utils.types.interfaces import BandwidthString
+from gso.utils.types.virtual_identifiers import VC_ID, VLAN_ID
+
+
+class Layer2CircuitSideBlockInactive(
+    ProductBlockModel,
+    lifecycle=[SubscriptionLifecycle.INITIAL],
+    product_block_name="Layer2CircuitSideBlock",
+):
+    """One inactive side of a Layer 2 Circuit."""
+
+    sbp: ServiceBindingPortInactive
+
+
+class Layer2CircuitSideBlockProvisioning(
+    Layer2CircuitSideBlockInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]
+):
+    """One provisioning side of a Layer 2 Circuit."""
+
+    sbp: ServiceBindingPortProvisioning
+
+
+class Layer2CircuitSideBlock(Layer2CircuitSideBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
+    """One side of a Layer 2 Circuit."""
+
+    sbp: ServiceBindingPort
+
+
+Layer2CircuitSideBlockType = TypeVar(
+    "Layer2CircuitSideBlockType",
+    "Layer2CircuitSideBlockInactive",
+    "Layer2CircuitSideBlockProvisioning",
+    "Layer2CircuitSideBlock",
+)
+
+Layer2CircuitSides = Annotated[
+    list[Layer2CircuitSideBlockType],
+    AfterValidator(validate_unique_list),
+    Len(min_length=2, max_length=2),
+    Doc("A list of two Layer 2 Circuit sides."),
+]
+
+
+class Layer2CircuitType(strEnum):
+    """The two types of Layer 2 Circuit."""
+
+    TAGGED = "TAGGED"
+    UNTAGGED = "UNTAGGED"
+
+
+class Layer2CircuitBlockInactive(
+    ProductBlockModel, lifecycle=[SubscriptionLifecycle.INITIAL], product_block_name="Layer2CircuitBlock"
+):
+    """An inactive Layer 2 Circuit, see :class:`Layer2CircuitBlock`."""
+
+    layer_2_circuit_sides: Layer2CircuitSides[Layer2CircuitSideBlockInactive]
+    virtual_circuit_id: VC_ID | None = None
+    layer_2_circuit_type: Layer2CircuitType | None = None
+    vlan_range_lower_bound: VLAN_ID | None = None
+    vlan_range_upper_bound: VLAN_ID | None = None
+    policer_enabled: bool | None = None
+    bandwidth: BandwidthString | None = None
+
+
+class Layer2CircuitBlockProvisioning(Layer2CircuitBlockInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
+    """A provisioning Layer 2 Circuit, see :class:`Layer2CircuitBlock`."""
+
+    layer_2_circuit_sides: Layer2CircuitSides[Layer2CircuitSideBlockProvisioning]
+    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
+    bandwidth: BandwidthString | None = None
+
+
+class Layer2CircuitBlock(Layer2CircuitBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
+    """An active Layer 2 Circuit."""
+
+    #: The two sides that the Layer 2 Circuit is connected to.
+    layer_2_circuit_sides: Layer2CircuitSides[Layer2CircuitSideBlock]
+    #: Virtual Circuit ID of this Layer 2 Circuit.
+    virtual_circuit_id: VC_ID
+    #: The type of circuit, can be tagged or untagged.
+    layer_2_circuit_type: Layer2CircuitType
+    #: If tagged, the lower and upper bounds will set the :term:`VLAN` range.
+    vlan_range_lower_bound: VLAN_ID | None = None
+    #: Lower and Upper bounds are including.
+    vlan_range_upper_bound: VLAN_ID | None = None
+    #: Whether this Layer 2 Circuit is policed.
+    policer_enabled: bool
+    #: If policed, the bandwidth of the policer is stored.
+    bandwidth: BandwidthString | None = None

gso/products/product_blocks/service_binding_port.py

@@ -3,8 +3,6 @@
 A service binding port is used to logically attach an edge port to a customer service using a :term:`VLAN`.
 """
 
-from typing import Annotated
-
 from orchestrator.domain.base import ProductBlockModel
 from orchestrator.types import SubscriptionLifecycle
 from pydantic import Field
@@ -13,8 +11,7 @@ from gso.products.product_blocks.bgp_session import BGPSession, BGPSessionInacti
 from gso.products.product_blocks.edge_port import EdgePortBlock, EdgePortBlockInactive, EdgePortBlockProvisioning
 from gso.utils.shared_enums import SBPType
 from gso.utils.types.ip_address import IPv4AddressType, IPV4Netmask, IPv6AddressType, IPV6Netmask
-
-VLAN_ID = Annotated[int, Field(gt=0, lt=4096)]
+from gso.utils.types.virtual_identifiers import VLAN_ID
 
 
 class ServiceBindingPortInactive(

gso/products/product_types/layer_2_circuit.py

+"""Product type for a Layer 2 circuit."""
+
+from orchestrator.domain.base import SubscriptionModel
+from orchestrator.types import SubscriptionLifecycle
+
+from gso.products.product_blocks.layer_2_circuit import (
+    Layer2CircuitBlock,
+    Layer2CircuitBlockInactive,
+    Layer2CircuitBlockProvisioning,
+)
+
+
+class Layer2CircuitInactive(SubscriptionModel, is_base=True):
+    """An inactive Layer 2 Circuit."""
+
+    layer_2_circuit: Layer2CircuitBlockInactive
+
+
+class Layer2CircuitProvisioning(Layer2CircuitInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
+    """A Layer 2 Circuit that is provisioning."""
+
+    layer_2_circuit: Layer2CircuitBlockProvisioning
+
+
+class Layer2Circuit(Layer2CircuitProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
+    """An active Layer 2 Circuit."""
+
+    layer_2_circuit: Layer2CircuitBlock
+
+
+class ImportedLayer2CircuitInactive(SubscriptionModel, is_base=True):
+    """An imported, inactive Layer 2 Circuit."""
+
+    layer_2_circuit: Layer2CircuitBlockInactive
+
+
+class ImportedLayer2Circuit(
+    ImportedLayer2CircuitInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING, SubscriptionLifecycle.ACTIVE]
+):
+    """An imported Layer 2 Circuit."""
+
+    layer_2_circuit: Layer2CircuitBlock

gso/services/lso_client.py

@@ -10,11 +10,11 @@ from typing import Any, Literal, TypedDict
 import requests
 from orchestrator import step
 from orchestrator.config.assignee import Assignee
+from orchestrator.forms import FormPage
 from orchestrator.types import State
 from orchestrator.utils.errors import ProcessFailureError
 from orchestrator.workflow import Step, StepList, begin, callback_step, conditional, inputstep
 from pydantic import ConfigDict
-from pydantic_forms.core import FormPage
 from pydantic_forms.types import FormGenerator
 from pydantic_forms.validators import Label, LongText, ReadOnlyField
 

gso/utils/types/interfaces.py

@@ -100,3 +100,26 @@ class PhysicalPortCapacity(strEnum):
     TEN_GIGABIT_PER_SECOND = "10G"
     HUNDRED_GIGABIT_PER_SECOND = "100G"
     FOUR_HUNDRED_GIGABIT_PER_SECOND = "400G"
+
+
+def bandwidth_string_is_valid(bandwidth_string: str) -> str:
+    """Expect a bandwidth definition to follow the pattern of an int followed by a single letter.
+
+    If this string does not consist of a number followed by a single
+    """
+    msg = f"Expected a network capacity, e.g. 40G or 200M. Got: {bandwidth_string}"
+    if len(bandwidth_string) < 2:  # noqa: PLR2004 not a magic value
+        raise ValueError(msg)
+
+    if bandwidth_string[-1:] not in "K" "M" "G" "T":
+        raise ValueError(msg)
+
+    try:
+        int(bandwidth_string[:-1])  # Try parsing the bandwidth number
+    except ValueError as e:
+        raise ValueError(msg) from e
+
+    return bandwidth_string
+
+
+BandwidthString = Annotated[str, AfterValidator(bandwidth_string_is_valid)]

gso/utils/types/virtual_identifiers.py

+"""Annotated types for virtual identifiers such as :term:`VLAN` ID or Virtual Circuit ID."""
+
+from typing import Annotated
+
+from pydantic import Field
+from typing_extensions import Doc
+
+VLAN_ID = Annotated[int, Field(gt=0, lt=4096)]
+VC_ID = Annotated[
+    int,
+    Field(gt=0, le=2147483648),
+    Doc(
+        "A Virtual Circuit ID, the upper limit comes from the highest number that a service ID could be in Nokia srOS."
+    ),
+]

gso/utils/workflow_steps.py

@@ -5,12 +5,12 @@ from typing import Any
 
 from orchestrator import inputstep, step
 from orchestrator.config.assignee import Assignee
+from orchestrator.forms import FormPage
 from orchestrator.types import State, UUIDstr
 from orchestrator.utils.errors import ProcessFailureError
 from orchestrator.utils.json import json_dumps
 from orchestrator.workflow import StepList, conditional
 from pydantic import ConfigDict
-from pydantic_forms.core import FormPage
 from pydantic_forms.types import FormGenerator
 from pydantic_forms.validators import Label
 

gso/workflows/__init__.py

@@ -124,3 +124,6 @@ LazyWorkflowInstance(
 )
 LazyWorkflowInstance("gso.workflows.nren_l3_core_service.import_nren_l3_core_service", "import_nren_l3_core_service")
 LazyWorkflowInstance("gso.workflows.nren_l3_core_service.migrate_nren_l3_core_service", "migrate_nren_l3_core_service")
+
+# Layer 2 Circuit workflows
+LazyWorkflowInstance("gso.workflows.layer_2_circuit.create_layer_2_circuit", "create_layer_2_circuit")

gso/workflows/l2_circuit/__init__.py

+"""Workflows for Layer 2 Circuits."""

gso/workflows/l2_circuit/create_layer_2_circuit.py

+"""Workflow for creating a new Layer 2 Circuit."""
+
+from typing import Any
+from uuid import uuid4
+
+from orchestrator import step, workflow
+from orchestrator.forms import FormPage
+from orchestrator.targets import Target
+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 products.product_blocks.service_binding_port import ServiceBindingPortInactive
+from products.product_types.layer_2_circuit import Layer2CircuitInactive
+from pydantic import BaseModel, ConfigDict, Field
+from pydantic_forms.types import FormGenerator, State, UUIDstr
+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.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
+
+
+def initial_input_generator(product_name: str) -> FormGenerator:
+    """Gather input from the operator about a new Layer 2 Circuit subscription."""
+
+    class CreateLayer2CircuitServicePage(FormPage):
+        model_config = ConfigDict(title=f"{product_name}")
+
+        tt_number: TTNumber
+        partner: partner_choice()  # type: ignore[valid-type]
+        divider: Divider = Field(None, exclude=True)
+
+        layer_2_circuit_type: Layer2CircuitType
+        policer_enabled: bool = False
+
+    initial_user_input = yield CreateLayer2CircuitServicePage
+
+    class Layer2CircuitSideSelection(BaseModel):
+        edge_port: active_edge_port_selector(partner_id=initial_user_input.partner)  # type: ignore[valid-type]
+        divider: Divider = Field(None, exclude=True)
+
+        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):
+        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(int) = None
+            vlan_range_upper_bound: ReadOnlyField(int) = None
+
+        vlan_divider: Divider = Field(None, exclude=True)
+
+        if initial_user_input.policer_enabled:
+            policer_bandwidth: BandwidthString
+        else:
+            policer_bandwidth: ReadOnlyField(str) = None
+
+        policer_divider: Divider = Field(None, exclude=True)
+
+        layer_2_circuit_side_a: Layer2CircuitSideSelection
+        side_divider: Divider = Field(None, exclude=True)
+        layer_2_circuit_side_b: Layer2CircuitSideSelection
+
+    layer_2_circuit_input = yield Layer2CircuitServiceSidesPage
+
+    return {"product_name": product_name} | initial_user_input.model_dump() | layer_2_circuit_input.model_dump()
+
+
+@step("Create subscription")
+def create_subscription(product: UUIDstr, partner: str) -> State:
+    """Create a new subscription object in the database."""
+    subscription = Layer2CircuitInactive.from_product_id(product, partner)
+
+    return {"subscription": subscription, "subscription_id": subscription.subscription_id}
+
+
+@step("Initialize subscription")
+def initialize_subscription(
+    subscription: Layer2CircuitInactive,
+    layer_2_circuit_side_a: dict[str, Any],
+    layer_2_circuit_side_b: dict[str, Any],
+    virtual_circuit_id: VC_ID | None,
+    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,
+) -> 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))
+        for circuit_side in [layer_2_circuit_side_a, layer_2_circuit_side_b]
+    ]
+    subscription.layer_2_circuit.virtual_circuit_id = virtual_circuit_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
+
+    return {"subscription": subscription}
+
+
+@workflow(
+    "Create Layer 2 Circuit Service",
+    initial_input_form=wrap_create_initial_input_form(initial_input_generator),
+    target=Target.CREATE,
+)
+def create_layer_2_circuit() -> StepList:
+    """Create a new Layer 2 Circuit service subscription."""
+    return (
+        begin
+        >> create_subscription
+        >> store_process_subscription(Target.CREATE)
+        >> initialize_subscription
+        >> set_status(SubscriptionLifecycle.PROVISIONING)
+        >> resync
+        >> done
+    )

gso/workflows/nren_l3_core_service/create_imported_nren_l3_core_service.py

@@ -14,13 +14,14 @@ from pydantic_forms.types import UUIDstr
 from gso.products import ProductName
 from gso.products.product_blocks.bgp_session import BGPSession, IPFamily
 from gso.products.product_blocks.nren_l3_core_service import NRENAccessPortInactive
-from gso.products.product_blocks.service_binding_port import VLAN_ID, ServiceBindingPortInactive
+from gso.products.product_blocks.service_binding_port import ServiceBindingPortInactive
 from gso.products.product_types.edge_port import EdgePort
 from gso.products.product_types.nren_l3_core_service import ImportedNRENL3CoreServiceInactive, NRENL3CoreServiceType
 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.ip_address import IPAddress, IPv4AddressType, IPV4Netmask, IPv6AddressType, IPV6Netmask
+from gso.utils.types.virtual_identifiers import VLAN_ID
 
 
 def initial_input_form_generator() -> FormGenerator:

gso/workflows/nren_l3_core_service/create_nren_l3_core_service.py

@@ -15,7 +15,7 @@ from pydantic_forms.validators import Divider
 
 from gso.products.product_blocks.bgp_session import BGPSession, IPFamily
 from gso.products.product_blocks.nren_l3_core_service import NRENAccessPortInactive
-from gso.products.product_blocks.service_binding_port import VLAN_ID, ServiceBindingPortInactive
+from gso.products.product_blocks.service_binding_port import ServiceBindingPortInactive
 from gso.products.product_types.edge_port import EdgePort
 from gso.products.product_types.nren_l3_core_service import NRENL3CoreService, NRENL3CoreServiceInactive
 from gso.services.lso_client import LSOState, lso_interaction
@@ -26,6 +26,7 @@ from gso.utils.helpers import (
 from gso.utils.shared_enums import APType, SBPType
 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 VLAN_ID
 
 
 def initial_input_form_generator(product_name: str) -> FormGenerator:

gso/workflows/nren_l3_core_service/migrate_nren_l3_core_service.py

@@ -4,12 +4,12 @@ from typing import Annotated
 
 from annotated_types import Len
 from orchestrator import workflow
+from orchestrator.forms import FormPage
 from orchestrator.targets import Target
 from orchestrator.workflow import StepList, begin, done, step
 from orchestrator.workflows.steps import resync, store_process_subscription, unsync
 from orchestrator.workflows.utils import wrap_modify_initial_input_form
 from pydantic import AfterValidator, BaseModel, ConfigDict, Field
-from pydantic_forms.core import FormPage
 from pydantic_forms.types import FormGenerator, State, UUIDstr
 from pydantic_forms.validators import Choice, Divider
 

gso/workflows/nren_l3_core_service/modify_nren_l3_core_service.py

@@ -16,12 +16,13 @@ from pydantic_forms.validators import Divider, Label
 
 from gso.products.product_blocks.bgp_session import BGPSession, IPFamily
 from gso.products.product_blocks.nren_l3_core_service import NRENAccessPort
-from gso.products.product_blocks.service_binding_port import VLAN_ID, ServiceBindingPort
+from gso.products.product_blocks.service_binding_port import ServiceBindingPort
 from gso.products.product_types.edge_port import EdgePort
 from gso.products.product_types.nren_l3_core_service import NRENL3CoreService
 from gso.utils.helpers import active_edge_port_selector
 from gso.utils.shared_enums import APType, SBPType
 from gso.utils.types.ip_address import IPv4AddressType, IPV4Netmask, IPv6AddressType, IPV6Netmask
+from gso.utils.types.virtual_identifiers import VLAN_ID
 
 
 def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:

test/fixtures.py

@@ -5,9 +5,9 @@ from uuid import uuid4
 import pytest
 from orchestrator import step, workflow
 from orchestrator.config.assignee import Assignee
+from orchestrator.forms import FormPage
 from orchestrator.types import UUIDstr
 from orchestrator.workflow import done, init, inputstep
-from pydantic_forms.core import FormPage
 from pydantic_forms.types import FormGenerator
 from pydantic_forms.validators import Choice