<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.
# -----------------------------------------------------------------------------
"""Generic utility functions for testing.

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

import os
from time import sleep
from contextlib import contextmanager
from pprint import pformat

import enaml
from configobj import ConfigObj
from enaml.application import timed_call
from enaml.qt.qt_application import QtApplication
from enaml.widgets.api import Window, Dialog
with enaml.imports():
    from enaml.stdlib.message_box import MessageBox


APP_PREFERENCES = os.path.join('app', 'preferences')
APP_DIR_CONFIG = 'app_directory.ini'


def ecpy_path():
    """Get the ecpy path as determined by the sys_path fixture.

    """
    from .fixtures import ECPY
    assert ECPY
    return ECPY


def process_app_events():
    """Manually run the Qt event loop so that windows are shown and event
    propagated.

    """
    qapp = QtApplication.instance()._qapp
    qapp.flush()
    qapp.processEvents()
    qapp.sendPostedEvents()


def get_window(cls=Window):
    """Convenience function running the event loop and returning the first
    window found in the set of active windows.

    Parameters
    ----------
    cls : type, optional
        Type of the window which should be returned.

    """
    sleep(0.1)
    process_app_events()
    w_ = None
    for w in Window.windows:
        if isinstance(w, cls):
            w_ = w
            break

    return w_


def close_all_windows():
    """Close all opened windows.

    This should be used by all tests creating windows in a teardown step.

    """
    process_app_events()
    sleep(0.1)
    while Window.windows:
        for window in list(Window.windows):
            window.close()
        process_app_events()
        sleep(0.02)


@contextmanager
def handle_dialog(op='accept', custom=lambda x: x, cls=Dialog, time=100,
                  skip_answer=False):
    """Automatically close a dialog opened during the context.

    Parameters
    ----------
    op : {'accept', 'reject'}, optional
        Whether to accept or reject the dialog.

    custom : callable, optional
        Callable taking as only argument the dialog, called before accepting
        or rejecting the dialog.

    cls : type, optional
        Dialog class to identify.

    time : float, optional
        Time to wait before handling the dialog in ms.

    skip_answer : bool, optional
        Skip answering to the dialog. If this is True the handler should handle
        the answer itself.

    """
    def close_dialog():
        i = 0
        while True:
            dial = get_window(cls)
            if dial is not None:
                break
            elif i > 10:
                raise Exception('Dialog timeout')
            sleep(0.1)
            i += 1

        try:
            custom(dial)
        finally:
            process_app_events()
            from .fixtures import DIALOG_SLEEP
            sleep(DIALOG_SLEEP)
            getattr(dial, op)()
    timed_call(time, close_dialog)
    yield
    process_app_events()


@contextmanager
def handle_question(answer):
    """Handle question dialog.
<commit_msg>testing: handle dialog block execution till handler called (#95)<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.
# -----------------------------------------------------------------------------
"""Generic utility functions for testing.

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

import os
from time import sleep
from contextlib import contextmanager
from pprint import pformat

import enaml
from configobj import ConfigObj
from enaml.application import timed_call
from enaml.qt.qt_application import QtApplication
from enaml.widgets.api import Window, Dialog
with enaml.imports():
    from enaml.stdlib.message_box import MessageBox


APP_PREFERENCES = os.path.join('app', 'preferences')
APP_DIR_CONFIG = 'app_directory.ini'


def ecpy_path():
    """Get the ecpy path as determined by the sys_path fixture.

    """
    from .fixtures import ECPY
    assert ECPY
    return ECPY


def process_app_events():
    """Manually run the Qt event loop so that windows are shown and event
    propagated.

    """
    qapp = QtApplication.instance()._qapp
    qapp.flush()
    qapp.processEvents()
    qapp.sendPostedEvents()


def get_window(cls=Window):
    """Convenience function running the event loop and returning the first
    window found in the set of active windows.

    Parameters
    ----------
    cls : type, optional
        Type of the window which should be returned.

    """
    sleep(0.1)
    process_app_events()
    w_ = None
    for w in Window.windows:
        if isinstance(w, cls):
            w_ = w
            break

    return w_


def close_all_windows():
    """Close all opened windows.

    This should be used by all tests creating windows in a teardown step.

    """
    process_app_events()
    sleep(0.1)
    while Window.windows:
        for window in list(Window.windows):
            window.close()
        process_app_events()
        sleep(0.02)


class ScheduledClosing(object):
    """Scheduled closing of dialog.

    """

    def __init__(self, cls, custom, op):
        self.called = False
        self.cls = cls
        self.custom = custom
        self.op = op

    def __call__(self):
        i = 0
        while True:
            dial = get_window(self.cls)
            if dial is not None:
                break
            elif i > 10:
                raise Exception('Dialog timeout')
            sleep(0.1)
            i += 1

        try:
            self.custom(dial)
        finally:
            process_app_events()
            from .fixtures import DIALOG_SLEEP
            sleep(DIALOG_SLEEP)
            getattr(dial, self.op)()
            self.called = True


@contextmanager
def handle_dialog(op='accept', custom=lambda x: x, cls=Dialog, time=100,
                  skip_answer=False):
    """Automatically close a dialog opened during the context.

    Parameters
    ----------
    op : {'accept', 'reject'}, optional
        Whether to accept or reject the dialog.

    custom : callable, optional
        Callable taking as only argument the dialog, called before accepting
        or rejecting the dialog.

    cls : type, optional
        Dialog class to identify.

    time : float, optional
        Time to wait before handling the dialog in ms.

    skip_answer : bool, optional
        Skip answering to the dialog. If this is True the handler should handle
        the answer itself.

    """
    sch = ScheduledClosing(cls, custom, op)
    timed_call(time, sch)
    try:
        yield
    except Exception:
        raise
    else:
        while not sch.called:
            process_app_events()


@contextmanager
def handle_question(answer):
    """Handle question dialog.

{'fraction_non_alphanumeric': 0.11817212896349587, 'fraction_numerical': 0.005861977084998668, 'mean_word_length': 2.027016129032258, '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': '3996327', 'n_tokens_mistral': 2043, 'n_tokens_neox': 1951, 'n_words': 713}