端到端测试 Django 频道后台任务

Question

我正在尝试对使用通道和Websockets的Django应用程序进行端到端测试。就我而言，请求会触发复杂的后台计算，我想观察它的效果。据我了解，在 Django 测试套件中没有办法运行 worker，这将消耗 Django 通道的后台任务。

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

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

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...