1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
|
from gc import collect
from inspect import isawaitable
from pytest import mark, raises
from graphql import graphql_sync
from graphql.execution import execute, execute_sync
from graphql.language import parse
from graphql.type import GraphQLField, GraphQLObjectType, GraphQLSchema, GraphQLString
from graphql.validation import validate
def describe_execute_synchronously_when_possible():
def _resolve_sync(root_value, _info):
return root_value
async def _resolve_async(root_value, _info):
return root_value
schema = GraphQLSchema(
GraphQLObjectType(
"Query",
{
"syncField": GraphQLField(GraphQLString, resolve=_resolve_sync),
"asyncField": GraphQLField(GraphQLString, resolve=_resolve_async),
},
),
GraphQLObjectType(
"Mutation",
{"syncMutationField": GraphQLField(GraphQLString, resolve=_resolve_sync)},
),
)
def does_not_return_an_awaitable_for_initial_errors():
doc = "fragment Example on Query { syncField }"
assert execute(schema, parse(doc), "rootValue") == (
None,
[{"message": "Must provide an operation."}],
)
def does_not_return_an_awaitable_if_fields_are_all_synchronous():
doc = "query Example { syncField }"
assert execute(schema, parse(doc), "rootValue") == (
{"syncField": "rootValue"},
None,
)
def does_not_return_an_awaitable_if_mutation_fields_are_all_synchronous():
doc = "mutation Example { syncMutationField }"
assert execute(schema, parse(doc), "rootValue") == (
{"syncMutationField": "rootValue"},
None,
)
@mark.asyncio
async def returns_an_awaitable_if_any_field_is_asynchronous():
doc = "query Example { syncField, asyncField }"
result = execute(schema, parse(doc), "rootValue")
assert isawaitable(result)
assert await result == (
{"syncField": "rootValue", "asyncField": "rootValue"},
None,
)
def describe_execute_sync():
def does_not_return_an_awaitable_for_sync_execution():
doc = "query Example { syncField }"
result = execute_sync(schema, document=parse(doc), root_value="rootValue")
assert result == (
{"syncField": "rootValue"},
None,
)
def does_not_throw_if_not_encountering_async_execution_with_check_sync():
doc = "query Example { syncField }"
result = execute_sync(
schema, document=parse(doc), root_value="rootValue", check_sync=True
)
assert result == (
{"syncField": "rootValue"},
None,
)
@mark.asyncio
@mark.filterwarnings("ignore:.* was never awaited:RuntimeWarning")
async def throws_if_encountering_async_execution_with_check_sync():
doc = "query Example { syncField, asyncField }"
with raises(RuntimeError) as exc_info:
execute_sync(
schema, document=parse(doc), root_value="rootValue", check_sync=True
)
msg = str(exc_info.value)
assert msg == "GraphQL execution failed to complete synchronously."
@mark.asyncio
@mark.filterwarnings("ignore:.* was never awaited:RuntimeWarning")
async def throws_if_encountering_async_operation_without_check_sync():
doc = "query Example { syncField, asyncField }"
result = execute_sync(schema, document=parse(doc), root_value="rootValue")
assert result == (
{"syncField": "rootValue", "asyncField": None},
[
{
"message": "String cannot represent value:"
" <coroutine _resolve_async>",
"locations": [(1, 28)],
"path": ["asyncField"],
}
],
)
# garbage collect coroutine in order to not postpone the warning
del result
collect()
def describe_graphql_sync():
def reports_errors_raised_during_schema_validation():
bad_schema = GraphQLSchema()
result = graphql_sync(schema=bad_schema, source="{ __typename }")
assert result == (None, [{"message": "Query root type must be provided."}])
def does_not_return_an_awaitable_for_syntax_errors():
doc = "fragment Example on Query { { { syncField }"
assert graphql_sync(schema, doc) == (
None,
[
{
"message": "Syntax Error: Expected Name, found '{'.",
"locations": [(1, 29)],
}
],
)
def does_not_return_an_awaitable_for_validation_errors():
doc = "fragment Example on Query { unknownField }"
validation_errors = validate(schema, parse(doc))
result = graphql_sync(schema, doc)
assert result == (None, validation_errors)
def does_not_return_an_awaitable_for_sync_execution():
doc = "query Example { syncField }"
assert graphql_sync(schema, doc, "rootValue") == (
{"syncField": "rootValue"},
None,
)
def does_not_throw_if_not_encountering_async_operation_with_check_sync():
doc = "query Example { syncField }"
assert graphql_sync(schema, doc, "rootValue") == (
{"syncField": "rootValue"},
None,
)
@mark.asyncio
@mark.filterwarnings("ignore:.* was never awaited:RuntimeWarning")
async def throws_if_encountering_async_operation_with_check_sync():
doc = "query Example { syncField, asyncField }"
with raises(RuntimeError) as exc_info:
graphql_sync(schema, doc, "rootValue", check_sync=True)
msg = str(exc_info.value)
assert msg == "GraphQL execution failed to complete synchronously."
@mark.asyncio
@mark.filterwarnings("ignore:.* was never awaited:RuntimeWarning")
async def throws_if_encountering_async_operation_without_check_sync():
doc = "query Example { syncField, asyncField }"
result = graphql_sync(schema, doc, "rootValue")
assert result == (
{"syncField": "rootValue", "asyncField": None},
[
{
"message": "String cannot represent value:"
" <coroutine _resolve_async>",
"locations": [(1, 28)],
"path": ["asyncField"],
}
],
)
# garbage collect coroutine in order to not postpone the warning
del result
collect()
|
