adopt upstream unit tests for httpchecksum.apply_request_checksum, .handle_checksum_body, and .StreamingChecksumBody

Author: jakob-keller

tests/boto_tests/unit/test_httpchecksum.py

@@ -14,17 +14,15 @@
 from unittest import mock
 
 import pytest
-from botocore.httpchecksum import (
-    AwsChunkedWrapperError,
-    Crc32Checksum,
-    FlexibleChecksumError,
-)
+from botocore.exceptions import AwsChunkedWrapperError, FlexibleChecksumError
+from botocore.httpchecksum import Crc32Checksum
 from botocore.model import OperationModel, StringShape, StructureShape
 
 from aiobotocore.awsrequest import AioAWSResponse
 from aiobotocore.config import AioConfig
 from aiobotocore.httpchecksum import (
     AioAwsChunkedWrapper,
+    StreamingChecksumBody,
     apply_request_checksum,
     handle_checksum_body,
 )
@@ -523,3 +521,66 @@ def test_wrapper_can_only_seek_to_start(self):
             wrapper.seek(0, whence=1)
         with pytest.raises(AwsChunkedWrapperError):
             wrapper.seek(1, whence=2)
+
+
+class TestStreamingChecksumBody:
+    @pytest.fixture(scope="session")
+    def raw_bytes(self):
+        return b"hello world"
+
+    @pytest.fixture
+    def fake_body(self, raw_bytes):
+        return AsyncBytesIO(raw_bytes)
+
+    @pytest.fixture
+    def make_wrapper(self, fake_body):
+        def make_wrapper(checksum):
+            return StreamingChecksumBody(
+                fake_body,
+                None,
+                Crc32Checksum(),
+                checksum,
+            )
+
+        return make_wrapper
+
+    @pytest.fixture
+    def wrapper(self, make_wrapper):
+        return make_wrapper("DUoRhQ==")
+
+    async def test_basic_read_good(self, raw_bytes, wrapper):
+        actual = await wrapper.read()
+        assert actual == raw_bytes
+
+    async def test_many_reads_good(self, raw_bytes, wrapper):
+        actual = b""
+        actual += await wrapper.read(5)
+        actual += await wrapper.read(5)
+        actual += await wrapper.read(1)
+        assert actual == raw_bytes
+
+    async def test_basic_read_bad(self, make_wrapper):
+        wrapper = make_wrapper("duorhq==")
+        with pytest.raises(FlexibleChecksumError):
+            await wrapper.read()
+
+    async def test_many_reads_bad(self, make_wrapper):
+        wrapper = make_wrapper("duorhq==")
+        await wrapper.read(5)
+        await wrapper.read(6)
+        # Whole body has been read, next read signals the end of the stream and
+        # validates the checksum of the body contents read
+        with pytest.raises(FlexibleChecksumError):
+            await wrapper.read(1)
+
+    async def test_handles_variable_padding(self, raw_bytes, make_wrapper):
+        # This digest is equivalent but with more padding
+        wrapper = make_wrapper("DUoRhQ=====")
+        actual = await wrapper.read()
+        assert actual == raw_bytes
+
+    async def test_iter_raises_error(self, make_wrapper):
+        wrapper = make_wrapper("duorhq==")
+        with pytest.raises(FlexibleChecksumError):
+            async for chunk in wrapper:
+                pass