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
|
import asyncio
import pytest
from graphql import ExecutionResult, GraphQLError, subscribe
from gql import Client, gql
from .fixtures import reviews
from .schema import StarWarsSchema
subscription_str = """
subscription ListenEpisodeReviews($ep: Episode!) {
reviewAdded(episode: $ep) {
stars,
commentary,
episode
}
}
"""
async def await_if_coroutine(obj):
"""Function to make tests work for graphql-core versions before and after 3.3.0a3"""
if asyncio.iscoroutine(obj):
return await obj
return obj
@pytest.mark.asyncio
async def test_subscription_support():
# reset review data for this test
reviews[6] = [
{"stars": 3, "commentary": "Was expecting more stuff", "episode": 6},
{"stars": 5, "commentary": "This is a great movie!", "episode": 6},
]
subs = gql(subscription_str)
params = {"ep": "JEDI"}
expected = [{**review, "episode": "JEDI"} for review in reviews[6]]
ai = await await_if_coroutine(
subscribe(StarWarsSchema, subs.document, variable_values=params)
)
result = [result.data["reviewAdded"] async for result in ai]
assert result == expected
@pytest.mark.asyncio
async def test_subscription_support_using_client():
# reset review data for this test
reviews[6] = [
{"stars": 3, "commentary": "Was expecting more stuff", "episode": 6},
{"stars": 5, "commentary": "This is a great movie!", "episode": 6},
]
subs = gql(subscription_str)
subs.variable_values = {"ep": "JEDI"}
expected = [{**review, "episode": "JEDI"} for review in reviews[6]]
async with Client(schema=StarWarsSchema) as session:
results = [
result["reviewAdded"]
async for result in await await_if_coroutine(
session.subscribe(subs, parse_result=False)
)
]
assert results == expected
subscription_invalid_str = """
subscription ListenEpisodeReviews($ep: Episode!) {
qsdfqsdfqsdf
}
"""
@pytest.mark.asyncio
async def test_subscription_support_using_client_invalid_field():
subs = gql(subscription_invalid_str)
subs.variable_values = {"ep": "JEDI"}
async with Client(schema=StarWarsSchema) as session:
# We subscribe directly from the transport to avoid local validation
results = [
result
async for result in await await_if_coroutine(
session.transport.subscribe(subs)
)
]
assert len(results) == 1
result = results[0]
assert isinstance(result, ExecutionResult)
assert result.data is None
assert isinstance(result.errors, list)
assert len(result.errors) == 1
error = result.errors[0]
assert isinstance(error, GraphQLError)
assert error.message == "The subscription field 'qsdfqsdfqsdf' is not defined."
|
