GraphQLWebsocketCommunicator

A test communicator for GraphQL over Websockets.

import pytest
from strawberry.channels.testing import GraphQLWebsocketCommunicator
from myapp.asgi import application


@pytest.fixture
async def gql_communicator():
    async with GraphQLWebsocketCommunicator(application, path="/graphql") as client:
        yield client


async def test_subscribe_echo(gql_communicator):
    async for res in gql_communicator.subscribe(
        query='subscription { echo(message: "Hi") }'
    ):
        assert res.data == {"echo": "Hi"}

Constructor:

def __init__(
    self,
    application: ASGIApplication,
    path: str,
    headers: List[Tuple[bytes, bytes]] | None = None,
    protocol: str = GRAPHQL_TRANSPORT_WS_PROTOCOL,
    connection_params: dict = ExprDict,
    kwargs: Any = {},
) -> None:
  ...

Methods:

• gql_init

  Signature:

  def gql_init(self) -> None:
    ...

• subscribe

  Signature:

  def subscribe(self, query: str, variables: Dict | None = None) -> ExecutionResult | AsyncIterator[ExecutionResult]:
    ...

  Parameters:

  1. query:

     Type

     str

  2. variables:

     Type

     Dict | None

     Default

     None

• process_errors

  Reconstructs a GraphQLError from a FormattedGraphQLError.

  Signature:

  def process_errors(self, errors: List[GraphQLFormattedError]) -> List[GraphQLError]:
    ...

  Parameters:

  1. errors:

     Type

     List[GraphQLFormattedError]

Attributes:

1. protocol:

2. connection_params: