Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -12,3 +12,5 @@ To run a test with only a subset of configured factories, use the ``loop_factori
@pytest.mark.asyncio(loop_factories=["custom"])
async def test_only_with_custom_event_loop():
pass

If a requested factory name is not available from the hook, the test variant for that factory is skipped.
2 changes: 1 addition & 1 deletion docs/reference/markers/index.rst
Original file line number Diff line number Diff line change
Expand Up @@ -43,7 +43,7 @@ Subpackages do not share the loop with their parent package.

Tests marked with *session* scope share the same event loop, even if the tests exist in different packages.

The ``pytest.mark.asyncio`` marker also accepts a ``loop_factories`` keyword argument to select a subset of configured event loop factories for a test. If ``loop_factories`` contains unknown names, pytest-asyncio raises a ``pytest.UsageError`` during collection.
The ``pytest.mark.asyncio`` marker also accepts a ``loop_factories`` keyword argument to select a subset of configured event loop factories for a test. If ``loop_factories`` contains names not available from the hook, those test variants are skipped.

.. |auto mode| replace:: *auto mode*
.. _auto mode: ../../concepts.html#auto-mode
Expand Down
41 changes: 25 additions & 16 deletions pytest_asyncio/plugin.py
Original file line number Diff line number Diff line change
Expand Up @@ -698,30 +698,39 @@ def pytest_generate_tests(metafunc: pytest.Metafunc) -> None:
)
return

factory_params: Iterable[object]
factory_ids: Iterable[str]
if marker_selected_factory_names is None:
effective_factories = hook_factories
factory_params = hook_factories.values()
factory_ids = hook_factories.keys()
else:
missing_factory_names = tuple(
name for name in marker_selected_factory_names if name not in hook_factories
)
if missing_factory_names:
msg = (
f"Unknown factory name(s) {missing_factory_names}."
f" Available names: {', '.join(hook_factories)}."
# Iterate in marker order to preserve explicit user selection
# order.
factory_ids = marker_selected_factory_names
factory_params = [
(
hook_factories[name]
if name in hook_factories
else pytest.param(
None,
marks=pytest.mark.skip(
reason=(
f"Loop factory {name!r} is not available."
f" Available factories:"
f" {', '.join(hook_factories)}."
),
),
)
)
raise pytest.UsageError(msg)
# Build the mapping in marker order to preserve explicit user
# selection order in parametrization.
effective_factories = {
name: hook_factories[name] for name in marker_selected_factory_names
}
for name in marker_selected_factory_names
]
metafunc.fixturenames.append(_asyncio_loop_factory.__name__)
default_loop_scope = _get_default_test_loop_scope(metafunc.config)
loop_scope = marker_loop_scope or default_loop_scope
metafunc.parametrize(
_asyncio_loop_factory.__name__,
effective_factories.values(),
ids=effective_factories.keys(),
factory_params,
ids=factory_ids,
indirect=True,
scope=loop_scope,
)
Expand Down
79 changes: 70 additions & 9 deletions tests/test_loop_factory_parametrization.py
Original file line number Diff line number Diff line change
Expand Up @@ -371,7 +371,7 @@ async def test_runs_only_with_uvloop():
result.assert_outcomes(passed=1)


def test_asyncio_marker_loop_factories_unknown_name_errors(pytester: Pytester) -> None:
def test_unavailable_factory_skips_with_reason(pytester: Pytester) -> None:
pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function")
pytester.makeconftest(dedent("""\
import asyncio
Expand All @@ -385,16 +385,77 @@ def pytest_asyncio_loop_factories(config, item):
pytest_plugins = "pytest_asyncio"

@pytest.mark.asyncio(loop_factories=["missing"])
async def test_errors():
async def test_skipped():
assert True
"""))
result = pytester.runpytest("--asyncio-mode=strict")
result.assert_outcomes(errors=1)
result.stdout.fnmatch_lines(
[
"*Unknown factory name(s)*Available names:*",
]
)
result = pytester.runpytest("--asyncio-mode=strict", "-rs")
result.assert_outcomes(skipped=1)
result.stdout.fnmatch_lines(["*SKIPPED*Loop factory 'missing' is not available*"])


def test_partial_intersection_runs_available_and_skips_missing(
pytester: Pytester,
) -> None:
pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function")
pytester.makeconftest(dedent("""\
import asyncio

class CustomEventLoop(asyncio.SelectorEventLoop):
pass

def pytest_asyncio_loop_factories(config, item):
return {
"available": CustomEventLoop,
"other": asyncio.new_event_loop,
}
"""))
pytester.makepyfile(dedent("""\
import asyncio
import pytest

pytest_plugins = "pytest_asyncio"

@pytest.mark.asyncio(loop_factories=["available", "missing"])
async def test_runs_with_available():
assert type(asyncio.get_running_loop()).__name__ == "CustomEventLoop"
"""))
result = pytester.runpytest("--asyncio-mode=strict", "-rs")
result.assert_outcomes(passed=1, skipped=1)
result.stdout.fnmatch_lines(["*SKIPPED*Loop factory 'missing' is not available*"])


def test_platform_conditional_factories(pytester: Pytester) -> None:
pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function")
pytester.makeconftest(dedent("""\
import asyncio
import sys

def pytest_asyncio_loop_factories(config, item):
factories = {"default": asyncio.new_event_loop}
if sys.platform == "a_platform_that_does_not_exist":
factories["exotic"] = asyncio.new_event_loop
return factories
"""))
pytester.makepyfile(dedent("""\
import pytest

pytest_plugins = "pytest_asyncio"

@pytest.mark.asyncio(loop_factories=["exotic"])
async def test_exotic_only():
assert True

@pytest.mark.asyncio(loop_factories=["default"])
async def test_default_only():
assert True

@pytest.mark.asyncio(loop_factories=["default", "exotic"])
async def test_both():
assert True
"""))
result = pytester.runpytest("--asyncio-mode=strict", "-rs")
result.assert_outcomes(passed=2, skipped=2)
result.stdout.fnmatch_lines(["*SKIPPED*Loop factory 'exotic' is not available*"])


def test_asyncio_marker_loop_factories_without_hook_errors(
Expand Down
Loading