From 19d690548b92b71dd3abdc65553d422cb2bceea4 Mon Sep 17 00:00:00 2001 From: Jeffrey 'Alex' Clark Date: Thu, 23 Apr 2026 17:29:52 -0400 Subject: [PATCH 1/5] PYTHON-5779 Increase code coverage for compression_support.py from 64.36% to >=80% --- test/test_compression_support.py | 255 +++++++++++++++++++++++++++++++ 1 file changed, 255 insertions(+) create mode 100644 test/test_compression_support.py diff --git a/test/test_compression_support.py b/test/test_compression_support.py new file mode 100644 index 0000000000..810221af0a --- /dev/null +++ b/test/test_compression_support.py @@ -0,0 +1,255 @@ +# Copyright 2026-present MongoDB, Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Unit tests for compression_support.py.""" + +from __future__ import annotations + +import sys +import unittest +from unittest.mock import patch + +sys.path[0:0] = [""] + +from pymongo.compression_support import ( + CompressionSettings, + SnappyContext, + ZlibContext, + ZstdContext, + _have_snappy, + _have_zlib, + _have_zstd, + decompress, + validate_compressors, + validate_zlib_compression_level, +) + + +class TestHaveSnappy(unittest.TestCase): + def test_returns_true_when_available(self): + try: + import snappy + except ImportError: + self.skipTest("python-snappy not installed") + self.assertTrue(_have_snappy()) + + def test_returns_false_on_import_error(self): + with patch.dict(sys.modules, {"snappy": None}): + self.assertFalse(_have_snappy()) + + +class TestHaveZlib(unittest.TestCase): + def test_returns_true_when_available(self): + self.assertTrue(_have_zlib()) + + def test_returns_false_on_import_error(self): + with patch.dict(sys.modules, {"zlib": None}): + self.assertFalse(_have_zlib()) + + +class TestHaveZstd(unittest.TestCase): + def test_returns_bool(self): + result = _have_zstd() + self.assertIsInstance(result, bool) + + def test_returns_false_when_unavailable_pre_314(self): + if sys.version_info >= (3, 14): + self.skipTest("Python 3.14+ uses compression.zstd") + with patch.dict(sys.modules, {"backports": None, "backports.zstd": None}): + self.assertFalse(_have_zstd()) + + def test_returns_false_when_unavailable_314_plus(self): + if sys.version_info < (3, 14): + self.skipTest("Only applies to Python 3.14+") + with patch.dict(sys.modules, {"compression": None, "compression.zstd": None}): + self.assertFalse(_have_zstd()) + + +class TestValidateCompressors(unittest.TestCase): + def test_string_input_single(self): + result = validate_compressors(None, "zlib") + self.assertEqual(result, ["zlib"]) + + def test_string_input_comma_separated(self): + result = validate_compressors(None, "zlib,snappy") + self.assertIn("zlib", result) + + def test_iterable_input(self): + result = validate_compressors(None, ["zlib"]) + self.assertEqual(result, ["zlib"]) + + def test_unsupported_compressor_warns_and_removes(self): + with self.assertWarns(UserWarning) as ctx: + result = validate_compressors(None, ["bogus"]) + self.assertEqual(result, []) + self.assertIn("Unsupported compressor: bogus", str(ctx.warning)) + + def test_snappy_unavailable_warns_and_removes(self): + with patch("pymongo.compression_support._have_snappy", return_value=False): + with self.assertWarns(UserWarning) as ctx: + result = validate_compressors(None, ["snappy"]) + self.assertEqual(result, []) + self.assertIn("python-snappy", str(ctx.warning)) + + def test_zlib_unavailable_warns_and_removes(self): + with patch("pymongo.compression_support._have_zlib", return_value=False): + with self.assertWarns(UserWarning) as ctx: + result = validate_compressors(None, ["zlib"]) + self.assertEqual(result, []) + self.assertIn("zlib", str(ctx.warning)) + + def test_zstd_unavailable_warns_and_removes_pre_314(self): + if sys.version_info >= (3, 14): + self.skipTest("Python 3.14+ uses different warning message") + with patch("pymongo.compression_support._have_zstd", return_value=False): + with self.assertWarns(UserWarning) as ctx: + result = validate_compressors(None, ["zstd"]) + self.assertEqual(result, []) + self.assertIn("backports.zstd", str(ctx.warning)) + + def test_zstd_unavailable_warns_and_removes_314_plus(self): + if sys.version_info < (3, 14): + self.skipTest("Only applies to Python 3.14+") + with patch("pymongo.compression_support._have_zstd", return_value=False): + with self.assertWarns(UserWarning) as ctx: + result = validate_compressors(None, ["zstd"]) + self.assertEqual(result, []) + self.assertIn("compression.zstd", str(ctx.warning)) + + def test_valid_zlib_always_included(self): + result = validate_compressors(None, ["zlib"]) + self.assertEqual(result, ["zlib"]) + + def test_multiple_valid_compressors(self): + result = validate_compressors(None, ["zlib"]) + self.assertIn("zlib", result) + + def test_empty_list_returns_empty(self): + result = validate_compressors(None, []) + self.assertEqual(result, []) + + +class TestValidateZlibCompressionLevel(unittest.TestCase): + def test_valid_minimum(self): + self.assertEqual(validate_zlib_compression_level("level", -1), -1) + + def test_valid_zero(self): + self.assertEqual(validate_zlib_compression_level("level", 0), 0) + + def test_valid_maximum(self): + self.assertEqual(validate_zlib_compression_level("level", 9), 9) + + def test_valid_midrange(self): + self.assertEqual(validate_zlib_compression_level("level", 5), 5) + + def test_non_integer_raises_type_error(self): + with self.assertRaises(TypeError) as ctx: + validate_zlib_compression_level("level", "abc") + self.assertIn("must be an integer", str(ctx.exception)) + + def test_too_low_raises_value_error(self): + with self.assertRaises(ValueError) as ctx: + validate_zlib_compression_level("level", -2) + self.assertIn("must be between -1 and 9", str(ctx.exception)) + + def test_too_high_raises_value_error(self): + with self.assertRaises(ValueError) as ctx: + validate_zlib_compression_level("level", 10) + self.assertIn("must be between -1 and 9", str(ctx.exception)) + + def test_string_integer_is_coerced(self): + self.assertEqual(validate_zlib_compression_level("level", "5"), 5) + + +class TestCompressionSettings(unittest.TestCase): + def _make(self, compressors=None, level=-1): + return CompressionSettings(compressors or [], level) + + def test_get_context_none_when_empty(self): + settings = self._make() + self.assertIsNone(settings.get_compression_context([])) + + def test_get_context_none_when_none(self): + settings = self._make() + self.assertIsNone(settings.get_compression_context(None)) + + def test_get_context_snappy(self): + settings = self._make() + ctx = settings.get_compression_context(["snappy"]) + self.assertIsInstance(ctx, SnappyContext) + + def test_get_context_zlib(self): + settings = self._make(level=6) + ctx = settings.get_compression_context(["zlib"]) + self.assertIsInstance(ctx, ZlibContext) + self.assertEqual(ctx.level, 6) + + def test_get_context_zstd(self): + settings = self._make() + ctx = settings.get_compression_context(["zstd"]) + self.assertIsInstance(ctx, ZstdContext) + + def test_get_context_uses_first_compressor(self): + settings = self._make(level=1) + ctx = settings.get_compression_context(["zlib", "snappy"]) + self.assertIsInstance(ctx, ZlibContext) + + def test_get_context_unknown_returns_none(self): + settings = self._make() + ctx = settings.get_compression_context(["unknown"]) + self.assertIsNone(ctx) + + +class TestZlibContext(unittest.TestCase): + def test_compress_and_decompress_roundtrip(self): + import zlib + + ctx = ZlibContext(level=-1) + data = b"hello world" * 100 + compressed = ctx.compress(data) + self.assertEqual(zlib.decompress(compressed), data) + + def test_compress_level_stored(self): + ctx = ZlibContext(level=6) + self.assertEqual(ctx.level, 6) + + def test_compressor_id(self): + self.assertEqual(ZlibContext.compressor_id, 2) + + +class TestDecompress(unittest.TestCase): + def test_unknown_compressor_id_raises(self): + with self.assertRaises(ValueError) as ctx: + decompress(b"data", 99) + self.assertIn("Unknown compressorId 99", str(ctx.exception)) + + def test_zlib_roundtrip(self): + import zlib + + data = b"hello world" + compressed = zlib.compress(data) + result = decompress(compressed, ZlibContext.compressor_id) + self.assertEqual(result, data) + + def test_zlib_with_memoryview(self): + import zlib + + data = b"test data" + compressed = zlib.compress(data) + result = decompress(memoryview(compressed), ZlibContext.compressor_id) + self.assertEqual(result, data) + + +if __name__ == "__main__": + unittest.main() From 10598e3f7b4bd44167535b20338fd1fd5ac16e76 Mon Sep 17 00:00:00 2001 From: Jeffrey 'Alex' Clark Date: Thu, 23 Apr 2026 18:44:14 -0400 Subject: [PATCH 2/5] PYTHON-5779 Use 'from test import unittest' to match test suite convention --- test/test_compression_support.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/test/test_compression_support.py b/test/test_compression_support.py index 810221af0a..c2e7d09480 100644 --- a/test/test_compression_support.py +++ b/test/test_compression_support.py @@ -17,11 +17,12 @@ from __future__ import annotations import sys -import unittest from unittest.mock import patch sys.path[0:0] = [""] +from test import unittest + from pymongo.compression_support import ( CompressionSettings, SnappyContext, From 0721490be4e799e2a8d25948e0cd4e68e72eb437 Mon Sep 17 00:00:00 2001 From: Jeffrey 'Alex' Clark Date: Thu, 23 Apr 2026 19:37:48 -0400 Subject: [PATCH 3/5] PYTHON-5779 Fix environment-dependent snappy test, strengthen multiple-compressor test --- test/test_compression_support.py | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/test/test_compression_support.py b/test/test_compression_support.py index c2e7d09480..b1cf573983 100644 --- a/test/test_compression_support.py +++ b/test/test_compression_support.py @@ -83,8 +83,10 @@ def test_string_input_single(self): self.assertEqual(result, ["zlib"]) def test_string_input_comma_separated(self): - result = validate_compressors(None, "zlib,snappy") + with patch("pymongo.compression_support._have_snappy", return_value=True): + result = validate_compressors(None, "zlib,snappy") self.assertIn("zlib", result) + self.assertIn("snappy", result) def test_iterable_input(self): result = validate_compressors(None, ["zlib"]) @@ -132,9 +134,10 @@ def test_valid_zlib_always_included(self): result = validate_compressors(None, ["zlib"]) self.assertEqual(result, ["zlib"]) - def test_multiple_valid_compressors(self): - result = validate_compressors(None, ["zlib"]) - self.assertIn("zlib", result) + def test_multiple_valid_compressors_preserves_order(self): + with patch("pymongo.compression_support._have_snappy", return_value=True): + result = validate_compressors(None, ["zlib", "snappy"]) + self.assertEqual(result, ["zlib", "snappy"]) def test_empty_list_returns_empty(self): result = validate_compressors(None, []) From 9b482d5110c36def27e5feb820b2e0dac9d172b0 Mon Sep 17 00:00:00 2001 From: Jeffrey 'Alex' Clark Date: Tue, 5 May 2026 21:53:16 -0400 Subject: [PATCH 4/5] Noah feedback --- test/test_compression_support.py | 27 +-------------------------- 1 file changed, 1 insertion(+), 26 deletions(-) diff --git a/test/test_compression_support.py b/test/test_compression_support.py index b1cf573983..8ad65cd505 100644 --- a/test/test_compression_support.py +++ b/test/test_compression_support.py @@ -17,6 +17,7 @@ from __future__ import annotations import sys +import zlib from unittest.mock import patch sys.path[0:0] = [""] @@ -51,19 +52,12 @@ def test_returns_false_on_import_error(self): class TestHaveZlib(unittest.TestCase): - def test_returns_true_when_available(self): - self.assertTrue(_have_zlib()) - def test_returns_false_on_import_error(self): with patch.dict(sys.modules, {"zlib": None}): self.assertFalse(_have_zlib()) class TestHaveZstd(unittest.TestCase): - def test_returns_bool(self): - result = _have_zstd() - self.assertIsInstance(result, bool) - def test_returns_false_when_unavailable_pre_314(self): if sys.version_info >= (3, 14): self.skipTest("Python 3.14+ uses compression.zstd") @@ -130,10 +124,6 @@ def test_zstd_unavailable_warns_and_removes_314_plus(self): self.assertEqual(result, []) self.assertIn("compression.zstd", str(ctx.warning)) - def test_valid_zlib_always_included(self): - result = validate_compressors(None, ["zlib"]) - self.assertEqual(result, ["zlib"]) - def test_multiple_valid_compressors_preserves_order(self): with patch("pymongo.compression_support._have_snappy", return_value=True): result = validate_compressors(None, ["zlib", "snappy"]) @@ -148,15 +138,9 @@ class TestValidateZlibCompressionLevel(unittest.TestCase): def test_valid_minimum(self): self.assertEqual(validate_zlib_compression_level("level", -1), -1) - def test_valid_zero(self): - self.assertEqual(validate_zlib_compression_level("level", 0), 0) - def test_valid_maximum(self): self.assertEqual(validate_zlib_compression_level("level", 9), 9) - def test_valid_midrange(self): - self.assertEqual(validate_zlib_compression_level("level", 5), 5) - def test_non_integer_raises_type_error(self): with self.assertRaises(TypeError) as ctx: validate_zlib_compression_level("level", "abc") @@ -217,8 +201,6 @@ def test_get_context_unknown_returns_none(self): class TestZlibContext(unittest.TestCase): def test_compress_and_decompress_roundtrip(self): - import zlib - ctx = ZlibContext(level=-1) data = b"hello world" * 100 compressed = ctx.compress(data) @@ -228,9 +210,6 @@ def test_compress_level_stored(self): ctx = ZlibContext(level=6) self.assertEqual(ctx.level, 6) - def test_compressor_id(self): - self.assertEqual(ZlibContext.compressor_id, 2) - class TestDecompress(unittest.TestCase): def test_unknown_compressor_id_raises(self): @@ -239,16 +218,12 @@ def test_unknown_compressor_id_raises(self): self.assertIn("Unknown compressorId 99", str(ctx.exception)) def test_zlib_roundtrip(self): - import zlib - data = b"hello world" compressed = zlib.compress(data) result = decompress(compressed, ZlibContext.compressor_id) self.assertEqual(result, data) def test_zlib_with_memoryview(self): - import zlib - data = b"test data" compressed = zlib.compress(data) result = decompress(memoryview(compressed), ZlibContext.compressor_id) From a0af3a61d7811ab98024b705a18a98a1cf350eaa Mon Sep 17 00:00:00 2001 From: Jeffrey 'Alex' Clark Date: Wed, 6 May 2026 18:42:27 -0400 Subject: [PATCH 5/5] Copilot feedback --- test/test_compression_support.py | 98 +++++++++++++++++--------------- 1 file changed, 51 insertions(+), 47 deletions(-) diff --git a/test/test_compression_support.py b/test/test_compression_support.py index 8ad65cd505..aa767de245 100644 --- a/test/test_compression_support.py +++ b/test/test_compression_support.py @@ -17,7 +17,6 @@ from __future__ import annotations import sys -import zlib from unittest.mock import patch sys.path[0:0] = [""] @@ -38,52 +37,22 @@ ) -class TestHaveSnappy(unittest.TestCase): - def test_returns_true_when_available(self): - try: - import snappy - except ImportError: - self.skipTest("python-snappy not installed") - self.assertTrue(_have_snappy()) - - def test_returns_false_on_import_error(self): - with patch.dict(sys.modules, {"snappy": None}): - self.assertFalse(_have_snappy()) - - -class TestHaveZlib(unittest.TestCase): - def test_returns_false_on_import_error(self): - with patch.dict(sys.modules, {"zlib": None}): - self.assertFalse(_have_zlib()) - - -class TestHaveZstd(unittest.TestCase): - def test_returns_false_when_unavailable_pre_314(self): - if sys.version_info >= (3, 14): - self.skipTest("Python 3.14+ uses compression.zstd") - with patch.dict(sys.modules, {"backports": None, "backports.zstd": None}): - self.assertFalse(_have_zstd()) - - def test_returns_false_when_unavailable_314_plus(self): - if sys.version_info < (3, 14): - self.skipTest("Only applies to Python 3.14+") - with patch.dict(sys.modules, {"compression": None, "compression.zstd": None}): - self.assertFalse(_have_zstd()) - - class TestValidateCompressors(unittest.TestCase): def test_string_input_single(self): - result = validate_compressors(None, "zlib") + with patch("pymongo.compression_support._have_zlib", return_value=True): + result = validate_compressors(None, "zlib") self.assertEqual(result, ["zlib"]) def test_string_input_comma_separated(self): - with patch("pymongo.compression_support._have_snappy", return_value=True): + with patch("pymongo.compression_support._have_zlib", return_value=True), patch( + "pymongo.compression_support._have_snappy", return_value=True + ): result = validate_compressors(None, "zlib,snappy") - self.assertIn("zlib", result) - self.assertIn("snappy", result) + self.assertEqual(result, ["zlib", "snappy"]) def test_iterable_input(self): - result = validate_compressors(None, ["zlib"]) + with patch("pymongo.compression_support._have_zlib", return_value=True): + result = validate_compressors(None, ["zlib"]) self.assertEqual(result, ["zlib"]) def test_unsupported_compressor_warns_and_removes(self): @@ -125,7 +94,9 @@ def test_zstd_unavailable_warns_and_removes_314_plus(self): self.assertIn("compression.zstd", str(ctx.warning)) def test_multiple_valid_compressors_preserves_order(self): - with patch("pymongo.compression_support._have_snappy", return_value=True): + with patch("pymongo.compression_support._have_zlib", return_value=True), patch( + "pymongo.compression_support._have_snappy", return_value=True + ): result = validate_compressors(None, ["zlib", "snappy"]) self.assertEqual(result, ["zlib", "snappy"]) @@ -178,10 +149,9 @@ def test_get_context_snappy(self): self.assertIsInstance(ctx, SnappyContext) def test_get_context_zlib(self): - settings = self._make(level=6) + settings = self._make() ctx = settings.get_compression_context(["zlib"]) self.assertIsInstance(ctx, ZlibContext) - self.assertEqual(ctx.level, 6) def test_get_context_zstd(self): settings = self._make() @@ -189,7 +159,7 @@ def test_get_context_zstd(self): self.assertIsInstance(ctx, ZstdContext) def test_get_context_uses_first_compressor(self): - settings = self._make(level=1) + settings = self._make() ctx = settings.get_compression_context(["zlib", "snappy"]) self.assertIsInstance(ctx, ZlibContext) @@ -200,16 +170,18 @@ def test_get_context_unknown_returns_none(self): class TestZlibContext(unittest.TestCase): + def setUp(self): + if not _have_zlib(): + self.skipTest("zlib not available") + def test_compress_and_decompress_roundtrip(self): + import zlib + ctx = ZlibContext(level=-1) data = b"hello world" * 100 compressed = ctx.compress(data) self.assertEqual(zlib.decompress(compressed), data) - def test_compress_level_stored(self): - ctx = ZlibContext(level=6) - self.assertEqual(ctx.level, 6) - class TestDecompress(unittest.TestCase): def test_unknown_compressor_id_raises(self): @@ -218,17 +190,49 @@ def test_unknown_compressor_id_raises(self): self.assertIn("Unknown compressorId 99", str(ctx.exception)) def test_zlib_roundtrip(self): + if not _have_zlib(): + self.skipTest("zlib not available") + import zlib + data = b"hello world" compressed = zlib.compress(data) result = decompress(compressed, ZlibContext.compressor_id) self.assertEqual(result, data) def test_zlib_with_memoryview(self): + if not _have_zlib(): + self.skipTest("zlib not available") + import zlib + data = b"test data" compressed = zlib.compress(data) result = decompress(memoryview(compressed), ZlibContext.compressor_id) self.assertEqual(result, data) + def test_snappy_roundtrip(self): + if not _have_snappy(): + self.skipTest("python-snappy not installed") + data = b"hello world" * 50 + compressed = SnappyContext.compress(data) + result = decompress(compressed, SnappyContext.compressor_id) + self.assertEqual(result, data) + + def test_snappy_with_memoryview(self): + if not _have_snappy(): + self.skipTest("python-snappy not installed") + data = b"hello world" * 50 + compressed = SnappyContext.compress(data) + result = decompress(memoryview(compressed), SnappyContext.compressor_id) + self.assertEqual(result, data) + + def test_zstd_roundtrip(self): + if not _have_zstd(): + self.skipTest("zstd not available") + data = b"hello world" * 50 + compressed = ZstdContext.compress(data) + result = decompress(compressed, ZstdContext.compressor_id) + self.assertEqual(result, data) + if __name__ == "__main__": unittest.main()