diff --git a/test/conftest.py b/test/conftest.py index 3211217acb46436f6287aa5e48fe6393493c3283..888926fc6aa1119e4781cbcae0f64dd3f0cfca9e 100644 --- a/test/conftest.py +++ b/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, diff --git a/test/fixtures/__init__.py b/test/fixtures/__init__.py index 3f183fb2e8778f758d7bba96a459d1df29ac717b..1fa5c4cf587c59a80b5838d6729524e44fc27c6a 100644 --- a/test/fixtures/__init__.py +++ b/test/fixtures/__init__.py @@ -1,6 +1,7 @@ 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", ] diff --git a/test/fixtures/layer_2_circuit_fixtures.py b/test/fixtures/layer_2_circuit_fixtures.py new file mode 100644 index 0000000000000000000000000000000000000000..b0e73227f7305d0aacf1e3f42a8c196969221a86 --- /dev/null +++ b/test/fixtures/layer_2_circuit_fixtures.py @@ -0,0 +1,116 @@ +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