AsyncWebsocketConsumer中日志记录异常问题排查求助

测试错误信息 #

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
from asgiref.sync import sync_to_async

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().connect()调用移到sync_to_async调用之后可以解决问题
• 移除sync_to_async（将sync_function改为异步）也可以解决问题
• 推测sync_to_async调用过程中，assertLogs注入的日志处理器被移除，导致第一条日志能被测试捕获，第二条不行
• 尝试调试sync_to_async调用，但因对异步底层机制不熟悉，无法在异步循环中监控日志对象

问题原因及解决方法 #

问题核心是super().connect()与sync_to_async的线程切换冲突：调用super().connect()时，Channels会完成WebSocket握手流程，此时可能触发线程上下文切换，导致assertLogs临时添加的日志处理器被清理。后续sync_to_async执行同步函数后回到异步上下文时，日志处理器已丢失，第二条日志无法被捕获。

可行解决方案： #

1. 调整调用顺序：把super().connect()移到所有日志输出和sync_to_async调用之后，确保日志都在assertLogs的上下文内生成：

   async def connect(self):
       logger = logging.getLogger(__name__)
       logger.info("Works")
       await sync_to_async(sync_function)()
       logger.info("Doesn't work")
       await super().connect()  # 移至最后执行
   

2. 替换为异步函数：如果sync_function可以改写成异步实现，直接去掉sync_to_async包裹，避免线程切换带来的日志上下文问题。
3. 手动管理日志处理器：若必须保留原调用顺序，可在测试中手动添加/移除日志处理器，替代assertLogs的上下文管理：

   async def test_fixed(self):
       communicator = WebsocketCommunicator(TestConsumer.as_asgi(), '/ws/device')
       logger = logging.getLogger('websocket.camera_consumer')
       log_capture = logging.handlers.MemoryHandler(1024)
       logger.addHandler(log_capture)
       logger.setLevel(logging.INFO)
       
       try:
           await communicator.connect(timeout=10)
           logs = [f"{record.levelname}:{record.name}:{record.getMessage()}" for record in log_capture.buffer]
           self.assertIn("INFO:websocket.camera_consumer:Doesn't work", logs)
       finally:
           logger.removeHandler(log_capture)
           await communicator.disconnect()
   

内容的提问来源于stack exchange，提问作者Kenny Loveall