I have a Django project that's using Channels for websocket communication and I've stumbled upon what I think might be a bug, but I'm not certain so I'm hoping someone that understands this better than I do can help explain what's happening here:
Test error message:
AssertionError: "INFO:websocket.camera_consumer:Doesn't work" not found in ['INFO:websocket.camera_consumer:Works']
Test that's failing:
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")
What I've learned so far:
- Moving the getLogger call outside the
connectfunction doesn't fix it - Moving the
supercall below thesync_to_asynccall does fix it - Removing the
sync_to_asyncfunction (and makingsync_functionasync) also fixes it - Somewhere along the way in the
sync_to_asyncfunction call, the logger handler object that theassertLogsinjects gets removed so the first log message gets routed to the test, but the second log message does not - I've attempted to step-though the sync_to_async call, however I can't figure out how to monitor the logger object while inside the async loop (I'm still relatively new-ish to async code so not super well versed in how it works under the hood).