fix: Support for custom loaders in runner

Now all built-in loaders are supported as an argument to `runner.prepare`

docs/changelog.rst

@@ -6,6 +6,11 @@ Changelog
 `Unreleased`_
 -------------
 
+Fixed
+~~~~~
+
+- Support for custom loaders in ``runner``. Now all built-in loaders are supported as an argument to ``runner.prepare``. `#496`_
+
 `1.1.1`_ - 2020-04-12
 ---------------------
 
@@ -892,6 +897,7 @@ Fixed
 .. _0.3.0: https://github.com/kiwicom/schemathesis/compare/v0.2.0...v0.3.0
 .. _0.2.0: https://github.com/kiwicom/schemathesis/compare/v0.1.0...v0.2.0
 
+.. _#496: https://github.com/kiwicom/schemathesis/issues/496
 .. _#489: https://github.com/kiwicom/schemathesis/issues/489
 .. _#473: https://github.com/kiwicom/schemathesis/issues/473
 .. _#469: https://github.com/kiwicom/schemathesis/issues/469

src/schemathesis/loaders.py

@@ -84,6 +84,7 @@ def from_file(
     *,
     app: Any = None,
     validate_schema: bool = True,
+    **kwargs: Any,  # needed in runner to have compatible API across all loaders
 ) -> BaseSchema:
     """Load a file content and parse to schema instance.
 
@@ -168,6 +169,7 @@ def from_wsgi(
     endpoint: Optional[Filter] = None,
     tag: Optional[Filter] = None,
     validate_schema: bool = True,
+    **kwargs: Any,
 ) -> BaseSchema:
     client = Client(app, WSGIResponse)
     response = client.get(schema_path, headers={"User-Agent": USER_AGENT})
@@ -202,6 +204,7 @@ def from_aiohttp(
     tag: Optional[Filter] = None,
     *,
     validate_schema: bool = True,
+    **kwargs: Any,
 ) -> BaseSchema:
     from .extra._aiohttp import run_server  # pylint: disable=import-outside-toplevel
 
@@ -210,4 +213,6 @@ def from_aiohttp(
     url = urljoin(app_url, schema_path)
     if not base_url:
         base_url = app_url
-    return from_uri(url, base_url=base_url, method=method, endpoint=endpoint, tag=tag, validate_schema=validate_schema)
+    return from_uri(
+        url, base_url=base_url, method=method, endpoint=endpoint, tag=tag, validate_schema=validate_schema, **kwargs
+    )

src/schemathesis/runner/__init__.py

@@ -14,7 +14,7 @@
 
 
 def prepare(  # pylint: disable=too-many-arguments
-    schema_uri: str,
+    schema_uri: Union[str, Dict[str, Any]],
     *,
     # Runtime behavior
     checks: Iterable[CheckFunction] = DEFAULT_CHECKS,
@@ -80,7 +80,7 @@ def prepare(  # pylint: disable=too-many-arguments
 
 def execute_from_schema(
     *,
-    schema_uri: str,
+    schema_uri: Union[str, Dict[str, Any]],
     loader: Callable = loaders.from_uri,
     base_url: Optional[str] = None,
     endpoint: Optional[Filter] = None,
@@ -176,7 +176,7 @@ def execute_from_schema(
 
 
 def load_schema(
-    schema_uri: str,
+    schema_uri: Union[str, Dict[str, Any]],
     *,
     base_url: Optional[str] = None,
     loader: Callable = loaders.from_uri,
@@ -194,7 +194,9 @@ def load_schema(
     """Load schema via specified loader and parameters."""
     loader_options = dict_true_values(base_url=base_url, endpoint=endpoint, method=method, tag=tag, app=app)
 
-    if file_exists(schema_uri):
+    if isinstance(schema_uri, dict):
+        loader = loaders.from_dict
+    elif file_exists(schema_uri):
         loader = loaders.from_path
     elif app is not None and not urlparse(schema_uri).netloc:
         # If `schema` is not an existing filesystem path or an URL then it is considered as an endpoint with
@@ -203,7 +205,7 @@ def load_schema(
     else:
         loader_options.update(dict_true_values(headers=headers, auth=auth, auth_type=auth_type))
 
-    if "base_url" not in loader_options:
+    if "base_url" not in loader_options and not isinstance(schema_uri, dict):
         loader_options["base_url"] = get_base_url(schema_uri)
     if loader is loaders.from_uri and loader_options.get("auth"):
         loader_options["auth"] = get_requests_auth(loader_options["auth"], loader_options.pop("auth_type", None))

test/runner/test_runner.py

@@ -1,3 +1,4 @@
+import json
 from typing import Dict, Optional
 
 import pytest
@@ -8,7 +9,7 @@
 from requests.auth import HTTPDigestAuth
 
 import schemathesis
-from schemathesis import from_uri, from_wsgi
+from schemathesis import loaders
 from schemathesis.checks import (
     DEFAULT_CHECKS,
     content_type_conformance,
@@ -21,7 +22,7 @@
 from schemathesis.runner.impl.core import get_wsgi_auth
 
 
-def execute(schema_uri, checks=DEFAULT_CHECKS, loader=from_uri, **options) -> events.Finished:
+def execute(schema_uri, checks=DEFAULT_CHECKS, loader=loaders.from_uri, **options) -> events.Finished:
     generator = prepare(schema_uri=schema_uri, checks=checks, loader=loader, **options)
     all_events = list(generator)
     return all_events[-1]
@@ -87,7 +88,7 @@ def args(request, mocker):
         app_path = request.getfixturevalue("loadable_flask_app")
         # To have simpler tests it is easier to reuse already imported application for inspection
         mocker.patch("schemathesis.runner.import_app", return_value=app)
-        kwargs = {"schema_uri": "/swagger.yaml", "app": app_path, "loader": from_wsgi}
+        kwargs = {"schema_uri": "/swagger.yaml", "app": app_path, "loader": loaders.from_wsgi}
     return app, kwargs
 
 
@@ -448,3 +449,50 @@ def test_auth_loader_options(base_url, schema_url, app):
     execute(schema_url, base_url=base_url, auth=("test", "test"), auth_type="basic")
     schema_request = get_schema_requests(app)
     assert schema_request[0].headers["Authorization"] == "Basic dGVzdDp0ZXN0"
+
+
+@pytest.fixture()
+def raw_schema(app):
+    return app["config"]["schema_data"]
+
+
+@pytest.fixture()
+def json_string(raw_schema):
+    return json.dumps(raw_schema)
+
+
+@pytest.fixture()
+def schema_path(json_string, tmp_path):
+    path = tmp_path / "schema.json"
+    path.write_text(json_string)
+    return str(path)
+
+
+@pytest.fixture()
+def relative_schema_url():
+    return "/swagger.yaml"
+
+
+@pytest.mark.parametrize(
+    "loader, fixture",
+    (
+        (loaders.from_dict, "raw_schema"),
+        (loaders.from_file, "json_string"),
+        (loaders.from_path, "schema_path"),
+        (loaders.from_wsgi, "relative_schema_url"),
+        (loaders.from_aiohttp, "relative_schema_url"),
+    ),
+)
+@pytest.mark.endpoints("success")
+def test_custom_loader(request, loader, fixture):
+    schema = request.getfixturevalue(fixture)
+    kwargs = {}
+    if loader is loaders.from_wsgi:
+        kwargs["app"] = request.getfixturevalue("loadable_flask_app")
+    else:
+        if loader is loaders.from_aiohttp:
+            kwargs["app"] = request.getfixturevalue("loadable_aiohttp_app")
+        kwargs["base_url"] = request.getfixturevalue("base_url")
+    init, *others, finished = prepare(schema, loader=loader, headers={"TEST": "foo"}, **kwargs)
+    assert not finished.has_errors
+    assert not finished.has_failures