Add async_ decorators.

Author: maafy6

src/pytest_bdd/asyncio.py

@@ -0,0 +1,3 @@
+from pytest_bdd.steps import async_given, async_then, async_when
+
+__all__ = ["async_given", "async_when", "async_then"]

src/pytest_bdd/steps.py

@@ -83,6 +83,7 @@ def given(
     converters: dict[str, Callable] | None = None,
     target_fixture: str | None = None,
     stacklevel: int = 1,
+    is_async: bool = False,
 ) -> Callable:
     """Given step decorator.
 
@@ -95,14 +96,36 @@ def given(
 
     :return: Decorator function for the step.
     """
-    return step(name, GIVEN, converters=converters, target_fixture=target_fixture, stacklevel=stacklevel)
+    return step(
+        name, GIVEN, converters=converters, target_fixture=target_fixture, stacklevel=stacklevel, is_async=is_async
+    )
+
+
+def async_given(
+    name: str | StepParser,
+    converters: dict[str, Callable] | None = None,
+    target_fixture: str | None = None,
+    stacklevel: int = 1,
+) -> Callable:
+    """Async Given step decorator.
+
+    :param name: Step name or a parser object.
+    :param converters: Optional `dict` of the argument or parameter converters in form
+                       {<param_name>: <converter function>}.
+    :param target_fixture: Target fixture name to replace by steps definition function.
+    :param stacklevel: Stack level to find the caller frame. This is used when injecting the step definition fixture.
+
+    :return: Decorator function for the step.
+    """
+    return given(name, converters=converters, target_fixture=target_fixture, stacklevel=stacklevel, is_async=True)
 
 
 def when(
     name: str | StepParser,
     converters: dict[str, Callable] | None = None,
     target_fixture: str | None = None,
     stacklevel: int = 1,
+    is_async: bool = False,
 ) -> Callable:
     """When step decorator.
 
@@ -115,14 +138,59 @@ def when(
 
     :return: Decorator function for the step.
     """
-    return step(name, WHEN, converters=converters, target_fixture=target_fixture, stacklevel=stacklevel)
+    return step(
+        name, WHEN, converters=converters, target_fixture=target_fixture, stacklevel=stacklevel, is_async=is_async
+    )
+
+
+def async_when(
+    name: str | StepParser,
+    converters: dict[str, Callable] | None = None,
+    target_fixture: str | None = None,
+    stacklevel: int = 1,
+) -> Callable:
+    """When step decorator.
+
+    :param name: Step name or a parser object.
+    :param converters: Optional `dict` of the argument or parameter converters in form
+                       {<param_name>: <converter function>}.
+    :param target_fixture: Target fixture name to replace by steps definition function.
+    :param stacklevel: Stack level to find the caller frame. This is used when injecting the step definition fixture.
+    :param is_async: True if the step is asynchronous. (Default: False)
+
+    :return: Decorator function for the step.
+    """
+    return when(name, converters=converters, target_fixture=target_fixture, stacklevel=stacklevel, is_async=True)
 
 
 def then(
     name: str | StepParser,
     converters: dict[str, Callable] | None = None,
     target_fixture: str | None = None,
     stacklevel: int = 1,
+    is_async: bool = False,
+) -> Callable:
+    """Then step decorator.
+
+    :param name: Step name or a parser object.
+    :param converters: Optional `dict` of the argument or parameter converters in form
+                       {<param_name>: <converter function>}.
+    :param target_fixture: Target fixture name to replace by steps definition function.
+    :param stacklevel: Stack level to find the caller frame. This is used when injecting the step definition fixture.
+    :param is_async: True if the step is asynchronous. (Default: False)
+
+    :return: Decorator function for the step.
+    """
+    return step(
+        name, THEN, converters=converters, target_fixture=target_fixture, stacklevel=stacklevel, is_async=is_async
+    )
+
+
+def async_then(
+    name: str | StepParser,
+    converters: dict[str, Callable] | None = None,
+    target_fixture: str | None = None,
+    stacklevel: int = 1,
 ) -> Callable:
     """Then step decorator.
 
@@ -135,7 +203,7 @@ def then(
 
     :return: Decorator function for the step.
     """
-    return step(name, THEN, converters=converters, target_fixture=target_fixture, stacklevel=stacklevel)
+    return step(name, THEN, converters=converters, target_fixture=target_fixture, stacklevel=stacklevel, is_async=True)
 
 
 def step(
@@ -144,6 +212,7 @@ def step(
     converters: dict[str, Callable] | None = None,
     target_fixture: str | None = None,
     stacklevel: int = 1,
+    is_async: bool = False,
 ) -> Callable[[TCallable], TCallable]:
     """Generic step decorator.
 
@@ -168,10 +237,11 @@ def step(
     def decorator(func: TCallable) -> TCallable:
         parser = get_parser(name)
 
-        if inspect.isasyncgenfunction(func):
-            func = wrap_asyncgen(func)
-        elif inspect.iscoroutinefunction(func):
-            func = wrap_coroutine(func)
+        if is_async:
+            if inspect.isasyncgenfunction(func):
+                func = wrap_asyncgen(func)
+            elif inspect.iscoroutinefunction(func):
+                func = wrap_coroutine(func)
 
         context = StepFunctionContext(
             type=type_,

tests/feature/test_async_steps.py

@@ -25,43 +25,44 @@ def test_steps(pytester):
     pytester.makepyfile(
         textwrap.dedent(
             """\
-        from pytest_bdd import given, when, then, scenario
+        from pytest_bdd import scenario
+        from pytest_bdd.asyncio import async_given, async_when, async_then
 
         @scenario("steps.feature", "Executed step by step")
         def test_steps():
             pass
 
-        @given('I have a foo fixture with value "foo"', target_fixture="foo")
+        @async_given('I have a foo fixture with value "foo"', target_fixture="foo")
         async def _():
             return "foo"
 
 
-        @given("there is a list", target_fixture="results")
+        @async_given("there is a list", target_fixture="results")
         async def _():
             yield []
 
 
-        @when("I append 1 to the list")
+        @async_when("I append 1 to the list")
         async def _(results):
             results.append(1)
 
 
-        @when("I append 2 to the list")
+        @async_when("I append 2 to the list")
         async def _(results):
             results.append(2)
 
 
-        @when("I append 3 to the list")
+        @async_when("I append 3 to the list")
         async def _(results):
             results.append(3)
 
 
-        @then('foo should have value "foo"')
+        @async_then('foo should have value "foo"')
         async def _(foo):
             assert foo == "foo"
 
 
-        @then("the list should be [1, 2, 3]")
+        @async_then("the list should be [1, 2, 3]")
         async def _(results):
             assert results == [1, 2, 3]
 

tests/steps/test_common.py

@@ -5,11 +5,22 @@
 import pytest
 
 from pytest_bdd import given, parsers, then, when
+from pytest_bdd.asyncio import async_given, async_then, async_when
 from pytest_bdd.utils import collect_dumped_objects
 
 
-@pytest.mark.parametrize("step_fn, step_type", [(given, "given"), (when, "when"), (then, "then")])
-def test_given_when_then_delegate_to_step(step_fn: Callable[..., Any], step_type: str) -> None:
+@pytest.mark.parametrize(
+    "step_fn, step_type, is_async",
+    [
+        (given, "given", False),
+        (when, "when", False),
+        (then, "then", False),
+        (async_given, "given", True),
+        (async_when, "when", True),
+        (async_then, "then", True),
+    ],
+)
+def test_given_when_then_delegate_to_step(step_fn: Callable[..., Any], step_type: str, is_async: bool) -> None:
     """Test that @given, @when, @then just delegate the work to @step(...).
     This way we don't have to repeat integration tests for each step decorator.
     """
@@ -18,15 +29,17 @@ def test_given_when_then_delegate_to_step(step_fn: Callable[..., Any], step_type
     with mock.patch("pytest_bdd.steps.step", autospec=True) as step_mock:
         step_fn("foo")
 
-    step_mock.assert_called_once_with("foo", type_=step_type, converters=None, target_fixture=None, stacklevel=1)
+    step_mock.assert_called_once_with(
+        "foo", type_=step_type, converters=None, target_fixture=None, stacklevel=1, is_async=is_async
+    )
 
     # Advanced usage: step parser, converters, target_fixture, ...
     with mock.patch("pytest_bdd.steps.step", autospec=True) as step_mock:
         parser = parsers.re(r"foo (?P<n>\d+)")
         step_fn(parser, converters={"n": int}, target_fixture="foo_n", stacklevel=3)
 
     step_mock.assert_called_once_with(
-        name=parser, type_=step_type, converters={"n": int}, target_fixture="foo_n", stacklevel=3
+        name=parser, type_=step_type, converters={"n": int}, target_fixture="foo_n", stacklevel=3, is_async=is_async
     )