-
Neda Moeini authoredNeda Moeini authored
layer_2_circuit_fixtures.py 5.93 KiB
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 EdgePort, ProductName
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,
policer_burst_rate: 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
if policer_enabled:
subscription.layer_2_circuit.bandwidth = policer_bandwidth or faker.bandwidth()
subscription.layer_2_circuit.policer_burst_rate = policer_burst_rate or faker.bandwidth()
else:
subscription.layer_2_circuit.bandwidth = None
subscription.layer_2_circuit.policer_burst_rate = None
subscription.description = description or (
f"{subscription.product.name} - " f"{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