-
Notifications
You must be signed in to change notification settings - Fork 139
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[feat] Add support for package-scoped loops.
Signed-off-by: Michael Seifert <m.seifert@digitalernachschub.de>
- Loading branch information
Showing
4 changed files
with
226 additions
and
8 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,191 @@ | ||
| from textwrap import dedent | ||
|
|
||
| from pytest import Pytester | ||
|
|
||
|
|
||
| def test_asyncio_mark_provides_package_scoped_loop_strict_mode(pytester: Pytester): | ||
| package_name = pytester.path.name | ||
| pytester.makepyfile( | ||
| __init__="", | ||
| shared_module=dedent( | ||
| """\ | ||
| import asyncio | ||
| loop: asyncio.AbstractEventLoop = None | ||
| """ | ||
| ), | ||
| test_module_one=dedent( | ||
| f"""\ | ||
| import asyncio | ||
| import pytest | ||
| from {package_name} import shared_module | ||
| @pytest.mark.asyncio(scope="package") | ||
| async def test_remember_loop(): | ||
| shared_module.loop = asyncio.get_running_loop() | ||
| """ | ||
| ), | ||
| test_module_two=dedent( | ||
| f"""\ | ||
| import asyncio | ||
| import pytest | ||
| from {package_name} import shared_module | ||
| pytestmark = pytest.mark.asyncio(scope="package") | ||
| async def test_this_runs_in_same_loop(): | ||
| assert asyncio.get_running_loop() is shared_module.loop | ||
| class TestClassA: | ||
| async def test_this_runs_in_same_loop(self): | ||
| assert asyncio.get_running_loop() is shared_module.loop | ||
| """ | ||
| ), | ||
| ) | ||
| result = pytester.runpytest("--asyncio-mode=strict") | ||
| result.assert_outcomes(passed=3) | ||
|
|
||
|
|
||
| def test_raise_when_event_loop_fixture_is_requested_in_addition_to_scoped_loop( | ||
| pytester: Pytester, | ||
| ): | ||
| pytester.makepyfile( | ||
| __init__="", | ||
| test_raises=dedent( | ||
| """\ | ||
| import asyncio | ||
| import pytest | ||
| @pytest.mark.asyncio(scope="package") | ||
| async def test_remember_loop(event_loop): | ||
| pass | ||
| """ | ||
| ), | ||
| ) | ||
| result = pytester.runpytest("--asyncio-mode=strict") | ||
| result.assert_outcomes(errors=1) | ||
| result.stdout.fnmatch_lines("*MultipleEventLoopsRequestedError: *") | ||
|
|
||
|
|
||
| def test_asyncio_mark_respects_the_loop_policy( | ||
| pytester: Pytester, | ||
| ): | ||
| pytester.makepyfile( | ||
| __init__="", | ||
| conftest=dedent( | ||
| """\ | ||
| import pytest | ||
| from .custom_policy import CustomEventLoopPolicy | ||
| @pytest.fixture(scope="package") | ||
| def event_loop_policy(): | ||
| return CustomEventLoopPolicy() | ||
| """ | ||
| ), | ||
| custom_policy=dedent( | ||
| """\ | ||
| import asyncio | ||
| class CustomEventLoopPolicy(asyncio.DefaultEventLoopPolicy): | ||
| pass | ||
| """ | ||
| ), | ||
| test_uses_custom_policy=dedent( | ||
| """\ | ||
| import asyncio | ||
| import pytest | ||
| from .custom_policy import CustomEventLoopPolicy | ||
| pytestmark = pytest.mark.asyncio(scope="package") | ||
| async def test_uses_custom_event_loop_policy(): | ||
| assert isinstance( | ||
| asyncio.get_event_loop_policy(), | ||
| CustomEventLoopPolicy, | ||
| ) | ||
| """ | ||
| ), | ||
| test_also_uses_custom_policy=dedent( | ||
| """\ | ||
| import asyncio | ||
| import pytest | ||
| from .custom_policy import CustomEventLoopPolicy | ||
| pytestmark = pytest.mark.asyncio(scope="package") | ||
| async def test_also_uses_custom_event_loop_policy(): | ||
| assert isinstance( | ||
| asyncio.get_event_loop_policy(), | ||
| CustomEventLoopPolicy, | ||
| ) | ||
| """ | ||
| ), | ||
| ) | ||
| result = pytester.runpytest("--asyncio-mode=strict") | ||
| result.assert_outcomes(passed=2) | ||
|
|
||
|
|
||
| def test_asyncio_mark_respects_parametrized_loop_policies( | ||
| pytester: Pytester, | ||
| ): | ||
| pytester.makepyfile( | ||
| dedent( | ||
| """\ | ||
| import asyncio | ||
| import pytest | ||
| pytestmark = pytest.mark.asyncio(scope="module") | ||
| @pytest.fixture( | ||
| scope="module", | ||
| params=[ | ||
| asyncio.DefaultEventLoopPolicy(), | ||
| asyncio.DefaultEventLoopPolicy(), | ||
| ], | ||
| ) | ||
| def event_loop_policy(request): | ||
| return request.param | ||
| async def test_parametrized_loop(): | ||
| pass | ||
| """ | ||
| ) | ||
| ) | ||
| result = pytester.runpytest_subprocess("--asyncio-mode=strict") | ||
| result.assert_outcomes(passed=2) | ||
|
|
||
|
|
||
| def test_asyncio_mark_provides_module_scoped_loop_to_fixtures( | ||
| pytester: Pytester, | ||
| ): | ||
| pytester.makepyfile( | ||
| dedent( | ||
| """\ | ||
| import asyncio | ||
| import pytest | ||
| import pytest_asyncio | ||
| pytestmark = pytest.mark.asyncio(scope="module") | ||
| loop: asyncio.AbstractEventLoop | ||
| @pytest_asyncio.fixture(scope="module") | ||
| async def my_fixture(): | ||
| global loop | ||
| loop = asyncio.get_running_loop() | ||
| async def test_runs_is_same_loop_as_fixture(my_fixture): | ||
| global loop | ||
| assert asyncio.get_running_loop() is loop | ||
| """ | ||
| ) | ||
| ) | ||
| result = pytester.runpytest_subprocess("--asyncio-mode=strict") | ||
| result.assert_outcomes(passed=1) |