Select Git revision
test_classifier_routes.py 11.88 KiB
import contextlib
import json
import jsonschema
import pytest
DEFAULT_REQUEST_HEADERS = {
"Content-type": "application/json",
"Accept": ["application/json"]
}
JUNIPER_LINK_METADATA = {
"$schema": "http://json-schema.org/draft-07/schema#",
"type": "object",
"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
}
},
"type": "object",
"properties": {
"services": {
"type": "array",
"items": {"$ref": "#/definitions/service-info"}
},
"interface": {"$ref": "#/definitions/interface-info"},
"related-services": {
"type": "array",
"items": {"$ref": "#/definitions/service-info"}
}
},
# "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)
VPN_RR_PEER_INFO_KEYS = {'vpn-rr-peer-info'}
IX_PUBLIC_PEER_INFO_KEYS = {'ix-public-peer-info', 'interfaces'}
@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 = {
"$schema": "http://json-schema.org/draft-07/schema#",
"type": "object",
"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"}
}
}
}
},
"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"}
}
},
"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 == 404
@pytest.mark.parametrize('equipment,entity_name,card_id,port_number', [
('grv3.ams.nl.geant.net', '1-1.3.1-100GbE-ODU4-TTP1', '1-1', '3'),
('bogus-hostname.with&special.char',
'234-2345234.7878i234crazynamewithslash/1-2.3', '234-2345234', '7878')
])
def test_coriant_info(
client, mocker, 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: {'C': a, 'E': b, 'CID': c, 'P': d})
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': {
'C': CONNECTION,
'E': equipment,
'CID': card_id,
'P': port_number
}
}
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