diff --git a/graphql_server/aiohttp/graphqlview.py b/graphql_server/aiohttp/graphqlview.py
index 61d2a3d..a3db1d6 100644
--- a/graphql_server/aiohttp/graphqlview.py
+++ b/graphql_server/aiohttp/graphqlview.py
@@ -75,8 +75,8 @@ def get_context(self, request):
def get_middleware(self):
return self.middleware
- # This method can be static
- async def parse_body(self, request):
+ @staticmethod
+ async def parse_body(request):
content_type = request.content_type
# request.text() is the aiohttp equivalent to
# request.body.decode("utf8")
diff --git a/graphql_server/flask/graphqlview.py b/graphql_server/flask/graphqlview.py
index 1b33433..a417406 100644
--- a/graphql_server/flask/graphqlview.py
+++ b/graphql_server/flask/graphqlview.py
@@ -139,8 +139,8 @@ def dispatch_request(self):
content_type="application/json",
)
- # Flask
- def parse_body(self):
+ @staticmethod
+ def parse_body():
# We use mimetype here since we don't need the other
# information provided by content_type
content_type = request.mimetype
@@ -164,7 +164,8 @@ def should_display_graphiql(self):
return self.request_wants_html()
- def request_wants_html(self):
+ @staticmethod
+ def request_wants_html():
best = request.accept_mimetypes.best_match(["application/json", "text/html"])
return (
best == "text/html"
diff --git a/graphql_server/quart/__init__.py b/graphql_server/quart/__init__.py
new file mode 100644
index 0000000..8f5beaf
--- /dev/null
+++ b/graphql_server/quart/__init__.py
@@ -0,0 +1,3 @@
+from .graphqlview import GraphQLView
+
+__all__ = ["GraphQLView"]
diff --git a/graphql_server/quart/graphqlview.py b/graphql_server/quart/graphqlview.py
new file mode 100644
index 0000000..9993998
--- /dev/null
+++ b/graphql_server/quart/graphqlview.py
@@ -0,0 +1,201 @@
+import copy
+import sys
+from collections.abc import MutableMapping
+from functools import partial
+from typing import List
+
+from graphql import ExecutionResult
+from graphql.error import GraphQLError
+from graphql.type.schema import GraphQLSchema
+from quart import Response, render_template_string, request
+from quart.views import View
+
+from graphql_server import (
+ GraphQLParams,
+ HttpQueryError,
+ encode_execution_results,
+ format_error_default,
+ json_encode,
+ load_json_body,
+ run_http_query,
+)
+from graphql_server.render_graphiql import (
+ GraphiQLConfig,
+ GraphiQLData,
+ GraphiQLOptions,
+ render_graphiql_sync,
+)
+
+
+class GraphQLView(View):
+ schema = None
+ root_value = None
+ context = None
+ pretty = False
+ graphiql = False
+ graphiql_version = None
+ graphiql_template = None
+ graphiql_html_title = None
+ middleware = None
+ batch = False
+ enable_async = False
+ subscriptions = None
+ headers = None
+ default_query = None
+ header_editor_enabled = None
+ should_persist_headers = None
+
+ methods = ["GET", "POST", "PUT", "DELETE"]
+
+ format_error = staticmethod(format_error_default)
+ encode = staticmethod(json_encode)
+
+ def __init__(self, **kwargs):
+ super(GraphQLView, self).__init__()
+ for key, value in kwargs.items():
+ if hasattr(self, key):
+ setattr(self, key, value)
+
+ assert isinstance(
+ self.schema, GraphQLSchema
+ ), "A Schema is required to be provided to GraphQLView."
+
+ def get_root_value(self):
+ return self.root_value
+
+ def get_context(self):
+ context = (
+ copy.copy(self.context)
+ if self.context and isinstance(self.context, MutableMapping)
+ else {}
+ )
+ if isinstance(context, MutableMapping) and "request" not in context:
+ context.update({"request": request})
+ return context
+
+ def get_middleware(self):
+ return self.middleware
+
+ async def dispatch_request(self):
+ try:
+ request_method = request.method.lower()
+ data = await self.parse_body()
+
+ show_graphiql = request_method == "get" and self.should_display_graphiql()
+ catch = show_graphiql
+
+ pretty = self.pretty or show_graphiql or request.args.get("pretty")
+ all_params: List[GraphQLParams]
+ execution_results, all_params = run_http_query(
+ self.schema,
+ request_method,
+ data,
+ query_data=request.args,
+ batch_enabled=self.batch,
+ catch=catch,
+ # Execute options
+ run_sync=not self.enable_async,
+ root_value=self.get_root_value(),
+ context_value=self.get_context(),
+ middleware=self.get_middleware(),
+ )
+ exec_res = (
+ [
+ ex if ex is None or isinstance(ex, ExecutionResult) else await ex
+ for ex in execution_results
+ ]
+ if self.enable_async
+ else execution_results
+ )
+ result, status_code = encode_execution_results(
+ exec_res,
+ is_batch=isinstance(data, list),
+ format_error=self.format_error,
+ encode=partial(self.encode, pretty=pretty), # noqa
+ )
+
+ if show_graphiql:
+ graphiql_data = GraphiQLData(
+ result=result,
+ query=getattr(all_params[0], "query"),
+ variables=getattr(all_params[0], "variables"),
+ operation_name=getattr(all_params[0], "operation_name"),
+ subscription_url=self.subscriptions,
+ headers=self.headers,
+ )
+ graphiql_config = GraphiQLConfig(
+ graphiql_version=self.graphiql_version,
+ graphiql_template=self.graphiql_template,
+ graphiql_html_title=self.graphiql_html_title,
+ jinja_env=None,
+ )
+ graphiql_options = GraphiQLOptions(
+ default_query=self.default_query,
+ header_editor_enabled=self.header_editor_enabled,
+ should_persist_headers=self.should_persist_headers,
+ )
+ source = render_graphiql_sync(
+ data=graphiql_data, config=graphiql_config, options=graphiql_options
+ )
+ return await render_template_string(source)
+
+ return Response(result, status=status_code, content_type="application/json")
+
+ except HttpQueryError as e:
+ parsed_error = GraphQLError(e.message)
+ return Response(
+ self.encode(dict(errors=[self.format_error(parsed_error)])),
+ status=e.status_code,
+ headers=e.headers,
+ content_type="application/json",
+ )
+
+ @staticmethod
+ async def parse_body():
+ # We use mimetype here since we don't need the other
+ # information provided by content_type
+ content_type = request.mimetype
+ if content_type == "application/graphql":
+ refined_data = await request.get_data(raw=False)
+ return {"query": refined_data}
+
+ elif content_type == "application/json":
+ refined_data = await request.get_data(raw=False)
+ return load_json_body(refined_data)
+
+ elif content_type == "application/x-www-form-urlencoded":
+ return await request.form
+
+ # TODO: Fix this check
+ elif content_type == "multipart/form-data":
+ return await request.files
+
+ return {}
+
+ def should_display_graphiql(self):
+ if not self.graphiql or "raw" in request.args:
+ return False
+
+ return self.request_wants_html()
+
+ @staticmethod
+ def request_wants_html():
+ best = request.accept_mimetypes.best_match(["application/json", "text/html"])
+
+ # Needed as this was introduced at Quart 0.8.0: https://gitlab.com/pgjones/quart/-/issues/189
+ def _quality(accept, key: str) -> float:
+ for option in accept.options:
+ if accept._values_match(key, option.value):
+ return option.quality
+ return 0.0
+
+ if sys.version_info >= (3, 7):
+ return (
+ best == "text/html"
+ and request.accept_mimetypes[best]
+ > request.accept_mimetypes["application/json"]
+ )
+ else:
+ return best == "text/html" and _quality(
+ request.accept_mimetypes, best
+ ) > _quality(request.accept_mimetypes, "application/json")
diff --git a/setup.py b/setup.py
index 6295b99..7afe3bc 100644
--- a/setup.py
+++ b/setup.py
@@ -38,12 +38,17 @@
"aiohttp>=3.5.0,<4",
]
+install_quart_requires = [
+ "quart>=0.6.15"
+]
+
install_all_requires = \
install_requires + \
install_flask_requires + \
install_sanic_requires + \
install_webob_requires + \
- install_aiohttp_requires
+ install_aiohttp_requires + \
+ install_quart_requires
with open("graphql_server/version.py") as version_file:
version = search('version = "(.*)"', version_file.read()).group(1)
@@ -84,6 +89,7 @@
"sanic": install_sanic_requires,
"webob": install_webob_requires,
"aiohttp": install_aiohttp_requires,
+ "quart": install_quart_requires,
},
include_package_data=True,
zip_safe=False,
diff --git a/tests/flask/app.py b/tests/flask/app.py
index 01f6fa8..ec9e9d0 100644
--- a/tests/flask/app.py
+++ b/tests/flask/app.py
@@ -5,12 +5,12 @@
def create_app(path="/graphql", **kwargs):
- app = Flask(__name__)
- app.debug = True
- app.add_url_rule(
+ server = Flask(__name__)
+ server.debug = True
+ server.add_url_rule(
path, view_func=GraphQLView.as_view("graphql", schema=Schema, **kwargs)
)
- return app
+ return server
if __name__ == "__main__":
diff --git a/tests/flask/test_graphqlview.py b/tests/flask/test_graphqlview.py
index 961a8e0..d8d60b0 100644
--- a/tests/flask/test_graphqlview.py
+++ b/tests/flask/test_graphqlview.py
@@ -9,7 +9,7 @@
@pytest.fixture
-def app(request):
+def app():
# import app factory pattern
app = create_app()
@@ -269,7 +269,7 @@ def test_supports_post_url_encoded_query_with_string_variables(app, client):
assert response_json(response) == {"data": {"test": "Hello Dolly"}}
-def test_supports_post_json_quey_with_get_variable_values(app, client):
+def test_supports_post_json_query_with_get_variable_values(app, client):
response = client.post(
url_string(app, variables=json.dumps({"who": "Dolly"})),
data=json_dump_kwarg(query="query helloWho($who: String){ test(who: $who) }",),
@@ -533,20 +533,12 @@ def test_post_multipart_data(app, client):
def test_batch_allows_post_with_json_encoding(app, client):
response = client.post(
url_string(app),
- data=json_dump_kwarg_list(
- # id=1,
- query="{test}"
- ),
+ data=json_dump_kwarg_list(query="{test}"),
content_type="application/json",
)
assert response.status_code == 200
- assert response_json(response) == [
- {
- # 'id': 1,
- "data": {"test": "Hello World"}
- }
- ]
+ assert response_json(response) == [{"data": {"test": "Hello World"}}]
@pytest.mark.parametrize("app", [create_app(batch=True)])
@@ -554,7 +546,6 @@ def test_batch_supports_post_json_query_with_json_variables(app, client):
response = client.post(
url_string(app),
data=json_dump_kwarg_list(
- # id=1,
query="query helloWho($who: String){ test(who: $who) }",
variables={"who": "Dolly"},
),
@@ -562,12 +553,7 @@ def test_batch_supports_post_json_query_with_json_variables(app, client):
)
assert response.status_code == 200
- assert response_json(response) == [
- {
- # 'id': 1,
- "data": {"test": "Hello Dolly"}
- }
- ]
+ assert response_json(response) == [{"data": {"test": "Hello Dolly"}}]
@pytest.mark.parametrize("app", [create_app(batch=True)])
@@ -575,7 +561,6 @@ def test_batch_allows_post_with_operation_name(app, client):
response = client.post(
url_string(app),
data=json_dump_kwarg_list(
- # id=1,
query="""
query helloYou { test(who: "You"), ...shared }
query helloWorld { test(who: "World"), ...shared }
@@ -591,8 +576,5 @@ def test_batch_allows_post_with_operation_name(app, client):
assert response.status_code == 200
assert response_json(response) == [
- {
- # 'id': 1,
- "data": {"test": "Hello World", "shared": "Hello Everyone"}
- }
+ {"data": {"test": "Hello World", "shared": "Hello Everyone"}}
]
diff --git a/tests/quart/__init__.py b/tests/quart/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/tests/quart/app.py b/tests/quart/app.py
new file mode 100644
index 0000000..2313f99
--- /dev/null
+++ b/tests/quart/app.py
@@ -0,0 +1,18 @@
+from quart import Quart
+
+from graphql_server.quart import GraphQLView
+from tests.quart.schema import Schema
+
+
+def create_app(path="/graphql", **kwargs):
+ server = Quart(__name__)
+ server.debug = True
+ server.add_url_rule(
+ path, view_func=GraphQLView.as_view("graphql", schema=Schema, **kwargs)
+ )
+ return server
+
+
+if __name__ == "__main__":
+ app = create_app(graphiql=True)
+ app.run()
diff --git a/tests/quart/schema.py b/tests/quart/schema.py
new file mode 100644
index 0000000..eb51e26
--- /dev/null
+++ b/tests/quart/schema.py
@@ -0,0 +1,51 @@
+from graphql.type.definition import (
+ GraphQLArgument,
+ GraphQLField,
+ GraphQLNonNull,
+ GraphQLObjectType,
+)
+from graphql.type.scalars import GraphQLString
+from graphql.type.schema import GraphQLSchema
+
+
+def resolve_raises(*_):
+ raise Exception("Throws!")
+
+
+QueryRootType = GraphQLObjectType(
+ name="QueryRoot",
+ fields={
+ "thrower": GraphQLField(GraphQLNonNull(GraphQLString), resolve=resolve_raises),
+ "request": GraphQLField(
+ GraphQLNonNull(GraphQLString),
+ resolve=lambda obj, info: info.context["request"].args.get("q"),
+ ),
+ "context": GraphQLField(
+ GraphQLObjectType(
+ name="context",
+ fields={
+ "session": GraphQLField(GraphQLString),
+ "request": GraphQLField(
+ GraphQLNonNull(GraphQLString),
+ resolve=lambda obj, info: info.context["request"],
+ ),
+ },
+ ),
+ resolve=lambda obj, info: info.context,
+ ),
+ "test": GraphQLField(
+ type_=GraphQLString,
+ args={"who": GraphQLArgument(GraphQLString)},
+ resolve=lambda obj, info, who="World": "Hello %s" % who,
+ ),
+ },
+)
+
+MutationRootType = GraphQLObjectType(
+ name="MutationRoot",
+ fields={
+ "writeTest": GraphQLField(type_=QueryRootType, resolve=lambda *_: QueryRootType)
+ },
+)
+
+Schema = GraphQLSchema(QueryRootType, MutationRootType)
diff --git a/tests/quart/test_graphiqlview.py b/tests/quart/test_graphiqlview.py
new file mode 100644
index 0000000..12b001f
--- /dev/null
+++ b/tests/quart/test_graphiqlview.py
@@ -0,0 +1,87 @@
+import sys
+
+import pytest
+from quart import Quart, Response, url_for
+from quart.testing import QuartClient
+from werkzeug.datastructures import Headers
+
+from .app import create_app
+
+
+@pytest.fixture
+def app() -> Quart:
+ # import app factory pattern
+ app = create_app(graphiql=True)
+
+ # pushes an application context manually
+ # ctx = app.app_context()
+ # await ctx.push()
+ return app
+
+
+@pytest.fixture
+def client(app: Quart) -> QuartClient:
+ return app.test_client()
+
+
+@pytest.mark.asyncio
+async def execute_client(
+ app: Quart,
+ client: QuartClient,
+ method: str = "GET",
+ headers: Headers = None,
+ **extra_params
+) -> Response:
+ if sys.version_info >= (3, 7):
+ test_request_context = app.test_request_context("/", method=method)
+ else:
+ test_request_context = app.test_request_context(method, "/")
+ async with test_request_context:
+ string = url_for("graphql", **extra_params)
+ return await client.get(string, headers=headers)
+
+
+@pytest.mark.asyncio
+async def test_graphiql_is_enabled(app: Quart, client: QuartClient):
+ response = await execute_client(
+ app, client, headers=Headers({"Accept": "text/html"}), externals=False
+ )
+ assert response.status_code == 200
+
+
+@pytest.mark.asyncio
+async def test_graphiql_renders_pretty(app: Quart, client: QuartClient):
+ response = await execute_client(
+ app, client, headers=Headers({"Accept": "text/html"}), query="{test}"
+ )
+ assert response.status_code == 200
+ pretty_response = (
+ "{\n"
+ ' "data": {\n'
+ ' "test": "Hello World"\n'
+ " }\n"
+ "}".replace('"', '\\"').replace("\n", "\\n")
+ )
+ result = await response.get_data(raw=False)
+ assert pretty_response in result
+
+
+@pytest.mark.asyncio
+async def test_graphiql_default_title(app: Quart, client: QuartClient):
+ response = await execute_client(
+ app, client, headers=Headers({"Accept": "text/html"})
+ )
+ result = await response.get_data(raw=False)
+ assert "
GraphiQL" in result
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+ "app", [create_app(graphiql=True, graphiql_html_title="Awesome")]
+)
+async def test_graphiql_custom_title(app: Quart, client: QuartClient):
+ response = await execute_client(
+ app, client, headers=Headers({"Accept": "text/html"})
+ )
+ result = await response.get_data(raw=False)
+ assert "Awesome" in result
diff --git a/tests/quart/test_graphqlview.py b/tests/quart/test_graphqlview.py
new file mode 100644
index 0000000..4a24ace
--- /dev/null
+++ b/tests/quart/test_graphqlview.py
@@ -0,0 +1,732 @@
+import json
+import sys
+
+# from io import StringIO
+from urllib.parse import urlencode
+
+import pytest
+from quart import Quart, Response, url_for
+from quart.testing import QuartClient
+from werkzeug.datastructures import Headers
+
+from .app import create_app
+
+
+@pytest.fixture
+def app() -> Quart:
+ # import app factory pattern
+ app = create_app(graphiql=True)
+
+ # pushes an application context manually
+ # ctx = app.app_context()
+ # await ctx.push()
+ return app
+
+
+@pytest.fixture
+def client(app: Quart) -> QuartClient:
+ return app.test_client()
+
+
+@pytest.mark.asyncio
+async def execute_client(
+ app: Quart,
+ client: QuartClient,
+ method: str = "GET",
+ data: str = None,
+ headers: Headers = None,
+ **url_params
+) -> Response:
+ if sys.version_info >= (3, 7):
+ test_request_context = app.test_request_context("/", method=method)
+ else:
+ test_request_context = app.test_request_context(method, "/")
+ async with test_request_context:
+ string = url_for("graphql")
+
+ if url_params:
+ string += "?" + urlencode(url_params)
+
+ if method == "POST":
+ return await client.post(string, data=data, headers=headers)
+ elif method == "PUT":
+ return await client.put(string, data=data, headers=headers)
+ else:
+ return await client.get(string)
+
+
+def response_json(result):
+ return json.loads(result)
+
+
+def json_dump_kwarg(**kwargs) -> str:
+ return json.dumps(kwargs)
+
+
+def json_dump_kwarg_list(**kwargs):
+ return json.dumps([kwargs])
+
+
+@pytest.mark.asyncio
+async def test_allows_get_with_query_param(app: Quart, client: QuartClient):
+ response = await execute_client(app, client, query="{test}")
+
+ assert response.status_code == 200
+ result = await response.get_data(raw=False)
+ assert response_json(result) == {"data": {"test": "Hello World"}}
+
+
+@pytest.mark.asyncio
+async def test_allows_get_with_variable_values(app: Quart, client: QuartClient):
+ response = await execute_client(
+ app,
+ client,
+ query="query helloWho($who: String){ test(who: $who) }",
+ variables=json.dumps({"who": "Dolly"}),
+ )
+
+ assert response.status_code == 200
+ result = await response.get_data(raw=False)
+ assert response_json(result) == {"data": {"test": "Hello Dolly"}}
+
+
+@pytest.mark.asyncio
+async def test_allows_get_with_operation_name(app: Quart, client: QuartClient):
+ response = await execute_client(
+ app,
+ client,
+ query="""
+ query helloYou { test(who: "You"), ...shared }
+ query helloWorld { test(who: "World"), ...shared }
+ query helloDolly { test(who: "Dolly"), ...shared }
+ fragment shared on QueryRoot {
+ shared: test(who: "Everyone")
+ }
+ """,
+ operationName="helloWorld",
+ )
+
+ assert response.status_code == 200
+ result = await response.get_data(raw=False)
+ assert response_json(result) == {
+ "data": {"test": "Hello World", "shared": "Hello Everyone"}
+ }
+
+
+@pytest.mark.asyncio
+async def test_reports_validation_errors(app: Quart, client: QuartClient):
+ response = await execute_client(
+ app, client, query="{ test, unknownOne, unknownTwo }"
+ )
+
+ assert response.status_code == 400
+ result = await response.get_data(raw=False)
+ assert response_json(result) == {
+ "errors": [
+ {
+ "message": "Cannot query field 'unknownOne' on type 'QueryRoot'.",
+ "locations": [{"line": 1, "column": 9}],
+ "path": None,
+ },
+ {
+ "message": "Cannot query field 'unknownTwo' on type 'QueryRoot'.",
+ "locations": [{"line": 1, "column": 21}],
+ "path": None,
+ },
+ ]
+ }
+
+
+@pytest.mark.asyncio
+async def test_errors_when_missing_operation_name(app: Quart, client: QuartClient):
+ response = await execute_client(
+ app,
+ client,
+ query="""
+ query TestQuery { test }
+ mutation TestMutation { writeTest { test } }
+ """,
+ )
+
+ assert response.status_code == 400
+ result = await response.get_data(raw=False)
+ assert response_json(result) == {
+ "errors": [
+ {
+ "message": "Must provide operation name if query contains multiple operations.", # noqa: E501
+ "locations": None,
+ "path": None,
+ }
+ ]
+ }
+
+
+@pytest.mark.asyncio
+async def test_errors_when_sending_a_mutation_via_get(app: Quart, client: QuartClient):
+ response = await execute_client(
+ app,
+ client,
+ query="""
+ mutation TestMutation { writeTest { test } }
+ """,
+ )
+ assert response.status_code == 405
+ result = await response.get_data(raw=False)
+ assert response_json(result) == {
+ "errors": [
+ {
+ "message": "Can only perform a mutation operation from a POST request.",
+ "locations": None,
+ "path": None,
+ }
+ ]
+ }
+
+
+@pytest.mark.asyncio
+async def test_errors_when_selecting_a_mutation_within_a_get(
+ app: Quart, client: QuartClient
+):
+ response = await execute_client(
+ app,
+ client,
+ query="""
+ query TestQuery { test }
+ mutation TestMutation { writeTest { test } }
+ """,
+ operationName="TestMutation",
+ )
+
+ assert response.status_code == 405
+ result = await response.get_data(raw=False)
+ assert response_json(result) == {
+ "errors": [
+ {
+ "message": "Can only perform a mutation operation from a POST request.",
+ "locations": None,
+ "path": None,
+ }
+ ]
+ }
+
+
+@pytest.mark.asyncio
+async def test_allows_mutation_to_exist_within_a_get(app: Quart, client: QuartClient):
+ response = await execute_client(
+ app,
+ client,
+ query="""
+ query TestQuery { test }
+ mutation TestMutation { writeTest { test } }
+ """,
+ operationName="TestQuery",
+ )
+
+ assert response.status_code == 200
+ result = await response.get_data(raw=False)
+ assert response_json(result) == {"data": {"test": "Hello World"}}
+
+
+@pytest.mark.asyncio
+async def test_allows_post_with_json_encoding(app: Quart, client: QuartClient):
+ response = await execute_client(
+ app,
+ client,
+ method="POST",
+ data=json_dump_kwarg(query="{test}"),
+ headers=Headers({"Content-Type": "application/json"}),
+ )
+
+ assert response.status_code == 200
+ result = await response.get_data(raw=False)
+ assert response_json(result) == {"data": {"test": "Hello World"}}
+
+
+@pytest.mark.asyncio
+async def test_allows_sending_a_mutation_via_post(app: Quart, client: QuartClient):
+ response = await execute_client(
+ app,
+ client,
+ method="POST",
+ data=json_dump_kwarg(query="mutation TestMutation { writeTest { test } }"),
+ headers=Headers({"Content-Type": "application/json"}),
+ )
+
+ assert response.status_code == 200
+ result = await response.get_data(raw=False)
+ assert response_json(result) == {"data": {"writeTest": {"test": "Hello World"}}}
+
+
+@pytest.mark.asyncio
+async def test_allows_post_with_url_encoding(app: Quart, client: QuartClient):
+ response = await execute_client(
+ app,
+ client,
+ method="POST",
+ data=urlencode(dict(query="{test}")),
+ headers=Headers({"Content-Type": "application/x-www-form-urlencoded"}),
+ )
+
+ assert response.status_code == 200
+ result = await response.get_data(raw=False)
+ assert response_json(result) == {"data": {"test": "Hello World"}}
+
+
+@pytest.mark.asyncio
+async def test_supports_post_json_query_with_string_variables(
+ app: Quart, client: QuartClient
+):
+ response = await execute_client(
+ app,
+ client,
+ method="POST",
+ data=json_dump_kwarg(
+ query="query helloWho($who: String){ test(who: $who) }",
+ variables=json.dumps({"who": "Dolly"}),
+ ),
+ headers=Headers({"Content-Type": "application/json"}),
+ )
+
+ assert response.status_code == 200
+ result = await response.get_data(raw=False)
+ assert response_json(result) == {"data": {"test": "Hello Dolly"}}
+
+
+@pytest.mark.asyncio
+async def test_supports_post_json_query_with_json_variables(
+ app: Quart, client: QuartClient
+):
+ response = await execute_client(
+ app,
+ client,
+ method="POST",
+ data=json_dump_kwarg(
+ query="query helloWho($who: String){ test(who: $who) }",
+ variables={"who": "Dolly"},
+ ),
+ headers=Headers({"Content-Type": "application/json"}),
+ )
+
+ assert response.status_code == 200
+ result = await response.get_data(raw=False)
+ assert response_json(result) == {"data": {"test": "Hello Dolly"}}
+
+
+@pytest.mark.asyncio
+async def test_supports_post_url_encoded_query_with_string_variables(
+ app: Quart, client: QuartClient
+):
+ response = await execute_client(
+ app,
+ client,
+ method="POST",
+ data=urlencode(
+ dict(
+ query="query helloWho($who: String){ test(who: $who) }",
+ variables=json.dumps({"who": "Dolly"}),
+ )
+ ),
+ headers=Headers({"Content-Type": "application/x-www-form-urlencoded"}),
+ )
+
+ assert response.status_code == 200
+ result = await response.get_data(raw=False)
+ assert response_json(result) == {"data": {"test": "Hello Dolly"}}
+
+
+@pytest.mark.asyncio
+async def test_supports_post_json_query_with_get_variable_values(
+ app: Quart, client: QuartClient
+):
+ response = await execute_client(
+ app,
+ client,
+ method="POST",
+ data=json_dump_kwarg(query="query helloWho($who: String){ test(who: $who) }",),
+ headers=Headers({"Content-Type": "application/json"}),
+ variables=json.dumps({"who": "Dolly"}),
+ )
+
+ assert response.status_code == 200
+ result = await response.get_data(raw=False)
+ assert response_json(result) == {"data": {"test": "Hello Dolly"}}
+
+
+@pytest.mark.asyncio
+async def test_post_url_encoded_query_with_get_variable_values(
+ app: Quart, client: QuartClient
+):
+ response = await execute_client(
+ app,
+ client,
+ method="POST",
+ data=urlencode(dict(query="query helloWho($who: String){ test(who: $who) }",)),
+ headers=Headers({"Content-Type": "application/x-www-form-urlencoded"}),
+ variables=json.dumps({"who": "Dolly"}),
+ )
+
+ assert response.status_code == 200
+ result = await response.get_data(raw=False)
+ assert response_json(result) == {"data": {"test": "Hello Dolly"}}
+
+
+@pytest.mark.asyncio
+async def test_supports_post_raw_text_query_with_get_variable_values(
+ app: Quart, client: QuartClient
+):
+ response = await execute_client(
+ app,
+ client=client,
+ method="POST",
+ data="query helloWho($who: String){ test(who: $who) }",
+ headers=Headers({"Content-Type": "application/graphql"}),
+ variables=json.dumps({"who": "Dolly"}),
+ )
+
+ assert response.status_code == 200
+ result = await response.get_data(raw=False)
+ assert response_json(result) == {"data": {"test": "Hello Dolly"}}
+
+
+@pytest.mark.asyncio
+async def test_allows_post_with_operation_name(app: Quart, client: QuartClient):
+ response = await execute_client(
+ app,
+ client,
+ method="POST",
+ data=json_dump_kwarg(
+ query="""
+ query helloYou { test(who: "You"), ...shared }
+ query helloWorld { test(who: "World"), ...shared }
+ query helloDolly { test(who: "Dolly"), ...shared }
+ fragment shared on QueryRoot {
+ shared: test(who: "Everyone")
+ }
+ """,
+ operationName="helloWorld",
+ ),
+ headers=Headers({"Content-Type": "application/json"}),
+ )
+
+ assert response.status_code == 200
+ result = await response.get_data(raw=False)
+ assert response_json(result) == {
+ "data": {"test": "Hello World", "shared": "Hello Everyone"}
+ }
+
+
+@pytest.mark.asyncio
+async def test_allows_post_with_get_operation_name(app: Quart, client: QuartClient):
+ response = await execute_client(
+ app,
+ client,
+ method="POST",
+ data="""
+ query helloYou { test(who: "You"), ...shared }
+ query helloWorld { test(who: "World"), ...shared }
+ query helloDolly { test(who: "Dolly"), ...shared }
+ fragment shared on QueryRoot {
+ shared: test(who: "Everyone")
+ }
+ """,
+ headers=Headers({"Content-Type": "application/graphql"}),
+ operationName="helloWorld",
+ )
+
+ assert response.status_code == 200
+ result = await response.get_data(raw=False)
+ assert response_json(result) == {
+ "data": {"test": "Hello World", "shared": "Hello Everyone"}
+ }
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("app", [create_app(pretty=True)])
+async def test_supports_pretty_printing(app: Quart, client: QuartClient):
+ response = await execute_client(app, client, query="{test}")
+
+ result = await response.get_data(raw=False)
+ assert result == ("{\n" ' "data": {\n' ' "test": "Hello World"\n' " }\n" "}")
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("app", [create_app(pretty=False)])
+async def test_not_pretty_by_default(app: Quart, client: QuartClient):
+ response = await execute_client(app, client, query="{test}")
+
+ result = await response.get_data(raw=False)
+ assert result == '{"data":{"test":"Hello World"}}'
+
+
+@pytest.mark.asyncio
+async def test_supports_pretty_printing_by_request(app: Quart, client: QuartClient):
+ response = await execute_client(app, client, query="{test}", pretty="1")
+
+ result = await response.get_data(raw=False)
+ assert result == ("{\n" ' "data": {\n' ' "test": "Hello World"\n' " }\n" "}")
+
+
+@pytest.mark.asyncio
+async def test_handles_field_errors_caught_by_graphql(app: Quart, client: QuartClient):
+ response = await execute_client(app, client, query="{thrower}")
+ assert response.status_code == 200
+ result = await response.get_data(raw=False)
+ assert response_json(result) == {
+ "errors": [
+ {
+ "locations": [{"column": 2, "line": 1}],
+ "path": ["thrower"],
+ "message": "Throws!",
+ }
+ ],
+ "data": None,
+ }
+
+
+@pytest.mark.asyncio
+async def test_handles_syntax_errors_caught_by_graphql(app: Quart, client: QuartClient):
+ response = await execute_client(app, client, query="syntaxerror")
+ assert response.status_code == 400
+ result = await response.get_data(raw=False)
+ assert response_json(result) == {
+ "errors": [
+ {
+ "locations": [{"column": 1, "line": 1}],
+ "message": "Syntax Error: Unexpected Name 'syntaxerror'.",
+ "path": None,
+ }
+ ]
+ }
+
+
+@pytest.mark.asyncio
+async def test_handles_errors_caused_by_a_lack_of_query(
+ app: Quart, client: QuartClient
+):
+ response = await execute_client(app, client)
+
+ assert response.status_code == 400
+ result = await response.get_data(raw=False)
+ assert response_json(result) == {
+ "errors": [
+ {"message": "Must provide query string.", "locations": None, "path": None}
+ ]
+ }
+
+
+@pytest.mark.asyncio
+async def test_handles_batch_correctly_if_is_disabled(app: Quart, client: QuartClient):
+ response = await execute_client(
+ app,
+ client,
+ method="POST",
+ data="[]",
+ headers=Headers({"Content-Type": "application/json"}),
+ )
+
+ assert response.status_code == 400
+ result = await response.get_data(raw=False)
+ assert response_json(result) == {
+ "errors": [
+ {
+ "message": "Batch GraphQL requests are not enabled.",
+ "locations": None,
+ "path": None,
+ }
+ ]
+ }
+
+
+@pytest.mark.asyncio
+async def test_handles_incomplete_json_bodies(app: Quart, client: QuartClient):
+ response = await execute_client(
+ app,
+ client,
+ method="POST",
+ data='{"query":',
+ headers=Headers({"Content-Type": "application/json"}),
+ )
+
+ assert response.status_code == 400
+ result = await response.get_data(raw=False)
+ assert response_json(result) == {
+ "errors": [
+ {"message": "POST body sent invalid JSON.", "locations": None, "path": None}
+ ]
+ }
+
+
+@pytest.mark.asyncio
+async def test_handles_plain_post_text(app: Quart, client: QuartClient):
+ response = await execute_client(
+ app,
+ client,
+ method="POST",
+ data="query helloWho($who: String){ test(who: $who) }",
+ headers=Headers({"Content-Type": "text/plain"}),
+ variables=json.dumps({"who": "Dolly"}),
+ )
+ assert response.status_code == 400
+ result = await response.get_data(raw=False)
+ assert response_json(result) == {
+ "errors": [
+ {"message": "Must provide query string.", "locations": None, "path": None}
+ ]
+ }
+
+
+@pytest.mark.asyncio
+async def test_handles_poorly_formed_variables(app: Quart, client: QuartClient):
+ response = await execute_client(
+ app,
+ client,
+ query="query helloWho($who: String){ test(who: $who) }",
+ variables="who:You",
+ )
+ assert response.status_code == 400
+ result = await response.get_data(raw=False)
+ assert response_json(result) == {
+ "errors": [
+ {"message": "Variables are invalid JSON.", "locations": None, "path": None}
+ ]
+ }
+
+
+@pytest.mark.asyncio
+async def test_handles_unsupported_http_methods(app: Quart, client: QuartClient):
+ response = await execute_client(app, client, method="PUT", query="{test}")
+ assert response.status_code == 405
+ result = await response.get_data(raw=False)
+ assert response.headers["Allow"] in ["GET, POST", "HEAD, GET, POST, OPTIONS"]
+ assert response_json(result) == {
+ "errors": [
+ {
+ "message": "GraphQL only supports GET and POST requests.",
+ "locations": None,
+ "path": None,
+ }
+ ]
+ }
+
+
+@pytest.mark.asyncio
+async def test_passes_request_into_request_context(app: Quart, client: QuartClient):
+ response = await execute_client(app, client, query="{request}", q="testing")
+
+ assert response.status_code == 200
+ result = await response.get_data(raw=False)
+ assert response_json(result) == {"data": {"request": "testing"}}
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("app", [create_app(context={"session": "CUSTOM CONTEXT"})])
+async def test_passes_custom_context_into_context(app: Quart, client: QuartClient):
+ response = await execute_client(app, client, query="{context { session request }}")
+
+ assert response.status_code == 200
+ result = await response.get_data(raw=False)
+ res = response_json(result)
+ assert "data" in res
+ assert "session" in res["data"]["context"]
+ assert "request" in res["data"]["context"]
+ assert "CUSTOM CONTEXT" in res["data"]["context"]["session"]
+ assert "Request" in res["data"]["context"]["request"]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("app", [create_app(context="CUSTOM CONTEXT")])
+async def test_context_remapped_if_not_mapping(app: Quart, client: QuartClient):
+ response = await execute_client(app, client, query="{context { session request }}")
+
+ assert response.status_code == 200
+ result = await response.get_data(raw=False)
+ res = response_json(result)
+ assert "data" in res
+ assert "session" in res["data"]["context"]
+ assert "request" in res["data"]["context"]
+ assert "CUSTOM CONTEXT" not in res["data"]["context"]["request"]
+ assert "Request" in res["data"]["context"]["request"]
+
+
+# @pytest.mark.asyncio
+# async def test_post_multipart_data(app: Quart, client: QuartClient):
+# query = "mutation TestMutation { writeTest { test } }"
+# response = await execute_client(
+# app,
+# client,
+# method='POST',
+# data={"query": query, "file": (StringIO(), "text1.txt")},
+# headers=Headers({"Content-Type": "multipart/form-data"})
+# )
+#
+# assert response.status_code == 200
+# result = await response.get_data()
+# assert response_json(result) == {
+# "data": {u"writeTest": {u"test": u"Hello World"}}
+# }
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("app", [create_app(batch=True)])
+async def test_batch_allows_post_with_json_encoding(app: Quart, client: QuartClient):
+ response = await execute_client(
+ app,
+ client,
+ method="POST",
+ data=json_dump_kwarg_list(query="{test}"),
+ headers=Headers({"Content-Type": "application/json"}),
+ )
+
+ assert response.status_code == 200
+ result = await response.get_data(raw=False)
+ assert response_json(result) == [{"data": {"test": "Hello World"}}]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("app", [create_app(batch=True)])
+async def test_batch_supports_post_json_query_with_json_variables(
+ app: Quart, client: QuartClient
+):
+ response = await execute_client(
+ app,
+ client,
+ method="POST",
+ data=json_dump_kwarg_list(
+ query="query helloWho($who: String){ test(who: $who) }",
+ variables={"who": "Dolly"},
+ ),
+ headers=Headers({"Content-Type": "application/json"}),
+ )
+
+ assert response.status_code == 200
+ result = await response.get_data(raw=False)
+ assert response_json(result) == [{"data": {"test": "Hello Dolly"}}]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("app", [create_app(batch=True)])
+async def test_batch_allows_post_with_operation_name(app: Quart, client: QuartClient):
+ response = await execute_client(
+ app,
+ client,
+ method="POST",
+ data=json_dump_kwarg_list(
+ # id=1,
+ query="""
+ query helloYou { test(who: "You"), ...shared }
+ query helloWorld { test(who: "World"), ...shared }
+ query helloDolly { test(who: "Dolly"), ...shared }
+ fragment shared on QueryRoot {
+ shared: test(who: "Everyone")
+ }
+ """,
+ operationName="helloWorld",
+ ),
+ headers=Headers({"Content-Type": "application/json"}),
+ )
+
+ assert response.status_code == 200
+ result = await response.get_data(raw=False)
+ assert response_json(result) == [
+ {"data": {"test": "Hello World", "shared": "Hello Everyone"}}
+ ]