feat: Stateful testing via Open API links for the pytest runner

docs/changelog.rst

@@ -6,11 +6,18 @@ Changelog
 `Unreleased`_
 -------------
 
+Added
+~~~~~
+
+- Stateful testing via Open API links for the ``pytest`` runner. `#616`_
+
 Changed
 ~~~~~~~
 
 - Check name is no longer displayed in the CLI output, since its verbose message is already displayed. This change
   also simplifies the internal structure of the runner events.
+- The ``stateful`` argument type in the ``runner.prepare`` is ``Optional[Stateful]`` instead of ``Optional[str]``. Use
+  ``schemathesis.Stateful`` enum.
 
 `2.4.1`_ - 2020-09-17
 ---------------------
@@ -1368,6 +1375,7 @@ Fixed
 .. _#621: https://github.com/schemathesis/schemathesis/issues/621
 .. _#618: https://github.com/schemathesis/schemathesis/issues/618
 .. _#617: https://github.com/schemathesis/schemathesis/issues/617
+.. _#616: https://github.com/schemathesis/schemathesis/issues/616
 .. _#614: https://github.com/schemathesis/schemathesis/issues/614
 .. _#612: https://github.com/schemathesis/schemathesis/issues/612
 .. _#600: https://github.com/schemathesis/schemathesis/issues/600

docs/stateful.rst

@@ -3,8 +3,8 @@
 Stateful testing
 ================
 
-By default, Schemathesis generates random data for all endpoints in your schema. With Schemathesis's `stateful testing`
-Schemathesis CLI will try to reuse data from requests that were sent and responses received for generating requests to
+By default, Schemathesis generates random data for all endpoints in your schema. With Schemathesis's ``stateful testing``
+Schemathesis will try to reuse data from requests that were sent and responses received for generating requests to
 other endpoints.
 
 Open API Links
@@ -61,9 +61,11 @@ Based on this definition, Schemathesis will:
 In this case, it is much more likely that instead of a 404 response for a randomly-generated ``user_id`` we'll receive
 something else - for example, HTTP codes 200 or 500.
 
-By default, stateful testing is disabled. You can add it via ``--stateful=links`` CLI option. Please, note that more
+By default, stateful testing is disabled. You can add it via the ``--stateful=links`` CLI option or with the ``stateful=Stateful.links`` argument to ``parametrize``. Please, note that more
 different algorithms for stateful testing might be implemented in the future.
 
+CLI:
+
 .. code:: bash
 
     schemathesis run --stateful=links http://0.0.0.0/schema.yaml
@@ -80,9 +82,32 @@ different algorithms for stateful testing might be implemented in the future.
 
     ...
 
+Python tests:
+
+.. code:: python
+
+    from schemathesis import Stateful
+
+    @schema.parametrize(stateful=Stateful.links)
+    def test_api(case):
+        response = case.call()
+        ...
+
 Each additional test will be indented and prefixed with ``->`` in the CLI output.
-You can specify recursive links if you want. The default recursion depth limit is ``5``, it can be changed with
-``--stateful-recursion-limit=<N>`` CLI option.
+You can specify recursive links if you want. The default recursion depth limit is ``5``, it can be changed with the
+``--stateful-recursion-limit=<N>`` CLI option or with the ``stateful_recursion_limit=<N>`` argument to ``parametrize``.
+
+**NOTE**. If you use stateful testing in Python tests, make sure you use the ``case.call`` method that automatically stores the response for further usage.
+Alternatively, you could use ``case.store_response`` and store the received response by hand:
+
+.. code:: python
+
+    @schema.parametrize(stateful=Stateful.links)
+    def test_api(case):
+        response = case.call()  # stores the response automatically
+        # OR, store it manually
+        response = requests.request(**case.as_requests_kwargs())
+        case.store_response(response)
 
 Even though this feature appears only in Open API 3.0 specification, under Open API 2.0, you can use it
 via the ``x-links`` extension, the syntax is the same, but you need to use the ``x-links`` keyword instead of ``links``.

src/schemathesis/__init__.py

@@ -5,5 +5,6 @@
 from .models import Case
 from .specs import graphql
 from .specs.openapi._hypothesis import init_default_strategies, register_string_format
+from .stateful import Stateful
 
 init_default_strategies()

src/schemathesis/_hypothesis.py

@@ -9,14 +9,20 @@
 from .exceptions import InvalidSchema
 from .hooks import GLOBAL_HOOK_DISPATCHER, HookContext, HookDispatcher
 from .models import Case, Endpoint
+from .stateful import Feedback, Stateful
 
 
 def create_test(
     endpoint: Endpoint, test: Callable, settings: Optional[hypothesis.settings] = None, seed: Optional[int] = None
 ) -> Callable:
     """Create a Hypothesis test."""
     hook_dispatcher = getattr(test, "_schemathesis_hooks", None)
-    strategy = endpoint.as_strategy(hooks=hook_dispatcher)
+    feedback: Optional[Feedback]
+    if endpoint.schema.stateful == Stateful.links:
+        feedback = Feedback(endpoint.schema.stateful, endpoint)
+    else:
+        feedback = None
+    strategy = endpoint.as_strategy(hooks=hook_dispatcher, feedback=feedback)
     wrapped_test = hypothesis.given(case=strategy)(test)
     if seed is not None:
         wrapped_test = hypothesis.seed(seed)(wrapped_test)
@@ -25,6 +31,7 @@ def create_test(
     setup_default_deadline(wrapped_test)
     if settings is not None:
         wrapped_test = settings(wrapped_test)
+    wrapped_test._schemathesis_feedback = feedback  # type: ignore
     return add_examples(wrapped_test, endpoint, hook_dispatcher=hook_dispatcher)
 
 

src/schemathesis/cli/__init__.py

@@ -11,10 +11,12 @@
 from .. import checks as checks_module
 from .. import runner
 from .. import targets as targets_module
+from ..constants import DEFAULT_STATEFUL_RECURSION_LIMIT
 from ..fixups import ALL_FIXUPS
 from ..hooks import GLOBAL_HOOK_DISPATCHER, HookContext, HookDispatcher, HookScope
 from ..models import CheckFunction
-from ..runner import DEFAULT_STATEFUL_RECURSION_LIMIT, events
+from ..runner import events
+from ..stateful import Stateful
 from ..targets import Target
 from ..types import Filter
 from . import callbacks, cassettes, output
@@ -198,7 +200,12 @@ def schemathesis(pre_run: Optional[str] = None) -> None:
     multiple=True,
     type=click.Choice(list(ALL_FIXUPS) + ["all"]),
 )
-@click.option("--stateful", help="Utilize stateful testing capabilities.", type=click.Choice(["links"]))
+@click.option(
+    "--stateful",
+    help="Utilize stateful testing capabilities.",
+    type=click.Choice([item.name for item in Stateful]),
+    callback=callbacks.convert_stateful,
+)
 @click.option(
     "--stateful-recursion-limit",
     help="Limit recursion depth for stateful testing.",
@@ -255,7 +262,7 @@ def run(  # pylint: disable=too-many-arguments
     show_errors_tracebacks: bool = False,
     store_network_log: Optional[click.utils.LazyFile] = None,
     fixups: Tuple[str] = (),  # type: ignore
-    stateful: Optional[str] = None,
+    stateful: Optional[Stateful] = None,
     stateful_recursion_limit: int = DEFAULT_STATEFUL_RECURSION_LIMIT,
     hypothesis_deadline: Optional[Union[int, NotSet]] = None,
     hypothesis_derandomize: Optional[bool] = None,

src/schemathesis/cli/callbacks.py

@@ -8,6 +8,7 @@
 from requests import PreparedRequest, RequestException
 
 from .. import utils
+from ..stateful import Stateful
 
 
 def validate_schema(ctx: click.core.Context, param: click.core.Parameter, raw_value: str) -> str:
@@ -110,6 +111,12 @@ def convert_verbosity(
     return hypothesis.Verbosity[value]
 
 
+def convert_stateful(ctx: click.core.Context, param: click.core.Parameter, value: Optional[str]) -> Optional[Stateful]:
+    if value is None:
+        return value
+    return Stateful[value]
+
+
 @contextmanager
 def reraise_format_error(raw_value: str) -> Generator[None, None, None]:
     try:

src/schemathesis/constants.py

@@ -9,3 +9,4 @@
 
 USER_AGENT = f"schemathesis/{__version__}"
 DEFAULT_DEADLINE = 500  # pragma: no mutate
+DEFAULT_STATEFUL_RECURSION_LIMIT = 5  # pragma: no mutate

src/schemathesis/extra/pytest_plugin.py

@@ -1,18 +1,21 @@
 from functools import partial
-from typing import Any, Callable, Generator, List, Optional, Type, cast
+from typing import Any, Callable, Generator, List, Optional, Type, Union, cast
 
 import pytest
 from _pytest import fixtures, nodes
 from _pytest.config import hookimpl
 from _pytest.fixtures import FuncFixtureInfo
 from _pytest.nodes import Node
 from _pytest.python import Class, Function, FunctionDefinition, Metafunc, Module, PyCollector
+from _pytest.runner import runtestprotocol
+from _pytest.warning_types import PytestWarning
 from hypothesis.errors import InvalidArgument  # pylint: disable=ungrouped-imports
 from packaging import version
 
 from .._hypothesis import create_test
 from ..exceptions import InvalidSchema
 from ..models import Endpoint
+from ..stateful import Feedback
 from ..utils import is_schemathesis_test
 
 USE_FROM_PARENT = version.parse(pytest.__version__) >= version.parse("5.4.0")
@@ -53,7 +56,15 @@ def _gen_items(self, endpoint: Endpoint) -> Generator[Function, None, None]:
         metafunc = self._parametrize(cls, definition, fixtureinfo)
 
         if not metafunc._calls:
-            yield create(SchemathesisFunction, name=name, parent=self.parent, callobj=funcobj, fixtureinfo=fixtureinfo)
+            yield create(
+                SchemathesisFunction,
+                name=name,
+                parent=self.parent,
+                callobj=funcobj,
+                fixtureinfo=fixtureinfo,
+                test_func=self.test_function,
+                originalname=self.name,
+            )
         else:
             fixtures.add_funcarg_pseudo_fixture_def(self.parent, metafunc, fixturemanager)
             fixtureinfo.prune_dependency_tree()
@@ -68,6 +79,7 @@ def _gen_items(self, endpoint: Endpoint) -> Generator[Function, None, None]:
                     fixtureinfo=fixtureinfo,
                     keywords={callspec.id: True},
                     originalname=name,
+                    test_func=self.test_function,
                 )
 
     def _get_class_parent(self) -> Optional[Type]:
@@ -105,14 +117,73 @@ def collect(self) -> List[Function]:  # type: ignore
             pytest.fail("Error during collection")
 
 
+NOT_USED_STATEFUL_TESTING_MESSAGE = (
+    "You are using stateful testing, but no responses were stored during the test! "
+    "Please, use `case.call` or `case.store_response` in your test to enable stateful tests."
+)
+
+
 class SchemathesisFunction(Function):  # pylint: disable=too-many-ancestors
+    def __init__(
+        self, *args: Any, test_func: Callable, test_name: Optional[str] = None, recursion_level: int = 0, **kwargs: Any
+    ) -> None:
+        super().__init__(*args, **kwargs)
+        self.test_function = test_func
+        self.test_name = test_name
+        self.recursion_level = recursion_level
+
     def _getobj(self) -> partial:
         """Tests defined as methods require `self` as the first argument.
 
         This method is called only for this case.
         """
         return partial(self.obj, self.parent.obj)
 
+    @property
+    def feedback(self) -> Optional[Feedback]:
+        return getattr(self.obj, "_schemathesis_feedback", None)
+
+    def warn_if_stateful_responses_not_stored(self) -> None:
+        feedback = self.feedback
+        if feedback is not None and not feedback.stateful_tests:
+            self.warn(PytestWarning(NOT_USED_STATEFUL_TESTING_MESSAGE))
+
+    def _get_stateful_tests(self) -> List["SchemathesisFunction"]:
+        feedback = self.feedback
+        recursion_level = self.recursion_level
+        if feedback is None or recursion_level >= feedback.endpoint.schema.stateful_recursion_limit:
+            return []
+        previous_test_name = self.test_name or f"{feedback.endpoint.method}:{feedback.endpoint.full_path}"
+
+        def make_test(
+            endpoint: Endpoint, test: Union[Callable, InvalidSchema], previous_tests: str
+        ) -> SchemathesisFunction:
+            test_name = f"{previous_tests} -> {endpoint.method}:{endpoint.full_path}"
+            return create(
+                self.__class__,
+                name=f"{self.originalname}[{test_name}]",
+                parent=self.parent,
+                callspec=getattr(self, "callspec", None),
+                callobj=test,
+                fixtureinfo=self._fixtureinfo,
+                keywords=self.keywords,
+                originalname=self.originalname,
+                test_func=self.test_function,
+                test_name=test_name,
+                recursion_level=recursion_level + 1,
+            )
+
+        return [
+            make_test(endpoint, test, previous_test_name)
+            for (endpoint, test) in feedback.get_stateful_tests(self.test_function, None, None)
+        ]
+
+    def add_stateful_tests(self) -> None:
+        idx = self.session.items.index(self) + 1
+        tests = self._get_stateful_tests()
+        self.session.items[idx:idx] = tests
+        self.session.testscollected += len(tests)
+
 
 @hookimpl(hookwrapper=True)  # type:ignore # pragma: no mutate
 def pytest_pycollect_makeitem(collector: nodes.Collector, name: str, obj: Any) -> Generator[None, Any, None]:
@@ -135,3 +206,13 @@ def pytest_pyfunc_call(pyfuncitem):  # type:ignore
         outcome.get_result()
     except InvalidArgument as exc:
         pytest.fail(exc.args[0])
+
+
+def pytest_runtest_protocol(item: Function, nextitem: Optional[Function]) -> bool:
+    item.ihook.pytest_runtest_logstart(nodeid=item.nodeid, location=item.location)
+    runtestprotocol(item, nextitem=nextitem)
+    item.ihook.pytest_runtest_logfinish(nodeid=item.nodeid, location=item.location)
+    if isinstance(item, SchemathesisFunction):
+        item.warn_if_stateful_responses_not_stored()
+        item.add_stateful_tests()
+    return True

src/schemathesis/models.py

@@ -24,7 +24,13 @@
 if TYPE_CHECKING:
     from .hooks import HookDispatcher
     from .schemas import BaseSchema
-    from .stateful import StatefulTest
+    from .stateful import Feedback, Stateful, StatefulTest
+
+
+MISSING_STATEFUL_ARGUMENT_MESSAGE = (
+    "To use `store_response` you need to enable stateful testing by adding "
+    "`stateful=Stateful.links` to your `parametrize` call."
+)
 
 
 @attr.s(slots=True)  # pragma: no mutate
@@ -39,6 +45,8 @@ class Case:
     body: Optional[Body] = attr.ib(default=None)  # pragma: no mutate
     form_data: Optional[FormData] = attr.ib(default=None)  # pragma: no mutate
 
+    feedback: "Feedback" = attr.ib(repr=False, default=None)
+
     @property
     def path(self) -> str:
         return self.endpoint.path
@@ -177,8 +185,15 @@ def call(
         response = session.request(**data)  # type: ignore
         if close_session:
             session.close()
+        if self.feedback:
+            self.store_response(response)
         return response
 
+    def store_response(self, response: GenericResponse) -> None:
+        if self.feedback is None:
+            raise RuntimeError(MISSING_STATEFUL_ARGUMENT_MESSAGE)
+        self.feedback.add_test_case(self, response)
+
     def as_werkzeug_kwargs(self, headers: Optional[Dict[str, str]] = None) -> Dict[str, Any]:
         """Convert the case into a dictionary acceptable by werkzeug.Client."""
         final_headers = self.headers.copy() if self.headers is not None else {}
@@ -212,7 +227,10 @@ def call_wsgi(self, app: Any = None, headers: Optional[Dict[str, str]] = None, *
         data = self.as_werkzeug_kwargs(headers)
         client = werkzeug.Client(application, WSGIResponse)
         with cookie_handler(client, self.cookies):
-            return client.open(**data, **kwargs)
+            response = client.open(**data, **kwargs)
+        if self.feedback:
+            self.store_response(response)
+        return response
 
     def call_asgi(
         self,
@@ -341,14 +359,16 @@ class Endpoint:
     def full_path(self) -> str:
         return self.schema.get_full_path(self.path)
 
-    def as_strategy(self, hooks: Optional["HookDispatcher"] = None) -> SearchStrategy:
-        return self.schema.get_case_strategy(self, hooks)
+    def as_strategy(
+        self, hooks: Optional["HookDispatcher"] = None, feedback: Optional["Feedback"] = None
+    ) -> SearchStrategy:
+        return self.schema.get_case_strategy(self, hooks, feedback)
 
     def get_strategies_from_examples(self) -> List[SearchStrategy[Case]]:
         """Get examples from endpoint."""
         return self.schema.get_strategies_from_examples(self)
 
-    def get_stateful_tests(self, response: GenericResponse, stateful: Optional[str]) -> Sequence["StatefulTest"]:
+    def get_stateful_tests(self, response: GenericResponse, stateful: Optional["Stateful"]) -> Sequence["StatefulTest"]:
         return self.schema.get_stateful_tests(response, self, stateful)
 
     def get_hypothesis_conversions(self, location: str) -> Optional[Callable]: