diff --git a/test/per_router/conftest.py b/test/per_router/conftest.py index 03646f222e8289c61411e2752b59233ca5d5c756..3c6943cf2ead2807b6fc7992b246a518cc143456 100644 --- a/test/per_router/conftest.py +++ b/test/per_router/conftest.py @@ -1,8 +1,10 @@ import glob +import json import os import re from lxml import etree +import pytest import inventory_provider @@ -13,6 +15,50 @@ TEST_DATA_DIRNAME = os.path.realpath(os.path.join( "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 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_redis(mocker): + mocker.patch( + 'inventory_provider.tasks.common.redis.StrictRedis', + MockedRedis) + return MockedRedis() + + def pytest_generate_tests(metafunc): def _junosspace_hosts(): diff --git a/test/per_router/test_celery_worker.py b/test/per_router/test_celery_worker.py index fa8037043e5e8b3ba3e85ae2489c0f0f14bcb0d9..43d0b445196b2775cae71493502d1c803fc200af 100644 --- a/test/per_router/test_celery_worker.py +++ b/test/per_router/test_celery_worker.py @@ -9,14 +9,7 @@ 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')) +from inventory_provider.tasks.common import get_redis # class MockedRedis(object): # @@ -38,52 +31,18 @@ TEST_DATA_DIRNAME = os.path.realpath(os.path.join( # 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] - +def backend_db(): + return get_redis({ + 'redis': {'hostname': None, 'port': None} + }).db @pytest.fixture def mocked_worker_module( - mocker, data_config, cached_test_data, mocked_netifaces): + mocker, mocked_redis, 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( @@ -96,31 +55,29 @@ def mocked_worker_module( '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] + del backend_db()['netconf:' + router] worker.netconf_refresh_config(router) - assert MockedRedis.db['netconf:' + router] + assert backend_db()['netconf:' + router] def test_snmp_refresh_interfaces(mocked_worker_module, router): - del MockedRedis.db['snmp-interfaces:' + router] + del backend_db()['snmp-interfaces:' + router] worker.snmp_refresh_interfaces(router, 'fake-community') - assert MockedRedis.db['snmp-interfaces:' + router] + assert backend_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 + saved_data[key] = backend_db().pop(key) + assert 'netconf:' + router not in backend_db() + assert 'snmp-interfaces:' + router not in backend_db() def _mocked_netconf_refresh_config_apply(args): key = 'netconf:' + args[0] - MockedRedis.db[key] = saved_data[key] + backend_db()[key] = saved_data[key] mocker.patch( 'inventory_provider.tasks.worker.netconf_refresh_config.apply', _mocked_netconf_refresh_config_apply) @@ -128,7 +85,7 @@ def test_reload_router_config(mocked_worker_module, router, mocker): def _mocked_snmp_refresh_interfaces_apply(args): assert len(args) == 2 key = 'snmp-interfaces:' + args[0] - MockedRedis.db[key] = saved_data[key] + backend_db()[key] = saved_data[key] mocker.patch( 'inventory_provider.tasks.worker.snmp_refresh_interfaces.apply', _mocked_snmp_refresh_interfaces_apply) @@ -140,5 +97,5 @@ def test_reload_router_config(mocked_worker_module, router, mocker): _mocked_update_status) worker.reload_router_config(router) - assert 'netconf:' + router in MockedRedis.db - assert 'snmp-interfaces:' + router in MockedRedis.db + assert 'netconf:' + router in backend_db() + assert 'snmp-interfaces:' + router in backend_db() diff --git a/test/per_router/test_data_routes.py b/test/per_router/test_data_routes.py index 2d9dc30430c878284c4750e5b22d8c479694c6af..caa270348c7a6844a128c5067c94087d0acc7448 100644 --- a/test/per_router/test_data_routes.py +++ b/test/per_router/test_data_routes.py @@ -7,68 +7,68 @@ import jsonschema import inventory_provider -TEST_DATA_DIRNAME = os.path.realpath(os.path.join( - inventory_provider.__path__[0], - "..", - "test", - "data")) - +# TEST_DATA_DIRNAME = os.path.realpath(os.path.join( +# inventory_provider.__path__[0], +# "..", +# "test", +# "data")) +# DEFAULT_REQUEST_HEADERS = { "Content-type": "application/json", "Accept": ["application/json"] } -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, key, value): - MockedRedis.db[key] = value - - def get(self, key): - return MockedRedis.db[key].encode('utf-8') - - # def hget(self, key, field): - # value = MockedRedis.db[key] - # return value[field].encode('utf-8') - # # return json.dumps(value[field]).encode('utf-8') - # - # def hgetall(self, key): - # result = {} - # for k, v in MockedRedis.db[key].items(): - # result[k.encode('utf-8')] \ - # = json.dumps(v).encode('utf-8') - # return result - # - # def keys(self, *args, **kwargs): - # return list([k.encode("utf-8") for k in MockedRedis.db.keys()]) - - 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))]) - - -@pytest.fixture -def client_with_mocked_data(mocker, client): - mocker.patch( - 'inventory_provider.routes.common.redis.StrictRedis', - MockedRedis) - return client - +# 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, key, value): +# MockedRedis.db[key] = value +# +# def get(self, key): +# return MockedRedis.db[key].encode('utf-8') +# +# # def hget(self, key, field): +# # value = MockedRedis.db[key] +# # return value[field].encode('utf-8') +# # # return json.dumps(value[field]).encode('utf-8') +# # +# # def hgetall(self, key): +# # result = {} +# # for k, v in MockedRedis.db[key].items(): +# # result[k.encode('utf-8')] \ +# # = json.dumps(v).encode('utf-8') +# # return result +# # +# # def keys(self, *args, **kwargs): +# # return list([k.encode("utf-8") for k in MockedRedis.db.keys()]) +# +# 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))]) + + +# @pytest.fixture +# def client_with_mocked_data(mocker, client): +# mocker.patch( +# 'inventory_provider.routes.common.redis.StrictRedis', +# MockedRedis) +# return client +# def test_router_interfaces(router, client_with_mocked_data): diff --git a/test/per_router/test_poller_routes.py b/test/per_router/test_poller_routes.py index b22bad258916ee119efceac81437bb318a025eb5..dae876447cc8a13417dd8bea5e81d7deb9718329 100644 --- a/test/per_router/test_poller_routes.py +++ b/test/per_router/test_poller_routes.py @@ -6,50 +6,50 @@ import jsonschema import inventory_provider -TEST_DATA_DIRNAME = os.path.realpath(os.path.join( - inventory_provider.__path__[0], - "..", - "test", - "data")) - +# TEST_DATA_DIRNAME = os.path.realpath(os.path.join( +# inventory_provider.__path__[0], +# "..", +# "test", +# "data")) +# DEFAULT_REQUEST_HEADERS = { "Content-type": "application/json", "Accept": ["application/json"] } -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, key, value): - MockedRedis.db[key] = value - - def get(self, key): - value = MockedRedis.db.get(key) - if value is None: - return None - return value.encode('utf-8') - - def keys(self, *args, **kwargs): - return list([k.encode("utf-8") for k in MockedRedis.db.keys()]) - - -@pytest.fixture -def client_with_mocked_data(mocker, client): - mocker.patch( - 'inventory_provider.routes.common.redis.StrictRedis', - MockedRedis) - return client - +# 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, key, value): +# MockedRedis.db[key] = value +# +# def get(self, key): +# value = MockedRedis.db.get(key) +# if value is None: +# return None +# return value.encode('utf-8') +# +# def keys(self, *args, **kwargs): +# return list([k.encode("utf-8") for k in MockedRedis.db.keys()]) +# +# +# @pytest.fixture +# def client_with_mocked_data(mocker, client): +# mocker.patch( +# 'inventory_provider.routes.common.redis.StrictRedis', +# MockedRedis) +# return client +# def test_router_interfaces(router, client_with_mocked_data):