test_imports.py

from unittest.mock import patch
from uuid import uuid4

import pytest
from orchestrator.db import SubscriptionTable
from orchestrator.services import subscriptions

from gso.products.product_blocks.iptrunk import IptrunkType, PhyPortCapacity
from gso.products.product_blocks.router import RouterRole, RouterVendor
from gso.products.product_blocks.site import SiteTier
from gso.utils.helpers import iso_from_ipv4

SITE_IMPORT_ENDPOINT = "/api/v1/imports/sites"
ROUTER_IMPORT_ENDPOINT = "/api/v1/imports/routers"
IPTRUNK_IMPORT_API_URL = "/api/v1/imports/iptrunks"


@pytest.fixture()
def iptrunk_data(nokia_router_subscription_factory, faker):
    router_side_a = nokia_router_subscription_factory()
    router_side_b = nokia_router_subscription_factory()
    return {
        "customer": "GÉANT",
        "geant_s_sid": faker.geant_sid(),
        "iptrunk_type": IptrunkType.DARK_FIBER,
        "iptrunk_description": faker.sentence(),
        "iptrunk_speed": PhyPortCapacity.HUNDRED_GIGABIT_PER_SECOND,
        "iptrunk_minimum_links": 5,
        "iptrunk_isis_metric": 500,
        "side_a_node_id": router_side_a,
        "side_a_ae_iface": faker.network_interface(),
        "side_a_ae_geant_a_sid": faker.geant_sid(),
        "side_a_ae_members": [
            {
                "interface_name": faker.network_interface(),
                "interface_description": faker.sentence(),
            }
            for _ in range(5)
        ],
        "side_b_node_id": router_side_b,
        "side_b_ae_iface": faker.network_interface(),
        "side_b_ae_geant_a_sid": faker.geant_sid(),
        "side_b_ae_members": [
            {
                "interface_name": faker.network_interface(),
                "interface_description": faker.sentence(),
            }
            for _ in range(5)
        ],
        "iptrunk_ipv4_network": str(faker.ipv4(network=True)),
        "iptrunk_ipv6_network": str(faker.ipv6(network=True)),
    }


@pytest.fixture()
def mock_routers(iptrunk_data):
    with patch("gso.services.subscriptions.get_active_router_subscriptions") as mock_get_active_router_subscriptions:

        def _active_router_subscriptions(*args, **kwargs):
            if kwargs["includes"] == ["subscription_id", "description"]:
                return [
                    {
                        "subscription_id": iptrunk_data["side_a_node_id"],
                        "description": "iptrunk_sideA_node_id description",
                    },
                    {
                        "subscription_id": iptrunk_data["side_b_node_id"],
                        "description": "iptrunk_sideB_node_id description",
                    },
                    {
                        "subscription_id": str(uuid4()),
                        "description": "random description",
                    },
                ]
            return [
                {"subscription_id": iptrunk_data["side_a_node_id"]},
                {"subscription_id": iptrunk_data["side_b_node_id"]},
                {"subscription_id": str(uuid4())},
            ]

        mock_get_active_router_subscriptions.side_effect = _active_router_subscriptions
        yield mock_get_active_router_subscriptions


@patch("gso.api.v1.imports._start_process")
def test_import_iptrunk_successful_with_mocked_process(mock_start_process, test_client, mock_routers, iptrunk_data):
    mock_start_process.return_value = "123e4567-e89b-12d3-a456-426655440000"
    response = test_client.post(IPTRUNK_IMPORT_API_URL, json=iptrunk_data)

    assert response.status_code == 201
    assert response.json()["pid"] == "123e4567-e89b-12d3-a456-426655440000"


@pytest.fixture()
def site_data(faker):
    return {
        "site_name": faker.site_name(),
        "site_city": faker.city(),
        "site_country": faker.country(),
        "site_country_code": faker.country_code(),
        "site_latitude": float(faker.latitude()),
        "site_longitude": float(faker.longitude()),
        "site_bgp_community_id": faker.pyint(),
        "site_internal_id": faker.pyint(),
        "site_tier": SiteTier.TIER1,
        "site_ts_address": faker.ipv4(),
        "customer": "GÉANT",
    }


@pytest.fixture()
def router_data(faker, site_data):
    mock_ipv4 = faker.ipv4()
    return {
        "hostname": "127.0.0.1",
        "router_role": RouterRole.PE,
        "router_vendor": RouterVendor.JUNIPER,
        "router_site": site_data["site_name"],
        "ts_port": 1234,
        "customer": "GÉANT",
        "router_lo_ipv4_address": mock_ipv4,
        "router_lo_ipv6_address": faker.ipv6(),
        "router_lo_iso_address": iso_from_ipv4(mock_ipv4),
    }


def test_import_site_endpoint(test_client, site_data):
    assert SubscriptionTable.query.all() == []
    # Post data to the endpoint
    response = test_client.post(SITE_IMPORT_ENDPOINT, json=site_data)
    assert response.status_code == 201
    assert "detail" in response.json()
    assert "pid" in response.json()
    subscription = subscriptions.retrieve_subscription_by_subscription_instance_value(
        resource_type="site_name",
        value=site_data["site_name"],
    )
    assert subscription is not None


def test_import_site_endpoint_with_existing_site(test_client, site_data):
    response = test_client.post(SITE_IMPORT_ENDPOINT, json=site_data)
    assert SubscriptionTable.query.count() == 1
    assert response.status_code == 201

    response = test_client.post(SITE_IMPORT_ENDPOINT, json=site_data)
    assert response.status_code == 422
    assert SubscriptionTable.query.count() == 1


def test_import_site_endpoint_with_invalid_data(test_client, site_data):
    # invalid data, missing site_latitude and invalid site_longitude
    site_data.pop("site_latitude")
    site_data["site_longitude"] = "invalid"
    assert SubscriptionTable.query.count() == 0
    response = test_client.post(SITE_IMPORT_ENDPOINT, json=site_data)
    assert response.status_code == 422
    assert SubscriptionTable.query.count() == 0
    response = response.json()
    assert response["detail"][0]["loc"] == ["body", "site_latitude"]
    assert response["detail"][0]["msg"] == "field required"
    assert response["detail"][1]["loc"] == ["body", "site_longitude"]
    assert response["detail"][1]["msg"] == "value is not a valid float"


def test_import_router_endpoint(test_client, site_data, router_data):
    # Create a site first
    response = test_client.post(SITE_IMPORT_ENDPOINT, json=site_data)
    assert response.status_code == 201
    assert SubscriptionTable.query.count() == 1

    response = test_client.post(ROUTER_IMPORT_ENDPOINT, json=router_data)
    assert response.status_code == 201
    assert SubscriptionTable.query.count() == 2


def test_import_router_endpoint_with_invalid_data(test_client, site_data, router_data):
    response = test_client.post(SITE_IMPORT_ENDPOINT, json=site_data)
    assert response.status_code == 201
    assert SubscriptionTable.query.count() == 1

    # invalid data, missing hostname and invalid router_lo_ipv6_address
    router_data.pop("hostname")
    router_data["router_lo_ipv6_address"] = "invalid"
    response = test_client.post(ROUTER_IMPORT_ENDPOINT, json=router_data)
    assert response.status_code == 422
    assert SubscriptionTable.query.count() == 1
    response = response.json()
    assert response["detail"][0]["loc"] == ["body", "hostname"]
    assert response["detail"][0]["msg"] == "field required"
    assert response["detail"][1]["loc"] == ["body", "router_lo_ipv6_address"]
    assert response["detail"][1]["msg"] == "value is not a valid IPv6 address"


def test_import_iptrunk_successful_with_real_process(test_client, mock_routers, iptrunk_data):
    response = test_client.post(IPTRUNK_IMPORT_API_URL, json=iptrunk_data)
    assert response.status_code == 201

    response = response.json()
    assert "detail" in response
    assert "pid" in response

    subscription = subscriptions.retrieve_subscription_by_subscription_instance_value(
        resource_type="geant_s_sid",
        value=iptrunk_data["geant_s_sid"],
    )
    assert subscription is not None


@patch("gso.api.v1.imports._start_process")
def test_import_iptrunk_invalid_customer(mock_start_process, test_client, mock_routers, iptrunk_data):
    iptrunk_data["customer"] = "not_existing_customer"
    mock_start_process.return_value = "123e4567-e89b-12d3-a456-426655440000"
    response = test_client.post(IPTRUNK_IMPORT_API_URL, json=iptrunk_data)

    assert response.status_code == 422
    assert response.json() == {
        "detail": [
            {
                "loc": ["body", "customer"],
                "msg": "Customer not_existing_customer not found",
                "type": "value_error",
            },
        ],
    }


@patch("gso.api.v1.imports._start_process")
def test_import_iptrunk_invalid_router_id_side_a_and_b(mock_start_process, test_client, iptrunk_data):
    iptrunk_data["side_a_node_id"] = "NOT FOUND"
    iptrunk_data["side_b_node_id"] = "NOT FOUND"

    mock_start_process.return_value = "123e4567-e89b-12d3-a456-426655440000"
    response = test_client.post(IPTRUNK_IMPORT_API_URL, json=iptrunk_data)

    assert response.status_code == 422
    assert response.json() == {
        "detail": [
            {
                "loc": ["body", "side_a_node_id"],
                "msg": f"Router {iptrunk_data['side_a_node_id']} not found",
                "type": "value_error",
            },
            {
                "loc": ["body", "side_b_node_id"],
                "msg": f"Router {iptrunk_data['side_b_node_id']} not found",
                "type": "value_error",
            },
        ],
    }


@patch("gso.api.v1.imports._start_process")
def test_import_iptrunk_non_unique_members_side_a(mock_start_process, test_client, mock_routers, iptrunk_data, faker):
    mock_start_process.return_value = "123e4567-e89b-12d3-a456-426655440000"

    repeat_interface_a = {
        "interface_name": faker.network_interface(),
        "interface_description": faker.sentence(),
    }
    repeat_interface_b = {
        "interface_name": faker.network_interface(),
        "interface_description": faker.sentence(),
    }
    iptrunk_data["side_a_ae_members"] = [repeat_interface_a for _ in range(5)]
    iptrunk_data["side_b_ae_members"] = [repeat_interface_b for _ in range(5)]

    response = test_client.post(IPTRUNK_IMPORT_API_URL, json=iptrunk_data)

    assert response.status_code == 422
    assert response.json() == {
        "detail": [
            {
                "loc": ["body", "side_a_ae_members"],
                "msg": "Items must be unique",
                "type": "value_error",
            },
            {
                "loc": ["body", "side_b_ae_members"],
                "msg": "Items must be unique",
                "type": "value_error",
            },
            {
                "loc": ["body", "__root__"],
                "msg": "Side A members should be at least 5 (iptrunk_minimum_links)",
                "type": "value_error",
            },
        ],
    }


@patch("gso.api.v1.imports._start_process")
def test_import_iptrunk_fails_on_side_a_member_count_mismatch(
    mock_start_process,
    test_client,
    mock_routers,
    iptrunk_data,
):
    mock_start_process.return_value = "123e4567-e89b-12d3-a456-426655440000"

    iptrunk_data["side_a_ae_members"].remove(iptrunk_data["side_a_ae_members"][0])

    response = test_client.post(IPTRUNK_IMPORT_API_URL, json=iptrunk_data)

    assert response.status_code == 422
    assert response.json() == {
        "detail": [
            {
                "loc": ["body", "__root__"],
                "msg": "Side A members should be at least 5 (iptrunk_minimum_links)",
                "type": "value_error",
            },
        ],
    }


@patch("gso.api.v1.imports._start_process")
def test_import_iptrunk_fails_on_side_a_and_b_members_mismatch(
    mock_start_process,
    test_client,
    iptrunk_data,
    mock_routers,
):
    mock_start_process.return_value = "123e4567-e89b-12d3-a456-426655440000"

    iptrunk_data["side_b_ae_members"].remove(iptrunk_data["side_b_ae_members"][0])

    response = test_client.post(IPTRUNK_IMPORT_API_URL, json=iptrunk_data)

    assert response.status_code == 422
    assert response.json() == {
        "detail": [
            {
                "loc": ["body", "__root__"],
                "msg": "Mismatch between Side A and B members",
                "type": "value_error",
            },
        ],
    }