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_run_probe(this_dir):
config_dir = os.path.join(this_dir, 'data', 'config', 'fping')
daemon = vaping.daemon.Vaping(config_dir=config_dir)
probes = daemon.config.get('probes', None)
fping = plugin.get_probe(probes[0], daemon.plugin_context)
msg = fping.probe()
print(msg)
def test_run_probe(config_dir):
config_dir = os.path.join(config_dir, "fping_mtr")
daemon = vaping.daemon.Vaping(config_dir=config_dir)
probes = daemon.config.get("probes", None)
mtr = plugin.get_probe(probes[0], daemon.plugin_context)
msg = mtr.probe()
print(msg)
def test_probe_plugin_name(config_dir):
"""
checks that vaping correct errors if a probe is named the same as a plugin
"""
config_dir = os.path.join(config_dir, "dupe")
with pytest.raises(ValueError) as excinfo:
vaping.daemon.Vaping(config_dir=config_dir)
assert "probes may not share names with plugins" in str(excinfo.value)
def test_start_stop(this_dir):
config_dir = os.path.join(this_dir, "data", "config", "fping")
vaping.daemon.Vaping(config=vaping.Config(read=config_dir))
# daemon._main()
def test_plugin_context():
data = {
'1': 'two'
}
cfg = vaping.Config(data=data)
ctx = vaping.daemon.PluginContext(cfg)
assert data == ctx.config.data
# test immutable
ctx.config.data['1'] = 'three'
assert data != ctx.config.data
def mk_daemon(ctx):
"""
Return a daemon.Vaping instance
**Arguments**
- ctx (`Context`): vaping `click` context instance
**Returns**
`vaping.daemon.Vaping` instance
"""
if not ctx.config.meta:
raise ValueError("no config specified, please use specify a home directory")
return vaping.daemon.Vaping(ctx.config)
"""
kwargs = {
'pidfile': self.pidfile,
'working_directory': self.home_dir,
}
# FIXME - doesn't work
if not detach:
kwargs.update({
'detach_process': False,
'files_preserve': [0,1,2],
'stdout': sys.stdout,
'stderr': sys.stderr,
})
ctx = daemon.DaemonContext(**kwargs)
with ctx:
self._main()