How to use the vaping.plugin.instantiate function in vaping

To help you get started, we’ve selected a few vaping examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github 20c / vaping / tests / test_plugin.py View on Github external
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']

    obj = plugin.get_instance({'fancy_copy': {'var0': 'luggage'}}, None)
    assert 'reeb' == obj.config['str0']
github 20c / vaping / tests / test_plugin.py View on Github external
def test_emission_queuing():
    plugin.instantiate(config["plugin"], None)
    queue_tester = plugin.get_instance("queue_tester")
    emit_store1 = plugin.get_output("emit_store1", None)
    emit_store2 = plugin.get_output("emit_store2", None)

    # daemon handles this, do it manually here since
    # we are not starting the daemon
    queue_tester._emit = [emit_store1, emit_store2]

    message = queue_tester.probe()
    queue_tester.queue_emission(message)

    queue_tester.send_emission()
    assert emit_store1.store[0] == message
    assert emit_store2.store == []

    queue_tester.send_emission()
github 20c / vaping / vaping / daemon.py View on Github external
if not os.path.exists(self.home_dir):
            raise ValueError("home directory '{}' does not exist".format(self.home_dir))

        if not os.access(self.home_dir, os.W_OK):
            raise ValueError("home directory '{}' is not writable".format(self.home_dir))

        # change to home for working dir
        os.chdir(self.home_dir)

        # instantiate all defined plugins
        # TODO remove and let them lazy init?
        plugins = self.config.get('plugins', None)
        if not plugins:
            raise ValueError('no plugins specified')

        plugin.instantiate(self.config['plugins'], self.plugin_context)

        # check that probes don't name clash with plugins
        for probe in self.config.get('probes', []):
            if plugin.exists(probe["name"]):
                raise ValueError("probes may not share names with plugins ({})".format(probe["name"]))

        # TODO move to daemon
        pidname = vcfg.get('pidfile', 'vaping.pid')
        self.pidfile = pidfile.PidFile(pidname=pidname, piddir=self.home_dir)