asyncio_monkey/tests.py at master · litwisha/asyncio_monkey · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
import asyncio
from functools import partial
from unittest import mock

import pytest

from asyncio_monkey import (
    PY_360, patch_all, patch_log_destroy_pending, patch_get_event_loop,
)


def create_task(*, loop=None):
    if loop is None:
        loop = asyncio.get_event_loop()

    try:
        return loop.create_task
    except AttributeError:
        try:
            return partial(asyncio.ensure_future, loop=loop)
        except AttributeError:
            return partial(getattr(asyncio, 'async'), loop=loop)


def test_patch_log_destroy_pending():
    assert not hasattr(asyncio.Task, 'patched')

    fut = asyncio.Future()
    task = create_task()(fut)

    assert task._log_destroy_pending

    task._log_destroy_pending = False

    assert not task._log_destroy_pending

    task.cancel()
    fut.cancel()

    ###

    patch_log_destroy_pending()
    patch_log_destroy_pending()

    ###

    assert hasattr(asyncio.Task, 'patched')

    fut = asyncio.Future()
    task = create_task()(fut)

    assert task._log_destroy_pending

    task._log_destroy_pending = False

    assert task._log_destroy_pending

    task.cancel()
    fut.cancel()


def test_get_event_loop():
    if not PY_360:
        return

    @asyncio.coroutine
    def coro():
        loop = asyncio.get_event_loop()

        return loop

    asyncio.set_event_loop(None)

    loop = asyncio.new_event_loop()

    ###

    assert not hasattr(asyncio.get_event_loop, 'patched')

    running_loop = loop.run_until_complete(coro())

    assert running_loop is loop

    ###

    patch_get_event_loop()
    patch_get_event_loop()

    ###

    assert hasattr(asyncio.get_event_loop, 'patched')

    with pytest.raises(RuntimeError):
        loop.run_until_complete(coro())

    ###

    loop.close()


def test_patch_all():
    with mock.patch('asyncio_monkey.patch_get_event_loop') as mocked_patch_get_event_loop, mock.patch('asyncio_monkey.patch_log_destroy_pending') as mocked_patch_log_destroy_pending:  # noqa

        patch_all()

        assert mocked_patch_get_event_loop.called_once()
        assert mocked_patch_log_destroy_pending.called_once()