From 2144fa258db7b727e6d0855ac88d4214af53ac09 Mon Sep 17 00:00:00 2001
From: Karel van Klink <karel.vanklink@geant.org>
Date: Mon, 28 Oct 2024 09:30:27 +0100
Subject: [PATCH] Add Layer 2 Circuit product blocks and creation workflow
---
.../product_blocks/layer_2_circuit.py | 110 ++++++++++++++
.../product_blocks/service_binding_port.py | 5 +-
gso/products/product_types/layer_2_circuit.py | 42 ++++++
gso/services/lso_client.py | 2 +-
gso/utils/types/interfaces.py | 23 +++
gso/utils/types/virtual_identifiers.py | 15 ++
gso/utils/workflow_steps.py | 2 +-
gso/workflows/__init__.py | 3 +
gso/workflows/l2_circuit/__init__.py | 1 +
.../l2_circuit/create_layer_2_circuit.py | 134 ++++++++++++++++++
.../create_imported_nren_l3_core_service.py | 3 +-
.../create_nren_l3_core_service.py | 3 +-
.../migrate_nren_l3_core_service.py | 2 +-
.../modify_nren_l3_core_service.py | 3 +-
test/fixtures.py | 2 +-
15 files changed, 339 insertions(+), 11 deletions(-)
create mode 100644 gso/products/product_blocks/layer_2_circuit.py
create mode 100644 gso/products/product_types/layer_2_circuit.py
create mode 100644 gso/utils/types/virtual_identifiers.py
create mode 100644 gso/workflows/l2_circuit/__init__.py
create mode 100644 gso/workflows/l2_circuit/create_layer_2_circuit.py
diff --git a/gso/products/product_blocks/layer_2_circuit.py b/gso/products/product_blocks/layer_2_circuit.py
new file mode 100644
index 00000000..854d58d9
--- /dev/null
+++ b/gso/products/product_blocks/layer_2_circuit.py
@@ -0,0 +1,110 @@
+"""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
diff --git a/gso/products/product_blocks/service_binding_port.py b/gso/products/product_blocks/service_binding_port.py
index 54098382..bb90fcc1 100644
--- a/gso/products/product_blocks/service_binding_port.py
+++ b/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(
diff --git a/gso/products/product_types/layer_2_circuit.py b/gso/products/product_types/layer_2_circuit.py
new file mode 100644
index 00000000..cbb98ee4
--- /dev/null
+++ b/gso/products/product_types/layer_2_circuit.py
@@ -0,0 +1,42 @@
+"""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
diff --git a/gso/services/lso_client.py b/gso/services/lso_client.py
index cae5dd9c..f501ed70 100644
--- a/gso/services/lso_client.py
+++ b/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
diff --git a/gso/utils/types/interfaces.py b/gso/utils/types/interfaces.py
index 15c91167..2251b49e 100644
--- a/gso/utils/types/interfaces.py
+++ b/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)]
diff --git a/gso/utils/types/virtual_identifiers.py b/gso/utils/types/virtual_identifiers.py
new file mode 100644
index 00000000..2208ca74
--- /dev/null
+++ b/gso/utils/types/virtual_identifiers.py
@@ -0,0 +1,15 @@
+"""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."
+ ),
+]
diff --git a/gso/utils/workflow_steps.py b/gso/utils/workflow_steps.py
index 9b82b69e..6979d6aa 100644
--- a/gso/utils/workflow_steps.py
+++ b/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
diff --git a/gso/workflows/__init__.py b/gso/workflows/__init__.py
index 362f96ab..9e141d64 100644
--- a/gso/workflows/__init__.py
+++ b/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")
diff --git a/gso/workflows/l2_circuit/__init__.py b/gso/workflows/l2_circuit/__init__.py
new file mode 100644
index 00000000..53538548
--- /dev/null
+++ b/gso/workflows/l2_circuit/__init__.py
@@ -0,0 +1 @@
+"""Workflows for Layer 2 Circuits."""
diff --git a/gso/workflows/l2_circuit/create_layer_2_circuit.py b/gso/workflows/l2_circuit/create_layer_2_circuit.py
new file mode 100644
index 00000000..eb18b7b0
--- /dev/null
+++ b/gso/workflows/l2_circuit/create_layer_2_circuit.py
@@ -0,0 +1,134 @@
+"""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
+ )
diff --git a/gso/workflows/nren_l3_core_service/create_imported_nren_l3_core_service.py b/gso/workflows/nren_l3_core_service/create_imported_nren_l3_core_service.py
index f5a2e2ea..35ab8563 100644
--- a/gso/workflows/nren_l3_core_service/create_imported_nren_l3_core_service.py
+++ b/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:
diff --git a/gso/workflows/nren_l3_core_service/create_nren_l3_core_service.py b/gso/workflows/nren_l3_core_service/create_nren_l3_core_service.py
index 2ac1708d..be55ae6a 100644
--- a/gso/workflows/nren_l3_core_service/create_nren_l3_core_service.py
+++ b/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:
diff --git a/gso/workflows/nren_l3_core_service/migrate_nren_l3_core_service.py b/gso/workflows/nren_l3_core_service/migrate_nren_l3_core_service.py
index 5d42bccd..6ecd489b 100644
--- a/gso/workflows/nren_l3_core_service/migrate_nren_l3_core_service.py
+++ b/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
diff --git a/gso/workflows/nren_l3_core_service/modify_nren_l3_core_service.py b/gso/workflows/nren_l3_core_service/modify_nren_l3_core_service.py
index 4331de2b..afb7d69b 100644
--- a/gso/workflows/nren_l3_core_service/modify_nren_l3_core_service.py
+++ b/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:
diff --git a/test/fixtures.py b/test/fixtures.py
index e2b905c1..da10b0d1 100644
--- a/test/fixtures.py
+++ b/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
--
GitLab