django通道WebsocketCommunicator超时错误

Question

我正在尝试运行以下测试：

测试.py

from rest_framework.test import APITestCase
from myapp.routing import application
from channels.testing import WebsocketCommunicator
from account.models import User
from rest_framework.authtoken.models import Token
class Tests(APITestCase):
def setUp(self):
self.user = User.objects.create(email='[e-mail]', 
password='a password')
self.token, created = Token.objects.get_or_create(user=self.user)
async def test_connect(self):
communicator = WebsocketCommunicator(application, f"/ws/user/{self.token}/")
connected, subprotocol = await communicator.connect()
self.assertTrue(connected)
await communicator.disconnect()

application是channels.routing.ProtocolTypeRouter的样板实例(类似于此处：https://channels.readthedocs.io/en/latest/topics/routing.html)。生产中一切顺利。测试退出时出现以下错误：

Traceback (most recent call last):
File "/home/projects/myapp/myapp-env/lib/python3.7/site-packages/asgiref/testing.py", line 74, in receive_output
return await self.output_queue.get()
File "/usr/lib/python3.7/asyncio/queues.py", line 159, in get
await getter
concurrent.futures._base.CancelledError
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
File "/home/projects/myapp/myapp-env/lib/python3.7/site-packages/asgiref/sync.py", line 223, in __call__
return call_result.result()
File "/usr/lib/python3.7/concurrent/futures/_base.py", line 428, in result
return self.__get_result()
File "/usr/lib/python3.7/concurrent/futures/_base.py", line 384, in __get_result
raise self._exception
File "/home/projects/myapp/myapp-env/lib/python3.7/site-packages/asgiref/sync.py", line 292, in main_wrap
result = await self.awaitable(*args, **kwargs)
File "/home/projects/myapp/myapp-api/app/tests.py", line 35, in test_connect
connected, subprotocol = await communicator.connect()
File "/home/projects/myapp/myapp-env/lib/python3.7/site-packages/channels/testing/websocket.py", line 36, in connect
response = await self.receive_output(timeout)
File "/home/projects/myapp/myapp-env/lib/python3.7/site-packages/asgiref/testing.py", line 85, in receive_output
raise e
File "/home/projects/myapp/myapp-env/lib/python3.7/site-packages/asgiref/testing.py", line 74, in receive_output
return await self.output_queue.get()
File "/home/projects/myapp/myapp-env/lib/python3.7/site-packages/asgiref/timeout.py", line 66, in __aexit__
self._do_exit(exc_type)
File "/home/projects/myapp/myapp-env/lib/python3.7/site-packages/asgiref/timeout.py", line 103, in _do_exit
raise asyncio.TimeoutError
concurrent.futures._base.TimeoutError
----------------------------------------------------------------------
Ran 1 test in 1.026s

我尝试过python 3.7.5、3.8.0和3.9.9版本，使用带有django 3.2.10的通道3.0.4和通道redis 3.3.1(设置.py中的'BACKEND': 'channels_redis.core.RedisChannelLayer'(。错误仍然存在。我做错了什么？

Answer 1

我也遇到了同样的问题。APITestCase或TestCase不允许事务，您必须使用django测试中的SimpleTestCase，并将数据库设置为所有数据库。只要有这样的区别，我想它就会起作用。请注意，事务将在测试之间保存，而不会在测试后回滚。

from django.test import SimpleTestCase
from myapp.routing import application
from channels.testing import WebsocketCommunicator
from account.models import User
from rest_framework.authtoken.models import Token

class Tests(SimpleTestCase):
databases = '__all__'
def setUp(self):
self.user = User.objects.create(email='[e-mail]', password='a password')
self.token, created = Token.objects.get_or_create(user=self.user)
async def test_connect(self):
communicator = WebsocketCommunicator(application, f"/ws/user/{self.token}/")
connected, subprotocol = await communicator.connect()
self.assertTrue(connected)
await communicator.disconnect()

以下是SimpleTestCase的信息https://docs.djangoproject.com/en/4.0/topics/testing/tools/

Answer 2

我遇到了类似的问题，解决方案通常也是为了测试而模拟您的生产路由器，即在实例化Communicator时，还应该添加生产中使用的任何中间件或附加组件。例如，在我的asgi.py中，我有：

application = ProtocolTypeRouter(
{
"http": get_asgi_application(),
"websocket": AllowedHostsOriginValidator(
jwt_auth_middleware_stack(URLRouter(chat.routing.websocket_urlpatterns)),
),
}
)

我的通信器实例化如下：

communicator = WebsocketCommunicator(jwt_auth_middleware_stack(URLRouter(websocket_urlpatterns)),
f"/ws/chat/{chat.id}/?token={token}")

我的网址是：

websocket_urlpatterns = [
path("ws/chat/<str:chat_id>/", consumers.AsyncChatConsumer.as_asgi())
]