refactor(test): Schema loading tests

test/conftest.py

@@ -13,7 +13,7 @@
 from .apps import _graphql as graphql
 from .apps import openapi
 from .apps.openapi.schema import OpenAPIVersion, Operation
-from .utils import get_schema_path, make_schema
+from .utils import SIMPLE_PATH, get_schema_path, make_schema
 
 pytest_plugins = ["pytester", "aiohttp.pytest_plugin", "pytest_mock"]
 
@@ -90,6 +90,11 @@ def app(openapi_version, _app, reset_app):
     return _app
 
 
+@pytest.fixture
+def simple_schema_path():
+    return SIMPLE_PATH
+
+
 @pytest.fixture
 def open_api_2():
     return OpenAPIVersion("2.0")

test/loaders/conftest.py

@@ -0,0 +1,26 @@
+import pytest
+from hypothesis import given, settings
+
+
+def make_inner(get_response):
+    # Minimal smoke test to check whether `call_*` methods work successfully
+    def inner(strategy):
+        @given(case=strategy)
+        @settings(max_examples=1)
+        def test(case):
+            response = get_response(case)
+            assert response.status_code == 200
+
+        test()
+
+    return inner
+
+
+@pytest.fixture
+def run_asgi_test():
+    return make_inner(lambda c: c.call_asgi())
+
+
+@pytest.fixture
+def run_wsgi_test():
+    return make_inner(lambda c: c.call_wsgi())

test/loaders/test_common.py

@@ -0,0 +1,224 @@
+"""Tests for common schema loading logic shared by all loaders."""
+import json
+import platform
+from contextlib import suppress
+
+import pytest
+from requests import Response
+from yarl import URL
+
+import schemathesis
+from schemathesis.constants import USER_AGENT
+from schemathesis.runner import events, prepare
+
+
+@pytest.mark.parametrize(
+    "loader",
+    (
+        schemathesis.openapi.from_asgi,
+        schemathesis.openapi.from_wsgi,
+        schemathesis.graphql.from_asgi,
+        schemathesis.graphql.from_wsgi,
+    ),
+)
+def test_absolute_urls_for_apps(loader):
+    # When an absolute URL passed to a ASGI / WSGI loader
+    # Then it should be rejected
+    with pytest.raises(ValueError, match="Schema path should be relative for WSGI/ASGI loaders"):
+        loader("http://127.0.0.1:1/schema.json", app=None)  # actual app doesn't matter here
+
+
+@pytest.mark.parametrize(
+    "loader", (schemathesis.openapi.from_dict, schemathesis.openapi.from_pytest_fixture, schemathesis.graphql.from_dict)
+)
+def test_invalid_code_sample_style(loader):
+    with pytest.raises(ValueError, match="Invalid value for code sample style: ruby. Available styles: python, curl"):
+        loader({}, code_sample_style="ruby")
+
+
+@pytest.mark.parametrize(
+    "loader, url_fixture, expected",
+    (
+        (schemathesis.openapi.from_uri, "openapi3_schema_url", "http://127.0.0.1:8081/schema.yaml"),
+        (schemathesis.graphql.from_url, "graphql_url", "http://127.0.0.1:8081/graphql"),
+    ),
+)
+def test_port_override(request, loader, url_fixture, expected):
+    url = request.getfixturevalue(url_fixture)
+    # When the user overrides `port`
+    schema = loader(url, port=8081)
+    operation = next(schema.get_all_operations()).ok()
+    # Then the base_url should be taken from `url` and the port should be overridden
+    assert operation.base_url == expected
+
+
+def to_ipv6(url):
+    url = URL(url)
+    parts = list(map(int, url.host.split(".")))
+    ipv6_host = "2002:{:02x}{:02x}:{:02x}{:02x}::".format(*parts)
+    return str(url.with_host("[%s]" % ipv6_host))
+
+
+@pytest.mark.parametrize(
+    "loader, url_fixture, target, expected",
+    (
+        (
+            schemathesis.openapi.from_uri,
+            "openapi3_schema_url",
+            "schemathesis.specs.openapi.loaders.requests.get",
+            "http://[2002:7f00:1::]:8081/schema.yaml",
+        ),
+        (
+            schemathesis.graphql.from_url,
+            "graphql_url",
+            "schemathesis.specs.graphql.loaders.requests.post",
+            "http://[2002:7f00:1::]:8081/graphql",
+        ),
+    ),
+)
+def test_port_override_with_ipv6(request, loader, url_fixture, target, mocker, expected):
+    url = request.getfixturevalue(url_fixture)
+    raw_schema = loader(url).raw_schema
+    response = Response()
+    response.status_code = 200
+    response._content = json.dumps({"data": raw_schema} if url_fixture == "graphql_url" else raw_schema).encode("utf-8")
+    mocker.patch(target, return_value=response)
+
+    url = to_ipv6(url)
+    schema = loader(url, validate_schema=False, port=8081)
+    operation = next(schema.get_all_operations()).ok()
+    assert operation.base_url == expected
+
+
+@pytest.mark.parametrize(
+    "loader, url_fixture",
+    (
+        (schemathesis.openapi.from_uri, "openapi3_schema_url"),
+        (schemathesis.graphql.from_url, "graphql_url"),
+    ),
+)
+@pytest.mark.parametrize("base_url", ("http://example.com/", "http://example.com"))
+def test_base_url_override(request, loader, url_fixture, base_url):
+    url = request.getfixturevalue(url_fixture)
+    # When the user overrides base_url
+    schema = loader(url, base_url=base_url)
+    operation = next(schema.get_all_operations()).ok()
+    # Then the overridden value should not have a trailing slash
+    assert operation.base_url == "http://example.com"
+
+
+@pytest.mark.parametrize(
+    "target, loader",
+    (
+        ("schemathesis.specs.openapi.loaders.requests.get", schemathesis.openapi.from_uri),
+        ("schemathesis.specs.graphql.loaders.requests.post", schemathesis.graphql.from_url),
+    ),
+)
+def test_uri_loader_custom_kwargs(mocker, target, loader):
+    # All custom kwargs are passed to `requests` as is
+    mocked = mocker.patch(target)
+    with suppress(Exception):
+        loader("http://127.0.0.1:8000", verify=False, headers={"X-Test": "foo"})
+    assert mocked.call_args[1]["verify"] is False
+    assert mocked.call_args[1]["headers"] == {"X-Test": "foo", "User-Agent": USER_AGENT}
+
+
+@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 "/schema.yaml"
+
+
+@pytest.mark.parametrize(
+    "loader, fixture",
+    (
+        (schemathesis.openapi.from_dict, "raw_schema"),
+        (schemathesis.openapi.from_file, "json_string"),
+        (schemathesis.openapi.from_path, "schema_path"),
+        (schemathesis.openapi.from_wsgi, "relative_schema_url"),
+        (schemathesis.openapi.from_aiohttp, "relative_schema_url"),
+    ),
+)
+@pytest.mark.operations("success")
+def test_non_default_loader(openapi_version, request, loader, fixture):
+    schema = request.getfixturevalue(fixture)
+    kwargs = {}
+    if loader is schemathesis.openapi.from_wsgi:
+        kwargs["app"] = request.getfixturevalue("loadable_flask_app")
+    else:
+        if loader is schemathesis.openapi.from_aiohttp:
+            kwargs["app"] = request.getfixturevalue("loadable_aiohttp_app")
+        kwargs["base_url"] = request.getfixturevalue("base_url")
+    # Common kwargs combinations for loaders should work without errors
+    init, *others, finished = prepare(schema, loader=loader, headers={"TEST": "foo"}, **kwargs)
+    assert not finished.has_errors
+    assert not finished.has_failures
+
+
+FROM_DICT_ERROR_MESSAGE = "Dictionary as a schema is allowed only with `from_dict` loader"
+
+
+@pytest.mark.parametrize(
+    "loader, schema, message",
+    (
+        (schemathesis.openapi.from_uri, {}, FROM_DICT_ERROR_MESSAGE),
+        (schemathesis.openapi.from_dict, "", "Schema should be a dictionary for `from_dict` loader"),
+        (schemathesis.graphql.from_dict, "", "Schema should be a dictionary for `from_dict` loader"),
+        (schemathesis.openapi.from_wsgi, {}, FROM_DICT_ERROR_MESSAGE),
+        (schemathesis.openapi.from_file, {}, FROM_DICT_ERROR_MESSAGE),
+        (schemathesis.openapi.from_path, {}, FROM_DICT_ERROR_MESSAGE),
+        (schemathesis.graphql.from_wsgi, {}, FROM_DICT_ERROR_MESSAGE),
+    ),
+)
+def test_validation(loader, schema, message):
+    # When incorrect schema is passed to a loader
+    with pytest.raises(ValueError, match=message):
+        # Then it should be rejected
+        list(prepare(schema, loader=loader))
+
+
+def test_custom_loader(swagger_20, openapi2_base_url):
+    swagger_20.base_url = openapi2_base_url
+    # Custom loaders are not validated
+    *others, finished = list(prepare({}, loader=lambda *args, **kwargs: swagger_20))
+    assert not finished.has_errors
+    assert not finished.has_failures
+
+
+def test_from_path_loader_ignore_network_parameters(openapi2_base_url):
+    # When `from_path` loader is used
+    # And network-related parameters are passed
+    all_events = list(
+        prepare(
+            openapi2_base_url,
+            loader=schemathesis.openapi.from_path,
+            auth=("user", "password"),
+            headers={"X-Foo": "Bar"},
+            auth_type="basic",
+        )
+    )
+    # Then those parameters should be ignored during schema loading
+    # And a proper error message should be displayed
+    assert len(all_events) == 1
+    assert isinstance(all_events[0], events.InternalError)
+    if platform.system() == "Windows":
+        exception_type = "builtins.OSError"
+    else:
+        exception_type = "builtins.FileNotFoundError"
+    assert all_events[0].exception_type == exception_type

test/loaders/test_graphql.py

@@ -0,0 +1,18 @@
+"""GraphQL specific loader behavior."""
+from schemathesis.specs.graphql import loaders
+
+
+def test_graphql_asgi_loader(graphql_path, fastapi_graphql_app, run_asgi_test):
+    # When an ASGI app is loaded via `from_asgi`
+    schema = loaders.from_asgi(graphql_path, fastapi_graphql_app)
+    strategy = schema[graphql_path]["POST"].as_strategy()
+    # Then it should successfully make calls via `call_asgi`
+    run_asgi_test(strategy)
+
+
+def test_graphql_wsgi_loader(graphql_path, graphql_app, run_wsgi_test):
+    # When a WSGI app is loaded via `from_wsgi`
+    schema = loaders.from_wsgi(graphql_path, graphql_app)
+    strategy = schema[graphql_path]["POST"].as_strategy()
+    # Then it should successfully make calls via `call_wsgi`
+    run_wsgi_test(strategy)

test/loaders/test_openapi.py

@@ -0,0 +1,79 @@
+"""OpenAPI specific loader behavior."""
+import json
+
+import pytest
+
+from schemathesis.specs.openapi import loaders
+from schemathesis.specs.openapi.schemas import OpenApi30, SwaggerV20
+
+
+def test_openapi_asgi_loader(fastapi_app, run_asgi_test):
+    # When an ASGI app is loaded via `from_asgi`
+    schema = loaders.from_asgi("/openapi.json", fastapi_app)
+    strategy = schema["/users"]["GET"].as_strategy()
+    # Then it should successfully make calls via `call_asgi`
+    run_asgi_test(strategy)
+
+
+def test_openapi_wsgi_loader(flask_app, run_wsgi_test):
+    # When a WSGI app is loaded via `from_wsgi`
+    schema = loaders.from_wsgi("/schema.yaml", flask_app)
+    strategy = schema["/success"]["GET"].as_strategy()
+    # Then it should successfully make calls via `call_wsgi`
+    run_wsgi_test(strategy)
+
+
+@pytest.mark.parametrize(
+    "version, expected",
+    (
+        ("20", SwaggerV20),
+        ("30", OpenApi30),
+    ),
+)
+def test_force_open_api_version(version, expected):
+    schema = {
+        # Invalid schema, but it happens in real applications
+        "swagger": "2.0",
+        "openapi": "3.0.0",
+    }
+    loaded = loaders.from_dict(schema, force_schema_version=version, validate_schema=False)
+    assert isinstance(loaded, expected)
+
+
+def test_number_deserializing(testdir):
+    # When numbers in a schema are written in scientific notation but without a dot
+    # (achieved by dumping the schema with json.dumps)
+    schema = {
+        "openapi": "3.0.2",
+        "info": {"title": "Test", "description": "Test", "version": "0.1.0"},
+        "paths": {
+            "/teapot": {
+                "get": {
+                    "summary": "Test",
+                    "parameters": [
+                        {
+                            "name": "key",
+                            "in": "query",
+                            "required": True,
+                            "schema": {"type": "number", "multipleOf": 0.00001},
+                        }
+                    ],
+                    "responses": {"200": {"description": "OK"}},
+                }
+            }
+        },
+    }
+
+    schema_path = testdir.makefile(".yaml", schema=json.dumps(schema))
+    # Then yaml loader should parse them without schema validation errors
+    parsed = loaders.from_path(str(schema_path))
+    # and the value should be a number
+    value = parsed.raw_schema["paths"]["/teapot"]["get"]["parameters"][0]["schema"]["multipleOf"]
+    assert isinstance(value, float)
+
+
+def test_unsupported_type():
+    # When Schemathesis can't detect the Open API spec version
+    with pytest.raises(ValueError, match="^Unsupported schema type$"):
+        # Then it raises an error
+        loaders.from_dict({})