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
from gso.products.product_blocks.router import RouterRole, RouterVendor
from gso.products.product_blocks.site import SiteTier
from gso.utils.types.phy_port import PhyPortCapacity

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(router_subscription_factory, faker):
    router_side_a = router_subscription_factory()
    router_side_b = router_subscription_factory()
    return {
        "customer": "GÉANT",
        "geant_s_sid": faker.pystr(),
        "iptrunk_type": IptrunkType.DARK_FIBER,
        "iptrunk_description": faker.sentence(),
        "iptrunk_speed": PhyPortCapacity.HUNDRED_GIGABIT_PER_SECOND,
        "iptrunk_minimum_links": 5,
        "iptrunk_sideA_node_id": router_side_a,
        "iptrunk_sideA_ae_iface": faker.pystr(),
        "iptrunk_sideA_ae_geant_a_sid": faker.pystr(),
        "iptrunk_sideA_ae_members": [faker.pystr() for _ in range(5)],
        "iptrunk_sideA_ae_members_descriptions": [faker.sentence() for _ in range(5)],
        "iptrunk_sideB_node_id": router_side_b,
        "iptrunk_sideB_ae_iface": faker.pystr(),
        "iptrunk_sideB_ae_geant_a_sid": faker.pystr(),
        "iptrunk_sideB_ae_members": [faker.pystr() for _ in range(5)],
        "iptrunk_sideB_ae_members_descriptions": [faker.sentence() for _ in range(5)],
        "iptrunk_ipv4_network": str(faker.ipv4_network()),
        "iptrunk_ipv6_network": str(faker.ipv6_network()),
    }


@pytest.fixture
def mock_routers(iptrunk_data):
    first_call = [iptrunk_data["iptrunk_sideA_node_id"], iptrunk_data["iptrunk_sideB_node_id"], str(uuid4())]
    side_effects = [
        first_call,
        first_call,
        [
            (iptrunk_data["iptrunk_sideA_node_id"], "iptrunk_sideA_node_id description"),
            (iptrunk_data["iptrunk_sideB_node_id"], "iptrunk_sideB_node_id description"),
            (str(uuid4()), "random description"),
        ],
    ]
    with patch("gso.services.subscriptions.get_active_router_subscriptions") as mock_get_active_router_subscriptions:
        mock_get_active_router_subscriptions.side_effect = side_effects
        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, iptrunk_data, mock_routers):
    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.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):
    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",
        "is_ias_connected": True,
        "router_lo_ipv4_address": faker.ipv4(),
        "router_lo_ipv6_address": faker.ipv6(),
        "router_lo_iso_address": "TestAddress",
    }


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 == 409
    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, iptrunk_data, mock_routers):
    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, iptrunk_data, mock_routers):
    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):
    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", "iptrunk_sideA_node_id"], "msg": "Router not found", "type": "value_error"},
            {"loc": ["body", "iptrunk_sideB_node_id"], "msg": "Router 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, iptrunk_data, mock_routers):
    mock_start_process.return_value = "123e4567-e89b-12d3-a456-426655440000"

    iptrunk_data["iptrunk_sideA_ae_members"] = [5, 5, 5, 5, 5]
    iptrunk_data["iptrunk_sideB_ae_members"] = [4, 4, 4, 5, 5]

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

    assert response.status_code == 422
    assert response.json() == {
        "detail": [
            {"loc": ["body", "iptrunk_sideA_ae_members"], "msg": "Items must be unique", "type": "value_error"},
            {"loc": ["body", "iptrunk_sideB_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_iptrunk_import_fails_on_side_a_member_count_mismatch(
    mock_start_process, test_client, iptrunk_data, mock_routers
):
    mock_start_process.return_value = "123e4567-e89b-12d3-a456-426655440000"

    iptrunk_data["iptrunk_sideA_ae_members"].remove(iptrunk_data["iptrunk_sideA_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_iptrunk_import_fails_on_side_a_member_description_mismatch(
    mock_start_process, test_client, iptrunk_data, mock_routers
):
    mock_start_process.return_value = "123e4567-e89b-12d3-a456-426655440000"

    iptrunk_data["iptrunk_sideA_ae_members_descriptions"].remove(
        iptrunk_data["iptrunk_sideA_ae_members_descriptions"][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 in Side A members and their descriptions",
                "type": "value_error",
            }
        ]
    }


@patch("gso.api.v1.imports._start_process")
def test_iptrunk_import_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["iptrunk_sideB_ae_members"].remove(iptrunk_data["iptrunk_sideB_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"}]
    }


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

    iptrunk_data["iptrunk_sideB_ae_members_descriptions"].remove(
        iptrunk_data["iptrunk_sideB_ae_members_descriptions"][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 in Side B members and their descriptions",
                "type": "value_error",
            }
        ]
    }