test_helpers.py

from unittest.mock import patch

import pytest
from orchestrator.types import SubscriptionLifecycle

from gso.annotations.tt_number import validate_tt_number
from gso.products import Router
from gso.products.product_blocks.iptrunk import IptrunkInterfaceBlock
from gso.products.product_blocks.router import RouterRole
from gso.utils.helpers import (
    available_interfaces_choices_including_current_members,
    generate_inventory_for_active_routers,
)
from gso.utils.shared_enums import Vendor
from gso.utils.types.tt_number import validate_tt_number


@pytest.fixture()
def mock_router():
    """Fixture to mock the Router class."""
    with patch("gso.utils.helpers.Router") as mock:
        yield mock


@pytest.fixture()
def mock_netbox_client():
    """Fixture to mock the NetboxClient class."""
    with patch("gso.utils.helpers.NetboxClient") as mock:
        yield mock


@pytest.fixture()
def generate_tt_numbers(faker, request):
    """Get a Generator for valid and invalid tt numbers."""
    valid_count = request.param.get("valid", 0)
    invalid_count = request.param.get("invalid", 0)

    valid_data = [(faker.tt_number(), True) for _ in range(valid_count)]
    invalid_data = [(faker.sentence(), False) for _ in range(invalid_count)]

    return valid_data + invalid_data


def test_non_nokia_router_returns_none(mock_router, faker):
    mock_router.from_subscription.return_value.router.vendor = Vendor.JUNIPER
    result = available_interfaces_choices_including_current_members(faker.uuid4(), "10G", [])
    assert result is None


def test_nokia_router_with_no_interfaces_returns_empty_choice(mock_router, mock_netbox_client, faker):
    mock_router.from_subscription.return_value.router.vendor = Vendor.NOKIA
    mock_netbox_client().get_available_interfaces.return_value = iter([])
    result = available_interfaces_choices_including_current_members(faker.uuid4(), "10G", [])
    assert len(result) == 0


def test_nokia_router_with_interfaces_returns_choice(mock_router, mock_netbox_client, faker):
    mock_router.from_subscription.return_value.router.vendor = Vendor.NOKIA
    mock_netbox_client().get_available_interfaces.return_value = iter(
        [
            {"name": "interface1", "module": {"display": "module1"}, "description": "desc1"},
            {"name": "interface2", "module": {"display": "module2"}, "description": "desc2"},
        ],
    )
    mock_netbox_client().get_interface_by_name_and_device.return_value = {
        "name": "interface3",
        "module": {"display": "module3"},
        "description": "desc3",
    }
    interfaces = [
        IptrunkInterfaceBlock(
            interface_name="interface3",
            interface_description="desc3",
            owner_subscription_id=faker.uuid4(),
            subscription_instance_id=faker.uuid4(),
        ),
    ]

    result = available_interfaces_choices_including_current_members(faker.uuid4(), "10G", interfaces)

    assert len(result) == 3
    assert hasattr(result, "interface1")
    assert hasattr(result, "interface2")
    assert hasattr(result, "interface3")


@pytest.mark.parametrize("generate_tt_numbers", [{"valid": 5, "invalid": 3}], indirect=True)
def test_tt_number(generate_tt_numbers):
    """Test different TT numbers."""
    for tt_number, is_valid in generate_tt_numbers:
        if is_valid:
            assert validate_tt_number(tt_number) == tt_number
        else:
            err_msg = (
                f"The given TT number: {tt_number} is not valid. "
                f" A valid TT number starts with 'TT#' followed by 16 digits."
            )

            with pytest.raises(ValueError, match=err_msg):
                validate_tt_number(tt_number)


def test_generate_inventory_for_active_routers_with_single_active_router(nokia_router_subscription_factory):
    """Test the generation of inventory for a single active P router."""
    router = Router.from_subscription(nokia_router_subscription_factory(router_role=RouterRole.P))
    expected_result = {
        "all": {
            "hosts": {
                router.router.router_fqdn: {
                    "lo4": str(router.router.router_lo_ipv4_address),
                    "lo6": str(router.router.router_lo_ipv6_address),
                    "vendor": str(router.router.vendor),
                }
            }
        }
    }
    assert generate_inventory_for_active_routers(RouterRole.P) == expected_result


def test_generate_inventory_for_active_routers_with_multiple_routers(nokia_router_subscription_factory):
    """Test the generation of inventory for multiple active P and PE routers"""
    for _ in range(5):
        nokia_router_subscription_factory(router_role=RouterRole.P)
    for _ in range(3):
        nokia_router_subscription_factory(router_role=RouterRole.PE)
    nokia_router_subscription_factory(status=SubscriptionLifecycle.TERMINATED)
    nokia_router_subscription_factory(status=SubscriptionLifecycle.INITIAL)
    #  Test the generation of inventory for multiple active P routers.
    inventory = generate_inventory_for_active_routers(RouterRole.P)
    assert len(inventory["all"]["hosts"]) == 5
    inventory = generate_inventory_for_active_routers(RouterRole.PE)
    assert len(inventory["all"]["hosts"]) == 3


def test_generate_inventory_for_active_routers_with_excluded_router(nokia_router_subscription_factory):
    """Test the generation of inventory for active P routers with an excluded router."""
    for _ in range(5):
        nokia_router_subscription_factory(router_role=RouterRole.P)
    router = nokia_router_subscription_factory(router_role=RouterRole.P)
    excluded_routers = [Router.from_subscription(router).router.router_fqdn]
    inventory = generate_inventory_for_active_routers(RouterRole.P, exclude_routers=excluded_routers)
    assert len(inventory["all"]["hosts"]) == 5  # 6 P routers, the last one is excluded, so 5 P routers are left.