Roll version of gsutil to 4.15.

third_party/gsutil/gslib/tests/test_cp.py

 # -*- coding: utf-8 -*-
 # Copyright 2013 Google Inc. 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.
 # You may obtain a copy of the License at
 #
 #     http://www.apache.org/licenses/LICENSE-2.0
 #
 # Unless required by applicable law or agreed to in writing, software
@@ skipping 15 matching lines @@
 import pkgutil
 import random
 import re
 import string
 import sys
 
 from apitools.base.py import exceptions as apitools_exceptions
 import boto
 from boto import storage_uri
 from boto.exception import ResumableTransferDisposition
-from boto.exception import ResumableUploadException
 from boto.exception import StorageResponseError
 from boto.storage_uri import BucketStorageUri
+import crcmod
 
 from gslib.cloud_api import ResumableDownloadException
 from gslib.cloud_api import ResumableUploadException
 from gslib.cloud_api import ResumableUploadStartOverException
+from gslib.commands.config import DEFAULT_SLICED_OBJECT_DOWNLOAD_THRESHOLD
 from gslib.copy_helper import GetTrackerFilePath
 from gslib.copy_helper import TrackerFileType
 from gslib.cs_api_map import ApiSelector
 from gslib.gcs_json_api import GcsJsonApi
 from gslib.hashing_helper import CalculateMd5FromContents
 from gslib.storage_url import StorageUrlFromString
 import gslib.tests.testcase as testcase
 from gslib.tests.testcase.base import NotParallelizable
 from gslib.tests.testcase.integration_testcase import SkipForS3
 from gslib.tests.util import GenerationFromURI as urigen
 from gslib.tests.util import HAS_S3_CREDS
 from gslib.tests.util import ObjectToURI as suri
-from gslib.tests.util import PerformsFileToObjectUpload
+from gslib.tests.util import SequentialAndParallelTransfer
 from gslib.tests.util import SetBotoConfigForTest
 from gslib.tests.util import unittest
 from gslib.third_party.storage_apitools import storage_v1_messages as apitools_messages
 from gslib.tracker_file import DeleteTrackerFile
 from gslib.tracker_file import GetRewriteTrackerFilePath
+from gslib.tracker_file import GetSlicedDownloadTrackerFilePaths
 from gslib.util import EIGHT_MIB
+from gslib.util import HumanReadableToBytes
 from gslib.util import IS_WINDOWS
 from gslib.util import MakeHumanReadable
 from gslib.util import ONE_KIB
 from gslib.util import ONE_MIB
 from gslib.util import Retry
 from gslib.util import START_CALLBACK_PER_BYTES
+from gslib.util import UsingCrcmodExtension
 from gslib.util import UTF8
 
 
 # Custom test callbacks must be pickleable, and therefore at global scope.
 class _HaltingCopyCallbackHandler(object):
   """Test callback handler for intentionally stopping a resumable transfer."""
 
   def __init__(self, is_upload, halt_at_byte):
     self._is_upload = is_upload
     self._halt_at_byte = halt_at_byte
@@ skipping 54 matching lines @@
           '%s/%s transferred.\r\n' % (
               self._startover_at_byte,
               MakeHumanReadable(total_bytes_transferred),
               MakeHumanReadable(total_size)))
       self.started_over_once = True
       raise boto.exception.ResumableUploadException(
           'Forcing upload start over',
           ResumableTransferDisposition.START_OVER)
 
 
+class _HaltOneComponentCopyCallbackHandler(object):
+  """Test callback handler for stopping part of a sliced download."""
+
+  def __init__(self, halt_at_byte):
+    self._last_progress_byte = None
+    self._halt_at_byte = halt_at_byte
+
+  # pylint: disable=invalid-name
+  # pylint: disable=unused-argument
+  def call(self, current_progress_byte, total_size_unused):
+    """Forcibly exits if the passed the halting point since the last call."""
+    if (self._last_progress_byte is not None and
+        self._last_progress_byte < self._halt_at_byte < current_progress_byte):
+      sys.stderr.write('Halting transfer.\r\n')
+      raise ResumableDownloadException('Artifically halting download.')
+    self._last_progress_byte = current_progress_byte
+
+
 class _DeleteBucketThenStartOverCopyCallbackHandler(object):
   """Test callback handler that deletes bucket then raises start-over."""
 
   def __init__(self, startover_at_byte, bucket_uri):
     self._startover_at_byte = startover_at_byte
     self._bucket_uri = bucket_uri
     self.started_over_once = False
 
   # pylint: disable=invalid-name
   def call(self, total_bytes_transferred, total_size):
@@ skipping 80 matching lines @@
   """Integration tests for cp command."""
 
   # For tests that artificially halt, we need to ensure at least one callback
   # occurs.
   halt_size = START_CALLBACK_PER_BYTES * 2
 
   def _get_test_file(self, name):
     contents = pkgutil.get_data('gslib', 'tests/test_data/%s' % name)
     return self.CreateTempFile(file_name=name, contents=contents)
 
-  @PerformsFileToObjectUpload
+  @SequentialAndParallelTransfer
   def test_noclobber(self):
     key_uri = self.CreateObject(contents='foo')
     fpath = self.CreateTempFile(contents='bar')
     stderr = self.RunGsUtil(['cp', '-n', fpath, suri(key_uri)],
                             return_stderr=True)
     self.assertIn('Skipping existing item: %s' % suri(key_uri), stderr)
     self.assertEqual(key_uri.get_contents_as_string(), 'foo')
     stderr = self.RunGsUtil(['cp', '-n', suri(key_uri), fpath],
                             return_stderr=True)
     with open(fpath, 'r') as f:
       self.assertIn('Skipping existing item: %s' % suri(f), stderr)
       self.assertEqual(f.read(), 'bar')
 
   def test_dest_bucket_not_exist(self):
     fpath = self.CreateTempFile(contents='foo')
     invalid_bucket_uri = (
         '%s://%s' % (self.default_provider, self.nonexistent_bucket_name))
     stderr = self.RunGsUtil(['cp', fpath, invalid_bucket_uri],
                             expected_status=1, return_stderr=True)
-    self.assertIn('does not exist.', stderr)
+    self.assertIn('does not exist', stderr)
 
   def test_copy_in_cloud_noclobber(self):
     bucket1_uri = self.CreateBucket()
     bucket2_uri = self.CreateBucket()
     key_uri = self.CreateObject(bucket_uri=bucket1_uri, contents='foo')
     stderr = self.RunGsUtil(['cp', suri(key_uri), suri(bucket2_uri)],
                             return_stderr=True)
     # Rewrite API may output an additional 'Copying' progress notification.
     self.assertGreaterEqual(stderr.count('Copying'), 1)
     self.assertLessEqual(stderr.count('Copying'), 2)
     stderr = self.RunGsUtil(['cp', '-n', suri(key_uri), suri(bucket2_uri)],
                             return_stderr=True)
     self.assertIn('Skipping existing item: %s' %
                   suri(bucket2_uri, key_uri.object_name), stderr)
 
-  @PerformsFileToObjectUpload
+  @SequentialAndParallelTransfer
   def test_streaming(self):
     bucket_uri = self.CreateBucket()
     stderr = self.RunGsUtil(['cp', '-', '%s' % suri(bucket_uri, 'foo')],
                             stdin='bar', return_stderr=True)
     self.assertIn('Copying from <STDIN>', stderr)
     key_uri = bucket_uri.clone_replace_name('foo')
     self.assertEqual(key_uri.get_contents_as_string(), 'bar')
 
   def test_streaming_multiple_arguments(self):
     bucket_uri = self.CreateBucket()
     stderr = self.RunGsUtil(['cp', '-', '-', suri(bucket_uri)],
                             stdin='bar', return_stderr=True, expected_status=1)
     self.assertIn('Multiple URL strings are not supported with streaming',
                   stderr)
 
   # TODO: Implement a way to test both with and without using magic file.
 
-  @PerformsFileToObjectUpload
+  @SequentialAndParallelTransfer
   def test_detect_content_type(self):
     """Tests local detection of content type."""
     bucket_uri = self.CreateBucket()
     dsturi = suri(bucket_uri, 'foo')
 
     self.RunGsUtil(['cp', self._get_test_file('test.mp3'), dsturi])
 
     # Use @Retry as hedge against bucket listing eventual consistency.
     @Retry(AssertionError, tries=3, timeout_secs=1)
     def _Check1():
@@ skipping 61 matching lines @@
                     self._get_test_file('test.gif'), dsturi])
 
     # Use @Retry as hedge against bucket listing eventual consistency.
     @Retry(AssertionError, tries=3, timeout_secs=1)
     def _Check2():
       stdout = self.RunGsUtil(['ls', '-L', dsturi], return_stdout=True)
       self.assertRegexpMatches(stdout, r'Content-Type:\s+text/plain')
     _Check2()
 
   @unittest.skipIf(IS_WINDOWS, 'magicfile is not available on Windows.')
-  @PerformsFileToObjectUpload
+  @SequentialAndParallelTransfer
   def test_magicfile_override(self):
     """Tests content type override with magicfile value."""
     bucket_uri = self.CreateBucket()
     dsturi = suri(bucket_uri, 'foo')
     fpath = self.CreateTempFile(contents='foo/bar\n')
     self.RunGsUtil(['cp', fpath, dsturi])
 
     # Use @Retry as hedge against bucket listing eventual consistency.
     @Retry(AssertionError, tries=3, timeout_secs=1)
     def _Check1():
       stdout = self.RunGsUtil(['ls', '-L', dsturi], return_stdout=True)
       use_magicfile = boto.config.getbool('GSUtil', 'use_magicfile', False)
       content_type = ('text/plain' if use_magicfile
                       else 'application/octet-stream')
       self.assertRegexpMatches(stdout, r'Content-Type:\s+%s' % content_type)
     _Check1()
 
-  @PerformsFileToObjectUpload
+  @SequentialAndParallelTransfer
   def test_content_type_mismatches(self):
     """Tests overriding content type when it does not match the file type."""
     bucket_uri = self.CreateBucket()
     dsturi = suri(bucket_uri, 'foo')
     fpath = self.CreateTempFile(contents='foo/bar\n')
 
     self.RunGsUtil(['-h', 'Content-Type:image/gif', 'cp',
                     self._get_test_file('test.mp3'), dsturi])
 
     # Use @Retry as hedge against bucket listing eventual consistency.
@@ skipping 15 matching lines @@
 
     self.RunGsUtil(['-h', 'Content-Type:image/gif', 'cp', fpath, dsturi])
 
     # Use @Retry as hedge against bucket listing eventual consistency.
     @Retry(AssertionError, tries=3, timeout_secs=1)
     def _Check3():
       stdout = self.RunGsUtil(['ls', '-L', dsturi], return_stdout=True)
       self.assertRegexpMatches(stdout, r'Content-Type:\s+image/gif')
     _Check3()
 
-  @PerformsFileToObjectUpload
+  @SequentialAndParallelTransfer
   def test_content_type_header_case_insensitive(self):
     """Tests that content type header is treated with case insensitivity."""
     bucket_uri = self.CreateBucket()
     dsturi = suri(bucket_uri, 'foo')
     fpath = self._get_test_file('test.gif')
 
     self.RunGsUtil(['-h', 'content-Type:text/plain', 'cp',
                     fpath, dsturi])
 
     # Use @Retry as hedge against bucket listing eventual consistency.
     @Retry(AssertionError, tries=3, timeout_secs=1)
     def _Check1():
       stdout = self.RunGsUtil(['ls', '-L', dsturi], return_stdout=True)
       self.assertRegexpMatches(stdout, r'Content-Type:\s+text/plain')
       self.assertNotRegexpMatches(stdout, r'image/gif')
     _Check1()
 
     self.RunGsUtil(['-h', 'CONTENT-TYPE:image/gif',
                     '-h', 'content-type:image/gif',
                     'cp', fpath, dsturi])
 
     # Use @Retry as hedge against bucket listing eventual consistency.
     @Retry(AssertionError, tries=3, timeout_secs=1)
     def _Check2():
       stdout = self.RunGsUtil(['ls', '-L', dsturi], return_stdout=True)
       self.assertRegexpMatches(stdout, r'Content-Type:\s+image/gif')
       self.assertNotRegexpMatches(stdout, r'image/gif,\s*image/gif')
     _Check2()
 
-  @PerformsFileToObjectUpload
+  @SequentialAndParallelTransfer
   def test_other_headers(self):
     """Tests that non-content-type headers are applied successfully on copy."""
     bucket_uri = self.CreateBucket()
     dst_uri = suri(bucket_uri, 'foo')
     fpath = self._get_test_file('test.gif')
 
     self.RunGsUtil(['-h', 'Cache-Control:public,max-age=12',
                     '-h', 'x-%s-meta-1:abcd' % self.provider_custom_meta, 'cp',
                     fpath, dst_uri])
 
     stdout = self.RunGsUtil(['ls', '-L', dst_uri], return_stdout=True)
     self.assertRegexpMatches(stdout, r'Cache-Control\s*:\s*public,max-age=12')
     self.assertRegexpMatches(stdout, r'Metadata:\s*1:\s*abcd')
 
     dst_uri2 = suri(bucket_uri, 'bar')
     self.RunGsUtil(['cp', dst_uri, dst_uri2])
     # Ensure metadata was preserved across copy.
     stdout = self.RunGsUtil(['ls', '-L', dst_uri2], return_stdout=True)
     self.assertRegexpMatches(stdout, r'Cache-Control\s*:\s*public,max-age=12')
     self.assertRegexpMatches(stdout, r'Metadata:\s*1:\s*abcd')
 
-  @PerformsFileToObjectUpload
+  @SequentialAndParallelTransfer
   def test_versioning(self):
     """Tests copy with versioning."""
     bucket_uri = self.CreateVersionedBucket()
     k1_uri = self.CreateObject(bucket_uri=bucket_uri, contents='data2')
     k2_uri = self.CreateObject(bucket_uri=bucket_uri, contents='data1')
     g1 = urigen(k2_uri)
     self.RunGsUtil(['cp', suri(k1_uri), suri(k2_uri)])
     k2_uri = bucket_uri.clone_replace_name(k2_uri.object_name)
     k2_uri = bucket_uri.clone_replace_key(k2_uri.get_key())
     g2 = urigen(k2_uri)
@@ skipping 22 matching lines @@
                     k2_uri.versionless_uri])
     self.RunGsUtil(['cp', k2_uri.versionless_uri, fpath])
     with open(fpath, 'r') as f:
       self.assertEqual(f.read(), 'data1')
 
     # Attempt to specify a version-specific URI for destination.
     stderr = self.RunGsUtil(['cp', fpath, k2_uri.uri], return_stderr=True,
                             expected_status=1)
     self.assertIn('cannot be the destination for gsutil cp', stderr)
 
+  def test_versioning_no_parallelism(self):
+    """Tests that copy all-versions errors when parallelism is enabled."""
+    stderr = self.RunGsUtil(
+        ['-m', 'cp', '-A', suri(self.nonexistent_bucket_name, 'foo'),
+         suri(self.nonexistent_bucket_name, 'bar')],
+        expected_status=1, return_stderr=True)
+    self.assertIn('-m option is not supported with the cp -A flag', stderr)
+
   @SkipForS3('S3 lists versioned objects in reverse timestamp order.')
   def test_recursive_copying_versioned_bucket(self):
-    """Tests that cp -R with versioned buckets copies all versions in order."""
+    """Tests cp -R with versioned buckets."""
     bucket1_uri = self.CreateVersionedBucket()
     bucket2_uri = self.CreateVersionedBucket()
+    bucket3_uri = self.CreateVersionedBucket()
 
     # Write two versions of an object to the bucket1.
     self.CreateObject(bucket_uri=bucket1_uri, object_name='k', contents='data0')
     self.CreateObject(bucket_uri=bucket1_uri, object_name='k',
                       contents='longer_data1')
 
     self.AssertNObjectsInBucket(bucket1_uri, 2, versioned=True)
     self.AssertNObjectsInBucket(bucket2_uri, 0, versioned=True)
+    self.AssertNObjectsInBucket(bucket3_uri, 0, versioned=True)
 
     # Recursively copy to second versioned bucket.
-    self.RunGsUtil(['cp', '-R', suri(bucket1_uri, '*'), suri(bucket2_uri)])
+    # -A flag should copy all versions in order.
+    self.RunGsUtil(['cp', '-R', '-A', suri(bucket1_uri, '*'),
+                    suri(bucket2_uri)])
 
     # Use @Retry as hedge against bucket listing eventual consistency.
     @Retry(AssertionError, tries=3, timeout_secs=1)
     def _Check2():
       """Validates the results of the cp -R."""
       listing1 = self.RunGsUtil(['ls', '-la', suri(bucket1_uri)],
                                 return_stdout=True).split('\n')
       listing2 = self.RunGsUtil(['ls', '-la', suri(bucket2_uri)],
                                 return_stdout=True).split('\n')
       # 2 lines of listing output, 1 summary line, 1 empty line from \n split.
@@ skipping 10 matching lines @@
 
       # Similarly for second object in each bucket.
       size1, _, uri_str1, _ = listing1[1].split()
       self.assertEquals(size1, str(len('longer_data1')))
       self.assertEquals(storage_uri(uri_str1).object_name, 'k')
       size2, _, uri_str2, _ = listing2[1].split()
       self.assertEquals(size2, str(len('longer_data1')))
       self.assertEquals(storage_uri(uri_str2).object_name, 'k')
     _Check2()
 
-  @PerformsFileToObjectUpload
+    # Recursively copy to second versioned bucket with no -A flag.
+    # This should copy only the live object.
+    self.RunGsUtil(['cp', '-R', suri(bucket1_uri, '*'),
+                    suri(bucket3_uri)])
+
+    # Use @Retry as hedge against bucket listing eventual consistency.
+    @Retry(AssertionError, tries=3, timeout_secs=1)
+    def _Check3():
+      """Validates the results of the cp -R."""
+      listing1 = self.RunGsUtil(['ls', '-la', suri(bucket1_uri)],
+                                return_stdout=True).split('\n')
+      listing2 = self.RunGsUtil(['ls', '-la', suri(bucket3_uri)],
+                                return_stdout=True).split('\n')
+      # 2 lines of listing output, 1 summary line, 1 empty line from \n split.
+      self.assertEquals(len(listing1), 4)
+      # 1 lines of listing output, 1 summary line, 1 empty line from \n split.
+      self.assertEquals(len(listing2), 3)
+
+      # Live (second) object in bucket 1 should match the single live object.
+      size1, _, uri_str1, _ = listing2[0].split()
+      self.assertEquals(size1, str(len('longer_data1')))
+      self.assertEquals(storage_uri(uri_str1).object_name, 'k')
+    _Check3()
+
+  @SequentialAndParallelTransfer
   @SkipForS3('Preconditions not supported for S3.')
   def test_cp_generation_zero_match(self):
     """Tests that cp handles an object-not-exists precondition header."""
     bucket_uri = self.CreateBucket()
     fpath1 = self.CreateTempFile(contents='data1')
     # Match 0 means only write the object if it doesn't already exist.
     gen_match_header = 'x-goog-if-generation-match:0'
 
     # First copy should succeed.
     # TODO: This can fail (rarely) if the server returns a 5xx but actually
     # commits the bytes. If we add restarts on small uploads, handle this
     # case.
     self.RunGsUtil(['-h', gen_match_header, 'cp', fpath1, suri(bucket_uri)])
 
     # Second copy should fail with a precondition error.
     stderr = self.RunGsUtil(['-h', gen_match_header, 'cp', fpath1,
                              suri(bucket_uri)],
                             return_stderr=True, expected_status=1)
     self.assertIn('PreconditionException', stderr)
 
-  @PerformsFileToObjectUpload
+  @SequentialAndParallelTransfer
   @SkipForS3('Preconditions not supported for S3.')
   def test_cp_v_generation_match(self):
     """Tests that cp -v option handles the if-generation-match header."""
     bucket_uri = self.CreateVersionedBucket()
     k1_uri = self.CreateObject(bucket_uri=bucket_uri, contents='data1')
     g1 = k1_uri.generation
 
     tmpdir = self.CreateTempDir()
     fpath1 = self.CreateTempFile(tmpdir=tmpdir, contents='data2')
 
@@ skipping 11 matching lines @@
     # Specifiying a generation with -n should fail before the request hits the
     # server.
     stderr = self.RunGsUtil(['-h', gen_match_header, 'cp', '-n', fpath1,
                              suri(k1_uri)],
                             return_stderr=True, expected_status=1)
 
     self.assertIn('ArgumentException', stderr)
     self.assertIn('Specifying x-goog-if-generation-match is not supported '
                   'with cp -n', stderr)
 
-  @PerformsFileToObjectUpload
+  @SequentialAndParallelTransfer
   def test_cp_nv(self):
     """Tests that cp -nv works when skipping existing file."""
     bucket_uri = self.CreateVersionedBucket()
     k1_uri = self.CreateObject(bucket_uri=bucket_uri, contents='data1')
 
     tmpdir = self.CreateTempDir()
     fpath1 = self.CreateTempFile(tmpdir=tmpdir, contents='data2')
 
     # First copy should succeed.
     self.RunGsUtil(['cp', '-nv', fpath1, suri(k1_uri)])
 
     # Second copy should skip copying.
     stderr = self.RunGsUtil(['cp', '-nv', fpath1, suri(k1_uri)],
                             return_stderr=True)
     self.assertIn('Skipping existing item:', stderr)
 
-  @PerformsFileToObjectUpload
+  @SequentialAndParallelTransfer
   @SkipForS3('S3 lists versioned objects in reverse timestamp order.')
   def test_cp_v_option(self):
     """"Tests that cp -v returns the created object's version-specific URI."""
     bucket_uri = self.CreateVersionedBucket()
     k1_uri = self.CreateObject(bucket_uri=bucket_uri, contents='data1')
     k2_uri = self.CreateObject(bucket_uri=bucket_uri, contents='data2')
 
     # Case 1: Upload file to object using one-shot PUT.
     tmpdir = self.CreateTempDir()
     fpath1 = self.CreateTempFile(tmpdir=tmpdir, contents='data1')
@@ skipping 38 matching lines @@
     @Retry(AssertionError, tries=3, timeout_secs=1)
     def _Check1():
       stdout = self.RunGsUtil(['ls', '-a', dst_str], return_stdout=True)
       lines = stdout.split('\n')
       # Final (most recent) object should match the "Created:" URI. This is
       # in second-to-last line (last line is '\n').
       self.assertGreater(len(lines), 2)
       self.assertEqual(created_uri, lines[-2])
     _Check1()
 
-  @PerformsFileToObjectUpload
+  @SequentialAndParallelTransfer
   def test_stdin_args(self):
     """Tests cp with the -I option."""
     tmpdir = self.CreateTempDir()
     fpath1 = self.CreateTempFile(tmpdir=tmpdir, contents='data1')
     fpath2 = self.CreateTempFile(tmpdir=tmpdir, contents='data2')
     bucket_uri = self.CreateBucket()
     self.RunGsUtil(['cp', '-I', suri(bucket_uri)],
                    stdin='\n'.join((fpath1, fpath2)))
 
     # Use @Retry as hedge against bucket listing eventual consistency.
@@ skipping 124 matching lines @@
     public_read_acl = self.RunGsUtil(['acl', 'get', suri(key_uri)],
                                      return_stdout=True)
 
     @Retry(AssertionError, tries=3, timeout_secs=1)
     def _Check():
       uri = suri(bucket2_uri, key_uri.object_name)
       new_acl_json = self.RunGsUtil(['acl', 'get', uri], return_stdout=True)
       self.assertEqual(public_read_acl, new_acl_json)
     _Check()
 
-  @PerformsFileToObjectUpload
+  @SequentialAndParallelTransfer
   def test_canned_acl_upload(self):
     """Tests uploading a file with a canned ACL."""
     bucket1_uri = self.CreateBucket()
     key_uri = self.CreateObject(bucket_uri=bucket1_uri, contents='foo')
     # Set public-read on the object so we can compare the ACLs.
     self.RunGsUtil(['acl', 'set', 'public-read', suri(key_uri)])
     public_read_acl = self.RunGsUtil(['acl', 'get', suri(key_uri)],
                                      return_stdout=True)
 
     file_name = 'bar'
@@ skipping 24 matching lines @@
     key_uri = self.CreateObject(bucket_uri=bucket_uri, contents=contents)
     stdout = self.RunGsUtil(['cp', suri(key_uri), '-'], return_stdout=True)
     self.assertIn(contents, stdout)
 
   def test_cp_local_file_to_local_stream(self):
     contents = 'content'
     fpath = self.CreateTempFile(contents=contents)
     stdout = self.RunGsUtil(['cp', fpath, '-'], return_stdout=True)
     self.assertIn(contents, stdout)
 
-  @PerformsFileToObjectUpload
+  @SequentialAndParallelTransfer
   def test_cp_zero_byte_file(self):
     dst_bucket_uri = self.CreateBucket()
     src_dir = self.CreateTempDir()
     fpath = os.path.join(src_dir, 'zero_byte')
     with open(fpath, 'w') as unused_out_file:
       pass  # Write a zero byte file
     self.RunGsUtil(['cp', fpath, suri(dst_bucket_uri)])
 
     @Retry(AssertionError, tries=3, timeout_secs=1)
     def _Check1():
       stdout = self.RunGsUtil(['ls', suri(dst_bucket_uri)], return_stdout=True)
       self.assertIn(os.path.basename(fpath), stdout)
     _Check1()
 
     download_path = os.path.join(src_dir, 'zero_byte_download')
     self.RunGsUtil(['cp', suri(dst_bucket_uri, 'zero_byte'), download_path])
     self.assertTrue(os.stat(download_path))
 
   def test_copy_bucket_to_bucket(self):
-    """Tests that recursively copying from bucket to bucket.
+    """Tests recursively copying from bucket to bucket.
 
     This should produce identically named objects (and not, in particular,
     destination objects named by the version-specific URI from source objects).
     """
     src_bucket_uri = self.CreateVersionedBucket()
     dst_bucket_uri = self.CreateVersionedBucket()
     self.CreateObject(bucket_uri=src_bucket_uri, object_name='obj0',
                       contents='abc')
     self.CreateObject(bucket_uri=src_bucket_uri, object_name='obj1',
                       contents='def')
@@ skipping 48 matching lines @@
     self.CreateObject(bucket_uri=src_bucket_uri, object_name='obj0',
                       contents='abc')
     self.CreateObject(bucket_uri=src_bucket_uri, object_name='obj1',
                       contents='def')
 
     # Create a placeholder like what can be left over by web GUI tools.
     key_uri = src_bucket_uri.clone_replace_name('/')
     key_uri.set_contents_from_string('')
     self.AssertNObjectsInBucket(src_bucket_uri, 3)
 
-    stderr = self.RunGsUtil(['cp', '-R', suri(src_bucket_uri), dst_dir],
-                            return_stderr=True)
-    self.assertIn('Skipping cloud sub-directory placeholder object', stderr)
+    self.RunGsUtil(['cp', '-R', suri(src_bucket_uri), dst_dir])
     dir_list = []
     for dirname, _, filenames in os.walk(dst_dir):
       for filename in filenames:
         dir_list.append(os.path.join(dirname, filename))
     dir_list = sorted(dir_list)
     self.assertEqual(len(dir_list), 2)
     self.assertEqual(os.path.join(dst_dir, src_bucket_uri.bucket_name,
                                   'obj0'), dir_list[0])
     self.assertEqual(os.path.join(dst_dir, src_bucket_uri.bucket_name,
                                   'obj1'), dir_list[1])
@@ skipping 24 matching lines @@
     @Retry(AssertionError, tries=3, timeout_secs=1)
     def _Check1():
       stdout = self.RunGsUtil(['ls', '-L', suri(bucket_uri)],
                               return_stdout=True)
       self.assertRegexpMatches(stdout,
                                r'Hash\s+\(md5\):\s+%s' % re.escape(file_md5))
     _Check1()
 
   @unittest.skipIf(IS_WINDOWS,
                    'Unicode handling on Windows requires mods to site-packages')
-  @PerformsFileToObjectUpload
+  @SequentialAndParallelTransfer
   def test_cp_manifest_upload_unicode(self):
     return self._ManifestUpload('foo-unicöde', 'bar-unicöde',
                                 'manifest-unicöde')
 
-  @PerformsFileToObjectUpload
+  @SequentialAndParallelTransfer
   def test_cp_manifest_upload(self):
     """Tests uploading with a mnifest file."""
     return self._ManifestUpload('foo', 'bar', 'manifest')
 
   def _ManifestUpload(self, file_name, object_name, manifest_name):
     """Tests uploading with a manifest file."""
     bucket_uri = self.CreateBucket()
     dsturi = suri(bucket_uri, object_name)
 
     fpath = self.CreateTempFile(file_name=file_name, contents='bar')
@@ skipping 20 matching lines @@
     if self.RunGsUtil == testcase.GsUtilIntegrationTestCase.RunGsUtil:
       # Check that we didn't do automatic parallel uploads - compose doesn't
       # calculate the MD5 hash. Since RunGsUtil is overriden in
       # TestCpParallelUploads to force parallel uploads, we can check which
       # method was used.
       self.assertEqual(results[4], 'rL0Y20zC+Fzt72VPzMSk2A==')  # md5
     self.assertEqual(int(results[6]), 3)  # Source Size
     self.assertEqual(int(results[7]), 3)  # Bytes Transferred
     self.assertEqual(results[8], 'OK')  # Result
 
-  @PerformsFileToObjectUpload
+  @SequentialAndParallelTransfer
   def test_cp_manifest_download(self):
     """Tests downloading with a manifest file."""
     key_uri = self.CreateObject(contents='foo')
     fpath = self.CreateTempFile(contents='')
     logpath = self.CreateTempFile(contents='')
     # Ensure the file is empty.
     open(logpath, 'w').close()
     self.RunGsUtil(['cp', '-L', logpath, suri(key_uri), fpath],
                    return_stdout=True)
     with open(logpath, 'r') as f:
       lines = f.readlines()
     self.assertEqual(len(lines), 2)
 
     expected_headers = ['Source', 'Destination', 'Start', 'End', 'Md5',
                         'UploadId', 'Source Size', 'Bytes Transferred',
                         'Result', 'Description']
     self.assertEqual(expected_headers, lines[0].strip().split(','))
     results = lines[1].strip().split(',')
     self.assertEqual(results[0][:5], '%s://' %
                      self.default_provider)      # source
     self.assertEqual(results[1][:7], 'file://')  # destination
     date_format = '%Y-%m-%dT%H:%M:%S.%fZ'
     start_date = datetime.datetime.strptime(results[2], date_format)
     end_date = datetime.datetime.strptime(results[3], date_format)
     self.assertEqual(end_date > start_date, True)
-    self.assertEqual(results[4], 'rL0Y20zC+Fzt72VPzMSk2A==')  # md5
     self.assertEqual(int(results[6]), 3)  # Source Size
     # Bytes transferred might be more than 3 if the file was gzipped, since
     # the minimum gzip header is 10 bytes.
     self.assertGreaterEqual(int(results[7]), 3)  # Bytes Transferred
     self.assertEqual(results[8], 'OK')  # Result
 
-  @PerformsFileToObjectUpload
+  @SequentialAndParallelTransfer
   def test_copy_unicode_non_ascii_filename(self):
     key_uri = self.CreateObject(contents='foo')
     # Make file large enough to cause a resumable upload (which hashes filename
     # to construct tracker filename).
     fpath = self.CreateTempFile(file_name=u'Аудиоархив',
                                 contents='x' * 3 * 1024 * 1024)
     fpath_bytes = fpath.encode(UTF8)
     stderr = self.RunGsUtil(['cp', fpath_bytes, suri(key_uri)],
                             return_stderr=True)
     self.assertIn('Copying file:', stderr)
 
   # Note: We originally one time implemented a test
   # (test_copy_invalid_unicode_filename) that invalid unicode filenames were
   # skipped, but it turns out os.walk() on MacOS doesn't have problems with
   # such files (so, failed that test). Given that, we decided to remove the
   # test.
 
+  @SequentialAndParallelTransfer
   def test_gzip_upload_and_download(self):
     bucket_uri = self.CreateBucket()
     contents = 'x' * 10000
     tmpdir = self.CreateTempDir()
     self.CreateTempFile(file_name='test.html', tmpdir=tmpdir, contents=contents)
     self.CreateTempFile(file_name='test.js', tmpdir=tmpdir, contents=contents)
     self.CreateTempFile(file_name='test.txt', tmpdir=tmpdir, contents=contents)
     # Test that copying specifying only 2 of the 3 prefixes gzips the correct
     # files, and test that including whitespace in the extension list works.
     self.RunGsUtil(['cp', '-z', 'js, html',
@@ skipping 16 matching lines @@
 
   def test_upload_with_subdir_and_unexpanded_wildcard(self):
     fpath1 = self.CreateTempFile(file_name=('tmp', 'x', 'y', 'z'))
     bucket_uri = self.CreateBucket()
     wildcard_uri = '%s*' % fpath1[:-5]
     stderr = self.RunGsUtil(['cp', '-R', wildcard_uri, suri(bucket_uri)],
                             return_stderr=True)
     self.assertIn('Copying file:', stderr)
     self.AssertNObjectsInBucket(bucket_uri, 1)
 
+  @SequentialAndParallelTransfer
   def test_cp_object_ending_with_slash(self):
     """Tests that cp works with object names ending with slash."""
     tmpdir = self.CreateTempDir()
     bucket_uri = self.CreateBucket()
     self.CreateObject(bucket_uri=bucket_uri,
                       object_name='abc/',
                       contents='dir')
     self.CreateObject(bucket_uri=bucket_uri,
                       object_name='abc/def',
                       contents='def')
@@ skipping 36 matching lines @@
 
   def test_cp_multithreaded_wildcard(self):
     """Tests that cp -m works with a wildcard."""
     num_test_files = 5
     tmp_dir = self.CreateTempDir(test_files=num_test_files)
     bucket_uri = self.CreateBucket()
     wildcard_uri = '%s%s*' % (tmp_dir, os.sep)
     self.RunGsUtil(['-m', 'cp', wildcard_uri, suri(bucket_uri)])
     self.AssertNObjectsInBucket(bucket_uri, num_test_files)
 
+  @SequentialAndParallelTransfer
   def test_cp_duplicate_source_args(self):
     """Tests that cp -m works when a source argument is provided twice."""
     object_contents = 'edge'
     object_uri = self.CreateObject(object_name='foo', contents=object_contents)
     tmp_dir = self.CreateTempDir()
     self.RunGsUtil(['-m', 'cp', suri(object_uri), suri(object_uri), tmp_dir])
     with open(os.path.join(tmp_dir, 'foo'), 'r') as in_fp:
       contents = in_fp.read()
       # Contents should be not duplicated.
       self.assertEqual(contents, object_contents)
@@ skipping 182 matching lines @@
                                   contents='a' * ONE_KIB)
       stderr = self.RunGsUtil(['cp', fpath, suri(bucket_uri)],
                               expected_status=1, return_stderr=True)
       self.assertIn('ResumableUploadAbortException', stderr)
 
   # This temporarily changes the tracker directory to unwritable which
   # interferes with any parallel running tests that use the tracker directory.
   @NotParallelizable
   @SkipForS3('No resumable upload support for S3.')
   @unittest.skipIf(IS_WINDOWS, 'chmod on dir unsupported on Windows.')
-  @PerformsFileToObjectUpload
+  @SequentialAndParallelTransfer
   def test_cp_unwritable_tracker_file(self):
     """Tests a resumable upload with an unwritable tracker file."""
     bucket_uri = self.CreateBucket()
     tracker_filename = GetTrackerFilePath(
         StorageUrlFromString(suri(bucket_uri, 'foo')),
         TrackerFileType.UPLOAD, self.test_api)
     tracker_dir = os.path.dirname(tracker_filename)
     fpath = self.CreateTempFile(file_name='foo', contents='a' * ONE_KIB)
     boto_config_for_test = ('GSUtil', 'resumable_threshold', str(ONE_KIB))
     save_mod = os.stat(tracker_dir).st_mode
 
     try:
       os.chmod(tracker_dir, 0)
       with SetBotoConfigForTest([boto_config_for_test]):
         stderr = self.RunGsUtil(['cp', fpath, suri(bucket_uri)],
                                 expected_status=1, return_stderr=True)
         self.assertIn('Couldn\'t write tracker file', stderr)
     finally:
       os.chmod(tracker_dir, save_mod)
       if os.path.exists(tracker_filename):
         os.unlink(tracker_filename)
 
   # This temporarily changes the tracker directory to unwritable which
   # interferes with any parallel running tests that use the tracker directory.
   @NotParallelizable
   @unittest.skipIf(IS_WINDOWS, 'chmod on dir unsupported on Windows.')
+  @SequentialAndParallelTransfer
   def test_cp_unwritable_tracker_file_download(self):
     """Tests downloads with an unwritable tracker file."""
     object_uri = self.CreateObject(contents='foo' * ONE_KIB)
     tracker_filename = GetTrackerFilePath(
         StorageUrlFromString(suri(object_uri)),
         TrackerFileType.DOWNLOAD, self.test_api)
     tracker_dir = os.path.dirname(tracker_filename)
     fpath = self.CreateTempFile()
     save_mod = os.stat(tracker_dir).st_mode
 
@@ skipping 28 matching lines @@
                                suri(object_uri), fpath],
                               expected_status=1, return_stderr=True)
       self.assertIn('Artifically halting download.', stderr)
       tracker_filename = GetTrackerFilePath(
           StorageUrlFromString(fpath), TrackerFileType.DOWNLOAD, self.test_api)
       self.assertTrue(os.path.isfile(tracker_filename))
       stderr = self.RunGsUtil(['cp', suri(object_uri), fpath],
                               return_stderr=True)
       self.assertIn('Resuming download', stderr)
 
+  @SequentialAndParallelTransfer
   def test_cp_resumable_download_etag_differs(self):
     """Tests that download restarts the file when the source object changes.
 
     This causes the etag not to match.
     """
     bucket_uri = self.CreateBucket()
     object_uri = self.CreateObject(bucket_uri=bucket_uri, object_name='foo',
-                                   contents='a' * self.halt_size)
+                                   contents='abc' * self.halt_size)
     fpath = self.CreateTempFile()
     test_callback_file = self.CreateTempFile(
         contents=pickle.dumps(_HaltingCopyCallbackHandler(False, 5)))
     boto_config_for_test = ('GSUtil', 'resumable_threshold', str(ONE_KIB))
     with SetBotoConfigForTest([boto_config_for_test]):
       # This will create a tracker file with an ETag.
       stderr = self.RunGsUtil(['cp', '--testcallbackfile', test_callback_file,
                                suri(object_uri), fpath],
                               expected_status=1, return_stderr=True)
       self.assertIn('Artifically halting download.', stderr)
       # Create a new object with different contents - it should have a
       # different ETag since the content has changed.
       object_uri = self.CreateObject(bucket_uri=bucket_uri, object_name='foo',
                                      contents='b' * self.halt_size)
       stderr = self.RunGsUtil(['cp', suri(object_uri), fpath],
                               return_stderr=True)
       self.assertNotIn('Resuming download', stderr)
 
+  # TODO: Enable this test for sequential downloads when their tracker files are
+  # modified to contain the source object generation.
+  @unittest.skipUnless(UsingCrcmodExtension(crcmod),
+                       'Sliced download requires fast crcmod.')
+  @SkipForS3('No sliced download support for S3.')
+  def test_cp_resumable_download_generation_differs(self):
+    """Tests that a resumable download restarts if the generation differs."""
+    bucket_uri = self.CreateBucket()
+    file_contents = 'abcd' * self.halt_size
+    object_uri = self.CreateObject(bucket_uri=bucket_uri, object_name='foo',
+                                   contents=file_contents)
+    fpath = self.CreateTempFile()
+
+    test_callback_file = self.CreateTempFile(
+        contents=pickle.dumps(_HaltingCopyCallbackHandler(False, 5)))
+
+    boto_config_for_test = [
+        ('GSUtil', 'resumable_threshold', str(self.halt_size)),
+        ('GSUtil', 'sliced_object_download_threshold', str(self.halt_size)),
+        ('GSUtil', 'sliced_object_download_max_components', '3')]
+
+    with SetBotoConfigForTest(boto_config_for_test):
+      stderr = self.RunGsUtil(['cp', '--testcallbackfile', test_callback_file,
+                               suri(object_uri), suri(fpath)],
+                              return_stderr=True, expected_status=1)
+      self.assertIn('Artifically halting download.', stderr)
+
+      # Overwrite the object with an identical object, increasing
+      # the generation but leaving other metadata the same.
+      identical_file = self.CreateTempFile(contents=file_contents)
+      self.RunGsUtil(['cp', suri(identical_file), suri(object_uri)])
+
+      stderr = self.RunGsUtil(['cp', suri(object_uri), suri(fpath)],
+                              return_stderr=True)
+      self.assertIn('Restarting download from scratch', stderr)
+      with open(fpath, 'r') as f:
+        self.assertEqual(f.read(), file_contents, 'File contents differ')
+
   def test_cp_resumable_download_file_larger(self):
     """Tests download deletes the tracker file when existing file is larger."""
     bucket_uri = self.CreateBucket()
     fpath = self.CreateTempFile()
     object_uri = self.CreateObject(bucket_uri=bucket_uri, object_name='foo',
                                    contents='a' * self.halt_size)
     test_callback_file = self.CreateTempFile(
         contents=pickle.dumps(_HaltingCopyCallbackHandler(False, 5)))
     boto_config_for_test = ('GSUtil', 'resumable_threshold', str(ONE_KIB))
     with SetBotoConfigForTest([boto_config_for_test]):
       stderr = self.RunGsUtil(['cp', '--testcallbackfile', test_callback_file,
                                suri(object_uri), fpath],
                               expected_status=1, return_stderr=True)
       self.assertIn('Artifically halting download.', stderr)
-      with open(fpath, 'w') as larger_file:
+      with open(fpath + '_.gstmp', 'w') as larger_file:
         for _ in range(self.halt_size * 2):
           larger_file.write('a')
       stderr = self.RunGsUtil(['cp', suri(object_uri), fpath],
                               expected_status=1, return_stderr=True)
       self.assertNotIn('Resuming download', stderr)
-      self.assertIn('is larger', stderr)
       self.assertIn('Deleting tracker file', stderr)
 
   def test_cp_resumable_download_content_differs(self):
     """Tests that we do not re-download when tracker file matches existing file.
 
     We only compare size, not contents, so re-download should not occur even
     though the contents are technically different. However, hash validation on
     the file should still occur and we will delete the file then because
     the hashes differ.
     """
     bucket_uri = self.CreateBucket()
     tmp_dir = self.CreateTempDir()
-    fpath = self.CreateTempFile(tmpdir=tmp_dir, contents='abcd' * ONE_KIB)
+    fpath = self.CreateTempFile(tmpdir=tmp_dir)
+    temp_download_file = fpath + '_.gstmp'
+    with open(temp_download_file, 'w') as fp:
+      fp.write('abcd' * ONE_KIB)
+
     object_uri = self.CreateObject(bucket_uri=bucket_uri, object_name='foo',
                                    contents='efgh' * ONE_KIB)
     stdout = self.RunGsUtil(['ls', '-L', suri(object_uri)], return_stdout=True)
     etag_match = re.search(r'\s*ETag:\s*(.*)', stdout)
     self.assertIsNotNone(etag_match, 'Could not get object ETag')
     self.assertEqual(len(etag_match.groups()), 1,
                      'Did not match expected single ETag')
     etag = etag_match.group(1)
 
     tracker_filename = GetTrackerFilePath(
         StorageUrlFromString(fpath), TrackerFileType.DOWNLOAD, self.test_api)
     try:
       with open(tracker_filename, 'w') as tracker_fp:
         tracker_fp.write(etag)
       boto_config_for_test = ('GSUtil', 'resumable_threshold', str(ONE_KIB))
       with SetBotoConfigForTest([boto_config_for_test]):
         stderr = self.RunGsUtil(['cp', suri(object_uri), fpath],
                                 return_stderr=True, expected_status=1)
-        self.assertIn('Download already complete for file', stderr)
+        self.assertIn('Download already complete', stderr)
         self.assertIn('doesn\'t match cloud-supplied digest', stderr)
         # File and tracker file should be deleted.
+        self.assertFalse(os.path.isfile(temp_download_file))
+        self.assertFalse(os.path.isfile(tracker_filename))
+        # Permanent file should not have been created.
         self.assertFalse(os.path.isfile(fpath))
-        self.assertFalse(os.path.isfile(tracker_filename))
     finally:
       if os.path.exists(tracker_filename):
         os.unlink(tracker_filename)
 
   def test_cp_resumable_download_content_matches(self):
     """Tests download no-ops when tracker file matches existing file."""
     bucket_uri = self.CreateBucket()
     tmp_dir = self.CreateTempDir()
+    fpath = self.CreateTempFile(tmpdir=tmp_dir)
     matching_contents = 'abcd' * ONE_KIB
-    fpath = self.CreateTempFile(tmpdir=tmp_dir, contents=matching_contents)
+    temp_download_file = fpath + '_.gstmp'
+    with open(temp_download_file, 'w') as fp:
+      fp.write(matching_contents)
+
     object_uri = self.CreateObject(bucket_uri=bucket_uri, object_name='foo',
                                    contents=matching_contents)
     stdout = self.RunGsUtil(['ls', '-L', suri(object_uri)], return_stdout=True)
     etag_match = re.search(r'\s*ETag:\s*(.*)', stdout)
     self.assertIsNotNone(etag_match, 'Could not get object ETag')
     self.assertEqual(len(etag_match.groups()), 1,
                      'Did not match expected single ETag')
     etag = etag_match.group(1)
     tracker_filename = GetTrackerFilePath(
         StorageUrlFromString(fpath), TrackerFileType.DOWNLOAD, self.test_api)
     with open(tracker_filename, 'w') as tracker_fp:
       tracker_fp.write(etag)
     try:
       boto_config_for_test = ('GSUtil', 'resumable_threshold', str(ONE_KIB))
       with SetBotoConfigForTest([boto_config_for_test]):
         stderr = self.RunGsUtil(['cp', suri(object_uri), fpath],
                                 return_stderr=True)
-        self.assertIn('Download already complete for file', stderr)
+        self.assertIn('Download already complete', stderr)
         # Tracker file should be removed after successful hash validation.
         self.assertFalse(os.path.isfile(tracker_filename))
     finally:
       if os.path.exists(tracker_filename):
         os.unlink(tracker_filename)
 
   def test_cp_resumable_download_tracker_file_not_matches(self):
     """Tests that download overwrites when tracker file etag does not match."""
     bucket_uri = self.CreateBucket()
     tmp_dir = self.CreateTempDir()
@@ skipping 21 matching lines @@
         with open(fpath, 'r') as in_fp:
           contents = in_fp.read()
           self.assertEqual(contents, 'efgh' * ONE_KIB,
                            'File not overwritten when it should have been '
                            'due to a non-matching tracker file.')
         self.assertFalse(os.path.isfile(tracker_filename))
     finally:
       if os.path.exists(tracker_filename):
         os.unlink(tracker_filename)
 
+  @SequentialAndParallelTransfer
   def test_cp_resumable_download_gzip(self):
     """Tests that download can be resumed successfully with a gzipped file."""
     # Generate some reasonably incompressible data.  This compresses to a bit
     # around 128K in practice, but we assert specifically below that it is
     # larger than self.halt_size to guarantee that we can halt the download
     # partway through.
     object_uri = self.CreateObject()
     random.seed(0)
     contents = str([random.choice(string.ascii_letters)
                     for _ in xrange(ONE_KIB * 128)])
@@ skipping 20 matching lines @@
     fpath2 = self.CreateTempFile()
     test_callback_file = self.CreateTempFile(
         contents=pickle.dumps(_HaltingCopyCallbackHandler(False, 5)))
 
     boto_config_for_test = ('GSUtil', 'resumable_threshold', str(ONE_KIB))
     with SetBotoConfigForTest([boto_config_for_test]):
       stderr = self.RunGsUtil(['cp', '--testcallbackfile', test_callback_file,
                                suri(object_uri), suri(fpath2)],
                               return_stderr=True, expected_status=1)
       self.assertIn('Artifically halting download.', stderr)
+      self.assertIn('Downloading to temp gzip filename', stderr)
+
+      # Tracker files will have different names depending on if we are
+      # downloading sequentially or in parallel.
+      sliced_download_threshold = HumanReadableToBytes(
+          boto.config.get('GSUtil', 'sliced_object_download_threshold',
+                          DEFAULT_SLICED_OBJECT_DOWNLOAD_THRESHOLD))
+      sliced_download = (len(contents) > sliced_download_threshold
+                         and sliced_download_threshold > 0
+                         and UsingCrcmodExtension(crcmod))
+      if sliced_download:
+        trackerfile_type = TrackerFileType.SLICED_DOWNLOAD
+      else:
+        trackerfile_type = TrackerFileType.DOWNLOAD
       tracker_filename = GetTrackerFilePath(
-          StorageUrlFromString(fpath2), TrackerFileType.DOWNLOAD, self.test_api)
-      self.assertTrue(os.path.isfile(tracker_filename))
-      self.assertIn('Downloading to temp gzip filename', stderr)
+          StorageUrlFromString(fpath2), trackerfile_type, self.test_api)
+
       # We should have a temporary gzipped file, a tracker file, and no
       # final file yet.
+      self.assertTrue(os.path.isfile(tracker_filename))
       self.assertTrue(os.path.isfile('%s_.gztmp' % fpath2))
       stderr = self.RunGsUtil(['cp', suri(object_uri), suri(fpath2)],
                               return_stderr=True)
       self.assertIn('Resuming download', stderr)
       with open(fpath2, 'r') as f:
         self.assertEqual(f.read(), contents, 'File contents did not match.')
       self.assertFalse(os.path.isfile(tracker_filename))
       self.assertFalse(os.path.isfile('%s_.gztmp' % fpath2))
 
+  @SequentialAndParallelTransfer
+  def test_cp_resumable_download_check_hashes_never(self):
+    """Tests that resumble downloads work with check_hashes = never."""
+    bucket_uri = self.CreateBucket()
+    contents = 'abcd' * self.halt_size
+    object_uri = self.CreateObject(bucket_uri=bucket_uri, object_name='foo',
+                                   contents=contents)
+    fpath = self.CreateTempFile()
+    test_callback_file = self.CreateTempFile(
+        contents=pickle.dumps(_HaltingCopyCallbackHandler(False, 5)))
+
+    boto_config_for_test = [('GSUtil', 'resumable_threshold', str(ONE_KIB)),
+                            ('GSUtil', 'check_hashes', 'never')]
+    with SetBotoConfigForTest(boto_config_for_test):
+      stderr = self.RunGsUtil(['cp', '--testcallbackfile', test_callback_file,
+                               suri(object_uri), fpath],
+                              expected_status=1, return_stderr=True)
+      self.assertIn('Artifically halting download.', stderr)
+      stderr = self.RunGsUtil(['cp', suri(object_uri), fpath],
+                              return_stderr=True)
+      self.assertIn('Resuming download', stderr)
+      self.assertIn('Found no hashes to validate object downloaded', stderr)
+      with open(fpath, 'r') as f:
+        self.assertEqual(f.read(), contents, 'File contents did not match.')
+
   @SkipForS3('No resumable upload support for S3.')
   def test_cp_resumable_upload_bucket_deleted(self):
     """Tests that a not found exception is raised if bucket no longer exists."""
     bucket_uri = self.CreateBucket()
     fpath = self.CreateTempFile(contents='a' * 2 * ONE_KIB)
     boto_config_for_test = ('GSUtil', 'resumable_threshold', str(ONE_KIB))
     test_callback_file = self.CreateTempFile(
         contents=pickle.dumps(
             _DeleteBucketThenStartOverCopyCallbackHandler(5, bucket_uri)))
 
     with SetBotoConfigForTest([boto_config_for_test]):
       stderr = self.RunGsUtil(['cp', '--testcallbackfile', test_callback_file,
                                fpath, suri(bucket_uri)], return_stderr=True,
                               expected_status=1)
     self.assertIn('Deleting bucket', stderr)
     self.assertIn('bucket does not exist', stderr)
 
+  @SkipForS3('No sliced download support for S3.')
+  def test_cp_sliced_download(self):
+    """Tests that sliced object download works in the general case."""
+    bucket_uri = self.CreateBucket()
+    object_uri = self.CreateObject(bucket_uri=bucket_uri, object_name='foo',
+                                   contents='abc' * ONE_KIB)
+    fpath = self.CreateTempFile()
+
+    # Force fast crcmod to return True to test the basic sliced download
+    # scenario, ensuring that if the user installs crcmod, it will work.
+    boto_config_for_test = [
+        ('GSUtil', 'resumable_threshold', str(ONE_KIB)),
+        ('GSUtil', 'test_assume_fast_crcmod', 'True'),
+        ('GSUtil', 'sliced_object_download_threshold', str(ONE_KIB)),
+        ('GSUtil', 'sliced_object_download_max_components', '3')]
+
+    with SetBotoConfigForTest(boto_config_for_test):
+      self.RunGsUtil(['cp', suri(object_uri), fpath])
+
+      # Each tracker file should have been deleted.
+      tracker_filenames = GetSlicedDownloadTrackerFilePaths(
+          StorageUrlFromString(fpath), self.test_api)
+      for tracker_filename in tracker_filenames:
+        self.assertFalse(os.path.isfile(tracker_filename))
+
+      with open(fpath, 'r') as f:
+        self.assertEqual(f.read(), 'abc' * ONE_KIB, 'File contents differ')
+
+  @unittest.skipUnless(UsingCrcmodExtension(crcmod),
+                       'Sliced download requires fast crcmod.')
+  @SkipForS3('No sliced download support for S3.')
+  def test_cp_unresumable_sliced_download(self):
+    """Tests sliced download works when resumability is disabled."""
+    bucket_uri = self.CreateBucket()
+    object_uri = self.CreateObject(bucket_uri=bucket_uri, object_name='foo',
+                                   contents='abcd' * self.halt_size)
+    fpath = self.CreateTempFile()
+    test_callback_file = self.CreateTempFile(
+        contents=pickle.dumps(_HaltingCopyCallbackHandler(False, 5)))
+
+    boto_config_for_test = [
+        ('GSUtil', 'resumable_threshold', str(self.halt_size*5)),
+        ('GSUtil', 'sliced_object_download_threshold', str(self.halt_size)),
+        ('GSUtil', 'sliced_object_download_max_components', '4')]
+
+    with SetBotoConfigForTest(boto_config_for_test):
+      stderr = self.RunGsUtil(['cp', '--testcallbackfile', test_callback_file,
+                               suri(object_uri), suri(fpath)],
+                              return_stderr=True, expected_status=1)
+      self.assertIn('not downloaded successfully', stderr)
+      # Temporary download file should exist.
+      self.assertTrue(os.path.isfile(fpath + '_.gstmp'))
+
+      # No tracker files should exist.
+      tracker_filenames = GetSlicedDownloadTrackerFilePaths(
+          StorageUrlFromString(fpath), self.test_api)
+      for tracker_filename in tracker_filenames:
+        self.assertFalse(os.path.isfile(tracker_filename))
+
+    # Perform the entire download, without resuming.
+    with SetBotoConfigForTest(boto_config_for_test):
+      stderr = self.RunGsUtil(['cp', suri(object_uri), suri(fpath)],
+                              return_stderr=True)
+      self.assertNotIn('Resuming download', stderr)
+      # Temporary download file should have been deleted.
+      self.assertFalse(os.path.isfile(fpath + '_.gstmp'))
+      with open(fpath, 'r') as f:
+        self.assertEqual(f.read(), 'abcd' * self.halt_size,
+                         'File contents differ')
+
+  @unittest.skipUnless(UsingCrcmodExtension(crcmod),
+                       'Sliced download requires fast crcmod.')
+  @SkipForS3('No sliced download support for S3.')
+  def test_cp_sliced_download_resume(self):
+    """Tests that sliced object download is resumable."""
+    bucket_uri = self.CreateBucket()
+    object_uri = self.CreateObject(bucket_uri=bucket_uri, object_name='foo',
+                                   contents='abc' * self.halt_size)
+    fpath = self.CreateTempFile()
+    test_callback_file = self.CreateTempFile(
+        contents=pickle.dumps(_HaltingCopyCallbackHandler(False, 5)))
+
+    boto_config_for_test = [
+        ('GSUtil', 'resumable_threshold', str(self.halt_size)),
+        ('GSUtil', 'sliced_object_download_threshold', str(self.halt_size)),
+        ('GSUtil', 'sliced_object_download_max_components', '3')]
+
+    with SetBotoConfigForTest(boto_config_for_test):
+      stderr = self.RunGsUtil(['cp', '--testcallbackfile', test_callback_file,
+                               suri(object_uri), suri(fpath)],
+                              return_stderr=True, expected_status=1)
+      self.assertIn('not downloaded successfully', stderr)
+
+      # Each tracker file should exist.
+      tracker_filenames = GetSlicedDownloadTrackerFilePaths(
+          StorageUrlFromString(fpath), self.test_api)
+      for tracker_filename in tracker_filenames:
+        self.assertTrue(os.path.isfile(tracker_filename))
+
+      stderr = self.RunGsUtil(['cp', suri(object_uri), fpath],
+                              return_stderr=True)
+      self.assertIn('Resuming download', stderr)
+
+      # Each tracker file should have been deleted.
+      tracker_filenames = GetSlicedDownloadTrackerFilePaths(
+          StorageUrlFromString(fpath), self.test_api)
+      for tracker_filename in tracker_filenames:
+        self.assertFalse(os.path.isfile(tracker_filename))
+
+      with open(fpath, 'r') as f:
+        self.assertEqual(f.read(), 'abc' * self.halt_size,
+                         'File contents differ')
+
+  @unittest.skipUnless(UsingCrcmodExtension(crcmod),
+                       'Sliced download requires fast crcmod.')
+  @SkipForS3('No sliced download support for S3.')
+  def test_cp_sliced_download_partial_resume(self):
+    """Test sliced download resumability when some components are finished."""
+    bucket_uri = self.CreateBucket()
+    object_uri = self.CreateObject(bucket_uri=bucket_uri, object_name='foo',
+                                   contents='abc' * self.halt_size)
+    fpath = self.CreateTempFile()
+    test_callback_file = self.CreateTempFile(
+        contents=pickle.dumps(_HaltOneComponentCopyCallbackHandler(5)))
+
+    boto_config_for_test = [
+        ('GSUtil', 'resumable_threshold', str(self.halt_size)),
+        ('GSUtil', 'sliced_object_download_threshold', str(self.halt_size)),
+        ('GSUtil', 'sliced_object_download_max_components', '3')]
+
+    with SetBotoConfigForTest(boto_config_for_test):
+      stderr = self.RunGsUtil(['cp', '--testcallbackfile', test_callback_file,
+                               suri(object_uri), suri(fpath)],
+                              return_stderr=True, expected_status=1)
+      self.assertIn('not downloaded successfully', stderr)
+
+      # Each tracker file should exist.
+      tracker_filenames = GetSlicedDownloadTrackerFilePaths(
+          StorageUrlFromString(fpath), self.test_api)
+      for tracker_filename in tracker_filenames:
+        self.assertTrue(os.path.isfile(tracker_filename))
+
+      stderr = self.RunGsUtil(['cp', suri(object_uri), fpath],
+                              return_stderr=True)
+      self.assertIn('Resuming download', stderr)
+      self.assertIn('Download already complete', stderr)
+
+      # Each tracker file should have been deleted.
+      tracker_filenames = GetSlicedDownloadTrackerFilePaths(
+          StorageUrlFromString(fpath), self.test_api)
+      for tracker_filename in tracker_filenames:
+        self.assertFalse(os.path.isfile(tracker_filename))
+
+      with open(fpath, 'r') as f:
+        self.assertEqual(f.read(), 'abc' * self.halt_size,
+                         'File contents differ')
+
+  @unittest.skipUnless(UsingCrcmodExtension(crcmod),
+                       'Sliced download requires fast crcmod.')
+  @SkipForS3('No sliced download support for S3.')
+  def test_cp_sliced_download_resume_content_differs(self):
+    """Tests differing file contents are detected by sliced downloads."""
+    bucket_uri = self.CreateBucket()
+    object_uri = self.CreateObject(bucket_uri=bucket_uri, object_name='foo',
+                                   contents='abc' * self.halt_size)
+    fpath = self.CreateTempFile(contents='')
+    test_callback_file = self.CreateTempFile(
+        contents=pickle.dumps(_HaltingCopyCallbackHandler(False, 5)))
+
+    boto_config_for_test = [
+        ('GSUtil', 'resumable_threshold', str(self.halt_size)),
+        ('GSUtil', 'sliced_object_download_threshold', str(self.halt_size)),
+        ('GSUtil', 'sliced_object_download_max_components', '3')]
+
+    with SetBotoConfigForTest(boto_config_for_test):
+      stderr = self.RunGsUtil(['cp', '--testcallbackfile', test_callback_file,
+                               suri(object_uri), suri(fpath)],
+                              return_stderr=True, expected_status=1)
+      self.assertIn('not downloaded successfully', stderr)
+
+      # Temporary download file should exist.
+      self.assertTrue(os.path.isfile(fpath + '_.gstmp'))
+
+      # Each tracker file should exist.
+      tracker_filenames = GetSlicedDownloadTrackerFilePaths(
+          StorageUrlFromString(fpath), self.test_api)
+      for tracker_filename in tracker_filenames:
+        self.assertTrue(os.path.isfile(tracker_filename))
+
+      with open(fpath + '_.gstmp', 'r+b') as f:
+        f.write('altered file contents')
+
+      stderr = self.RunGsUtil(['cp', suri(object_uri), fpath],
+                              return_stderr=True, expected_status=1)
+      self.assertIn('Resuming download', stderr)
+      self.assertIn('doesn\'t match cloud-supplied digest', stderr)
+      self.assertIn('HashMismatchException: crc32c', stderr)
+
+      # Each tracker file should have been deleted.
+      tracker_filenames = GetSlicedDownloadTrackerFilePaths(
+          StorageUrlFromString(fpath), self.test_api)
+      for tracker_filename in tracker_filenames:
+        self.assertFalse(os.path.isfile(tracker_filename))
+
+      # Temporary file should have been deleted due to hash mismatch.
+      self.assertFalse(os.path.isfile(fpath + '_.gstmp'))
+      # Final file should not exist.
+      self.assertFalse(os.path.isfile(fpath))
+
+  @unittest.skipUnless(UsingCrcmodExtension(crcmod),
+                       'Sliced download requires fast crcmod.')
+  @SkipForS3('No sliced download support for S3.')
+  def test_cp_sliced_download_component_size_changed(self):
+    """Tests sliced download doesn't break when the boto config changes.
+
+    If the number of components used changes cross-process, the download should
+    be restarted.
+    """
+    bucket_uri = self.CreateBucket()
+    object_uri = self.CreateObject(bucket_uri=bucket_uri, object_name='foo',
+                                   contents='abcd' * self.halt_size)
+    fpath = self.CreateTempFile()
+    test_callback_file = self.CreateTempFile(
+        contents=pickle.dumps(_HaltingCopyCallbackHandler(False, 5)))
+
+    boto_config_for_test = [
+        ('GSUtil', 'resumable_threshold', str(self.halt_size)),
+        ('GSUtil', 'sliced_object_download_threshold', str(self.halt_size)),
+        ('GSUtil', 'sliced_object_download_component_size',
+         str(self.halt_size//4)),
+        ('GSUtil', 'sliced_object_download_max_components', '4')]
+
+    with SetBotoConfigForTest(boto_config_for_test):
+      stderr = self.RunGsUtil(['cp', '--testcallbackfile', test_callback_file,
+                               suri(object_uri), suri(fpath)],
+                              return_stderr=True, expected_status=1)
+      self.assertIn('not downloaded successfully', stderr)
+
+    boto_config_for_test = [
+        ('GSUtil', 'resumable_threshold', str(self.halt_size)),
+        ('GSUtil', 'sliced_object_download_threshold', str(self.halt_size)),
+        ('GSUtil', 'sliced_object_download_component_size',
+         str(self.halt_size//2)),
+        ('GSUtil', 'sliced_object_download_max_components', '2')]
+
+    with SetBotoConfigForTest(boto_config_for_test):
+      stderr = self.RunGsUtil(['cp', suri(object_uri), fpath],
+                              return_stderr=True)
+      self.assertIn('Sliced download tracker file doesn\'t match ', stderr)
+      self.assertIn('Restarting download from scratch', stderr)
+      self.assertNotIn('Resuming download', stderr)
+
+  @unittest.skipUnless(UsingCrcmodExtension(crcmod),
+                       'Sliced download requires fast crcmod.')
+  @SkipForS3('No sliced download support for S3.')
+  def test_cp_sliced_download_disabled_cross_process(self):
+    """Tests temporary files are not orphaned if sliced download is disabled.
+
+    Specifically, temporary files should be deleted when the corresponding
+    non-sliced download is completed.
+    """
+    bucket_uri = self.CreateBucket()
+    object_uri = self.CreateObject(bucket_uri=bucket_uri, object_name='foo',
+                                   contents='abcd' * self.halt_size)
+    fpath = self.CreateTempFile()
+    test_callback_file = self.CreateTempFile(
+        contents=pickle.dumps(_HaltingCopyCallbackHandler(False, 5)))
+
+    boto_config_for_test = [
+        ('GSUtil', 'resumable_threshold', str(self.halt_size)),
+        ('GSUtil', 'sliced_object_download_threshold', str(self.halt_size)),
+        ('GSUtil', 'sliced_object_download_max_components', '4')]
+
+    with SetBotoConfigForTest(boto_config_for_test):
+      stderr = self.RunGsUtil(['cp', '--testcallbackfile', test_callback_file,
+                               suri(object_uri), suri(fpath)],
+                              return_stderr=True, expected_status=1)
+      self.assertIn('not downloaded successfully', stderr)
+      # Temporary download file should exist.
+      self.assertTrue(os.path.isfile(fpath + '_.gstmp'))
+
+      # Each tracker file should exist.
+      tracker_filenames = GetSlicedDownloadTrackerFilePaths(
+          StorageUrlFromString(fpath), self.test_api)
+      for tracker_filename in tracker_filenames:
+        self.assertTrue(os.path.isfile(tracker_filename))
+
+    # Disable sliced downloads by increasing the threshold
+    boto_config_for_test = [
+        ('GSUtil', 'resumable_threshold', str(self.halt_size)),
+        ('GSUtil', 'sliced_object_download_threshold', str(self.halt_size*5)),
+        ('GSUtil', 'sliced_object_download_max_components', '4')]
+
+    with SetBotoConfigForTest(boto_config_for_test):
+      stderr = self.RunGsUtil(['cp', suri(object_uri), fpath],
+                              return_stderr=True)
+      self.assertNotIn('Resuming download', stderr)
+      # Temporary download file should have been deleted.
+      self.assertFalse(os.path.isfile(fpath + '_.gstmp'))
+
+      # Each tracker file should have been deleted.
+      for tracker_filename in tracker_filenames:
+        self.assertFalse(os.path.isfile(tracker_filename))
+      with open(fpath, 'r') as f:
+        self.assertEqual(f.read(), 'abcd' * self.halt_size)
+
   @SkipForS3('No resumable upload support for S3.')
   def test_cp_resumable_upload_start_over_http_error(self):
     for start_over_error in (404, 410):
       self.start_over_error_test_helper(start_over_error)
 
   def start_over_error_test_helper(self, http_error_num):
     bucket_uri = self.CreateBucket()
     fpath = self.CreateTempFile(contents='a' * 2 * ONE_KIB)
     boto_config_for_test = ('GSUtil', 'resumable_threshold', str(ONE_KIB))
     if self.test_api == ApiSelector.JSON:
@@ skipping 284 matching lines @@
   def test_cp_upload_respects_no_hashes(self):
     bucket_uri = self.CreateBucket()
     fpath = self.CreateTempFile(contents='abcd')
     with SetBotoConfigForTest([('GSUtil', 'check_hashes', 'never')]):
       log_handler = self.RunCommand('cp', [fpath, suri(bucket_uri)],
                                     return_log_handler=True)
     warning_messages = log_handler.messages['warning']
     self.assertEquals(1, len(warning_messages))
     self.assertIn('Found no hashes to validate object upload',
                   warning_messages[0])