Skip to content
Snippets Groups Projects
test_classifier_routes.py 14.62 KiB
import contextlib
import json
import jsonschema
import pytest

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

LOCATIONS_DEFINITIONS = {
    "location-endpoint": {
        "type": "object",
        "properties": {
            "equipment": {"type": "string"},
            "name": {"type": "string"},
            "abbreviation": {"type": "string"}
        },
        "required": ["equipment", "name", "abbreviation"],
        "additionalProperties": False
    },
    "location": {
        "type": "object",
        "properties": {
            "a": {"$ref": "#/definitions/location-endpoint"},
            "b": {"$ref": "#/definitions/location-endpoint"}
        },
        "required": ["a"],
        "additionalProperties": False
    },
    "locations-list": {
        "type": "array",
        "items": {"$ref": "#/definitions/location"}
    }
}

JUNIPER_LINK_METADATA_DEFINITIONS = {
    "ip-address": {
        "type": "string",
        "oneOf": [
            {"pattern": r'^(\d+\.){3}\d+$'},
            {"pattern": r'^([a-f\d]{4}:){7}[a-f\d]{4}$'}
        ]
    },
    "ipv4-interface-address": {
        "type": "string",
        "pattern": r'^(\d+\.){3}\d+/\d+$'
    },
    "ipv6-interface-address": {
        "type": "string",
        "pattern": r'^[a-f\d:]+/\d+$'
    },
    "interface-info": {
        "type": "object",
        "properties": {
            "name": {"type": "string"},
            "description": {"type": "string"},
            "ipv4": {
                "type": "array",
                "items": {"$ref": "#/definitions/ipv4-interface-address"}
            },
            "ipv6": {
                "type": "array",
                "items": {"$ref": "#/definitions/ipv6-interface-address"}
            },

            # TODO: check what's changed: added to make tests pass
            'bundle': {"type": "array"}
        },
        "required": ["name", "description", "ipv4", "ipv6"],
        "additionalProperties": False
    },
    "service-info": {
        "type": "object",
        "properties": {
            "id": {"type": "integer"},
            "name": {"type": "string"},
            "status": {
                "type": "string",
                "enum": ["operational", "installed", "planned", "ordered"]
            },
            "circuit_type": {
                "type": "string",
                "enum": ["path", "service", "l2circuit"]
            },
            "service_type": {"type": "string"},
            "project": {"type": "string"},
            "equipment": {"type": "string"},
            "pop": {"type": "string"},
            "pop_abbreviation": {"type": "string"},

            "other_end_pop": {"type": "string"},
            "other_end_pop_abbreviation": {"type": "string"},
            "other_end_equipment": {"type": "string"},
            "port": {"type": "string"},
            "other_end_port": {"type": "string"},
            "logical_unit": {
                "oneOf": [
                    {"type": "integer"},
                    {"type": "string", "maxLength": 0}
                ]
            },
            "other_end_logical_unit": {
                "oneOf": [
                    {"type": "integer"},
                    {"type": "string", "maxLength": 0}
                ]
            },
            "manufacturer": {
                "type": "string",
                "enum": ["juniper", "coriant", "infinera",
                         "cisco", "hewlett packard",
                         "corsa", "graham smith uk ltd",
                         "unknown", ""]
            },
            "card_id": {"type": "string"},
            "other_end_card_id": {"type": "string"},
            "interface_name": {"type": "string"},
            "other_end_interface_name": {"type": "string"},

            # TODO: check what's changed: added to make tests pass
            'other_end_pop_name': {"type": "string"},
            'pop_name': {"type": "string"}
        },
        # TODO: modify service-info so that "" entries are just omitted
        #       (... rather than requiring 'oneOf')
        # TODO: put 'other_end_*' params in a sub dictionary
        # "required": [
        #     "id", "name", "status",
        #     "circuit_type", "service_type",
        #     "project", "port", "manufacturer",
        #     "equipment", "logical_unit", "card_id", "interface_name"
        # ],
        "additionalProperties": False
    },
    "related-service-info": {
        "type": "object",
        "properties": {
            "name": {"type": "string"},
            "status": {
                "type": "string",
                "enum": ["operational", "installed", "planned", "ordered"]
            },
            "circuit_type": {
                "type": "string",
                "enum": ["path", "service", "l2circuit"]
            },
            "project": {"type": "string"}
        },
        "required": ["name", "status", "circuit_type", "project"],
        "additionalProperties": False
    }
}

JUNIPER_LINK_METADATA = {
    "$schema": "http://json-schema.org/draft-07/schema#",
    "type": "object",

    "definitions": {
        **JUNIPER_LINK_METADATA_DEFINITIONS, **LOCATIONS_DEFINITIONS
    },

    "type": "object",
    "properties": {
        "services": {
            "type": "array",
            "items": {"$ref": "#/definitions/service-info"}
        },
        "interface": {"$ref": "#/definitions/interface-info"},
        "related-services": {
            "type": "array",
            "items": {"$ref": "#/definitions/related-service-info"}
        },
        "locations": {"$ref": "#/definitions/locations-list"}
    },
    # "required": ["interface"],
    "additionalProperties": False
}


def test_juniper_link_info(client):
    rv = client.get(
        '/classifier/juniper-link-info/mx1.ams.nl.geant.net/ae15.1500',
        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, JUNIPER_LINK_METADATA)


def test_juniper_link_info_not_found(client):
    rv = client.get(
        '/classifier/juniper-link-info/'
        'mx1.ams.nl.geant.net/unknown-interface-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, JUNIPER_LINK_METADATA)
    assert response_data == {
        'interface': {
            'name': 'unknown-interface-name',
            'description': '',
            'ipv4': [],
            'ipv6': [],
            'bundle': []
        },
        'locations': [{
            'a': {
                'equipment': 'mx1.ams.nl.geant.net',
                'name': 'Amsterdam',
                'abbreviation': 'ams'}
        }]
    }


VPN_RR_PEER_INFO_KEYS = {'vpn-rr-peer-info', 'locations'}
IX_PUBLIC_PEER_INFO_KEYS = {'ix-public-peer-info', 'interfaces', 'locations'}


@pytest.mark.parametrize('peer_address,expected_response_keys', [
    ('109.105.110.54', VPN_RR_PEER_INFO_KEYS),
    ('2001:07f8:001c:024a:0000:0000:316e:0001', IX_PUBLIC_PEER_INFO_KEYS),
    ('2001:07f8:000b:0100:01d1:a5d1:0310:0029', IX_PUBLIC_PEER_INFO_KEYS),
    ('195.66.224.238', IX_PUBLIC_PEER_INFO_KEYS),
]
)
def test_peer_info(
        client, peer_address, expected_response_keys):
    response_schema_definitions = {
        "ip-address": {
            "type": "string",
            "oneOf": [
                {"pattern": r'^(\d+\.){3}\d+$'},
                {"pattern": r'^([a-f\d]{4}:){7}[a-f\d]{4}$'}
            ]
        },
        "interface-address": {
            "type": "string",
            "oneOf": [
                {"pattern": r'^(\d+\.){3}\d+/\d+$'},
                {"pattern": r'^[a-f\d:]+/\d+$'}
            ]
        },
        "vpn-rr-peer": {
            "type": "object",
            "properties": {
                "name": {"$ref": "#/definitions/ip-address"},
                "description": {"type": "string"},
                "peer-as": {"type": "integer"},
                "router": {"type": "string"}
            },
            "required": ["name", "description"],
            "additionalProperties": False
        },
        "ix-public-peer": {
            "type": "object",
            "properties": {
                "name": {"$ref": "#/definitions/ip-address"},
                "description": {"type": "string"},
                "router": {"type": "string"},
                "as": {
                    "type": "object",
                    "properties": {
                        "local": {"type": "integer"},
                        "peer": {"type": "integer"},
                    },
                    "required": ["local", "peer"],
                    "additionalProperties": False
                }
            },
            "required": ["name", "description", "as"],
            "additionalProperties": False
        },
        "ix-public-peer-list": {
            "type": "array",
            "items": {"$ref": "#/definitions/ip-address"}
        },
        "ix-public-peer-info": {
            "type": "object",
            "properties": {
                "peer": {"$ref": "#/definitions/ix-public-peer"},
                "group": {"$ref": "#/definitions/ix-public-peer-list"},
                "router": {"$ref": "#/definitions/ix-public-peer-list"}
            },
            "required": ["peer", "group", "router"],
            "additionalProperties": False
        },
        "interface-info": {
            "type": "object",
            "properties": {
                "name": {"$ref": "#/definitions/ip-address"},
                "interface address": {
                    "$ref": "#/definitions/interface-address"},
                "interface name": {"type": "string"},
                "router": {"type": "string"}
            },
            "required": [
                "name", "interface address", "interface name", "router"],
            "additionalProperties": False
        },
        "service-info": {
            "type": "object"
        },
        "interface-lookup-info": {
            "type": "object",
            "properties": {
                "interface": {"$ref": "#/definitions/interface-info"},
                "services": {
                    "type": "array",
                    "items": {"$ref": "#/definitions/service-info"}
                }
            }
        }
    }
    response_schema = {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "type": "object",

        "definitions": {
            **response_schema_definitions, **LOCATIONS_DEFINITIONS
        },

        "type": "object",
        "properties": {
            "ix-public-peer-info": {
                "$ref": "#/definitions/ix-public-peer-info"},
            "vpn-rr-peer-info": {"$ref": "#/definitions/vpn-rr-peer"},
            "interfaces": {
                "type": "array",
                "items": {"$ref": "#/definitions/interface-lookup-info"}
            },
            "locations": {"$ref": "#/definitions/locations-list"}
        },
        "additionalProperties": False
    }

    rv = client.get(
        '/classifier/peer-info/%s' % peer_address,
        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, response_schema)

    assert set(response_data.keys()) == expected_response_keys


def test_peer_invalid_address(client):
    rv = client.get(
        '/classifier/peer-info/1.2.3.4.5',
        headers=DEFAULT_REQUEST_HEADERS)
    assert rv.status_code == 422


def test_peer_not_found(client):
    rv = client.get(
        '/classifier/peer-info/1.2.3.4',
        headers=DEFAULT_REQUEST_HEADERS)
    assert rv.status_code == 200
    response_data = json.loads(rv.data.decode('utf-8'))
    assert response_data == {'locations': []}


@pytest.mark.parametrize('id_,equipment,entity_name,card_id,port_number', [
    ('123', 'grv3.ams.nl.geant.net', '1-1.3.1-100GbE-ODU4-TTP1', '1-1', '3'),
    ('123', 'bogus-hostname.with&special.char',
     '234-2345234.7878i234crazynamewithslash/1-2.3', '234-2345234', '7878')
])
def test_coriant_info(
        client, mocker, id_, equipment, entity_name, card_id, port_number):
    """
    just check that entity_name is correctly parsed and the correct
    method is called, but mock out all sql access
    """
    CONNECTION = 'bogus connection'

    @contextlib.contextmanager
    def mocked_connection(ignored):
        yield CONNECTION

    mocker.patch(
        'inventory_provider.db.db.connection', mocked_connection)
    mocker.patch(
        'inventory_provider.db.opsdb.lookup_coriant_path',
        lambda a, b, c, d: {
            'id': '123',
            'C': a,
            'E': b,
            'CID': c,
            'P': d,
            'a': {
                'equipment name': 'abc',
                'pop': {
                    'name': 'zzz',
                    'abbreviation': '123'
                }
            },
            'b': {
                'equipment name': 'ddd',
                'pop': {
                    'name': 'aaa',
                    'abbreviation': '999'
                }
            }
        })

    rv = client.get(
        '/classifier/coriant-info/{equipment}/{entity}'.format(
            equipment=equipment,
            entity=entity_name),
        headers=DEFAULT_REQUEST_HEADERS)

    assert rv.status_code == 200
    assert rv.is_json
    response_data = json.loads(rv.data.decode('utf-8'))

    expected_response = {
        'equipment name': equipment,
        'card id': card_id,
        'port number': port_number,
        'path': {
            'id': id_,
            'C': CONNECTION,
            'E': equipment,
            'CID': card_id,
            'P': port_number,
            'a': {
                'equipment name': 'abc',
                'pop': {'name': 'zzz', 'abbreviation': '123'}},
            'b': {
                'equipment name': 'ddd',
                'pop': {'name': 'aaa', 'abbreviation': '999'}}
        },
        'locations': [{
            'a': {'equipment': 'abc', 'name': 'zzz', 'abbreviation': '123'},
            'b': {'equipment': 'ddd', 'name': 'aaa', 'abbreviation': '999'}
        }]
    }

    assert response_data == expected_response


def test_coriant_info_not_found(client, mocker):
    """
    just check the correct method is called, but mock out all sql access
    """

    @contextlib.contextmanager
    def mocked_connection(ignored):
        yield None

    mocker.patch(
        'inventory_provider.db.db.connection', mocked_connection)
    mocker.patch(
        'inventory_provider.db.opsdb.lookup_coriant_path',
        lambda a, b, c, d: None)

    rv = client.get(
        '/classifier/coriant-info/aaa/unparseableentitystring',
        headers=DEFAULT_REQUEST_HEADERS)

    assert rv.status_code == 404