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",
}
]
}