Merge branch 'deduplicate' into channels2

Author: SmileyChris

.travis.yml

@@ -10,10 +10,11 @@ deploy:
     tags: true
 install: pip install -U tox-travis
 language: python
+dist: focal
 python:
+- 3.9
 - 3.8
 - 3.7
 - 3.6
-- 3.5
 - 2.7
 script: tox

graphql_ws/base_async.py

@@ -54,7 +54,10 @@ async def resolve(
     else:
         items = None
     if items is not None:
-        children = [resolve(child, _container=data, _key=key) for key, child in items]
+        children = [
+            asyncio.ensure_future(resolve(child, _container=data, _key=key))
+            for key, child in items
+        ]
         if children:
             await asyncio.wait(children)
 
@@ -90,10 +93,7 @@ def remember_task(self, task):
 
     async def unsubscribe(self, op_id):
         async_iterator = super().unsubscribe(op_id)
-        if (
-            getattr(async_iterator, "future", None)
-            and async_iterator.future.cancel()
-        ):
+        if getattr(async_iterator, "future", None) and async_iterator.future.cancel():
             await async_iterator.future
 
     async def unsubscribe_all(self):

graphql_ws/base_sync.py

@@ -65,7 +65,11 @@ def __init__(
         self.send_message = send_message
 
     def on_next(self, value):
-        self.send_execution_result(self.connection_context, self.op_id, value)
+        if isinstance(value, Exception):
+            send_method = self.send_error
+        else:
+            send_method = self.send_execution_result
+        send_method(self.connection_context, self.op_id, value)
 
     def on_completed(self):
         self.send_message(self.connection_context, self.op_id, GQL_COMPLETE)

setup.cfg

@@ -15,15 +15,12 @@ classifiers =
     License :: OSI Approved :: MIT License
     Natural Language :: English
     Programming Language :: Python :: 2
-    Programming Language :: Python :: 2.6
     Programming Language :: Python :: 2.7
     Programming Language :: Python :: 3
-    Programming Language :: Python :: 3.3
-    Programming Language :: Python :: 3.4
-    Programming Language :: Python :: 3.5
     Programming Language :: Python :: 3.6
     Programming Language :: Python :: 3.7
     Programming Language :: Python :: 3.8
+    Programming Language :: Python :: 3.9
 
 [options]
 zip_safe = False
@@ -95,4 +92,4 @@ omit =
 [coverage:report]
 exclude_lines =
     pragma: no cover
-    @abstract
+    @abstract

tests/conftest.py

@@ -2,7 +2,5 @@
 
 if sys.version_info > (3,):
     collect_ignore = ["test_django_channels.py"]
-    if sys.version_info < (3, 6):
-        collect_ignore.append("test_gevent.py")
 else:
     collect_ignore = ["test_aiohttp.py", "test_base_async.py"]

tests/test_aiohttp.py

@@ -1,15 +1,22 @@
+try:
+    from aiohttp import WSMsgType
+    from graphql_ws.aiohttp import AiohttpConnectionContext, AiohttpSubscriptionServer
+except ImportError:  # pragma: no cover
+    WSMsgType = None
+
 from unittest import mock
 
 import pytest
-from aiohttp import WSMsgType
 
-from graphql_ws.aiohttp import AiohttpConnectionContext, AiohttpSubscriptionServer
 from graphql_ws.base import ConnectionClosedException
 
+if_aiohttp_installed = pytest.mark.skipif(
+    WSMsgType is None, reason="aiohttp is not installed"
+)
+
 
 class AsyncMock(mock.Mock):
     def __call__(self, *args, **kwargs):
-
         async def coro():
             return super(AsyncMock, self).__call__(*args, **kwargs)
 
@@ -24,6 +31,7 @@ def mock_ws():
     return ws
 
 
+@if_aiohttp_installed
 @pytest.mark.asyncio
 class TestConnectionContext:
     async def test_receive_good_data(self, mock_ws):
@@ -69,5 +77,6 @@ async def test_close(self, mock_ws):
         mock_ws.close.assert_called_with(code=123)
 
 
+@if_aiohttp_installed
 def test_subscription_server_smoke():
     AiohttpSubscriptionServer(schema=None)

tests/test_base.py

@@ -8,6 +8,7 @@
 import pytest
 
 from graphql_ws import base
+from graphql_ws.base_sync import SubscriptionObserver
 
 
 def test_not_implemented():
@@ -77,3 +78,35 @@ def test_context_operations():
     assert not context.has_operation(1)
     # Removing a non-existant operation fails silently.
     context.remove_operation(999)
+
+
+def test_observer_data():
+    ws = mock.Mock()
+    context = base.BaseConnectionContext(ws)
+    send_result, send_error, send_message = mock.Mock(), mock.Mock(), mock.Mock()
+    observer = SubscriptionObserver(
+        connection_context=context,
+        op_id=1,
+        send_execution_result=send_result,
+        send_error=send_error,
+        send_message=send_message,
+    )
+    observer.on_next('data')
+    assert send_result.called
+    assert not send_error.called
+
+
+def test_observer_exception():
+    ws = mock.Mock()
+    context = base.BaseConnectionContext(ws)
+    send_result, send_error, send_message = mock.Mock(), mock.Mock(), mock.Mock()
+    observer = SubscriptionObserver(
+        connection_context=context,
+        op_id=1,
+        send_execution_result=send_result,
+        send_error=send_error,
+        send_message=send_message,
+    )
+    observer.on_next(TypeError('some bad message'))
+    assert send_error.called
+    assert not send_result.called

tests/test_base_async.py

@@ -15,24 +15,25 @@ async def __call__(self, *args, **kwargs):
         return super().__call__(*args, **kwargs)
 
 
-class TestServer(base_async.BaseAsyncSubscriptionServer):
+class TstServer(base_async.BaseAsyncSubscriptionServer):
     def handle(self, *args, **kwargs):
-        pass
+        pass  # pragma: no cover
 
 
 @pytest.fixture
 def server():
-    return TestServer(schema=None)
+    return TstServer(schema=None)
 
 
-async def test_terminate(server: TestServer):
+async def test_terminate(server: TstServer):
     context = AsyncMock()
     await server.on_connection_terminate(connection_context=context, op_id=1)
     context.close.assert_called_with(1011)
 
 
-async def test_send_error(server: TestServer):
+async def test_send_error(server: TstServer):
     context = AsyncMock()
+    context.has_operation = mock.Mock()
     await server.send_error(connection_context=context, op_id=1, error="test error")
     context.send.assert_called_with(
         {"id": 1, "type": "error", "payload": {"message": "test error"}}

tox.ini

@@ -1,15 +1,15 @@
 [tox]
-envlist = 
+envlist =
     coverage_setup
-    py27, py35, py36, py37, py38, flake8
+    py27, py36, py37, py38, py39, flake8
     coverage_report
 
 [travis]
 python =
-    3.8: py38, flake8
+    3.9: py39, flake8
+    3.8: py38
     3.7: py37
     3.6: py36
-    3.5: py35
     2.7: py27
 
 [testenv]
@@ -33,4 +33,4 @@ commands =
   coverage html
   coverage xml
   coverage report --include="tests/*" --fail-under=100 -m
-  coverage report --omit="tests/*"  # --fail-under=90 -m
+  coverage report --omit="tests/*"  # --fail-under=90 -m