From 9f6a7bcde31e3901e9c8dc9c0d2f71624fec28d3 Mon Sep 17 00:00:00 2001 From: Nate Prewitt Date: Thu, 27 Jul 2023 12:54:10 -0600 Subject: [PATCH] Refactor unit tests --- botocore/compress.py | 9 +- tests/__init__.py | 24 +++ tests/unit/test_compress.py | 338 +++++++++++++++--------------------- 3 files changed, 167 insertions(+), 204 deletions(-) diff --git a/botocore/compress.py b/botocore/compress.py index 13c74636a2..1f8577e84b 100644 --- a/botocore/compress.py +++ b/botocore/compress.py @@ -16,9 +16,10 @@ """ -import gzip import io import logging +from gzip import GzipFile +from gzip import compress as gzip_compress from botocore.compat import urlencode from botocore.utils import determine_content_length @@ -88,9 +89,9 @@ def _get_body_size(body): def _gzip_compress_body(body): if isinstance(body, str): - return gzip.compress(body.encode('utf-8')) + return gzip_compress(body.encode('utf-8')) elif isinstance(body, (bytes, bytearray)): - return gzip.compress(body) + return gzip_compress(body) elif hasattr(body, 'read'): if hasattr(body, 'seek') and hasattr(body, 'tell'): current_position = body.tell() @@ -102,7 +103,7 @@ def _gzip_compress_body(body): def _gzip_compress_fileobj(body): compressed_obj = io.BytesIO() - with gzip.GzipFile(fileobj=compressed_obj, mode='wb') as gz: + with GzipFile(fileobj=compressed_obj, mode='wb') as gz: while True: chunk = body.read(8192) if not chunk: diff --git a/tests/__init__.py b/tests/__init__.py index 33307c0081..7390d7f7f1 100644 --- a/tests/__init__.py +++ b/tests/__init__.py @@ -587,6 +587,30 @@ def __exit__(self, *args, **kwargs): self.datetime_patcher.stop() +class PatchObject(ContextDecorator): + """ + Context manager for patching out objects in specific classes for tests. + + :type module: module + :param module: reference to imported module to patch + + :type obj: str + :param obj: object within the module to replace. + + :type new: Any + :param new: Any functional replacement for object to be patched. + """ + + def __init__(self, module, obj, new): + self.obj_patcher = mock.patch.object(module, obj, new) + + def __enter__(self, *args, **kwargs): + self.obj_patcher.start() + + def __exit__(self, *args, **kwargs): + self.obj_patcher.stop() + + def patch_load_service_model( session, monkeypatch, service_model_json, ruleset_json ): diff --git a/tests/unit/test_compress.py b/tests/unit/test_compress.py index a7044ff7df..4c46a43ce4 100644 --- a/tests/unit/test_compress.py +++ b/tests/unit/test_compress.py @@ -12,14 +12,14 @@ # language governing permissions and limitations under the License. import gzip import io -from copy import deepcopy +import sys import pytest -from botocore.compat import urlencode +import botocore from botocore.compress import COMPRESSION_MAPPING, maybe_compress_request from botocore.config import Config -from tests import mock +from tests import PatchObject, mock def _make_op( @@ -51,19 +51,29 @@ def _make_op( True, {'requiresLength': True}, ) + + REQUEST_BODY = ( b'Action=PutMetricData&Version=2010-08-01&Namespace=Namespace' b'&MetricData.member.1.MetricName=metric&MetricData.member.1.Unit=Bytes' b'&MetricData.member.1.Value=128' ) - -REQUEST_BODY_STRING = REQUEST_BODY.decode('utf-8') +REQUEST_BODY_COMPRESSED = ( + b'\x1f\x8b\x08\x00\x00\x00\x00\x00\x02\x13sL.\xc9\xcc\xcf\xb3\r(-\xf1M-)' + b'\xcaLvI,IT\x0bK-*\x06\x89\x1a\x19\x18\x1a\xe8\x1aX\xe8\x1a\x18\xaa\xf9%' + b'\xe6\xa6\x16\x17$&\xa7\xda\xc2Yj\x08\x1dz\xb9\xa9\xb9I\xa9Ez\x86z\x101\x90' + b'\x1a\xdb\\0\x13\xab\xaa\xd0\xbc\xcc\x12[\xa7\xca\x92\xd4b\xac\xd2a\x899\xa5' + b'\xa9\xb6\x86F\x16\x00\x1e\xdd\t\xfd\x9e\x00\x00\x00' +) +REQUEST_BODY_GZIPFILE = ( + b'\x1f\x8b\x08\x00\x00\x00\x00\x00\x02\xffsL.\xc9\xcc\xcf\xb3\r(-\xf1M-)' + b'\xcaLvI,IT\x0bK-*\x06\x89\x1a\x19\x18\x1a\xe8\x1aX\xe8\x1a\x18\xaa\xf9%' + b'\xe6\xa6\x16\x17$&\xa7\xda\xc2Yj\x08\x1dz\xb9\xa9\xb9I\xa9Ez\x86z\x101\x90' + b'\x1a\xdb\\0\x13\xab\xaa\xd0\xbc\xcc\x12[\xa7\xca\x92\xd4b\xac\xd2a\x899\xa5' + b'\xa9\xb6\x86F\x16\x00\x1e\xdd\t\xfd\x9e\x00\x00\x00' +) -COMPRESSION_CONFIG_10240_BYTES = Config( - disable_request_compression=False, - request_min_compression_size_bytes=10240, -) COMPRESSION_CONFIG_128_BYTES = Config( disable_request_compression=False, request_min_compression_size_bytes=128, @@ -82,7 +92,7 @@ def read(self, size=None): return self._buffer.read(size) -def _request_dict(body=b'', headers=None): +def _request_dict(body=REQUEST_BODY, headers=None): if headers is None: headers = {} @@ -92,45 +102,24 @@ def _request_dict(body=b'', headers=None): } -def default_request_dict(): - return _request_dict(REQUEST_BODY) - - -def request_dict_string(): - return _request_dict(REQUEST_BODY_STRING) - - -def request_dict_bytearray(): - return _request_dict(bytearray(REQUEST_BODY)) - - -def request_dict_with_content_encoding_header(): - return _request_dict( - REQUEST_BODY, {'foo': b'bar', 'Content-Encoding': 'identity'} - ) - - -def request_dict_string_io(): - return _request_dict(io.StringIO(REQUEST_BODY_STRING)) - - -def request_dict_bytes_io(): - return _request_dict(io.BytesIO(REQUEST_BODY)) - - def request_dict_non_seekable_text_stream(): - return _request_dict(NonSeekableStream(io.StringIO(REQUEST_BODY_STRING))) + stream = NonSeekableStream(io.StringIO(REQUEST_BODY.decode('utf-8'))) + return _request_dict(stream) def request_dict_non_seekable_bytes_stream(): return _request_dict(NonSeekableStream(io.BytesIO(REQUEST_BODY))) -def request_dict_dict(): - return _request_dict({'foo': 'bar'}) +class StaticGzipFile(gzip.GzipFile): + def __init__(self, *args, **kwargs): + kwargs['mtime'] = 0 + super().__init__(*args, **kwargs) -DECOMPRESSION_METHOD_MAP = {'gzip': gzip.decompress} +def static_compress(*args, **kwargs): + kwargs['mtime'] = 0 + return gzip.compress(*args, **kwargs) def _bad_compression(body): @@ -141,122 +130,88 @@ def _bad_compression(body): MOCK_COMPRESSION.update(COMPRESSION_MAPPING) -def _assert_compression_body(original_body, compressed_body, encoding): - if hasattr(original_body, 'read'): - original_body = original_body.read() - compressed_body = compressed_body.read() - if isinstance(original_body, dict): - original_body = urlencode(original_body, doseq=True, encoding='utf-8') - if isinstance(original_body, str): - original_body = original_body.encode('utf-8') - decompress = DECOMPRESSION_METHOD_MAP[encoding] - assert original_body == decompress(compressed_body) +def _assert_compression_body(compressed_body, expected_body): + data = compressed_body + if hasattr(compressed_body, 'read'): + data = compressed_body.read() + assert data == expected_body -def _assert_compression_header(headers, encoding): - assert ( - 'Content-Encoding' in headers - and encoding in headers['Content-Encoding'] - ) +def _assert_compression_header(headers, encoding='gzip'): + assert 'Content-Encoding' in headers + assert encoding in headers['Content-Encoding'] -def assert_compression(original_body, request_dict, encoding): - compressed_body = request_dict['body'] - headers = request_dict['headers'] - _assert_compression_body(original_body, compressed_body, encoding) - _assert_compression_header(headers, encoding) +def assert_request_compressed(request_dict, expected_body): + _assert_compression_body(request_dict['body'], expected_body) + _assert_compression_header(request_dict['headers']) @pytest.mark.parametrize( - 'config, request_dict, operation_model, encoding', + 'request_dict, operation_model, expected_body', [ ( - COMPRESSION_CONFIG_128_BYTES, - default_request_dict(), + _request_dict(), OP_WITH_COMPRESSION, - 'gzip', + REQUEST_BODY_COMPRESSED, ), ( - COMPRESSION_CONFIG_128_BYTES, - default_request_dict(), + _request_dict(), OP_MULTIPLE_COMPRESSIONS, - 'gzip', + REQUEST_BODY_COMPRESSED, ), ( - COMPRESSION_CONFIG_10240_BYTES, - default_request_dict(), + _request_dict(), STREAMING_OP_WITH_COMPRESSION, - 'gzip', + REQUEST_BODY_COMPRESSED, ), ( - COMPRESSION_CONFIG_128_BYTES, - request_dict_bytearray(), - OP_WITH_COMPRESSION, - 'gzip', - ), - ( - COMPRESSION_CONFIG_128_BYTES, - request_dict_with_content_encoding_header(), + _request_dict(bytearray(REQUEST_BODY)), OP_WITH_COMPRESSION, - 'gzip', + REQUEST_BODY_COMPRESSED, ), ( - COMPRESSION_CONFIG_128_BYTES, - request_dict_string(), + _request_dict(headers={'Content-Encoding': 'identity'}), OP_WITH_COMPRESSION, - 'gzip', + REQUEST_BODY_COMPRESSED, ), ( - COMPRESSION_CONFIG_128_BYTES, - request_dict_bytes_io(), + _request_dict(REQUEST_BODY.decode('utf-8')), OP_WITH_COMPRESSION, - 'gzip', + REQUEST_BODY_COMPRESSED, ), ( - COMPRESSION_CONFIG_128_BYTES, - request_dict_string_io(), + _request_dict(io.BytesIO(REQUEST_BODY)), OP_WITH_COMPRESSION, - 'gzip', + REQUEST_BODY_GZIPFILE, ), ( - COMPRESSION_CONFIG_1_BYTE, - request_dict_dict(), + _request_dict(io.StringIO(REQUEST_BODY.decode('utf-8'))), OP_WITH_COMPRESSION, - 'gzip', + REQUEST_BODY_GZIPFILE, ), - ], -) -def test_compression(config, request_dict, operation_model, encoding): - original_body = request_dict['body'] - maybe_compress_request(config, request_dict, operation_model) - assert_compression(original_body, request_dict, encoding) - - -@pytest.mark.parametrize( - 'config, request_dict, operation_model, encoding', - [ ( - COMPRESSION_CONFIG_10240_BYTES, request_dict_non_seekable_bytes_stream(), STREAMING_OP_WITH_COMPRESSION, - 'gzip', + REQUEST_BODY_GZIPFILE, ), ( - COMPRESSION_CONFIG_10240_BYTES, request_dict_non_seekable_text_stream(), STREAMING_OP_WITH_COMPRESSION, - 'gzip', + REQUEST_BODY_GZIPFILE, ), ], ) -def test_compression_non_seekable_streams( - config, request_dict, operation_model, encoding -): - # since the body can't be reset, we must make a copy - # of the original body to test against - original_body = deepcopy(request_dict['body']) - maybe_compress_request(config, request_dict, operation_model) - assert_compression(original_body, request_dict, encoding) +@PatchObject(botocore.compress, 'GzipFile', StaticGzipFile) +@PatchObject(botocore.compress, 'gzip_compress', static_compress) +@pytest.mark.skipif( + sys.version_info < (3, 8), reason="requires python3.8 or higher" +) +def test_compression(request_dict, operation_model, expected_body): + maybe_compress_request( + COMPRESSION_CONFIG_128_BYTES, request_dict, operation_model + ) + assert_request_compressed(request_dict, expected_body) @pytest.mark.parametrize( @@ -265,9 +220,9 @@ def test_compression_non_seekable_streams( ( Config( disable_request_compression=True, - request_min_compression_size_bytes=1000, + request_min_compression_size_bytes=1, ), - default_request_dict(), + _request_dict(), OP_WITH_COMPRESSION, ), ( @@ -275,128 +230,111 @@ def test_compression_non_seekable_streams( disable_request_compression=False, request_min_compression_size_bytes=256, ), - default_request_dict(), + _request_dict(), + OP_WITH_COMPRESSION, + ), + ( + Config( + disable_request_compression=False, + request_min_compression_size_bytes=1, + signature_version='v2', + ), + _request_dict(), OP_WITH_COMPRESSION, ), ( - COMPRESSION_CONFIG_10240_BYTES, - default_request_dict(), + COMPRESSION_CONFIG_128_BYTES, + _request_dict(), STREAMING_OP_WITH_COMPRESSION_REQUIRES_LENGTH, ), ( - COMPRESSION_CONFIG_10240_BYTES, - default_request_dict(), + COMPRESSION_CONFIG_128_BYTES, + _request_dict(), OP_NO_COMPRESSION, ), ( COMPRESSION_CONFIG_128_BYTES, - default_request_dict(), + _request_dict(), OP_UNKNOWN_COMPRESSION, ), - ( - COMPRESSION_CONFIG_10240_BYTES, - request_dict_string(), - OP_WITH_COMPRESSION, - ), - ( - COMPRESSION_CONFIG_10240_BYTES, - request_dict_bytearray(), - OP_WITH_COMPRESSION, - ), - ( - COMPRESSION_CONFIG_10240_BYTES, - request_dict_bytes_io(), - OP_WITH_COMPRESSION, - ), - ( - COMPRESSION_CONFIG_10240_BYTES, - request_dict_string_io(), - OP_WITH_COMPRESSION, - ), ( COMPRESSION_CONFIG_128_BYTES, - request_dict_with_content_encoding_header(), - OP_UNKNOWN_COMPRESSION, + _request_dict(headers={'Content-Encoding': 'identity'}), + OP_NO_COMPRESSION, ), ( - COMPRESSION_CONFIG_1_BYTE, + COMPRESSION_CONFIG_128_BYTES, request_dict_non_seekable_bytes_stream(), OP_WITH_COMPRESSION, ), - ( - COMPRESSION_CONFIG_1_BYTE, - request_dict_non_seekable_text_stream(), - OP_WITH_COMPRESSION, - ), - ( - COMPRESSION_CONFIG_1_BYTE, - _request_dict(1), - OP_WITH_COMPRESSION, - ), - ( - COMPRESSION_CONFIG_1_BYTE, - _request_dict(1.0), - OP_WITH_COMPRESSION, - ), - ( - COMPRESSION_CONFIG_1_BYTE, - _request_dict(object()), - OP_WITH_COMPRESSION, - ), - ( - COMPRESSION_CONFIG_1_BYTE, - _request_dict(True), - OP_WITH_COMPRESSION, - ), - ( - COMPRESSION_CONFIG_1_BYTE, - _request_dict(None), - OP_WITH_COMPRESSION, - ), - ( - COMPRESSION_CONFIG_1_BYTE, - _request_dict(set()), - OP_WITH_COMPRESSION, - ), ], ) def test_no_compression(config, request_dict, operation_model): ce_header = request_dict['headers'].get('Content-Encoding') original_body = request_dict['body'] maybe_compress_request(config, request_dict, operation_model) - assert request_dict['body'] is original_body - assert ce_header is request_dict['headers'].get('Content-Encoding') + assert request_dict['body'] == original_body + assert ce_header == request_dict['headers'].get('Content-Encoding') -def test_dict_no_compression(): - request_dict = request_dict_dict() - original_body = request_dict['body'] +@pytest.mark.parametrize( + 'operation_model, expected_body', + [ + ( + OP_WITH_COMPRESSION, + ( + b'\x1f\x8b\x08\x00\x00\x00\x00\x00\x02\x13K\xcb' + b'\xcf\xb7MJ,\x02\x00v\x8e5\x1c\x07\x00\x00\x00' + ), + ), + (OP_NO_COMPRESSION, {'foo': 'bar'}), + ], +) +@PatchObject(botocore.compress, 'gzip_compress', static_compress) +@pytest.mark.skipif( + sys.version_info < (3, 8), reason="requires python3.8 or higher" +) +def test_dict_compression(operation_model, expected_body): + request_dict = _request_dict({'foo': 'bar'}) maybe_compress_request( - COMPRESSION_CONFIG_128_BYTES, request_dict, OP_WITH_COMPRESSION + COMPRESSION_CONFIG_1_BYTE, request_dict, operation_model ) body = request_dict['body'] - encoded_body = urlencode(original_body, doseq=True, encoding='utf-8') - assert body == encoded_body.encode('utf-8') + assert body == expected_body -@pytest.mark.parametrize( - 'request_dict', - [request_dict_string_io(), request_dict_bytes_io()], -) -def test_body_streams_position_reset(request_dict): +@pytest.mark.parametrize('body', [1, object(), True, 1.0]) +def test_maybe_compress_bad_types(body): + request_dict = _request_dict(body) + maybe_compress_request( + COMPRESSION_CONFIG_1_BYTE, request_dict, OP_WITH_COMPRESSION + ) + assert request_dict['body'] == body + + +@PatchObject(botocore.compress, 'GzipFile', StaticGzipFile) +def test_body_streams_position_reset(): + request_dict = _request_dict(io.BytesIO(REQUEST_BODY)) maybe_compress_request( COMPRESSION_CONFIG_128_BYTES, request_dict, OP_WITH_COMPRESSION, ) assert request_dict['body'].tell() == 0 + assert 'Content-Encoding' in request_dict['headers'] + assert request_dict['body'].read() == REQUEST_BODY_GZIPFILE +@PatchObject(botocore.compress, 'gzip_compress', static_compress) +@PatchObject(botocore.compress, 'COMPRESSION_MAPPING', MOCK_COMPRESSION) +@pytest.mark.skipif( + sys.version_info < (3, 8), reason="requires python3.8 or higher" +) def test_only_compress_once(): - with mock.patch('botocore.compress.COMPRESSION_MAPPING', MOCK_COMPRESSION): - request_dict = default_request_dict() - body = request_dict['body'] - maybe_compress_request( - COMPRESSION_CONFIG_128_BYTES, request_dict, OP_WITH_COMPRESSION - ) - assert_compression(body, request_dict, 'gzip') + request_dict = _request_dict() + maybe_compress_request( + COMPRESSION_CONFIG_128_BYTES, + request_dict, + OP_WITH_COMPRESSION, + ) + assert_request_compressed(request_dict, REQUEST_BODY_COMPRESSED)