Merging in latest boto.

tests/s3/test_resumable_downloads.py

 #!/usr/bin/env python
 #
 # Copyright 2010 Google Inc.
 #
 # Permission is hereby granted, free of charge, to any person obtaining a
 # copy of this software and associated documentation files (the
 # "Software"), to deal in the Software without restriction, including
 # without limitation the rights to use, copy, modify, merge, publish, dis-
 # tribute, sublicense, and/or sell copies of the Software, and to permit
 # persons to whom the Software is furnished to do so, subject to the fol-
@@ skipping 27 matching lines @@
 import time
 import unittest
 
 import boto
 from boto import storage_uri
 from boto.s3.resumable_download_handler import get_cur_file_size
 from boto.s3.resumable_download_handler import ResumableDownloadHandler
 from boto.exception import ResumableTransferDisposition
 from boto.exception import ResumableDownloadException
 from boto.exception import StorageResponseError
-from boto.tests.cb_test_harnass import CallbackTestHarnass
+from cb_test_harnass import CallbackTestHarnass
+
+# We don't use the OAuth2 authentication plugin directly; importing it here
+# ensures that it's loaded and available by default.
+try:
+  from oauth2_plugin import oauth2_plugin
+except ImportError:
+  # Do nothing - if user doesn't have OAuth2 configured it doesn't matter;
+  # and if they do, the tests will fail (as they should in that case).
+  pass
 
 
 class ResumableDownloadTests(unittest.TestCase):
     """
     Resumable download test suite.
     """
 
     def get_suite_description(self):
         return 'Resumable download test suite'
 
@@ skipping 47 matching lines @@
         return (string_data, key)
 
     @classmethod
     def set_up_class(cls, debug):
         """
         Initializes test suite.
         """
 
         # Create the test bucket.
         hostname = socket.gethostname().split('.')[0]
-        uri_base_str = 'gs://res_download_test_%s_%s_%s' % (
+        uri_base_str = 'gs://res-download-test-%s-%s-%s' % (
             hostname, os.getpid(), int(time.time()))
-        cls.src_bucket_uri = storage_uri('%s_dst' % uri_base_str)
+        cls.src_bucket_uri = storage_uri('%s-dst' % uri_base_str)
         cls.src_bucket_uri.create_bucket()
 
         # Create test source objects.
         cls.empty_src_key_size = 0
         (cls.empty_src_key_as_string, cls.empty_src_key) = (
             cls.build_test_input_object('empty', cls.empty_src_key_size,
                                         debug=debug))
         cls.small_src_key_size = 2 * 1024  # 2 KB.
         (cls.small_src_key_as_string, cls.small_src_key) = (
             cls.build_test_input_object('small', cls.small_src_key_size,
@@ skipping 76 matching lines @@
             tracker_file_name=self.tracker_file_name, num_retries=0)
         try:
             self.small_src_key.get_contents_to_file(
                 self.dst_fp, cb=harnass.call,
                 res_download_handler=res_download_handler)
             self.fail('Did not get expected ResumableDownloadException')
         except ResumableDownloadException, e:
             # We'll get a ResumableDownloadException at this point because
             # of CallbackTestHarnass (above). Check that the tracker file was
             # created correctly.
-            self.assertEqual(e.disposition, ResumableTransferDisposition.ABORT)
+            self.assertEqual(e.disposition,
+                             ResumableTransferDisposition.ABORT_CUR_PROCESS)
             self.assertTrue(os.path.exists(self.tracker_file_name))
             f = open(self.tracker_file_name)
             etag_line = f.readline()
             m = re.search(ResumableDownloadHandler.ETAG_REGEX, etag_line)
             f.close()
             self.assertTrue(m)
 
     def test_retryable_exception_recovery(self):
         """
         Tests handling of a retryable exception
         """
         # Test one of the RETRYABLE_EXCEPTIONS.
         exception = ResumableDownloadHandler.RETRYABLE_EXCEPTIONS[0]
         harnass = CallbackTestHarnass(exception=exception)
         res_download_handler = ResumableDownloadHandler(num_retries=1)
         self.small_src_key.get_contents_to_file(
             self.dst_fp, cb=harnass.call,
             res_download_handler=res_download_handler)
         # Ensure downloaded object has correct content.
         self.assertEqual(self.small_src_key_size,
                          get_cur_file_size(self.dst_fp))
         self.assertEqual(self.small_src_key_as_string,
                          self.small_src_key.get_contents_as_string())
 
+    def test_broken_pipe_recovery(self):
+        """
+        Tests handling of a Broken Pipe (which interacts with an httplib bug)
+        """
+        exception = IOError(errno.EPIPE, "Broken pipe")
+        harnass = CallbackTestHarnass(exception=exception)
+        res_download_handler = ResumableDownloadHandler(num_retries=1)
+        self.small_src_key.get_contents_to_file(
+            self.dst_fp, cb=harnass.call,
+            res_download_handler=res_download_handler)
+        # Ensure downloaded object has correct content.
+        self.assertEqual(self.small_src_key_size,
+                         get_cur_file_size(self.dst_fp))
+        self.assertEqual(self.small_src_key_as_string,
+                         self.small_src_key.get_contents_as_string())
+
     def test_non_retryable_exception_handling(self):
         """
         Tests resumable download that fails with a non-retryable exception
         """
         harnass = CallbackTestHarnass(
             exception=OSError(errno.EACCES, 'Permission denied'))
         res_download_handler = ResumableDownloadHandler(num_retries=1)
         try:
             self.small_src_key.get_contents_to_file(
                 self.dst_fp, cb=harnass.call,
@@ skipping 46 matching lines @@
         harnass = CallbackTestHarnass(
             fail_after_n_bytes=self.larger_src_key_size/2, num_times_to_fail=2)
         res_download_handler = ResumableDownloadHandler(
             tracker_file_name=self.tracker_file_name, num_retries=0)
         try:
             self.larger_src_key.get_contents_to_file(
                 self.dst_fp, cb=harnass.call,
                 res_download_handler=res_download_handler)
             self.fail('Did not get expected ResumableDownloadException')
         except ResumableDownloadException, e:
-            self.assertEqual(e.disposition, ResumableTransferDisposition.ABORT)
+            self.assertEqual(e.disposition,
+                             ResumableTransferDisposition.ABORT_CUR_PROCESS)
             # Ensure a tracker file survived.
             self.assertTrue(os.path.exists(self.tracker_file_name))
         # Try it one more time; this time should succeed.
         self.larger_src_key.get_contents_to_file(
             self.dst_fp, cb=harnass.call,
             res_download_handler=res_download_handler)
         self.assertEqual(self.larger_src_key_size,
                          get_cur_file_size(self.dst_fp))
         self.assertEqual(self.larger_src_key_as_string,
                          self.larger_src_key.get_contents_as_string())
@@ skipping 46 matching lines @@
         # retries (initial download request will establish expected size to
         # download server).
         res_download_handler = ResumableDownloadHandler(
             tracker_file_name=self.tracker_file_name, num_retries=0)
         try:
             self.larger_src_key.get_contents_to_file(
                 self.dst_fp, cb=harnass.call,
                 res_download_handler=res_download_handler)
             self.fail('Did not get expected ResumableDownloadException')
         except ResumableDownloadException, e:
-            self.assertEqual(e.disposition, ResumableTransferDisposition.ABORT)
+            # First abort (from harnass-forced failure) should be
+            # ABORT_CUR_PROCESS.
+            self.assertEqual(e.disposition, ResumableTransferDisposition.ABORT_CUR_PROCESS)
             # Ensure a tracker file survived.
             self.assertTrue(os.path.exists(self.tracker_file_name))
         # Try it again, this time with different src key (simulating an
         # object that changes sizes between downloads).
         try:
             self.small_src_key.get_contents_to_file(
                 self.dst_fp, res_download_handler=res_download_handler)
             self.fail('Did not get expected ResumableDownloadException')
         except ResumableDownloadException, e:
+            # This abort should be a hard abort (object size changing during
+            # transfer).
             self.assertEqual(e.disposition, ResumableTransferDisposition.ABORT)
             self.assertNotEqual(
                 e.message.find('md5 signature doesn\'t match etag'), -1)
 
     def test_download_with_file_content_change_during_download(self):
         """
         Tests resumable download on an object where the file content changes
         without changing length while download in progress
         """
         harnass = CallbackTestHarnass(
             fail_after_n_bytes=self.larger_src_key_size/2, num_times_to_fail=2)
         # Set up first process' ResumableDownloadHandler not to do any
         # retries (initial download request will establish expected size to
         # download server).
         res_download_handler = ResumableDownloadHandler(
             tracker_file_name=self.tracker_file_name, num_retries=0)
         dst_filename = self.dst_fp.name
         try:
             self.larger_src_key.get_contents_to_file(
                 self.dst_fp, cb=harnass.call,
                 res_download_handler=res_download_handler)
             self.fail('Did not get expected ResumableDownloadException')
         except ResumableDownloadException, e:
-            self.assertEqual(e.disposition, ResumableTransferDisposition.ABORT)
+            # First abort (from harnass-forced failure) should be
+            # ABORT_CUR_PROCESS.
+            self.assertEqual(e.disposition,
+                             ResumableTransferDisposition.ABORT_CUR_PROCESS)
             # Ensure a tracker file survived.
             self.assertTrue(os.path.exists(self.tracker_file_name))
         # Before trying again change the first byte of the file fragment
         # that was already downloaded.
         orig_size = get_cur_file_size(self.dst_fp)
         self.dst_fp.seek(0, os.SEEK_SET)
         self.dst_fp.write('a')
         # Ensure the file size didn't change.
         self.assertEqual(orig_size, get_cur_file_size(self.dst_fp))
         try:
             self.larger_src_key.get_contents_to_file(
                 self.dst_fp, cb=harnass.call,
                 res_download_handler=res_download_handler)
             self.fail('Did not get expected ResumableDownloadException')
         except ResumableDownloadException, e:
+            # This abort should be a hard abort (file content changing during
+            # transfer).
             self.assertEqual(e.disposition, ResumableTransferDisposition.ABORT)
             self.assertNotEqual(
                 e.message.find('md5 signature doesn\'t match etag'), -1)
             # Ensure the bad data wasn't left around.
             self.assertFalse(os.path.exists(dst_filename))
 
     def test_download_with_invalid_tracker_etag(self):
         """
         Tests resumable download with a tracker file containing an invalid etag
         """
@@ skipping 80 matching lines @@
     # that do it, with camelCase versions of these names).
     try:
         print 'Setting up %s...' % test_class.get_suite_description()
         test_class.set_up_class(debug)
         print 'Running %s...' % test_class.get_suite_description()
         unittest.TextTestRunner(verbosity=2).run(suite)
     finally:
         print 'Cleaning up after %s...' % test_class.get_suite_description()
         test_class.tear_down_class()
         print ''