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 logging
import pytest
from inventory_provider.tasks import worker


class MockedRedis(object):

    db = {}

    def __init__(self, *args, **kwargs):
        pass

    def hset(self, name, key, value):
        MockedRedis.db.setdefault(name, {})[key] = value


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

    worker.InventoryTask.config = data_config
    worker.InventoryTask.logger = logging.getLogger()

    MockedRedis.db = {}

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


def test_juniper_refresh_bgp(
        mocked_worker_module, mocker, cached_test_data):

    def _mocked_fetch_bpg_config(hostname, _):
        return cached_test_data[hostname]["bgp"]

    mocker.patch(
        'inventory_provider.tasks.worker.juniper.fetch_bgp_config',
        _mocked_fetch_bpg_config)

    for hostname in cached_test_data.keys():
        assert 'hostname' not in MockedRedis.db
        worker.juniper_refresh_bgp(hostname)
        assert MockedRedis.db[hostname]['bgp']


def test_juniper_refresh_interfaces(
        mocked_worker_module, mocker, cached_test_data):

    def _mocked_fetch_interfaces(hostname, _):
        return cached_test_data[hostname]["interfaces"]

    mocker.patch(
        'inventory_provider.tasks.worker.juniper.fetch_interfaces',
        _mocked_fetch_interfaces)

    for hostname in cached_test_data.keys():
        assert 'hostname' not in MockedRedis.db
        worker.juniper_refresh_interfaces(hostname)
        assert MockedRedis.db[hostname]['interfaces']


def test_juniper_refresh_vrr(
        mocked_worker_module, mocker, cached_test_data):

    def _mocked_fetch_vrr_config(hostname, _):
        return cached_test_data[hostname]["vrr"]

    mocker.patch(
        'inventory_provider.tasks.worker.juniper.fetch_vrr_config',
        _mocked_fetch_vrr_config)

    for hostname in cached_test_data.keys():
        assert 'hostname' not in MockedRedis.db
        worker.juniper_refresh_vrr(hostname)
        assert MockedRedis.db[hostname]['vrr']


def test_snmp_refresh_interfaces(
        mocked_worker_module, mocker, cached_test_data):

    def _mocked_snmp_interfaces(hostname, community, _):
        return cached_test_data[hostname]["snmp-interfaces"]

    mocker.patch(
        'inventory_provider.tasks.worker.snmp.get_router_interfaces',
        _mocked_snmp_interfaces)

    for hostname in cached_test_data.keys():
        assert 'hostname' not in MockedRedis.db
        worker.snmp_refresh_interfaces(hostname, 'fake-community')
        assert MockedRedis.db[hostname]['snmp-interfaces']