test_celery_worker.py

"""
just checks that the worker methods call the right functions
and some data ends up in the right place ... otherwise not very detailed
"""
import json
import os
import re

from lxml import etree
import pytest
from inventory_provider.tasks import worker

import inventory_provider

TEST_DATA_DIRNAME = os.path.realpath(os.path.join(
    inventory_provider.__path__[0],
    '..',
    'test',
    'data'))

# class MockedRedis(object):
#
#     db = {}
#
#     def __init__(self, *args, **kwargs):
#         pass
#
#     def set(self, name, value):
#         assert isinstance(value, str)
#         if name.startswith('netconf:'):
#             etree.fromstring(value)
#         elif name.startswith('snmp-interfaces:'):
#             obj = json.loads(value)
#             assert isinstance(obj, list)
#         MockedRedis.db[name] = value
#
#     def get(self, key):
#         return MockedRedis.db[key].encode('utf-8')


class MockedRedis(object):

    db = None

    def __init__(self, *args, **kwargs):
        if MockedRedis.db is None:
            test_data_filename = os.path.join(
                TEST_DATA_DIRNAME,
                "router-info.json")
            with open(test_data_filename) as f:
                MockedRedis.db = json.loads(f.read())

    def set(self, name, value):
        MockedRedis.db[name] = value

    def get(self, name):
        value = MockedRedis.db.get(name, None)
        if value is None:
            return None
        return value.encode('utf-8')

    def keys(self, glob=None):
        if not glob:
            return list([k.encode("utf-8") for k in MockedRedis.db.keys()])
        m = re.match(r'^([^*]+)\*$', glob)
        assert m  # all expected global are like this
        return list([
            k.encode("utf-8") for k in MockedRedis.db.keys()
            if k.startswith(m.group(1))])

    def delete(self, key):
        if isinstance(key, bytes):
            key = key.decode('utf-8')
        del MockedRedis.db[key]


@pytest.fixture
def mocked_worker_module(
        mocker, data_config, cached_test_data, mocked_netifaces):

    worker.InventoryTask.config = data_config

    mocker.patch(
        'inventory_provider.tasks.common.redis.StrictRedis',
        MockedRedis)

    def _mocked_snmp_interfaces(hostname, community, _):
        return json.loads(cached_test_data['snmp-interfaces:' + hostname])
    mocker.patch(
        'inventory_provider.snmp.get_router_interfaces',
        _mocked_snmp_interfaces)

    def _mocked_load_juniper_netconf_config(hostname, _):
        return etree.XML(cached_test_data['netconf:' + hostname])
    mocker.patch(
        'inventory_provider.juniper.load_config',
        _mocked_load_juniper_netconf_config)

    MockedRedis()  # create an instances so db is initialized


def test_netconf_refresh_config(mocked_worker_module, router):
    del MockedRedis.db['netconf:' + router]
    worker.netconf_refresh_config(router)
    assert MockedRedis.db['netconf:' + router]


def test_snmp_refresh_interfaces(mocked_worker_module, router):
    del MockedRedis.db['snmp-interfaces:' + router]
    worker.snmp_refresh_interfaces(router, 'fake-community')
    assert MockedRedis.db['snmp-interfaces:' + router]


def test_reload_router_config(mocked_worker_module, router, mocker):
    saved_data = {}
    for key in ('netconf:' + router, 'snmp-interfaces:' + router):
        saved_data[key] = MockedRedis.db.pop(key)
    assert 'netconf:' + router not in MockedRedis.db
    assert 'snmp-interfaces:' + router not in MockedRedis.db

    def _mocked_netconf_refresh_config_apply(args):
        key = 'netconf:' + args[0]
        MockedRedis.db[key] = saved_data[key]
    mocker.patch(
        'inventory_provider.tasks.worker.netconf_refresh_config.apply',
        _mocked_netconf_refresh_config_apply)

    def _mocked_snmp_refresh_interfaces_apply(args):
        assert len(args) == 2
        key = 'snmp-interfaces:' + args[0]
        MockedRedis.db[key] = saved_data[key]
    mocker.patch(
        'inventory_provider.tasks.worker.snmp_refresh_interfaces.apply',
        _mocked_snmp_refresh_interfaces_apply)

    mocker.patch(
        'inventory_provider.tasks.worker.snmp_refresh_interfaces.apply',
        _mocked_snmp_refresh_interfaces_apply)

    worker.reload_router_config(router)
    assert 'netconf:' + router in MockedRedis.db
    assert 'snmp-interfaces:' + router in MockedRedis.db