<commit_before>"""Tests for aiohttp/worker.py"""
import asyncio
import sys
from unittest import mock

import pytest

from aiohttp import helpers

base_worker = pytest.importorskip('aiohttp.worker')
pytest.importorskip('uvloop')


class BaseTestWorker:

    def __init__(self):
        self.servers = []
        self.exit_code = 0
        self.cfg = mock.Mock()
        self.cfg.graceful_timeout = 100


class AsyncioWorker(BaseTestWorker, base_worker.GunicornWebWorker):
    pass


class UvloopWorker(BaseTestWorker, base_worker.GunicornUVLoopWebWorker):

    def __init__(self):
        if sys.version_info < (3, 5) \
                or sys.platform in ('win32', 'cygwin', 'cli'):
            raise pytest.skip("uvloop requires Python 3.5 and *nix.")

        super().__init__()


@pytest.fixture(params=[AsyncioWorker, UvloopWorker])
def worker(request):
    return request.param()


def test_init_process(worker):
    with mock.patch('aiohttp.worker.asyncio') as m_asyncio:
        try:
            worker.init_process()
        except TypeError:
            pass

        assert m_asyncio.get_event_loop.return_value.close.called
        assert m_asyncio.new_event_loop.called
        assert m_asyncio.set_event_loop.called


def test_run(worker, loop):
    worker.loop = loop
    worker._run = mock.Mock(
        wraps=asyncio.coroutine(lambda: None))
    with pytest.raises(SystemExit):
        worker.run()

    assert worker._run.called
    assert loop.is_closed()


def test_handle_quit(worker):<commit_msg>Fix test_worker problem with pytest==3.0<commit_after>"""Tests for aiohttp/worker.py"""
import asyncio
import sys
from unittest import mock

import pytest

from aiohttp import helpers

base_worker = pytest.importorskip('aiohttp.worker')

try:
    import uvloop
except ImportError:
    uvloop = None


class BaseTestWorker:

    def __init__(self):
        self.servers = []
        self.exit_code = 0
        self.cfg = mock.Mock()
        self.cfg.graceful_timeout = 100


class AsyncioWorker(BaseTestWorker, base_worker.GunicornWebWorker):
    pass


PARAMS = [AsyncioWorker]
if uvloop is not None:
    class UvloopWorker(BaseTestWorker, base_worker.GunicornUVLoopWebWorker):
        pass

    PARAMS.append(UvloopWorker)


@pytest.fixture(params=PARAMS)
def worker(request):
    return request.param()


def test_init_process(worker):
    with mock.patch('aiohttp.worker.asyncio') as m_asyncio:
        try:
            worker.init_process()
        except TypeError:
            pass

        assert m_asyncio.get_event_loop.return_value.close.called
        assert m_asyncio.new_event_loop.called
        assert m_asyncio.set_event_loop.called


def test_run(worker, loop):
    worker.loop = loop
    worker._run = mock.Mock(
        wraps=asyncio.coroutine(lambda: None))
    with pytest.raises(SystemExit):
        worker.run()

    assert worker._run.called
    assert loop.is_closed()


def test_handle_quit(worker):
{'fraction_non_alphanumeric': 0.09486569194151649, 'fraction_numerical': 0.005440326419585175, 'mean_word_length': 3.1612446958981613, 'pattern_counts': {'":': 0, '<': 4, '<?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': '5860743', 'n_tokens_mistral': 972, 'n_tokens_neox': 939, 'n_words': 195}