diff --git a/changelog.d/+event_loop_policy_removed.removed.rst b/changelog.d/+event_loop_policy_removed.removed.rst new file mode 100644 index 00000000..dd03c8e4 --- /dev/null +++ b/changelog.d/+event_loop_policy_removed.removed.rst @@ -0,0 +1 @@ +Removed the *event_loop_policy* fixture. Defining a fixture with this name now raises a usage error at collection time instead of silently having no effect. Use the ``pytest_asyncio_loop_factories`` hook to customize event loop creation instead. See the :doc:`migration guide `. diff --git a/changelog.d/+hook_based_rewrite.changed.rst b/changelog.d/+hook_based_rewrite.changed.rst new file mode 100644 index 00000000..9ffa238f --- /dev/null +++ b/changelog.d/+hook_based_rewrite.changed.rst @@ -0,0 +1 @@ +Reimplemented test collection and dispatch using plain pytest hooks (``pytest_pycollect_makeitem``, ``pytest_generate_tests``, ``pytest_runtest_setup``, ``pytest_pyfunc_call``) instead of a custom ``pytest.Item`` subclass hierarchy. ``pytest_asyncio.is_async_test`` now returns a plain ``bool`` instead of a ``TypeIs``-narrowed type, since there is no longer a subclass to narrow to. Event loop lifecycle management (one ``asyncio.Runner`` per active loop scope) is unchanged internally. diff --git a/changelog.d/+scope_kwarg_removed.removed.rst b/changelog.d/+scope_kwarg_removed.removed.rst new file mode 100644 index 00000000..7f2e3ac8 --- /dev/null +++ b/changelog.d/+scope_kwarg_removed.removed.rst @@ -0,0 +1 @@ +Removed the deprecated *scope* keyword argument to the ``asyncio`` marker. Use ``pytest.mark.asyncio(loop_scope="…")`` instead. diff --git a/changelog.d/+version_floor_unchanged.downstream.rst b/changelog.d/+version_floor_unchanged.downstream.rst new file mode 100644 index 00000000..edd527ab --- /dev/null +++ b/changelog.d/+version_floor_unchanged.downstream.rst @@ -0,0 +1 @@ +The minimum supported Python (3.10) and pytest (8.4) versions are unchanged despite this being a major version bump; nothing in the v2 rewrite requires newer versions of either. diff --git a/changelog.d/1514.added.rst b/changelog.d/1514.added.rst new file mode 100644 index 00000000..7eb037cd --- /dev/null +++ b/changelog.d/1514.added.rst @@ -0,0 +1 @@ +Added ``PytestAsyncioLoopScopeMismatchWarning``, raised when a test or fixture (transitively) depends on an async fixture whose effective *loop_scope* differs from its own. Running a test and a fixture it depends on under different event loops can silently break objects (such as ``asyncio.Future``, ``asyncio.Task``, or ``asyncio.Lock``) bound to the loop they were created on. See :doc:`/reference/warnings`. diff --git a/docs/how-to-guides/index.rst b/docs/how-to-guides/index.rst index b67a4125..27562aae 100644 --- a/docs/how-to-guides/index.rst +++ b/docs/how-to-guides/index.rst @@ -7,6 +7,7 @@ How-To Guides migrate_from_0_21 migrate_from_0_23 + migrate_from_1_x change_fixture_loop change_default_fixture_loop change_default_test_loop @@ -16,7 +17,6 @@ How-To Guides run_class_tests_in_same_loop run_module_tests_in_same_loop run_package_tests_in_same_loop - multiple_loops parametrize_with_asyncio uvloop test_item_is_async diff --git a/docs/how-to-guides/migrate_from_1_x.rst b/docs/how-to-guides/migrate_from_1_x.rst new file mode 100644 index 00000000..15894912 --- /dev/null +++ b/docs/how-to-guides/migrate_from_1_x.rst @@ -0,0 +1,9 @@ +.. _how_to_guides/migrate_from_1_x: + +====================================== +How to migrate from pytest-asyncio v1 +====================================== +1. If your test suite (or a plugin/conftest it uses) defines an *event_loop_policy* fixture, remove it. The fixture no longer has any effect; defining one now raises a usage error at collection time. Move the customization into a ``pytest_asyncio_loop_factories`` hook implementation instead. See :doc:`custom_loop_factory` and :doc:`uvloop`. +2. Change all occurrences of ``pytest.mark.asyncio(scope="…")`` to ``pytest.mark.asyncio(loop_scope="…")``. The deprecated *scope* keyword argument to the *asyncio* marker is no longer accepted at all. +3. Expect a new ``PytestAsyncioLoopScopeMismatchWarning`` to appear if a test or fixture (transitively) depends on an async fixture with a different effective *loop_scope*. This is new in v2 and can surface pre-existing bugs in test suites that already use *loop_scope*, sometimes in bulk, on first upgrade: a test and a fixture running on different event loops can silently break objects (such as ``asyncio.Future``, ``asyncio.Task``, or ``asyncio.Lock``) bound to the loop they were created on. See :doc:`../reference/warnings` for how to fix or, if the mismatch is intentional, silence individual instances. +4. If your code relies on the return type of ``pytest_asyncio.is_async_test``, note that it now returns a plain ``bool`` rather than a ``TypeIs`` type guard, since the check is no longer an ``isinstance`` check against a pytest-asyncio-specific ``Item`` subclass. diff --git a/docs/how-to-guides/multiple_loops.rst b/docs/how-to-guides/multiple_loops.rst deleted file mode 100644 index 7ebf0616..00000000 --- a/docs/how-to-guides/multiple_loops.rst +++ /dev/null @@ -1,14 +0,0 @@ -====================================== -How to test with different event loops -====================================== - -.. warning:: - - Overriding the *event_loop_policy* fixture is deprecated and will be removed in a future version of pytest-asyncio. Use the ``pytest_asyncio_loop_factories`` hook instead. See :doc:`custom_loop_factory` for details. - -Parametrizing the *event_loop_policy* fixture parametrizes all async tests. The following example causes all async tests to run multiple times, once for each event loop in the fixture parameters: - -.. include:: multiple_loops_example.py - :code: python - -You may choose to limit the scope of the fixture to *package,* *module,* or *class,* if you only want a subset of your tests to run with different event loops. diff --git a/docs/how-to-guides/multiple_loops_example.py b/docs/how-to-guides/multiple_loops_example.py deleted file mode 100644 index 2083e8b6..00000000 --- a/docs/how-to-guides/multiple_loops_example.py +++ /dev/null @@ -1,29 +0,0 @@ -import asyncio -import warnings - -with warnings.catch_warnings(): - warnings.simplefilter("ignore", DeprecationWarning) - from asyncio import DefaultEventLoopPolicy - -import pytest - - -class CustomEventLoopPolicy(DefaultEventLoopPolicy): - pass - - -@pytest.fixture( - scope="session", - params=( - CustomEventLoopPolicy(), - CustomEventLoopPolicy(), - ), -) -def event_loop_policy(request): - return request.param - - -@pytest.mark.asyncio -@pytest.mark.filterwarnings("ignore::DeprecationWarning") -async def test_uses_custom_event_loop_policy(): - assert isinstance(asyncio.get_event_loop_policy(), CustomEventLoopPolicy) diff --git a/docs/how-to-guides/uvloop.rst b/docs/how-to-guides/uvloop.rst index 03143be5..8239d94d 100644 --- a/docs/how-to-guides/uvloop.rst +++ b/docs/how-to-guides/uvloop.rst @@ -18,24 +18,3 @@ Define a ``pytest_asyncio_loop_factories`` hook in your *conftest.py* that maps :doc:`custom_loop_factory` More details on the ``pytest_asyncio_loop_factories`` hook, including per-test factory selection and multiple factory parametrization. - -Using the event_loop_policy fixture ------------------------------------ - -.. note:: - - ``asyncio.AbstractEventLoopPolicy`` is deprecated as of Python 3.14 (removal planned for 3.16), and ``uvloop.EventLoopPolicy`` will be removed alongside it. Overriding the *event_loop_policy* fixture is also deprecated in pytest-asyncio. Prefer the hook approach above. - -For older versions of Python and uvloop, you can override the *event_loop_policy* fixture in your *conftest.py:* - -.. code-block:: python - - import pytest - import uvloop - - - @pytest.fixture(scope="session") - def event_loop_policy(): - return uvloop.EventLoopPolicy() - -You may choose to limit the scope of the fixture to *package,* *module,* or *class,* if you only want a subset of your tests to run with uvloop. diff --git a/docs/reference/fixtures/event_loop_policy_example.py b/docs/reference/fixtures/event_loop_policy_example.py deleted file mode 100644 index e8642527..00000000 --- a/docs/reference/fixtures/event_loop_policy_example.py +++ /dev/null @@ -1,23 +0,0 @@ -import asyncio -import warnings - -with warnings.catch_warnings(): - warnings.simplefilter("ignore", DeprecationWarning) - from asyncio import DefaultEventLoopPolicy - -import pytest - - -class CustomEventLoopPolicy(DefaultEventLoopPolicy): - pass - - -@pytest.fixture(scope="module") -def event_loop_policy(request): - return CustomEventLoopPolicy() - - -@pytest.mark.asyncio(loop_scope="module") -@pytest.mark.filterwarnings("ignore::DeprecationWarning") -async def test_uses_custom_event_loop_policy(): - assert isinstance(asyncio.get_event_loop_policy(), CustomEventLoopPolicy) diff --git a/docs/reference/fixtures/event_loop_policy_parametrized_example.py b/docs/reference/fixtures/event_loop_policy_parametrized_example.py deleted file mode 100644 index 19552d81..00000000 --- a/docs/reference/fixtures/event_loop_policy_parametrized_example.py +++ /dev/null @@ -1,28 +0,0 @@ -import asyncio -import warnings - -with warnings.catch_warnings(): - warnings.simplefilter("ignore", DeprecationWarning) - from asyncio import DefaultEventLoopPolicy - -import pytest - - -class CustomEventLoopPolicy(DefaultEventLoopPolicy): - pass - - -@pytest.fixture( - params=( - DefaultEventLoopPolicy(), - CustomEventLoopPolicy(), - ), -) -def event_loop_policy(request): - return request.param - - -@pytest.mark.asyncio -@pytest.mark.filterwarnings("ignore::DeprecationWarning") -async def test_uses_custom_event_loop_policy(): - assert isinstance(asyncio.get_event_loop_policy(), DefaultEventLoopPolicy) diff --git a/docs/reference/fixtures/index.rst b/docs/reference/fixtures/index.rst index 2791a484..ade01d18 100644 --- a/docs/reference/fixtures/index.rst +++ b/docs/reference/fixtures/index.rst @@ -2,29 +2,6 @@ Fixtures ======== -event_loop_policy -================= - -.. warning:: - - Overriding the *event_loop_policy* fixture is deprecated and will be removed in a future version of pytest-asyncio. Use the ``pytest_asyncio_loop_factories`` hook instead. See :doc:`../hooks` for details. - -Returns the event loop policy used to create asyncio event loops. -The default return value is *asyncio.get_event_loop_policy().* - -This fixture can be overridden when a different event loop policy should be used. - -.. include:: event_loop_policy_example.py - :code: python - -Multiple policies can be provided via fixture parameters. -The fixture is automatically applied to all pytest-asyncio tests. -Therefore, all tests managed by pytest-asyncio are run once for each fixture parameter. -The following example runs the test with different event loop policies. - -.. include:: event_loop_policy_parametrized_example.py - :code: python - unused_tcp_port =============== Finds and yields a single unused TCP port on the localhost interface. Useful for diff --git a/docs/reference/index.rst b/docs/reference/index.rst index 5c3095f7..81af9865 100644 --- a/docs/reference/index.rst +++ b/docs/reference/index.rst @@ -11,6 +11,7 @@ Reference hooks markers/index decorators/index + warnings changelog This section of the documentation provides descriptions of the individual parts provided by pytest-asyncio. diff --git a/docs/reference/warnings.rst b/docs/reference/warnings.rst new file mode 100644 index 00000000..983c0bb0 --- /dev/null +++ b/docs/reference/warnings.rst @@ -0,0 +1,36 @@ +======== +Warnings +======== + +PytestAsyncioLoopScopeMismatchWarning +====================================== +Warns that a test or fixture requests an async fixture whose effective *loop_scope* differs from its own. + +Every async test and every async fixture runs on an event loop determined by its effective *loop_scope* (its own explicit *loop_scope* argument, else the configured default, else its pytest caching scope). When a test or fixture (transitively) depends on an async fixture with a *different* effective *loop_scope*, the two run on different event loops. This can silently break objects -- such as ``asyncio.Future``, ``asyncio.Task``, or ``asyncio.Lock`` -- that are bound to the loop they were created on. + +.. code-block:: python + + import pytest + + import pytest_asyncio + + + @pytest_asyncio.fixture(loop_scope="module") + async def async_fixture(): ... + + + @pytest.mark.asyncio(loop_scope="function") + async def test_uses_fixture_from_a_different_loop(async_fixture): + # async_fixture ran on the module-scoped loop, but this test + # runs on its own function-scoped loop: PytestAsyncioLoopScopeMismatchWarning + ... + +If the mismatch is intentional, silence the warning for a specific test with the standard *filterwarnings* marker: + +.. code-block:: python + + @pytest.mark.asyncio(loop_scope="function") + @pytest.mark.filterwarnings( + "ignore::pytest_asyncio.PytestAsyncioLoopScopeMismatchWarning" + ) + async def test_uses_fixture_from_a_different_loop(async_fixture): ... diff --git a/pytest_asyncio/__init__.py b/pytest_asyncio/__init__.py index abd62e15..e0d84c65 100644 --- a/pytest_asyncio/__init__.py +++ b/pytest_asyncio/__init__.py @@ -4,8 +4,8 @@ from importlib.metadata import version -from .plugin import fixture, is_async_test +from .plugin import PytestAsyncioLoopScopeMismatchWarning, fixture, is_async_test __version__ = version(__name__) -__all__ = ("fixture", "is_async_test") +__all__ = ("PytestAsyncioLoopScopeMismatchWarning", "fixture", "is_async_test") diff --git a/pytest_asyncio/_collection.py b/pytest_asyncio/_collection.py new file mode 100644 index 00000000..ad9508ef --- /dev/null +++ b/pytest_asyncio/_collection.py @@ -0,0 +1,238 @@ +"""Collection of asyncio tests: tag collected async functions as pytest-asyncio's.""" + +from __future__ import annotations + +import inspect +from collections.abc import Collection, Generator, Sequence +from typing import Literal + +import pluggy +import pytest +from pytest import Function, Item, PytestCollectionWarning + +from ._config import _get_default_test_loop_scope +from ._hooks import _is_coroutine_or_asyncgen, _ScopeName +from ._markers import ( + _collect_hook_loop_factories, + _parse_asyncio_marker, + _resolve_asyncio_marker, +) +from ._mismatch import _compute_loop_scope_mismatches +from ._runner import _asyncio_loop_factory + +_AsyncFunctionCategory = Literal["coroutine", "asyncgen", "staticmethod", "hypothesis"] +_SynchronizationTargetAttr = Literal["obj", "inner_test"] + +asyncio_test_key: pytest.StashKey[bool] = pytest.StashKey() +loop_scope_key: pytest.StashKey[_ScopeName] = pytest.StashKey() +loop_scope_mismatches_key: pytest.StashKey[list[tuple[str, _ScopeName]]] = ( + pytest.StashKey() +) +_hypothesis_version_incompatible_key: pytest.StashKey[bool] = pytest.StashKey() + + +def _is_hypothesis_wrapped_coroutine(func: object) -> bool: + hypothesis_handle = getattr(func, "hypothesis", None) + return bool( + getattr(func, "is_hypothesis_test", False) + and hypothesis_handle is not None + and inspect.iscoroutinefunction(hypothesis_handle.inner_test) + ) + + +def _classify_async_function(func: object) -> _AsyncFunctionCategory | None: + """ + Classify a collected test function, mirroring the precedence of + pytest-asyncio's historical Item-subclass hierarchy exactly (first + matching category wins, in this order). In particular, a + staticmethod-wrapped async generator is classified "staticmethod", not + "asyncgen" -- the unsupported-async-generator rejection below only ever + applied to plain (non-staticmethod) async generators, and that quirk is + preserved here rather than "fixed". + + Returns None if `func` isn't async-shaped at all, in which case + pytest-asyncio does not take ownership of it. + """ + if inspect.iscoroutinefunction(func): + return "coroutine" + if inspect.isasyncgenfunction(func): + return "asyncgen" + if isinstance(func, staticmethod) and _is_coroutine_or_asyncgen(func.__func__): + return "staticmethod" + if _is_hypothesis_wrapped_coroutine(func): + return "hypothesis" + return None + + +def _synchronization_target( + item: Function, +) -> tuple[object, _SynchronizationTargetAttr]: + """ + Return the (obj, attr) pair that needs to be monkeypatched with a + synchronized wrapper while the test runs. Normally this is the item + itself, but a Hypothesis-wrapped (`@given`) coroutine lives at + `item.obj.hypothesis.inner_test`, not `item.obj`. + """ + if _classify_async_function(item.obj) == "hypothesis": + return item.obj.hypothesis, "inner_test" + return item, "obj" + + +def _compute_test_loop_scope(item: Function) -> _ScopeName: + marker = item.get_closest_marker("asyncio") + assert marker is not None + loop_scope = marker.kwargs.get("loop_scope") + if loop_scope is not None: + return loop_scope + return _get_default_test_loop_scope(item.config) + + +# The function name needs to start with "pytest_" +# see https://github.com/pytest-dev/pytest/issues/11307 +@pytest.hookimpl(specname="pytest_pycollect_makeitem", hookwrapper=True) +def pytest_pycollect_makeitem_tag_async_items( + collector: pytest.Module | pytest.Class, name: str, obj: object +) -> Generator[None, pluggy.Result, None]: + """ + Tags coroutines, async generators, and their staticmethod/Hypothesis-wrapped + variants -- collected as plain pytest.Function items -- as pytest-asyncio + tests, and computes/stashes the loop scope each one runs under. + """ + hook_result = yield + try: + node_or_list_of_nodes: ( + pytest.Item | pytest.Collector | list[pytest.Item | pytest.Collector] | None + ) = hook_result.get_result() + except BaseException as e: + hook_result.force_exception(e) + return + if not node_or_list_of_nodes: + return + if isinstance(node_or_list_of_nodes, Sequence): + node_iterator = iter(node_or_list_of_nodes) + else: + # Treat single node as a single-element iterable + node_iterator = iter((node_or_list_of_nodes,)) + for node in node_iterator: + if not isinstance(node, Function): + continue + category = _classify_async_function(node.obj) + if category is None: + continue + if _resolve_asyncio_marker(node) is None: + continue + node.stash[asyncio_test_key] = True + if category == "asyncgen": + unsupported_item_type_message = ( + f"Tests based on asynchronous generators are not supported. " + f"{node.name} will be ignored." + ) + node.warn(PytestCollectionWarning(unsupported_item_type_message)) + node.add_marker( + pytest.mark.xfail(run=False, reason=unsupported_item_type_message) + ) + if ( + category == "hypothesis" + and not getattr(node.obj, "hypothesis", False) + and getattr(node.obj, "is_hypothesis_test", False) + ): + node.stash[_hypothesis_version_incompatible_key] = True + loop_scope = _compute_test_loop_scope(node) + node.stash[loop_scope_key] = loop_scope + node.stash[loop_scope_mismatches_key] = _compute_loop_scope_mismatches( + node, loop_scope, node.config + ) + # Insert (rather than append) so that _fillfixtures(), which resolves + # item.fixturenames strictly in order, resolves the loop factory + # before any other same-scope async fixture. A loop-factory variant + # change must tear down the stale runner (and, via the finalizer + # chaining in pytest_fixture_setup, any other async fixture cached + # under it) before those fixtures are resolved for this item -- and + # that can only happen if something actually requests the (side + # effect free) loop factory value early. The runner fixture id itself + # is *appended*, deliberately staying last: unlike the loop factory, + # creating/entering the runner has real side effects (it becomes the + # current asyncio event loop), so it must not be resolved ahead of + # ordinary same-scope fixtures that assume no loop is current yet. + if _asyncio_loop_factory.__name__ not in node.fixturenames: + node.fixturenames.insert(0, _asyncio_loop_factory.__name__) + runner_fixture_id = f"_{loop_scope}_scoped_runner" + if runner_fixture_id not in node.fixturenames: + node.fixturenames.append(runner_fixture_id) + + +@pytest.hookimpl(tryfirst=True) +def pytest_generate_tests(metafunc: pytest.Metafunc) -> None: + if _classify_async_function(metafunc.definition.obj) is None: + return + + asyncio_marker = _resolve_asyncio_marker(metafunc.definition) + if asyncio_marker is None: + return + marker_loop_scope, marker_selected_factory_names = _parse_asyncio_marker( + asyncio_marker + ) + + hook_factories = _collect_hook_loop_factories(metafunc.config, metafunc.definition) + if hook_factories is None: + if marker_selected_factory_names is not None: + raise pytest.UsageError( + "mark.asyncio 'loop_factories' requires at least one " + "pytest_asyncio_loop_factories hook implementation." + ) + return + + factory_params: Collection[object] + factory_ids: Collection[str] + if marker_selected_factory_names is None: + factory_params = hook_factories.values() + factory_ids = hook_factories.keys() + else: + # 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)}." + ), + ), + ) + ) + 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 + # pytest.HIDDEN_PARAM was added in pytest 8.4 + hide_id = len(factory_ids) == 1 and hasattr(pytest, "HIDDEN_PARAM") + metafunc.parametrize( + _asyncio_loop_factory.__name__, + factory_params, + ids=(pytest.HIDDEN_PARAM,) if hide_id else factory_ids, + indirect=True, + scope=loop_scope, + ) + + +@pytest.hookimpl +def pytest_runtest_setup(item: Item) -> None: + if not is_async_test(item): + return + if item.stash.get(_hypothesis_version_incompatible_key, False): + pytest.fail( + f"test function `{item!r}` is using Hypothesis, but pytest-asyncio " + "only works with Hypothesis 3.64.0 or later." + ) + + +def is_async_test(item: Item) -> bool: + """Returns whether a test item is a pytest-asyncio test""" + return item.stash.get(asyncio_test_key, False) diff --git a/pytest_asyncio/_config.py b/pytest_asyncio/_config.py new file mode 100644 index 00000000..fb16f6c0 --- /dev/null +++ b/pytest_asyncio/_config.py @@ -0,0 +1,151 @@ +"""CLI/ini configuration: asyncio_mode, asyncio_debug, and loop-scope defaults.""" + +from __future__ import annotations + +import enum +import warnings +from typing import Any + +import pytest +from _pytest.scope import Scope +from pytest import Config, Parser, PytestDeprecationWarning, PytestPluginManager + +from ._hooks import PytestAsyncioSpecs + + +class Mode(str, enum.Enum): + AUTO = "auto" + STRICT = "strict" + + +ASYNCIO_MODE_HELP = """\ +'auto' - for automatically handling all async functions by the plugin +'strict' - for autoprocessing disabling (useful if different async frameworks \ +should be tested together, e.g. \ +both pytest-asyncio and pytest-trio are used in the same project) +""" + + +def pytest_addoption(parser: Parser, pluginmanager: PytestPluginManager) -> None: + pluginmanager.add_hookspecs(PytestAsyncioSpecs) + group = parser.getgroup("asyncio") + group.addoption( + "--asyncio-mode", + dest="asyncio_mode", + default=None, + metavar="MODE", + help=ASYNCIO_MODE_HELP, + ) + group.addoption( + "--asyncio-debug", + dest="asyncio_debug", + action="store_true", + default=None, + help="enable asyncio debug mode for the default event loop", + ) + parser.addini( + "asyncio_mode", + help="default value for --asyncio-mode", + default="strict", + ) + parser.addini( + "asyncio_debug", + help="enable asyncio debug mode for the default event loop", + type="bool", + default="false", + ) + parser.addini( + "asyncio_default_fixture_loop_scope", + type="string", + help="default scope of the asyncio event loop used to execute async fixtures", + default=None, + ) + parser.addini( + "asyncio_default_test_loop_scope", + type="string", + help="default scope of the asyncio event loop used to execute tests", + default="function", + ) + + +def _get_asyncio_mode(config: Config) -> Mode: + val = config.getoption("asyncio_mode") + if val is None: + val = config.getini("asyncio_mode") + try: + return Mode(val) + except ValueError as e: + modes = ", ".join(m.value for m in Mode) + raise pytest.UsageError( + f"{val!r} is not a valid asyncio_mode. Valid modes: {modes}." + ) from e + + +def _get_asyncio_debug(config: Config) -> bool: + val = config.getoption("asyncio_debug") + if val is None: + val = config.getini("asyncio_debug") + + if isinstance(val, bool): + return val + else: + return val == "true" + + +_DEFAULT_FIXTURE_LOOP_SCOPE_UNSET = """\ +The configuration option "asyncio_default_fixture_loop_scope" is unset. +The event loop scope for asynchronous fixtures will default to the "fixture" caching \ +scope. Future versions of pytest-asyncio will default the loop scope for asynchronous \ +fixtures to "function" scope. Set the default fixture loop scope explicitly in order \ +to avoid unexpected behavior in the future. Valid fixture loop scopes are: \ +"function", "class", "module", "package", "session" +""" + + +def _validate_scope(scope: str | None, option_name: str) -> None: + if scope is None: + return + valid_scopes = [s.value for s in Scope] + if scope not in valid_scopes: + raise pytest.UsageError( + f"{scope!r} is not a valid {option_name}. " + f"Valid scopes are: {', '.join(valid_scopes)}." + ) + + +def pytest_configure(config: Config) -> None: + default_fixture_loop_scope = config.getini("asyncio_default_fixture_loop_scope") + _validate_scope(default_fixture_loop_scope, "asyncio_default_fixture_loop_scope") + if not default_fixture_loop_scope: + warnings.warn(PytestDeprecationWarning(_DEFAULT_FIXTURE_LOOP_SCOPE_UNSET)) + + default_test_loop_scope = config.getini("asyncio_default_test_loop_scope") + _validate_scope(default_test_loop_scope, "asyncio_default_test_loop_scope") + config.addinivalue_line( + "markers", + "asyncio: " + "mark the test as a coroutine, it will be " + "run using an asyncio event loop", + ) + + +@pytest.hookimpl(tryfirst=True) +def pytest_report_header(config: Config) -> list[str]: + """Add asyncio config to pytest header.""" + mode = _get_asyncio_mode(config) + debug = _get_asyncio_debug(config) + default_fixture_loop_scope = config.getini("asyncio_default_fixture_loop_scope") + default_test_loop_scope = _get_default_test_loop_scope(config) + header = [ + f"mode={mode}", + f"debug={debug}", + f"asyncio_default_fixture_loop_scope={default_fixture_loop_scope}", + f"asyncio_default_test_loop_scope={default_test_loop_scope}", + ] + return [ + "asyncio: " + ", ".join(header), + ] + + +def _get_default_test_loop_scope(config: Config) -> Any: + return config.getini("asyncio_default_test_loop_scope") diff --git a/pytest_asyncio/_dispatch.py b/pytest_asyncio/_dispatch.py new file mode 100644 index 00000000..05fd7c33 --- /dev/null +++ b/pytest_asyncio/_dispatch.py @@ -0,0 +1,112 @@ +"""Dispatch of asyncio tests: synchronized execution plus pre-flight warnings.""" + +from __future__ import annotations + +import contextvars +import functools +import warnings +from collections.abc import Callable +from types import CoroutineType + +import pytest +from pytest import Function, MonkeyPatch, PytestDeprecationWarning + +from ._collection import ( + _is_coroutine_or_asyncgen, + _synchronization_target, + is_async_test, + loop_scope_key, + loop_scope_mismatches_key, +) +from ._config import Mode, _get_asyncio_mode +from ._fixtures import _is_asyncio_fixture_function +from ._mismatch import PytestAsyncioLoopScopeMismatchWarning + + +def _synchronize_coroutine( + func: Callable[..., CoroutineType], + runner, + context: contextvars.Context, +): + """ + Return a sync wrapper around a coroutine executing it in the + specified runner and context. + """ + + @functools.wraps(func) + def inner(*args, **kwargs): + coro = func(*args, **kwargs) + runner.run(coro, context=context) + + return inner + + +@pytest.hookimpl(tryfirst=True, hookwrapper=True) +def pytest_pyfunc_call(pyfuncitem: Function) -> object | None: + """Pytest hook called before a test case is run.""" + if pyfuncitem.get_closest_marker("asyncio") is not None: + if is_async_test(pyfuncitem): + asyncio_mode = _get_asyncio_mode(pyfuncitem.config) + for fixname, fixtures in pyfuncitem._fixtureinfo.name2fixturedefs.items(): + # name2fixturedefs is a dict between fixture name and a list of matching + # fixturedefs. The last entry in the list is closest and the one used. + func = fixtures[-1].func + if ( + asyncio_mode == Mode.STRICT + and _is_coroutine_or_asyncgen(func) + and not _is_asyncio_fixture_function(func) + ): + warnings.warn( + PytestDeprecationWarning( + f"asyncio test {pyfuncitem.name!r} requested async " + "@pytest.fixture " + f"{fixname!r} in strict mode. " + "You might want to use @pytest_asyncio.fixture or switch " + "to auto mode. " + "This will become an error in future versions of " + "pytest-asyncio." + ), + stacklevel=1, + ) + # no stacklevel points at the users code, so we set stacklevel=1 + # so it at least indicates that it's the plugin complaining. + # Pytest gives the test file & name in the warnings summary at least + + loop_scope = pyfuncitem.stash[loop_scope_key] + for fixture_name, fixture_loop_scope in pyfuncitem.stash.get( + loop_scope_mismatches_key, () + ): + warnings.warn( + PytestAsyncioLoopScopeMismatchWarning( + f"{pyfuncitem.nodeid}: the test's effective loop_scope " + f"{loop_scope!r} differs from fixture {fixture_name!r}'s " + f"effective loop_scope {fixture_loop_scope!r}. The fixture " + "will run on a different event loop than the test, which " + "can silently break objects (e.g. asyncio.Future, Task, or " + "Lock) bound to the loop they were created on." + ), + stacklevel=1, + ) + runner_fixture_id = f"_{loop_scope}_scoped_runner" + runner = pyfuncitem._request.getfixturevalue(runner_fixture_id) # type: ignore[attr-defined] + context = contextvars.copy_context() + target_obj, target_attr = _synchronization_target(pyfuncitem) + synchronized_obj = _synchronize_coroutine( + getattr(target_obj, target_attr), runner, context + ) + with MonkeyPatch.context() as c: + c.setattr(target_obj, target_attr, synchronized_obj) + yield + return None + else: + pyfuncitem.warn( + pytest.PytestWarning( + f"The test {pyfuncitem} is marked with '@pytest.mark.asyncio' " + "but it is not an async function. " + "Please remove the asyncio mark. " + "If the test is not marked explicitly, " + "check for global marks applied via 'pytestmark'." + ) + ) + yield + return None diff --git a/pytest_asyncio/_fixtures.py b/pytest_asyncio/_fixtures.py new file mode 100644 index 00000000..a3c8cdab --- /dev/null +++ b/pytest_asyncio/_fixtures.py @@ -0,0 +1,322 @@ +"""The @pytest_asyncio.fixture decorator and the machinery that runs async fixtures.""" + +from __future__ import annotations + +import contextvars +import functools +import inspect +from collections.abc import ( + AsyncIterable, + AsyncIterator, + Awaitable, + Callable, + Generator, + Iterable, +) +from types import AsyncGeneratorType, CoroutineType +from typing import Any, ParamSpec, TypeVar, cast, overload + +import pytest +from _pytest.fixtures import resolve_fixture_function +from pytest import Config, FixtureDef, FixtureRequest, MonkeyPatch + +from ._config import Mode, _get_asyncio_mode +from ._hooks import _is_coroutine_or_asyncgen, _ScopeName +from ._runner import Runner, _temporary_event_loop + +_R = TypeVar("_R", bound=Awaitable | AsyncIterable | AsyncIterator) +_P = ParamSpec("_P") +FixtureFunction = Callable[_P, _R] + + +@overload +def fixture( + fixture_function: FixtureFunction[_P, _R], + *, + scope: _ScopeName | Callable[[str, Config], _ScopeName] = ..., + loop_scope: _ScopeName | None = ..., + params: Iterable[object] | None = ..., + autouse: bool = ..., + ids: ( + Iterable[str | float | int | bool | None] + | Callable[[Any], object | None] + | None + ) = ..., + name: str | None = ..., +) -> FixtureFunction[_P, _R]: ... + + +@overload +def fixture( + fixture_function: None = ..., + *, + scope: _ScopeName | Callable[[str, Config], _ScopeName] = ..., + loop_scope: _ScopeName | None = ..., + params: Iterable[object] | None = ..., + autouse: bool = ..., + ids: ( + Iterable[str | float | int | bool | None] + | Callable[[Any], object | None] + | None + ) = ..., + name: str | None = None, +) -> Callable[[FixtureFunction[_P, _R]], FixtureFunction[_P, _R]]: ... + + +def fixture( + fixture_function: FixtureFunction[_P, _R] | None = None, + loop_scope: _ScopeName | None = None, + **kwargs: Any, +) -> ( + FixtureFunction[_P, _R] + | Callable[[FixtureFunction[_P, _R]], FixtureFunction[_P, _R]] +): + if fixture_function is not None: + _make_asyncio_fixture_function(fixture_function, loop_scope) + return pytest.fixture(fixture_function, **kwargs) + + else: + + @functools.wraps(fixture) + def inner(fixture_function: FixtureFunction[_P, _R]) -> FixtureFunction[_P, _R]: + return fixture(fixture_function, loop_scope=loop_scope, **kwargs) + + return inner + + +def _is_asyncio_fixture_function(obj: Any) -> bool: + obj = getattr(obj, "__func__", obj) # instance method maybe? + return getattr(obj, "_force_asyncio_fixture", False) + + +def _make_asyncio_fixture_function(obj: Any, loop_scope: _ScopeName | None) -> None: + if hasattr(obj, "__func__"): + # instance method, check the function object + obj = obj.__func__ + obj._force_asyncio_fixture = True + obj._loop_scope = loop_scope + + +def _owns_fixture(func: Any, mode: Mode) -> bool: + """ + Returns whether pytest-asyncio takes ownership of (wraps/synchronizes) + the given fixture function.""" + if _is_asyncio_fixture_function(func): + return True + if mode is not Mode.AUTO: + return False + return _is_coroutine_or_asyncgen(func) + + +def _effective_fixture_loop_scope( + func: Any, cache_scope: str, config: Config +) -> _ScopeName: + """ + The loop scope an async fixture actually runs its body under: its own + explicit loop_scope kwarg, else the configured fixture-loop-scope + default, else its pytest cache scope.""" + default_loop_scope = config.getini("asyncio_default_fixture_loop_scope") + return cast( + "_ScopeName", + getattr(func, "_loop_scope", None) or default_loop_scope or cache_scope, + ) + + +def _fixture_synchronizer( + fixturedef: FixtureDef, runner: Runner, request: FixtureRequest +) -> Callable: + """Returns a synchronous function evaluating the specified fixture.""" + fixture_function = resolve_fixture_function(fixturedef, request) + if inspect.isasyncgenfunction(fixturedef.func): + return _wrap_asyncgen_fixture(fixture_function, runner, request) # type: ignore[arg-type] + elif inspect.iscoroutinefunction(fixturedef.func): + return _wrap_async_fixture(fixture_function, runner, request) # type: ignore[arg-type] + elif inspect.isgeneratorfunction(fixturedef.func): + return _wrap_syncgen_fixture(fixture_function, runner) # type: ignore[arg-type] + else: + return _wrap_sync_fixture(fixture_function, runner) # type: ignore[arg-type] + + +SyncGenFixtureParams = ParamSpec("SyncGenFixtureParams") +SyncGenFixtureYieldType = TypeVar("SyncGenFixtureYieldType") + + +def _wrap_syncgen_fixture( + fixture_function: Callable[ + SyncGenFixtureParams, Generator[SyncGenFixtureYieldType] + ], + runner: Runner, +) -> Callable[SyncGenFixtureParams, Generator[SyncGenFixtureYieldType]]: + @functools.wraps(fixture_function) + def _syncgen_fixture_wrapper( + *args: SyncGenFixtureParams.args, + **kwargs: SyncGenFixtureParams.kwargs, + ) -> Generator[SyncGenFixtureYieldType]: + with _temporary_event_loop(runner.get_loop()): + yield from fixture_function(*args, **kwargs) + + return _syncgen_fixture_wrapper + + +SyncFixtureParams = ParamSpec("SyncFixtureParams") +SyncFixtureReturnType = TypeVar("SyncFixtureReturnType") + + +def _wrap_sync_fixture( + fixture_function: Callable[SyncFixtureParams, SyncFixtureReturnType], + runner: Runner, +) -> Callable[SyncFixtureParams, SyncFixtureReturnType]: + @functools.wraps(fixture_function) + def _sync_fixture_wrapper( + *args: SyncFixtureParams.args, + **kwargs: SyncFixtureParams.kwargs, + ) -> SyncFixtureReturnType: + with _temporary_event_loop(runner.get_loop()): + return fixture_function(*args, **kwargs) + + return _sync_fixture_wrapper + + +AsyncGenFixtureParams = ParamSpec("AsyncGenFixtureParams") +AsyncGenFixtureYieldType = TypeVar("AsyncGenFixtureYieldType") + + +def _wrap_asyncgen_fixture( + fixture_function: Callable[ + AsyncGenFixtureParams, AsyncGeneratorType[AsyncGenFixtureYieldType, Any] + ], + runner: Runner, + request: FixtureRequest, +) -> Callable[AsyncGenFixtureParams, AsyncGenFixtureYieldType]: + @functools.wraps(fixture_function) + def _asyncgen_fixture_wrapper( + *args: AsyncGenFixtureParams.args, + **kwargs: AsyncGenFixtureParams.kwargs, + ): + gen_obj = fixture_function(*args, **kwargs) + + async def setup(): + res = await gen_obj.__anext__() + return res + + context = contextvars.copy_context() + result = runner.run(setup(), context=context) + + reset_contextvars = _apply_contextvar_changes(context) + + def finalizer() -> None: + """Yield again, to finalize.""" + + async def async_finalizer() -> None: + try: + await gen_obj.__anext__() + except StopAsyncIteration: + pass + else: + msg = "Async generator fixture didn't stop." + msg += "Yield only once." + raise ValueError(msg) + + runner.run(async_finalizer(), context=context) + if reset_contextvars is not None: + reset_contextvars() + + request.addfinalizer(finalizer) + return result + + return _asyncgen_fixture_wrapper + + +AsyncFixtureParams = ParamSpec("AsyncFixtureParams") +AsyncFixtureReturnType = TypeVar("AsyncFixtureReturnType") + + +def _wrap_async_fixture( + fixture_function: Callable[ + AsyncFixtureParams, CoroutineType[Any, Any, AsyncFixtureReturnType] + ], + runner: Runner, + request: FixtureRequest, +) -> Callable[AsyncFixtureParams, AsyncFixtureReturnType]: + @functools.wraps(fixture_function) + def _async_fixture_wrapper( + *args: AsyncFixtureParams.args, + **kwargs: AsyncFixtureParams.kwargs, + ): + async def setup(): + res = await fixture_function(*args, **kwargs) + return res + + context = contextvars.copy_context() + result = runner.run(setup(), context=context) + + # Copy the context vars modified by the setup task into the current + # context, and (if needed) add a finalizer to reset them. + # + # Note that this is slightly different from the behavior of a non-async + # fixture, which would rely on the fixture author to add a finalizer + # to reset the variables. In this case, the author of the fixture can't + # write such a finalizer because they have no way to capture the Context + # in which the setup function was run, so we need to do it for them. + reset_contextvars = _apply_contextvar_changes(context) + if reset_contextvars is not None: + request.addfinalizer(reset_contextvars) + + return result + + return _async_fixture_wrapper + + +def _apply_contextvar_changes( + context: contextvars.Context, +) -> Callable[[], None] | None: + """ + Copy contextvar changes from the given context to the current context. + + If any contextvars were modified by the fixture, return a finalizer that + will restore them. + """ + context_tokens = [] + for var in context: + try: + if var.get() is context.get(var): + # This variable is not modified, so leave it as-is. + continue + except LookupError: + # This variable isn't yet set in the current context at all. + pass + token = var.set(context.get(var)) + context_tokens.append((var, token)) + + if not context_tokens: + return None + + def restore_contextvars(): + while context_tokens: + var, token = context_tokens.pop() + var.reset(token) + + return restore_contextvars + + +@pytest.hookimpl(wrapper=True) +def pytest_fixture_setup(fixturedef: FixtureDef, request) -> object | None: + asyncio_mode = _get_asyncio_mode(request.config) + if not _owns_fixture(fixturedef.func, asyncio_mode): + return (yield) + loop_scope = _effective_fixture_loop_scope( + fixturedef.func, fixturedef.scope, request.config + ) + runner_fixture_id = f"_{loop_scope}_scoped_runner" + runner = request.getfixturevalue(runner_fixture_id) + # Prevent the runner closing before the fixture's async teardown. + runner_fixturedef = request._get_active_fixturedef(runner_fixture_id) + runner_fixturedef.addfinalizer( + functools.partial(fixturedef.finish, request=request) + ) + synchronizer = _fixture_synchronizer(fixturedef, runner, request) + _make_asyncio_fixture_function(synchronizer, loop_scope) + with MonkeyPatch.context() as c: + c.setattr(fixturedef, "func", synchronizer) + hook_result = yield + return hook_result diff --git a/pytest_asyncio/_hooks.py b/pytest_asyncio/_hooks.py new file mode 100644 index 00000000..e9785fb9 --- /dev/null +++ b/pytest_asyncio/_hooks.py @@ -0,0 +1,35 @@ +"""Hook specifications and shared type aliases for pytest-asyncio.""" + +from __future__ import annotations + +import inspect +from asyncio import AbstractEventLoop +from collections.abc import Callable, Mapping +from typing import Literal, TypeAlias + +import pluggy +from pytest import Config, Item + +_ScopeName = Literal["session", "package", "module", "class", "function"] +LoopFactory: TypeAlias = Callable[[], AbstractEventLoop] + + +def _is_coroutine_or_asyncgen(obj: object) -> bool: + return inspect.iscoroutinefunction(obj) or inspect.isasyncgenfunction(obj) + + +class PytestAsyncioError(Exception): + """Base class for exceptions raised by pytest-asyncio""" + + +hookspec = pluggy.HookspecMarker("pytest") + + +class PytestAsyncioSpecs: + @hookspec(firstresult=True) + def pytest_asyncio_loop_factories( + self, + config: Config, + item: Item, + ) -> Mapping[str, LoopFactory] | None: + raise NotImplementedError # pragma: no cover diff --git a/pytest_asyncio/_markers.py b/pytest_asyncio/_markers.py new file mode 100644 index 00000000..7639b23a --- /dev/null +++ b/pytest_asyncio/_markers.py @@ -0,0 +1,86 @@ +"""Parsing, validation, and mode-aware resolution of @pytest.mark.asyncio.""" + +from __future__ import annotations + +from collections.abc import Mapping, Sequence + +import pytest +from pytest import Config, Function, Item, Mark + +from ._config import Mode, _get_asyncio_mode +from ._hooks import LoopFactory, _ScopeName + +_INVALID_LOOP_FACTORIES = """\ +pytest_asyncio_loop_factories must return a non-empty mapping of \ +factory names to callables. +""" + + +def _collect_hook_loop_factories( + config: Config, + item: Item, +) -> dict[str, LoopFactory] | None: + hook_caller = item.ihook.pytest_asyncio_loop_factories + if not hook_caller.get_hookimpls(): + return None + + result = hook_caller(config=config, item=item) + if result is None or not isinstance(result, Mapping): + raise pytest.UsageError(_INVALID_LOOP_FACTORIES) + # Copy into an isolated snapshot so later mutations of the hook's + # original container do not affect parametrization. + factories = dict(result) + if not factories or any( + not isinstance(name, str) or not name or not callable(factory) + for name, factory in factories.items() + ): + raise pytest.UsageError(_INVALID_LOOP_FACTORIES) + return factories + + +_INVALID_LOOP_FACTORIES_KWARG = """\ +mark.asyncio 'loop_factories' must be a non-empty sequence of strings. +""" + + +def _parse_asyncio_marker( + asyncio_marker: Mark, +) -> tuple[_ScopeName | None, Sequence[str] | None]: + assert asyncio_marker.name == "asyncio" + _validate_asyncio_marker(asyncio_marker) + scope = asyncio_marker.kwargs.get("loop_scope") + if scope is not None: + assert scope in {"function", "class", "module", "package", "session"} + marker_value = asyncio_marker.kwargs.get("loop_factories") + if marker_value is None: + return scope, None + if isinstance(marker_value, str) or not isinstance(marker_value, Sequence): + raise ValueError(_INVALID_LOOP_FACTORIES_KWARG) + if not marker_value or any( + not isinstance(factory_name, str) or not factory_name + for factory_name in marker_value + ): + raise ValueError(_INVALID_LOOP_FACTORIES_KWARG) + return scope, marker_value + + +def _validate_asyncio_marker(asyncio_marker: Mark) -> None: + if asyncio_marker.args or ( + asyncio_marker.kwargs + and set(asyncio_marker.kwargs) - {"loop_scope", "loop_factories"} + ): + msg = ( + "mark.asyncio accepts only keyword arguments 'loop_scope' and" + " 'loop_factories'." + ) + raise ValueError(msg) + + +def _resolve_asyncio_marker(item: Function) -> Mark | None: + marker = item.get_closest_marker("asyncio") + if marker is not None: + return marker + if _get_asyncio_mode(item.config) == Mode.AUTO: + item.add_marker("asyncio") + return item.get_closest_marker("asyncio") + return None diff --git a/pytest_asyncio/_mismatch.py b/pytest_asyncio/_mismatch.py new file mode 100644 index 00000000..175d73d1 --- /dev/null +++ b/pytest_asyncio/_mismatch.py @@ -0,0 +1,55 @@ +""" +Detects tests/fixtures whose effective loop_scope differs from a fixture +they (transitively) depend on -- i.e. a test or fixture that ends up running +on a different event loop than an async fixture it uses. See issue #1514.""" + +from __future__ import annotations + +import pytest +from pytest import Config, Function + +from ._config import _get_asyncio_mode +from ._fixtures import _effective_fixture_loop_scope, _owns_fixture +from ._hooks import _ScopeName + + +class PytestAsyncioLoopScopeMismatchWarning(pytest.PytestWarning): + """ + Warns that a test or fixture requests an async fixture whose effective + loop_scope differs from its own. The two will run on different event + loops, which can silently break objects (e.g. asyncio.Future, Task, or + Lock) that are bound to the loop they were created on. + """ + + +def _compute_loop_scope_mismatches( + item: Function, test_loop_scope: _ScopeName, config: Config +) -> list[tuple[str, _ScopeName]]: + """ + Walk the full transitive closure of `item`'s fixture dependencies and + return (fixture_name, fixture_loop_scope) for every async-owned fixture + whose effective loop_scope differs from `test_loop_scope`. + + `names_closure` is pytest's own deduplicated transitive closure (each + fixture name appears at most once, however many paths reach it), so a + single pass here can't produce duplicate entries for the same fixture + even in a diamond-shaped dependency graph. + """ + mode = _get_asyncio_mode(config) + mismatches: list[tuple[str, _ScopeName]] = [] + for name in item._fixtureinfo.names_closure: + fixturedefs = item._fixtureinfo.name2fixturedefs.get(name) + if not fixturedefs: + continue + # The last entry is the closest/effective one, same convention used + # elsewhere in the plugin (e.g. pytest_pyfunc_call's strict-mode + # check). + fixturedef = fixturedefs[-1] + if not _owns_fixture(fixturedef.func, mode): + continue + fixture_loop_scope = _effective_fixture_loop_scope( + fixturedef.func, fixturedef.scope, config + ) + if fixture_loop_scope != test_loop_scope: + mismatches.append((name, fixture_loop_scope)) + return mismatches diff --git a/pytest_asyncio/_runner.py b/pytest_asyncio/_runner.py new file mode 100644 index 00000000..e7a02e9f --- /dev/null +++ b/pytest_asyncio/_runner.py @@ -0,0 +1,149 @@ +"""Event loop / asyncio.Runner lifecycle management, scoped by pytest fixture scope.""" + +from __future__ import annotations + +import asyncio +import contextlib +import sys +import traceback +import warnings +from asyncio import AbstractEventLoop +from collections.abc import Callable, Iterator +from typing import TYPE_CHECKING + +import pytest +from pytest import FixtureRequest + +from ._config import _get_asyncio_debug +from ._hooks import LoopFactory, _ScopeName + +if sys.version_info >= (3, 11): + from asyncio import Runner +else: + from backports.asyncio.runner import Runner + +if TYPE_CHECKING: + # AbstractEventLoopPolicy is deprecated and scheduled for removal in Python 3.16 + # Import it for type checking only to avoid raising a DeprecationWarning. + from asyncio import AbstractEventLoopPolicy + + +@contextlib.contextmanager +def _temporary_event_loop(loop: AbstractEventLoop) -> Iterator[None]: + try: + old_loop = _get_event_loop_no_warn() + except RuntimeError: + old_loop = None + if old_loop is loop: + yield + return + _set_event_loop(loop) + try: + yield + finally: + _set_event_loop(old_loop) + + +@contextlib.contextmanager +def _temporary_event_loop_policy( + policy: AbstractEventLoopPolicy, +) -> Iterator[None]: + old_loop_policy = _get_event_loop_policy() + _set_event_loop_policy(policy) + try: + yield + finally: + _set_event_loop_policy(old_loop_policy) + + +def _get_event_loop_policy() -> AbstractEventLoopPolicy: + with warnings.catch_warnings(): + warnings.simplefilter("ignore", DeprecationWarning) + return asyncio.get_event_loop_policy() + + +def _set_event_loop_policy(policy: AbstractEventLoopPolicy) -> None: + with warnings.catch_warnings(): + warnings.simplefilter("ignore", DeprecationWarning) + asyncio.set_event_loop_policy(policy) + + +def _get_event_loop_no_warn( + policy: AbstractEventLoopPolicy | None = None, +) -> asyncio.AbstractEventLoop: + with warnings.catch_warnings(): + warnings.simplefilter("ignore", DeprecationWarning) + if policy is not None: + return policy.get_event_loop() + else: + return asyncio.get_event_loop() + + +def _set_event_loop(loop: AbstractEventLoop | None) -> None: + with warnings.catch_warnings(): + warnings.simplefilter("ignore", DeprecationWarning) + asyncio.set_event_loop(loop) + + +_RUNNER_TEARDOWN_WARNING = """\ +An exception occurred during teardown of an asyncio.Runner. \ +The reason is likely that you closed the underlying event loop in a test, \ +which prevents the cleanup of asynchronous generators by the runner. +This warning will become an error in future versions of pytest-asyncio. \ +Please ensure that your tests don't close the event loop. \ +Here is the traceback of the exception triggered during teardown: +%s +""" + + +def _create_scoped_runner_fixture(scope: _ScopeName) -> Callable: + @pytest.fixture( + scope=scope, + name=f"_{scope}_scoped_runner", + ) + def _scoped_runner( + _asyncio_loop_factory, + request: FixtureRequest, + ) -> Iterator[Runner]: + new_loop_policy = _get_event_loop_policy() + debug_mode = _get_asyncio_debug(request.config) + with _temporary_event_loop_policy(new_loop_policy): + runner = Runner( + debug=debug_mode, + loop_factory=_asyncio_loop_factory, + ).__enter__() + if _asyncio_loop_factory is not None: + _set_event_loop(runner.get_loop()) + try: + yield runner + except Exception as e: + runner.__exit__(type(e), e, e.__traceback__) + else: + with warnings.catch_warnings(): + warnings.filterwarnings( + "ignore", ".*BaseEventLoop.shutdown_asyncgens.*", RuntimeWarning + ) + try: + runner.__exit__(None, None, None) + except RuntimeError: + warnings.warn( + _RUNNER_TEARDOWN_WARNING % traceback.format_exc(), + RuntimeWarning, + ) + finally: + if _asyncio_loop_factory is not None: + _set_event_loop(None) + + return _scoped_runner + + +_function_scoped_runner = _create_scoped_runner_fixture("function") +_class_scoped_runner = _create_scoped_runner_fixture("class") +_module_scoped_runner = _create_scoped_runner_fixture("module") +_package_scoped_runner = _create_scoped_runner_fixture("package") +_session_scoped_runner = _create_scoped_runner_fixture("session") + + +@pytest.fixture(scope="session") +def _asyncio_loop_factory(request: FixtureRequest) -> LoopFactory | None: + return getattr(request, "param", None) diff --git a/pytest_asyncio/_usage_checks.py b/pytest_asyncio/_usage_checks.py new file mode 100644 index 00000000..5390f618 --- /dev/null +++ b/pytest_asyncio/_usage_checks.py @@ -0,0 +1,20 @@ +"""Collection-time usage checks: fail fast on removed public API.""" + +from __future__ import annotations + +import pytest + +_EVENT_LOOP_POLICY_FIXTURE_REMOVED_MESSAGE = """\ +The "event_loop_policy" fixture was removed in pytest-asyncio 2.0. Defining \ +a fixture with this name no longer has any effect on event loop creation. \ +Use the "pytest_asyncio_loop_factories" hook to customize event loop \ +creation instead. See the migration guide: \ +https://pytest-asyncio.readthedocs.io/en/stable/how-to-guides/migrate_from_1_x.html\ +""" + + +@pytest.hookimpl +def pytest_collection_finish(session: pytest.Session) -> None: + fixturedefs = session._fixturemanager._arg2fixturedefs.get("event_loop_policy", ()) + if fixturedefs: + raise pytest.UsageError(_EVENT_LOOP_POLICY_FIXTURE_REMOVED_MESSAGE) diff --git a/pytest_asyncio/plugin.py b/pytest_asyncio/plugin.py index 38b75e41..d1e83178 100644 --- a/pytest_asyncio/plugin.py +++ b/pytest_asyncio/plugin.py @@ -2,1087 +2,58 @@ from __future__ import annotations -import asyncio import contextlib -import contextvars -import enum -import functools -import inspect import socket -import sys -import traceback -import warnings -from asyncio import AbstractEventLoop -from collections.abc import ( - AsyncIterable, - AsyncIterator, - Awaitable, - Callable, - Collection, - Generator, - Iterable, - Iterator, - Mapping, - Sequence, -) -from types import AsyncGeneratorType, CoroutineType -from typing import ( - TYPE_CHECKING, - Any, - Literal, - ParamSpec, - TypeAlias, - TypeVar, - overload, -) +from collections.abc import Callable -import pluggy import pytest -from _pytest.fixtures import resolve_fixture_function -from _pytest.scope import Scope -from pytest import ( - Config, - FixtureDef, - FixtureRequest, - Function, - Item, - Mark, - MonkeyPatch, - Parser, - PytestCollectionWarning, - PytestDeprecationWarning, - PytestPluginManager, -) - -if sys.version_info >= (3, 11): - from asyncio import Runner -else: - from backports.asyncio.runner import Runner - -if sys.version_info >= (3, 13): - from typing import TypeIs -else: - from typing_extensions import TypeIs - -if TYPE_CHECKING: - # AbstractEventLoopPolicy is deprecated and scheduled for removal in Python 3.16 - # Import it for type checking only to avoid raising a DeprecationWarning. - from asyncio import AbstractEventLoopPolicy - -_ScopeName = Literal["session", "package", "module", "class", "function"] -_R = TypeVar("_R", bound=Awaitable | AsyncIterable | AsyncIterator) -_P = ParamSpec("_P") -FixtureFunction = Callable[_P, _R] -LoopFactory: TypeAlias = Callable[[], AbstractEventLoop] - - -class PytestAsyncioError(Exception): - """Base class for exceptions raised by pytest-asyncio""" - - -class Mode(str, enum.Enum): - AUTO = "auto" - STRICT = "strict" - - -hookspec = pluggy.HookspecMarker("pytest") - - -class PytestAsyncioSpecs: - @hookspec(firstresult=True) - def pytest_asyncio_loop_factories( - self, - config: Config, - item: Item, - ) -> Mapping[str, LoopFactory] | None: - raise NotImplementedError # pragma: no cover - - -ASYNCIO_MODE_HELP = """\ -'auto' - for automatically handling all async functions by the plugin -'strict' - for autoprocessing disabling (useful if different async frameworks \ -should be tested together, e.g. \ -both pytest-asyncio and pytest-trio are used in the same project) -""" - - -def pytest_addoption(parser: Parser, pluginmanager: PytestPluginManager) -> None: - pluginmanager.add_hookspecs(PytestAsyncioSpecs) - group = parser.getgroup("asyncio") - group.addoption( - "--asyncio-mode", - dest="asyncio_mode", - default=None, - metavar="MODE", - help=ASYNCIO_MODE_HELP, - ) - group.addoption( - "--asyncio-debug", - dest="asyncio_debug", - action="store_true", - default=None, - help="enable asyncio debug mode for the default event loop", - ) - parser.addini( - "asyncio_mode", - help="default value for --asyncio-mode", - default="strict", - ) - parser.addini( - "asyncio_debug", - help="enable asyncio debug mode for the default event loop", - type="bool", - default="false", - ) - parser.addini( - "asyncio_default_fixture_loop_scope", - type="string", - help="default scope of the asyncio event loop used to execute async fixtures", - default=None, - ) - parser.addini( - "asyncio_default_test_loop_scope", - type="string", - help="default scope of the asyncio event loop used to execute tests", - default="function", - ) - - -@overload -def fixture( - fixture_function: FixtureFunction[_P, _R], - *, - scope: _ScopeName | Callable[[str, Config], _ScopeName] = ..., - loop_scope: _ScopeName | None = ..., - params: Iterable[object] | None = ..., - autouse: bool = ..., - ids: ( - Iterable[str | float | int | bool | None] - | Callable[[Any], object | None] - | None - ) = ..., - name: str | None = ..., -) -> FixtureFunction[_P, _R]: ... - - -@overload -def fixture( - fixture_function: None = ..., - *, - scope: _ScopeName | Callable[[str, Config], _ScopeName] = ..., - loop_scope: _ScopeName | None = ..., - params: Iterable[object] | None = ..., - autouse: bool = ..., - ids: ( - Iterable[str | float | int | bool | None] - | Callable[[Any], object | None] - | None - ) = ..., - name: str | None = None, -) -> Callable[[FixtureFunction[_P, _R]], FixtureFunction[_P, _R]]: ... - - -def fixture( - fixture_function: FixtureFunction[_P, _R] | None = None, - loop_scope: _ScopeName | None = None, - **kwargs: Any, -) -> ( - FixtureFunction[_P, _R] - | Callable[[FixtureFunction[_P, _R]], FixtureFunction[_P, _R]] -): - if fixture_function is not None: - _make_asyncio_fixture_function(fixture_function, loop_scope) - return pytest.fixture(fixture_function, **kwargs) - - else: - - @functools.wraps(fixture) - def inner(fixture_function: FixtureFunction[_P, _R]) -> FixtureFunction[_P, _R]: - return fixture(fixture_function, loop_scope=loop_scope, **kwargs) - - return inner - - -def _is_asyncio_fixture_function(obj: Any) -> bool: - obj = getattr(obj, "__func__", obj) # instance method maybe? - return getattr(obj, "_force_asyncio_fixture", False) - - -def _make_asyncio_fixture_function(obj: Any, loop_scope: _ScopeName | None) -> None: - if hasattr(obj, "__func__"): - # instance method, check the function object - obj = obj.__func__ - obj._force_asyncio_fixture = True - obj._loop_scope = loop_scope - - -def _is_coroutine_or_asyncgen(obj: Any) -> bool: - return inspect.iscoroutinefunction(obj) or inspect.isasyncgenfunction(obj) - - -def _get_asyncio_mode(config: Config) -> Mode: - val = config.getoption("asyncio_mode") - if val is None: - val = config.getini("asyncio_mode") - try: - return Mode(val) - except ValueError as e: - modes = ", ".join(m.value for m in Mode) - raise pytest.UsageError( - f"{val!r} is not a valid asyncio_mode. Valid modes: {modes}." - ) from e - - -def _get_asyncio_debug(config: Config) -> bool: - val = config.getoption("asyncio_debug") - if val is None: - val = config.getini("asyncio_debug") - - if isinstance(val, bool): - return val - else: - return val == "true" - - -_INVALID_LOOP_FACTORIES = """\ -pytest_asyncio_loop_factories must return a non-empty mapping of \ -factory names to callables. -""" - - -def _collect_hook_loop_factories( - config: Config, - item: Item, -) -> dict[str, LoopFactory] | None: - hook_caller = item.ihook.pytest_asyncio_loop_factories - if not hook_caller.get_hookimpls(): - return None - - result = hook_caller(config=config, item=item) - if result is None or not isinstance(result, Mapping): - raise pytest.UsageError(_INVALID_LOOP_FACTORIES) - # Copy into an isolated snapshot so later mutations of the hook's - # original container do not affect parametrization. - factories = dict(result) - if not factories or any( - not isinstance(name, str) or not name or not callable(factory) - for name, factory in factories.items() - ): - raise pytest.UsageError(_INVALID_LOOP_FACTORIES) - return factories - - -_DEFAULT_FIXTURE_LOOP_SCOPE_UNSET = """\ -The configuration option "asyncio_default_fixture_loop_scope" is unset. -The event loop scope for asynchronous fixtures will default to the "fixture" caching \ -scope. Future versions of pytest-asyncio will default the loop scope for asynchronous \ -fixtures to "function" scope. Set the default fixture loop scope explicitly in order \ -to avoid unexpected behavior in the future. Valid fixture loop scopes are: \ -"function", "class", "module", "package", "session" -""" - - -def _validate_scope(scope: str | None, option_name: str) -> None: - if scope is None: - return - valid_scopes = [s.value for s in Scope] - if scope not in valid_scopes: - raise pytest.UsageError( - f"{scope!r} is not a valid {option_name}. " - f"Valid scopes are: {', '.join(valid_scopes)}." - ) - - -def pytest_configure(config: Config) -> None: - default_fixture_loop_scope = config.getini("asyncio_default_fixture_loop_scope") - _validate_scope(default_fixture_loop_scope, "asyncio_default_fixture_loop_scope") - if not default_fixture_loop_scope: - warnings.warn(PytestDeprecationWarning(_DEFAULT_FIXTURE_LOOP_SCOPE_UNSET)) - - default_test_loop_scope = config.getini("asyncio_default_test_loop_scope") - _validate_scope(default_test_loop_scope, "asyncio_default_test_loop_scope") - config.addinivalue_line( - "markers", - "asyncio: " - "mark the test as a coroutine, it will be " - "run using an asyncio event loop", - ) - - -@pytest.hookimpl(tryfirst=True) -def pytest_report_header(config: Config) -> list[str]: - """Add asyncio config to pytest header.""" - mode = _get_asyncio_mode(config) - debug = _get_asyncio_debug(config) - default_fixture_loop_scope = config.getini("asyncio_default_fixture_loop_scope") - default_test_loop_scope = _get_default_test_loop_scope(config) - header = [ - f"mode={mode}", - f"debug={debug}", - f"asyncio_default_fixture_loop_scope={default_fixture_loop_scope}", - f"asyncio_default_test_loop_scope={default_test_loop_scope}", - ] - return [ - "asyncio: " + ", ".join(header), - ] - - -def _fixture_synchronizer( - fixturedef: FixtureDef, runner: Runner, request: FixtureRequest -) -> Callable: - """Returns a synchronous function evaluating the specified fixture.""" - fixture_function = resolve_fixture_function(fixturedef, request) - if inspect.isasyncgenfunction(fixturedef.func): - return _wrap_asyncgen_fixture(fixture_function, runner, request) # type: ignore[arg-type] - elif inspect.iscoroutinefunction(fixturedef.func): - return _wrap_async_fixture(fixture_function, runner, request) # type: ignore[arg-type] - elif inspect.isgeneratorfunction(fixturedef.func): - return _wrap_syncgen_fixture(fixture_function, runner) # type: ignore[arg-type] - else: - return _wrap_sync_fixture(fixture_function, runner) # type: ignore[arg-type] - - -SyncGenFixtureParams = ParamSpec("SyncGenFixtureParams") -SyncGenFixtureYieldType = TypeVar("SyncGenFixtureYieldType") - - -def _wrap_syncgen_fixture( - fixture_function: Callable[ - SyncGenFixtureParams, Generator[SyncGenFixtureYieldType] - ], - runner: Runner, -) -> Callable[SyncGenFixtureParams, Generator[SyncGenFixtureYieldType]]: - @functools.wraps(fixture_function) - def _syncgen_fixture_wrapper( - *args: SyncGenFixtureParams.args, - **kwargs: SyncGenFixtureParams.kwargs, - ) -> Generator[SyncGenFixtureYieldType]: - with _temporary_event_loop(runner.get_loop()): - yield from fixture_function(*args, **kwargs) - - return _syncgen_fixture_wrapper - - -SyncFixtureParams = ParamSpec("SyncFixtureParams") -SyncFixtureReturnType = TypeVar("SyncFixtureReturnType") - - -def _wrap_sync_fixture( - fixture_function: Callable[SyncFixtureParams, SyncFixtureReturnType], - runner: Runner, -) -> Callable[SyncFixtureParams, SyncFixtureReturnType]: - @functools.wraps(fixture_function) - def _sync_fixture_wrapper( - *args: SyncFixtureParams.args, - **kwargs: SyncFixtureParams.kwargs, - ) -> SyncFixtureReturnType: - with _temporary_event_loop(runner.get_loop()): - return fixture_function(*args, **kwargs) - - return _sync_fixture_wrapper - - -AsyncGenFixtureParams = ParamSpec("AsyncGenFixtureParams") -AsyncGenFixtureYieldType = TypeVar("AsyncGenFixtureYieldType") - - -def _wrap_asyncgen_fixture( - fixture_function: Callable[ - AsyncGenFixtureParams, AsyncGeneratorType[AsyncGenFixtureYieldType, Any] - ], - runner: Runner, - request: FixtureRequest, -) -> Callable[AsyncGenFixtureParams, AsyncGenFixtureYieldType]: - @functools.wraps(fixture_function) - def _asyncgen_fixture_wrapper( - *args: AsyncGenFixtureParams.args, - **kwargs: AsyncGenFixtureParams.kwargs, - ): - gen_obj = fixture_function(*args, **kwargs) - - async def setup(): - res = await gen_obj.__anext__() - return res - - context = contextvars.copy_context() - result = runner.run(setup(), context=context) - - reset_contextvars = _apply_contextvar_changes(context) - - def finalizer() -> None: - """Yield again, to finalize.""" - - async def async_finalizer() -> None: - try: - await gen_obj.__anext__() - except StopAsyncIteration: - pass - else: - msg = "Async generator fixture didn't stop." - msg += "Yield only once." - raise ValueError(msg) - - runner.run(async_finalizer(), context=context) - if reset_contextvars is not None: - reset_contextvars() - - request.addfinalizer(finalizer) - return result - - return _asyncgen_fixture_wrapper - - -AsyncFixtureParams = ParamSpec("AsyncFixtureParams") -AsyncFixtureReturnType = TypeVar("AsyncFixtureReturnType") - - -def _wrap_async_fixture( - fixture_function: Callable[ - AsyncFixtureParams, CoroutineType[Any, Any, AsyncFixtureReturnType] - ], - runner: Runner, - request: FixtureRequest, -) -> Callable[AsyncFixtureParams, AsyncFixtureReturnType]: - @functools.wraps(fixture_function) - def _async_fixture_wrapper( - *args: AsyncFixtureParams.args, - **kwargs: AsyncFixtureParams.kwargs, - ): - async def setup(): - res = await fixture_function(*args, **kwargs) - return res - - context = contextvars.copy_context() - result = runner.run(setup(), context=context) - - # Copy the context vars modified by the setup task into the current - # context, and (if needed) add a finalizer to reset them. - # - # Note that this is slightly different from the behavior of a non-async - # fixture, which would rely on the fixture author to add a finalizer - # to reset the variables. In this case, the author of the fixture can't - # write such a finalizer because they have no way to capture the Context - # in which the setup function was run, so we need to do it for them. - reset_contextvars = _apply_contextvar_changes(context) - if reset_contextvars is not None: - request.addfinalizer(reset_contextvars) - - return result - - return _async_fixture_wrapper - - -def _apply_contextvar_changes( - context: contextvars.Context, -) -> Callable[[], None] | None: - """ - Copy contextvar changes from the given context to the current context. - If any contextvars were modified by the fixture, return a finalizer that - will restore them. - """ - context_tokens = [] - for var in context: - try: - if var.get() is context.get(var): - # This variable is not modified, so leave it as-is. - continue - except LookupError: - # This variable isn't yet set in the current context at all. - pass - token = var.set(context.get(var)) - context_tokens.append((var, token)) - - if not context_tokens: - return None - - def restore_contextvars(): - while context_tokens: - var, token = context_tokens.pop() - var.reset(token) - - return restore_contextvars - - -class PytestAsyncioFunction(Function): - """Base class for all test functions managed by pytest-asyncio.""" - - @classmethod - def item_subclass_for(cls, item: Function, /) -> type[PytestAsyncioFunction] | None: - """ - Returns a subclass of PytestAsyncioFunction if there is a specialized subclass - for the specified function item. - - Return None if no specialized subclass exists for the specified item. - """ - for subclass in cls.__subclasses__(): - if subclass._can_substitute(item): - return subclass - return None - - @classmethod - def _from_function(cls, function: Function, /) -> Function: - """ - Instantiates this specific PytestAsyncioFunction type from the specified - Function item. - """ - assert function.get_closest_marker("asyncio") - assert function.parent is not None - subclass_instance = cls.from_parent( - function.parent, - name=function.name, - callspec=getattr(function, "callspec", None), - callobj=function.obj, - fixtureinfo=function._fixtureinfo, - keywords=function.keywords, - originalname=function.originalname, - ) - subclass_instance.own_markers = function.own_markers - assert subclass_instance.own_markers == function.own_markers - return subclass_instance - - @staticmethod - def _can_substitute(item: Function) -> bool: - """Returns whether the specified function can be replaced by this class""" - raise NotImplementedError() - - def setup(self) -> None: - runner_fixture_id = f"_{self._loop_scope}_scoped_runner" - if runner_fixture_id not in self.fixturenames: - self.fixturenames.append(runner_fixture_id) - # When loop factories are configured, resolve the loop factory - # fixture early so that a factory variant change cascades cache - # invalidation before any async fixture checks its cache. - hook_caller = self.config.hook.pytest_asyncio_loop_factories - if hook_caller.get_hookimpls(): - _ = self._request.getfixturevalue(_asyncio_loop_factory.__name__) - return super().setup() - - def runtest(self) -> None: - runner_fixture_id = f"_{self._loop_scope}_scoped_runner" - runner = self._request.getfixturevalue(runner_fixture_id) - context = contextvars.copy_context() - synchronized_obj = _synchronize_coroutine( - getattr(*self._synchronization_target_attr), runner, context - ) - with MonkeyPatch.context() as c: - c.setattr(*self._synchronization_target_attr, synchronized_obj) - super().runtest() - - @functools.cached_property - def _loop_scope(self) -> _ScopeName: - """ - Return the scope of the asyncio event loop this item is run in. - - The effective scope is determined lazily. It is identical to to the - `loop_scope` value of the closest `asyncio` pytest marker. If no such - marker is present, the the loop scope is determined by the configuration - value of `asyncio_default_test_loop_scope`, instead. - """ - marker = self.get_closest_marker("asyncio") - assert marker is not None - default_loop_scope = _get_default_test_loop_scope(self.config) - loop_scope = marker.kwargs.get("loop_scope") or marker.kwargs.get("scope") - if loop_scope is None: - return default_loop_scope - else: - return loop_scope - - @property - def _synchronization_target_attr(self) -> tuple[object, str]: - """ - Return the coroutine that needs to be synchronized during the test run. - - This method is intended to be overwritten by subclasses when they need to apply - the coroutine synchronizer to a value that's different from self.obj - e.g. the AsyncHypothesisTest subclass. - """ - return self, "obj" - - -class Coroutine(PytestAsyncioFunction): - """Pytest item created by a coroutine""" - - @staticmethod - def _can_substitute(item: Function) -> bool: - func = item.obj - return inspect.iscoroutinefunction(func) - - -class AsyncGenerator(PytestAsyncioFunction): - """Pytest item created by an asynchronous generator""" - - @staticmethod - def _can_substitute(item: Function) -> bool: - func = item.obj - return inspect.isasyncgenfunction(func) - - @classmethod - def _from_function(cls, function: Function, /) -> Function: - async_gen_item = super()._from_function(function) - unsupported_item_type_message = ( - f"Tests based on asynchronous generators are not supported. " - f"{function.name} will be ignored." - ) - async_gen_item.warn(PytestCollectionWarning(unsupported_item_type_message)) - async_gen_item.add_marker( - pytest.mark.xfail(run=False, reason=unsupported_item_type_message) - ) - return async_gen_item - - -class AsyncStaticMethod(PytestAsyncioFunction): - """ - Pytest item that is a coroutine or an asynchronous generator - decorated with staticmethod - """ - - @staticmethod - def _can_substitute(item: Function) -> bool: - func = item.obj - return isinstance(func, staticmethod) and _is_coroutine_or_asyncgen( - func.__func__ - ) - - -class AsyncHypothesisTest(PytestAsyncioFunction): - """ - Pytest item that is coroutine or an asynchronous generator decorated by - @hypothesis.given. - """ - - def setup(self) -> None: - if not getattr(self.obj, "hypothesis", False) and getattr( - self.obj, "is_hypothesis_test", False - ): - pytest.fail( - f"test function `{self!r}` is using Hypothesis, but pytest-asyncio " - "only works with Hypothesis 3.64.0 or later." - ) - return super().setup() - - @staticmethod - def _can_substitute(item: Function) -> bool: - func = item.obj - return ( - getattr(func, "is_hypothesis_test", False) # type: ignore[return-value] - and getattr(func, "hypothesis", None) - and inspect.iscoroutinefunction(func.hypothesis.inner_test) - ) - - @property - def _synchronization_target_attr(self) -> tuple[object, str]: - return self.obj.hypothesis, "inner_test" - - -def _resolve_asyncio_marker(item: Function) -> Mark | None: - marker = item.get_closest_marker("asyncio") - if marker is not None: - return marker - if _get_asyncio_mode(item.config) == Mode.AUTO: - item.add_marker("asyncio") - return item.get_closest_marker("asyncio") - return None - - -# The function name needs to start with "pytest_" -# see https://github.com/pytest-dev/pytest/issues/11307 -@pytest.hookimpl(specname="pytest_pycollect_makeitem", hookwrapper=True) -def pytest_pycollect_makeitem_convert_async_functions_to_subclass( - collector: pytest.Module | pytest.Class, name: str, obj: object -) -> Generator[None, pluggy.Result, None]: - """ - Converts coroutines and async generators collected as pytest.Functions - to AsyncFunction items. - """ - hook_result = yield - try: - node_or_list_of_nodes: ( - pytest.Item | pytest.Collector | list[pytest.Item | pytest.Collector] | None - ) = hook_result.get_result() - except BaseException as e: - hook_result.force_exception(e) - return - if not node_or_list_of_nodes: - return - if isinstance(node_or_list_of_nodes, Sequence): - node_iterator = iter(node_or_list_of_nodes) - else: - # Treat single node as a single-element iterable - node_iterator = iter((node_or_list_of_nodes,)) - updated_node_collection = [] - for node in node_iterator: - updated_item = node - if isinstance(node, Function): - specialized_item_class = PytestAsyncioFunction.item_subclass_for(node) - if ( - specialized_item_class is not None - and _resolve_asyncio_marker(node) is not None - ): - updated_item = specialized_item_class._from_function(node) - updated_node_collection.append(updated_item) - hook_result.force_result(updated_node_collection) - - -@pytest.hookimpl(tryfirst=True) -def pytest_generate_tests(metafunc: pytest.Metafunc) -> None: - specialized_item_class = PytestAsyncioFunction.item_subclass_for( - metafunc.definition - ) - if specialized_item_class is None: - return - - asyncio_marker = _resolve_asyncio_marker(metafunc.definition) - if asyncio_marker is None: - return - marker_loop_scope, marker_selected_factory_names = _parse_asyncio_marker( - asyncio_marker - ) - - hook_factories = _collect_hook_loop_factories(metafunc.config, metafunc.definition) - if hook_factories is None: - if marker_selected_factory_names is not None: - raise pytest.UsageError( - "mark.asyncio 'loop_factories' requires at least one " - "pytest_asyncio_loop_factories hook implementation." - ) - return - - factory_params: Collection[object] - factory_ids: Collection[str] - if marker_selected_factory_names is None: - factory_params = hook_factories.values() - factory_ids = hook_factories.keys() - else: - # 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)}." - ), - ), - ) - ) - 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 - # pytest.HIDDEN_PARAM was added in pytest 8.4 - hide_id = len(factory_ids) == 1 and hasattr(pytest, "HIDDEN_PARAM") - metafunc.parametrize( - _asyncio_loop_factory.__name__, - factory_params, - ids=(pytest.HIDDEN_PARAM,) if hide_id else factory_ids, - indirect=True, - scope=loop_scope, - ) - - -@contextlib.contextmanager -def _temporary_event_loop(loop: AbstractEventLoop) -> Iterator[None]: - try: - old_loop = _get_event_loop_no_warn() - except RuntimeError: - old_loop = None - if old_loop is loop: - yield - return - _set_event_loop(loop) - try: - yield - finally: - _set_event_loop(old_loop) - - -@contextlib.contextmanager -def _temporary_event_loop_policy( - policy: AbstractEventLoopPolicy, -) -> Iterator[None]: - old_loop_policy = _get_event_loop_policy() - _set_event_loop_policy(policy) - try: - yield - finally: - _set_event_loop_policy(old_loop_policy) - - -def _get_event_loop_policy() -> AbstractEventLoopPolicy: - with warnings.catch_warnings(): - warnings.simplefilter("ignore", DeprecationWarning) - return asyncio.get_event_loop_policy() - - -def _set_event_loop_policy(policy: AbstractEventLoopPolicy) -> None: - with warnings.catch_warnings(): - warnings.simplefilter("ignore", DeprecationWarning) - asyncio.set_event_loop_policy(policy) - - -def _get_event_loop_no_warn( - policy: AbstractEventLoopPolicy | None = None, -) -> asyncio.AbstractEventLoop: - with warnings.catch_warnings(): - warnings.simplefilter("ignore", DeprecationWarning) - if policy is not None: - return policy.get_event_loop() - else: - return asyncio.get_event_loop() - - -def _set_event_loop(loop: AbstractEventLoop | None) -> None: - with warnings.catch_warnings(): - warnings.simplefilter("ignore", DeprecationWarning) - asyncio.set_event_loop(loop) - - -@pytest.hookimpl(tryfirst=True, hookwrapper=True) -def pytest_pyfunc_call(pyfuncitem: Function) -> object | None: - """Pytest hook called before a test case is run.""" - if pyfuncitem.get_closest_marker("asyncio") is not None: - if is_async_test(pyfuncitem): - asyncio_mode = _get_asyncio_mode(pyfuncitem.config) - for fixname, fixtures in pyfuncitem._fixtureinfo.name2fixturedefs.items(): - # name2fixturedefs is a dict between fixture name and a list of matching - # fixturedefs. The last entry in the list is closest and the one used. - func = fixtures[-1].func - if ( - asyncio_mode == Mode.STRICT - and _is_coroutine_or_asyncgen(func) - and not _is_asyncio_fixture_function(func) - ): - warnings.warn( - PytestDeprecationWarning( - f"asyncio test {pyfuncitem.name!r} requested async " - "@pytest.fixture " - f"{fixname!r} in strict mode. " - "You might want to use @pytest_asyncio.fixture or switch " - "to auto mode. " - "This will become an error in future versions of " - "pytest-asyncio." - ), - stacklevel=1, - ) - # no stacklevel points at the users code, so we set stacklevel=1 - # so it at least indicates that it's the plugin complaining. - # Pytest gives the test file & name in the warnings summary at least - - else: - pyfuncitem.warn( - pytest.PytestWarning( - f"The test {pyfuncitem} is marked with '@pytest.mark.asyncio' " - "but it is not an async function. " - "Please remove the asyncio mark. " - "If the test is not marked explicitly, " - "check for global marks applied via 'pytestmark'." - ) - ) - yield - return None - - -def _synchronize_coroutine( - func: Callable[..., CoroutineType], - runner: asyncio.Runner, - context: contextvars.Context, -): - """ - Return a sync wrapper around a coroutine executing it in the - specified runner and context. - """ - - @functools.wraps(func) - def inner(*args, **kwargs): - coro = func(*args, **kwargs) - runner.run(coro, context=context) - - return inner - - -@pytest.hookimpl(wrapper=True) -def pytest_fixture_setup(fixturedef: FixtureDef, request) -> object | None: - if ( - fixturedef.argname == "event_loop_policy" - and fixturedef.func.__module__ != __name__ - ): - warnings.warn( - PytestDeprecationWarning(_EVENT_LOOP_POLICY_FIXTURE_DEPRECATION_WARNING), - ) - asyncio_mode = _get_asyncio_mode(request.config) - if not _is_asyncio_fixture_function(fixturedef.func): - if asyncio_mode == Mode.STRICT: - # Ignore async fixtures without explicit asyncio mark in strict mode - # This applies to pytest_trio fixtures, for example - return (yield) - if not _is_coroutine_or_asyncgen(fixturedef.func): - return (yield) - default_loop_scope = request.config.getini("asyncio_default_fixture_loop_scope") - loop_scope = ( - getattr(fixturedef.func, "_loop_scope", None) - or default_loop_scope - or fixturedef.scope - ) - runner_fixture_id = f"_{loop_scope}_scoped_runner" - runner = request.getfixturevalue(runner_fixture_id) - # Prevent the runner closing before the fixture's async teardown. - runner_fixturedef = request._get_active_fixturedef(runner_fixture_id) - runner_fixturedef.addfinalizer( - functools.partial(fixturedef.finish, request=request) - ) - synchronizer = _fixture_synchronizer(fixturedef, runner, request) - _make_asyncio_fixture_function(synchronizer, loop_scope) - with MonkeyPatch.context() as c: - c.setattr(fixturedef, "func", synchronizer) - hook_result = yield - return hook_result - - -_DUPLICATE_LOOP_SCOPE_DEFINITION_ERROR = """\ -An asyncio pytest marker defines both "scope" and "loop_scope", \ -but it should only use "loop_scope". -""" - -_MARKER_SCOPE_KWARG_DEPRECATION_WARNING = """\ -The "scope" keyword argument to the asyncio marker has been deprecated. \ -Please use the "loop_scope" argument instead. -""" - -_INVALID_LOOP_FACTORIES_KWARG = """\ -mark.asyncio 'loop_factories' must be a non-empty sequence of strings. -""" - -_EVENT_LOOP_POLICY_FIXTURE_DEPRECATION_WARNING = """\ -Overriding the "event_loop_policy" fixture is deprecated \ -and will be removed in a future version of pytest-asyncio. \ -Use the "pytest_asyncio_loop_factories" hook to customize event loop creation.\ -""" - - -def _parse_asyncio_marker( - asyncio_marker: Mark, -) -> tuple[_ScopeName | None, Sequence[str] | None]: - assert asyncio_marker.name == "asyncio" - _validate_asyncio_marker(asyncio_marker) - if "scope" in asyncio_marker.kwargs: - if "loop_scope" in asyncio_marker.kwargs: - raise pytest.UsageError(_DUPLICATE_LOOP_SCOPE_DEFINITION_ERROR) - warnings.warn(PytestDeprecationWarning(_MARKER_SCOPE_KWARG_DEPRECATION_WARNING)) - scope = asyncio_marker.kwargs.get("loop_scope") or asyncio_marker.kwargs.get( - "scope" - ) - if scope is not None: - assert scope in {"function", "class", "module", "package", "session"} - marker_value = asyncio_marker.kwargs.get("loop_factories") - if marker_value is None: - return scope, None - if isinstance(marker_value, str) or not isinstance(marker_value, Sequence): - raise ValueError(_INVALID_LOOP_FACTORIES_KWARG) - if not marker_value or any( - not isinstance(factory_name, str) or not factory_name - for factory_name in marker_value - ): - raise ValueError(_INVALID_LOOP_FACTORIES_KWARG) - return scope, marker_value - - -def _validate_asyncio_marker(asyncio_marker: Mark) -> None: - if asyncio_marker.args or ( - asyncio_marker.kwargs - and set(asyncio_marker.kwargs) - {"loop_scope", "scope", "loop_factories"} - ): - msg = ( - "mark.asyncio accepts only keyword arguments 'loop_scope' and" - " 'loop_factories'." - ) - raise ValueError(msg) - - -def _get_default_test_loop_scope(config: Config) -> Any: - return config.getini("asyncio_default_test_loop_scope") - - -_RUNNER_TEARDOWN_WARNING = """\ -An exception occurred during teardown of an asyncio.Runner. \ -The reason is likely that you closed the underlying event loop in a test, \ -which prevents the cleanup of asynchronous generators by the runner. -This warning will become an error in future versions of pytest-asyncio. \ -Please ensure that your tests don't close the event loop. \ -Here is the traceback of the exception triggered during teardown: -%s -""" - - -def _create_scoped_runner_fixture(scope: _ScopeName) -> Callable: - @pytest.fixture( - scope=scope, - name=f"_{scope}_scoped_runner", - ) - def _scoped_runner( - event_loop_policy, - _asyncio_loop_factory, - request: FixtureRequest, - ) -> Iterator[Runner]: - new_loop_policy = event_loop_policy - debug_mode = _get_asyncio_debug(request.config) - with _temporary_event_loop_policy(new_loop_policy): - runner = Runner( - debug=debug_mode, - loop_factory=_asyncio_loop_factory, - ).__enter__() - if _asyncio_loop_factory is not None: - _set_event_loop(runner.get_loop()) - try: - yield runner - except Exception as e: - runner.__exit__(type(e), e, e.__traceback__) - else: - with warnings.catch_warnings(): - warnings.filterwarnings( - "ignore", ".*BaseEventLoop.shutdown_asyncgens.*", RuntimeWarning - ) - try: - runner.__exit__(None, None, None) - except RuntimeError: - warnings.warn( - _RUNNER_TEARDOWN_WARNING % traceback.format_exc(), - RuntimeWarning, - ) - finally: - if _asyncio_loop_factory is not None: - _set_event_loop(None) - - return _scoped_runner - - -for scope in Scope: - globals()[f"_{scope.value}_scoped_runner"] = _create_scoped_runner_fixture( - scope.value - ) - - -@pytest.fixture(scope="session") -def _asyncio_loop_factory(request: FixtureRequest) -> LoopFactory | None: - return getattr(request, "param", None) - - -@pytest.fixture(scope="session", autouse=True) -def event_loop_policy() -> AbstractEventLoopPolicy: - """Return an instance of the policy used to create asyncio event loops.""" - return _get_event_loop_policy() - - -def is_async_test(item: Item) -> TypeIs[PytestAsyncioFunction]: - """Returns whether a test item is a pytest-asyncio test""" - return isinstance(item, PytestAsyncioFunction) +from ._collection import ( + is_async_test, + pytest_generate_tests, + pytest_pycollect_makeitem_tag_async_items, + pytest_runtest_setup, +) +from ._config import pytest_addoption, pytest_configure, pytest_report_header +from ._dispatch import pytest_pyfunc_call +from ._fixtures import fixture, pytest_fixture_setup +from ._hooks import PytestAsyncioError +from ._mismatch import PytestAsyncioLoopScopeMismatchWarning +from ._runner import ( + _asyncio_loop_factory, + _class_scoped_runner, + _function_scoped_runner, + _module_scoped_runner, + _package_scoped_runner, + _session_scoped_runner, +) +from ._usage_checks import pytest_collection_finish + +__all__ = [ + "PytestAsyncioError", + "PytestAsyncioLoopScopeMismatchWarning", + "_asyncio_loop_factory", + "_class_scoped_runner", + "_function_scoped_runner", + "_module_scoped_runner", + "_package_scoped_runner", + "_session_scoped_runner", + "fixture", + "is_async_test", + "pytest_addoption", + "pytest_collection_finish", + "pytest_configure", + "pytest_fixture_setup", + "pytest_generate_tests", + "pytest_pycollect_makeitem_tag_async_items", + "pytest_pyfunc_call", + "pytest_report_header", + "pytest_runtest_setup", + "unused_tcp_port", + "unused_tcp_port_factory", + "unused_udp_port", + "unused_udp_port_factory", +] def _unused_port(socket_type: int) -> int: diff --git a/tests/async_fixtures/test_shared_module_fixture.py b/tests/async_fixtures/test_shared_module_fixture.py index f37c0bb7..36741be1 100644 --- a/tests/async_fixtures/test_shared_module_fixture.py +++ b/tests/async_fixtures/test_shared_module_fixture.py @@ -28,5 +28,6 @@ async def test_shared_module_fixture_use_b(async_shared_module_fixture): assert async_shared_module_fixture is True """), ) - result = pytester.runpytest("--asyncio-mode=strict") + result = pytester.runpytest("--asyncio-mode=strict", "-W", "default") result.assert_outcomes(passed=2) + result.stdout.fnmatch_lines("*PytestAsyncioLoopScopeMismatchWarning*") diff --git a/tests/markers/test_class_scope.py b/tests/markers/test_class_scope.py index ce023fcc..746a501f 100644 --- a/tests/markers/test_class_scope.py +++ b/tests/markers/test_class_scope.py @@ -3,7 +3,6 @@ from __future__ import annotations import asyncio -import sys from textwrap import dedent import pytest @@ -98,83 +97,6 @@ async def test_this_runs_in_same_loop(self): result.assert_outcomes(passed=2) -def test_asyncio_mark_respects_the_loop_policy( - pytester: pytest.Pytester, -): - pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") - pytester.makepyfile(dedent("""\ - import asyncio - import pytest - - class CustomEventLoopPolicy(asyncio.DefaultEventLoopPolicy): - pass - - class TestUsesCustomEventLoop: - @pytest.fixture(scope="class") - @classmethod - def event_loop_policy(cls): - return CustomEventLoopPolicy() - - @pytest.mark.asyncio - async def test_uses_custom_event_loop_policy(self): - assert isinstance( - asyncio.get_event_loop_policy(), - CustomEventLoopPolicy, - ) - - @pytest.mark.asyncio - async def test_does_not_use_custom_event_loop_policy(): - assert not isinstance( - asyncio.get_event_loop_policy(), - CustomEventLoopPolicy, - ) - """)) - pytest_args = ["--asyncio-mode=strict"] - if sys.version_info >= (3, 14): - pytest_args.extend(["-W", "default"]) - result = pytester.runpytest(*pytest_args) - if sys.version_info >= (3, 14): - result.assert_outcomes(passed=2, warnings=4) - result.stdout.fnmatch_lines("*DefaultEventLoopPolicy*") - else: - result.assert_outcomes(passed=2) - - -def test_asyncio_mark_respects_parametrized_loop_policies( - pytester: pytest.Pytester, -): - pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") - pytester.makepyfile(dedent("""\ - import asyncio - - import pytest - - @pytest.fixture( - scope="class", - params=[ - asyncio.DefaultEventLoopPolicy(), - asyncio.DefaultEventLoopPolicy(), - ] - ) - def event_loop_policy(request): - return request.param - - @pytest.mark.asyncio(loop_scope="class") - class TestWithDifferentLoopPolicies: - async def test_parametrized_loop(self, request): - pass - """)) - pytest_args = ["--asyncio-mode=strict"] - if sys.version_info >= (3, 14): - pytest_args.extend(["-W", "default"]) - result = pytester.runpytest(*pytest_args) - if sys.version_info >= (3, 14): - result.assert_outcomes(passed=2, warnings=4) - result.stdout.fnmatch_lines("*DefaultEventLoopPolicy*") - else: - result.assert_outcomes(passed=2) - - def test_asyncio_mark_provides_class_scoped_loop_to_fixtures( pytester: pytest.Pytester, ): @@ -227,8 +149,9 @@ async def test_runs_in_different_loop_as_fixture(self, async_fixture): """), ) - result = pytester.runpytest("--asyncio-mode=strict") + result = pytester.runpytest("--asyncio-mode=strict", "-W", "default") result.assert_outcomes(passed=1) + result.stdout.fnmatch_lines("*PytestAsyncioLoopScopeMismatchWarning*") def test_asyncio_mark_handles_missing_event_loop_triggered_by_fixture( diff --git a/tests/markers/test_function_scope.py b/tests/markers/test_function_scope.py index 180d5c71..0ae0e25d 100644 --- a/tests/markers/test_function_scope.py +++ b/tests/markers/test_function_scope.py @@ -1,6 +1,5 @@ from __future__ import annotations -import sys from textwrap import dedent from pytest import Pytester @@ -61,144 +60,27 @@ async def test_raises(): """)) result = pytester.runpytest("--asyncio-mode=strict") result.assert_outcomes(errors=1) + result.stdout.fnmatch_lines( + "*mark.asyncio accepts only keyword arguments*'loop_scope'*'loop_factories'*" + ) -def test_warns_when_scope_argument_is_present(pytester: Pytester): +def test_raises_when_scope_argument_is_present(pytester: Pytester): pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") pytester.makepyfile(dedent("""\ import pytest @pytest.mark.asyncio(scope="function") - async def test_warns(): + async def test_raises(): ... """)) - result = pytester.runpytest("--asyncio-mode=strict", "-W", "default") - result.assert_outcomes(passed=1, warnings=1) - result.stdout.fnmatch_lines("*DeprecationWarning*") - - -def test_asyncio_mark_respects_the_loop_policy( - pytester: Pytester, -): - pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") - pytester.makepyfile( - dedent("""\ - import asyncio - import pytest - - pytestmark = pytest.mark.asyncio - - class CustomEventLoopPolicy(asyncio.DefaultEventLoopPolicy): - pass - - @pytest.fixture(scope="function") - def event_loop_policy(): - return CustomEventLoopPolicy() - - async def test_uses_custom_event_loop_policy(): - assert isinstance( - asyncio.get_event_loop_policy(), - CustomEventLoopPolicy, - ) - """), - ) - pytest_args = ["--asyncio-mode=strict"] - if sys.version_info >= (3, 14): - pytest_args.extend(["-W", "default"]) - result = pytester.runpytest(*pytest_args) - if sys.version_info >= (3, 14): - result.assert_outcomes(passed=1, warnings=3) - result.stdout.fnmatch_lines("*DefaultEventLoopPolicy*") - else: - result.assert_outcomes(passed=1) - - -def test_asyncio_mark_respects_parametrized_loop_policies( - pytester: Pytester, -): - pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") - pytester.makepyfile(dedent("""\ - import asyncio - - import pytest - - pytestmark = pytest.mark.asyncio - - class CustomEventLoopPolicy(asyncio.DefaultEventLoopPolicy): - pass - - @pytest.fixture( - scope="module", - params=[ - CustomEventLoopPolicy(), - CustomEventLoopPolicy(), - ], - ) - def event_loop_policy(request): - return request.param - - async def test_parametrized_loop(): - assert isinstance( - asyncio.get_event_loop_policy(), - CustomEventLoopPolicy, - ) - """)) - pytest_args = ["--asyncio-mode=strict"] - if sys.version_info >= (3, 14): - pytest_args.extend(["-W", "default"]) - result = pytester.runpytest(*pytest_args) - if sys.version_info >= (3, 14): - result.assert_outcomes(passed=2, warnings=5) - result.stdout.fnmatch_lines("*DefaultEventLoopPolicy*") - else: - result.assert_outcomes(passed=2) - - -def test_event_loop_policy_fixture_override_emits_deprecation_warning( - pytester: Pytester, -): - pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") - pytester.makepyfile( - dedent("""\ - import asyncio - import pytest - - pytestmark = pytest.mark.asyncio - - @pytest.fixture - def event_loop_policy(): - return asyncio.DefaultEventLoopPolicy() - - async def test_anything(): - pass - """), - ) - result = pytester.runpytest("--asyncio-mode=strict", "-W", "default") - result.assert_outcomes(passed=1) + result = pytester.runpytest("--asyncio-mode=strict") + result.assert_outcomes(errors=1) result.stdout.fnmatch_lines( - "*PytestDeprecationWarning*event_loop_policy*deprecated*" + "*mark.asyncio accepts only keyword arguments*'loop_scope'*'loop_factories'*" ) -def test_default_event_loop_policy_fixture_does_not_warn( - pytester: Pytester, -): - pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") - pytester.makepyfile( - dedent("""\ - import pytest - - pytestmark = pytest.mark.asyncio - - async def test_anything(): - pass - """), - ) - result = pytester.runpytest("--asyncio-mode=strict", "-W", "default") - result.assert_outcomes(passed=1) - result.stdout.no_fnmatch_line("*PytestDeprecationWarning*event_loop_policy*") - - def test_asyncio_mark_provides_function_scoped_loop_to_fixtures( pytester: Pytester, ): diff --git a/tests/markers/test_module_scope.py b/tests/markers/test_module_scope.py index a2662812..f4671dd2 100644 --- a/tests/markers/test_module_scope.py +++ b/tests/markers/test_module_scope.py @@ -1,6 +1,5 @@ from __future__ import annotations -import sys from textwrap import dedent from pytest import Pytester @@ -33,97 +32,6 @@ async def test_this_runs_in_same_loop(self): result.assert_outcomes(passed=3) -def test_asyncio_mark_respects_the_loop_policy( - pytester: Pytester, -): - pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") - pytester.makepyfile( - __init__="", - 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(loop_scope="module") - - @pytest.fixture(scope="module") - def event_loop_policy(): - return CustomEventLoopPolicy() - - async def test_uses_custom_event_loop_policy(): - assert isinstance( - asyncio.get_event_loop_policy(), - CustomEventLoopPolicy, - ) - """), - test_does_not_use_custom_policy=dedent("""\ - import asyncio - import pytest - - from .custom_policy import CustomEventLoopPolicy - - pytestmark = pytest.mark.asyncio(loop_scope="module") - - async def test_does_not_use_custom_event_loop_policy(): - assert not isinstance( - asyncio.get_event_loop_policy(), - CustomEventLoopPolicy, - ) - """), - ) - pytest_args = ["--asyncio-mode=strict"] - if sys.version_info >= (3, 14): - pytest_args.extend(["-W", "default"]) - result = pytester.runpytest(*pytest_args) - if sys.version_info >= (3, 14): - result.assert_outcomes(passed=2, warnings=4) - result.stdout.fnmatch_lines("*DefaultEventLoopPolicy*") - else: - result.assert_outcomes(passed=2) - - -def test_asyncio_mark_respects_parametrized_loop_policies( - pytester: Pytester, -): - pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") - pytester.makepyfile(dedent("""\ - import asyncio - - import pytest - - pytestmark = pytest.mark.asyncio(loop_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 - """)) - pytest_args = ["--asyncio-mode=strict"] - if sys.version_info >= (3, 14): - pytest_args.extend(["-W", "default"]) - result = pytester.runpytest(*pytest_args) - if sys.version_info >= (3, 14): - result.assert_outcomes(passed=2, warnings=4) - result.stdout.fnmatch_lines("*DefaultEventLoopPolicy*") - else: - result.assert_outcomes(passed=2) - - def test_asyncio_mark_provides_module_scoped_loop_to_fixtures( pytester: Pytester, ): @@ -177,8 +85,9 @@ async def test_runs_in_different_loop_as_fixture(self, async_fixture): """), ) - result = pytester.runpytest("--asyncio-mode=strict") + result = pytester.runpytest("--asyncio-mode=strict", "-W", "default") result.assert_outcomes(passed=1) + result.stdout.fnmatch_lines("*PytestAsyncioLoopScopeMismatchWarning*") def test_asyncio_mark_allows_combining_module_scoped_fixture_with_function_scoped_test( @@ -206,8 +115,9 @@ async def test_runs_in_different_loop_as_fixture(async_fixture): assert asyncio.get_running_loop() is not loop """), ) - result = pytester.runpytest("--asyncio-mode=strict") + result = pytester.runpytest("--asyncio-mode=strict", "-W", "default") result.assert_outcomes(passed=1) + result.stdout.fnmatch_lines("*PytestAsyncioLoopScopeMismatchWarning*") def test_allows_combining_module_scoped_asyncgen_fixture_with_function_scoped_test( @@ -235,8 +145,9 @@ async def test_runs_in_different_loop_as_fixture(async_fixture): assert asyncio.get_running_loop() is not loop """), ) - result = pytester.runpytest("--asyncio-mode=strict") + result = pytester.runpytest("--asyncio-mode=strict", "-W", "default") result.assert_outcomes(passed=1) + result.stdout.fnmatch_lines("*PytestAsyncioLoopScopeMismatchWarning*") def test_asyncio_mark_handles_missing_event_loop_triggered_by_fixture( diff --git a/tests/markers/test_package_scope.py b/tests/markers/test_package_scope.py index ff6aba4a..bebe4df5 100644 --- a/tests/markers/test_package_scope.py +++ b/tests/markers/test_package_scope.py @@ -1,6 +1,5 @@ from __future__ import annotations -import sys from textwrap import dedent from pytest import Pytester @@ -60,105 +59,6 @@ async def test_subpackage_runs_in_different_loop(): result.assert_outcomes(passed=4) -def test_asyncio_mark_respects_the_loop_policy( - pytester: Pytester, -): - pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") - 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(loop_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(loop_scope="package") - - async def test_also_uses_custom_event_loop_policy(): - assert isinstance( - asyncio.get_event_loop_policy(), - CustomEventLoopPolicy, - ) - """), - ) - pytest_args = ["--asyncio-mode=strict"] - if sys.version_info >= (3, 14): - pytest_args.extend(["-W", "default"]) - result = pytester.runpytest(*pytest_args) - if sys.version_info >= (3, 14): - result.assert_outcomes(passed=2, warnings=4) - result.stdout.fnmatch_lines("*DefaultEventLoopPolicy*") - else: - result.assert_outcomes(passed=2) - - -def test_asyncio_mark_respects_parametrized_loop_policies( - pytester: Pytester, -): - pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") - pytester.makepyfile( - __init__="", - test_parametrization=dedent("""\ - import asyncio - - import pytest - - pytestmark = pytest.mark.asyncio(loop_scope="package") - - @pytest.fixture( - scope="package", - params=[ - asyncio.DefaultEventLoopPolicy(), - asyncio.DefaultEventLoopPolicy(), - ], - ) - def event_loop_policy(request): - return request.param - - async def test_parametrized_loop(): - pass - """), - ) - pytest_args = ["--asyncio-mode=strict"] - if sys.version_info >= (3, 14): - pytest_args.extend(["-W", "default"]) - result = pytester.runpytest(*pytest_args) - if sys.version_info >= (3, 14): - result.assert_outcomes(passed=2, warnings=4) - result.stdout.fnmatch_lines("*DefaultEventLoopPolicy*") - else: - result.assert_outcomes(passed=2) - - def test_asyncio_mark_provides_package_scoped_loop_to_fixtures( pytester: Pytester, ): @@ -225,8 +125,9 @@ async def test_runs_in_different_loop_as_fixture(async_fixture): assert asyncio.get_running_loop() is not loop """), ) - result = pytester.runpytest("--asyncio-mode=strict") + result = pytester.runpytest("--asyncio-mode=strict", "-W", "default") result.assert_outcomes(passed=1) + result.stdout.fnmatch_lines("*PytestAsyncioLoopScopeMismatchWarning*") def test_asyncio_mark_allows_combining_package_scoped_fixture_with_class_scoped_test( @@ -255,8 +156,9 @@ async def test_runs_in_different_loop_as_fixture(self, async_fixture): assert asyncio.get_running_loop() is not loop """), ) - result = pytester.runpytest("--asyncio-mode=strict") + result = pytester.runpytest("--asyncio-mode=strict", "-W", "default") result.assert_outcomes(passed=1) + result.stdout.fnmatch_lines("*PytestAsyncioLoopScopeMismatchWarning*") def test_asyncio_mark_allows_combining_package_scoped_fixture_with_function_scoped_test( @@ -284,8 +186,9 @@ async def test_runs_in_different_loop_as_fixture(async_fixture): assert asyncio.get_running_loop() is not loop """), ) - result = pytester.runpytest("--asyncio-mode=strict") + result = pytester.runpytest("--asyncio-mode=strict", "-W", "default") result.assert_outcomes(passed=1) + result.stdout.fnmatch_lines("*PytestAsyncioLoopScopeMismatchWarning*") def test_asyncio_mark_handles_missing_event_loop_triggered_by_fixture( diff --git a/tests/markers/test_session_scope.py b/tests/markers/test_session_scope.py index 24566c5e..f2b5efb0 100644 --- a/tests/markers/test_session_scope.py +++ b/tests/markers/test_session_scope.py @@ -1,6 +1,5 @@ from __future__ import annotations -import sys from textwrap import dedent from pytest import Pytester @@ -61,105 +60,6 @@ async def test_subpackage_runs_in_same_loop(): result.assert_outcomes(passed=4) -def test_asyncio_mark_respects_the_loop_policy( - pytester: Pytester, -): - pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") - pytester.makepyfile( - __init__="", - conftest=dedent("""\ - import pytest - - from .custom_policy import CustomEventLoopPolicy - - @pytest.fixture(scope="session") - 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(loop_scope="session") - - 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(loop_scope="session") - - async def test_also_uses_custom_event_loop_policy(): - assert isinstance( - asyncio.get_event_loop_policy(), - CustomEventLoopPolicy, - ) - """), - ) - pytest_args = ["--asyncio-mode=strict"] - if sys.version_info >= (3, 14): - pytest_args.extend(["-W", "default"]) - result = pytester.runpytest(*pytest_args) - if sys.version_info >= (3, 14): - result.assert_outcomes(passed=2, warnings=4) - result.stdout.fnmatch_lines("*DefaultEventLoopPolicy*") - else: - result.assert_outcomes(passed=2) - - -def test_asyncio_mark_respects_parametrized_loop_policies( - pytester: Pytester, -): - pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") - pytester.makepyfile( - __init__="", - test_parametrization=dedent("""\ - import asyncio - - import pytest - - pytestmark = pytest.mark.asyncio(loop_scope="session") - - @pytest.fixture( - scope="session", - params=[ - asyncio.DefaultEventLoopPolicy(), - asyncio.DefaultEventLoopPolicy(), - ], - ) - def event_loop_policy(request): - return request.param - - async def test_parametrized_loop(): - pass - """), - ) - pytest_args = ["--asyncio-mode=strict"] - if sys.version_info >= (3, 14): - pytest_args.extend(["-W", "default"]) - result = pytester.runpytest(*pytest_args) - if sys.version_info >= (3, 14): - result.assert_outcomes(passed=2, warnings=4) - result.stdout.fnmatch_lines("*DefaultEventLoopPolicy*") - else: - result.assert_outcomes(passed=2) - - def test_asyncio_mark_provides_session_scoped_loop_to_fixtures( pytester: Pytester, ): @@ -228,8 +128,9 @@ async def test_runs_in_different_loop_as_fixture(async_fixture): assert asyncio.get_running_loop() is not loop """), ) - result = pytester.runpytest("--asyncio-mode=strict") + result = pytester.runpytest("--asyncio-mode=strict", "-W", "default") result.assert_outcomes(passed=1) + result.stdout.fnmatch_lines("*PytestAsyncioLoopScopeMismatchWarning*") def test_asyncio_mark_allows_combining_session_scoped_fixture_with_module_scoped_test( @@ -257,8 +158,9 @@ async def test_runs_in_different_loop_as_fixture(async_fixture): assert asyncio.get_running_loop() is not loop """), ) - result = pytester.runpytest("--asyncio-mode=strict") + result = pytester.runpytest("--asyncio-mode=strict", "-W", "default") result.assert_outcomes(passed=1) + result.stdout.fnmatch_lines("*PytestAsyncioLoopScopeMismatchWarning*") def test_asyncio_mark_allows_combining_session_scoped_fixture_with_class_scoped_test( @@ -287,8 +189,9 @@ async def test_runs_in_different_loop_as_fixture(self, async_fixture): assert asyncio.get_running_loop() is not loop """), ) - result = pytester.runpytest("--asyncio-mode=strict") + result = pytester.runpytest("--asyncio-mode=strict", "-W", "default") result.assert_outcomes(passed=1) + result.stdout.fnmatch_lines("*PytestAsyncioLoopScopeMismatchWarning*") def test_asyncio_mark_allows_combining_session_scoped_fixture_with_function_scoped_test( @@ -316,8 +219,9 @@ async def test_runs_in_different_loop_as_fixture(async_fixture): assert asyncio.get_running_loop() is not loop """), ) - result = pytester.runpytest("--asyncio-mode=strict") + result = pytester.runpytest("--asyncio-mode=strict", "-W", "default") result.assert_outcomes(passed=1) + result.stdout.fnmatch_lines("*PytestAsyncioLoopScopeMismatchWarning*") def test_allows_combining_session_scoped_asyncgen_fixture_with_function_scoped_test( @@ -346,8 +250,9 @@ async def test_runs_in_different_loop_as_fixture(async_fixture): assert asyncio.get_running_loop() is not loop """), ) - result = pytester.runpytest("--asyncio-mode=strict") + result = pytester.runpytest("--asyncio-mode=strict", "-W", "default") result.assert_outcomes(passed=1) + result.stdout.fnmatch_lines("*PytestAsyncioLoopScopeMismatchWarning*") def test_asyncio_mark_handles_missing_event_loop_triggered_by_fixture( diff --git a/tests/modes/test_strict_mode.py b/tests/modes/test_strict_mode.py index 86cc34ad..b369424c 100644 --- a/tests/modes/test_strict_mode.py +++ b/tests/modes/test_strict_mode.py @@ -149,7 +149,7 @@ async def test_anything(any_fixture): "test_anything" ), ( - "*/pytest_asyncio/plugin.py:*: PytestDeprecationWarning: " + "*/pytest_asyncio/_dispatch.py:*: PytestDeprecationWarning: " "asyncio test 'test_anything' requested async " "@pytest.fixture 'any_fixture' in strict mode. " "You might want to use @pytest_asyncio.fixture or switch to " @@ -196,7 +196,7 @@ async def test_anything(any_fixture): "test_anything" ), ( - "*/pytest_asyncio/plugin.py:*: PytestDeprecationWarning: " + "*/pytest_asyncio/_dispatch.py:*: PytestDeprecationWarning: " "*asyncio test 'test_anything' requested async " "@pytest.fixture 'any_fixture' in strict mode. " "You might want to use @pytest_asyncio.fixture or switch to " diff --git a/tests/test_event_loop_policy_removed.py b/tests/test_event_loop_policy_removed.py new file mode 100644 index 00000000..a0b5a979 --- /dev/null +++ b/tests/test_event_loop_policy_removed.py @@ -0,0 +1,87 @@ +from __future__ import annotations + +from textwrap import dedent + +from pytest import Pytester + + +def test_event_loop_policy_fixture_in_conftest_raises_usage_error(pytester: Pytester): + pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") + pytester.makeconftest(dedent("""\ + import asyncio + import pytest + + @pytest.fixture + def event_loop_policy(): + return asyncio.DefaultEventLoopPolicy() + """)) + pytester.makepyfile(dedent("""\ + import pytest + + @pytest.mark.asyncio + async def test_anything(): + pass + """)) + result = pytester.runpytest("--asyncio-mode=strict") + result.stderr.no_fnmatch_line("INTERNALERROR> *") + result.assert_outcomes(passed=0, failed=0, errors=0) + assert result.ret != 0 + result.stderr.fnmatch_lines('*"event_loop_policy" fixture was removed*') + + +def test_event_loop_policy_fixture_in_test_module_raises_usage_error( + pytester: Pytester, +): + pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") + pytester.makepyfile(dedent("""\ + import asyncio + import pytest + + @pytest.fixture + def event_loop_policy(): + return asyncio.DefaultEventLoopPolicy() + + @pytest.mark.asyncio + async def test_anything(): + pass + """)) + result = pytester.runpytest("--asyncio-mode=strict") + result.stderr.no_fnmatch_line("INTERNALERROR> *") + assert result.ret != 0 + result.stderr.fnmatch_lines('*"event_loop_policy" fixture was removed*') + + +def test_event_loop_policy_fixture_message_mentions_loop_factories_hook( + pytester: Pytester, +): + pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") + pytester.makeconftest(dedent("""\ + import asyncio + import pytest + + @pytest.fixture + def event_loop_policy(): + return asyncio.DefaultEventLoopPolicy() + """)) + pytester.makepyfile(dedent("""\ + import pytest + + @pytest.mark.asyncio + async def test_anything(): + pass + """)) + result = pytester.runpytest("--asyncio-mode=strict") + result.stderr.fnmatch_lines("*pytest_asyncio_loop_factories*") + + +def test_suite_without_event_loop_policy_fixture_runs_fine(pytester: Pytester): + pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") + pytester.makepyfile(dedent("""\ + import pytest + + @pytest.mark.asyncio + async def test_anything(): + pass + """)) + result = pytester.runpytest("--asyncio-mode=strict") + result.assert_outcomes(passed=1) diff --git a/tests/test_loop_factory_parametrization.py b/tests/test_loop_factory_parametrization.py index 224c9141..19589c46 100644 --- a/tests/test_loop_factory_parametrization.py +++ b/tests/test_loop_factory_parametrization.py @@ -783,8 +783,9 @@ def sync_fixture_captures_loop(): async def test_sync_fixture_and_test_see_same_loop(sync_fixture_captures_loop): assert sync_fixture_captures_loop == id(asyncio.get_running_loop()) """)) - result = pytester.runpytest("--asyncio-mode=strict") + result = pytester.runpytest("--asyncio-mode=strict", "-W", "default") result.assert_outcomes(passed=1) + result.stdout.fnmatch_lines("*PytestAsyncioLoopScopeMismatchWarning*") @pytest.mark.parametrize( @@ -840,8 +841,9 @@ def sync_generator_fixture(): async def test_generator_fixture_sees_correct_loop(sync_generator_fixture): assert sync_generator_fixture == id(asyncio.get_running_loop()) """)) - result = pytester.runpytest("--asyncio-mode=strict") + result = pytester.runpytest("--asyncio-mode=strict", "-W", "default") result.assert_outcomes(passed=1) + result.stdout.fnmatch_lines("*PytestAsyncioLoopScopeMismatchWarning*") @pytest.mark.parametrize("loop_scope", ("module", "package", "session")) diff --git a/tests/test_loop_scope_mismatch_warning.py b/tests/test_loop_scope_mismatch_warning.py new file mode 100644 index 00000000..6e637776 --- /dev/null +++ b/tests/test_loop_scope_mismatch_warning.py @@ -0,0 +1,181 @@ +from __future__ import annotations + +from textwrap import dedent + +from pytest import Pytester + + +def test_warns_when_fixture_loop_scope_is_wider_than_test(pytester: Pytester): + pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") + pytester.makepyfile(dedent("""\ + import pytest + import pytest_asyncio + + pytest_plugins = "pytest_asyncio" + + @pytest_asyncio.fixture(loop_scope="module") + async def wide_fixture(): + return 1 + + @pytest.mark.asyncio(loop_scope="function") + async def test_narrow(wide_fixture): + assert wide_fixture == 1 + """)) + result = pytester.runpytest("--asyncio-mode=strict", "-W", "default") + result.assert_outcomes(passed=1) + result.stdout.fnmatch_lines( + "*PytestAsyncioLoopScopeMismatchWarning*'function'*'wide_fixture'*'module'*" + ) + + +def test_warns_when_fixture_loop_scope_is_narrower_than_test(pytester: Pytester): + pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") + pytester.makepyfile(dedent("""\ + import pytest + import pytest_asyncio + + pytest_plugins = "pytest_asyncio" + + @pytest_asyncio.fixture(loop_scope="function") + async def narrow_fixture(): + return 1 + + @pytest.mark.asyncio(loop_scope="session") + async def test_wide(narrow_fixture): + assert narrow_fixture == 1 + """)) + result = pytester.runpytest("--asyncio-mode=strict", "-W", "default") + result.assert_outcomes(passed=1) + result.stdout.fnmatch_lines( + "*PytestAsyncioLoopScopeMismatchWarning*'session'*'narrow_fixture'*'function'*" + ) + + +def test_warns_for_mismatch_reachable_only_transitively(pytester: Pytester): + pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") + pytester.makepyfile(dedent("""\ + import pytest + import pytest_asyncio + + pytest_plugins = "pytest_asyncio" + + @pytest_asyncio.fixture(loop_scope="module") + async def async_fixture(): + return 1 + + @pytest.fixture + def sync_fixture(async_fixture): + return async_fixture + + @pytest.mark.asyncio(loop_scope="function") + async def test_uses_sync_fixture(sync_fixture): + assert sync_fixture == 1 + """)) + result = pytester.runpytest("--asyncio-mode=strict", "-W", "default") + result.assert_outcomes(passed=1) + result.stdout.fnmatch_lines( + "*PytestAsyncioLoopScopeMismatchWarning*'async_fixture'*" + ) + + +def test_diamond_shaped_dependency_warns_only_once(pytester: Pytester): + pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") + pytester.makepyfile(dedent("""\ + import pytest + import pytest_asyncio + + pytest_plugins = "pytest_asyncio" + + @pytest_asyncio.fixture(loop_scope="module") + async def shared_fixture(): + return 1 + + @pytest.fixture + def fixture_a(shared_fixture): + return shared_fixture + + @pytest.fixture + def fixture_b(shared_fixture): + return shared_fixture + + @pytest.mark.asyncio(loop_scope="function") + async def test_uses_both(fixture_a, fixture_b): + assert fixture_a == fixture_b == 1 + """)) + result = pytester.runpytest("--asyncio-mode=strict", "-W", "default") + result.assert_outcomes(passed=1) + warning_lines = [ + line + for line in result.stdout.lines + if "PytestAsyncioLoopScopeMismatchWarning" in line and "shared_fixture" in line + ] + assert len(warning_lines) == 1 + + +def test_no_warning_when_loop_scope_matches_despite_differing_cache_scope( + pytester: Pytester, +): + pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") + pytester.makepyfile(dedent("""\ + import pytest + import pytest_asyncio + + pytest_plugins = "pytest_asyncio" + + @pytest_asyncio.fixture(scope="module", loop_scope="session") + async def fixture_with_matching_loop_scope(): + return 1 + + @pytest.mark.asyncio(loop_scope="session") + async def test_matches(fixture_with_matching_loop_scope): + assert fixture_with_matching_loop_scope == 1 + """)) + result = pytester.runpytest("--asyncio-mode=strict", "-W", "default") + result.assert_outcomes(passed=1) + result.stdout.no_fnmatch_line("*PytestAsyncioLoopScopeMismatchWarning*") + + +def test_no_warning_for_sync_fixture_with_differing_cache_scope(pytester: Pytester): + pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") + pytester.makepyfile(dedent("""\ + import pytest + + pytest_plugins = "pytest_asyncio" + + @pytest.fixture(scope="module") + def plain_sync_fixture(): + return 1 + + @pytest.mark.asyncio(loop_scope="function") + async def test_uses_plain_sync_fixture(plain_sync_fixture): + assert plain_sync_fixture == 1 + """)) + result = pytester.runpytest("--asyncio-mode=strict", "-W", "default") + result.assert_outcomes(passed=1) + result.stdout.no_fnmatch_line("*PytestAsyncioLoopScopeMismatchWarning*") + + +def test_mismatch_warning_can_be_silenced_via_filterwarnings_marker( + pytester: Pytester, +): + pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") + pytester.makepyfile(dedent("""\ + import pytest + import pytest_asyncio + + pytest_plugins = "pytest_asyncio" + + @pytest_asyncio.fixture(loop_scope="module") + async def wide_fixture(): + return 1 + + @pytest.mark.asyncio(loop_scope="function") + @pytest.mark.filterwarnings( + "ignore::pytest_asyncio.PytestAsyncioLoopScopeMismatchWarning" + ) + async def test_narrow(wide_fixture): + assert wide_fixture == 1 + """)) + result = pytester.runpytest("--asyncio-mode=strict", "-W", "default") + result.assert_outcomes(passed=1) + result.stdout.no_fnmatch_line("*PytestAsyncioLoopScopeMismatchWarning*")