PrivacyScanData/S3/NewFind/s3transfer-develop/tests/integration/test_copy.py

# Copyright 2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"). You
# may not use this file except in compliance with the License. A copy of
# the License is located at
#
# http://aws.amazon.com/apache2.0/
#
# or in the "license" file accompanying this file. This file is
# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
# ANY KIND, either express or implied. See the License for the specific
# language governing permissions and limitations under the License.
from s3transfer.manager import TransferConfig
from tests import RecordingSubscriber
from tests.integration import BaseTransferManagerIntegTest


class TestCopy(BaseTransferManagerIntegTest):
    def setUp(self):
        super().setUp()
        self.multipart_threshold = 5 * 1024 * 1024
        self.config = TransferConfig(
            multipart_threshold=self.multipart_threshold
        )

    def test_copy_below_threshold(self):
        transfer_manager = self.create_transfer_manager(self.config)
        key = '1mb.txt'
        new_key = '1mb-copy.txt'

        filename = self.files.create_file_with_size(key, filesize=1024 * 1024)
        self.upload_file(filename, key)

        future = transfer_manager.copy(
            copy_source={'Bucket': self.bucket_name, 'Key': key},
            bucket=self.bucket_name,
            key=new_key,
        )

        future.result()
        self.assertTrue(self.object_exists(new_key))

    def test_copy_above_threshold(self):
        transfer_manager = self.create_transfer_manager(self.config)
        key = '20mb.txt'
        new_key = '20mb-copy.txt'

        filename = self.files.create_file_with_size(
            key, filesize=20 * 1024 * 1024
        )
        self.upload_file(filename, key)

        future = transfer_manager.copy(
            copy_source={'Bucket': self.bucket_name, 'Key': key},
            bucket=self.bucket_name,
            key=new_key,
        )

        future.result()
        self.assertTrue(self.object_exists(new_key))

    def test_progress_subscribers_on_copy(self):
        subscriber = RecordingSubscriber()
        transfer_manager = self.create_transfer_manager(self.config)
        key = '20mb.txt'
        new_key = '20mb-copy.txt'

        filename = self.files.create_file_with_size(
            key, filesize=20 * 1024 * 1024
        )
        self.upload_file(filename, key)

        future = transfer_manager.copy(
            copy_source={'Bucket': self.bucket_name, 'Key': key},
            bucket=self.bucket_name,
            key=new_key,
            subscribers=[subscriber],
        )

        future.result()
        # The callback should have been called enough times such that
        # the total amount of bytes we've seen (via the "amount"
        # arg to the callback function) should be the size
        # of the file we uploaded.
        self.assertEqual(subscriber.calculate_bytes_seen(), 20 * 1024 * 1024)