5
5
from graphql .language .parser import parse
6
6
from graphql .type import GraphQLField , GraphQLObjectType , GraphQLSchema , GraphQLString
7
7
8
-
9
- def describe_middleware ():
10
- def describe_with_manager ():
11
- def default ():
8
+ def _create_schema (tp : GraphQLObjectType , is_subscription : bool ) -> GraphQLSchema :
9
+ if is_subscription :
10
+ noop_type = GraphQLObjectType ("Noop" , {"noop" : GraphQLField (GraphQLString , resolve = lambda * _ : "noop" )})
11
+ return GraphQLSchema (query = noop_type , subscription = tp )
12
+ return GraphQLSchema (tp )
13
+ @pytest .mark .parametrize ("is_subscription" , [False , True ], ids = ["query" , "subscription" ])
14
+ def test_describe_middleware (is_subscription : bool ):
15
+
16
+ def test_test_describe_with_manager ():
17
+ def test_default ():
12
18
doc = parse ("{ field }" )
13
19
14
20
# noinspection PyMethodMayBeStatic
15
21
class Data :
16
- def field (self , _info ):
22
+ def test_field (self , _info ):
17
23
return "resolved"
18
24
19
25
test_type = GraphQLObjectType (
@@ -22,20 +28,20 @@ def field(self, _info):
22
28
23
29
middlewares = MiddlewareManager ()
24
30
result = execute (
25
- GraphQLSchema (test_type ), doc , Data (), middleware = middlewares
31
+ _create_schema (test_type , is_subscription ), doc , Data (), middleware = middlewares
26
32
)
27
33
28
34
assert result .data ["field" ] == "resolved" # type: ignore
29
35
30
- def single_function ():
36
+ def test_single_function ():
31
37
doc = parse ("{ first second }" )
32
38
33
39
# noinspection PyMethodMayBeStatic
34
40
class Data :
35
- def first (self , _info ):
41
+ def test_first (self , _info ):
36
42
return "one"
37
43
38
- def second (self , _info ):
44
+ def test_second (self , _info ):
39
45
return "two"
40
46
41
47
test_type = GraphQLObjectType (
@@ -51,12 +57,12 @@ def reverse_middleware(next_, *args, **kwargs):
51
57
52
58
middlewares = MiddlewareManager (reverse_middleware )
53
59
result = execute (
54
- GraphQLSchema (test_type ), doc , Data (), middleware = middlewares
60
+ _create_schema (test_type , is_subscription ), doc , Data (), middleware = middlewares
55
61
)
56
62
57
63
assert result .data == {"first" : "eno" , "second" : "owt" } # type: ignore
58
64
59
- def two_functions_and_field_resolvers ():
65
+ def test_two_functions_and_field_resolvers ():
60
66
doc = parse ("{ first second }" )
61
67
62
68
# noinspection PyMethodMayBeStatic
@@ -84,21 +90,21 @@ def capitalize_middleware(next_, *args, **kwargs):
84
90
85
91
middlewares = MiddlewareManager (reverse_middleware , capitalize_middleware )
86
92
result = execute (
87
- GraphQLSchema (test_type ), doc , Data (), middleware = middlewares
93
+ _create_schema (test_type , is_subscription ), doc , Data (), middleware = middlewares
88
94
)
89
95
90
96
assert result .data == {"first" : "Eno" , "second" : "Owt" } # type: ignore
91
97
92
98
@pytest .mark .asyncio ()
93
- async def single_async_function ():
99
+ async def test_single_async_function ():
94
100
doc = parse ("{ first second }" )
95
101
96
102
# noinspection PyMethodMayBeStatic
97
103
class Data :
98
- async def first (self , _info ):
104
+ async def test_first (self , _info ):
99
105
return "one"
100
106
101
- async def second (self , _info ):
107
+ async def test_second (self , _info ):
102
108
return "two"
103
109
104
110
test_type = GraphQLObjectType (
@@ -114,21 +120,21 @@ async def reverse_middleware(next_, *args, **kwargs):
114
120
115
121
middlewares = MiddlewareManager (reverse_middleware )
116
122
awaitable_result = execute (
117
- GraphQLSchema (test_type ), doc , Data (), middleware = middlewares
123
+ _create_schema (test_type , is_subscription ), doc , Data (), middleware = middlewares
118
124
)
119
125
assert isinstance (awaitable_result , Awaitable )
120
126
result = await awaitable_result
121
127
assert result .data == {"first" : "eno" , "second" : "owt" }
122
128
123
- def single_object ():
129
+ def test_single_object ():
124
130
doc = parse ("{ first second }" )
125
131
126
132
# noinspection PyMethodMayBeStatic
127
133
class Data :
128
- def first (self , _info ):
134
+ def test_first (self , _info ):
129
135
return "one"
130
136
131
- def second (self , _info ):
137
+ def test_second (self , _info ):
132
138
return "two"
133
139
134
140
test_type = GraphQLObjectType (
@@ -141,17 +147,17 @@ def second(self, _info):
141
147
142
148
class ReverseMiddleware :
143
149
# noinspection PyMethodMayBeStatic
144
- def resolve (self , next_ , * args , ** kwargs ):
150
+ def test_resolve (self , next_ , * args , ** kwargs ):
145
151
return next_ (* args , ** kwargs )[::- 1 ]
146
152
147
153
middlewares = MiddlewareManager (ReverseMiddleware ())
148
154
result = execute (
149
- GraphQLSchema (test_type ), doc , Data (), middleware = middlewares
155
+ _create_schema (test_type , is_subscription ), doc , Data (), middleware = middlewares
150
156
)
151
157
152
158
assert result .data == {"first" : "eno" , "second" : "owt" } # type: ignore
153
159
154
- def skip_middleware_without_resolve_method ():
160
+ def test_skip_middleware_without_resolve_method ():
155
161
class BadMiddleware :
156
162
pass # no resolve method here
157
163
@@ -167,12 +173,12 @@ class BadMiddleware:
167
173
middleware = MiddlewareManager (BadMiddleware ()),
168
174
) == ({"foo" : "bar" }, None )
169
175
170
- def with_function_and_object ():
176
+ def test_with_function_and_object ():
171
177
doc = parse ("{ field }" )
172
178
173
179
# noinspection PyMethodMayBeStatic
174
180
class Data :
175
- def field (self , _info ):
181
+ def test_field (self , _info ):
176
182
return "resolved"
177
183
178
184
test_type = GraphQLObjectType (
@@ -184,28 +190,28 @@ def reverse_middleware(next_, *args, **kwargs):
184
190
185
191
class CaptitalizeMiddleware :
186
192
# noinspection PyMethodMayBeStatic
187
- def resolve (self , next_ , * args , ** kwargs ):
193
+ def test_resolve (self , next_ , * args , ** kwargs ):
188
194
return next_ (* args , ** kwargs ).capitalize ()
189
195
190
196
middlewares = MiddlewareManager (reverse_middleware , CaptitalizeMiddleware ())
191
197
result = execute (
192
- GraphQLSchema (test_type ), doc , Data (), middleware = middlewares
198
+ _create_schema (test_type , is_subscription ), doc , Data (), middleware = middlewares
193
199
)
194
200
assert result .data == {"field" : "Devloser" } # type: ignore
195
201
196
202
middlewares = MiddlewareManager (CaptitalizeMiddleware (), reverse_middleware )
197
203
result = execute (
198
- GraphQLSchema (test_type ), doc , Data (), middleware = middlewares
204
+ _create_schema (test_type , is_subscription ), doc , Data (), middleware = middlewares
199
205
)
200
206
assert result .data == {"field" : "devloseR" } # type: ignore
201
207
202
208
@pytest .mark .asyncio ()
203
- async def with_async_function_and_object ():
209
+ async def test_with_async_function_and_object ():
204
210
doc = parse ("{ field }" )
205
211
206
212
# noinspection PyMethodMayBeStatic
207
213
class Data :
208
- async def field (self , _info ):
214
+ async def test_field (self , _info ):
209
215
return "resolved"
210
216
211
217
test_type = GraphQLObjectType (
@@ -217,59 +223,59 @@ async def reverse_middleware(next_, *args, **kwargs):
217
223
218
224
class CaptitalizeMiddleware :
219
225
# noinspection PyMethodMayBeStatic
220
- async def resolve (self , next_ , * args , ** kwargs ):
226
+ async def test_resolve (self , next_ , * args , ** kwargs ):
221
227
return (await next_ (* args , ** kwargs )).capitalize ()
222
228
223
229
middlewares = MiddlewareManager (reverse_middleware , CaptitalizeMiddleware ())
224
230
awaitable_result = execute (
225
- GraphQLSchema (test_type ), doc , Data (), middleware = middlewares
231
+ _create_schema (test_type , is_subscription ), doc , Data (), middleware = middlewares
226
232
)
227
233
assert isinstance (awaitable_result , Awaitable )
228
234
result = await awaitable_result
229
235
assert result .data == {"field" : "Devloser" }
230
236
231
237
middlewares = MiddlewareManager (CaptitalizeMiddleware (), reverse_middleware )
232
238
awaitable_result = execute (
233
- GraphQLSchema (test_type ), doc , Data (), middleware = middlewares
239
+ _create_schema (test_type , is_subscription ), doc , Data (), middleware = middlewares
234
240
)
235
241
assert isinstance (awaitable_result , Awaitable )
236
242
result = await awaitable_result
237
243
assert result .data == {"field" : "devloseR" }
238
244
239
- def describe_without_manager ():
240
- def no_middleware ():
245
+ def test_describe_without_manager ():
246
+ def test_no_middleware ():
241
247
doc = parse ("{ field }" )
242
248
243
249
# noinspection PyMethodMayBeStatic
244
250
class Data :
245
- def field (self , _info ):
251
+ def test_field (self , _info ):
246
252
return "resolved"
247
253
248
254
test_type = GraphQLObjectType (
249
255
"TestType" , {"field" : GraphQLField (GraphQLString )}
250
256
)
251
257
252
- result = execute (GraphQLSchema (test_type ), doc , Data (), middleware = None )
258
+ result = execute (_create_schema (test_type , is_subscription ), doc , Data (), middleware = None )
253
259
254
260
assert result .data ["field" ] == "resolved" # type: ignore
255
261
256
- def empty_middleware_list ():
262
+ def test_empty_middleware_list ():
257
263
doc = parse ("{ field }" )
258
264
259
265
# noinspection PyMethodMayBeStatic
260
266
class Data :
261
- def field (self , _info ):
267
+ def test_field (self , _info ):
262
268
return "resolved"
263
269
264
270
test_type = GraphQLObjectType (
265
271
"TestType" , {"field" : GraphQLField (GraphQLString )}
266
272
)
267
273
268
- result = execute (GraphQLSchema (test_type ), doc , Data (), middleware = [])
274
+ result = execute (_create_schema (test_type , is_subscription ), doc , Data (), middleware = [])
269
275
270
276
assert result .data ["field" ] == "resolved" # type: ignore
271
277
272
- def bad_middleware_object ():
278
+ def test_bad_middleware_object ():
273
279
doc = parse ("{ field }" )
274
280
275
281
test_type = GraphQLObjectType (
@@ -279,7 +285,7 @@ def bad_middleware_object():
279
285
with pytest .raises (TypeError ) as exc_info :
280
286
# noinspection PyTypeChecker
281
287
execute (
282
- GraphQLSchema (test_type ),
288
+ _create_schema (test_type , is_subscription ),
283
289
doc ,
284
290
None ,
285
291
middleware = cast (Middleware , {"bad" : "value" }),
@@ -291,12 +297,12 @@ def bad_middleware_object():
291
297
" Got {'bad': 'value'} instead."
292
298
)
293
299
294
- def list_of_functions ():
300
+ def test_list_of_functions ():
295
301
doc = parse ("{ field }" )
296
302
297
303
# noinspection PyMethodMayBeStatic
298
304
class Data :
299
- def field (self , _info ):
305
+ def test_field (self , _info ):
300
306
return "resolved"
301
307
302
308
test_type = GraphQLObjectType (
@@ -306,11 +312,11 @@ def field(self, _info):
306
312
log = []
307
313
308
314
class LogMiddleware :
309
- def __init__ (self , name ):
315
+ def test___init__ (self , name ):
310
316
self .name = name
311
317
312
318
# noinspection PyMethodMayBeStatic
313
- def resolve (self , next_ , * args , ** kwargs ):
319
+ def test_resolve (self , next_ , * args , ** kwargs ):
314
320
log .append (f"enter { self .name } " )
315
321
value = next_ (* args , ** kwargs )
316
322
log .append (f"exit { self .name } " )
@@ -319,7 +325,7 @@ def resolve(self, next_, *args, **kwargs):
319
325
middlewares = [LogMiddleware ("A" ), LogMiddleware ("B" ), LogMiddleware ("C" )]
320
326
321
327
result = execute (
322
- GraphQLSchema (test_type ), doc , Data (), middleware = middlewares
328
+ _create_schema (test_type , is_subscription ), doc , Data (), middleware = middlewares
323
329
)
324
330
assert result .data == {"field" : "resolved" } # type: ignore
325
331
0 commit comments