-
Karel van Klink authored
vendor is moved back into the router product block
Karel van Klink authoredvendor is moved back into the router product block
test_imports.py 12.15 KiB
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,
"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()),
"iptrunk_ipv6_network": str(faker.ipv6_network()),
}
@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",
},
],
}