test_classifier_routes.py 9.46 KiB
import contextlib
import json
import jsonschema
import pytest
from inventory_provider.routes.classifier_schema import \
JUNIPER_LINK_RESPONSE_SCHEMA, PEER_INFO_RESPONSE_SCHEMA
DEFAULT_REQUEST_HEADERS = {
"Content-type": "application/json",
"Accept": ["application/json"]
}
def test_juniper_link_info(client):
rv = client.get(
'/classifier/juniper-link-info/mx1.ams.nl.geant.net/ae16.100',
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_RESPONSE_SCHEMA)
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_RESPONSE_SCHEMA)
assert response_data == {
'interface': {
'name': 'unknown-interface-name',
'description': '',
'ipv4': [],
'ipv6': [],
'bundle': [],
'bundle_members': []
},
'locations': [{
'a': {
'equipment': 'mx1.ams.nl.geant.net',
'name': 'Amsterdam',
'abbreviation': 'ams'}
}]
}
def test_juniper_link_unknown_router(client):
rv = client.get(
'/classifier/juniper-link-info/'
'unknown-router/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_RESPONSE_SCHEMA)
assert response_data == {
'interface': {
'name': 'unknown-interface-name',
'description': '',
'ipv4': [],
'ipv6': [],
'bundle': [],
'bundle_members': []
},
'locations': []
}
VPN_RR_PEER_INFO_KEYS = {'vpn-rr-peer-info', 'locations', 'snmp'}
IX_PUBLIC_PEER_INFO_KEYS = {
'ix-public-peer-info', 'interfaces', 'locations', 'snmp', 'asn'}
@pytest.mark.parametrize('peer_address,expected_response_keys', [
# MDVPN, asn=13092
('147.91.0.117', VPN_RR_PEER_INFO_KEYS | {'asn'}),
# MDVPN, no asn
('62.40.96.18', VPN_RR_PEER_INFO_KEYS | {'interfaces'}),
('2001:07f8:0036:0000:0000:3417:0000:0001', IX_PUBLIC_PEER_INFO_KEYS),
('2001:07f8:0030:0000:0002:0002:0003:2934', IX_PUBLIC_PEER_INFO_KEYS),
('195.66.227.154', IX_PUBLIC_PEER_INFO_KEYS),
('149.29.9.9', {'interfaces', 'locations', 'snmp', 'asn'}),
('62.40.125.142', {'interfaces', 'locations', 'snmp'})
])
def test_peer_info(
client, peer_address, expected_response_keys):
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, PEER_INFO_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
# TODO: schema validation
"""
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
def test_mtc_info(client):
rv = client.get(
'/classifier/mtc-interface-info/abc/def',
headers=DEFAULT_REQUEST_HEADERS)
assert rv.status_code == 200
assert rv.is_json
response_data = json.loads(rv.data.decode('utf-8'))
assert response_data
# schema seems wrong ...
# jsonschema.validate(response_data, MTC_INTERFACE_INFO_RESPONSE_SCHEMA)
def test_infinera_unparseable_fiberlink(client):
rv = client.get(
'/classifier/infinera-fiberlink-info/unparseableentitystring/aaa',
headers=DEFAULT_REQUEST_HEADERS)
assert rv.status_code == 422
rv = client.get(
'/classifier/infinera-fiberlink-info/XXX-OLA1-XXX02-DTNX10-1/aaa',
headers=DEFAULT_REQUEST_HEADERS)
assert rv.status_code == 422
def test_infinera_fiberlink_not_found(client):
rv = client.get(
'/classifier/infinera-fiberlink-info/'
'XXX-OLA1-XXX02-DTNX10-1/1-A-2-L1_3-A-2-L1',
headers=DEFAULT_REQUEST_HEADERS)
assert rv.status_code == 404
def test_infinera_fiberlink(client, mocker):
# mocker.patch('inventory_provider.routes.classifier.json.dumps')
mocked_redis = mocker.patch(
'inventory_provider.routes.classifier.common.get_current_redis')
mg = mocked_redis.return_value.get
mg.side_effect = [
False,
b"[{\"ne\": \"XXX-OLA1-1\", \"df_route\": \"end1-end2-dfroute\", \"df_route_id\": 1234, \"df_status\": \"Operational\", \"pop\": \"POP 1\", \"pop_abbreviation\": \"p1\"}]", # noqa
b"[{\"ne\": \"XXX02-DTNX10-1-3\", \"df_route\": \"end1-end2-dfroute\", \"df_route_id\": 1234, \"df_status\": \"Operational\", \"pop\": \"POP 2\", \"pop_abbreviation\": \"p2\"}]", # noqa
]
mocked_tls = mocker.patch(
'inventory_provider.routes.classifier.get_top_level_services')
mocked_tls.return_value = {'a': 'A'}
rv = client.get(
'/classifier/infinera-fiberlink-info/'
'XXX-OLA1-XXX02-DTNX10-1/1-A-2-L1_3-A-2-L1',
headers=DEFAULT_REQUEST_HEADERS)
mg.assert_any_call(
'classifier-cache:fiberlink:XXX-OLA1-XXX02-DTNX10-1:1-A-2-L1_3-A-2-L1')
mg.assert_any_call('opsdb:ne_fibre_spans:XXX-OLA1-1')
mg.assert_any_call('opsdb:ne_fibre_spans:XXX02-DTNX10-1-3')
mocked_tls.assert_called_with(1234, mocker.ANY)
expected = {
'ends': {
'a': {
'pop': 'POP 1',
'pop_abbreviation': 'p1',
'equipment': 'XXX-OLA1'
},
'b': {
'pop': 'POP 2',
'pop_abbreviation': 'p2',
'equipment': 'XXX02-DTNX10-1'
},
},
'df_route': {
'id': 1234,
'name': 'end1-end2-dfroute',
'status': 'Operational',
},
'related-services': {'a': 'A'}
}
assert rv.status_code == 200
assert rv.get_data(as_text=True) == json.dumps(expected)