Errors whilst testing with the WebsocketCommunicator

[jensjeflensje]

Hi everyone, I'm posting this issue to talk about the WebsocketCommunicator class. I'm unable to get it to work using the documentation provided. I'm unsure if this is just a gap in the documentation, or actually a bug. At least it should be an actionable problem. I am expecting the first test's .connect() to fail in the caught way (not quite sure if TimeoutError is the correct exception though), and the second to just succeed, as the docs suggest.

consumers.py:

from <my django app>.asgi import application


WEBSOCKET_PATH = "api/ws"


class CoreWebsocketTest(TestCase):

    def setUp(self):
        self.communicator = WebsocketCommunicator(application,
                                                  WEBSOCKET_PATH)

    async def test_connection_no_login(self):
        try:
            await self.communicator.connect()
        except TimeoutError:
            pass
        else:
            raise Exception("Client connected with no auth.")

    async def test_connection(self):
        connected, _ = await self.communicator.connect()
        self.assertTrue(connected)

(I understand that it's not actually logging in right now to actually test that stuff. I wanted to make it as stripped down as possible to test)

routing.py:

from django.urls import path

from . import consumers

websocket_urlpatterns = [
    path('api/ws',
         consumers.CoreWebsocketConsumer.as_asgi()),
]

I've tried versions v3.0.5 and v4.0.0, both suffered from the same issue. I think I narrowed the problem down to line 36 of channels.testing.websocket:

    async def connect(self, timeout=1):
        """
        Trigger the connection code.

        On an accepted connection, returns (True, <chosen-subprotocol>)
        On a rejected connection, returns (False, <close-code>)
        """
        await self.send_input({"type": "websocket.connect"})
        response = await self.receive_output(timeout) # <-----
        if response["type"] == "websocket.close":
            return (False, response.get("code", 1000))
        else:
            return (True, response.get("subprotocol", None))

This also suggests the issue is inside asgiref, as receive_output is from asgiref. Please let me know if that's a better place to post this :).

asgi.py

django_asgi_app = get_asgi_application()

application = ProtocolTypeRouter({
    "http": django_asgi_app,
    "websocket": AllowedHostsOriginValidator(
        AuthMiddlewareStack(
            URLRouter(
                <my django app>.websocket.routing.websocket_urlpatterns
            )
        )
    ),
})

I get this error:

Error
Traceback (most recent call last):
  File "/usr/local/lib/python3.10/site-packages/asgiref/testing.py", line 74, in receive_output
    return await self.output_queue.get()
  File "/usr/local/lib/python3.10/asyncio/queues.py", line 159, in get
    await getter
asyncio.exceptions.CancelledError

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/usr/local/lib/python3.10/site-packages/asgiref/testing.py", line 73, in receive_output
    async with async_timeout(timeout):
  File "/usr/local/lib/python3.10/site-packages/asgiref/timeout.py", line 65, in __aexit__
    self._do_exit(exc_type)
  File "/usr/local/lib/python3.10/site-packages/asgiref/timeout.py", line 102, in _do_exit
    raise asyncio.TimeoutError
asyncio.exceptions.TimeoutError

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/usr/local/lib/python3.10/site-packages/asgiref/sync.py", line 218, in __call__
    return call_result.result()
  File "/usr/local/lib/python3.10/concurrent/futures/_base.py", line 439, in result
    return self.__get_result()
  File "/usr/local/lib/python3.10/concurrent/futures/_base.py", line 391, in __get_result
    raise self._exception
  File "/usr/local/lib/python3.10/site-packages/asgiref/sync.py", line 284, in main_wrap
    result = await self.awaitable(*args, **kwargs)
  File "/opt/project/backend/src/<my django app>/websocket/tests.py", line 57, in test_connection_with_login
    connected, _ = await self.communicator.connect()
  File "/usr/local/lib/python3.10/site-packages/channels/testing/websocket.py", line 36, in connect
    response = await self.receive_output(timeout)
  File "/usr/local/lib/python3.10/site-packages/asgiref/testing.py", line 82, in receive_output
    await self.future
RuntimeError: Task <Task pending name='Task-14' coro=<AsyncToSync.main_wrap() running at /usr/local/lib/python3.10/site-packages/asgiref/sync.py:284> cb=[_run_until_complete_cb() at /usr/local/lib/python3.10/asyncio/base_events.py:184]> got Future <Task cancelling name='Task-13' coro=<CoreWebsocketConsumer() running at /usr/local/lib/python3.10/site-packages/channels/consumer.py:92>> attached to a different loop

pip freeze output:

amqp==5.1.1
asgiref==3.5.2
asttokens==2.0.8
async-timeout==4.0.2
attrs==22.1.0
autobahn==22.7.1
Automat==20.2.0
backcall==0.2.0
billiard==3.6.4.0
bleach==5.0.1
boto3==1.24.3
botocore==1.27.91
celery==5.2.7
certifi==2022.9.24
cffi==1.15.1
channels==4.0.0
channels-redis==4.0.0
charset-normalizer==2.1.1
click==8.1.3
click-didyoumean==0.3.0
click-plugins==1.1.1
click-repl==0.2.0
constantly==15.1.0
cryptography==38.0.1
cssselect2==0.7.0
daphne==4.0.0
decorator==5.1.1
Deprecated==1.2.13
Django==4.0.5
django-anymail==8.6
django-axes==5.35.0
django-bleach==3.0.0
django-cors-headers==3.13.0
django-environ==0.8.1
django-filter==21.1
django-hashid-field==3.3.4
django-influxdb-metrics==1.5.0
django-ipware==4.0.2
django-polymorphic==3.1.0
django-redis==5.2.0
django-storages==1.12.3
djangorestframework==3.13.1
easy-thumbnails==2.8.1
executing==1.1.1
factory-boy==3.2.1
Faker==15.1.1
ffmpeg-python==0.2.0
flake8==4.0.1
freezegun==1.2.1
future==0.18.2
gconfigs==0.1.5
gevent==21.12.0
greenlet==1.1.3.post0
gunicorn==20.1.0
hashids==1.3.1
hyperlink==21.0.0
idna==3.4
incremental==22.10.0
influxdb==5.3.1
ipython==8.4.0
jedi==0.18.1
jmespath==1.0.1
kombu==5.2.4
lxml==4.9.1
matplotlib-inline==0.1.6
mccabe==0.6.1
memory-profiler==0.60.0
monday==1.3.0
msgpack==1.0.4
mux-python==3.5.0
packaging==21.3
parso==0.8.3
pexpect==4.8.0
pickleshare==0.7.5
Pillow==9.1.1
prompt-toolkit==3.0.31
psutil==5.9.2
psycopg2-binary==2.9.3
ptyprocess==0.7.0
pure-eval==0.2.2
pyasn1==0.4.8
pyasn1-modules==0.2.8
pycodestyle==2.8.0
pycparser==2.21
pyflakes==2.4.0
Pygments==2.13.0
PyJWT==2.4.0
pyOpenSSL==22.1.0
pyparsing==3.0.9
python-dateutil==2.8.2
python-server-metrics==0.2.2
pytz==2022.4
redis==4.3.4
reportlab==3.6.11
requests==2.28.1
s3transfer==0.6.0
sentry-sdk==1.5.12
service-identity==21.1.0
six==1.16.0
sqlparse==0.4.3
stack-data==0.5.1
svglib==1.4.1
tblib==1.7.0
tinycss2==1.1.1
tld==0.12.6
traitlets==5.4.0
Twisted==22.8.0
txaio==22.2.1
typing_extensions==4.4.0
urllib3==1.26.9
vine==5.0.0
wcwidth==0.2.5
webencodings==0.5.1
wrapt==1.14.1
zope.event==4.5.0
zope.interface==5.5.0

[dr-ftvkun]

The problem is that you created communicator() in sync setUp() method and tried to reuse it in async test.

The communicator inits its input and output Queue with one event loop but the async tests are being executed in the another one. Hope it helps someone.

[jensjeflensje]

How would I fix this? I still want to use my setUp method to create the communicator, as putting it in every function would not be that elegant. I'm not able to make my setup async, right?

[dr-ftvkun]

How would I fix this? I still want to use my setUp method to create the communicator, as putting it in every function would not be that elegant. I'm not able to make my setup async, right?

The only worked solution for me was to init it in smth like async def _create_communicator() and use explicitly in the beginning of each test.

For example, I tried to use async_to_sync(_init_communicator_in_setUp)() right in setUp() but was out of luck with the other errors.