From e8a795e6d93c0f04d5e07dcf4c19718dab4343dc Mon Sep 17 00:00:00 2001 From: cacosandon Date: Mon, 2 Sep 2024 23:47:58 -0400 Subject: [PATCH] tests(channels/generic): regression test for double check of text message None --- tests/test_generic_websocket.py | 41 +++++++++++++++++++++++++++++++++ 1 file changed, 41 insertions(+) diff --git a/tests/test_generic_websocket.py b/tests/test_generic_websocket.py index 73cdb4862..2b85bc884 100644 --- a/tests/test_generic_websocket.py +++ b/tests/test_generic_websocket.py @@ -478,3 +478,44 @@ async def connect(self): assert msg["type"] == "websocket.close" assert msg["code"] == 4007 assert msg["reason"] == "test reason" + +@pytest.mark.django_db +@pytest.mark.asyncio +async def test_websocket_receive_with_none_text(): + """ + Tests that the receive method handles messages with None text data correctly. + """ + + class TestConsumer(WebsocketConsumer): + def receive(self, text_data=None, bytes_data=None): + if text_data: + self.send(text_data="Received text: " + text_data) + elif bytes_data: + self.send(text_data=f"Received bytes of length: {len(bytes_data)}") + + app = TestConsumer() + + # Open a connection + communicator = WebsocketCommunicator(app, "/testws/") + connected, _ = await communicator.connect() + assert connected + + # Simulate Hypercorn behavior (both 'text' and 'bytes' keys present, but 'text' is None) + await communicator.send_input({"type": "websocket.receive", "text": None, "bytes": b"test data"}) + response = await communicator.receive_output() + assert response["type"] == "websocket.send" + assert response["text"] == "Received bytes of length: 9" + + # Test with only 'bytes' key (simulating uvicorn/daphne behavior) + await communicator.send_input({"type": "websocket.receive", "bytes": b"more data"}) + response = await communicator.receive_output() + assert response["type"] == "websocket.send" + assert response["text"] == "Received bytes of length: 9" + + # Test with valid text data + await communicator.send_input({"type": "websocket.receive", "text": "Hello, world!"}) + response = await communicator.receive_output() + assert response["type"] == "websocket.send" + assert response["text"] == "Received text: Hello, world!" + + await communicator.disconnect()