Skip to content
Snippets Groups Projects
Commit 43d788a7 authored by Erik Reid's avatar Erik Reid
Browse files

move mocked redis to common conftest

parent e19d48c8
No related branches found
No related tags found
No related merge requests found
Hide whitespace changes
Inline Side-by-side
test/per_router/conftest.py
+ 46
− 0
View file @ 43d788a7
import glob import glob
import json
import os import os
import re import re
from lxml import etree from lxml import etree
import pytest
import inventory_provider import inventory_provider
...@@ -13,6 +15,50 @@ TEST_DATA_DIRNAME = os.path.realpath(os.path.join( ...@@ -13,6 +15,50 @@ TEST_DATA_DIRNAME = os.path.realpath(os.path.join(
"data")) "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 pytest_generate_tests(metafunc):
def _junosspace_hosts(): def _junosspace_hosts():
......
test/per_router/test_celery_worker.py
+ 18
− 61
View file @ 43d788a7
...@@ -9,14 +9,7 @@ import re ...@@ -9,14 +9,7 @@ import re
from lxml import etree from lxml import etree
import pytest import pytest
from inventory_provider.tasks import worker from inventory_provider.tasks import worker
from inventory_provider.tasks.common import get_redis
import inventory_provider
TEST_DATA_DIRNAME = os.path.realpath(os.path.join(
inventory_provider.__path__[0],
'..',
'test',
'data'))
# class MockedRedis(object): # class MockedRedis(object):
# #
...@@ -38,52 +31,18 @@ TEST_DATA_DIRNAME = os.path.realpath(os.path.join( ...@@ -38,52 +31,18 @@ TEST_DATA_DIRNAME = os.path.realpath(os.path.join(
# return MockedRedis.db[key].encode('utf-8') # return MockedRedis.db[key].encode('utf-8')
class MockedRedis(object): def backend_db():
return get_redis({
db = None 'redis': {'hostname': None, 'port': None}
}).db
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 @pytest.fixture
def mocked_worker_module( 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 worker.InventoryTask.config = data_config
mocker.patch(
'inventory_provider.tasks.common.redis.StrictRedis',
MockedRedis)
def _mocked_snmp_interfaces(hostname, community, _): def _mocked_snmp_interfaces(hostname, community, _):
return json.loads(cached_test_data['snmp-interfaces:' + hostname]) return json.loads(cached_test_data['snmp-interfaces:' + hostname])
mocker.patch( mocker.patch(
...@@ -96,31 +55,29 @@ def mocked_worker_module( ...@@ -96,31 +55,29 @@ def mocked_worker_module(
'inventory_provider.juniper.load_config', 'inventory_provider.juniper.load_config',
_mocked_load_juniper_netconf_config) _mocked_load_juniper_netconf_config)
MockedRedis() # create an instances so db is initialized
def test_netconf_refresh_config(mocked_worker_module, router): def test_netconf_refresh_config(mocked_worker_module, router):
del MockedRedis.db['netconf:' + router] del backend_db()['netconf:' + router]
worker.netconf_refresh_config(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): 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') 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): def test_reload_router_config(mocked_worker_module, router, mocker):
saved_data = {} saved_data = {}
for key in ('netconf:' + router, 'snmp-interfaces:' + router): for key in ('netconf:' + router, 'snmp-interfaces:' + router):
saved_data[key] = MockedRedis.db.pop(key) saved_data[key] = backend_db().pop(key)
assert 'netconf:' + router not in MockedRedis.db assert 'netconf:' + router not in backend_db()
assert 'snmp-interfaces:' + router not in MockedRedis.db assert 'snmp-interfaces:' + router not in backend_db()
def _mocked_netconf_refresh_config_apply(args): def _mocked_netconf_refresh_config_apply(args):
key = 'netconf:' + args[0] key = 'netconf:' + args[0]
MockedRedis.db[key] = saved_data[key] backend_db()[key] = saved_data[key]
mocker.patch( mocker.patch(
'inventory_provider.tasks.worker.netconf_refresh_config.apply', 'inventory_provider.tasks.worker.netconf_refresh_config.apply',
_mocked_netconf_refresh_config_apply) _mocked_netconf_refresh_config_apply)
...@@ -128,7 +85,7 @@ def test_reload_router_config(mocked_worker_module, router, mocker): ...@@ -128,7 +85,7 @@ def test_reload_router_config(mocked_worker_module, router, mocker):
def _mocked_snmp_refresh_interfaces_apply(args): def _mocked_snmp_refresh_interfaces_apply(args):
assert len(args) == 2 assert len(args) == 2
key = 'snmp-interfaces:' + args[0] key = 'snmp-interfaces:' + args[0]
MockedRedis.db[key] = saved_data[key] backend_db()[key] = saved_data[key]
mocker.patch( mocker.patch(
'inventory_provider.tasks.worker.snmp_refresh_interfaces.apply', 'inventory_provider.tasks.worker.snmp_refresh_interfaces.apply',
_mocked_snmp_refresh_interfaces_apply) _mocked_snmp_refresh_interfaces_apply)
...@@ -140,5 +97,5 @@ def test_reload_router_config(mocked_worker_module, router, mocker): ...@@ -140,5 +97,5 @@ def test_reload_router_config(mocked_worker_module, router, mocker):
_mocked_update_status) _mocked_update_status)
worker.reload_router_config(router) worker.reload_router_config(router)
assert 'netconf:' + router in MockedRedis.db assert 'netconf:' + router in backend_db()
assert 'snmp-interfaces:' + router in MockedRedis.db assert 'snmp-interfaces:' + router in backend_db()
test/per_router/test_data_routes.py
+ 56
− 56
View file @ 43d788a7
...@@ -7,68 +7,68 @@ import jsonschema ...@@ -7,68 +7,68 @@ import jsonschema
import inventory_provider import inventory_provider
TEST_DATA_DIRNAME = os.path.realpath(os.path.join( # TEST_DATA_DIRNAME = os.path.realpath(os.path.join(
inventory_provider.__path__[0], # inventory_provider.__path__[0],
"..", # "..",
"test", # "test",
"data")) # "data"))
#
DEFAULT_REQUEST_HEADERS = { DEFAULT_REQUEST_HEADERS = {
"Content-type": "application/json", "Content-type": "application/json",
"Accept": ["application/json"] "Accept": ["application/json"]
} }
class MockedRedis(object): # class MockedRedis(object):
#
db = None # db = None
#
def __init__(self, *args, **kwargs): # def __init__(self, *args, **kwargs):
if MockedRedis.db is None: # if MockedRedis.db is None:
test_data_filename = os.path.join( # test_data_filename = os.path.join(
TEST_DATA_DIRNAME, # TEST_DATA_DIRNAME,
"router-info.json") # "router-info.json")
with open(test_data_filename) as f: # with open(test_data_filename) as f:
MockedRedis.db = json.loads(f.read()) # MockedRedis.db = json.loads(f.read())
#
def set(self, key, value): # def set(self, key, value):
MockedRedis.db[key] = value # MockedRedis.db[key] = value
#
def get(self, key): # def get(self, key):
return MockedRedis.db[key].encode('utf-8') # return MockedRedis.db[key].encode('utf-8')
#
# def hget(self, key, field): # # def hget(self, key, field):
# value = MockedRedis.db[key] # # value = MockedRedis.db[key]
# return value[field].encode('utf-8') # # return value[field].encode('utf-8')
# # return json.dumps(value[field]).encode('utf-8') # # # return json.dumps(value[field]).encode('utf-8')
# # #
# def hgetall(self, key): # # def hgetall(self, key):
# result = {} # # result = {}
# for k, v in MockedRedis.db[key].items(): # # for k, v in MockedRedis.db[key].items():
# result[k.encode('utf-8')] \ # # result[k.encode('utf-8')] \
# = json.dumps(v).encode('utf-8') # # = json.dumps(v).encode('utf-8')
# return result # # return result
# # #
# def keys(self, *args, **kwargs): # # def keys(self, *args, **kwargs):
# return list([k.encode("utf-8") for k in MockedRedis.db.keys()]) # # return list([k.encode("utf-8") for k in MockedRedis.db.keys()])
#
def keys(self, glob=None): # def keys(self, glob=None):
if not glob: # if not glob:
return list([k.encode("utf-8") for k in MockedRedis.db.keys()]) # return list([k.encode("utf-8") for k in MockedRedis.db.keys()])
m = re.match(r'^([^*]+)\*$', glob) # m = re.match(r'^([^*]+)\*$', glob)
assert m # all expected global are like this # assert m # all expected global are like this
return list([ # return list([
k.encode("utf-8") for k in MockedRedis.db.keys() # k.encode("utf-8") for k in MockedRedis.db.keys()
if k.startswith(m.group(1))]) # if k.startswith(m.group(1))])
@pytest.fixture # @pytest.fixture
def client_with_mocked_data(mocker, client): # def client_with_mocked_data(mocker, client):
mocker.patch( # mocker.patch(
'inventory_provider.routes.common.redis.StrictRedis', # 'inventory_provider.routes.common.redis.StrictRedis',
MockedRedis) # MockedRedis)
return client # return client
#
def test_router_interfaces(router, client_with_mocked_data): def test_router_interfaces(router, client_with_mocked_data):
......
test/per_router/test_poller_routes.py
+ 38
− 38
View file @ 43d788a7
...@@ -6,50 +6,50 @@ import jsonschema ...@@ -6,50 +6,50 @@ import jsonschema
import inventory_provider import inventory_provider
TEST_DATA_DIRNAME = os.path.realpath(os.path.join( # TEST_DATA_DIRNAME = os.path.realpath(os.path.join(
inventory_provider.__path__[0], # inventory_provider.__path__[0],
"..", # "..",
"test", # "test",
"data")) # "data"))
#
DEFAULT_REQUEST_HEADERS = { DEFAULT_REQUEST_HEADERS = {
"Content-type": "application/json", "Content-type": "application/json",
"Accept": ["application/json"] "Accept": ["application/json"]
} }
class MockedRedis(object): # class MockedRedis(object):
#
db = None # db = None
#
def __init__(self, *args, **kwargs): # def __init__(self, *args, **kwargs):
if MockedRedis.db is None: # if MockedRedis.db is None:
test_data_filename = os.path.join( # test_data_filename = os.path.join(
TEST_DATA_DIRNAME, # TEST_DATA_DIRNAME,
"router-info.json") # "router-info.json")
with open(test_data_filename) as f: # with open(test_data_filename) as f:
MockedRedis.db = json.loads(f.read()) # MockedRedis.db = json.loads(f.read())
#
def set(self, key, value): # def set(self, key, value):
MockedRedis.db[key] = value # MockedRedis.db[key] = value
#
def get(self, key): # def get(self, key):
value = MockedRedis.db.get(key) # value = MockedRedis.db.get(key)
if value is None: # if value is None:
return None # return None
return value.encode('utf-8') # return value.encode('utf-8')
#
def keys(self, *args, **kwargs): # def keys(self, *args, **kwargs):
return list([k.encode("utf-8") for k in MockedRedis.db.keys()]) # return list([k.encode("utf-8") for k in MockedRedis.db.keys()])
#
#
@pytest.fixture # @pytest.fixture
def client_with_mocked_data(mocker, client): # def client_with_mocked_data(mocker, client):
mocker.patch( # mocker.patch(
'inventory_provider.routes.common.redis.StrictRedis', # 'inventory_provider.routes.common.redis.StrictRedis',
MockedRedis) # MockedRedis)
return client # return client
#
def test_router_interfaces(router, client_with_mocked_data): def test_router_interfaces(router, client_with_mocked_data):
......
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment