WIP

src/schemathesis/_hypothesis.py

@@ -119,7 +119,12 @@ def get_case_strategy(endpoint: Endpoint) -> st.SearchStrategy:
     Path & endpoint are static, the others are JSON schemas.
     """
     strategies = {}
-    static_kwargs = {"path": endpoint.path, "method": endpoint.method, "base_url": endpoint.base_url}
+    static_kwargs = {
+        "path": endpoint.path,
+        "method": endpoint.method,
+        "base_url": endpoint.base_url,
+        "wsgi_client": endpoint.wsgi_client,
+    }
     try:
         for parameter in PARAMETERS:
             value = getattr(endpoint, parameter)

src/schemathesis/loaders.py

@@ -49,13 +49,16 @@ def from_file(
     method: Optional[Filter] = None,
     endpoint: Optional[Filter] = None,
     tag: Optional[Filter] = None,
+    wsgi_client: Optional[Client] = None,
 ) -> BaseSchema:
     """Load a file content and parse to schema instance.
 
     `file` could be a file descriptor, string or bytes.
     """
     raw = yaml.safe_load(file)
-    return from_dict(raw, location=location, base_url=base_url, method=method, endpoint=endpoint, tag=tag)
+    return from_dict(
+        raw, location=location, base_url=base_url, method=method, endpoint=endpoint, tag=tag, wsgi_client=wsgi_client
+    )
 
 
 def from_dict(
@@ -65,13 +68,30 @@ def from_dict(
     method: Optional[Filter] = None,
     endpoint: Optional[Filter] = None,
     tag: Optional[Filter] = None,
+    wsgi_client: Optional[Client] = None,
 ) -> BaseSchema:
     """Get a proper abstraction for the given raw schema."""
     if "swagger" in raw_schema:
-        return SwaggerV20(raw_schema, location=location, base_url=base_url, method=method, endpoint=endpoint, tag=tag)
+        return SwaggerV20(
+            raw_schema,
+            location=location,
+            base_url=base_url,
+            method=method,
+            endpoint=endpoint,
+            tag=tag,
+            wsgi_client=wsgi_client,
+        )
 
     if "openapi" in raw_schema:
-        return OpenApi30(raw_schema, location=location, base_url=base_url, method=method, endpoint=endpoint, tag=tag)
+        return OpenApi30(
+            raw_schema,
+            location=location,
+            base_url=base_url,
+            method=method,
+            endpoint=endpoint,
+            tag=tag,
+            wsgi_client=wsgi_client,
+        )
     raise ValueError("Unsupported schema type")
 
 

src/schemathesis/models.py

@@ -6,7 +6,9 @@
 
 import attr
 import requests
+import werkzeug
 from hypothesis.searchstrategy import SearchStrategy
+from werkzeug import Client
 
 from .exceptions import InvalidSchema
 from .types import Body, Cookies, FormData, Headers, PathParameters, Query
@@ -28,6 +30,7 @@ class Case:
     query: Query = attr.ib(default=None)  # pragma: no mutate
     body: Optional[Body] = attr.ib(default=None)  # pragma: no mutate
     form_data: FormData = attr.ib(default=None)  # pragma: no mutate
+    wsgi_client: Optional[Client] = attr.ib(default=None)  # pragma: no mutate
 
     @property
     def formatted_path(self) -> str:
@@ -87,6 +90,19 @@ def as_requests_kwargs(self, base_url: Optional[str] = None) -> Dict[str, Any]:
             **extra,
         }
 
+    def as_werkzeug_kwargs(self) -> Dict[str, Any]:
+        extra: Dict[str, Optional[Dict]]
+        if self.form_data:
+            extra = {"data": self.form_data}
+        else:
+            extra = {"json": self.body}
+        return {
+            "method": self.method,
+            "path": self.formatted_path,
+            "headers": self.headers,
+            "query_string": self.query ** extra,
+        }
+
     def call(
         self, base_url: Optional[str] = None, session: Optional[requests.Session] = None, **kwargs: Any
     ) -> requests.Response:
@@ -104,6 +120,11 @@ def call(
             session.close()
         return response
 
+    def wsgi_call(self, **kwargs: Any) -> werkzeug.Response:
+        data = self.as_werkzeug_kwargs()
+        response = self.wsgi_client.open(**data, **kwargs)
+        return response
+
 
 def empty_object() -> Dict[str, Any]:
     return {"properties": {}, "additionalProperties": False, "type": "object", "required": []}
@@ -123,6 +144,7 @@ class Endpoint:
     query: Query = attr.ib(default=None)  # pragma: no mutate
     body: Body = attr.ib(default=None)  # pragma: no mutate
     form_data: FormData = attr.ib(default=None)  # pragma: no mutate
+    wsgi_client: Optional[Client] = None
 
     def as_strategy(self) -> SearchStrategy:
         from ._hypothesis import get_case_strategy  # pylint: disable=import-outside-toplevel

src/schemathesis/schemas.py

@@ -17,6 +17,7 @@
 import hypothesis
 import jsonschema
 from requests.structures import CaseInsensitiveDict
+from werkzeug import Client
 
 from schemathesis.exceptions import InvalidSchema
 from schemathesis.models import empty_object
@@ -36,6 +37,7 @@ class BaseSchema(Mapping):
     method: Optional[Filter] = attr.ib(default=None)  # pragma: no mutate
     endpoint: Optional[Filter] = attr.ib(default=None)  # pragma: no mutate
     tag: Optional[Filter] = attr.ib(default=None)  # pragma: no mutate
+    wsgi_client: Optional[Client] = attr.ib(default=None)  # pragma: no mutate
 
     def __iter__(self) -> Iterator:
         return iter(self.endpoints)
@@ -82,6 +84,7 @@ def get_all_tests(
         """Generate all endpoints and Hypothesis tests for them."""
         test: Union[Callable, InvalidSchema]
         for endpoint in self.get_all_endpoints():
+            endpoint.wsgi_client = self.wsgi_client
             try:
                 test = create_test(endpoint, func, settings, seed=seed)
             except InvalidSchema as exc: