diff --git a/docs/changelog.rst b/docs/changelog.rst index 1b055dae32..cb99506afa 100644 --- a/docs/changelog.rst +++ b/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 diff --git a/src/schemathesis/loaders.py b/src/schemathesis/loaders.py index 91d8873ed8..b8b99737fc 100644 --- a/src/schemathesis/loaders.py +++ b/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 + ) diff --git a/src/schemathesis/runner/__init__.py b/src/schemathesis/runner/__init__.py index ea24324c29..ceb7559af8 100644 --- a/src/schemathesis/runner/__init__.py +++ b/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)) diff --git a/test/runner/test_runner.py b/test/runner/test_runner.py index 7e7778af70..51cfdc3cc0 100644 --- a/test/runner/test_runner.py +++ b/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