feat: WSGI applications support

README.rst

@@ -77,6 +77,21 @@ To speed up the testing process Schemathesis provides ``-w/--workers`` option fo
 
 In the example above all tests will be distributed among 8 worker threads.
 
+If you'd like to test your WSGI app (Flask for example) then there is ``--app`` option for you:
+
+.. code:: bash
+
+    schemathesis run --app=importable.path:app /swagger.json
+
+You need to specify an importable path to the module where your app instance resides and a variable name after ``:`` that points
+to your app. **Note**, app factories are not supported. The schema location could be:
+
+- A full URL;
+- An existing filesystem path;
+- In-app endpoint with schema.
+
+This method is significantly faster than the default one, which involves network.
+
 For the full list of options, run:
 
 .. code:: bash
@@ -162,20 +177,22 @@ look like this:
     @schema.parametrize()
     def test_no_server_errors(case):
         # `requests` will make an appropriate call under the hood
-        response = case.call()
+        response = case.call()  # use `call_wsgi` if you used `schemathesis.from_wsgi`
         assert response.status_code < 500
 
 
 It consists of four main parts:
 
-1. Schema preparation; ``schemathesis`` package provides multiple ways to initialize the schema - ``from_path``, ``from_dict``, ``from_uri``, ``from_file``.
+1. Schema preparation; ``schemathesis`` package provides multiple ways to initialize the schema - ``from_path``, ``from_dict``, ``from_uri``, ``from_file`` and ``from_wsgi``*.
 
 2. Test parametrization; ``@schema.parametrize()`` generates separate tests for all endpoint/method combination available in the schema.
 
 3. A network call to the running application; ``case.call`` does it.
 
 4. Verifying a property you'd like to test; In the example, we verify that any app response will not indicate a server-side error (HTTP codes 5xx).
 
+**NOTE**. Look for ``from_wsgi`` usage `below <https://github.com/kiwicom/schemathesis#wsgi>`_
+
 Run the tests:
 
 .. code:: bash
@@ -236,6 +253,31 @@ To narrow down the scope of the schemathesis tests it is possible to filter by m
 
 The acceptable values are regexps or list of regexps (matched with ``re.search``).
 
+WSGI applications support
+~~~~~~~~~~~~~~~~~~~~~~~~~
+
+Schemathesis supports making calls to WSGI-compliant applications instead of real network calls, in this case
+the test execution will go much faster.
+
+.. code:: python
+
+    app = Flask("test_app")
+
+    @app.route("/schema.json")
+    def schema():
+        return {...}
+
+    @app.route("/v1/users", methods=["GET"])
+    def users():
+        return jsonify([{"name": "Robin"}])
+
+    schema = schemathesis.from_wsgi("/schema.json", app)
+
+    @schema.parametrize()
+    def test_no_server_errors(case):
+        response = case.call_wsgi()
+        assert response.status_code < 500
+
 Explicit examples
 ~~~~~~~~~~~~~~~~~
 

docs/changelog.rst

@@ -6,6 +6,11 @@ Changelog
 `Unreleased`_
 -------------
 
+Added
+~~~~~
+
+- WSGI apps support. `#31`_
+
 `0.19.1`_ - 2019-12-11
 ----------------------
 
@@ -565,6 +570,7 @@ Fixed
 .. _#40: https://github.com/kiwicom/schemathesis/issues/40
 .. _#35: https://github.com/kiwicom/schemathesis/issues/35
 .. _#34: https://github.com/kiwicom/schemathesis/issues/34
+.. _#31: https://github.com/kiwicom/schemathesis/issues/31
 .. _#30: https://github.com/kiwicom/schemathesis/issues/30
 .. _#29: https://github.com/kiwicom/schemathesis/issues/29
 .. _#28: https://github.com/kiwicom/schemathesis/issues/28

pyproject.toml

@@ -35,6 +35,7 @@ pytest-subtests = "^0.2.1"
 requests = "^2.22"
 click = "^7.0"
 importlib_metadata = "^1.1"
+werkzeug = "^0.16.0"
 
 [tool.poetry.dev-dependencies]
 coverage = "^4.5"
@@ -44,6 +45,7 @@ pytest-mock = "^1.11.0"
 pytest-asyncio = "^0.10.0"
 pytest-xdist = "^1.30"
 typing_extensions = "^3.7"
+flask = "^1.1"
 
 [tool.poetry.plugins]
 pytest11 = {schemathesis = "schemathesis.extra.pytest_plugin"}

src/schemathesis/__init__.py

@@ -1,7 +1,7 @@
 from ._hypothesis import init_default_strategies, register_string_format
 from .cli import register_check
 from .constants import __version__
-from .loaders import Parametrizer, from_dict, from_file, from_path, from_pytest_fixture, from_uri
+from .loaders import Parametrizer, from_dict, from_file, from_path, from_pytest_fixture, from_uri, from_wsgi
 from .models import Case
 
 init_default_strategies()

src/schemathesis/_hypothesis.py

@@ -168,6 +168,7 @@ def _get_case_strategy(
         "path": endpoint.path,
         "method": endpoint.method,
         "base_url": endpoint.base_url,
+        "app": endpoint.app,
         **extra_static_parameters,
     }
     if endpoint.method == "GET":

src/schemathesis/cli/__init__.py

@@ -2,16 +2,17 @@
 import traceback
 from contextlib import contextmanager
 from enum import Enum
-from typing import Callable, Dict, Generator, Iterable, List, Optional, Tuple, cast
+from typing import Any, Callable, Dict, Generator, Iterable, List, Optional, Tuple, cast
+from urllib.parse import urlparse
 
 import click
 import hypothesis
 import requests
 from requests import exceptions
-from requests.auth import HTTPDigestAuth
 
 from .. import models, runner, utils
-from ..loaders import from_path
+from ..exceptions import HTTPError
+from ..loaders import from_path, from_wsgi
 from ..runner import events
 from ..types import Filter
 from ..utils import dict_not_none_values, dict_true_values
@@ -87,10 +88,11 @@ def schemathesis(pre_run: Optional[str] = None) -> None:
 @click.option(
     "--base-url",
     "-b",
-    help="Base URL address of the API, required for SCHEMA if specified by file. ",
+    help="Base URL address of the API, required for SCHEMA if specified by file.",
     type=str,
     callback=callbacks.validate_base_url,
 )
+@click.option("--app", help="WSGI application to test", type=str, callback=callbacks.validate_app)
 @click.option("--request-timeout", help="Timeout in milliseconds for network requests during the test run.", type=int)
 @click.option(
     "--hypothesis-deadline",
@@ -130,6 +132,7 @@ def run(  # pylint: disable=too-many-arguments
     tags: Optional[Filter] = None,
     workers_num: int = DEFAULT_WORKERS,
     base_url: Optional[str] = None,
+    app: Any = None,
     request_timeout: Optional[int] = None,
     hypothesis_deadline: Optional[int] = None,
     hypothesis_derandomize: Optional[bool] = None,
@@ -148,12 +151,13 @@ def run(  # pylint: disable=too-many-arguments
 
     selected_checks = tuple(check for check in runner.checks.ALL_CHECKS if check.__name__ in checks)
 
-    if auth and auth_type == "digest":
-        auth = HTTPDigestAuth(*auth)  # type: ignore
+    if auth is None:
+        # Auth type doesn't matter if auth is not passed
+        auth_type = None  # type: ignore
 
     options = dict_true_values(
-        api_options=dict_true_values(auth=auth, headers=headers, request_timeout=request_timeout),
-        loader_options=dict_true_values(base_url=base_url, endpoint=endpoints, method=methods, tag=tags),
+        api_options=dict_true_values(auth=auth, auth_type=auth_type, headers=headers, request_timeout=request_timeout),
+        loader_options=dict_true_values(base_url=base_url, endpoint=endpoints, method=methods, tag=tags, app=app),
         hypothesis_options=dict_not_none_values(
             deadline=hypothesis_deadline,
             derandomize=hypothesis_derandomize,
@@ -167,12 +171,14 @@ def run(  # pylint: disable=too-many-arguments
     )
 
     with abort_on_network_errors():
+        options.update({"checks": selected_checks, "workers_num": workers_num})
         if pathlib.Path(schema).is_file():
-            prepared_runner = runner.prepare(
-                schema, checks=selected_checks, workers_num=workers_num, loader=from_path, **options
-            )
-        else:
-            prepared_runner = runner.prepare(schema, checks=selected_checks, workers_num=workers_num, **options)
+            options["loader"] = from_path
+        elif app is not None and not urlparse(schema).netloc:
+            # If `schema` is not an existing filesystem path or an URL then it is considered as an endpoint with
+            # the given app
+            options["loader"] = from_wsgi
+        prepared_runner = runner.prepare(schema, **options)
     execute(prepared_runner, workers_num)
 
 
@@ -205,13 +211,11 @@ def abort_on_network_errors() -> Generator[None, None, None]:
         message = utils.format_exception(exc)
         click.secho(f"Error: {message}", fg="red")
         raise click.Abort
-    except exceptions.HTTPError as exc:
+    except HTTPError as exc:
         if exc.response.status_code == 404:
-            click.secho(f"Schema was not found at {exc.request.url}", fg="red")
+            click.secho(f"Schema was not found at {exc.url}", fg="red")
             raise click.Abort
-        click.secho(
-            f"Failed to load schema, code {exc.response.status_code} was returned from {exc.request.url}", fg="red"
-        )
+        click.secho(f"Failed to load schema, code {exc.response.status_code} was returned from {exc.url}", fg="red")
         raise click.Abort