edge_port_fixtures.py

import pytest
from orchestrator.db import (
    db,
)
from orchestrator.domain import SubscriptionModel
from orchestrator.types import SubscriptionLifecycle, UUIDstr

from gso.products import ProductName, Router
from gso.products.product_blocks.edge_port import (
    EdgePortAEMemberBlock,
    EdgePortType,
    EncapsulationType,
)
from gso.products.product_types.edge_port import EdgePortInactive, ImportedEdgePortInactive
from gso.services import subscriptions
from gso.utils.types.interfaces import PhysicalPortCapacity


@pytest.fixture()
def edge_port_subscription_factory(faker, partner_factory, nokia_router_subscription_factory):
    def subscription_create(
        description=None,
        partner: dict | None = None,
        start_date="2023-05-24T00:00:00+00:00",
        node=None,
        name=None,
        edge_port_description=None,
        encapsulation=EncapsulationType.DOT1Q,
        mac_address=None,
        member_speed=PhysicalPortCapacity.HUNDRED_GIGABIT_PER_SECOND,
        minimum_links=None,
        edge_port_type=EdgePortType.PUBLIC,
        geant_ga_id=None,
        edge_port_ae_members=None,
        status: SubscriptionLifecycle | None = None,
        *,
        enable_lacp=True,
        ignore_if_down=False,
        is_imported=True,
    ) -> UUIDstr:
        partner = partner or partner_factory()
        edge_port_node = Router.from_subscription(nokia_router_subscription_factory()).router
        if is_imported:
            product_id = subscriptions.get_product_id_by_name(ProductName.EDGE_PORT)
            edge_port_subscription = EdgePortInactive.from_product_id(
                product_id, customer_id=partner["partner_id"], insync=True
            )
        else:
            product_id = subscriptions.get_product_id_by_name(ProductName.IMPORTED_EDGE_PORT)
            edge_port_subscription = ImportedEdgePortInactive.from_product_id(
                product_id, customer_id=partner["partner_id"], insync=True
            )

        edge_port_subscription.edge_port.edge_port_description = description or faker.text(max_nb_chars=30)
        edge_port_subscription.edge_port.edge_port_geant_ga_id = geant_ga_id or faker.geant_sid()
        edge_port_subscription.edge_port.edge_port_node = node or edge_port_node
        edge_port_subscription.edge_port.edge_port_name = name or f"lag-{faker.pyint(21, 50)}"
        edge_port_subscription.edge_port.edge_port_description = edge_port_description or faker.sentence()
        edge_port_subscription.edge_port.edge_port_enable_lacp = enable_lacp
        edge_port_subscription.edge_port.edge_port_encapsulation = encapsulation
        edge_port_subscription.edge_port.edge_port_mac_address = mac_address or faker.mac_address()
        edge_port_subscription.edge_port.edge_port_member_speed = member_speed
        edge_port_subscription.edge_port.edge_port_minimum_links = minimum_links or faker.pyint(1, 2)
        edge_port_subscription.edge_port.edge_port_type = edge_port_type
        edge_port_subscription.edge_port.edge_port_ignore_if_down = ignore_if_down
        edge_port_subscription.edge_port.edge_port_geant_ga_id = geant_ga_id
        edge_port_subscription.edge_port.edge_port_ae_members = edge_port_ae_members or [
            EdgePortAEMemberBlock.new(
                faker.uuid4(),
                interface_name="Interface2",
                interface_description=faker.sentence(),
            ),
            EdgePortAEMemberBlock.new(
                faker.uuid4(),
                interface_name="Interface3",
                interface_description=faker.sentence(),
            ),
        ]
        edge_port_subscription = SubscriptionModel.from_other_lifecycle(
            edge_port_subscription,
            SubscriptionLifecycle.ACTIVE,
        )

        if status:
            edge_port_subscription.status = status

        edge_port_subscription.description = description or faker.text(max_nb_chars=30)
        edge_port_subscription.start_date = start_date
        edge_port_subscription.save()
        db.session.commit()

        return str(edge_port_subscription.subscription_id)

    return subscription_create