Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_init(this_dir):
config_dir = os.path.join(this_dir, 'data', 'config', 'vodka')
vaping.daemon.Vaping(config_dir=config_dir)
plugin.get_instance("vodka")
def test_update_and_create():
"""
test that update() and create() are called accordingly during
emit()
"""
inst = plugin.get_instance(config, None)
t = time.time()
inst.emit({
"type" : "test",
"source" : "test_update_and_create",
"ts" : t,
"data" : [
{ "test" : 123, "a" : "row", "b" : "1" },
{ "test" : 456, "a" : "row", "b" : "2" }
]
})
assert inst.created == True
assert inst.updated["row-1-test"] == (t, 123)
assert inst.updated["row-2-test"] == (t, 456)
def test_whisper(tmpdir):
"""
test whisper-db creation and update from emit
"""
config["filename"] = str(tmpdir.mkdir("whisper").join(FILENAME_TEMPLATE))
inst = plugin.get_instance(config, None)
inst.start()
i = 0
t = int(time.time())-10
# emit and insert 10 data points for 2 graphs at 1 second intervals
while i < 10:
inst.emit({
"type" : "test",
"source" : "test_whisper",
"ts" : t+i,
"data" : [
{ "test" : 123+i, "id" : 1 },
{ "test" : 456+i, "id" : 2 }
]
})
i += 1
def test_time_parser(line,result,raises):
inst = plugin.get_instance(config_time_parser, None)
if raises:
with pytest.raises(ValueError) as exception_info:
_result = inst.process_line(line, {})
assert str(exception_info).find(raises) > -1
else:
_result = inst.process_line(line, {})
if not result:
assert _result == result
else:
dt = datetime.datetime.fromtimestamp(_result["ts"])
assert dt.strftime("%Y.%m.%d %H:%M:%S") == result
def test_parse_line(line,result,raises):
inst = plugin.get_instance(config, None)
if raises:
with pytest.raises(ValueError) as exception_info:
_result = inst.parse_line(line)
assert str(exception_info).find(raises) > -1
else:
_result = inst.parse_line(line)
assert _result == result
def test_plugin_instance():
with pytest.raises(ValueError):
plugin.new_plugin({}, None)
with pytest.raises(ValueError):
plugin.get_instance('nonexistant', None)
with pytest.raises(ValueError):
plugin.get_instance(['unparsable'], None)
plugin.instantiate(config['plugin'], None)
for each in config['plugin']:
if 'name' not in each:
continue
obj = plugin.get_instance(each['name'], None)
for k,v in list(each.items()):
assert v == obj.config[k]
obj = plugin.get_instance(anon_config, None)
assert obj.config == anon_config
# copy ctor
obj = plugin.get_instance('fancy_copy', None)
assert 'reeb' == obj.config['str0']
def test_plugin_instance():
with pytest.raises(ValueError):
plugin.new_plugin({}, None)
with pytest.raises(ValueError):
plugin.get_instance('nonexistant', None)
with pytest.raises(ValueError):
plugin.get_instance(['unparsable'], None)
plugin.instantiate(config['plugin'], None)
for each in config['plugin']:
if 'name' not in each:
continue
obj = plugin.get_instance(each['name'], None)
for k,v in list(each.items()):
assert v == obj.config[k]
obj = plugin.get_instance(anon_config, None)
assert obj.config == anon_config
# copy ctor
obj = plugin.get_instance('fancy_copy', None)