mirror of
https://github.com/graphql-python/graphene.git
synced 2024-11-24 18:43:55 +03:00
35 lines
672 B
Python
35 lines
672 B
Python
from pytest import mark
|
|
|
|
from graphene import ObjectType, Int, String, Schema
|
|
|
|
|
|
class Query(ObjectType):
|
|
a = String()
|
|
|
|
|
|
class Subscription(ObjectType):
|
|
count_to_ten = Int()
|
|
|
|
async def subscribe_count_to_ten(root, info):
|
|
count = 0
|
|
while count < 10:
|
|
count += 1
|
|
yield count
|
|
|
|
|
|
schema = Schema(query=Query, subscription=Subscription)
|
|
|
|
|
|
@mark.asyncio
|
|
async def test_subscription():
|
|
subscription = """
|
|
subscription {
|
|
countToTen
|
|
}
|
|
"""
|
|
result = await schema.subscribe(subscription)
|
|
count = 0
|
|
async for item in result:
|
|
count = item.data["countToTen"]
|
|
assert count == 10
|