File: test_event_loop_fixture.py

package info (click to toggle)
python-pytest-asyncio 1.1.0-1
  • links: PTS, VCS
  • area: main
  • in suites:
  • size: 616 kB
  • sloc: python: 2,421; makefile: 24; sh: 1
file content (143 lines) | stat: -rw-r--r-- 4,374 bytes parent folder | download | duplicates (2)
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
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
from __future__ import annotations

from textwrap import dedent

from pytest import Pytester


def test_event_loop_fixture_respects_event_loop_policy(pytester: Pytester):
    pytester.makeconftest(
        dedent(
            """\
            '''Defines and sets a custom event loop policy'''
            import asyncio
            from asyncio import DefaultEventLoopPolicy, SelectorEventLoop

            class TestEventLoop(SelectorEventLoop):
                pass

            class TestEventLoopPolicy(DefaultEventLoopPolicy):
                def new_event_loop(self):
                    return TestEventLoop()

            # This statement represents a code which sets a custom event loop policy
            asyncio.set_event_loop_policy(TestEventLoopPolicy())
            """
        )
    )
    pytester.makepyfile(
        dedent(
            """\
            '''Tests that any externally provided event loop policy remains unaltered'''
            import asyncio

            import pytest


            @pytest.mark.asyncio
            async def test_uses_loop_provided_by_custom_policy():
                '''Asserts that test cases use the event loop
                provided by the custom event loop policy'''
                assert type(asyncio.get_event_loop()).__name__ == "TestEventLoop"


            @pytest.mark.asyncio
            async def test_custom_policy_is_not_overwritten():
                '''
                Asserts that any custom event loop policy stays the same
                across test cases.
                '''
                assert type(asyncio.get_event_loop()).__name__ == "TestEventLoop"
            """
        )
    )
    result = pytester.runpytest_subprocess("--asyncio-mode=strict")
    result.assert_outcomes(passed=2)


def test_event_loop_fixture_handles_unclosed_async_gen(
    pytester: Pytester,
):
    pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function")
    pytester.makepyfile(
        dedent(
            """\
            import asyncio
            import pytest

            pytest_plugins = 'pytest_asyncio'

            @pytest.mark.asyncio
            async def test_something():
                async def generator_fn():
                    yield
                    yield

                gen = generator_fn()
                await gen.__anext__()
            """
        )
    )
    result = pytester.runpytest_subprocess("--asyncio-mode=strict", "-W", "default")
    result.assert_outcomes(passed=1, warnings=0)


def test_closing_event_loop_in_sync_fixture_teardown_raises_warning(
    pytester: Pytester,
):
    pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function")
    pytester.makepyfile(
        dedent(
            """\
            import asyncio
            import pytest
            import pytest_asyncio
            pytest_plugins = 'pytest_asyncio'

            @pytest_asyncio.fixture
            async def _event_loop():
                return asyncio.get_running_loop()

            @pytest.fixture
            def close_event_loop(_event_loop):
                yield
                # fixture has its own cleanup code
                _event_loop.close()

            @pytest.mark.asyncio
            async def test_something(close_event_loop):
                await asyncio.sleep(0.01)
            """
        )
    )
    result = pytester.runpytest_subprocess("--asyncio-mode=strict")
    result.assert_outcomes(passed=1, warnings=1)
    result.stdout.fnmatch_lines(
        ["*An exception occurred during teardown of an asyncio.Runner*"]
    )


def test_event_loop_fixture_asyncgen_error(
    pytester: Pytester,
):
    pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function")
    pytester.makepyfile(
        dedent(
            """\
            import asyncio
            import pytest

            pytest_plugins = 'pytest_asyncio'

            @pytest.mark.asyncio
            async def test_something():
                # mock shutdown_asyncgen failure
                loop = asyncio.get_running_loop()
                async def fail():
                    raise RuntimeError("mock error cleaning up...")
                loop.shutdown_asyncgens = fail
            """
        )
    )
    result = pytester.runpytest_subprocess("--asyncio-mode=strict", "-W", "default")
    result.assert_outcomes(passed=1, warnings=1)