AsyncWebsocketConsumer 中的日志记录有什么问题？

Question

我有一个 Django 项目，它使用 Channels 进行 websocket 通信，我偶然发现了我认为可能是一个错误的地方，但我不确定，所以我希望有人比我更了解这一点，可以帮助解释发生了什么这里：

测试错误信息：

AssertionError: "INFO:websocket.camera_consumer:Doesn't work" not found in ['INFO:websocket.camera_consumer:Works']

失败的测试：

class TestCameraConsumerTests(TransactionTestCase):
    async def test_fails(self):
        communicator = WebsocketCommunicator(TestConsumer.as_asgi(), '/ws/device')

        with self.assertLogs(f'websocket.camera_consumer', level=logging.INFO) as logs:
            await communicator.connect(timeout=10)

        await communicator.disconnect()
        self.assertIn(f"INFO:websocket.camera_consumer:Doesn't work", logs.output)

websocket.camera_consumer.py

：

import logging
from channels.generic.websocket import AsyncWebsocketConsumer

def sync_function():
    pass

class TestConsumer(AsyncWebsocketConsumer):
    async def connect(self):
        await super().connect()

        logger = logging.getLogger(__name__)

        logger.info("Works")
        await sync_to_async(sync_function)()
        logger.info("Doesn't work")

到目前为止我学到了什么：

• 将 getLogger 调用移到

  connect

  函数之外并不能解决问题
• 将

  super

  调用移至

  sync_to_async

  调用下方确实修复它
• 删除

  sync_to_async

  函数（并使

  sync_function

  异步）也修复它
• 在

  sync_to_async

  函数调用过程中的某个地方，

  assertLogs

  注入的记录器处理程序对象被删除，因此第一个日志消息被路由到测试，但第二个日志消息则没有
• 我尝试单步执行sync_to_async调用，但是我不知道如何在异步循环内监视记录器对象（我对异步代码还比较陌生，所以不太熟悉如何它在幕后工作）。

Answer 1

乍一看，我发现你的测试存在问题： 您正在上下文之外使用

logs

var。 对于我来说，如果我想测试代码中的日志记录，我只需模拟模块中的记录器并检查它是否被正确调用。它总是有效并且易于理解。