<commit_before>import multiprocessing
import threading
import Queue

import pytest
from mock import patch

import pplugins

DEFAULT = "default"


def test_pluginmanager_abstract():
    with pytest.raises(TypeError):
        pplugins.PluginManager()


def test_pluginrunner_abstract():
    with pytest.raises(TypeError):
        pplugins.PluginRunner()


def test_plugin_abstract():
    with pytest.raises(TypeError):
        pplugins.Plugin()


def test_pluginerror_plugin_property():
    try:
        raise pplugins.PluginError("Generic error", "test")
    except pplugins.PluginError as e:
        assert e.plugin == "test"


def test_pluginerror_str_contains_plugin_name():
    plugin_name = "test"

    try:
        raise pplugins.PluginError(None, plugin_name)
    except pplugins.PluginError as e:
        assert plugin_name in str(e)


def test_plugininterface_add_message():
    msg = "TEST"

    with patch.object(Queue.Queue, 'put', return_Value=None) as mock_method:
        queue = Queue.Queue()

        pi = pplugins.PluginInterface(None, queue)
        pi.add_message(msg)

    mock_method.assert_called_once_with(msg)


@pytest.mark.parametrize("params,exp_block,exp_timeout", (
    ({}, True, None),
    ({'block': None, 'timeout': None}, None, None),
    ({'block': True, 'timeout': False}, True, False),
    ({'block': False, 'timeout': True}, False, True)
))
def test_plugininterface_get_event(params, exp_block, exp_timeout):
    return_value = 'TEST'
    with patch.object(
            Queue.Queue, 'get', return_value=return_value) as mock_method:
        queue = Queue.Queue()

        pi = pplugins.PluginInterface(queue, None)
        assert pi.get_event(**params) == return_value

    mock_method.assert_called_once_with(exp_block, exp_timeout)


@patch.multiple(pplugins.Plugin, __abstractmethods__=set())
@patch.object(pplugins.Plugin, 'run')
def test_plugin_constructor(mock_method):
    interface = "TEST"

    plugin = pplugins.Plugin(interface)
    assert plugin.interface == interface

    mock_method.assert_called_once_with()


@patch.multiple(pplugins.PluginRunner, __abstractmethods__=set())
def test_pluginrunner_daemon_flag():
    pr = pplugins.PluginRunner(None, None, None)

    # multiprocessing daemon flag
    assert pr.daemon is True


@pytest.mark.parametrize("plugin", (
    type('Module', (), {}),
    # Testing that pplugins.Plugin doesn't count as a valid plugin
    pplugins,
))
@patch.multiple(pplugins.PluginRunner, __abstractmethods__=set())
def test_pluginrunner_run_exception(plugin):
    pr = pplugins.PluginRunner('test', None, None)
    with patch.object(pplugins.PluginRunner, '_load_plugin',
                      return_value=plugin) as load_plugin_mock:
        with pytest.raises(pplugins.PluginError) as excinfo:
            pr.run()

    # assert the overrided _load_plugin() got called
    load_plugin_mock.assert_called_once_with()

    # assert that an exception was raised that we were unable to find a plugin
    assert str(pr.plugin_class) in str(excinfo.value)
    assert 'Unable to find' in str(excinfo.value)


@pytest.fixture(params=["EmptyPlugin", "ErrorPlugin"])
def plugin(request):
    class EmptyPlugin(pplugins.Plugin):
        def run(self):
            pass

    class ErrorPlugin(pplugins.Plugin):
        def run(self):
            raise Exception

    # yeah, yeah, it's eval()
    return type('Module', (), {request.param: eval(request.param)})


@patch.multiple(pplugins.PluginRunner, __abstractmethods__=set())
def test_pluginrunner_run(plugin):
    pr = pplugins.PluginRunner(None, None, None)
    with patch.object(pplugins.PluginRunner, '_load_plugin',
                      return_value=plugin):
        pr.run()


@patch.multiple(pplugins.PluginManager, __abstractmethods__=set())
def test_pluginmanager_constructor():
    threads = threading.active_count()
    pplugins.PluginManager()

    # plugin manager should not have started the reaping thread when called
    # through the constructor
    assert threading.active_count() == threads


@patch.multiple(pplugins.PluginManager, __abstractmethods__=set())
def test_pluginmanager_contextmanager():
    threads = threading.active_count()
    with pplugins.PluginManager():
        # assert that reaping thread was started
        assert threading.active_count() == threads + 1

    # assert that reaping thread was stopped
    assert threading.active_count() == threads


@patch.multiple(pplugins.PluginManager, __abstractmethods__=set())
@patch.object(multiprocessing.Process, 'is_alive', return_value=False)
def test_pluginmanager_reap_plugins(mock_is_alive):
    pm = pplugins.PluginManager()
    plugins = pm.plugins

    pm.plugins = dict(test={'process': multiprocessing.Process()},
                      **pm.plugins)
    pm.reap_plugins()

    mock_is_alive.assert_called_once_with()
    assert pm.plugins == plugins<commit_msg>Clean up reap_plugins() test<commit_after>import multiprocessing
import threading
import Queue

import pytest
from mock import patch

import pplugins

DEFAULT = "default"


def test_pluginmanager_abstract():
    with pytest.raises(TypeError):
        pplugins.PluginManager()


def test_pluginrunner_abstract():
    with pytest.raises(TypeError):
        pplugins.PluginRunner()


def test_plugin_abstract():
    with pytest.raises(TypeError):
        pplugins.Plugin()


def test_pluginerror_plugin_property():
    try:
        raise pplugins.PluginError("Generic error", "test")
    except pplugins.PluginError as e:
        assert e.plugin == "test"


def test_pluginerror_str_contains_plugin_name():
    plugin_name = "test"

    try:
        raise pplugins.PluginError(None, plugin_name)
    except pplugins.PluginError as e:
        assert plugin_name in str(e)


def test_plugininterface_add_message():
    msg = "TEST"

    with patch.object(Queue.Queue, 'put', return_Value=None) as mock_method:
        queue = Queue.Queue()

        pi = pplugins.PluginInterface(None, queue)
        pi.add_message(msg)

    mock_method.assert_called_once_with(msg)


@pytest.mark.parametrize("params,exp_block,exp_timeout", (
    ({}, True, None),
    ({'block': None, 'timeout': None}, None, None),
    ({'block': True, 'timeout': False}, True, False),
    ({'block': False, 'timeout': True}, False, True)
))
def test_plugininterface_get_event(params, exp_block, exp_timeout):
    return_value = 'TEST'
    with patch.object(
            Queue.Queue, 'get', return_value=return_value) as mock_method:
        queue = Queue.Queue()

        pi = pplugins.PluginInterface(queue, None)
        assert pi.get_event(**params) == return_value

    mock_method.assert_called_once_with(exp_block, exp_timeout)


@patch.multiple(pplugins.Plugin, __abstractmethods__=set())
@patch.object(pplugins.Plugin, 'run')
def test_plugin_constructor(mock_method):
    interface = "TEST"

    plugin = pplugins.Plugin(interface)
    assert plugin.interface == interface

    mock_method.assert_called_once_with()


@patch.multiple(pplugins.PluginRunner, __abstractmethods__=set())
def test_pluginrunner_daemon_flag():
    pr = pplugins.PluginRunner(None, None, None)

    # multiprocessing daemon flag
    assert pr.daemon is True


@pytest.mark.parametrize("plugin", (
    type('Module', (), {}),
    # Testing that pplugins.Plugin doesn't count as a valid plugin
    pplugins,
))
@patch.multiple(pplugins.PluginRunner, __abstractmethods__=set())
def test_pluginrunner_run_exception(plugin):
    pr = pplugins.PluginRunner('test', None, None)
    with patch.object(pplugins.PluginRunner, '_load_plugin',
                      return_value=plugin) as load_plugin_mock:
        with pytest.raises(pplugins.PluginError) as excinfo:
            pr.run()

    # assert the overrided _load_plugin() got called
    load_plugin_mock.assert_called_once_with()

    # assert that an exception was raised that we were unable to find a plugin
    assert str(pr.plugin_class) in str(excinfo.value)
    assert 'Unable to find' in str(excinfo.value)


@pytest.fixture(params=["EmptyPlugin", "ErrorPlugin"])
def plugin(request):
    class EmptyPlugin(pplugins.Plugin):
        def run(self):
            pass

    class ErrorPlugin(pplugins.Plugin):
        def run(self):
            raise Exception

    # yeah, yeah, it's eval()
    return type('Module', (), {request.param: eval(request.param)})


@patch.multiple(pplugins.PluginRunner, __abstractmethods__=set())
def test_pluginrunner_run(plugin):
    pr = pplugins.PluginRunner(None, None, None)
    with patch.object(pplugins.PluginRunner, '_load_plugin',
                      return_value=plugin):
        pr.run()


@patch.multiple(pplugins.PluginManager, __abstractmethods__=set())
def test_pluginmanager_constructor():
    threads = threading.active_count()
    pplugins.PluginManager()

    # plugin manager should not have started the reaping thread when called
    # through the constructor
    assert threading.active_count() == threads


@patch.multiple(pplugins.PluginManager, __abstractmethods__=set())
def test_pluginmanager_contextmanager():
    threads = threading.active_count()
    with pplugins.PluginManager():
        # assert that reaping thread was started
        assert threading.active_count() == threads + 1

    # assert that reaping thread was stopped
    assert threading.active_count() == threads


@patch.multiple(pplugins.PluginManager, __abstractmethods__=set())
def test_pluginmanager_reap_plugins():
    pm = pplugins.PluginManager()
    plugins = dict(test={'process': multiprocessing.Process()},
                   **pm.plugins)

    pm.plugins = plugins
    with patch.object(multiprocessing.Process, 'is_alive', return_value=False):
        pm.reap_plugins()

    assert pm.plugins == {}

    pm.plugins = plugins
    with patch.object(multiprocessing.Process, 'is_alive', return_value=True):
        pm.reap_plugins()

    assert pm.plugins == plugins
{'fraction_non_alphanumeric': 0.1083594460729809, 'fraction_numerical': 0.00020216314565854644, 'mean_word_length': 3.9077380952380953, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2528629', 'n_tokens_mistral': 2987, 'n_tokens_neox': 2982, 'n_words': 651}