Implement functional tests for flexible checksums support for high-le…

…vel S3 commands.
aws · Sep 20, 2024 · bba1392 · bba1392
1 parent 762e37d
commit bba1392
Show file tree

Hide file tree

Showing 3 changed files with 144 additions and 20 deletions.
diff --git a/tests/functional/s3/test_cp_command.py b/tests/functional/s3/test_cp_command.py
@@ -761,20 +761,70 @@ def test_cp_with_error_and_warning_permissions(self):
         self.assertIn('warning: File has an invalid timestamp.', stderr)
 
     def test_upload_with_flexible_checksum_crc32(self):
-        # TODO Ahmed
-        pass
+        full_path = self.files.create_file('foo.txt', 'contents')
+        cmdline = (
+                '%s %s s3://bucket/key.txt --checksum-algorithm CRC32' % (
+            self.prefix, full_path))
+        self.run_cmd(cmdline, expected_rc=0)
+        self.assert_in_operations_called(
+            ('PutObject', {
+                'Bucket': 'bucket',
+                'Key': 'key.txt',
+                'ChecksumAlgorithm': 'CRC32',
+                'Body': mock.ANY,
+                'ContentType': 'text/plain'
+            })
+        )
 
     def test_upload_with_flexible_checksum_crc32c(self):
-        # TODO Ahmed
-        pass
+        full_path = self.files.create_file('foo.txt', 'contents')
+        cmdline = (
+                '%s %s s3://bucket/key.txt --checksum-algorithm CRC32C' % (
+            self.prefix, full_path))
+        self.run_cmd(cmdline, expected_rc=0)
+        self.assert_in_operations_called(
+            ('PutObject', {
+                'Bucket': 'bucket',
+                'Key': 'key.txt',
+                'ChecksumAlgorithm': 'CRC32C',
+                'Body': mock.ANY,
+                'ContentType': 'text/plain'
+            })
+        )
 
     def test_download_with_flexible_checksum_crc32(self):
-        # TODO Ahmed
-        pass
+        self.parsed_responses = [
+            self.head_object_response(),
+            {
+                'ContentLength': '100',
+                'LastModified': '00:00:00Z',
+                'ETag': 'foo-1',
+                'ChecksumCRC32': 'Tq0H4g==',
+                'Body': BytesIO(b'foo')
+            }
+        ]
+        cmdline = '%s s3://bucket/foo %s --checksum-mode ENABLED' \
+                  % (self.prefix, self.files.rootdir)
+        self.run_cmd(cmdline, expected_rc=0)
+        self.assertEqual(self.operations_called[1][0].name, 'GetObject')
+        self.assertEqual(self.operations_called[1][1]['ChecksumMode'], 'ENABLED')
 
     def test_download_with_flexible_checksum_crc32c(self):
-        # TODO Ahmed
-        pass
+        self.parsed_responses = [
+            self.head_object_response(),
+            {
+                'ContentLength': '100',
+                'LastModified': '00:00:00Z',
+                'ETag': 'foo-1',
+                'ChecksumCRC32C': 'checksum',
+                'Body': BytesIO(b'foo')
+            }
+        ]
+        cmdline = '%s s3://bucket/foo %s --checksum-mode ENABLED' \
+                  % (self.prefix, self.files.rootdir)
+        self.run_cmd(cmdline, expected_rc=0)
+        self.assertEqual(self.operations_called[1][0].name, 'GetObject')
+        self.assertEqual(self.operations_called[1][1]['ChecksumMode'], 'ENABLED')
 
 
 class TestStreamingCPCommand(BaseAWSCommandParamsTest):

diff --git a/tests/functional/s3/test_mv_command.py b/tests/functional/s3/test_mv_command.py
@@ -245,12 +245,38 @@ def test_mv_does_not_delete_source_on_failed_put_tagging(self):
         )
 
     def test_upload_with_flexible_checksum_crc32(self):
-        # TODO Ahmed
-        pass
+        full_path = self.files.create_file('foo.txt', 'contents')
+        cmdline = (
+                '%s %s s3://bucket/key.txt --checksum-algorithm CRC32' % (
+            self.prefix, full_path))
+        self.run_cmd(cmdline, expected_rc=0)
+        self.assert_in_operations_called(
+            ('PutObject', {
+                'Bucket': 'bucket',
+                'Key': 'key.txt',
+                'ChecksumAlgorithm': 'CRC32',
+                'Body': mock.ANY,
+                'ContentType': 'text/plain'
+            })
+        )
 
     def test_download_with_flexible_checksum_crc32(self):
-        # TODO Ahmed
-        pass
+        self.parsed_responses = [
+            self.head_object_response(),
+            {
+                'ContentLength': '100',
+                'LastModified': '00:00:00Z',
+                'ETag': 'foo-1',
+                'ChecksumCRC32': 'checksum',
+                'Body': BytesIO(b'foo')
+            },
+            self.delete_object_response()
+        ]
+        cmdline = '%s s3://bucket/foo %s --checksum-mode ENABLED' \
+                  % (self.prefix, self.files.rootdir)
+        self.run_cmd(cmdline, expected_rc=0)
+        self.assertEqual(self.operations_called[1][0].name, 'GetObject')
+        self.assertEqual(self.operations_called[1][1]['ChecksumMode'], 'ENABLED')
 
 
 class TestMvWithCRTClient(BaseCRTTransferClientTest):

diff --git a/tests/functional/s3/test_sync_command.py b/tests/functional/s3/test_sync_command.py
@@ -354,20 +354,68 @@ def test_with_copy_props(self):
         )
 
     def test_upload_with_flexible_checksum_sha1(self):
-        # TODO Ahmed
-        pass
+        self.files.create_file('foo.txt', 'contents')
+        cmdline = (
+                '%s %s s3://bucket/ --checksum-algorithm SHA1' % (
+            self.prefix, self.files.rootdir))
+        self.run_cmd(cmdline, expected_rc=0)
+        self.assert_in_operations_called(
+            ('PutObject', {
+                'Bucket': 'bucket',
+                'Key': 'foo.txt',
+                'ChecksumAlgorithm': 'SHA1',
+                'Body': mock.ANY,
+                'ContentType': 'text/plain'
+            })
+        )
 
     def test_upload_with_flexible_checksum_sha256(self):
-        # TODO Ahmed
-        pass
+        self.files.create_file('foo.txt', 'contents')
+        cmdline = (
+                '%s %s s3://bucket/ --checksum-algorithm SHA256' % (
+            self.prefix, self.files.rootdir))
+        self.run_cmd(cmdline, expected_rc=0)
+        self.assert_in_operations_called(
+            ('PutObject', {
+                'Bucket': 'bucket',
+                'Key': 'foo.txt',
+                'ChecksumAlgorithm': 'SHA256',
+                'Body': mock.ANY,
+                'ContentType': 'text/plain'
+            })
+        )
 
     def test_download_with_flexible_checksum_sha1(self):
-        # TODO Ahmed
-        pass
+        self.parsed_responses = [
+            self.head_object_response(),
+            {
+                'ContentLength': '100',
+                'LastModified': '00:00:00Z',
+                'ETag': 'foo-1',
+                'ChecksumSHA1': 'checksum',
+                'Body': BytesIO(b'foo')
+            }
+        ]
+        cmdline = '%s s3://bucket/foo %s --checksum-mode ENABLED' \
+                  % (self.prefix, self.files.rootdir)
+        self.run_cmd(cmdline, expected_rc=0)
+        self.assertEqual(self.operations_called[0][0].name, 'ListObjectsV2')
 
     def test_download_with_flexible_checksum_sha256(self):
-        # TODO Ahmed
-        pass
+        self.parsed_responses = [
+            self.head_object_response(),
+            {
+                'ContentLength': '100',
+                'LastModified': '00:00:00Z',
+                'ETag': 'foo-1',
+                'ChecksumSHA256': 'checksum',
+                'Body': BytesIO(b'foo')
+            }
+        ]
+        cmdline = '%s s3://bucket/foo %s --checksum-mode ENABLED' \
+                  % (self.prefix, self.files.rootdir)
+        self.run_cmd(cmdline, expected_rc=0)
+        self.assertEqual(self.operations_called[0][0].name, 'ListObjectsV2')
 
 
 class TestSyncSourceRegion(BaseS3CLIRunnerTest):