使用 Django Channels 和 pytest-asyncio 测试消费者方法是否可以引发异常

Question

使用 Django 和 Channels 2，我有一个可以通过频道组访问并可能引发异常的使用者方法。就像这个微不足道的：

from channels.generic.websocket import WebsocketConsumer
from asgiref.sync import async_to_sync

class DummyConsumer(WebsocketConsumer):
    def connect(self):
        async_to_sync(self.channel_layer.group_add)(
            "dummy",
            self.channel_name,
        )
        self.accept()

    def will_raise(self, event):
        raise ValueError('value error')

    def disconnect(self, code):
        async_to_sync(self.channel_layer.group_discard)(
            "dummy",
            self.channel_name,
        )

我想使用 pytest-asyncio 测试这个方法。由于可以使用 pytest.raises 捕获协程的异常，我天真地认为这样的事情就足够了：

import pytest
from channels.testing import WebsocketCommunicator
from channels.layers import get_channel_layer
from app.consumers import DummyConsumer
channel_layer = get_channel_layer()

@pytest.fixture
async def communicator():
    communicator = WebsocketCommunicator(DummyConsumer, "ws/dummy/")
    await communicator.connect()
    yield communicator
    await communicator.disconnect()

@pytest.mark.asyncio
async def test_will_raise(communicator):
    with pytest.raises(ValueError):
        await channel_layer.group_send('dummy', {
            'type': 'will_raise'
        })

但是测试以一种非常混乱的方式失败了（截断的输出）：

================== ERRORS ==================
___ ERROR at teardown of test_will_raise ___
...
>       raise ValueError('value error')
E       ValueError: value error

app/consumers.py:28: ValueError
================= FAILURES =================
_____________ test_will_raise ______________
...
            await channel_layer.group_send('dummy', {
>               'type': 'will_raise'
            })
E           Failed: DID NOT RAISE <class 'ValueError'>

app/tests_dummy.py:21: Failed
==== 1 failed, 1 error in 1.47 seconds =====

那么，我该怎么办？从消费者方法中引发异常是一个糟糕的设计吗？

Answer 1

A channel_layer 有两个站点。一个站点将数据发送到 channel_layer，另一个站点接收数据。发送站点没有从接收站点得到任何响应。这意味着，如果接收站点引发异常，发送站点将看不到它。

在您的测试中，您正在测试发送站点。它向 channel_layer 发送一条消息，但正如所解释的那样，这不会引发异常。

要测试是否引发了异常，您必须编写一个连接到您的使用者的测试。它可能看起来像这样：

channel_layer = get_channel_layer()

@pytest.mark.asyncio
async def test_will_raise():
    communicator = WebsocketCommunicator(DummyConsumer, "ws/dummy/")
    await communicator.connect()

    await channel_layer.group_send('dummy', {
            'type': 'will_raise'
        })

    with pytest.raises(ValueError):
        await communicator.wait()

如您所见，当您发送到 channel_layer 时不会发生异常，但在侦听 channel_layer 的通信器上会发生异常。另见：https://channels.readthedocs.io/en/latest/topics/testing.html#wait

另请注意，测试不会调用 communicator.disconnect()。当通信器内部发生异常时，不必调用 disconnect()。请参阅此标题下方绿色 "Important" 框中的第二句话：https://channels.readthedocs.io/en/latest/topics/testing.html#websocketcommunicator

You do not, however, have to disconnect() if your app already raised an error.

使用 Django Channels 和 pytest-asyncio 测试消费者方法是否可以引发异常

Test that a consumer method can raise an exception with Django Channels and pytest-asyncio

python

python-3.x

python-asyncio

django-channels

pytest-asyncio