# -*- coding: utf-8 -*-
# Copyright 2013 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.

# The following tests are performed to ensure that the commands work.
# It does not check every possible parameter that can be thrown as
# those are checked by tests in other classes
import os
import random
import platform
import contextlib
import time
import stat
import signal
import string
import socket
import tempfile
import shutil
import copy

import botocore.session
from botocore.exceptions import ClientError
from awscli.compat import six
from nose.plugins.attrib import attr

from awscli.testutils import unittest, get_stdout_encoding
from awscli.testutils import aws as _aws
from awscli.testutils import BaseS3CLICommand
from tests.integration.customizations.s3 import create_bucket as _create_bucket
from awscli.customizations.s3.transferconfig import DEFAULTS
from awscli.customizations.scalarparse import add_scalar_parsers


@contextlib.contextmanager
def cd(directory):
    original = os.getcwd()
    try:
        os.chdir(directory)
        yield
    finally:
        os.chdir(original)


def aws(command, collect_memory=False, env_vars=None, wait_for_finish=True,
        input_data=None, input_file=None):
    if not env_vars:
        env_vars = os.environ.copy()
        env_vars['AWS_DEFAULT_REGION'] = "us-west-2"
    return _aws(command, collect_memory=collect_memory, env_vars=env_vars,
                wait_for_finish=wait_for_finish, input_data=input_data,
                input_file=input_file)


def _running_on_rhel():
    return (
        hasattr(platform, 'linux_distribution') and
        platform.linux_distribution()[0] == 'Red Hat Enterprise Linux Server')


class TestMoveCommand(BaseS3CLICommand):

    def test_mv_local_to_s3(self):
        bucket_name = self.create_bucket()
        full_path = self.files.create_file('foo.txt', 'this is foo.txt')
        p = aws('s3 mv %s s3://%s/foo.txt' % (full_path,
                                              bucket_name))
        self.assert_no_errors(p)
        # When we move an object, the local file is gone:
        self.assertTrue(not os.path.exists(full_path))
        # And now resides in s3.
        self.assert_key_contents_equal(bucket_name, 'foo.txt',
                                       'this is foo.txt')

    def test_mv_s3_to_local(self):
        bucket_name = self.create_bucket()
        self.put_object(bucket_name, 'foo.txt', 'this is foo.txt')
        full_path = self.files.full_path('foo.txt')
        self.assertTrue(self.key_exists(bucket_name, key_name='foo.txt'))
        p = aws('s3 mv s3://%s/foo.txt %s' % (bucket_name, full_path))
        self.assert_no_errors(p)
        self.assertTrue(os.path.exists(full_path))
        with open(full_path, 'r') as f:
            self.assertEqual(f.read(), 'this is foo.txt')
        # The s3 file should not be there anymore.
        self.assertTrue(not self.key_exists(bucket_name, key_name='foo.txt'))

    def test_mv_s3_to_s3(self):
        from_bucket = self.create_bucket()
        to_bucket = self.create_bucket()
        self.put_object(from_bucket, 'foo.txt', 'this is foo.txt')

        p = aws('s3 mv s3://%s/foo.txt s3://%s/foo.txt' % (from_bucket,
                                                           to_bucket))
        self.assert_no_errors(p)
        contents = self.get_key_contents(to_bucket, 'foo.txt')
        self.assertEqual(contents, 'this is foo.txt')
        # And verify that the object no longer exists in the from_bucket.
        self.assertTrue(not self.key_exists(from_bucket, key_name='foo.txt'))

    @attr('slow')
    def test_mv_s3_to_s3_multipart(self):
        from_bucket = self.create_bucket()
        to_bucket = self.create_bucket()
        file_contents = six.BytesIO(b'abcd' * (1024 * 1024 * 10))
        self.put_object(from_bucket, 'foo.txt', file_contents)

        p = aws('s3 mv s3://%s/foo.txt s3://%s/foo.txt' % (from_bucket,
                                                           to_bucket))
        self.assert_no_errors(p)
        self.assert_key_contents_equal(to_bucket, 'foo.txt', file_contents)
        # And verify that the object no longer exists in the from_bucket.
        self.assertTrue(not self.key_exists(from_bucket, key_name='foo.txt'))

    def test_mv_s3_to_s3_multipart_recursive(self):
        from_bucket = self.create_bucket()
        to_bucket = self.create_bucket()

        large_file_contents = six.BytesIO(b'abcd' * (1024 * 1024 * 10))
        small_file_contents = 'small file contents'
        self.put_object(from_bucket, 'largefile', large_file_contents)
        self.put_object(from_bucket, 'smallfile', small_file_contents)

        p = aws('s3 mv s3://%s/ s3://%s/ --recursive' % (from_bucket,
                                                         to_bucket))
        self.assert_no_errors(p)
        # Nothing's in the from_bucket.
        self.assertTrue(not self.key_exists(from_bucket,
                                            key_name='largefile'))
        self.assertTrue(not self.key_exists(from_bucket,
                                            key_name='smallfile'))

        # And both files are in the to_bucket.
        self.assertTrue(self.key_exists(to_bucket, key_name='largefile'))
        self.assertTrue(self.key_exists(to_bucket, key_name='smallfile'))

        # And the contents are what we expect.
        self.assert_key_contents_equal(to_bucket, 'smallfile',
                                       small_file_contents)
        self.assert_key_contents_equal(to_bucket, 'largefile',
                                       large_file_contents)

    @attr('slow')
    def test_mv_with_large_file(self):
        bucket_name = self.create_bucket()
        # 40MB will force a multipart upload.
        file_contents = six.BytesIO(b'abcd' * (1024 * 1024 * 10))
        foo_txt = self.files.create_file(
            'foo.txt', file_contents.getvalue().decode('utf-8'))
        p = aws('s3 mv %s s3://%s/foo.txt' % (foo_txt, bucket_name))
        self.assert_no_errors(p)
        # When we move an object, the local file is gone:
        self.assertTrue(not os.path.exists(foo_txt))
        # And now resides in s3.
        self.assert_key_contents_equal(bucket_name, 'foo.txt', file_contents)

        # Now verify we can download this file.
        p = aws('s3 mv s3://%s/foo.txt %s' % (bucket_name, foo_txt))
        self.assert_no_errors(p)
        self.assertTrue(os.path.exists(foo_txt))
        self.assertEqual(os.path.getsize(foo_txt),
                         len(file_contents.getvalue()))

    def test_mv_to_nonexistent_bucket(self):
        full_path = self.files.create_file('foo.txt', 'this is foo.txt')
        p = aws('s3 mv %s s3://bad-noexist-13143242/foo.txt' % (full_path,))
        self.assertEqual(p.rc, 1)

    def test_cant_move_file_onto_itself_small_file(self):
        # We don't even need a remote file in this case.  We can
        # immediately validate that we can't move a file onto itself.
        bucket_name = self.create_bucket()
        self.put_object(bucket_name, key_name='key.txt', contents='foo')
        p = aws('s3 mv s3://%s/key.txt s3://%s/key.txt' %
                (bucket_name, bucket_name))
        self.assertEqual(p.rc, 255)
        self.assertIn('Cannot mv a file onto itself', p.stderr)

    def test_cant_move_large_file_onto_itself(self):
        # At the API level, you can multipart copy an object onto itself,
        # but a mv command doesn't make sense because a mv is just a
        # cp + an rm of the src file.  We should be consistent and
        # not allow large files to be mv'd onto themselves.
        file_contents = six.BytesIO(b'a' * (1024 * 1024 * 10))
        bucket_name = self.create_bucket()
        self.put_object(bucket_name, key_name='key.txt',
                        contents=file_contents)
        p = aws('s3 mv s3://%s/key.txt s3://%s/key.txt' %
                (bucket_name, bucket_name))
        self.assertEqual(p.rc, 255)
        self.assertIn('Cannot mv a file onto itself', p.stderr)


class TestRm(BaseS3CLICommand):
    @unittest.skipIf(platform.system() not in ['Darwin', 'Linux'],
                     'Newline in filename test not valid on windows.')
    # Windows won't let you do this.  You'll get:
    # [Errno 22] invalid mode ('w') or filename:
    # 'c:\\windows\\temp\\tmp0fv8uu\\foo\r.txt'
    def test_rm_with_newlines(self):
        bucket_name = self.create_bucket()

        # Note the carriage return in the key name.
        foo_txt = self.files.create_file('foo\r.txt', 'this is foo.txt')
        p = aws('s3 cp %s s3://%s/foo\r.txt' % (foo_txt, bucket_name))
        self.assert_no_errors(p)

        # Make sure object is in bucket.
        self.assertTrue(self.key_exists(bucket_name, key_name='foo\r.txt'))

        # Then delete the file.
        p = aws('s3 rm s3://%s/ --recursive' % (bucket_name,))

        # And verify it's gone.
        self.assertFalse(self.key_exists(bucket_name, key_name='foo\r.txt'))

    def test_rm_with_page_size(self):
        bucket_name = self.create_bucket()
        self.put_object(bucket_name, 'foo.txt', contents='hello world')
        self.put_object(bucket_name, 'bar.txt', contents='hello world2')
        p = aws('s3 rm s3://%s/ --recursive --page-size 1' % bucket_name)
        self.assert_no_errors(p)

        self.assertFalse(self.key_exists(bucket_name, key_name='foo.txt'))
        self.assertFalse(self.key_exists(bucket_name, key_name='bar.txt'))


class TestCp(BaseS3CLICommand):

    def test_cp_to_and_from_s3(self):
        # This tests the ability to put a single file in s3
        # move it to a different bucket.
        # and download the file locally
        bucket_name = self.create_bucket()

        # copy file into bucket.
        foo_txt = self.files.create_file('foo.txt', 'this is foo.txt')
        p = aws('s3 cp %s s3://%s/foo.txt' % (foo_txt, bucket_name))
        self.assert_no_errors(p)

        # Make sure object is in bucket.
        self.assertTrue(self.key_exists(bucket_name, key_name='foo.txt'))
        self.assertEqual(
            self.get_key_contents(bucket_name, key_name='foo.txt'),
            'this is foo.txt')

        self.assertEqual(
            self.content_type_for_key(bucket_name, key_name='foo.txt'),
            'text/plain')

        # Make a new name for the file and copy it locally.
        full_path = self.files.full_path('bar.txt')
        p = aws('s3 cp s3://%s/foo.txt %s' % (bucket_name, full_path))
        self.assert_no_errors(p)

        with open(full_path, 'r') as f:
            self.assertEqual(f.read(), 'this is foo.txt')

    def test_cp_without_trailing_slash(self):
        # There's a unit test for this, but we still want to verify this
        # with an integration test.
        bucket_name = self.create_bucket()

        # copy file into bucket.
        foo_txt = self.files.create_file('foo.txt', 'this is foo.txt')
        # Note that the destination has no trailing slash.
        p = aws('s3 cp %s s3://%s' % (foo_txt, bucket_name))
        self.assert_no_errors(p)

        # Make sure object is in bucket.
        self.assertTrue(self.key_exists(bucket_name, key_name='foo.txt'))
        self.assertEqual(
            self.get_key_contents(bucket_name, key_name='foo.txt'),
            'this is foo.txt')

    @attr('slow')
    def test_cp_s3_s3_multipart(self):
        from_bucket = self.create_bucket()
        to_bucket = self.create_bucket()
        file_contents = six.BytesIO(b'abcd' * (1024 * 1024 * 10))
        self.put_object(from_bucket, 'foo.txt', file_contents)

        p = aws('s3 cp s3://%s/foo.txt s3://%s/foo.txt' %
                (from_bucket, to_bucket))
        self.assert_no_errors(p)
        self.assert_key_contents_equal(to_bucket, 'foo.txt', file_contents)
        self.assertTrue(self.key_exists(from_bucket, key_name='foo.txt'))

    def test_guess_mime_type(self):
        bucket_name = self.create_bucket()
        bar_png = self.files.create_file('bar.jpeg', 'fake png image')
        p = aws('s3 cp %s s3://%s/bar.jpeg' % (bar_png, bucket_name))
        self.assert_no_errors(p)

        # We should have correctly guessed the content type based on the
        # filename extension.
        self.assertEqual(
            self.content_type_for_key(bucket_name, key_name='bar.jpeg'),
            'image/jpeg')

    @attr('slow')
    def test_download_large_file(self):
        # This will force a multipart download.
        bucket_name = self.create_bucket()
        foo_contents = six.BytesIO(b'abcd' * (1024 * 1024 * 10))
        self.put_object(bucket_name, key_name='foo.txt',
                        contents=foo_contents)
        local_foo_txt = self.files.full_path('foo.txt')
        p = aws('s3 cp s3://%s/foo.txt %s' % (bucket_name, local_foo_txt))
        self.assert_no_errors(p)
        self.assertEqual(os.path.getsize(local_foo_txt),
                         len(foo_contents.getvalue()))

    @attr('slow')
    @unittest.skipIf(platform.system() not in ['Darwin', 'Linux'],
                     'SIGINT not supported on Windows.')
    def test_download_ctrl_c_does_not_hang(self):
        bucket_name = self.create_bucket()
        foo_contents = six.BytesIO(b'abcd' * (1024 * 1024 * 20))
        self.put_object(bucket_name, key_name='foo.txt',
                        contents=foo_contents)
        local_foo_txt = self.files.full_path('foo.txt')
        process = aws('s3 cp s3://%s/foo.txt %s' %
                      (bucket_name, local_foo_txt), wait_for_finish=False)
        # Give it some time to start up and enter it's main task loop.
        time.sleep(1)
        # The process has 60 seconds to finish after being sent a Ctrl+C,
        # otherwise the test fails.
        process.send_signal(signal.SIGINT)
        deadline = time.time() + 60
        while time.time() < deadline:
            rc = process.poll()
            if rc is not None:
                break
            time.sleep(1)
        else:
            process.kill()
            self.fail("CLI did not exist within 30 seconds of "
                      "receiving a Ctrl+C")
        # A Ctrl+C should have a non-zero RC.
        # We either caught the process in
        # its main polling loop (rc=1), or it was successfully terminated by
        # the SIGINT (rc=-2).
        self.assertIn(process.returncode, [1, -2])

    def test_cp_to_nonexistent_bucket(self):
        foo_txt = self.files.create_file('foo.txt', 'this is foo.txt')
        p = aws('s3 cp %s s3://noexist-bucket-foo-bar123/foo.txt' % (foo_txt,))
        self.assertEqual(p.rc, 1)

    def test_cp_empty_file(self):
        bucket_name = self.create_bucket()
        foo_txt = self.files.create_file('foo.txt', contents='')
        p = aws('s3 cp %s s3://%s/' % (foo_txt, bucket_name))
        self.assertEqual(p.rc, 0)
        self.assertNotIn('failed', p.stderr)
        self.assertTrue(self.key_exists(bucket_name, 'foo.txt'))

    def test_download_non_existent_key(self):
        p = aws('s3 cp s3://jasoidfjasdjfasdofijasdf/foo.txt foo.txt')
        self.assertEqual(p.rc, 1)
        expected_err_msg = (
            'A client error (404) occurred when calling the '
            'HeadObject operation: Key "foo.txt" does not exist')
        self.assertIn(expected_err_msg, p.stderr)

    def test_download_encrypted_kms_object(self):
        bucket_name = self.create_bucket(region='eu-central-1')
        extra_args = {
            'ServerSideEncryption': 'aws:kms',
            'SSEKMSKeyId': 'alias/aws/s3'
        }
        object_name = 'foo.txt'
        contents = 'this is foo.txt'
        self.put_object(bucket_name, object_name, contents,
                        extra_args=extra_args)
        local_filename = self.files.full_path('foo.txt')
        p = aws('s3 cp s3://%s/%s %s --region eu-central-1' %
                (bucket_name, object_name, local_filename))
        self.assertEqual(p.rc, 0)
        # Assert that the file was downloaded properly.
        with open(local_filename, 'r') as f:
            self.assertEqual(f.read(), contents)

    def test_website_redirect_ignore_paramfile(self):
        bucket_name = self.create_bucket()
        foo_txt = self.files.create_file('foo.txt', 'bar')
        website_redirect = 'http://someserver'
        p = aws('s3 cp %s s3://%s/foo.txt --website-redirect %s' %
                (foo_txt, bucket_name, website_redirect))
        self.assert_no_errors(p)

        # Ensure that the web address is used as opposed to the contents
        # of the web address. We can check via a head object.
        response = self.head_object(bucket_name, 'foo.txt')
        self.assertEqual(response['WebsiteRedirectLocation'], website_redirect)

    @attr('slow')
    def test_copy_large_file_signature_v4(self):
        # Just verify that we can upload a large file to a region
        # that uses signature version 4.
        bucket_name = self.create_bucket(region='eu-central-1')
        num_mb = 200
        foo_txt = self.files.create_file('foo.txt', '')
        with open(foo_txt, 'wb') as f:
            for i in range(num_mb):
                f.write(b'a' * 1024 * 1024)

        p = aws('s3 cp %s s3://%s/ --region eu-central-1' % (
            foo_txt, bucket_name))
        self.assert_no_errors(p)
        self.assertTrue(self.key_exists(bucket_name, key_name='foo.txt'))

    def test_copy_metadata_directive(self):
        # Copy the same style of parsing as the CLI session. This is needed
        # For comparing expires timestamp.
        add_scalar_parsers(self.session)
        bucket_name = self.create_bucket()
        original_key = 'foo.txt'
        new_key = 'bar.txt'
        metadata = {
            'ContentType': 'foo',
            'ContentDisposition': 'foo',
            'ContentEncoding': 'foo',
            'ContentLanguage': 'foo',
            'CacheControl': '90',
            'Expires': '0'
        }
        self.put_object(bucket_name, original_key, contents='foo',
                        extra_args=metadata)
        p = aws('s3 cp s3://%s/%s s3://%s/%s' %
                (bucket_name, original_key, bucket_name, new_key))
        self.assert_no_errors(p)
        response = self.head_object(bucket_name, new_key)
        # These values should have the metadata of the source object
        metadata_ref = copy.copy(metadata)
        metadata_ref['Expires'] = 'Thu, 01 Jan 1970 00:00:00 GMT'
        for name, value in metadata_ref.items():
            self.assertEqual(response[name], value)

        # Use REPLACE to wipe out all of the metadata.
        p = aws('s3 cp s3://%s/%s s3://%s/%s --metadata-directive REPLACE' %
                (bucket_name, original_key, bucket_name, new_key))
        self.assert_no_errors(p)
        response = self.head_object(bucket_name, new_key)
        # Make sure all of the original metadata is gone.
        for name, value in metadata_ref.items():
            self.assertNotEqual(response.get(name), value)

        # Use REPLACE to wipe out all of the metadata but include a new
        # metadata value.
        p = aws('s3 cp s3://%s/%s s3://%s/%s --metadata-directive REPLACE '
                '--content-type bar' %
                (bucket_name, original_key, bucket_name, new_key))
        self.assert_no_errors(p)
        response = self.head_object(bucket_name, new_key)
        # Make sure the content type metadata is included
        self.assertEqual(response['ContentType'], 'bar')
        # Make sure all of the original metadata is gone.
        for name, value in metadata_ref.items():
            self.assertNotEqual(response.get(name), value)


class TestSync(BaseS3CLICommand):
    def test_sync_with_plus_chars_paginate(self):
        # This test ensures pagination tokens are url decoded.
        # 1. Create > 2 files with '+' in the filename.
        # 2. Sync up to s3 while the page size is 2.
        # 3. Sync up to s3 while the page size is 2.
        # 4. Verify nothing was synced up down from s3 in step 3.
        bucket_name = self.create_bucket()
        filenames = []
        for i in range(4):
            # Create a file with a space char and a '+' char in the filename.
            # We're interested in testing the filename comparisons, not the
            # mtime comparisons so we're setting the mtime to some time
            # in the past to avoid mtime comparisons interfering with
            # test results.
            mtime = time.time() - 300
            filenames.append(
                self.files.create_file('foo +%06d' % i,
                                       contents='',
                                       mtime=mtime))
        p = aws('s3 sync %s s3://%s/ --page-size 2' %
                (self.files.rootdir, bucket_name))
        self.assert_no_errors(p)
        time.sleep(1)
        p2 = aws('s3 sync %s s3://%s/ --page-size 2'
                 % (self.files.rootdir, bucket_name))
        self.assertNotIn('upload:', p2.stdout)
        self.assertEqual('', p2.stdout)

    def test_s3_to_s3_sync_with_plus_char_paginate(self):
        keynames = []
        for i in range(4):
            keyname = 'foo+%d' % i
            keynames.append(keyname)
            self.files.create_file(keyname, contents='')

        bucket_name = self.create_bucket()
        bucket_name_2 = self.create_bucket()

        p = aws('s3 sync %s s3://%s' % (self.files.rootdir, bucket_name))
        self.assert_no_errors(p)
        for key in keynames:
            self.assertTrue(self.key_exists(bucket_name, key))

        p = aws('s3 sync s3://%s/ s3://%s/ --page-size 2' %
                (bucket_name, bucket_name_2))
        self.assert_no_errors(p)
        for key in keynames:
            self.assertTrue(self.key_exists(bucket_name_2, key))

        p2 = aws('s3 sync s3://%s/ s3://%s/ --page-size 2' %
                 (bucket_name, bucket_name_2))
        self.assertNotIn('copy:', p2.stdout)
        self.assertEqual('', p2.stdout)

    def test_sync_no_resync(self):
        self.files.create_file('xyz123456789', contents='test1')
        self.files.create_file(os.path.join('xyz1', 'test'), contents='test2')
        self.files.create_file(os.path.join('xyz', 'test'), contents='test3')
        bucket_name = self.create_bucket()

        p = aws('s3 sync %s s3://%s' % (self.files.rootdir, bucket_name))
        self.assert_no_errors(p)
        time.sleep(2)
        self.assertTrue(self.key_exists(bucket_name, 'xyz123456789'))
        self.assertTrue(self.key_exists(bucket_name, 'xyz1/test'))
        self.assertTrue(self.key_exists(bucket_name, 'xyz/test'))

        p2 = aws('s3 sync %s s3://%s/' % (self.files.rootdir, bucket_name))
        self.assertNotIn('upload:', p2.stdout)
        self.assertEqual('', p2.stdout)

    def test_sync_to_from_s3(self):
        bucket_name = self.create_bucket()
        foo_txt = self.files.create_file('foo.txt', 'foo contents')
        bar_txt = self.files.create_file('bar.txt', 'bar contents')

        # Sync the directory and the bucket.
        p = aws('s3 sync %s s3://%s' % (self.files.rootdir, bucket_name))
        self.assert_no_errors(p)

        # Ensure both files are in the bucket.
        self.assertTrue(self.key_exists(bucket_name, 'foo.txt'))
        self.assertTrue(self.key_exists(bucket_name, 'bar.txt'))

        # Sync back down.  First remote the local files.
        os.remove(foo_txt)
        os.remove(bar_txt)
        p = aws('s3 sync s3://%s %s' % (bucket_name, self.files.rootdir))
        # The files should be back now.
        self.assertTrue(os.path.isfile(foo_txt))
        self.assertTrue(os.path.isfile(bar_txt))
        with open(foo_txt, 'r') as f:
            self.assertEqual(f.read(), 'foo contents')
        with open(bar_txt, 'r') as f:
            self.assertEqual(f.read(), 'bar contents')

    def test_sync_to_nonexistent_bucket(self):
        self.files.create_file('foo.txt', 'foo contents')
        self.files.create_file('bar.txt', 'bar contents')

        # Sync the directory and the bucket.
        p = aws('s3 sync %s s3://noexist-bkt-nme-1412' % (self.files.rootdir,))
        self.assertEqual(p.rc, 1)

    def test_sync_with_empty_files(self):
        self.files.create_file('foo.txt', 'foo contents')
        self.files.create_file('bar.txt', contents='')
        bucket_name = self.create_bucket()
        p = aws('s3 sync %s s3://%s/' % (self.files.rootdir, bucket_name))
        self.assertEqual(p.rc, 0)
        self.assertNotIn('failed', p.stderr)
        self.assertTrue(
            self.key_exists(bucket_name=bucket_name, key_name='bar.txt'))

    def test_sync_with_delete_option_with_same_prefix(self):
        # Test for issue 440 (https://github.com/aws/aws-cli/issues/440)
        # First, we need to create a directory structure that has a dir with
        # the same prefix as some of the files:
        #
        #  test/foo.txt
        #  test-123.txt
        #  test-321.txt
        #  test.txt
        bucket_name = self.create_bucket()
        # create test/foo.txt
        nested_dir = os.path.join(self.files.rootdir, 'test')
        os.mkdir(nested_dir)
        self.files.create_file(os.path.join(nested_dir, 'foo.txt'),
                               contents='foo.txt contents')
        # Then create test-123.txt, test-321.txt, test.txt.
        self.files.create_file('test-123.txt', 'test-123.txt contents')
        self.files.create_file('test-321.txt', 'test-321.txt contents')
        self.files.create_file('test.txt', 'test.txt contents')

        # Now sync this content up to s3.
        # Allow settling time so that we have a different time between
        # source and destination.
        time.sleep(2)
        p = aws('s3 sync %s s3://%s/' % (self.files.rootdir, bucket_name))
        self.assert_no_errors(p)

        # Now here's the issue.  If we try to sync the contents down
        # with the --delete flag we should *not* see any output, the
        # sync operation should determine that nothing is different and
        # therefore do nothing.  We can just use --dryrun to show the issue.
        p = aws('s3 sync s3://%s/ %s --dryrun --delete' % (
            bucket_name, self.files.rootdir))
        self.assert_no_errors(p)
        # These assertion methods will give better error messages than just
        # checking if the output is empty.
        self.assertNotIn('download:', p.stdout)
        self.assertNotIn('delete:', p.stdout)
        self.assertEqual('', p.stdout)


class TestSourceRegion(BaseS3CLICommand):
    def extra_setup(self):
        name_comp = []
        # This creates a non DNS compatible bucket name by making two random
        # sequences of characters and joining them with a period and
        # adding a .com at the end.
        for i in range(2):
            name_comp.append(''.join(random.sample(string.ascii_lowercase +
                                                   string.digits, 10)))
        self.src_name = '.'.join(name_comp + ['com'])
        name_comp = []
        for i in range(2):
            name_comp.append(''.join(random.sample(string.ascii_lowercase +
                                                   string.digits, 10)))
        self.dest_name = '.'.join(name_comp + ['com'])
        self.src_region = 'us-west-1'
        self.dest_region = 'us-east-1'
        self.src_bucket = self.create_bucket(self.src_name, self.src_region)
        self.dest_bucket = self.create_bucket(self.dest_name, self.dest_region)

    def test_fail_without_region(self):
        self.files.create_file('foo.txt', 'foo')
        p = aws('s3 sync %s s3://%s/ --region %s' %
                (self.files.rootdir, self.src_bucket, self.src_region))
        self.assert_no_errors(p)
        p2 = aws('s3 sync s3://%s/ s3://%s/ --region %s' %
                 (self.src_bucket, self.dest_bucket, self.src_region))
        self.assertEqual(p2.rc, 1, p2.stdout)
        self.assertIn('PermanentRedirect', p2.stderr)

    def test_cp_region(self):
        self.files.create_file('foo.txt', 'foo')
        p = aws('s3 sync %s s3://%s/ --region %s' %
                (self.files.rootdir, self.src_bucket, self.src_region))
        self.assert_no_errors(p)
        p2 = aws('s3 cp s3://%s/ s3://%s/ --region %s --source-region %s '
                 '--recursive' %
                 (self.src_bucket, self.dest_bucket, self.dest_region,
                  self.src_region))
        self.assertEqual(p2.rc, 0, p2.stdout)
        self.assertTrue(
            self.key_exists(bucket_name=self.dest_bucket, key_name='foo.txt'))

    def test_sync_region(self):
        self.files.create_file('foo.txt', 'foo')
        p = aws('s3 sync %s s3://%s/ --region %s' %
                (self.files.rootdir, self.src_bucket, self.src_region))
        self.assert_no_errors(p)
        p2 = aws('s3 sync s3://%s/ s3://%s/ --region %s --source-region %s ' %
                 (self.src_bucket, self.dest_bucket, self.dest_region,
                  self.src_region))
        self.assertEqual(p2.rc, 0, p2.stdout)
        self.assertTrue(
            self.key_exists(bucket_name=self.dest_bucket, key_name='foo.txt'))

    def test_mv_region(self):
        self.files.create_file('foo.txt', 'foo')
        p = aws('s3 sync %s s3://%s/ --region %s' %
                (self.files.rootdir, self.src_bucket, self.src_region))
        self.assert_no_errors(p)
        p2 = aws('s3 mv s3://%s/ s3://%s/ --region %s --source-region %s '
                 '--recursive' %
                 (self.src_bucket, self.dest_bucket, self.dest_region,
                  self.src_region))
        self.assertEqual(p2.rc, 0, p2.stdout)
        self.assertTrue(
            self.key_exists(bucket_name=self.dest_bucket, key_name='foo.txt'))
        self.assertFalse(
            self.key_exists(bucket_name=self.src_bucket, key_name='foo.txt'))

    @attr('slow')
    def test_mv_large_file_region(self):
        foo_txt = self.files.create_file('foo.txt', 'a' * 1024 * 1024 * 10)
        p = aws('s3 cp %s s3://%s/foo.txt --region %s' %
                (foo_txt, self.src_bucket, self.src_region))
        self.assert_no_errors(p)

        p2 = aws(
            's3 mv s3://%s/foo.txt s3://%s/ --region %s --source-region %s ' %
            (self.src_bucket, self.dest_bucket, self.dest_region,
             self.src_region)
        )
        self.assert_no_errors(p2)
        self.assertTrue(
            self.key_exists(bucket_name=self.dest_bucket, key_name='foo.txt'))
        self.assertFalse(
            self.key_exists(bucket_name=self.src_bucket, key_name='foo.txt'))


class TestWarnings(BaseS3CLICommand):
    def extra_setup(self):
        self.bucket_name = self.create_bucket()

    def test_no_exist(self):
        filename = os.path.join(self.files.rootdir, "no-exists-file")
        p = aws('s3 cp %s s3://%s/' % (filename, self.bucket_name))
        self.assertEqual(p.rc, 2, p.stderr)
        self.assertIn('warning: Skipping file %s. File does not exist.' %
                      filename, p.stderr)

    @unittest.skipIf(platform.system() not in ['Darwin', 'Linux'],
                     'Read permissions tests only supported on mac/linux')
    def test_no_read_access(self):
        if os.geteuid() == 0:
            self.skipTest('Cannot completely remove read access as root user.')
        self.files.create_file('foo.txt', 'foo')
        filename = os.path.join(self.files.rootdir, 'foo.txt')
        permissions = stat.S_IMODE(os.stat(filename).st_mode)
        # Remove read permissions
        permissions = permissions ^ stat.S_IREAD
        os.chmod(filename, permissions)
        p = aws('s3 cp %s s3://%s/' % (filename, self.bucket_name))
        self.assertEqual(p.rc, 2, p.stderr)
        self.assertIn('warning: Skipping file %s. File/Directory is '
                      'not readable.' % filename, p.stderr)

    @unittest.skipIf(platform.system() not in ['Darwin', 'Linux'],
                     'Special files only supported on mac/linux')
    def test_is_special_file(self):
        file_path = os.path.join(self.files.rootdir, 'foo')
        # Use socket for special file.
        sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        sock.bind(file_path)
        p = aws('s3 cp %s s3://%s/' % (file_path, self.bucket_name))
        self.assertEqual(p.rc, 2, p.stderr)
        self.assertIn(("warning: Skipping file %s. File is character "
                       "special device, block special device, FIFO, or "
                       "socket." % file_path), p.stderr)


@unittest.skipIf(platform.system() not in ['Darwin', 'Linux'],
                 'Symlink tests only supported on mac/linux')
class TestSymlinks(BaseS3CLICommand):
    """
    This class test the ability to follow or not follow symlinks.
    """
    def extra_setup(self):
        self.bucket_name = self.create_bucket()
        self.nested_dir = os.path.join(self.files.rootdir, 'realfiles')
        os.mkdir(self.nested_dir)
        self.sample_file = \
            self.files.create_file(os.path.join(self.nested_dir, 'foo.txt'),
                                   contents='foo.txt contents')
        # Create a symlink to foo.txt.
        os.symlink(self.sample_file, os.path.join(self.files.rootdir,
                                                  'a-goodsymlink'))
        # Create a bad symlink.
        os.symlink('non-existent-file', os.path.join(self.files.rootdir,
                                                     'b-badsymlink'))
        # Create a symlink to directory where foo.txt is.
        os.symlink(self.nested_dir, os.path.join(self.files.rootdir,
                                                 'c-goodsymlink'))

    def test_no_follow_symlinks(self):
        p = aws('s3 sync %s s3://%s/ --no-follow-symlinks' % (
            self.files.rootdir, self.bucket_name))
        self.assert_no_errors(p)
        self.assertTrue(not self.key_exists(self.bucket_name,
                        'a-goodsymlink'))
        self.assertTrue(not self.key_exists(self.bucket_name,
                        'b-badsymlink'))
        self.assertTrue(not self.key_exists(self.bucket_name,
                        'c-goodsymlink/foo.txt'))
        self.assertEqual(self.get_key_contents(self.bucket_name,
                                               key_name='realfiles/foo.txt'),
                         'foo.txt contents')

    def test_follow_symlinks(self):
        # Get rid of the bad symlink first.
        os.remove(os.path.join(self.files.rootdir, 'b-badsymlink'))
        p = aws('s3 sync %s s3://%s/ --follow-symlinks' %
                (self.files.rootdir, self.bucket_name))
        self.assert_no_errors(p)
        self.assertEqual(self.get_key_contents(self.bucket_name,
                                               key_name='a-goodsymlink'),
                         'foo.txt contents')
        self.assertTrue(not self.key_exists(self.bucket_name,
                        'b-badsymlink'))
        self.assertEqual(
            self.get_key_contents(self.bucket_name,
                                  key_name='c-goodsymlink/foo.txt'),
            'foo.txt contents')
        self.assertEqual(self.get_key_contents(self.bucket_name,
                                               key_name='realfiles/foo.txt'),
                         'foo.txt contents')

    def test_follow_symlinks_default(self):
        # Get rid of the bad symlink first.
        os.remove(os.path.join(self.files.rootdir, 'b-badsymlink'))
        p = aws('s3 sync %s s3://%s/' %
                (self.files.rootdir, self.bucket_name))
        self.assert_no_errors(p)
        self.assertEqual(self.get_key_contents(self.bucket_name,
                                               key_name='a-goodsymlink'),
                         'foo.txt contents')
        self.assertTrue(not self.key_exists(self.bucket_name,
                        'b-badsymlink'))
        self.assertEqual(
            self.get_key_contents(self.bucket_name,
                                  key_name='c-goodsymlink/foo.txt'),
            'foo.txt contents')
        self.assertEqual(self.get_key_contents(self.bucket_name,
                                               key_name='realfiles/foo.txt'),
                         'foo.txt contents')

    def test_bad_symlink(self):
        p = aws('s3 sync %s s3://%s/' % (self.files.rootdir, self.bucket_name))
        self.assertEqual(p.rc, 2, p.stderr)
        self.assertIn('warning: Skipping file %s. File does not exist.' %
                      os.path.join(self.files.rootdir, 'b-badsymlink'),
                      p.stderr)


class TestUnicode(BaseS3CLICommand):
    """
    The purpose of these tests are to ensure that the commands can handle
    unicode characters in both keyname and from those generated for both
    uploading and downloading files.
    """

    def test_cp(self):
        bucket_name = self.create_bucket()
        local_example1_txt = \
            self.files.create_file(u'\u00e9xample.txt', 'example1 contents')
        s3_example1_txt = 's3://%s/%s' % (bucket_name,
                                          os.path.basename(local_example1_txt))
        local_example2_txt = self.files.full_path(u'\u00e9xample2.txt')

        p = aws('s3 cp %s %s' % (local_example1_txt, s3_example1_txt))
        self.assert_no_errors(p)

        # Download the file to the second example2.txt filename.
        p = aws('s3 cp %s %s --quiet' % (s3_example1_txt, local_example2_txt))
        self.assert_no_errors(p)
        with open(local_example2_txt, 'rb') as f:
            self.assertEqual(f.read(), b'example1 contents')

    def test_recursive_cp(self):
        bucket_name = self.create_bucket()
        local_example1_txt = self.files.create_file(u'\u00e9xample.txt',
                                                    'example1 contents')
        local_example2_txt = self.files.create_file(u'\u00e9xample2.txt',
                                                    'example2 contents')
        p = aws('s3 cp %s s3://%s --recursive --quiet' % (
            self.files.rootdir, bucket_name))
        self.assert_no_errors(p)

        os.remove(local_example1_txt)
        os.remove(local_example2_txt)

        p = aws('s3 cp s3://%s %s --recursive --quiet' % (
            bucket_name, self.files.rootdir))
        self.assert_no_errors(p)
        self.assertEqual(open(local_example1_txt).read(), 'example1 contents')
        self.assertEqual(open(local_example2_txt).read(), 'example2 contents')


class TestLs(BaseS3CLICommand):
    """
    This tests using the ``ls`` command.
    """
    def test_ls_bucket(self):
        p = aws('s3 ls')
        self.assert_no_errors(p)

    def test_ls_with_no_env_vars(self):
        # By default, the aws() function injects
        # an AWS_DEFAULT_REGION into the env var of the
        # process.  We're verifying that a region does *not*
        # need to be set anywhere.  If we provide our
        # own environ dict, then the aws() function won't
        # inject a region.
        env = os.environ.copy()
        p = aws('s3 ls', env_vars=env)
        self.assert_no_errors(p)

    def test_ls_bucket_with_s3_prefix(self):
        p = aws('s3 ls s3://')
        self.assert_no_errors(p)

    def test_ls_non_existent_bucket(self):
        p = aws('s3 ls s3://foobara99842u4wbts829381')
        self.assertEqual(p.rc, 255)
        self.assertIn(
            ('A client error (NoSuchBucket) occurred when calling the '
             'ListObjects operation: The specified bucket does not exist'),
            p.stderr)
        # There should be no stdout if we can't find the bucket.
        self.assertEqual(p.stdout, '')

    def test_ls_with_prefix(self):
        bucket_name = self.create_bucket()
        self.put_object(bucket_name, 'foo.txt', 'contents')
        self.put_object(bucket_name, 'foo', 'contents')
        self.put_object(bucket_name, 'bar.txt', 'contents')
        self.put_object(bucket_name, 'subdir/foo.txt', 'contents')
        p = aws('s3 ls s3://%s' % bucket_name)
        self.assertIn('PRE subdir/', p.stdout)
        self.assertIn('8 foo.txt', p.stdout)
        self.assertIn('8 foo', p.stdout)
        self.assertIn('8 bar.txt', p.stdout)

    def test_ls_recursive(self):
        bucket_name = self.create_bucket()
        self.put_object(bucket_name, 'foo.txt', 'contents')
        self.put_object(bucket_name, 'foo', 'contents')
        self.put_object(bucket_name, 'bar.txt', 'contents')
        self.put_object(bucket_name, 'subdir/foo.txt', 'contents')
        p = aws('s3 ls s3://%s --recursive' % bucket_name)
        self.assertIn('8 foo.txt', p.stdout)
        self.assertIn('8 foo', p.stdout)
        self.assertIn('8 bar.txt', p.stdout)
        self.assertIn('8 subdir/foo.txt', p.stdout)

    def test_ls_without_prefix(self):
        # The ls command does not require an s3:// prefix,
        # we're always listing s3 contents.
        bucket_name = self.create_bucket()
        self.put_object(bucket_name, 'foo.txt', 'contents')
        p = aws('s3 ls %s' % bucket_name)
        self.assertEqual(p.rc, 0)
        self.assertIn('foo.txt', p.stdout)

    def test_only_prefix(self):
        bucket_name = self.create_bucket()
        self.put_object(bucket_name, 'temp/foo.txt', 'contents')
        p = aws('s3 ls s3://%s/temp/foo.txt' % bucket_name)
        self.assertEqual(p.rc, 0)
        self.assertIn('foo.txt', p.stdout)

    def test_ls_empty_bucket(self):
        bucket_name = self.create_bucket()
        p = aws('s3 ls %s' % bucket_name)
        # There should not be an error thrown for checking the contents of
        # an empty bucket because no key was specified.
        self.assertEqual(p.rc, 0)

    def test_ls_fail(self):
        bucket_name = self.create_bucket()
        p = aws('s3 ls s3://%s/foo' % bucket_name)
        self.assertEqual(p.rc, 1)

    def test_ls_fail_recursive(self):
        bucket_name = self.create_bucket()
        p = aws('s3 ls s3://%s/bar --recursive' % bucket_name)
        self.assertEqual(p.rc, 1)


class TestMbRb(BaseS3CLICommand):
    """
    Tests primarily using ``rb`` and ``mb`` command.
    """
    def extra_setup(self):
        self.bucket_name = 'awscli-s3integ-' + str(random.randint(1, 1000))

    def test_mb_rb(self):
        p = aws('s3 mb s3://%s' % self.bucket_name)
        self.assert_no_errors(p)

        # Give the bucket time to form.
        time.sleep(1)
        response = self.list_buckets()
        self.assertIn(self.bucket_name, [b['Name'] for b in response])

        p = aws('s3 rb s3://%s' % self.bucket_name)
        self.assert_no_errors(p)

    def test_fail_mb_rb(self):
        # Choose a bucket name that already exists.
        p = aws('s3 mb s3://mybucket')
        self.assertIn("BucketAlreadyExists", p.stderr)
        self.assertEqual(p.rc, 1)


class TestOutput(BaseS3CLICommand):
    """
    This ensures that arguments that affect output i.e. ``--quiet`` and
    ``--only-show-errors`` behave as expected.
    """
    def test_normal_output(self):
        # Make a bucket.
        bucket_name = self.create_bucket()
        foo_txt = self.files.create_file('foo.txt', 'foo contents')

        # Copy file into bucket.
        p = aws('s3 cp %s s3://%s/' % (foo_txt, bucket_name))
        self.assertEqual(p.rc, 0)
        # Check that there were no errors and that parts of the expected
        # progress message are written to stdout.
        self.assert_no_errors(p)
        self.assertIn('upload', p.stdout)
        self.assertIn('s3://%s/foo.txt' % bucket_name, p.stdout)

    def test_normal_output_quiet(self):
        # Make a bucket.
        bucket_name = self.create_bucket()
        foo_txt = self.files.create_file('foo.txt', 'foo contents')

        # Copy file into bucket.
        p = aws('s3 cp %s s3://%s/ --quiet' % (foo_txt, bucket_name))
        self.assertEqual(p.rc, 0)
        # Check that nothing was printed to stdout.
        self.assertEqual('', p.stdout)

    def test_normal_output_only_show_errors(self):
        # Make a bucket.
        bucket_name = self.create_bucket()
        foo_txt = self.files.create_file('foo.txt', 'foo contents')

        # Copy file into bucket.
        p = aws('s3 cp %s s3://%s/ --only-show-errors' % (foo_txt,
                                                          bucket_name))
        self.assertEqual(p.rc, 0)
        # Check that nothing was printed to stdout.
        self.assertEqual('', p.stdout)

    def test_error_output(self):
        foo_txt = self.files.create_file('foo.txt', 'foo contents')

        # Copy file into bucket.
        p = aws('s3 cp %s s3://non-existant-bucket/' % foo_txt)
        # Check that there were errors and that the error was print to stderr.
        self.assertEqual(p.rc, 1)
        self.assertIn('upload failed', p.stderr)

    def test_error_ouput_quiet(self):
        foo_txt = self.files.create_file('foo.txt', 'foo contents')

        # Copy file into bucket.
        p = aws('s3 cp %s s3://non-existant-bucket/ --quiet' % foo_txt)
        # Check that there were errors and that the error was not
        # print to stderr.
        self.assertEqual(p.rc, 1)
        self.assertEqual('', p.stderr)

    def test_error_ouput_only_show_errors(self):
        foo_txt = self.files.create_file('foo.txt', 'foo contents')

        # Copy file into bucket.
        p = aws('s3 cp %s s3://non-existant-bucket/ --only-show-errors'
                % foo_txt)
        # Check that there were errors and that the error was print to stderr.
        self.assertEqual(p.rc, 1)
        self.assertIn('upload failed', p.stderr)

    def test_error_and_success_output_only_show_errors(self):
        # Make a bucket.
        bucket_name = self.create_bucket()

        # Create one file.
        self.files.create_file('f', 'foo contents')

        # Create another file that has a slightly longer name than the first.
        self.files.create_file('bar.txt', 'bar contents')

        # Create a prefix that will cause the second created file to have a key
        # longer than 1024 bytes which is not allowed in s3.
        long_prefix = 'd' * 1022

        p = aws('s3 cp %s s3://%s/%s/ --only-show-errors --recursive'
                % (self.files.rootdir, bucket_name, long_prefix))

        # Check that there was at least one error.
        self.assertEqual(p.rc, 1)

        # Check that there was nothing written to stdout for successful upload.
        self.assertEqual('', p.stdout)

        # Check that the failed message showed up in stderr.
        self.assertIn('upload failed', p.stderr)

        # Ensure the expected successful key exists in the bucket.
        self.assertTrue(self.key_exists(bucket_name, long_prefix + '/f'))


class TestDryrun(BaseS3CLICommand):
    """
    This ensures that dryrun works.
    """
    def test_dryrun(self):
        # Make a bucket.
        bucket_name = self.create_bucket()
        foo_txt = self.files.create_file('foo.txt', 'foo contents')

        # Copy file into bucket.
        p = aws('s3 cp %s s3://%s/ --dryrun' % (foo_txt, bucket_name))
        self.assertEqual(p.rc, 0)
        self.assert_no_errors(p)
        self.assertFalse(self.key_exists(bucket_name, 'foo.txt'))

    def test_dryrun_large_files(self):
        bucket_name = self.create_bucket()
        foo_txt = self.files.create_file('foo.txt', 'a' * 1024 * 1024 * 10)

        # Copy file into bucket.
        p = aws('s3 cp %s s3://%s/ --dryrun' % (foo_txt, bucket_name))
        self.assertEqual(p.rc, 0)
        self.assert_no_errors(p)
        self.assertFalse(
            self.key_exists(bucket_name, 'foo.txt'),
            "The key 'foo.txt' exists in S3. It looks like the --dryrun "
            "argument was not obeyed.")

    def test_dryrun_download_large_file(self):
        bucket_name = self.create_bucket()
        full_path = self.files.create_file('largefile', 'a' * 1024 * 1024 * 10)
        with open(full_path, 'rb') as body:
            self.put_object(bucket_name, 'foo.txt', body)

        foo_txt = self.files.full_path('foo.txt')
        p = aws('s3 cp s3://%s/foo.txt %s --dryrun' % (bucket_name, foo_txt))
        self.assertEqual(p.rc, 0)
        self.assert_no_errors(p)
        self.assertFalse(
            os.path.exists(foo_txt),
            "The file 'foo.txt' exists locally. It looks like the --dryrun "
            "argument was not obeyed.")


@unittest.skipIf(platform.system() not in ['Darwin', 'Linux'],
                 'Memory tests only supported on mac/linux')
class TestMemoryUtilization(BaseS3CLICommand):
    # These tests verify the memory utilization and growth are what we expect.
    def extra_setup(self):
        self.num_threads = DEFAULTS['max_concurrent_requests']
        self.chunk_size = DEFAULTS['multipart_chunksize']
        expected_memory_usage = self.num_threads * self.chunk_size
        # margin for things like python VM overhead, botocore service
        # objects, etc.  1.5 is really generous, perhaps over time this can be
        # lowered.
        runtime_margin = 1.5
        self.max_mem_allowed = runtime_margin * expected_memory_usage

    def assert_max_memory_used(self, process, max_mem_allowed, full_command):
        peak_memory = max(process.memory_usage)
        if peak_memory > max_mem_allowed:
            failure_message = (
                'Exceeded max memory allowed (%s MB) for command '
                '"%s": %s MB' % (self.max_mem_allowed / 1024.0 / 1024.0,
                                 full_command,
                                 peak_memory / 1024.0 / 1024.0))
            self.fail(failure_message)

    @attr('slow')
    def test_transfer_single_large_file(self):
        # 40MB will force a multipart upload.
        bucket_name = self.create_bucket()
        file_contents = 'abcdabcd' * (1024 * 1024 * 10)
        foo_txt = self.files.create_file('foo.txt', file_contents)
        full_command = 's3 mv %s s3://%s/foo.txt' % (foo_txt, bucket_name)
        p = aws(full_command, collect_memory=True)
        self.assert_no_errors(p)
        self.assert_max_memory_used(p, self.max_mem_allowed, full_command)

        # Verify downloading it back down obeys memory utilization.
        download_full_command = 's3 mv s3://%s/foo.txt %s' % (
            bucket_name, foo_txt)
        p = aws(download_full_command, collect_memory=True)
        self.assert_no_errors(p)
        self.assert_max_memory_used(p, self.max_mem_allowed,
                                    download_full_command)

    # Some versions of RHEL allocate memory in a way where free'd memory isn't
    # given back to the OS.  We haven't seen behavior as bad as RHEL's to the
    # point where this test fails on other distros, so for now we're disabling
    # the test on RHEL until we come up with a better way to collect
    # memory usage.
    @attr('slow')
    @unittest.skipIf(_running_on_rhel(),
                     'Streaming memory tests no supported on RHEL.')
    def test_stream_large_file(self):
        """
        This tests to ensure that streaming files for both uploads and
        downloads do not use too much memory.  Note that streaming uploads
        will use slightly more memory than usual but should not put the
        entire file into memory.
        """
        bucket_name = self.create_bucket()

        # Create a 200 MB file that will be streamed
        num_mb = 200
        foo_txt = self.files.create_file('foo.txt', '')
        with open(foo_txt, 'wb') as f:
            for i in range(num_mb):
                f.write(b'a' * 1024 * 1024)

        # The current memory threshold is set at about the peak amount for
        # performing a streaming upload of a file larger than 100 MB. So
        # this maximum needs to be bumped up.  The maximum memory allowance
        # is increased by two chunksizes because that is the maximum
        # amount of chunks that will be queued while not being operated on
        # by a thread when performing a streaming multipart upload.
        max_mem_allowed = self.max_mem_allowed + 2 * self.chunk_size

        full_command = 's3 cp - s3://%s/foo.txt' % bucket_name
        with open(foo_txt, 'rb') as f:
            p = aws(full_command, input_file=f, collect_memory=True)
            self.assert_no_errors(p)
            self.assert_max_memory_used(p, max_mem_allowed, full_command)

        # Now perform a streaming download of the file.
        full_command = 's3 cp s3://%s/foo.txt - > %s' % (bucket_name, foo_txt)
        p = aws(full_command, collect_memory=True)
        self.assert_no_errors(p)
        # Use the ususal bar for maximum memory usage since a streaming
        # download's memory usage should be comparable to non-streaming
        # transfers.
        self.assert_max_memory_used(p, self.max_mem_allowed, full_command)


class TestWebsiteConfiguration(BaseS3CLICommand):
    def test_create_website_index_configuration(self):
        bucket_name = self.create_bucket()
        # Supply only --index-document argument.
        full_command = 's3 website %s --index-document index.html' % \
            (bucket_name)
        p = aws(full_command)
        self.assertEqual(p.rc, 0)
        self.assert_no_errors(p)
        # Verify we have a bucket website configured.
        parsed = self.client.get_bucket_website(Bucket=bucket_name)
        self.assertEqual(parsed['IndexDocument']['Suffix'], 'index.html')
        self.assertNotIn('ErrorDocument', parsed)
        self.assertNotIn('RoutingRules', parsed)
        self.assertNotIn('RedirectAllRequestsTo', parsed)

    def test_create_website_index_and_error_configuration(self):
        bucket_name = self.create_bucket()
        # Supply both --index-document and --error-document arguments.
        p = aws('s3 website %s --index-document index.html '
                '--error-document error.html' % bucket_name)
        self.assertEqual(p.rc, 0)
        self.assert_no_errors(p)
        # Verify we have a bucket website configured.
        parsed = self.client.get_bucket_website(Bucket=bucket_name)
        self.assertEqual(parsed['IndexDocument']['Suffix'], 'index.html')
        self.assertEqual(parsed['ErrorDocument']['Key'], 'error.html')
        self.assertNotIn('RoutingRules', parsed)
        self.assertNotIn('RedirectAllRequestsTo', parsed)


class TestIncludeExcludeFilters(BaseS3CLICommand):
    def assert_no_files_would_be_uploaded(self, p):
        self.assert_no_errors(p)
        # There should be no output.
        self.assertEqual(p.stdout, '')
        self.assertEqual(p.stderr, '')

    def test_basic_exclude_filter_for_single_file(self):
        full_path = self.files.create_file('foo.txt', 'this is foo.txt')
        # With no exclude we should upload the file.
        p = aws('s3 cp %s s3://random-bucket-name/ --dryrun' % full_path)
        self.assert_no_errors(p)
        self.assertIn('(dryrun) upload:', p.stdout)

        p2 = aws("s3 cp %s s3://random-bucket-name/ --dryrun --exclude '*'"
                 % full_path)
        self.assert_no_files_would_be_uploaded(p2)

    def test_explicitly_exclude_single_file(self):
        full_path = self.files.create_file('foo.txt', 'this is foo.txt')
        p = aws('s3 cp %s s3://random-bucket-name/'
                ' --dryrun --exclude foo.txt'
                % full_path)
        self.assert_no_files_would_be_uploaded(p)

    def test_cwd_doesnt_matter(self):
        full_path = self.files.create_file('foo.txt', 'this is foo.txt')
        tempdir = tempfile.mkdtemp()
        self.addCleanup(shutil.rmtree, tempdir)
        with cd(tempdir):
            p = aws("s3 cp %s s3://random-bucket-name/ --dryrun --exclude '*'"
                    % full_path)
        self.assert_no_files_would_be_uploaded(p)

    def test_recursive_exclude(self):
        # create test/foo.txt
        nested_dir = os.path.join(self.files.rootdir, 'test')
        os.mkdir(nested_dir)
        self.files.create_file(os.path.join(nested_dir, 'foo.txt'),
                               contents='foo.txt contents')
        # Then create test-123.txt, test-321.txt, test.txt.
        self.files.create_file('test-123.txt', 'test-123.txt contents')
        self.files.create_file('test-321.txt', 'test-321.txt contents')
        self.files.create_file('test.txt', 'test.txt contents')
        # An --exclude test* should exclude everything here.
        p = aws("s3 cp %s s3://random-bucket-name/ --dryrun --exclude '*' "
                "--recursive" % self.files.rootdir)
        self.assert_no_files_would_be_uploaded(p)

        # We can include the test directory though.
        p = aws("s3 cp %s s3://random-bucket-name/ --dryrun "
                "--exclude '*' --include 'test/*' --recursive"
                % self.files.rootdir)
        self.assert_no_errors(p)
        self.assertRegexpMatches(p.stdout,
                                 r'\(dryrun\) upload:.*test/foo.txt.*')

    def test_s3_filtering(self):
        # Should behave the same as local file filtering.
        bucket_name = self.create_bucket()
        self.put_object(bucket_name, key_name='foo.txt')
        self.put_object(bucket_name, key_name='bar.txt')
        self.put_object(bucket_name, key_name='baz.jpg')
        p = aws("s3 rm s3://%s/ --dryrun --exclude '*' --recursive"
                % bucket_name)
        self.assert_no_files_would_be_uploaded(p)

        p = aws(
            "s3 rm s3://%s/ --dryrun --exclude '*.jpg' --exclude '*.txt' "
            "--recursive" % bucket_name)
        self.assert_no_files_would_be_uploaded(p)

        p = aws("s3 rm s3://%s/ --dryrun --exclude '*.txt' --recursive"
                % bucket_name)
        self.assert_no_errors(p)
        self.assertRegexpMatches(p.stdout, r'\(dryrun\) delete:.*baz.jpg.*')
        self.assertNotIn(p.stdout, 'bar.txt')
        self.assertNotIn(p.stdout, 'foo.txt')

    def test_exclude_filter_with_delete(self):
        # Test for: https://github.com/aws/aws-cli/issues/778
        bucket_name = self.create_bucket()
        self.files.create_file('foo.txt', 'contents')
        second = self.files.create_file('bar.py', 'contents')
        p = aws("s3 sync %s s3://%s/" % (self.files.rootdir, bucket_name))
        self.assert_no_errors(p)
        self.assertTrue(self.key_exists(bucket_name, key_name='bar.py'))
        os.remove(second)
        # We now have the same state as specified in the bug:
        # local           remote
        # -----           ------
        #
        # foo.txt         foo.txt
        #                 bar.py
        #
        # If we now run --exclude '*.py' --delete, then we should *not*
        # delete bar.py and the remote side.
        p = aws("s3 sync %s s3://%s/ --exclude '*.py' --delete" % (
            self.files.rootdir, bucket_name))
        self.assert_no_errors(p)
        self.assertTrue(
            self.key_exists(bucket_name, key_name='bar.py'),
            ("The --delete flag was not applied to the receiving "
             "end, the 'bar.py' file was deleted even though it"
             " was excluded."))

    def test_exclude_filter_with_relative_path(self):
        # Same test as test_exclude_filter_with_delete, except we don't
        # use an absolute path on the source dir.
        bucket_name = self.create_bucket()
        self.files.create_file('foo.txt', 'contents')
        second = self.files.create_file('bar.py', 'contents')
        p = aws("s3 sync %s s3://%s/" % (self.files.rootdir, bucket_name))
        self.assert_no_errors(p)
        self.assertTrue(self.key_exists(bucket_name, key_name='bar.py'))
        os.remove(second)
        cwd = os.getcwd()
        try:
            os.chdir(self.files.rootdir)
            # Note how we're using "." for the source directory.
            p = aws("s3 sync . s3://%s/ --exclude '*.py' --delete"
                    % bucket_name)
        finally:
            os.chdir(cwd)
        self.assert_no_errors(p)
        self.assertTrue(
            self.key_exists(bucket_name, key_name='bar.py'),
            ("The --delete flag was not applied to the receiving "
             "end, the 'bar.py' file was deleted even though"
             " it was excluded."))

    def test_filter_s3_with_prefix(self):
        bucket_name = self.create_bucket()
        self.put_object(bucket_name, key_name='temp/test')
        p = aws('s3 cp s3://%s/temp/ %s --recursive --exclude test --dryrun'
                % (bucket_name, self.files.rootdir))
        self.assert_no_files_would_be_uploaded(p)

    def test_filter_no_resync(self):
        # This specifically tests for the issue described here:
        # https://github.com/aws/aws-cli/issues/794
        bucket_name = self.create_bucket()
        dir_name = os.path.join(self.files.rootdir, 'temp')
        self.files.create_file(os.path.join(dir_name, 'test.txt'),
                               contents='foo')
        # Sync a local directory to an s3 prefix.
        p = aws('s3 sync %s s3://%s/temp' % (dir_name, bucket_name))
        self.assert_no_errors(p)
        self.assertTrue(self.key_exists(bucket_name, key_name='temp/test.txt'))

        # Nothing should be synced down if filters are used.
        p = aws("s3 sync s3://%s/temp %s --exclude '*' --include test.txt"
                % (bucket_name, dir_name))
        self.assert_no_files_would_be_uploaded(p)


class TestFileWithSpaces(BaseS3CLICommand):
    def test_upload_download_file_with_spaces(self):
        bucket_name = self.create_bucket()
        filename = self.files.create_file('with space.txt', 'contents')
        p = aws('s3 cp %s s3://%s/ --recursive' % (self.files.rootdir,
                                                   bucket_name))
        self.assert_no_errors(p)
        os.remove(filename)
        # Now download the file back down locally.
        p = aws('s3 cp s3://%s/ %s --recursive' % (bucket_name,
                                                   self.files.rootdir))
        self.assert_no_errors(p)
        self.assertEqual(os.listdir(self.files.rootdir)[0], 'with space.txt')

    def test_sync_file_with_spaces(self):
        bucket_name = self.create_bucket()
        bucket_name = self.create_bucket()
        self.files.create_file('with space.txt',
                               'contents', mtime=time.time() - 300)
        p = aws('s3 sync %s s3://%s/' % (self.files.rootdir,
                                         bucket_name))
        self.assert_no_errors(p)
        time.sleep(1)
        # Now syncing again should *not* trigger any uploads (i.e we should
        # get nothing on stdout).
        p2 = aws('s3 sync %s s3://%s/' % (self.files.rootdir,
                                          bucket_name))
        self.assertEqual(p2.stdout, '')
        self.assertEqual(p2.stderr, '')
        self.assertEqual(p2.rc, 0)


class TestStreams(BaseS3CLICommand):
    def test_upload(self):
        """
        This tests uploading a small stream from stdin.
        """
        bucket_name = self.create_bucket()
        p = aws('s3 cp - s3://%s/stream' % bucket_name,
                input_data=b'This is a test')
        self.assert_no_errors(p)
        self.assertTrue(self.key_exists(bucket_name, 'stream'))
        self.assertEqual(self.get_key_contents(bucket_name, 'stream'),
                         'This is a test')

    def test_unicode_upload(self):
        """
        This tests being able to upload unicode from stdin.
        """
        unicode_str = u'\u00e9 This is a test'
        byte_str = unicode_str.encode('utf-8')
        bucket_name = self.create_bucket()
        p = aws('s3 cp - s3://%s/stream' % bucket_name,
                input_data=byte_str)
        self.assert_no_errors(p)
        self.assertTrue(self.key_exists(bucket_name, 'stream'))
        self.assertEqual(self.get_key_contents(bucket_name, 'stream'),
                         unicode_str)

    @attr('slow')
    def test_multipart_upload(self):
        """
        This tests the ability to multipart upload streams from stdin.
        The data has some unicode in it to avoid having to do a seperate
        multipart upload test just for unicode.
        """

        bucket_name = self.create_bucket()
        data = u'\u00e9bcd' * (1024 * 1024 * 10)
        data_encoded = data.encode('utf-8')
        p = aws('s3 cp - s3://%s/stream' % bucket_name,
                input_data=data_encoded)
        self.assert_no_errors(p)
        self.assertTrue(self.key_exists(bucket_name, 'stream'))
        self.assert_key_contents_equal(bucket_name, 'stream', data)

    def test_download(self):
        """
        This tests downloading a small stream from stdout.
        """
        bucket_name = self.create_bucket()
        p = aws('s3 cp - s3://%s/stream' % bucket_name,
                input_data=b'This is a test')
        self.assert_no_errors(p)

        p = aws('s3 cp s3://%s/stream -' % bucket_name)
        self.assert_no_errors(p)
        self.assertEqual(p.stdout, 'This is a test')

    def test_unicode_download(self):
        """
        This tests downloading a small unicode stream from stdout.
        """
        bucket_name = self.create_bucket()

        data = u'\u00e9 This is a test'
        data_encoded = data.encode('utf-8')
        p = aws('s3 cp - s3://%s/stream' % bucket_name,
                input_data=data_encoded)
        self.assert_no_errors(p)

        # Downloading the unicode stream to standard out.
        p = aws('s3 cp s3://%s/stream -' % bucket_name)
        self.assert_no_errors(p)
        self.assertEqual(p.stdout, data_encoded.decode(get_stdout_encoding()))

    @attr('slow')
    def test_multipart_download(self):
        """
        This tests the ability to multipart download streams to stdout.
        The data has some unicode in it to avoid having to do a seperate
        multipart download test just for unicode.
        """
        bucket_name = self.create_bucket()

        # First lets upload some data via streaming since
        # its faster and we do not have to write to a file!
        data = u'\u00e9bcd' * (1024 * 1024 * 10)
        data_encoded = data.encode('utf-8')
        p = aws('s3 cp - s3://%s/stream' % bucket_name,
                input_data=data_encoded)

        # Download the unicode stream to standard out.
        p = aws('s3 cp s3://%s/stream -' % bucket_name)
        self.assert_no_errors(p)
        self.assertEqual(p.stdout, data_encoded.decode(get_stdout_encoding()))


class TestLSWithProfile(BaseS3CLICommand):
    def extra_setup(self):
        self.config_file = os.path.join(self.files.rootdir, 'tmpconfig')
        with open(self.config_file, 'w') as f:
            creds = self.session.get_credentials()
            f.write(
                "[profile testprofile]\n"
                "aws_access_key_id=%s\n"
                "aws_secret_access_key=%s\n" % (
                    creds.access_key,
                    creds.secret_key)
            )
            if creds.token is not None:
                f.write("aws_session_token=%s\n" % creds.token)

    def test_can_ls_with_profile(self):
        env_vars = os.environ.copy()
        env_vars['AWS_CONFIG_FILE'] = self.config_file
        p = aws('s3 ls s3:// --profile testprofile', env_vars=env_vars)
        self.assert_no_errors(p)


class TestNoSignRequests(BaseS3CLICommand):
    def test_no_sign_request(self):
        bucket_name = self.create_bucket()
        self.put_object(bucket_name, 'foo', contents='bar',
                        extra_args={'ACL': 'public-read-write'})
        env_vars = os.environ.copy()
        env_vars['AWS_ACCESS_KEY_ID'] = 'foo'
        env_vars['AWS_SECRET_ACCESS_KEY'] = 'bar'
        p = aws('s3 cp s3://%s/foo %s/ --region %s' %
                (bucket_name, self.files.rootdir, self.region),
                env_vars=env_vars)
        # Should have credential issues
        self.assertEqual(p.rc, 1)

        p = aws('s3 cp s3://%s/foo %s/ --region %s --no-sign-request' %
                (bucket_name, self.files.rootdir, self.region),
                env_vars=env_vars)
        # Should be able to download the file when not signing the request.
        self.assert_no_errors(p)


class TestHonorsEndpointUrl(BaseS3CLICommand):
    def test_verify_endpoint_url_is_used(self):
        # We're going to verify this indirectly by looking at the
        # debug logs.  The endpoint url we specify should be in the
        # debug logs, and the endpoint url that botocore would have
        # used if we didn't provide the endpoint-url should not
        # be in the debug logs.  The other alternative is to actually
        # watch what connections are made in the process, which is not
        # easy.
        p = aws('s3 ls s3://dnscompat/ '
                '--endpoint-url http://localhost:51515 '
                '--debug')
        debug_logs = p.stderr
        original_hostname = 'dnscompat.s3.amazonaws.com'
        expected = 'localhost'
        self.assertNotIn(original_hostname, debug_logs,
                         '--endpoint-url is being ignored in s3 commands.')
        self.assertIn(expected, debug_logs)


if __name__ == "__main__":
    unittest.main()
