Merge branch 'master' into fix_passing_positional_arguments_in_transports

Author: leszekhanusz

docs/transports/websockets.rst

@@ -8,7 +8,14 @@ The websockets transport supports both:
  - the `Apollo websockets transport protocol`_.
  - the `GraphQL-ws websockets transport protocol`_
 
-It will detect the backend supported protocol from the response http headers returned.
+It will propose both subprotocols to the backend and detect the supported protocol
+from the response http headers returned by the backend.
+
+.. note::
+    For some backends (graphql-ws before `version 5.6.1`_ without backwards compatibility), it may be necessary to specify
+    only one subprotocol to the backend. It can be done by using
+    :code:`subprotocols=[WebsocketsTransport.GRAPHQLWS_SUBPROTOCOL]`
+    or :code:`subprotocols=[WebsocketsTransport.APOLLO_SUBPROTOCOL]` in the transport arguments.
 
 This transport allows to do multiple queries, mutations and subscriptions on the same websocket connection.
 
@@ -118,5 +125,6 @@ Here is an example with a ping sent every 60 seconds, expecting a pong within 10
         pong_timeout=10,
     )
 
+.. _version 5.6.1: https://github.com/enisdenjo/graphql-ws/releases/tag/v5.6.1
 .. _Apollo websockets transport protocol:  https://github.com/apollographql/subscriptions-transport-ws/blob/master/PROTOCOL.md
 .. _GraphQL-ws websockets transport protocol: https://github.com/enisdenjo/graphql-ws/blob/master/PROTOCOL.md

gql/client.py

@@ -583,8 +583,15 @@ async def __aenter__(self):
             self.session = AsyncClientSession(client=self)
 
         # Get schema from transport if needed
-        if self.fetch_schema_from_transport and not self.schema:
-            await self.session.fetch_schema()
+        try:
+            if self.fetch_schema_from_transport and not self.schema:
+                await self.session.fetch_schema()
+        except Exception:
+            # we don't know what type of exception is thrown here because it
+            # depends on the underlying transport; we just make sure that the
+            # transport is closed and re-raise the exception
+            await self.transport.close()
+            raise
 
         return self.session
 
@@ -605,8 +612,15 @@ def __enter__(self):
             self.session = SyncClientSession(client=self)
 
         # Get schema from transport if needed
-        if self.fetch_schema_from_transport and not self.schema:
-            self.session.fetch_schema()
+        try:
+            if self.fetch_schema_from_transport and not self.schema:
+                self.session.fetch_schema()
+        except Exception:
+            # we don't know what type of exception is thrown here because it
+            # depends on the underlying transport; we just make sure that the
+            # transport is closed and re-raise the exception
+            self.transport.close()
+            raise
 
         return self.session
 

gql/transport/websockets.py

@@ -3,7 +3,7 @@
 import logging
 from contextlib import suppress
 from ssl import SSLContext
-from typing import Any, Dict, Optional, Tuple, Union, cast
+from typing import Any, Dict, List, Optional, Tuple, Union, cast
 
 from graphql import DocumentNode, ExecutionResult, print_ast
 from websockets.datastructures import HeadersLike
@@ -46,6 +46,7 @@ def __init__(
         pong_timeout: Optional[Union[int, float]] = None,
         answer_pings: bool = True,
         connect_args: Dict[str, Any] = {},
+        subprotocols: Optional[List[Subprotocol]] = None,
     ) -> None:
         """Initialize the transport with the given parameters.
 
@@ -71,6 +72,9 @@ def __init__(
             (for the graphql-ws protocol).
             By default: True
         :param connect_args: Other parameters forwarded to websockets.connect
+        :param subprotocols: list of subprotocols sent to the
+            backend in the 'subprotocols' http header.
+            By default: both apollo and graphql-ws subprotocols.
         """
 
         super().__init__(
@@ -105,10 +109,13 @@ def __init__(
         """pong_received is an asyncio Event which will fire  each time
         a pong is received with the graphql-ws protocol"""
 
-        self.supported_subprotocols = [
-            self.APOLLO_SUBPROTOCOL,
-            self.GRAPHQLWS_SUBPROTOCOL,
-        ]
+        if subprotocols is None:
+            self.supported_subprotocols = [
+                self.APOLLO_SUBPROTOCOL,
+                self.GRAPHQLWS_SUBPROTOCOL,
+            ]
+        else:
+            self.supported_subprotocols = subprotocols
 
     async def _wait_ack(self) -> None:
         """Wait for the connection_ack message. Keep alive messages are ignored"""
@@ -272,6 +279,7 @@ def _parse_answer_graphqlws(
             - instead of a unidirectional keep-alive (ka) message from server to client,
               there is now the possibility to send bidirectional ping/pong messages
             - connection_ack has an optional payload
+            - the 'error' answer type returns a list of errors instead of a single error
         """
 
         answer_type: str = ""
@@ -288,11 +296,11 @@ def _parse_answer_graphqlws(
 
                     payload = json_answer.get("payload")
 
-                    if not isinstance(payload, dict):
-                        raise ValueError("payload is not a dict")
-
                     if answer_type == "next":
 
+                        if not isinstance(payload, dict):
+                            raise ValueError("payload is not a dict")
+
                         if "errors" not in payload and "data" not in payload:
                             raise ValueError(
                                 "payload does not contain 'data' or 'errors' fields"
@@ -309,8 +317,11 @@ def _parse_answer_graphqlws(
 
                     elif answer_type == "error":
 
+                        if not isinstance(payload, list):
+                            raise ValueError("payload is not a list")
+
                         raise TransportQueryError(
-                            str(payload), query_id=answer_id, errors=[payload]
+                            str(payload[0]), query_id=answer_id, errors=payload
                         )
 
             elif answer_type in ["ping", "pong", "connection_ack"]:

tests/conftest.py

@@ -465,7 +465,9 @@ async def client_and_graphqlws_server(graphqlws_server):
     # Generate transport to connect to the server fixture
     path = "/graphql"
     url = f"ws://{graphqlws_server.hostname}:{graphqlws_server.port}{path}"
-    sample_transport = WebsocketsTransport(url=url)
+    sample_transport = WebsocketsTransport(
+        url=url, subprotocols=[WebsocketsTransport.GRAPHQLWS_SUBPROTOCOL],
+    )
 
     async with Client(transport=sample_transport) as session:
 

tests/test_client.py

@@ -200,3 +200,50 @@ def test_gql():
     client = Client(schema=schema)
     result = client.execute(query)
     assert result["user"] is None
+
+
+@pytest.mark.requests
+def test_sync_transport_close_on_schema_retrieval_failure():
+    """
+    Ensure that the transport session is closed if an error occurs when
+    entering the context manager (e.g., because schema retrieval fails)
+    """
+
+    from gql.transport.requests import RequestsHTTPTransport
+
+    transport = RequestsHTTPTransport(url="http://localhost/")
+    client = Client(transport=transport, fetch_schema_from_transport=True)
+
+    try:
+        with client:
+            pass
+    except Exception:
+        # we don't care what exception is thrown, we just want to check if the
+        # transport is closed afterwards
+        pass
+
+    assert client.transport.session is None
+
+
+@pytest.mark.aiohttp
+@pytest.mark.asyncio
+async def test_async_transport_close_on_schema_retrieval_failure():
+    """
+    Ensure that the transport session is closed if an error occurs when
+    entering the context manager (e.g., because schema retrieval fails)
+    """
+
+    from gql.transport.aiohttp import AIOHTTPTransport
+
+    transport = AIOHTTPTransport(url="http://localhost/")
+    client = Client(transport=transport, fetch_schema_from_transport=True)
+
+    try:
+        async with client:
+            pass
+    except Exception:
+        # we don't care what exception is thrown, we just want to check if the
+        # transport is closed afterwards
+        pass
+
+    assert client.transport.session is None

tests/test_graphqlws_exceptions.py

@@ -1,6 +1,4 @@
 import asyncio
-import json
-import types
 from typing import List
 
 import pytest
@@ -125,49 +123,29 @@ async def test_graphqlws_server_does_not_send_ack(
             pass
 
 
-invalid_payload_server_answer = (
-    '{"type":"error","id":"1","payload":{"message":"Must provide document"}}'
+invalid_query_server_answer = (
+    '{"id":"1","type":"error","payload":[{"message":"Cannot query field '
+    '\\"helo\\" on type \\"Query\\". Did you mean \\"hello\\"?",'
+    '"locations":[{"line":2,"column":3}]}]}'
 )
 
 
-async def server_invalid_payload(ws, path):
+async def server_invalid_query(ws, path):
     await WebSocketServerHelper.send_connection_ack(ws)
     result = await ws.recv()
     print(f"Server received: {result}")
-    await ws.send(invalid_payload_server_answer)
+    await ws.send(invalid_query_server_answer)
     await WebSocketServerHelper.wait_connection_terminate(ws)
     await ws.wait_closed()
 
 
 @pytest.mark.asyncio
-@pytest.mark.parametrize("graphqlws_server", [server_invalid_payload], indirect=True)
-@pytest.mark.parametrize("query_str", [invalid_query_str])
-async def test_graphqlws_sending_invalid_payload(
-    event_loop, client_and_graphqlws_server, query_str
-):
+@pytest.mark.parametrize("graphqlws_server", [server_invalid_query], indirect=True)
+async def test_graphqlws_sending_invalid_query(event_loop, client_and_graphqlws_server):
 
     session, server = client_and_graphqlws_server
 
-    # Monkey patching the _send_query method to send an invalid payload
-
-    async def monkey_patch_send_query(
-        self, document, variable_values=None, operation_name=None,
-    ) -> int:
-        query_id = self.next_query_id
-        self.next_query_id += 1
-
-        query_str = json.dumps(
-            {"id": str(query_id), "type": "subscribe", "payload": "BLAHBLAH"}
-        )
-
-        await self._send(query_str)
-        return query_id
-
-    session.transport._send_query = types.MethodType(
-        monkey_patch_send_query, session.transport
-    )
-
-    query = gql(query_str)
+    query = gql("{helo}")
 
     with pytest.raises(TransportQueryError) as exc_info:
         await session.execute(query)
@@ -178,7 +156,10 @@ async def monkey_patch_send_query(
 
     error = exception.errors[0]
 
-    assert error["message"] == "Must provide document"
+    assert (
+        error["message"]
+        == 'Cannot query field "helo" on type "Query". Did you mean "hello"?'
+    )
 
 
 not_json_answer = ["BLAHBLAH"]
@@ -188,6 +169,7 @@ async def monkey_patch_send_query(
 missing_id_answer_3 = ['{"type": "complete"}']
 data_without_payload = ['{"type": "next", "id":"1"}']
 error_without_payload = ['{"type": "error", "id":"1"}']
+error_with_payload_not_a_list = ['{"type": "error", "id":"1", "payload": "NOT A LIST"}']
 payload_is_not_a_dict = ['{"type": "next", "id":"1", "payload": "BLAH"}']
 empty_payload = ['{"type": "next", "id":"1", "payload": {}}']
 sending_bytes = [b"\x01\x02\x03"]
@@ -205,6 +187,7 @@ async def monkey_patch_send_query(
         data_without_payload,
         error_without_payload,
         payload_is_not_a_dict,
+        error_with_payload_not_a_list,
         empty_payload,
         sending_bytes,
     ],