<commit_before># -*- coding: utf-8 -*-
# -----------------------------------------------------------------------------
# Copyright 2015 by Ecpy Authors, see AUTHORS for more details.
#
# Distributed under the terms of the BSD license.
#
# The full license is in the file LICENCE, distributed with this software.
# -----------------------------------------------------------------------------
"""Test the working of the measure processor.

"""
from __future__ import (division, unicode_literals, print_function,
                        absolute_import)

from time import sleep

import enaml
import pytest
from future.builtins import str
from threading import Thread

from ecpy.measure.measure import Measure
from ecpy.tasks.api import RootTask

from ..util import (ErrorDialogException, process_app_events)

with enaml.imports():
    from enaml.workbench.ui.ui_manifest import UIManifest

    from ecpy.tasks.manager.manifest import TasksManagerManifest
    from .contributions import Flags


@pytest.fixture
def measure_with_tools(measure, tmpdir):
    """Create a measure with all dummy tools attached.

    """
    measure.add_tool('pre-hook', 'dummy')
    measure.add_tool('post-hook', 'dummy')
    measure.add_tool('monitor', 'dummy')
    measure.root_task.default_path = str(tmpdir)
    return measure


@pytest.fixture
def processor(windows, measure_workbench, measure):
    """Fixture starting the measure plugin and returning the processor.

    Use app because we need run the event loop

    """
    # measure ensures that contributions are there
    measure_workbench.register(UIManifest())
    measure_workbench.register(TasksManagerManifest())
    plugin = measure_workbench.get_plugin('ecpy.measure')
    plugin.selected_engine = 'dummy'
    return plugin.processor


def wait_and_process(waiting_function):
    """Call a function which can timeout and process app events.

    """
    i = 0
    while not waiting_function(timeout=0.001):
        process_app_events()
        i += 1
        if i > 1000:
            assert False
    process_app_events()


def test_setting_continuous_processing(processor):
    """Test that the post-setter does update the flag.

    """
    processor.continuous_processing = False
    assert not processor._state.test('continuous_processing')
    processor.continuous_processing = True
    assert processor._state.test('continuous_processing')


@pytest.mark.timeout(1)
def test_starting_measure_no_measure_enqueued(processor):
    """Test starting next measure in the queue when no measures are enqueued.

    """
    processor.start_measure(None)
    assert processor.active
    processor._thread.join()
    process_app_events()
    assert not processor.active<commit_msg>measure : make timeout for processor tests longer.<commit_after># -*- coding: utf-8 -*-
# -----------------------------------------------------------------------------
# Copyright 2015 by Ecpy Authors, see AUTHORS for more details.
#
# Distributed under the terms of the BSD license.
#
# The full license is in the file LICENCE, distributed with this software.
# -----------------------------------------------------------------------------
"""Test the working of the measure processor.

"""
from __future__ import (division, unicode_literals, print_function,
                        absolute_import)

from time import sleep

import enaml
import pytest
from future.builtins import str
from threading import Thread

from ecpy.measure.measure import Measure
from ecpy.tasks.api import RootTask

from ..util import (ErrorDialogException, process_app_events)

with enaml.imports():
    from enaml.workbench.ui.ui_manifest import UIManifest

    from ecpy.tasks.manager.manifest import TasksManagerManifest
    from .contributions import Flags


@pytest.fixture
def measure_with_tools(measure, tmpdir):
    """Create a measure with all dummy tools attached.

    """
    measure.add_tool('pre-hook', 'dummy')
    measure.add_tool('post-hook', 'dummy')
    measure.add_tool('monitor', 'dummy')
    measure.root_task.default_path = str(tmpdir)
    return measure


@pytest.fixture
def processor(windows, measure_workbench, measure):
    """Fixture starting the measure plugin and returning the processor.

    Use app because we need run the event loop

    """
    # measure ensures that contributions are there
    measure_workbench.register(UIManifest())
    measure_workbench.register(TasksManagerManifest())
    plugin = measure_workbench.get_plugin('ecpy.measure')
    plugin.selected_engine = 'dummy'
    return plugin.processor


def wait_and_process(waiting_function):
    """Call a function which can timeout and process app events.

    """
    i = 0
    while not waiting_function(timeout=0.001):
        process_app_events()
        i += 1
        if i > 2000:
            assert False
    process_app_events()


def test_setting_continuous_processing(processor):
    """Test that the post-setter does update the flag.

    """
    processor.continuous_processing = False
    assert not processor._state.test('continuous_processing')
    processor.continuous_processing = True
    assert processor._state.test('continuous_processing')


@pytest.mark.timeout(1)
def test_starting_measure_no_measure_enqueued(processor):
    """Test starting next measure in the queue when no measures are enqueued.

    """
    processor.start_measure(None)
    assert processor.active
    processor._thread.join()
    process_app_events()
    assert not processor.active
{'fraction_non_alphanumeric': 0.1350118591497902, 'fraction_numerical': 0.0058383506659368724, 'mean_word_length': 2.9695872556118754, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, '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': '5502819', 'n_tokens_mistral': 1470, 'n_tokens_neox': 1413, 'n_words': 439}