Repository: apache/airflow
File: providers/amazon/tests/unit/amazon/aws/hooks/test_s3.py (Python)

Code before:
                RequestPayer="requester",
            )

    @mock_aws
    def test_copy_object_with_kms_encryption(self, s3_bucket):
        mock_hook = S3Hook()
        with mock.patch.object(S3Hook, "get_conn") as get_conn:
            mock_hook.copy_object(
                "my_key",
                "my_key_encrypted",
                s3_bucket,
                s3_bucket,
                kms_key_id="arn:aws:kms:us-east-1:123456789012:key/abcd1234",
                kms_encryption_type="aws:kms",
            )
            get_conn.return_value.copy_object.assert_called_once_with(
                Bucket=s3_bucket,
                Key="my_key_encrypted",
                CopySource={"Bucket": s3_bucket, "Key": "my_key", "VersionId": None},
                ACL="private",
                SSEKMSKeyId="arn:aws:kms:us-east-1:123456789012:key/abcd1234",
                ServerSideEncryption="aws:kms",
            )

    @mock_aws
    def test_copy_object_with_kms_one_missing_raises(s3_bucket):
        hook = S3Hook()

        with pytest.raises(ValueError, match="kms_key_id and kms_encryption_type must both be specified"):
            hook.copy_object(
                source_bucket_key="my_key",
                dest_bucket_key="my_key_copy",
                source_bucket_name=s3_bucket,
                dest_bucket_name=s3_bucket,
                kms_key_id="arn:aws:kms:us-east-1:123456789012:key/abcd1234",
            )
        with pytest.raises(ValueError, match="kms_key_id and kms_encryption_type must both be specified"):
            hook.copy_object(
                source_bucket_key="my_key",
                dest_bucket_key="my_key_copy",
                source_bucket_name=s3_bucket,
                dest_bucket_name=s3_bucket,
                kms_encryption_type="aws:kms",
            )

    @mock_aws
    def test_delete_bucket_if_bucket_exist(self, s3_bucket):
        # assert if the bucket is created
        mock_hook = S3Hook()
        mock_hook.create_bucket(bucket_name=s3_bucket)
        assert mock_hook.check_for_bucket(bucket_name=s3_bucket)

Review comment (bug):
```suggestion
    def test_copy_object_with_kms_one_missing_raises(self, s3_bucket):
```

Interesting that the CI didn't fail until now because of that - could you please double-check?

Code after:
                RequestPayer="requester",
            )

    @mock_aws
    def test_copy_object_with_kms_encryption(self, s3_bucket):
        mock_hook = S3Hook()
        with mock.patch.object(S3Hook, "get_conn") as get_conn:
            mock_hook.copy_object(
                "my_key",
                "my_key_encrypted",
                s3_bucket,
                s3_bucket,
                kms_key_id="arn:aws:kms:us-east-1:123456789012:key/abcd1234",
                kms_encryption_type="aws:kms",
            )
            get_conn.return_value.copy_object.assert_called_once_with(
                Bucket=s3_bucket,
                Key="my_key_encrypted",
                CopySource={"Bucket": s3_bucket, "Key": "my_key", "VersionId": None},
                ACL="private",
                SSEKMSKeyId="arn:aws:kms:us-east-1:123456789012:key/abcd1234",
                ServerSideEncryption="aws:kms",
            )

    @mock_aws
    def test_copy_object_with_kms_one_missing_raises(self, s3_bucket):
        hook = S3Hook()

        with pytest.raises(ValueError, match="kms_key_id and kms_encryption_type must both be specified"):
            hook.copy_object(
                source_bucket_key="my_key",
                dest_bucket_key="my_key_copy",
                source_bucket_name=s3_bucket,
                dest_bucket_name=s3_bucket,
                kms_key_id="arn:aws:kms:us-east-1:123456789012:key/abcd1234",
            )
        with pytest.raises(ValueError, match="kms_key_id and kms_encryption_type must both be specified"):
            hook.copy_object(
                source_bucket_key="my_key",
                dest_bucket_key="my_key_copy",
                source_bucket_name=s3_bucket,
                dest_bucket_name=s3_bucket,
                kms_encryption_type="aws:kms",
            )

    @mock_aws
    def test_delete_bucket_if_bucket_exist(self, s3_bucket):
        # assert if the bucket is created
        mock_hook = S3Hook()
        mock_hook.create_bucket(bucket_name=s3_bucket)
        assert mock_hook.check_for_bucket(bucket_name=s3_bucket)