feat: ASGI support for GraphQL schemas

docs/changelog.rst

@@ -8,6 +8,7 @@ Changelog
 
 - Custom keyword arguments to ``schemathesis.graphql.from_url`` that are proxied to ``requests.post``.
 - ``from_wsgi`` loader for GraphQL apps. `#1097`_
+- ``from_asgi`` loader for GraphQL apps. `#1100`_
 - Support for ``app`` & ``base_url`` arguments for the ``from_pytest_fixture`` runner.
 
 **Changed**
@@ -1766,6 +1767,7 @@ Deprecated
 .. _0.3.0: https://github.com/schemathesis/schemathesis/compare/v0.2.0...v0.3.0
 .. _0.2.0: https://github.com/schemathesis/schemathesis/compare/v0.1.0...v0.2.0
 
+.. _#1100: https://github.com/schemathesis/schemathesis/issues/1100
 .. _#1097: https://github.com/schemathesis/schemathesis/issues/1097
 .. _#1093: https://github.com/schemathesis/schemathesis/issues/1093
 .. _#1073: https://github.com/schemathesis/schemathesis/issues/1073

src/schemathesis/specs/graphql/__init__.py

@@ -1 +1 @@
-from .loaders import from_dict, from_url, from_wsgi
+from .loaders import from_asgi, from_dict, from_url, from_wsgi

src/schemathesis/specs/graphql/loaders.py

@@ -1,6 +1,7 @@
 from typing import Any, Dict, Optional
 
 import requests
+from starlette.testclient import TestClient as ASGIClient
 from werkzeug import Client
 from yarl import URL
 
@@ -147,3 +148,28 @@ def from_wsgi(schema_path: str, app: Any, base_url: Optional[str] = None, **kwar
     response = client.post(schema_path, **kwargs)
     HTTPError.check_response(response, schema_path)
     return from_dict(raw_schema=response.json["data"], location=schema_path, base_url=base_url, app=app)
+
+
+def from_asgi(
+    schema_path: str,
+    app: Any,
+    base_url: Optional[str] = None,
+    **kwargs: Any,
+) -> GraphQLSchema:
+    """Load GraphQL schema from an ASGI app.
+
+    :param str schema_path: An in-app relative URL to the schema.
+    :param app: An ASGI app instance.
+    """
+    require_relative_url(schema_path)
+    setup_headers(kwargs)
+    kwargs.setdefault("json", {"query": INTROSPECTION_QUERY})
+    client = ASGIClient(app)
+    response = client.post(schema_path, **kwargs)
+    HTTPError.check_response(response, schema_path)
+    return from_dict(
+        response.json()["data"],
+        location=schema_path,
+        base_url=base_url,
+        app=app,
+    )

src/schemathesis/specs/graphql/schemas.py

@@ -4,6 +4,7 @@
 
 import attr
 import graphql
+import requests
 from hypothesis import strategies as st
 from hypothesis.strategies import SearchStrategy
 from hypothesis_graphql import strategies as gql_st
@@ -48,6 +49,15 @@ def validate_response(
         checks += additional_checks
         return super().validate_response(response, checks, code_sample_style=code_sample_style)
 
+    def call_asgi(
+        self,
+        app: Any = None,
+        base_url: Optional[str] = None,
+        headers: Optional[Dict[str, str]] = None,
+        **kwargs: Any,
+    ) -> requests.Response:
+        return super().call_asgi(app=app, base_url=base_url, headers=headers, **kwargs)
+
 
 @attr.s()  # pragma: no mutate
 class GraphQLSchema(BaseSchema):

test/apps/_graphql/__init__.py

@@ -1 +1 @@
-from . import _flask
+from . import _fastapi, _flask

test/apps/_graphql/_fastapi/__init__.py

@@ -0,0 +1,10 @@
+from fastapi import FastAPI
+from starlette.graphql import GraphQLApp
+
+from ..schema import schema
+
+
+def create_app(path="/graphql"):
+    app = FastAPI()
+    app.add_route(path, GraphQLApp(schema=schema))
+    return app

test/apps/_graphql/_fastapi/app.py

@@ -0,0 +1,3 @@
+from . import create_app
+
+app = create_app()

test/conftest.py

@@ -10,8 +10,8 @@
 from schemathesis.extra._flask import run_server as run_flask_server
 from schemathesis.specs.openapi import loaders as oas_loaders
 
-from .apps import _graphql
-from .apps.openapi import _aiohttp, _fastapi, _flask
+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
 
@@ -51,7 +51,7 @@ def pytest_configure(config):
 @pytest.fixture(scope="session")
 def _app():
     """A global AioHTTP application with configurable API operations."""
-    return _aiohttp.create_app(("success", "failure"))
+    return openapi._aiohttp.create_app(("success", "failure"))
 
 
 @pytest.fixture
@@ -70,7 +70,7 @@ def operations(request):
 @pytest.fixture
 def reset_app(_app, operations):
     def inner(version):
-        _aiohttp.reset_app(_app, operations, version)
+        openapi._aiohttp.reset_app(_app, operations, version)
 
     return inner
 
@@ -160,7 +160,7 @@ def graphql_path():
 
 @pytest.fixture(scope="session")
 def graphql_app(graphql_path):
-    return _graphql._flask.create_app(graphql_path)
+    return graphql._flask.create_app(graphql_path)
 
 
 @pytest.fixture()
@@ -506,7 +506,7 @@ def openapi_30():
 
 @pytest.fixture()
 def app_schema(openapi_version, operations):
-    return _aiohttp.make_openapi_schema(operations=operations, version=openapi_version)
+    return openapi._aiohttp.make_openapi_schema(operations=operations, version=openapi_version)
 
 
 @pytest.fixture()
@@ -583,7 +583,7 @@ def run_and_assert(*args, **kwargs):
 
 @pytest.fixture
 def wsgi_app_factory():
-    return _flask.create_app
+    return openapi._flask.create_app
 
 
 @pytest.fixture()
@@ -593,14 +593,19 @@ def flask_app(wsgi_app_factory, operations):
 
 @pytest.fixture
 def asgi_app_factory():
-    return _fastapi.create_app
+    return openapi._fastapi.create_app
 
 
 @pytest.fixture()
 def fastapi_app(asgi_app_factory):
     return asgi_app_factory()
 
 
+@pytest.fixture()
+def fastapi_graphql_app(graphql_path):
+    return graphql._fastapi.create_app(graphql_path)
+
+
 def make_importable(module):
     """Make the package importable by the inline CLI execution."""
     pkgroot = module.dirpath()

test/test_loaders.py

@@ -1,6 +1,7 @@
 import json
 
 import pytest
+from hypothesis import given, settings
 from requests.models import Response
 from yarl import URL
 
@@ -150,3 +151,31 @@ def test_absolute_urls_for_apps(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
+
+
+def test_graphql_wsgi_loader(graphql_path, graphql_app):
+    schema = schemathesis.graphql.from_wsgi(graphql_path, app=graphql_app)
+    strategy = schema[graphql_path]["POST"].as_strategy()
+
+    @given(case=strategy)
+    @settings(max_examples=1)
+    def test(case):
+        response = case.call_wsgi()
+        assert response.status_code == 200
+        assert "data" in response.json
+
+    test()
+
+
+def test_graphql_asgi_loader(graphql_path, fastapi_graphql_app):
+    schema = schemathesis.graphql.from_asgi(graphql_path, app=fastapi_graphql_app)
+    strategy = schema[graphql_path]["POST"].as_strategy()
+
+    @given(case=strategy)
+    @settings(max_examples=1)
+    def test(case):
+        response = case.call_asgi()
+        assert response.status_code == 200
+        assert "data" in response.json()
+
+    test()