端到端测试 Django Channels 后台任务

Question

我正在尝试对使用 Channels 和 Websockets 的 Django 应用程序执行端到端测试。在我的例子中，一个请求触发了一个复杂的后台计算，我想观察它的效果。据我了解，在Django测试套件中没有运行 workers的方式，这将消耗Django Channels的后台任务。

有什么方法可以使用正确的参数触发后台任务吗？我是否必须修改我的代码以使其可测试？

以下是我的设置的简化框架：

views.py

def calculation(request):
    # doing some simple calculation depending on request
    simple_result, parameter = simple_calculation(request)
    # trigger complex asynchronous calculation depending on parameter
    Channel('background-calculations').send({
        "parameter": parameter,
    })
    # return results of simple calculation
    return render(request, 'simple_reponse.html',{'simple_result': simple_result})

routing.py

channel_routing = [
    route("background-calculations", run_background_calculations),
    route("websocket.connect", ws_connect),
    route("websocket.receive", ws_message),
]

consumers.py

def run_background_calculations(message):
    # perform complex calculation depending on parameter from simple calculation
    result = complex_calculation(message)
    # update frontend via websocket
    Group("frontend-updates").send({
        "text": json.dumps({
            "result": result,
        })
    })

@channel_session
def ws_connect(message):
    message.reply_channel.send({"accept": True})
    Group("frontend-updates").add(message.reply_channel)


@channel_session_user
def ws_message(message):
    message.reply_channel.send({
        "text": message.content['text'],
    })

当我尝试使用客户端（例如 REST APIClient）访问视图时，我确实收到了响应，但后台任务从未执行过。没有办法测试整个链条吗？

Answer 1

所以我终于想通了。

如果有人发现这个问题，这里是执行端到端测试的方法：

from channels.test import Client
from rest_framework.test import APIClient

# setup
rest = APIClient()
channelclient = Client()
Group("frontend-updates").add(u"test-channel") 
rest.login(username="username",password="password")

class CopyTestCase(ChannelTestCase):
    def test_end_to_end(self):
        # trigger HTTP POST request
        response = rest.post("/calculate/")
        #...check response...

        # consume the background-calculation task (has to be done manually)
        channelclient.consume(u"background-calculations")

        # check data sent to the front-end as a result of this calculation
        message = testcase.get_next_message(u"test-channel")
        #...check message...

端到端测试 Django Channels 后台任务

Testing Django Channels background task end-to-end

django

testing

django-channels

views.py

routing.py

consumers.py