test_worker_utils.py

"""
tests of a few worker utilities
"""
import json
import re

import jsonschema
import pytest

from inventory_provider.tasks import worker
from inventory_provider.tasks import common


def backend_db():
    return common._get_redis({
        'redis': {
            'hostname': None,
            'port': None
        },
        'redis-databases': [0, 7]
    }).db


def test_build_interface_services(mocked_worker_module):
    """
    checks that valid interface service objects are created
    :param mocked_worker_module: fixture
    :return:
    """
    ifc_schema = {
        '$schema': 'http://json-schema.org/draft-07/schema#',
        'type': 'object',
        'properties': {
            'description': {'type': 'string'},
            'router': {'type': 'string'},
            'interface': {'type': 'string'},
            'users': {
                'type': 'array',
                'items': {'type': 'string'}
            }
        },
        'required': ['router', 'interface', 'description'],
        'additionalProperties': False
    }

    db = backend_db()  # also forces initialization
    worker._build_service_category_interface_list()

    seen_types = set()
    for k, v in db.items():
        if not k.startswith('interface-services:'):
            continue

        print(v)
        (_, type, router, ifc_name) = k.split(':')

        ifc_info = json.loads(v)
        jsonschema.validate(json.loads(v), ifc_schema)

        assert ifc_info['router'] == router
        assert ifc_info['interface'] == ifc_name

        seen_types.add(type)

    assert type in ('mdvpn',
                    'lhcone_peer',
                    're_peer',
                    're_cust',
                    'ias',
                    'lhcone',
                    'lhcone_cust',
                    'l2_circuits',
                    'automated_l2_circuits')

    expected_seen_types = set(['mdvpn',
                               'lhcone_peer',
                               're_peer',
                               're_cust',
                               'ias',
                               'lhcone',
                               'lhcone_cust',
                               'l2_circuits',
                               'automated_l2_circuits'])
    assert seen_types == expected_seen_types


def test_build_subnet_db(mocked_worker_module):
    """
    checks that valid reverse subnet objects are created
    :param mocked_worker_module: fixture
    :return:
    """

    address_schema = {
        '$schema': 'http://json-schema.org/draft-07/schema#',

        'type': 'object',
        'properties': {
            'name': {'type': 'string'},
            'interface address': {'type': 'string'},
            'interface name': {'type': 'string'},
            'router': {'type': 'string'}
        },
        'required': ['name', 'interface address', 'interface name', 'router'],
        'additionalProperties': False
    }

    db = backend_db()  # also forces initialization
    worker._build_subnet_db()

    found_record = False
    for key, value in db.items():

        if not key.startswith('reverse_interface_addresses:'):
            continue

        found_record = True

        m = re.match('^reverse_interface_addresses:(.+)', key)
        assert m
        address = m.group(1)

        value = json.loads(value)
        jsonschema.validate(value, address_schema)
        assert value['name'] == address

    assert found_record


# TODO: more tests
# TODO: then share with neteng to make sure the test data is sensible
# this data is just enough to be an example
#   ... seems the method is not yet ready for testing
@pytest.mark.parametrize('description,expected_categories', [
    [
        'SRV_MDVPN CUSTOMER SANET SRF9939441 | VPN-Proxy to NREN CE',
        {
            worker.PollerServiceCategory.MDVPN
        }
    ],
    [
        'SRV_L3VPN CUSTOMER ESNET LHCONE SRF9928635 | ASN293 | GEN-EEX',
        {
            worker.PollerServiceCategory.LHCONE_CUST
        }
    ],
    [
        'SRV_L3VPN RE KREONET LHCONE SRF17072 | ASN17579',
        {
            worker.PollerServiceCategory.LHCONE_PEER
        }
    ],
    [
        'SRV_IAS CUSTOMER ACONET SRF9947199 IASPS | ASN1853',
        {
            worker.PollerServiceCategory.IAS
        }
    ],
    [
        'SRV_GLOBAL CUSTOMER HBKU SRF18085 | ASN34945|',
        {
            worker.PollerServiceCategory.RE_CUST
        }
    ],
    [
        'SRV_GLOBAL RE_INTERCONNECT HBKU SRF18085 | ASN34945|',
        {
            worker.PollerServiceCategory.RE_PEER
        }
    ],
    [
        'SRV_MDVPN CUSTOMER LHCONE SRV_L3VPN CUSTOMER SRF18085 | ASN34945|',
        {
            worker.PollerServiceCategory.MDVPN,
            worker.PollerServiceCategory.LHCONE_CUST
        }
    ]
])
def test_interface_classification(description, expected_categories):
    categories = worker._classify_interface({'description': description})
    assert set(list(categories)) == expected_categories


# def _all_interfaces():
#     import redis
#     r = redis.StrictRedis(host='test-dashboard-storage01.geant.org')
#     for k in r.scan_iter('netconf-interfaces:*'):
#         k = k.decode('utf-8')
#         ifc = json.loads(r.get(k).decode('utf-8'))
#         fields = k.split(':')
#         ifc['router'] = fields[1]
#         yield ifc
#
#
# def test_123():
#     with open('interfaces.json', 'w') as f:
#         f.write(json.dumps(list(_all_interfaces())))