test_opsdb_queries.py 10.86 KiB
import json
import os
import jsonschema
import pytest
from inventory_provider.db import db
from inventory_provider.db import opsdb
pytestmark = pytest.mark.skipif(
'TEST_OPSDB_HOSTNAME' not in os.environ,
reason='TEST_OPSDB_HOSTNAME environment variable not found')
pytestmark = pytest.mark.skipif(
'TEST_OPSDB_DBNAME' not in os.environ,
reason='TEST_OPSDB_DBNAME environment variable not found')
pytestmark = pytest.mark.skipif(
'TEST_OPSDB_USERNAME' not in os.environ,
reason='TEST_OPSDB_USERNAME environment variable not found')
pytestmark = pytest.mark.skipif(
'TEST_OPSDB_PASSWORD' not in os.environ,
reason='TEST_OPSDB_PASSWORD environment variable not found')
@pytest.fixture
def db_params():
return {
'hostname': os.environ['TEST_OPSDB_HOSTNAME'],
'dbname': os.environ['TEST_OPSDB_DBNAME'],
'username': os.environ['TEST_OPSDB_USERNAME'],
'password': os.environ['TEST_OPSDB_PASSWORD'],
}
@pytest.fixture
def connection(db_params):
with db.connection(db_params) as c:
yield c
EQUIPMENT_LOCATION_METADATA = {
"$schema": "http://json-schema.org/draft-07/schema#",
"definitions": {
"pop-info": {
"type": "object",
"properties": {
"name": {"type": "string"},
"abbreviation": {"type": "string"},
"country": {"type": "string"},
"city": {"type": "string"},
"longitude": {"type": "number"},
"latitude": {"type": "number"}
},
"required": [
"name",
"abbreviation",
"country",
"city",
"longitude",
"latitude"
],
"additionalProperties": False
},
"equipment-info": {
"type": "object",
"properties": {
'equipment-name': {"type": "string"},
'status': {"type": "string"},
'pop': {"$ref": "#/definitions/pop-info"}
},
"required": [
"equipment-name",
"status",
"pop"
],
"additionalProperties": False
}
},
"type": "array",
"items": {"$ref": "#/definitions/equipment-info"}
}
CORIANT_PATH_METADATA = {
"$schema": "http://json-schema.org/draft-07/schema#",
"definitions": {
"pop-info": {
"type": "object",
"properties": {
"name": {"type": "string"},
"abbreviation": {"type": "string"},
"country": {"type": "string"},
"city": {"type": "string"},
"longitude": {"type": "number"},
"latitude": {"type": "number"}
},
"required": [
"name",
"abbreviation",
"country",
"city",
"longitude",
"latitude"
],
"additionalProperties": False
},
"endpoint": {
"type": "object",
"properties": {
"equipment name": {"type": "string"},
"card id": {"type": "string"},
"port number": {"type": "string"},
"pop": {"$ref": "#/definitions/pop-info"}
},
"required": ["equipment name", "card id", "port number", "pop"],
"additionalProperties": False
}
},
"type": "object",
"properties": {
'id': {"type": "integer"},
'category': {"type": "string"},
'circuit_type': {"type": "string"},
'service_type': {"type": "string"},
'peering_type': {"type": "string"},
'status': {"type": "string"},
'name': {"type": "string"},
'a': {"$ref": "#/definitions/endpoint"},
'b': {"$ref": "#/definitions/endpoint"}
},
"required": [
"id",
"category",
"circuit_type",
"service_type",
"peering_type",
"status",
"a",
"b"],
"additionalProperties": False
}
LG_ROUTERS_SCHEMA = {
"$schema": "http://json-schema.org/draft-07/schema#",
"definitions": {
"pop-info": {
"type": "object",
"properties": {
"name": {"type": "string"},
"abbreviation": {"type": "string"},
"country": {"type": "string"},
"country code": {"type": "string"},
"city": {"type": "string"},
"longitude": {"type": "number"},
"latitude": {"type": "number"}
},
"required": [
"name",
"abbreviation",
"country",
"country code",
"city",
"longitude",
"latitude"
],
"additionalProperties": False
},
"router": {
"type": "object",
"properties": {
"equipment name": {"type": "string"},
"type": {
"type": "string",
"enum": ["INTERNAL", "CORE"]
},
"pop": {"$ref": "#/definitions/pop-info"}
},
"required": ["equipment name", "type", "pop"],
"additionalProperties": False
}
},
"type": "array",
"items": {"$ref": "#/definitions/router"}
}
@pytest.mark.parametrize('equipment', [
'mx1.cbg.uk.geant.net',
'grv3.lon.uk.geant.net',
'mx1.gen.ch.geant.net'
])
def test_equipment_location(connection, equipment):
circuit = opsdb.lookup_pop_info(connection, equipment)
jsonschema.validate(circuit, EQUIPMENT_LOCATION_METADATA)
assert len(circuit) == 1
@pytest.mark.parametrize('equipment,card,port', [
('grv3.lon.uk.geant.net', '1-1', '3'),
('grv3.lon.uk.geant.net', '1-1', '5'),
('grv1.ams.nl.geant.net', '1-1', '1'),
('grv3.lon.uk.geant.net', '1-1', '1'),
])
def test_coriant_path(connection, equipment, card, port):
circuit = opsdb.lookup_coriant_path(connection, equipment, card, port)
jsonschema.validate(circuit, CORIANT_PATH_METADATA)
SERVICE_USER_LIST_SCHEMA = {
'$schema': 'http://json-schema.org/draft-07/schema#',
'definitions': {
'service-user': {
'type': 'object',
'properties': {
'service_id': {'type': 'integer'},
'user': {'type': 'string'}
},
'required': ['service_id', 'user'],
'additionalProperties': False
}
},
'type': 'array',
'items': {'$ref': '#/definitions/service-user'}
}
def test_get_service_users(connection):
service_id_list = [47673, 47675]
users = opsdb.get_service_users(connection, service_id_list)
users = list(users)
jsonschema.validate(users, SERVICE_USER_LIST_SCHEMA)
assert users
def test_get_all_service_users(connection, cached_test_data):
def _all_interfaces():
for k in cached_test_data.keys():
if not k.startswith('netconf-interfaces:'):
continue
(_, hostname, ifc_name) = k.split(':')
yield {'hostname': hostname, 'interface': ifc_name}
def _all_service_ids(interfaces):
for ifc in interfaces:
key = ('opsdb:interface_services'
f':{ifc["hostname"]}:{ifc["interface"]}')
if key not in cached_test_data:
print(f'warning: {key} not found in cached test data')
continue
for service in json.loads(cached_test_data[key]):
info = {'service_id': service['id']}
info.update(ifc)
yield info
ids = {s['service_id'] for s in _all_service_ids(_all_interfaces())}
assert len(ids) > 0
service_users = list(opsdb.get_service_users(connection, list(ids)))
jsonschema.validate(service_users, SERVICE_USER_LIST_SCHEMA)
assert service_users
# for user in opsdb.get_service_users(connection, list(ids)):
# services.setdefault(user['service_id'], []).append(user['user'])
#
# print([f'{k}: {v}' for k, v in services.items() if len(v) > 1])
def test_lookup_lg_routers(connection, cached_test_data):
routers = list(opsdb.lookup_lg_routers(connection))
jsonschema.validate(routers, LG_ROUTERS_SCHEMA)
assert routers # shouldn't be empty
CIRCUIT_INFO_SCHEMA = {
"$schema": "http://json-schema.org/draft-07/schema#",
"type": "object",
"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", "link-aggr-group"]
},
"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",
"intel", "dell", "mellanox technologies",
"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"}
},
"required": [
"id", "name", "status", "circuit_type",
"service_type", "project",
"equipment", "other_end_equipment",
"card_id", "other_end_card_id",
"port", "other_end_port",
"logical_unit", "other_end_logical_unit",
"interface_name", "other_end_interface_name"
# "pop", "other_end_pop",
# "pop_name", "other_end_pop_name",
# "pop_abbreviation", "other_end_pop_abbreviation",
],
"additionalProperties": False
}
def test_get_circuits(connection, cached_test_data):
import re
for circuit in opsdb.get_circuits(connection):
jsonschema.validate(circuit, CIRCUIT_INFO_SCHEMA)