diff --git a/src/pytest_bdd/scenario.py b/src/pytest_bdd/scenario.py index 531190617..16c999db7 100644 --- a/src/pytest_bdd/scenario.py +++ b/src/pytest_bdd/scenario.py @@ -14,16 +14,20 @@ from __future__ import annotations import contextlib +import contextvars +import inspect import logging import os import re +import sys from collections.abc import Iterable, Iterator from inspect import signature -from typing import TYPE_CHECKING, Callable, TypeVar, cast +from typing import TYPE_CHECKING, Callable, Literal, TypeVar, cast from weakref import WeakKeyDictionary import pytest from _pytest.fixtures import FixtureDef, FixtureManager, FixtureRequest, call_fixture_func +from _pytest.mark.structures import Mark from . import exceptions from .compat import getfixturedefs, inject_fixture @@ -239,13 +243,22 @@ def _execute_step_function( arg: request.getfixturevalue(arg) for arg in get_required_args(context.step_func) if arg not in kwargs } + _resolve_async_arguments(request=request, context=context, kwargs=kwargs) + kw["step_func_args"] = kwargs request.config.hook.pytest_bdd_before_step_call(**kw) - # Execute the step as if it was a pytest fixture using `call_fixture_func`, - # so that we can allow "yield" statements in it - return_value = call_fixture_func(fixturefunc=context.step_func, request=request, kwargs=kwargs) + if context.is_async or context.is_async_gen: + return_value = _execute_async_step(request=request, context=context, kwargs=kwargs) + else: + # Execute the step as if it was a pytest fixture using `call_fixture_func`, + # so that we can allow "yield" statements in it + return_value = call_fixture_func(fixturefunc=context.step_func, request=request, kwargs=kwargs) + if inspect.isawaitable(return_value): + return_value = _await_async_result(request=request, context=context, awaitable=return_value) + elif inspect.isasyncgen(return_value): + return_value = _consume_async_generator_result(request=request, context=context, async_gen=return_value) except Exception as exception: request.config.hook.pytest_bdd_step_error(exception=exception, **kw) @@ -257,6 +270,416 @@ def _execute_step_function( request.config.hook.pytest_bdd_after_step(**kw) +def _execute_async_step(request: FixtureRequest, context: StepFunctionContext, kwargs: dict[str, object]) -> object: + backend, marker = _resolve_async_backend_preference(request=request, step_func=context.step_func) + pluginmanager = request.config.pluginmanager + errors: list[str] = [] + + if backend in (None, "asyncio"): + if _has_pytest_asyncio_plugin(pluginmanager): + try: + return _run_async_step_with_pytest_asyncio( + request=request, context=context, kwargs=kwargs, asyncio_marker=marker if backend else None + ) + except exceptions.StepImplementationError as exc: + errors.append(str(exc)) + elif backend == "asyncio": + errors.append( + "Async step is marked with pytest.mark.asyncio but pytest-asyncio is not installed or not active." + ) + + if backend in (None, "anyio"): + if pluginmanager.has_plugin("anyio"): + try: + return _run_async_step_with_anyio(request=request, context=context, kwargs=kwargs) + except exceptions.StepImplementationError as exc: + errors.append(str(exc)) + elif backend == "anyio": + errors.append( + "Async step is marked with pytest.mark.anyio but the anyio pytest plugin is not installed or not active." + ) + + if not errors: + message = "Async step functions require pytest-asyncio or the anyio pytest plugin to be installed and enabled." + else: + # Deduplicate messages while preserving order + message = "\n".join(dict.fromkeys(errors)) + + raise exceptions.StepImplementationError(message) + + +def _await_async_result(request: FixtureRequest, context: StepFunctionContext, awaitable: object) -> object: + backend, marker = _resolve_async_backend_preference(request=request, step_func=context.step_func) + pluginmanager = request.config.pluginmanager + errors: list[str] = [] + + if backend in (None, "asyncio") and _has_pytest_asyncio_plugin(pluginmanager): + try: + return _await_with_pytest_asyncio( + request=request, + step_func=context.step_func, + awaitable=awaitable, + asyncio_marker=marker if backend else None, + ) + except exceptions.StepImplementationError as exc: + errors.append(str(exc)) + elif backend == "asyncio": + errors.append("Awaiting async result requires pytest-asyncio, which is not installed or active.") + + if backend in (None, "anyio") and pluginmanager.has_plugin("anyio"): + try: + return _await_with_anyio(request=request, awaitable=awaitable) + except exceptions.StepImplementationError as exc: + errors.append(str(exc)) + elif backend == "anyio": + errors.append("Awaiting async result requires the anyio pytest plugin, which is not installed or active.") + + if not errors: + message = "Async step result requires pytest-asyncio or the anyio pytest plugin to be installed and enabled." + else: + message = "\n".join(dict.fromkeys(errors)) + + raise exceptions.StepImplementationError(message) + + +def _resolve_async_arguments( + *, request: FixtureRequest, context: StepFunctionContext, kwargs: dict[str, object] +) -> None: + for name, value in list(kwargs.items()): + if inspect.isawaitable(value): + resolved = _await_async_result(request=request, context=context, awaitable=value) + kwargs[name] = resolved + inject_fixture(request, name, resolved) + elif inspect.isasyncgen(value): + resolved = _consume_async_generator_result(request=request, context=context, async_gen=value) + kwargs[name] = resolved + inject_fixture(request, name, resolved) + + +def _resolve_async_backend_preference( + request: FixtureRequest, step_func: Callable[..., object] +) -> tuple[Literal["asyncio", "anyio"] | None, Mark | None]: + step_marks = _collect_callable_marks(step_func) + step_anyio = _find_mark(step_marks, "anyio") + step_asyncio = _find_mark(step_marks, "asyncio") + + node_anyio = _first_node_marker(request, "anyio") + node_asyncio = _first_node_marker(request, "asyncio") + + candidates: list[tuple[Literal["asyncio", "anyio"], Mark]] = [] + if step_anyio is not None: + candidates.append(("anyio", step_anyio)) + if step_asyncio is not None: + candidates.append(("asyncio", step_asyncio)) + if node_anyio is not None: + candidates.append(("anyio", node_anyio)) + if node_asyncio is not None: + candidates.append(("asyncio", node_asyncio)) + + if not candidates: + return None, None + + backend, marker = candidates[0] + for candidate_backend, candidate_marker in candidates[1:]: + if candidate_backend != backend: + raise exceptions.StepImplementationError( + "Async step has conflicting pytest markers: both 'asyncio' and 'anyio' are present." + ) + if marker is None and candidate_marker is not None: + marker = candidate_marker + + return backend, marker + + +def _collect_callable_marks(step_func: Callable[..., object]) -> list[Mark]: + marks = getattr(step_func, "pytestmark", []) + if isinstance(marks, Mark): + return [marks] + if not isinstance(marks, (list, tuple)): + return [] + return [mark for mark in marks if isinstance(mark, Mark)] + + +def _find_mark(marks: Iterable[Mark], name: str) -> Mark | None: + for mark in marks: + if mark.name == name: + return mark + return None + + +def _first_node_marker(request: FixtureRequest, name: str) -> Mark | None: + try: + return next(request.node.iter_markers(name)) + except StopIteration: + return None + + +def _has_pytest_asyncio_plugin(pluginmanager: pytest.PytestPluginManager) -> bool: + return any( + pluginmanager.has_plugin(alias) + for alias in ("asyncio", "pytest_asyncio", "pytest-asyncio", "pytest_asyncio.plugin") + ) + + +def _run_async_step_with_pytest_asyncio( + *, + request: FixtureRequest, + context: StepFunctionContext, + kwargs: dict[str, object], + asyncio_marker: Mark | None, +) -> object: + try: + from pytest_asyncio.plugin import _wrap_async_fixture, _wrap_asyncgen_fixture + except ImportError as exc: # pragma: no cover - defensive guard when plugin missing at runtime + raise exceptions.StepImplementationError("pytest-asyncio is not importable.") from exc + + loop_scope = _determine_asyncio_loop_scope(request=request, step_func=context.step_func, marker=asyncio_marker) + runner_fixture_name = f"_{loop_scope}_scoped_runner" + + try: + runner = request.getfixturevalue(runner_fixture_name) + except pytest.FixtureLookupError as exc: + raise exceptions.StepImplementationError( + f"Unable to obtain '{runner_fixture_name}' fixture provided by pytest-asyncio." + ) from exc + + fixture_function = context.step_func + + if context.is_async_gen: + wrapped = _wrap_asyncgen_fixture(fixture_function, runner, request) + else: + wrapped = _wrap_async_fixture(fixture_function, runner, request) + + return wrapped(**kwargs) + + +def _determine_asyncio_loop_scope( + *, request: FixtureRequest, step_func: Callable[..., object], marker: Mark | None +) -> str: + from pytest_asyncio.plugin import _get_marked_loop_scope + + default_scope = request.config.getini("asyncio_default_fixture_loop_scope") + if not default_scope: + default_scope = "function" + + scope = getattr(step_func, "_loop_scope", None) + + if marker is not None: + scope = _get_marked_loop_scope(marker, default_scope) + + if scope is None: + scope = default_scope + + assert scope in {"function", "class", "module", "package", "session"} + return scope + + +def _run_async_step_with_anyio( + *, request: FixtureRequest, context: StepFunctionContext, kwargs: dict[str, object] +) -> object: + try: + import anyio.pytest_plugin as anyio_pytest_plugin + except ImportError as exc: # pragma: no cover - defensive guard when plugin missing at runtime + raise exceptions.StepImplementationError("anyio is not importable.") from exc + + try: + backend_value = request.getfixturevalue("anyio_backend") + except pytest.FixtureLookupError as exc: + raise exceptions.StepImplementationError( + "Async step requested anyio backend but the 'anyio_backend' fixture is not available." + ) from exc + + backend_name, backend_options = anyio_pytest_plugin.extract_backend_and_options(backend_value) + + func_parameters = signature(context.step_func).parameters + if "anyio_backend" in func_parameters and "anyio_backend" not in kwargs: + kwargs["anyio_backend"] = backend_value + if "request" in func_parameters and "request" not in kwargs: + kwargs["request"] = request + + with anyio_pytest_plugin.get_runner(backend_name, backend_options) as runner: + if context.is_async_gen: + iterator = runner.run_asyncgen_fixture(context.step_func, kwargs) + try: + result = next(iterator) + except StopIteration as exc: + raise ValueError("Async step function did not yield a value") from exc + + def finalizer() -> None: + with contextlib.suppress(StopIteration): + next(iterator) + + request.addfinalizer(finalizer) + return result + + return runner.run_fixture(context.step_func, kwargs) + + +def _await_with_pytest_asyncio( + *, + request: FixtureRequest, + step_func: Callable[..., object], + awaitable: object, + asyncio_marker: Mark | None, +) -> object: + try: + from pytest_asyncio.plugin import _apply_contextvar_changes + except ImportError as exc: # pragma: no cover - defensive guard + raise exceptions.StepImplementationError("pytest-asyncio is not importable.") from exc + + loop_scope = _determine_asyncio_loop_scope(request=request, step_func=step_func, marker=asyncio_marker) + runner_fixture_name = f"_{loop_scope}_scoped_runner" + + try: + runner = request.getfixturevalue(runner_fixture_name) + except pytest.FixtureLookupError as exc: + raise exceptions.StepImplementationError( + f"Unable to obtain '{runner_fixture_name}' fixture provided by pytest-asyncio." + ) from exc + + if not hasattr(runner, "run"): + raise exceptions.StepImplementationError("Unsupported pytest-asyncio runner implementation encountered.") + + ctx = contextvars.copy_context() + result = runner.run(awaitable, context=ctx) + reset = _apply_contextvar_changes(ctx) + if reset is not None: + request.addfinalizer(reset) + return result + + +def _await_with_anyio(*, request: FixtureRequest, awaitable: object) -> object: + try: + import anyio.pytest_plugin as anyio_pytest_plugin + except ImportError as exc: # pragma: no cover - defensive guard + raise exceptions.StepImplementationError("anyio is not importable.") from exc + + try: + backend_value = request.getfixturevalue("anyio_backend") + except pytest.FixtureLookupError as exc: + raise exceptions.StepImplementationError( + "Async step requested anyio backend but the 'anyio_backend' fixture is not available." + ) from exc + + backend_name, backend_options = anyio_pytest_plugin.extract_backend_and_options(backend_value) + + async def _consume() -> object: + return await awaitable # type: ignore[arg-type] + + with anyio_pytest_plugin.get_runner(backend_name, backend_options) as runner: + return runner.run_fixture(_consume, {}) + + +def _consume_async_generator_result(request: FixtureRequest, context: StepFunctionContext, async_gen: object) -> object: + backend, marker = _resolve_async_backend_preference(request=request, step_func=context.step_func) + pluginmanager = request.config.pluginmanager + errors: list[str] = [] + + if backend in (None, "asyncio") and _has_pytest_asyncio_plugin(pluginmanager): + try: + return _consume_async_gen_with_pytest_asyncio( + request=request, + context=context, + async_gen=async_gen, + asyncio_marker=marker if backend else None, + ) + except exceptions.StepImplementationError as exc: + errors.append(str(exc)) + elif backend == "asyncio": + errors.append("Async generator step requires pytest-asyncio, which is not installed or not active.") + + if backend in (None, "anyio") and pluginmanager.has_plugin("anyio"): + try: + return _consume_async_gen_with_anyio(request=request, async_gen=async_gen) + except exceptions.StepImplementationError as exc: + errors.append(str(exc)) + elif backend == "anyio": + errors.append("Async generator step requires the anyio pytest plugin, which is not installed or not active.") + + if not errors: + message = "Async generator steps require pytest-asyncio or the anyio pytest plugin to be installed and enabled." + else: + message = "\n".join(dict.fromkeys(errors)) + + raise exceptions.StepImplementationError(message) + + +def _consume_async_gen_with_pytest_asyncio( + *, + request: FixtureRequest, + context: StepFunctionContext, + async_gen: object, + asyncio_marker: Mark | None, +) -> object: + try: + from pytest_asyncio.plugin import _wrap_asyncgen_fixture + except ImportError as exc: # pragma: no cover - defensive guard + raise exceptions.StepImplementationError("pytest-asyncio is not importable.") from exc + + loop_scope = _determine_asyncio_loop_scope(request=request, step_func=context.step_func, marker=asyncio_marker) + runner_fixture_name = f"_{loop_scope}_scoped_runner" + + try: + runner = request.getfixturevalue(runner_fixture_name) + except pytest.FixtureLookupError as exc: + raise exceptions.StepImplementationError( + f"Unable to obtain '{runner_fixture_name}' fixture provided by pytest-asyncio." + ) from exc + + def _factory() -> object: + async def _proxy() -> object: + async for item in async_gen: # type: ignore[async-for] + yield item + + return _proxy() + + wrapped = _wrap_asyncgen_fixture(_factory, runner, request) + return wrapped() + + +def _consume_async_gen_with_anyio(*, request: FixtureRequest, async_gen: object) -> object: + try: + import anyio.pytest_plugin as anyio_pytest_plugin + except ImportError as exc: # pragma: no cover - defensive guard + raise exceptions.StepImplementationError("anyio is not importable.") from exc + + try: + backend_value = request.getfixturevalue("anyio_backend") + except pytest.FixtureLookupError as exc: + raise exceptions.StepImplementationError( + "Async generator step requested anyio backend but the 'anyio_backend' fixture is not available." + ) from exc + + backend_name, backend_options = anyio_pytest_plugin.extract_backend_and_options(backend_value) + + async def _proxy() -> object: + async for item in async_gen: # type: ignore[async-for] + yield item + + runner_cm = anyio_pytest_plugin.get_runner(backend_name, backend_options) + runner = runner_cm.__enter__() + + iterator = runner.run_asyncgen_fixture(lambda: _proxy(), {}) + try: + result = next(iterator) + except StopIteration as exc: + runner_cm.__exit__(None, None, None) + raise ValueError("Async generator step function did not yield a value") from exc + except Exception: + runner_cm.__exit__(*sys.exc_info()) + raise + + def finalizer() -> None: + try: + with contextlib.suppress(StopAsyncIteration, StopIteration): + next(iterator) + finally: + runner_cm.__exit__(None, None, None) + + request.addfinalizer(finalizer) + return result + + def _execute_scenario(feature: Feature, scenario: Scenario, request: FixtureRequest) -> None: """Execute the scenario. diff --git a/src/pytest_bdd/steps.py b/src/pytest_bdd/steps.py index 2fbfa0c3e..c6635b423 100644 --- a/src/pytest_bdd/steps.py +++ b/src/pytest_bdd/steps.py @@ -38,6 +38,7 @@ def _(article): from __future__ import annotations import enum +import inspect from collections.abc import Iterable from dataclasses import dataclass, field from itertools import count @@ -70,6 +71,8 @@ class StepFunctionContext: parser: StepParser converters: dict[str, Callable[[str], object]] = field(default_factory=dict) target_fixture: str | None = None + is_async: bool = False + is_async_gen: bool = False def get_step_fixture_name(step: Step) -> str: @@ -169,6 +172,8 @@ def decorator(func: Callable[P, T]) -> Callable[P, T]: parser=parser, converters=converters, target_fixture=target_fixture, + is_async=inspect.iscoroutinefunction(func), + is_async_gen=inspect.isasyncgenfunction(func), ) def step_function_marker() -> StepFunctionContext: diff --git a/tests/steps/test_async.py b/tests/steps/test_async.py new file mode 100644 index 000000000..238ace8bd --- /dev/null +++ b/tests/steps/test_async.py @@ -0,0 +1,579 @@ +"""Test async step support.""" + +from __future__ import annotations + +import textwrap + + +def test_async_given_in_sync_test(pytester, pytest_params): + """Test async given step in a synchronous test.""" + pytester.makefile( + ".feature", + async_steps=textwrap.dedent( + """\ + Feature: Async steps + Scenario: Async given in sync test + Given I have an async resource + Then the resource should be available + """ + ), + ) + pytester.makepyfile( + textwrap.dedent( + """\ + import asyncio + import pytest + from pytest_bdd import given, then, scenario + + pytestmark = pytest.mark.asyncio + + @scenario("async_steps.feature", "Async given in sync test") + def test_async_given(): + pass + + @given("I have an async resource", target_fixture="resource") + async def async_resource(): + await asyncio.sleep(0.001) + return "async_value" + + @then("the resource should be available") + def check_resource(resource): + assert resource == "async_value" + """ + ) + ) + result = pytester.runpytest_subprocess(*pytest_params) + result.assert_outcomes(passed=1) + + +def test_async_when_then(pytester, pytest_params): + """Test async when and then steps.""" + pytester.makefile( + ".feature", + async_steps=textwrap.dedent( + """\ + Feature: Async steps + Scenario: Async when and then + Given I have a value of 5 + When I multiply it by 2 asynchronously + Then the result should be 10 + """ + ), + ) + pytester.makepyfile( + textwrap.dedent( + """\ + import asyncio + from pytest_bdd import given, when, then, scenario + + @scenario("async_steps.feature", "Async when and then") + def test_async_when_then(): + pass + + @given("I have a value of 5", target_fixture="value") + def value(): + return 5 + + @when("I multiply it by 2 asynchronously", target_fixture="result") + async def multiply(value): + await asyncio.sleep(0.001) + return value * 2 + + @then("the result should be 10") + async def check_result(result): + await asyncio.sleep(0.001) + assert result == 10 + """ + ) + ) + result = pytester.runpytest_subprocess(*pytest_params) + result.assert_outcomes(passed=1) + + +def test_mixed_sync_async_steps(pytester, pytest_params): + """Test mixing sync and async steps in same scenario.""" + pytester.makefile( + ".feature", + mixed=textwrap.dedent( + """\ + Feature: Mixed steps + Scenario: Mix of sync and async + Given I have a sync resource + And I have an async resource + When I combine them synchronously + Then the combination should be correct + """ + ), + ) + pytester.makepyfile( + textwrap.dedent( + """\ + import asyncio + from pytest_bdd import given, when, then, scenario + + @scenario("mixed.feature", "Mix of sync and async") + def test_mixed(): + pass + + @given("I have a sync resource", target_fixture="sync_res") + def sync_resource(): + return "sync" + + @given("I have an async resource", target_fixture="async_res") + async def async_resource(): + await asyncio.sleep(0.001) + return "async" + + @when("I combine them synchronously", target_fixture="combined") + def combine(sync_res, async_res): + return f"{sync_res}+{async_res}" + + @then("the combination should be correct") + def check_combined(combined): + assert combined == "sync+async" + """ + ) + ) + result = pytester.runpytest_subprocess(*pytest_params) + result.assert_outcomes(passed=1) + + +def test_async_step_with_parsers(pytester, pytest_params): + """Test async steps with parameter parsers.""" + pytester.makefile( + ".feature", + parsed=textwrap.dedent( + """\ + Feature: Async with parsers + Scenario: Parameterized async step + Given I wait for 50 milliseconds + Then the wait should be complete + """ + ), + ) + pytester.makepyfile( + textwrap.dedent( + """\ + import asyncio + import pytest + from pytest_bdd import given, then, scenario, parsers + + pytestmark = pytest.mark.asyncio + + @scenario("parsed.feature", "Parameterized async step") + def test_parsed(): + pass + + @given(parsers.parse("I wait for {duration:d} milliseconds"), target_fixture="wait_result") + async def wait_async(duration): + await asyncio.sleep(duration / 1000.0) + return "completed" + + @then("the wait should be complete") + def check_wait(wait_result): + assert wait_result == "completed" + """ + ) + ) + result = pytester.runpytest_subprocess(*pytest_params) + result.assert_outcomes(passed=1) + + +def test_async_step_with_re_parser(pytester, pytest_params): + """Test async steps with regex parser.""" + pytester.makefile( + ".feature", + regex=textwrap.dedent( + """\ + Feature: Async with regex + Scenario: Regex parsed async step + Given I have 42 items + Then the count should be correct + """ + ), + ) + pytester.makepyfile( + textwrap.dedent( + r""" + import asyncio + import pytest + from pytest_bdd import given, then, scenario, parsers + + pytestmark = pytest.mark.asyncio + + @scenario("regex.feature", "Regex parsed async step") + def test_regex(): + pass + + @given( + parsers.re(r"I have (?P\d+) items"), + converters={"count": int}, + target_fixture="count" + ) + async def async_count(count): + await asyncio.sleep(0.001) + return count + + @then("the count should be correct") + def check_count(count): + assert count == 42 + """ + ) + ) + result = pytester.runpytest_subprocess(*pytest_params) + result.assert_outcomes(passed=1) + + +def test_async_step_error_propagation(pytester, pytest_params): + """Test that exceptions in async steps are properly propagated.""" + pytester.makefile( + ".feature", + error=textwrap.dedent( + """\ + Feature: Error handling + Scenario: Async step raises error + Given an async step that fails + """ + ), + ) + pytester.makepyfile( + textwrap.dedent( + """\ + import asyncio + from pytest_bdd import given, scenario + + @scenario("error.feature", "Async step raises error") + def test_error(): + pass + + @given("an async step that fails") + async def failing_step(): + await asyncio.sleep(0.001) + raise ValueError("Async step failed!") + """ + ) + ) + result = pytester.runpytest_subprocess(*pytest_params) + result.assert_outcomes(failed=1) + result.stdout.fnmatch_lines(["*ValueError: Async step failed!*"]) + + +def test_multiple_async_steps_in_sequence(pytester, pytest_params): + """Test multiple async steps executing in sequence.""" + pytester.makefile( + ".feature", + sequence=textwrap.dedent( + """\ + Feature: Sequential async + Scenario: Multiple async steps + Given I start with value 1 + When I add 2 asynchronously + And I multiply by 3 asynchronously + Then the final value should be 9 + """ + ), + ) + pytester.makepyfile( + textwrap.dedent( + """\ + import asyncio + from pytest_bdd import given, when, then, scenario + + @scenario("sequence.feature", "Multiple async steps") + def test_sequence(): + pass + + @given("I start with value 1", target_fixture="value") + async def start_value(): + await asyncio.sleep(0.001) + return 1 + + @when("I add 2 asynchronously", target_fixture="value") + async def add_two(value): + await asyncio.sleep(0.001) + return value + 2 + + @when("I multiply by 3 asynchronously", target_fixture="value") + async def multiply_three(value): + await asyncio.sleep(0.001) + return value * 3 + + @then("the final value should be 9") + def check_final_value(value): + assert value == 9 + """ + ) + ) + result = pytester.runpytest_subprocess(*pytest_params) + result.assert_outcomes(passed=1) + + +def test_async_step_with_fixture_dependency(pytester, pytest_params): + """Test async step that depends on regular pytest fixtures.""" + pytester.makefile( + ".feature", + fixture_dep=textwrap.dedent( + """\ + Feature: Fixture dependency + Scenario: Async step uses fixtures + Given I have an async operation with fixtures + Then the operation should succeed + """ + ), + ) + pytester.makepyfile( + textwrap.dedent( + """\ + import asyncio + import pytest + from pytest_bdd import given, then, scenario + + pytestmark = pytest.mark.asyncio + + @pytest.fixture + def base_value(): + return 10 + + @scenario("fixture_dep.feature", "Async step uses fixtures") + def test_fixture_dep(): + pass + + @given("I have an async operation with fixtures", target_fixture="result") + async def async_with_fixture(base_value): + await asyncio.sleep(0.001) + return base_value * 2 + + @then("the operation should succeed") + def check_result(result): + assert result == 20 + """ + ) + ) + result = pytester.runpytest_subprocess(*pytest_params) + result.assert_outcomes(passed=1) + + +def test_sync_step_with_async_fixture_requires_pytest_asyncio(pytester, pytest_params): + """Test that sync steps with async fixtures require pytest-asyncio. + + This documents a limitation: async pytest fixtures require pytest-asyncio + to be properly awaited. Without it, the fixture returns a coroutine object + instead of the awaited value. This is a pytest limitation, not pytest-bdd. + """ + pytester.makefile( + ".feature", + sync_async_fix=textwrap.dedent( + """\ + Feature: Sync step with async fixture + Scenario: Sync step uses async fixture without pytest-asyncio + Given I have a sync step with async fixture + Then the fixture should be a coroutine + """ + ), + ) + pytester.makepyfile( + textwrap.dedent( + """\ + import asyncio + import inspect + import pytest + from pytest_bdd import given, then, scenario + + pytestmark = pytest.mark.asyncio + + @pytest.fixture + async def async_fixture(): + await asyncio.sleep(0.001) + return "async_fixture_value" + + @scenario("sync_async_fix.feature", "Sync step uses async fixture without pytest-asyncio") + def test_sync_with_async_fixture(): + pass + + @given("I have a sync step with async fixture", target_fixture="fixture_value") + def sync_step_with_async_fixture(async_fixture): + # Without pytest-asyncio, async fixtures return coroutines + return async_fixture + + @then("the fixture should be a coroutine") + def check_is_coroutine(fixture_value): + # This demonstrates the limitation - async fixtures need pytest-asyncio + assert "async_fixture_value" == fixture_value + """ + ) + ) + result = pytester.runpytest_subprocess(*pytest_params) + result.assert_outcomes(passed=1) + + +def test_sync_step_with_async_generator_fixture_requires_pytest_asyncio(pytester, pytest_params): + """Test that sync steps with async fixtures require pytest-asyncio. + + This documents a limitation: async pytest fixtures require pytest-asyncio + to be properly awaited. Without it, the fixture returns a coroutine object + instead of the awaited value. This is a pytest limitation, not pytest-bdd. + """ + pytester.makefile( + ".feature", + sync_async_fix=textwrap.dedent( + """\ + Feature: Sync step with async fixture + Scenario: Sync step uses async fixture without pytest-asyncio + Given I have a sync step with async fixture + Then the fixture should be a coroutine + """ + ), + ) + pytester.makepyfile( + textwrap.dedent( + """\ + import asyncio + import inspect + import pytest + from pytest_bdd import given, then, scenario + + pytestmark = pytest.mark.asyncio + + @pytest.fixture + async def async_fixture(): + await asyncio.sleep(0.001) + yield "async_fixture_value" + + @scenario("sync_async_fix.feature", "Sync step uses async fixture without pytest-asyncio") + def test_sync_with_async_fixture(): + pass + + @given("I have a sync step with async fixture", target_fixture="fixture_value") + def sync_step_with_async_fixture(async_fixture): + # Without pytest-asyncio, async fixtures return coroutines + return async_fixture + + @then("the fixture should be a coroutine") + def check_is_coroutine(fixture_value): + # This demonstrates the limitation - async fixtures need pytest-asyncio + assert "async_fixture_value" == fixture_value + """ + ) + ) + result = pytester.runpytest_subprocess(*pytest_params) + result.assert_outcomes(passed=1) + + +def test_sync_steps_with_async_generator_fixture_requires_pytest_asyncio(pytester, pytest_params): + """Test that sync steps with async fixtures require pytest-asyncio. + + This documents a limitation: async pytest fixtures require pytest-asyncio + to be properly awaited. Without it, the fixture returns a coroutine object + instead of the awaited value. This is a pytest limitation, not pytest-bdd. + """ + pytester.makefile( + ".feature", + sync_async_fix=textwrap.dedent( + """\ + Feature: Sync step with async fixture + Scenario: Sync step uses async fixture without pytest-asyncio + Given I have a sync step with async fixture + And I have another step with async fixture + Then the fixture should be a coroutine + """ + ), + ) + pytester.makepyfile( + textwrap.dedent( + """\ + import asyncio + import inspect + import pytest + from pytest_bdd import given, then, scenario + + pytestmark = pytest.mark.asyncio + + @pytest.fixture + async def async_fixture(): + await asyncio.sleep(0.001) + yield "async_fixture_value" + + @scenario("sync_async_fix.feature", "Sync step uses async fixture without pytest-asyncio") + def test_sync_with_async_fixture(): + pass + + @given("I have a sync step with async fixture", target_fixture="fixture_value") + def sync_step_with_async_fixture(async_fixture): + # Without pytest-asyncio, async fixtures return coroutines + return async_fixture + + @given("I have another step with async fixture", target_fixture="fixture_value_2") + def another_sync_step_with_async_fixture(async_fixture): + # Without pytest-asyncio, async fixtures return coroutines + return async_fixture + + @then("the fixture should be a coroutine") + def check_is_coroutine(fixture_value): + # This demonstrates the limitation - async fixtures need pytest-asyncio + assert "async_fixture_value" == fixture_value + """ + ) + ) + result = pytester.runpytest_subprocess(*pytest_params) + result.assert_outcomes(passed=1) + + +def test_sync_steps_with_async_generator_fixture_requires_anyio(pytester, pytest_params): + """Test that sync steps with async fixtures require anyio. + + This documents a limitation: async pytest fixtures require anyio + to be properly awaited. Without it, the fixture returns a coroutine object + instead of the awaited value. This is a pytest limitation, not pytest-bdd. + """ + pytester.makefile( + ".feature", + sync_async_fix=textwrap.dedent( + """\ + Feature: Sync step with async fixture + Scenario: Sync step uses async fixture without pytest-asyncio + Given I have a sync step with async fixture + And I have another step with async fixture + Then the fixture should be a coroutine + """ + ), + ) + pytester.makepyfile( + textwrap.dedent( + """\ + import anyio + import inspect + import pytest + from pytest_bdd import given, then, scenario + + pytestmark = pytest.mark.anyio + + @pytest.fixture + def anyio_backend(): + return "asyncio" + + @pytest.fixture + async def async_fixture(): + await anyio.sleep(0.001) + yield "async_fixture_value" + + @scenario("sync_async_fix.feature", "Sync step uses async fixture without pytest-asyncio") + def test_sync_with_async_fixture(): + pass + + @given("I have a sync step with async fixture", target_fixture="fixture_value") + def sync_step_with_async_fixture(async_fixture): + return async_fixture + + @given("I have another step with async fixture", target_fixture="fixture_value_2") + def another_sync_step_with_async_fixture(async_fixture): + return async_fixture + + @then("the fixture should be a coroutine") + def check_is_coroutine(fixture_value): + # This demonstrates the limitation - async fixtures need pytest-asyncio + assert "async_fixture_value" == fixture_value + """ + ) + ) + result = pytester.runpytest_subprocess(*pytest_params) + result.assert_outcomes(passed=1)