时间戳

首页 > 解决方案 > 使用 pytest 进行 Django 通道 ProtocolTypeRouter 单元测试

django-channels - 使用 pytest 进行 Django 通道 ProtocolTypeRouter 单元测试

问题描述

我正在检查下一个项目的频道。但我坚持对基础进行单元测试。我想这一定是我一直忽略的东西。

版本：

路由.py

from channels.routing import ProtocolTypeRouter #, URLRouter
from Phoenix.ProtocolTestConsumer import ProtocolTestConsumer

application = ProtocolTypeRouter({
    # Empty for now (http->django views is added by default)

    "test": ProtocolTestConsumer,
})

协议测试消费者.py

import logging
logging.basicConfig(level=logging.DEBUG)

from django.conf import settings
from channels.consumer import SyncConsumer

class ProtocolTestConsumer(SyncConsumer):
    def test_a(self, event):
        logging.debug("consumer test_a {}".format(event))

    def test_b(self, event):
        logging.debug("consumer test_b {}".format(event))

    def test_c(self, event):
        logging.debug("consumer test_c {}".format(event))

test_ProtocolTestConsumer.py

import pytest
import logging

from Phoenix.ProtocolTestConsumer import ProtocolTestConsumer
from channels.testing import ApplicationCommunicator

communicator = ApplicationCommunicator(ProtocolTestConsumer, {"type": "test"})

@pytest.mark.asyncio
async def test_call_a():
    await communicator.send_input(
        {"type": "test.a",
         "text": "Blue"})
    logging.debug("pytest is printing this")

DJANGO_SETTINGS_MODULE=web.settings pytest -s -v

== test session starts ==
platform linux -- Python 3.7.1, pytest-4.1.1, py-1.7.0, pluggy-0.8.1 -- /home/user/src/web/Phoenix/venv-phoenix/bin/python3.7
cachedir: .pytest_cache
Django settings: web.settings (from environment variable)
rootdir: /home/user/src/web/Phoenix/web/Phoenix, inifile:
plugins: django-3.4.5, asyncio-0.10.0
collected 1 item                                                                                        

test_ProtocolTestConsumer.py::test_call_a DEBUG:asyncio:Using selector: EpollSelector
DEBUG:root:pytest is printing this
PASSED

== 1 passed in 0.24 seconds ==

我希望看到来自消费者的更多控制台输出。

标签： django-channels

解决方案

我在您的代码中看到了两个问题。首先，您的路由似乎缺少某些部分。在将消费者传递给协议类型路由器之前，应首先将消费者包装为 django url。像这样：

from channels.routing import ProtocolTypeRouter, URLRouter

application = ProtocolTypeRouter({
    # Empty for now (http->django views is added by default)

    "test": [
        path('test/', ProtocolTestConsumer)
    ]
})

第一个“测试”是做一个协议路由，而第二个是做一个普通的 django url 路由。

其次，在初始化你的通信器时，你应该使用消费者的 url：

ApplicationCommunicator(ProtocolTestConsumer, "test/")

编辑

仔细查看您的代码后，我发现它实际上是正确实现的，只是在较低级别上使其成为自己的协议，如 websocket 和 HTTP。但是，您的错误是在调用send_output. send_output 调用只是将消息放入队列中。然后receive_output检索队列顶部的第一条消息，该消息触发对消费者中相应方法的调用。

因此，您需要以这种方式在测试代码中接收输出：

@pytest.mark.asyncio
async def test_call_a():
    await communicator.send_input(
        {"type": "test.a",
         "text": "Blue"})
    logging.debug("pytest is printing this")
    response = await communicator.receive_output()

您可以查看 的实现WebsocketCommunicator以及HttpCommunicator有关它们如何发送和接收消息的更多详细信息

推荐阅读