Subscriptions testing

[sztadii]

Hi,

I could not find a way to test my already created subscription

import asyncio
from ariadne import SubscriptionType, make_executable_schema
from ariadne.asgi import GraphQL

type_def = """
    type Subscription {
        counter: Int!
    }
"""

subscription = SubscriptionType()

@subscription.source("counter")
async def counter_generator(obj, info):
    for i in range(5):
        await asyncio.sleep(1)
        yield i


@subscription.field("counter")
def counter_resolver(count, info):
    return count + 1


schema = make_executable_schema(type_def, subscription)
app = GraphQL(schema, debug=True)

I used it from docs example https://ariadnegraphql.org/docs/0.9.0/subscriptions

For testing other resolvers ( query and mutation ) I have used FastAPI client

from fastapi.testclient import TestClient
from app import get_app

app = get_app()
client = TestClient(app)
response = client.post("/", json=user_query_body)
data = response.json()
user = data["data"]["user"]
assert user == { "name": "Chris" }

and works perfectly.

Any examples on how we can do test subscription resolvers?

[rafalp]

Ariadne's test suite contains examples of how subscriptions can be tested:

from starlette.testclient import TestClient

from ariadne.asgi import (
    GQL_CONNECTION_ACK,
    GQL_CONNECTION_ERROR,
    GQL_CONNECTION_INIT,
    GQL_CONNECTION_TERMINATE,
    GQL_START,
    GQL_DATA,
    GQL_STOP,
    GQL_COMPLETE,
    GraphQL,
    WebSocketConnectionError,
)


def test_field_can_be_subscribed_using_websocket_connection(client):
    with client.websocket_connect("/", "graphql-ws") as ws:
        ws.send_json({"type": GQL_CONNECTION_INIT})
        ws.send_json(
            {
                "type": GQL_START,
                "id": "test1",
                "payload": {"query": "subscription { ping }"},
            }
        )
        response = ws.receive_json()
        assert response["type"] == GQL_CONNECTION_ACK
        response = ws.receive_json()
        assert response["type"] == GQL_DATA
        assert response["id"] == "test1"
        assert response["payload"]["data"] == {"ping": "pong"}
        ws.send_json({"type": GQL_STOP, "id": "test1"})
        response = ws.receive_json()
        assert response["type"] == GQL_COMPLETE
        assert response["id"] == "test1"
        ws.send_json({"type": GQL_CONNECTION_TERMINATE})

From my understanding fastapi.testclient.TestClient is actually an alias for starlette.testclient.TestClient, so test should be same for both of those libs.

[sztadii]

Thanks :)