From 443ba23b70cad371255f7d3f8b919c71cff9fcf8 Mon Sep 17 00:00:00 2001 From: Jeffrey 'Alex' Clark Date: Tue, 14 Jul 2026 11:56:24 -0400 Subject: [PATCH 1/5] PYTHON-5724 Add integration test exercising compression logic --- test/asynchronous/test_client.py | 50 +++++++++++++++++++++++++++++++- test/test_client.py | 50 +++++++++++++++++++++++++++++++- 2 files changed, 98 insertions(+), 2 deletions(-) diff --git a/test/asynchronous/test_client.py b/test/asynchronous/test_client.py index a7f66d9620..f6e6d7b722 100644 --- a/test/asynchronous/test_client.py +++ b/test/asynchronous/test_client.py @@ -71,7 +71,13 @@ from pymongo.asynchronous.topology import _ErrorContext from pymongo.client_options import ClientOptions from pymongo.common import _UUID_REPRESENTATIONS, CONNECT_TIMEOUT, MIN_SUPPORTED_WIRE_VERSION, has_c -from pymongo.compression_support import _have_snappy, _have_zstd +from pymongo.compression_support import ( + SnappyContext, + ZlibContext, + ZstdContext, + _have_snappy, + _have_zstd, +) from pymongo.driver_info import DriverInfo from pymongo.errors import ( AutoReconnect, @@ -1812,6 +1818,48 @@ def compression_settings(client): # No error await client.pymongo_test.test.find_one() + async def test_compression_commands(self): + # Ensure the compression logic is actually exercised end-to-end by + # sending commands with each available compressor negotiated. + candidates: list[tuple[str, type]] = [("zlib", ZlibContext)] + if _have_snappy(): + candidates.append(("snappy", SnappyContext)) + if _have_zstd(): + candidates.append(("zstd", ZstdContext)) + + for name, ctx_type in candidates: + with self.subTest(compressor=name): + # maxPoolSize=1 ensures the operations below reuse the same + # connection the spy is installed on. + client = await self.async_single_client(compressors=name, maxPoolSize=1) + # Trigger the connection handshake so the compressor is negotiated. + await client.admin.command("ping") + pool = await async_get_pool(client) + async with pool.checkout() as conn: + if conn.compression_context is None: + self.skipTest(f"server did not negotiate {name} compression") + self.assertIsInstance(conn.compression_context, ctx_type) + + # Spy on the compress method to confirm the outgoing message + # is actually compressed. + compressed = [] + original = conn.compression_context.compress + + def spy(data, _original=original, _sink=compressed): + _sink.append(data) + return _original(data) + + conn.compression_context.compress = spy + + # Round-trip a command large enough to compress. The response is + # decompressed via compression_support.decompress. + coll = client.pymongo_test.test_compression + await coll.delete_many({}) + await coll.insert_one({"x": "y" * 1024}) + doc = await coll.find_one({}, {"_id": 0}) + self.assertEqual(doc, {"x": "y" * 1024}) + self.assertTrue(compressed, "compress() was never called") + @async_client_context.require_sync async def test_reset_during_update_pool(self): client = await self.async_rs_or_single_client(minPoolSize=10) diff --git a/test/test_client.py b/test/test_client.py index 8f0da71321..0af6619dd6 100644 --- a/test/test_client.py +++ b/test/test_client.py @@ -60,7 +60,13 @@ from pymongo import event_loggers, message, monitoring from pymongo.client_options import ClientOptions from pymongo.common import _UUID_REPRESENTATIONS, CONNECT_TIMEOUT, MIN_SUPPORTED_WIRE_VERSION, has_c -from pymongo.compression_support import _have_snappy, _have_zstd +from pymongo.compression_support import ( + SnappyContext, + ZlibContext, + ZstdContext, + _have_snappy, + _have_zstd, +) from pymongo.driver_info import DriverInfo from pymongo.errors import ( AutoReconnect, @@ -1769,6 +1775,48 @@ def compression_settings(client): # No error client.pymongo_test.test.find_one() + def test_compression_commands(self): + # Ensure the compression logic is actually exercised end-to-end by + # sending commands with each available compressor negotiated. + candidates: list[tuple[str, type]] = [("zlib", ZlibContext)] + if _have_snappy(): + candidates.append(("snappy", SnappyContext)) + if _have_zstd(): + candidates.append(("zstd", ZstdContext)) + + for name, ctx_type in candidates: + with self.subTest(compressor=name): + # maxPoolSize=1 ensures the operations below reuse the same + # connection the spy is installed on. + client = self.single_client(compressors=name, maxPoolSize=1) + # Trigger the connection handshake so the compressor is negotiated. + client.admin.command("ping") + pool = get_pool(client) + with pool.checkout() as conn: + if conn.compression_context is None: + self.skipTest(f"server did not negotiate {name} compression") + self.assertIsInstance(conn.compression_context, ctx_type) + + # Spy on the compress method to confirm the outgoing message + # is actually compressed. + compressed = [] + original = conn.compression_context.compress + + def spy(data, _original=original, _sink=compressed): + _sink.append(data) + return _original(data) + + conn.compression_context.compress = spy + + # Round-trip a command large enough to compress. The response is + # decompressed via compression_support.decompress. + coll = client.pymongo_test.test_compression + coll.delete_many({}) + coll.insert_one({"x": "y" * 1024}) + doc = coll.find_one({}, {"_id": 0}) + self.assertEqual(doc, {"x": "y" * 1024}) + self.assertTrue(compressed, "compress() was never called") + @client_context.require_sync def test_reset_during_update_pool(self): client = self.rs_or_single_client(minPoolSize=10) From 35bd9b0a8cc65c4009b333643f38019d16c564cb Mon Sep 17 00:00:00 2001 From: Jeffrey 'Alex' Clark Date: Mon, 27 Jul 2026 13:28:42 -0400 Subject: [PATCH 2/5] PYTHON-5724 Assert response decompression and skip when compression unnegotiated --- test/asynchronous/test_client.py | 33 ++++++++++++++++++++++++-------- test/test_client.py | 33 ++++++++++++++++++++++++-------- 2 files changed, 50 insertions(+), 16 deletions(-) diff --git a/test/asynchronous/test_client.py b/test/asynchronous/test_client.py index 4529dc6be2..83ccfac7b3 100644 --- a/test/asynchronous/test_client.py +++ b/test/asynchronous/test_client.py @@ -58,7 +58,7 @@ ) from bson.son import SON from bson.tz_util import utc -from pymongo import event_loggers, message, monitoring +from pymongo import event_loggers, message, monitoring, network_layer from pymongo.asynchronous.command_cursor import AsyncCommandCursor from pymongo.asynchronous.cursor import AsyncCursor, CursorType from pymongo.asynchronous.database import AsyncDatabase @@ -1862,17 +1862,19 @@ async def test_compression_commands(self): if _have_zstd(): candidates.append(("zstd", ZstdContext)) + negotiated = [] for name, ctx_type in candidates: with self.subTest(compressor=name): # maxPoolSize=1 ensures the operations below reuse the same - # connection the spy is installed on. + # connection the spy is installed on, unless it is replaced. client = await self.async_single_client(compressors=name, maxPoolSize=1) # Trigger the connection handshake so the compressor is negotiated. await client.admin.command("ping") pool = await async_get_pool(client) async with pool.checkout() as conn: if conn.compression_context is None: - self.skipTest(f"server did not negotiate {name} compression") + continue + negotiated.append(name) self.assertIsInstance(conn.compression_context, ctx_type) # Spy on the compress method to confirm the outgoing message @@ -1886,14 +1888,29 @@ def spy(data, _original=original, _sink=compressed): conn.compression_context.compress = spy - # Round-trip a command large enough to compress. The response is - # decompressed via compression_support.decompress. + # Spy on the read path's decompress() to confirm the server's + # replies are actually compressed too. + decompressed = [] + original_decompress = network_layer.decompress + + def decompress_spy(data, compressor_id, _sink=decompressed): + _sink.append(compressor_id) + return original_decompress(data, compressor_id) + + # Round-trip a command large enough to compress. coll = client.pymongo_test.test_compression - await coll.delete_many({}) - await coll.insert_one({"x": "y" * 1024}) - doc = await coll.find_one({}, {"_id": 0}) + await coll.drop() + with patch.object(network_layer, "decompress", decompress_spy): + await coll.insert_one({"x": "y" * 1024}) + doc = await coll.find_one({}, {"_id": 0}) self.assertEqual(doc, {"x": "y" * 1024}) self.assertTrue(compressed, "compress() was never called") + self.assertTrue(decompressed, "decompress() was never called") + self.assertEqual(set(decompressed), {ctx_type.compressor_id}) + await coll.drop() + + if not negotiated: + self.skipTest("server did not negotiate compression for any compressor") @async_client_context.require_sync async def test_reset_during_update_pool(self): diff --git a/test/test_client.py b/test/test_client.py index c7c68058c9..152027b977 100644 --- a/test/test_client.py +++ b/test/test_client.py @@ -57,7 +57,7 @@ ) from bson.son import SON from bson.tz_util import utc -from pymongo import event_loggers, message, monitoring +from pymongo import event_loggers, message, monitoring, network_layer from pymongo.client_options import ClientOptions from pymongo.common import _UUID_REPRESENTATIONS, CONNECT_TIMEOUT, MIN_SUPPORTED_WIRE_VERSION, has_c from pymongo.compression_support import ( @@ -1819,17 +1819,19 @@ def test_compression_commands(self): if _have_zstd(): candidates.append(("zstd", ZstdContext)) + negotiated = [] for name, ctx_type in candidates: with self.subTest(compressor=name): # maxPoolSize=1 ensures the operations below reuse the same - # connection the spy is installed on. + # connection the spy is installed on, unless it is replaced. client = self.single_client(compressors=name, maxPoolSize=1) # Trigger the connection handshake so the compressor is negotiated. client.admin.command("ping") pool = get_pool(client) with pool.checkout() as conn: if conn.compression_context is None: - self.skipTest(f"server did not negotiate {name} compression") + continue + negotiated.append(name) self.assertIsInstance(conn.compression_context, ctx_type) # Spy on the compress method to confirm the outgoing message @@ -1843,14 +1845,29 @@ def spy(data, _original=original, _sink=compressed): conn.compression_context.compress = spy - # Round-trip a command large enough to compress. The response is - # decompressed via compression_support.decompress. + # Spy on the read path's decompress() to confirm the server's + # replies are actually compressed too. + decompressed = [] + original_decompress = network_layer.decompress + + def decompress_spy(data, compressor_id, _sink=decompressed): + _sink.append(compressor_id) + return original_decompress(data, compressor_id) + + # Round-trip a command large enough to compress. coll = client.pymongo_test.test_compression - coll.delete_many({}) - coll.insert_one({"x": "y" * 1024}) - doc = coll.find_one({}, {"_id": 0}) + coll.drop() + with patch.object(network_layer, "decompress", decompress_spy): + coll.insert_one({"x": "y" * 1024}) + doc = coll.find_one({}, {"_id": 0}) self.assertEqual(doc, {"x": "y" * 1024}) self.assertTrue(compressed, "compress() was never called") + self.assertTrue(decompressed, "decompress() was never called") + self.assertEqual(set(decompressed), {ctx_type.compressor_id}) + coll.drop() + + if not negotiated: + self.skipTest("server did not negotiate compression for any compressor") @client_context.require_sync def test_reset_during_update_pool(self): From 5b2407949fc08a805eafcc99b251f5a1b41aac18 Mon Sep 17 00:00:00 2001 From: Jeffrey 'Alex' Clark Date: Mon, 27 Jul 2026 13:58:21 -0400 Subject: [PATCH 3/5] PYTHON-5724 Rename spy sink args to _recorded --- test/asynchronous/test_client.py | 10 ++++++---- test/test_client.py | 10 ++++++---- 2 files changed, 12 insertions(+), 8 deletions(-) diff --git a/test/asynchronous/test_client.py b/test/asynchronous/test_client.py index 83ccfac7b3..6eb2a41d81 100644 --- a/test/asynchronous/test_client.py +++ b/test/asynchronous/test_client.py @@ -1882,8 +1882,10 @@ async def test_compression_commands(self): compressed = [] original = conn.compression_context.compress - def spy(data, _original=original, _sink=compressed): - _sink.append(data) + # Default args bind the current iteration's values so the + # closure does not late-bind the loop variables. + def spy(data, _original=original, _recorded=compressed): + _recorded.append(data) return _original(data) conn.compression_context.compress = spy @@ -1893,8 +1895,8 @@ def spy(data, _original=original, _sink=compressed): decompressed = [] original_decompress = network_layer.decompress - def decompress_spy(data, compressor_id, _sink=decompressed): - _sink.append(compressor_id) + def decompress_spy(data, compressor_id, _recorded=decompressed): + _recorded.append(compressor_id) return original_decompress(data, compressor_id) # Round-trip a command large enough to compress. diff --git a/test/test_client.py b/test/test_client.py index 152027b977..5becda94ab 100644 --- a/test/test_client.py +++ b/test/test_client.py @@ -1839,8 +1839,10 @@ def test_compression_commands(self): compressed = [] original = conn.compression_context.compress - def spy(data, _original=original, _sink=compressed): - _sink.append(data) + # Default args bind the current iteration's values so the + # closure does not late-bind the loop variables. + def spy(data, _original=original, _recorded=compressed): + _recorded.append(data) return _original(data) conn.compression_context.compress = spy @@ -1850,8 +1852,8 @@ def spy(data, _original=original, _sink=compressed): decompressed = [] original_decompress = network_layer.decompress - def decompress_spy(data, compressor_id, _sink=decompressed): - _sink.append(compressor_id) + def decompress_spy(data, compressor_id, _recorded=decompressed): + _recorded.append(compressor_id) return original_decompress(data, compressor_id) # Round-trip a command large enough to compress. From 621d1eb3da92a5731a9610b667faca746dea74d2 Mon Sep 17 00:00:00 2001 From: Jeffrey 'Alex' Clark Date: Mon, 27 Jul 2026 13:59:42 -0400 Subject: [PATCH 4/5] PYTHON-5724 Bind original decompress via default arg for consistency --- test/asynchronous/test_client.py | 6 ++++-- test/test_client.py | 6 ++++-- 2 files changed, 8 insertions(+), 4 deletions(-) diff --git a/test/asynchronous/test_client.py b/test/asynchronous/test_client.py index 6eb2a41d81..e5f3d63d11 100644 --- a/test/asynchronous/test_client.py +++ b/test/asynchronous/test_client.py @@ -1895,9 +1895,11 @@ def spy(data, _original=original, _recorded=compressed): decompressed = [] original_decompress = network_layer.decompress - def decompress_spy(data, compressor_id, _recorded=decompressed): + def decompress_spy( + data, compressor_id, _original=original_decompress, _recorded=decompressed + ): _recorded.append(compressor_id) - return original_decompress(data, compressor_id) + return _original(data, compressor_id) # Round-trip a command large enough to compress. coll = client.pymongo_test.test_compression diff --git a/test/test_client.py b/test/test_client.py index 5becda94ab..882be439e6 100644 --- a/test/test_client.py +++ b/test/test_client.py @@ -1852,9 +1852,11 @@ def spy(data, _original=original, _recorded=compressed): decompressed = [] original_decompress = network_layer.decompress - def decompress_spy(data, compressor_id, _recorded=decompressed): + def decompress_spy( + data, compressor_id, _original=original_decompress, _recorded=decompressed + ): _recorded.append(compressor_id) - return original_decompress(data, compressor_id) + return _original(data, compressor_id) # Round-trip a command large enough to compress. coll = client.pymongo_test.test_compression From b14b5cc118fdfc5040461c30542d689a2c501d1e Mon Sep 17 00:00:00 2001 From: Jeffrey 'Alex' Clark Date: Mon, 27 Jul 2026 14:12:16 -0400 Subject: [PATCH 5/5] PYTHON-5724 Close each client per subtest to isolate patched decompress --- test/asynchronous/test_client.py | 94 +++++++++++++++++--------------- test/test_client.py | 94 +++++++++++++++++--------------- 2 files changed, 100 insertions(+), 88 deletions(-) diff --git a/test/asynchronous/test_client.py b/test/asynchronous/test_client.py index e5f3d63d11..be3f8545de 100644 --- a/test/asynchronous/test_client.py +++ b/test/asynchronous/test_client.py @@ -1868,50 +1868,56 @@ async def test_compression_commands(self): # maxPoolSize=1 ensures the operations below reuse the same # connection the spy is installed on, unless it is replaced. client = await self.async_single_client(compressors=name, maxPoolSize=1) - # Trigger the connection handshake so the compressor is negotiated. - await client.admin.command("ping") - pool = await async_get_pool(client) - async with pool.checkout() as conn: - if conn.compression_context is None: - continue - negotiated.append(name) - self.assertIsInstance(conn.compression_context, ctx_type) - - # Spy on the compress method to confirm the outgoing message - # is actually compressed. - compressed = [] - original = conn.compression_context.compress - - # Default args bind the current iteration's values so the - # closure does not late-bind the loop variables. - def spy(data, _original=original, _recorded=compressed): - _recorded.append(data) - return _original(data) - - conn.compression_context.compress = spy - - # Spy on the read path's decompress() to confirm the server's - # replies are actually compressed too. - decompressed = [] - original_decompress = network_layer.decompress - - def decompress_spy( - data, compressor_id, _original=original_decompress, _recorded=decompressed - ): - _recorded.append(compressor_id) - return _original(data, compressor_id) - - # Round-trip a command large enough to compress. - coll = client.pymongo_test.test_compression - await coll.drop() - with patch.object(network_layer, "decompress", decompress_spy): - await coll.insert_one({"x": "y" * 1024}) - doc = await coll.find_one({}, {"_id": 0}) - self.assertEqual(doc, {"x": "y" * 1024}) - self.assertTrue(compressed, "compress() was never called") - self.assertTrue(decompressed, "decompress() was never called") - self.assertEqual(set(decompressed), {ctx_type.compressor_id}) - await coll.drop() + # Close each client before moving on: decompress() is patched + # globally below, so app traffic from a client left over from an + # earlier subtest could otherwise pollute the recorded ids. + try: + # Trigger the connection handshake so the compressor is negotiated. + await client.admin.command("ping") + pool = await async_get_pool(client) + async with pool.checkout() as conn: + if conn.compression_context is None: + continue + negotiated.append(name) + self.assertIsInstance(conn.compression_context, ctx_type) + + # Spy on the compress method to confirm the outgoing message + # is actually compressed. + compressed = [] + original = conn.compression_context.compress + + # Default args bind the current iteration's values so the + # closure does not late-bind the loop variables. + def spy(data, _original=original, _recorded=compressed): + _recorded.append(data) + return _original(data) + + conn.compression_context.compress = spy + + # Spy on the read path's decompress() to confirm the server's + # replies are actually compressed too. + decompressed = [] + original_decompress = network_layer.decompress + + def decompress_spy( + data, compressor_id, _original=original_decompress, _recorded=decompressed + ): + _recorded.append(compressor_id) + return _original(data, compressor_id) + + # Round-trip a command large enough to compress. + coll = client.pymongo_test.test_compression + await coll.drop() + with patch.object(network_layer, "decompress", decompress_spy): + await coll.insert_one({"x": "y" * 1024}) + doc = await coll.find_one({}, {"_id": 0}) + self.assertEqual(doc, {"x": "y" * 1024}) + self.assertTrue(compressed, "compress() was never called") + self.assertTrue(decompressed, "decompress() was never called") + self.assertEqual(set(decompressed), {ctx_type.compressor_id}) + await coll.drop() + finally: + await client.close() if not negotiated: self.skipTest("server did not negotiate compression for any compressor") diff --git a/test/test_client.py b/test/test_client.py index 882be439e6..befde70408 100644 --- a/test/test_client.py +++ b/test/test_client.py @@ -1825,50 +1825,56 @@ def test_compression_commands(self): # maxPoolSize=1 ensures the operations below reuse the same # connection the spy is installed on, unless it is replaced. client = self.single_client(compressors=name, maxPoolSize=1) - # Trigger the connection handshake so the compressor is negotiated. - client.admin.command("ping") - pool = get_pool(client) - with pool.checkout() as conn: - if conn.compression_context is None: - continue - negotiated.append(name) - self.assertIsInstance(conn.compression_context, ctx_type) - - # Spy on the compress method to confirm the outgoing message - # is actually compressed. - compressed = [] - original = conn.compression_context.compress - - # Default args bind the current iteration's values so the - # closure does not late-bind the loop variables. - def spy(data, _original=original, _recorded=compressed): - _recorded.append(data) - return _original(data) - - conn.compression_context.compress = spy - - # Spy on the read path's decompress() to confirm the server's - # replies are actually compressed too. - decompressed = [] - original_decompress = network_layer.decompress - - def decompress_spy( - data, compressor_id, _original=original_decompress, _recorded=decompressed - ): - _recorded.append(compressor_id) - return _original(data, compressor_id) - - # Round-trip a command large enough to compress. - coll = client.pymongo_test.test_compression - coll.drop() - with patch.object(network_layer, "decompress", decompress_spy): - coll.insert_one({"x": "y" * 1024}) - doc = coll.find_one({}, {"_id": 0}) - self.assertEqual(doc, {"x": "y" * 1024}) - self.assertTrue(compressed, "compress() was never called") - self.assertTrue(decompressed, "decompress() was never called") - self.assertEqual(set(decompressed), {ctx_type.compressor_id}) - coll.drop() + # Close each client before moving on: decompress() is patched + # globally below, so app traffic from a client left over from an + # earlier subtest could otherwise pollute the recorded ids. + try: + # Trigger the connection handshake so the compressor is negotiated. + client.admin.command("ping") + pool = get_pool(client) + with pool.checkout() as conn: + if conn.compression_context is None: + continue + negotiated.append(name) + self.assertIsInstance(conn.compression_context, ctx_type) + + # Spy on the compress method to confirm the outgoing message + # is actually compressed. + compressed = [] + original = conn.compression_context.compress + + # Default args bind the current iteration's values so the + # closure does not late-bind the loop variables. + def spy(data, _original=original, _recorded=compressed): + _recorded.append(data) + return _original(data) + + conn.compression_context.compress = spy + + # Spy on the read path's decompress() to confirm the server's + # replies are actually compressed too. + decompressed = [] + original_decompress = network_layer.decompress + + def decompress_spy( + data, compressor_id, _original=original_decompress, _recorded=decompressed + ): + _recorded.append(compressor_id) + return _original(data, compressor_id) + + # Round-trip a command large enough to compress. + coll = client.pymongo_test.test_compression + coll.drop() + with patch.object(network_layer, "decompress", decompress_spy): + coll.insert_one({"x": "y" * 1024}) + doc = coll.find_one({}, {"_id": 0}) + self.assertEqual(doc, {"x": "y" * 1024}) + self.assertTrue(compressed, "compress() was never called") + self.assertTrue(decompressed, "decompress() was never called") + self.assertEqual(set(decompressed), {ctx_type.compressor_id}) + coll.drop() + finally: + client.close() if not negotiated: self.skipTest("server did not negotiate compression for any compressor")