Update tests in respond to feedback.

tests/functional/s3/test_sync_command.py

@@ -367,6 +367,37 @@ def test_upload_with_checksum_algorithm_sha1(self):
  })
  )
 
+ def test_copy_with_checksum_algorithm_update_sha1(self):
+ cmdline = f'{self.prefix} s3://src-bucket/ s3://dest-bucket/ --checksum-algorithm SHA1'
+ self.parsed_responses = [
+ # Response for ListObjects on source bucket
+ self.list_objects_response(['mykey'], override_kwargs={'ChecksumAlgorithm': 'CRC32'}),
+ # Response for ListObjects on destination bucket
+ self.list_objects_response([]),
+ # Response for CopyObject
+ {
+ 'ChecksumSHA1': 'sha1-checksum'
+ }
+ ]
+ self.run_cmd(cmdline, expected_rc=0)
+ self.assert_operations_called(
+ [
+ self.list_objects_request('src-bucket'),
+ self.list_objects_request('dest-bucket'),
+ (
+ 'CopyObject', {
+ 'CopySource': {
+ 'Bucket': 'src-bucket',
+ 'Key': 'mykey'
+ },
+ 'Bucket': 'dest-bucket',
+ 'Key': 'mykey',
+ 'ChecksumAlgorithm': 'SHA1'
+ }
+ )
+ ]
+ )
+
  def test_upload_with_checksum_algorithm_sha256(self):
  self.files.create_file('foo.txt', 'contents')
  cmdline = f'{self.prefix} {self.files.rootdir} s3://bucket/ --checksum-algorithm SHA256'

tests/unit/customizations/s3/test_subcommands.py

@@ -404,8 +404,9 @@ def test_validate_checksum_algorithm_download_error(self):
  paths = ['s3://bucket/key', self.file_creator.rootdir]
  parameters = {'checksum_algorithm': 'CRC32'}
  cmd_params = CommandParameters('cp', parameters, '')
- with self.assertRaises(ParamValidationError):
+ with self.assertRaises(ParamValidationError) as cm:
  cmd_params.add_paths(paths)
+ self.assertIn('Expected checksum-algorithm parameter to be used with one of following path formats', cm.msg)
 
  def test_validate_checksum_algorithm_sync_download_error(self):
  paths = ['s3://bucket/key', self.file_creator.rootdir]
@@ -414,13 +415,6 @@ def test_validate_checksum_algorithm_sync_download_error(self):
  with self.assertRaises(ParamValidationError):
  cmd_params.add_paths(paths)
 
- def test_validate_checksum_algorithm_move_error(self):
- paths = ['s3://bucket/key', 's3://bucket2/key']
- parameters = {'checksum_algorithm': 'SHA1'}
- cmd_params = CommandParameters('mv', parameters, '')
- with self.assertRaises(ParamValidationError):
- cmd_params.add_paths(paths)
-
  def test_validate_checksum_mode_upload_error(self):
  paths = [self.file_creator.rootdir, 's3://bucket/key']
  parameters = {'checksum_mode': 'ENABLED'}

tests/unit/customizations/s3/test_utils.py

@@ -658,22 +658,50 @@ class TestRequestParamsMapperChecksumAlgorithm:
  def cli_params(self):
  return {'checksum_algorithm': 'CRC32'}
 
+ @pytest.fixture
+ def cli_params_no_algorithm(self):
+ return {}
+
  def test_put_object(self, cli_params):
  request_params = {}
  RequestParamsMapper.map_put_object_params(request_params, cli_params)
  assert request_params == {'ChecksumAlgorithm': 'CRC32'}
 
+ def test_put_object_no_checksum(self, cli_params_no_algorithm):
+ request_params = {}
+ RequestParamsMapper.map_put_object_params(request_params, cli_params_no_algorithm)
+ assert 'ChecksumAlgorithm' not in request_params
+
+ def test_copy_object(self, cli_params):
+ request_params = {}
+ RequestParamsMapper.map_copy_object_params(request_params, cli_params)
+ assert request_params == {'ChecksumAlgorithm': 'CRC32'}
+
+ def test_copy_object_no_checksum(self, cli_params_no_algorithm):
+ request_params = {}
+ RequestParamsMapper.map_put_object_params(request_params, cli_params_no_algorithm)
+ assert 'ChecksumAlgorithm' not in request_params
+
 
 class TestRequestParamsMapperChecksumMode:
  @pytest.fixture
  def cli_params(self):
  return {'checksum_mode': 'ENABLED'}
 
+ @pytest.fixture
+ def cli_params_no_checksum(self):
+ return {}
+
  def test_get_object(self, cli_params):
  request_params = {}
  RequestParamsMapper.map_get_object_params(request_params, cli_params)
  assert request_params == {'ChecksumMode': 'ENABLED'}
 
+ def test_get_object_no_checksums(self, cli_params_no_checksum):
+ request_params = {}
+ RequestParamsMapper.map_get_object_params(request_params, cli_params_no_checksum)
+ assert 'ChecksumMode' not in request_params
+
 
 class TestRequestParamsMapperRequestPayer(unittest.TestCase):
  def setUp(self):