Add fixture for L2Circuit creation

test/conftest.py

@@ -41,6 +41,7 @@ from test.fixtures import (  # noqa: F401
     iptrunk_side_subscription_factory,
     iptrunk_subscription_factory,
     lan_switch_interconnect_subscription_factory,
+    layer_2_circuit_subscription_factory,
     nren_access_port_factory,
     nren_l3_core_service_subscription_factory,
     office_router_subscription_factory,

test/fixtures/__init__.py

 from test.fixtures.edge_port_fixtures import edge_port_subscription_factory
 from test.fixtures.iptrunk_fixtures import iptrunk_side_subscription_factory, iptrunk_subscription_factory
 from test.fixtures.lan_switch_interconnect_fixtures import lan_switch_interconnect_subscription_factory
+from test.fixtures.layer_2_circuit_fixtures import layer_2_circuit_subscription_factory
 from test.fixtures.nren_l3_core_service_fixtures import (
     bgp_session_subscription_factory,
     nren_access_port_factory,
@@ -29,4 +30,5 @@ __all__ = [
     "site_subscription_factory",
     "super_pop_switch_subscription_factory",
     "switch_subscription_factory",
+    "layer_2_circuit_subscription_factory",
 ]

test/fixtures/layer_2_circuit_fixtures.py

+from uuid import uuid4
+
+import pytest
+from orchestrator.db import db
+from orchestrator.domain import SubscriptionModel
+from orchestrator.types import SubscriptionLifecycle, UUIDstr
+
+from gso.products import ProductName, EdgePort
+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.layer_2_circuit import (
+    ImportedLayer2Circuit,
+    Layer2CircuitInactive,
+    Layer2CircuitServiceType,
+)
+from gso.services import subscriptions
+from gso.utils.helpers import generate_unique_vc_id
+from gso.utils.shared_enums import SBPType
+from gso.utils.types.interfaces import BandwidthString
+from gso.utils.types.virtual_identifiers import VLAN_ID
+
+
+@pytest.fixture()
+def layer_2_circuit_subscription_factory(faker, geant_partner, edge_port_subscription_factory):
+    def create_subscription(
+            description: str | None = None,
+            partner: dict | None = None,
+            status: SubscriptionLifecycle | None = None,
+            start_date: str | None = "2024-01-01T10:20:30+01:02",
+            layer_2_circuit_service_type: Layer2CircuitServiceType | None = None,
+            layer_2_circuit_type: Layer2CircuitType = Layer2CircuitType.TAGGED,
+            vlan_range_lower_bound: VLAN_ID | None = None,
+            vlan_range_upper_bound: VLAN_ID | None = None,
+            policer_bandwidth: BandwidthString | None = None,
+            layer_2_circuit_side_a_edgeport: UUIDstr | None = None,
+            vlan_id_side_a: VLAN_ID | None = None,
+            layer_2_circuit_side_b_edgeport: UUIDstr | None = None,
+            vlan_id_side_b: VLAN_ID | None = None,
+            geant_sid: str | None = None,
+            *,
+            policer_enabled: bool = False,
+    ) -> UUIDstr:
+
+        # Assign default partner if none provided
+        if partner is None:
+            partner = geant_partner
+
+        # Select subscription type based on service type
+        match layer_2_circuit_service_type:
+            case Layer2CircuitServiceType.GEANT_PLUS:
+                product_id = subscriptions.get_product_id_by_name(ProductName.GEANT_PLUS)
+                subscription = Layer2CircuitInactive.from_product_id(
+                    product_id, customer_id=partner["partner_id"], insync=True
+                )
+            case Layer2CircuitServiceType.IMPORTED_GEANT_PLUS:
+                product_id = subscriptions.get_product_id_by_name(ProductName.IMPORTED_GEANT_PLUS)
+                subscription = ImportedLayer2Circuit.from_product_id(
+                    product_id, customer_id=partner["partner_id"], insync=True
+                )
+            case Layer2CircuitServiceType.EXPRESSROUTE:
+                product_id = subscriptions.get_product_id_by_name(ProductName.EXPRESSROUTE)
+                subscription = Layer2CircuitInactive.from_product_id(
+                    product_id, customer_id=partner["partner_id"], insync=True
+                )
+            case Layer2CircuitServiceType.IMPORTED_EXPRESSROUTE:
+                product_id = subscriptions.get_product_id_by_name(ProductName.IMPORTED_EXPRESSROUTE)
+                subscription = ImportedLayer2Circuit.from_product_id(
+                    product_id, customer_id=partner["partner_id"], insync=True
+                )
+            case _:
+                err = ValueError(f"Layer 2 Circuit Service type not found: {layer_2_circuit_service_type}")
+                raise ValueError(err)
+
+        layer_2_circuit_sides = []
+        for edge_port, vlan_id in [
+            (layer_2_circuit_side_a_edgeport or edge_port_subscription_factory(), vlan_id_side_a or faker.vlan_id()),
+            (layer_2_circuit_side_b_edgeport or edge_port_subscription_factory(), vlan_id_side_b or faker.vlan_id()),
+        ]:
+            sbp = ServiceBindingPortInactive.new(
+                uuid4(),
+                edge_port=EdgePort.from_subscription(edge_port).edge_port,
+                sbp_type=SBPType.L2,
+                vlan_id=vlan_id,
+                geant_sid=geant_sid or faker.geant_sid(),
+                is_tagged=layer_2_circuit_type == Layer2CircuitType.TAGGED,
+                custom_firewall_filters=False,
+            )
+            layer_2_circuit_side = Layer2CircuitSideBlockInactive.new(uuid4(), sbp=sbp)
+            layer_2_circuit_sides.append(layer_2_circuit_side)
+
+        subscription.layer_2_circuit.layer_2_circuit_sides = layer_2_circuit_sides
+        subscription.layer_2_circuit.virtual_circuit_id = generate_unique_vc_id()
+        subscription.layer_2_circuit.layer_2_circuit_type = layer_2_circuit_type
+        if layer_2_circuit_type == Layer2CircuitType.TAGGED:
+            subscription.layer_2_circuit.vlan_range_lower_bound = vlan_range_lower_bound or faker.vlan_id()
+            subscription.layer_2_circuit.vlan_range_upper_bound = vlan_range_upper_bound or faker.vlan_id()
+        else:
+            subscription.layer_2_circuit.vlan_range_lower_bound = None
+            subscription.layer_2_circuit.vlan_range_upper_bound = None
+
+        subscription.layer_2_circuit.policer_enabled = policer_enabled
+        subscription.layer_2_circuit.bandwidth = policer_bandwidth if policer_enabled else None
+        subscription.description = description or f"{subscription.product.name} - {subscription.layer_2_circuit.virtual_circuit_id}"
+
+        subscription = SubscriptionModel.from_other_lifecycle(subscription, SubscriptionLifecycle.ACTIVE)
+        subscription.insync = True
+        subscription.start_date = start_date
+        if status:
+            subscription.status = status
+
+        subscription.save()
+        db.session.commit()
+
+        return str(subscription.subscription_id)
+
+    return create_subscription