From 92dee75c51badcceab2b6dc85ed36e028c08cd46 Mon Sep 17 00:00:00 2001 From: Cycloctane Date: Sat, 4 Oct 2025 19:44:05 +0800 Subject: [PATCH] Refactor websocket frame compression tests in test_websocket_writer (#11546) --- CHANGES/11546.contrib.rst | 2 ++ tests/test_websocket_writer.py | 33 +++++++++++++++++++++++++++++---- 2 files changed, 31 insertions(+), 4 deletions(-) create mode 100644 CHANGES/11546.contrib.rst diff --git a/CHANGES/11546.contrib.rst b/CHANGES/11546.contrib.rst new file mode 100644 index 00000000000..5fda1bdbb98 --- /dev/null +++ b/CHANGES/11546.contrib.rst @@ -0,0 +1,2 @@ +Fixed ``test_send_compress_text`` failing when alternative zlib implementation +is used. (``zlib-ng`` in python 3.14 windows build) -- by :user:`Cycloctane`. diff --git a/tests/test_websocket_writer.py b/tests/test_websocket_writer.py index 467ed8e94ce..313290349a5 100644 --- a/tests/test_websocket_writer.py +++ b/tests/test_websocket_writer.py @@ -8,6 +8,7 @@ from aiohttp import WSMsgType from aiohttp._websocket.reader import WebSocketDataQueue from aiohttp.base_protocol import BaseProtocol +from aiohttp.compression_utils import ZLibBackend from aiohttp.http import WebSocketReader, WebSocketWriter @@ -86,24 +87,48 @@ async def test_send_text_masked( writer.transport.write.assert_called_with(b"\x81\x84\rg\xb3fy\x02\xcb\x12") # type: ignore[attr-defined] +@pytest.mark.usefixtures("parametrize_zlib_backend") async def test_send_compress_text( protocol: BaseProtocol, transport: asyncio.Transport ) -> None: + compress_obj = ZLibBackend.compressobj(level=ZLibBackend.Z_BEST_SPEED, wbits=-15) writer = WebSocketWriter(protocol, transport, compress=15) + + msg = ( + compress_obj.compress(b"text") + compress_obj.flush(ZLibBackend.Z_SYNC_FLUSH) + ).removesuffix(b"\x00\x00\xff\xff") await writer.send_frame(b"text", WSMsgType.TEXT) - writer.transport.write.assert_called_with(b"\xc1\x06*I\xad(\x01\x00") # type: ignore[attr-defined] + writer.transport.write.assert_called_with( # type: ignore[attr-defined] + b"\xc1" + len(msg).to_bytes(1, "big") + msg + ) + + msg = ( + compress_obj.compress(b"text") + compress_obj.flush(ZLibBackend.Z_SYNC_FLUSH) + ).removesuffix(b"\x00\x00\xff\xff") await writer.send_frame(b"text", WSMsgType.TEXT) - writer.transport.write.assert_called_with(b"\xc1\x05*\x01b\x00\x00") # type: ignore[attr-defined] + writer.transport.write.assert_called_with( # type: ignore[attr-defined] + b"\xc1" + len(msg).to_bytes(1, "big") + msg + ) +@pytest.mark.usefixtures("parametrize_zlib_backend") async def test_send_compress_text_notakeover( protocol: BaseProtocol, transport: asyncio.Transport ) -> None: + compress_obj = ZLibBackend.compressobj(level=ZLibBackend.Z_BEST_SPEED, wbits=-15) writer = WebSocketWriter(protocol, transport, compress=15, notakeover=True) + + msg = ( + compress_obj.compress(b"text") + compress_obj.flush(ZLibBackend.Z_FULL_FLUSH) + ).removesuffix(b"\x00\x00\xff\xff") await writer.send_frame(b"text", WSMsgType.TEXT) - writer.transport.write.assert_called_with(b"\xc1\x06*I\xad(\x01\x00") # type: ignore[attr-defined] + writer.transport.write.assert_called_with( # type: ignore[attr-defined] + b"\xc1" + len(msg).to_bytes(1, "big") + msg + ) await writer.send_frame(b"text", WSMsgType.TEXT) - writer.transport.write.assert_called_with(b"\xc1\x06*I\xad(\x01\x00") # type: ignore[attr-defined] + writer.transport.write.assert_called_with( # type: ignore[attr-defined] + b"\xc1" + len(msg).to_bytes(1, "big") + msg + ) async def test_send_compress_text_per_message(