conftest.py

import ast
import json
import netifaces
import os
import re
import shutil
import tempfile

from lxml import etree
import pytest

import inventory_provider
from inventory_provider import config
from inventory_provider.tasks import worker

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


def data_config_filename(tmp_dir_name):
    config = {
        "ops-db": {
            "hostname": "xxxxxxx.yyyyy.zzz",
            "dbname": "xxxxxx",
            "username": "xxxxxx",
            "password": "xxxxxxxx"
        },
        "oid_list.conf": os.path.join(
            tmp_dir_name,
            "oid_list.conf"),
        "ssh": {
            "username": "uSeR-NaMe",
            "private-key": "private-key-filename",
            "known-hosts": "known-hosts=filename"
        },
        "redis": {
            "hostname": "xxxxxx",
            "port": 6379
        },
        "junosspace": {
            "api": "bogus-url",
            "username": "bogus-username",
            "password": "bogus-password"
        }
    }

    shutil.copyfile(
        os.path.join(TEST_DATA_DIRNAME, 'oid_list.conf'),
        config['oid_list.conf']
    )

    filename = os.path.join(tmp_dir_name, "config.json")
    with open(filename, "w") as f:
        f.write(json.dumps(config))

    return filename


def _tmp_data_config():
    with tempfile.TemporaryDirectory() as tmpdir:
        with open(data_config_filename(tmpdir)) as f:
            return config.load(f)


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


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 delete(self, key):
        if isinstance(key, bytes):
            key = key.decode('utf-8')
        del MockedRedis.db[key]

    def scan_iter(self, glob=None):
        if not glob:
            for k in MockedRedis.db.keys():
                yield k.encode('utf-8')

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

    def keys(self, glob=None):
        return list(self.scan_iter(glob))

    def flushdb(self):
        # only called from testing routes (hopefully)
        pass


@pytest.fixture
def data_config():
    return _tmp_data_config()


@pytest.fixture
def cached_test_data():
    filename = os.path.join(TEST_DATA_DIRNAME, "router-info.json")
    with open(filename) as f:
        return json.loads(f.read())


@pytest.fixture
def app_config():
    with tempfile.TemporaryDirectory() as tmpdir:

        app_config_filename = os.path.join(tmpdir, "app.config")
        with open(app_config_filename, "w") as f:
            f.write("%s = '%s'\n" % (
                "INVENTORY_PROVIDER_CONFIG_FILENAME",
                data_config_filename(tmpdir)))
            f.write('ENABLE_TESTING_ROUTES = True\n')

        yield app_config_filename


@pytest.fixture
def client(app_config, mocker):

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

    os.environ["SETTINGS_FILENAME"] = app_config
    with inventory_provider.create_app().test_client() as c:
        yield c


@pytest.fixture
def mocked_redis(mocker):
    mocker.patch(
        'inventory_provider.tasks.common.redis.StrictRedis',
        MockedRedis)


@pytest.fixture
def client_with_mocked_data(client, mocked_redis):
    return client


NETIFACES_TEST_DATA_STRING = """{
    'lo0':  {{AF_INET}: [{'addr': '127.0.0.1', 'netmask': '255.0.0.0', 'peer': '127.0.0.1'}],
             {AF_INET6}: [{'addr': '::1', 'netmask': 'ffff:ffff:ffff:ffff:ffff:ffff:ffff:ffff/128', 'peer': '::1', 'flags': 0},
                 {'addr': 'fe80::1%lo0', 'netmask': 'ffff:ffff:ffff:ffff::/64', 'flags': 0}]},
    'eth0': {{AF_LINK}: [{'addr': '78:4f:43:76:73:ba'}],
             {AF_INET}: [{'addr': '83.97.92.239', 'netmask': '255.255.252.0', 'broadcast': '83.97.95.255'}],
             {AF_INET6}: [{'addr': 'fe80::250:56ff:fea1:8340', 'netmask': 'ffff:ffff:ffff:ffff::/64', 'flags': 1024},
                 {'addr': '2001:798:3::104', 'netmask': 'ffff:ffff:ffff:ffff::/64', 'flags': 1088}]}
}"""  # noqa E501


@pytest.fixture
def mocked_netifaces(mocker):
    s = NETIFACES_TEST_DATA_STRING
    for k, v in {
                'AF_INET': netifaces.AF_INET,
                'AF_INET6': netifaces.AF_INET6,
                'AF_LINK': netifaces.AF_LINK
            }.items():
        s = s.replace('{%s}' % k, str(v))
    data = ast.literal_eval(s)
    mocker.patch('netifaces.interfaces', lambda: data.keys())
    mocker.patch('netifaces.ifaddresses', lambda n: data[n])


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

    os.environ['BROKER_HOSTNAME'] = 'a.b.c'
    os.environ['BROKER_PORT'] = '123'
    os.environ['CELERY_DB_INDEX'] = '456'
    os.environ['BROKER_SCHEME'] = 'redis'

    worker.InventoryTask.config = data_config

    def _mocked_snmp_interfaces(hostname, community):
        return json.loads(cached_test_data['snmp-interfaces:' + hostname])
    mocker.patch(
        'inventory_provider.snmp.get_router_snmp_indexes',
        _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)