Блог

Главная - Вопросы по программированию - Тестирование подписок с использованием graphene.тест в Python Graphene

Тестирование подписок с использованием graphene.тест в Python Graphene

#python #graphql #graphene-python

#python #graphql #graphene-python

Вопрос:

В чем заключается идиоматический подход к тестированию подписок graphene-python ? Похоже, что client.execute опция in graphene.test подходит только для Query тестирования.

PS В документации есть пример выполнения подписки, но, похоже, он не является частью библиотеки тестирования (https://docs.graphene-python.org/en/latest/execution/subscriptions /).

Ответ №1:

Предварительная версия graphene (3) поддерживает подписки таким образом:

 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())

Источник: https://github.com/graphql-python/graphene/issues/1099 .

Комментарии:

1. Подписка из примера не обрабатывает события из других частей API. Вы знаете, как я мог бы сделать это для приложения для обмена сообщениями? Например, клиент 1 отправляет сообщение на сервер, сервер создает сообщение и отправляет через websocket вновь созданное сообщение? В graphene v2 я бы использовал слои каналов для передачи событий, но теперь я не уверен, как это сделать.

Ответ №2:

Вот как вы это сделаете в 2022 году с Graphene 3:

pip install pytest pytest-asyncio

 import pytest
from main.schema import schema

@pytest.mark.asyncio
async def test_realtime_updates():
    q = 'subscription { count }'
    result = await schema.subscribe(q)
    
    async for item in result:
        print(item)

    pytest.fail(pytrace=False)

Если у вас нет никакого приложения, вот минимальный пример с подписками:

pip установка graphene graphene_starlette3

 # Graphene Subscriptions demo

import graphene
from graphene import ResolveInfo
from starlette_graphene3 import GraphQLApp, make_playground_handler


class Query(graphene.ObjectType):
    hello = graphene.String()

class Subscription(graphene.ObjectType):
    count = graphene.Float()

    async def subscribe_count(_, info: ResolveInfo):
        for i in range(10):
            yield i


# GraphQL Schema
schema = graphene.Schema(
    query=Query,
    subscription=Subscription,
)


# ASGI application
# You'll probably want to mount it to a Starlette application
app = GraphQLApp(
    schema=schema,
    on_get=make_playground_handler(),
)

Запустите его:

 $ uvicorn run main:app --reload