如何解决Django channels3 测试从异步调用同步
我在使用 Django Channels 3 时遇到问题。我想从异步函数调用 http 测试客户端。
这是我的功能
def broadcast_message(client,message):
return client.post(
reverse("message-broadcast"),message,content_type="application/json")
这里是异步函数
@pytest.mark.asyncio
@pytest.mark.django_db(True)
async def test_end_to_end(client):
http_response = await loop.run_in_executor(
None,broadcast_message,client,test_message)
RuntimeError: Task <Task pending name='Task-1' coro=<test_end_to_end() running at /usr/src/app/tests/test_consumer_and_view.py:62> cb=[_run_until_complete_cb() at /usr/local/lib/python3.8/asyncio/base_events.py:184]> got Future <Future pending cb=[_chain_future.<locals>._call_check_cancel() at /usr/local/lib/python3.8/asyncio/futures.py:360]> attached to a different loop
tests/test_consumer_and_view.py:62: RuntimeError
我想了解发生了什么以及如何解决。谢谢。
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。