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 (将#修改为@）

django pytest-django python-asyncio