geant_ip_fixtures.py

import random
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.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 ServiceBindingPort
from gso.products.product_types.nren_l3_core_service import ImportedNRENL3CoreService, NRENL3CoreServiceInactive
from gso.services import subscriptions
from gso.utils.shared_enums import APType, SBPType
from gso.utils.types.ip_address import IPAddress


@pytest.fixture()
def bgp_session_subscription_factory(faker):
    def create_bgp_session(
        peer_address: IPAddress | None = None,
        bfd_interval: int = 2,
        bfd_multiplier: int = 2,
        families: list[IPFamily] | None = None,
        authentication_key: str | None = None,
        *,
        is_multi_hop: bool = False,
        has_custom_policies: bool = False,
        bfd_enabled: bool = True,
        multipath_enabled: bool | None = True,
        send_default_route: bool | None = True,
        is_passive: bool | None = False,
        rtbh_enabled: bool | None = False,
    ):
        return BGPSession.new(
            subscription_id=uuid4(),
            peer_address=peer_address or faker.ipv4(),
            bfd_enabled=bfd_enabled,
            families=families or [IPFamily.V4UNICAST],
            has_custom_policies=has_custom_policies,
            authentication_key=authentication_key or faker.password(),
            multipath_enabled=multipath_enabled,
            send_default_route=send_default_route,
            is_multi_hop=is_multi_hop,
            bfd_interval=bfd_interval,
            bfd_multiplier=bfd_multiplier,
            rtbh_enabled=rtbh_enabled,
            is_passive=is_passive,
        )

    return create_bgp_session


@pytest.fixture()
def service_binding_port_factory(faker, bgp_session_subscription_factory, edge_port_subscription_factory):
    def create_service_binding_port(
        sbp_bgp_session_list: list | None = None,
        geant_sid: str | None = None,
        sbp_type: SBPType = SBPType.L3,
        ipv4_address: str | None = None,
        ipv4_mask: int | None = None,
        ipv6_address: str | None = None,
        ipv6_mask: int | None = None,
        vlan_id: int | None = None,
        edge_port: EdgePort | None = None,
        *,
        custom_firewall_filters: bool = False,
        is_tagged: bool = False,
    ):
        return ServiceBindingPort.new(
            subscription_id=uuid4(),
            is_tagged=is_tagged,
            vlan_id=vlan_id or faker.pyint(min_value=1, max_value=4096),
            sbp_type=sbp_type,
            ipv4_address=ipv4_address or faker.ipv4(),
            ipv4_mask=ipv4_mask or faker.ipv4_netmask(),
            ipv6_address=ipv6_address or faker.ipv6(),
            ipv6_mask=ipv6_mask or faker.ipv6_netmask(),
            custom_firewall_filters=custom_firewall_filters,
            geant_sid=geant_sid or faker.geant_sid(),
            sbp_bgp_session_list=sbp_bgp_session_list
            or [
                bgp_session_subscription_factory(families=[IPFamily.V4UNICAST]),
                bgp_session_subscription_factory(families=[IPFamily.V6UNICAST], peer_address=faker.ipv6()),
            ],
            edge_port=edge_port or EdgePort.from_subscription(edge_port_subscription_factory()).edge_port,
        )

    return create_service_binding_port


@pytest.fixture()
def nren_access_port_factory(faker, service_binding_port_factory):
    def create_nren_access_port(
        nren_ap_type: APType | None = None,
        service_binding_port: ServiceBindingPort | None = None,
    ):
        return NRENAccessPort.new(
            subscription_id=uuid4(),
            nren_ap_type=nren_ap_type or random.choice(list(APType)),  # noqa: S311
            geant_ip_sbp=service_binding_port or service_binding_port_factory(),
        )

    return create_nren_access_port


@pytest.fixture()
def geant_ip_subscription_factory(
    faker,
    partner_factory,
    nren_access_port_factory,
):
    def create_geant_ip_subscription(
        description=None,
        partner: dict | None = None,
        nren_ap_list: list[NRENAccessPort] | None = None,
        start_date="2023-05-24T00:00:00+00:00",
        status: SubscriptionLifecycle | None = None,
        *,
        is_imported=True,
    ) -> UUIDstr:
        partner = partner or partner_factory()
        if is_imported:
            product_id = subscriptions.get_product_id_by_name(ProductName.GEANT_IP)
            geant_ip_subscription = NRENL3CoreServiceInactive.from_product_id(
                product_id, customer_id=partner["partner_id"], insync=True
            )
        else:
            product_id = subscriptions.get_product_id_by_name(ProductName.IMPORTED_GEANT_IP)
            geant_ip_subscription = ImportedNRENL3CoreService.from_product_id(
                product_id, customer_id=partner["partner_id"], insync=True
            )

        # Default nren_ap_list creation with primary and backup access ports
        geant_ip_subscription.geant_ip.geant_ip_ap_list = nren_ap_list or [
            nren_access_port_factory(nren_ap_type=APType.PRIMARY),
            nren_access_port_factory(nren_ap_type=APType.BACKUP),
        ]

        # Update subscription with description, start date, and status
        geant_ip_subscription = SubscriptionModel.from_other_lifecycle(
            geant_ip_subscription,
            SubscriptionLifecycle.ACTIVE,
        )
        geant_ip_subscription.description = description or faker.sentence()
        geant_ip_subscription.start_date = start_date
        geant_ip_subscription.status = status or SubscriptionLifecycle.ACTIVE
        geant_ip_subscription.save()

        db.session.commit()

        return str(geant_ip_subscription.subscription_id)

    return create_geant_ip_subscription