test_opsdb_queries.py

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
}


@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])