test_msr_routes.py

import json
import jsonschema

import pytest

from inventory_provider.routes.msr import PEERING_LIST_SCHEMA, \
    PEERING_GROUP_LIST_SCHEMA
from inventory_provider.routes.poller import SERVICES_LIST_SCHEMA

DEFAULT_REQUEST_HEADERS = {
    "Content-type": "application/json",
    "Accept": ["application/json"]
}


def test_access_services(client):
    rv = client.get(
        '/msr/access-services',
        headers=DEFAULT_REQUEST_HEADERS)
    assert rv.status_code == 200
    assert rv.is_json
    response_data = json.loads(rv.data.decode('utf-8'))
    jsonschema.validate(response_data, SERVICES_LIST_SCHEMA)

    assert response_data  # test data is non-empty
    assert all(s['type'] == 'GEANT IP' for s in response_data)


def test_logical_system_peerings_all(client):
    rv = client.get(
        '/msr/bgp/logical-system-peerings',
        headers=DEFAULT_REQUEST_HEADERS)
    assert rv.status_code == 200
    assert rv.is_json
    response_data = json.loads(rv.data.decode('utf-8'))
    jsonschema.validate(response_data, PEERING_LIST_SCHEMA)

    assert response_data  # test data is non-empty
    assert all('logical-system' in p for p in response_data)


@pytest.mark.parametrize('name', ['VRR', 'VPN-PROXY'])
def test_logical_system_peerings_specific(client, name):
    rv = client.get(
        f'/msr/bgp/logical-system-peerings/{name}',
        headers=DEFAULT_REQUEST_HEADERS)
    assert rv.status_code == 200
    assert rv.is_json
    response_data = json.loads(rv.data.decode('utf-8'))
    jsonschema.validate(response_data, PEERING_LIST_SCHEMA)

    assert response_data  # test data is non-empty
    assert all(p['logical-system'] == name for p in response_data)


@pytest.mark.parametrize('name', [
    'VRR1',
    'VPNPROXY',
    'vrr',
    ' vrr',
    'VPN PROXY'
])
def test_logical_system_peerings_404(client, name):
    rv = client.get(
        f'/msr/bgp/logical-system-peerings/{name}',
        headers=DEFAULT_REQUEST_HEADERS)
    assert rv.status_code == 404


def test_group_peerings_all(client):
    rv = client.get(
        '/msr/bgp/group-peerings',
        headers=DEFAULT_REQUEST_HEADERS)
    assert rv.status_code == 200
    assert rv.is_json
    response_data = json.loads(rv.data.decode('utf-8'))
    jsonschema.validate(response_data, PEERING_LIST_SCHEMA)

    assert response_data  # test data is non-empty


@pytest.mark.parametrize('name', ['BGPLU', 'eGEANT', 'eGEANT-mcast'])
def test_group_peerings_specific(client, name):
    rv = client.get(
        f'/msr/bgp/group-peerings/{name}',
        headers=DEFAULT_REQUEST_HEADERS)
    assert rv.status_code == 200
    assert rv.is_json
    response_data = json.loads(rv.data.decode('utf-8'))
    jsonschema.validate(response_data, PEERING_LIST_SCHEMA)

    assert response_data  # test data is non-empty
    assert all(p['group'] == name for p in response_data)


@pytest.mark.parametrize('name', ['EGEANT', 'eGEANT mcast'])
def test_group_peerings_404(client, name):
    rv = client.get(
        f'/msr/bgp/group-peerings/{name}',
        headers=DEFAULT_REQUEST_HEADERS)
    assert rv.status_code == 404


@pytest.mark.parametrize('name', ['PRACE-VPN', 'mdvpn'])
def test_routing_instance_peerings_specific(client, name):
    rv = client.get(
        f'/msr/bgp/routing-instance-peerings/{name}',
        headers=DEFAULT_REQUEST_HEADERS)
    assert rv.status_code == 200
    assert rv.is_json
    response_data = json.loads(rv.data.decode('utf-8'))
    jsonschema.validate(response_data, PEERING_LIST_SCHEMA)

    assert response_data  # test data is non-empty
    assert all(p['instance'] == name for p in response_data)


@pytest.mark.parametrize('name', ['xyz', 'MDVPN', 'PRACE'])
def test_routing_instance_peerings_404(client, name):
    rv = client.get(
        f'/msr/bgp/routing-instance-peerings/{name}',
        headers=DEFAULT_REQUEST_HEADERS)
    assert rv.status_code == 404


@pytest.mark.parametrize('uri', [
    '/msr/bgp/logical-systems',
    '/msr/bgp/groups',
    '/msr/bgp/routing-instances'])
def test_peerings_group_list(client, uri):
    rv = client.get(uri, headers=DEFAULT_REQUEST_HEADERS)
    assert rv.status_code == 200
    assert rv.is_json
    response_data = json.loads(rv.data.decode('utf-8'))
    jsonschema.validate(response_data, PEERING_GROUP_LIST_SCHEMA)

    assert response_data  # test data is non-empty