What is the idiomatic approach for testing subscriptions in graphene-python
? It seems that the client.execute
option in graphene.test
is only appropriate for Query
testing.
P.S. There is a subscription execution example in the documentation but it does not seem to be part of the testing library (https://docs.graphene-python.org/en/latest/execution/subscriptions/).
The pre-release version of graphene
(3) supports subscriptions in this way:
import asyncio
from datetime import datetime
from graphene import ObjectType, String, Schema, Field
class Query(ObjectType):
hello = String()
def resolve_hello(root, info):
return 'Hello, world!'
class Subscription(ObjectType):
time_of_day = Field(String)
async def resolve_time_of_day(root, info):
while True:
yield datetime.now().isoformat()
await asyncio.sleep(1)
schema = Schema(query=Query, subscription=Subscription)
async def main():
subscription = 'subscription { timeOfDay }'
result = await schema.execute_async(subscription)
async for item in result:
print(item)
asyncio.run(main())
Source: https://github.com/graphql-python/graphene/issues/1099.