# 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- # lowing conditions: # # The above copyright notice and this permission notice shall be included # in all copies or substantial portions of the Software. # # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS # OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABIL- # ITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT # SHALL THE AUTHOR BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, # WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS # IN THE SOFTWARE. """ Tests of Google Cloud Storage resumable uploads. """ import StringIO import errno import random import os import time import boto from boto import storage_uri from boto.gs.resumable_upload_handler import ResumableUploadHandler from boto.exception import InvalidUriError from boto.exception import ResumableTransferDisposition from boto.exception import ResumableUploadException from cb_test_harness import CallbackTestHarness from tests.integration.gs.testcase import GSTestCase SMALL_KEY_SIZE = 2 * 1024 # 2 KB. LARGE_KEY_SIZE = 500 * 1024 # 500 KB. LARGEST_KEY_SIZE = 1024 * 1024 # 1 MB. class ResumableUploadTests(GSTestCase): """Resumable upload test suite.""" def build_input_file(self, size): buf = [] # I manually construct the random data here instead of calling # os.urandom() because I want to constrain the range of data (in # this case to 0'..'9') so the test # code can easily overwrite part of the StringIO file with # known-to-be-different values. for i in range(size): buf.append(str(random.randint(0, 9))) file_as_string = ''.join(buf) return (file_as_string, StringIO.StringIO(file_as_string)) def make_small_file(self): return self.build_input_file(SMALL_KEY_SIZE) def make_large_file(self): return self.build_input_file(LARGE_KEY_SIZE) def make_tracker_file(self, tmpdir=None): if not tmpdir: tmpdir = self._MakeTempDir() tracker_file = os.path.join(tmpdir, 'tracker') return tracker_file def test_non_resumable_upload(self): """ Tests that non-resumable uploads work """ small_src_file_as_string, small_src_file = self.make_small_file() # Seek to end incase its the first test. small_src_file.seek(0, os.SEEK_END) dst_key = self._MakeKey(set_contents=False) try: dst_key.set_contents_from_file(small_src_file) self.fail("should fail as need to rewind the filepointer") except AttributeError: pass # Now try calling with a proper rewind. dst_key.set_contents_from_file(small_src_file, rewind=True) self.assertEqual(SMALL_KEY_SIZE, dst_key.size) self.assertEqual(small_src_file_as_string, dst_key.get_contents_as_string()) def test_upload_without_persistent_tracker(self): """ Tests a single resumable upload, with no tracker URI persistence """ res_upload_handler = ResumableUploadHandler() small_src_file_as_string, small_src_file = self.make_small_file() small_src_file.seek(0) dst_key = self._MakeKey(set_contents=False) dst_key.set_contents_from_file( small_src_file, res_upload_handler=res_upload_handler) self.assertEqual(SMALL_KEY_SIZE, dst_key.size) self.assertEqual(small_src_file_as_string, dst_key.get_contents_as_string()) def test_failed_upload_with_persistent_tracker(self): """ Tests that failed resumable upload leaves a correct tracker URI file """ harness = CallbackTestHarness() tracker_file_name = self.make_tracker_file() res_upload_handler = ResumableUploadHandler( tracker_file_name=tracker_file_name, num_retries=0) small_src_file_as_string, small_src_file = self.make_small_file() small_src_file.seek(0) dst_key = self._MakeKey(set_contents=False) try: dst_key.set_contents_from_file( small_src_file, cb=harness.call, res_upload_handler=res_upload_handler) self.fail('Did not get expected ResumableUploadException') except ResumableUploadException, e: # We'll get a ResumableUploadException at this point because # of CallbackTestHarness (above). Check that the tracker file was # created correctly. self.assertEqual(e.disposition, ResumableTransferDisposition.ABORT_CUR_PROCESS) self.assertTrue(os.path.exists(tracker_file_name)) f = open(tracker_file_name) uri_from_file = f.readline().strip() f.close() self.assertEqual(uri_from_file, res_upload_handler.get_tracker_uri()) def test_retryable_exception_recovery(self): """ Tests handling of a retryable exception """ # Test one of the RETRYABLE_EXCEPTIONS. exception = ResumableUploadHandler.RETRYABLE_EXCEPTIONS[0] harness = CallbackTestHarness(exception=exception) res_upload_handler = ResumableUploadHandler(num_retries=1) small_src_file_as_string, small_src_file = self.make_small_file() small_src_file.seek(0) dst_key = self._MakeKey(set_contents=False) dst_key.set_contents_from_file( small_src_file, cb=harness.call, res_upload_handler=res_upload_handler) # Ensure uploaded object has correct content. self.assertEqual(SMALL_KEY_SIZE, dst_key.size) self.assertEqual(small_src_file_as_string, dst_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") harness = CallbackTestHarness(exception=exception) res_upload_handler = ResumableUploadHandler(num_retries=1) small_src_file_as_string, small_src_file = self.make_small_file() small_src_file.seek(0) dst_key = self._MakeKey(set_contents=False) dst_key.set_contents_from_file( small_src_file, cb=harness.call, res_upload_handler=res_upload_handler) # Ensure uploaded object has correct content. self.assertEqual(SMALL_KEY_SIZE, dst_key.size) self.assertEqual(small_src_file_as_string, dst_key.get_contents_as_string()) def test_non_retryable_exception_handling(self): """ Tests a resumable upload that fails with a non-retryable exception """ harness = CallbackTestHarness( exception=OSError(errno.EACCES, 'Permission denied')) res_upload_handler = ResumableUploadHandler(num_retries=1) small_src_file_as_string, small_src_file = self.make_small_file() small_src_file.seek(0) dst_key = self._MakeKey(set_contents=False) try: dst_key.set_contents_from_file( small_src_file, cb=harness.call, res_upload_handler=res_upload_handler) self.fail('Did not get expected OSError') except OSError, e: # Ensure the error was re-raised. self.assertEqual(e.errno, 13) def test_failed_and_restarted_upload_with_persistent_tracker(self): """ Tests resumable upload that fails once and then completes, with tracker file """ harness = CallbackTestHarness() tracker_file_name = self.make_tracker_file() res_upload_handler = ResumableUploadHandler( tracker_file_name=tracker_file_name, num_retries=1) small_src_file_as_string, small_src_file = self.make_small_file() small_src_file.seek(0) dst_key = self._MakeKey(set_contents=False) dst_key.set_contents_from_file( small_src_file, cb=harness.call, res_upload_handler=res_upload_handler) # Ensure uploaded object has correct content. self.assertEqual(SMALL_KEY_SIZE, dst_key.size) self.assertEqual(small_src_file_as_string, dst_key.get_contents_as_string()) # Ensure tracker file deleted. self.assertFalse(os.path.exists(tracker_file_name)) def test_multiple_in_process_failures_then_succeed(self): """ Tests resumable upload that fails twice in one process, then completes """ res_upload_handler = ResumableUploadHandler(num_retries=3) small_src_file_as_string, small_src_file = self.make_small_file() small_src_file.seek(0) dst_key = self._MakeKey(set_contents=False) dst_key.set_contents_from_file( small_src_file, res_upload_handler=res_upload_handler) # Ensure uploaded object has correct content. self.assertEqual(SMALL_KEY_SIZE, dst_key.size) self.assertEqual(small_src_file_as_string, dst_key.get_contents_as_string()) def test_multiple_in_process_failures_then_succeed_with_tracker_file(self): """ Tests resumable upload that fails completely in one process, then when restarted completes, using a tracker file """ # Set up test harness that causes more failures than a single # ResumableUploadHandler instance will handle, writing enough data # before the first failure that some of it survives that process run. harness = CallbackTestHarness( fail_after_n_bytes=LARGE_KEY_SIZE/2, num_times_to_fail=2) tracker_file_name = self.make_tracker_file() res_upload_handler = ResumableUploadHandler( tracker_file_name=tracker_file_name, num_retries=1) larger_src_file_as_string, larger_src_file = self.make_large_file() larger_src_file.seek(0) dst_key = self._MakeKey(set_contents=False) try: dst_key.set_contents_from_file( larger_src_file, cb=harness.call, res_upload_handler=res_upload_handler) self.fail('Did not get expected ResumableUploadException') except ResumableUploadException, e: self.assertEqual(e.disposition, ResumableTransferDisposition.ABORT_CUR_PROCESS) # Ensure a tracker file survived. self.assertTrue(os.path.exists(tracker_file_name)) # Try it one more time; this time should succeed. larger_src_file.seek(0) dst_key.set_contents_from_file( larger_src_file, cb=harness.call, res_upload_handler=res_upload_handler) self.assertEqual(LARGE_KEY_SIZE, dst_key.size) self.assertEqual(larger_src_file_as_string, dst_key.get_contents_as_string()) self.assertFalse(os.path.exists(tracker_file_name)) # Ensure some of the file was uploaded both before and after failure. self.assertTrue(len(harness.transferred_seq_before_first_failure) > 1 and len(harness.transferred_seq_after_first_failure) > 1) def test_upload_with_inital_partial_upload_before_failure(self): """ Tests resumable upload that successfully uploads some content before it fails, then restarts and completes """ # Set up harness to fail upload after several hundred KB so upload # server will have saved something before we retry. harness = CallbackTestHarness( fail_after_n_bytes=LARGE_KEY_SIZE/2) res_upload_handler = ResumableUploadHandler(num_retries=1) larger_src_file_as_string, larger_src_file = self.make_large_file() larger_src_file.seek(0) dst_key = self._MakeKey(set_contents=False) dst_key.set_contents_from_file( larger_src_file, cb=harness.call, res_upload_handler=res_upload_handler) # Ensure uploaded object has correct content. self.assertEqual(LARGE_KEY_SIZE, dst_key.size) self.assertEqual(larger_src_file_as_string, dst_key.get_contents_as_string()) # Ensure some of the file was uploaded both before and after failure. self.assertTrue(len(harness.transferred_seq_before_first_failure) > 1 and len(harness.transferred_seq_after_first_failure) > 1) def test_empty_file_upload(self): """ Tests uploading an empty file (exercises boundary conditions). """ res_upload_handler = ResumableUploadHandler() empty_src_file = StringIO.StringIO('') empty_src_file.seek(0) dst_key = self._MakeKey(set_contents=False) dst_key.set_contents_from_file( empty_src_file, res_upload_handler=res_upload_handler) self.assertEqual(0, dst_key.size) def test_upload_retains_metadata(self): """ Tests that resumable upload correctly sets passed metadata """ res_upload_handler = ResumableUploadHandler() headers = {'Content-Type' : 'text/plain', 'x-goog-meta-abc' : 'my meta', 'x-goog-acl' : 'public-read'} small_src_file_as_string, small_src_file = self.make_small_file() small_src_file.seek(0) dst_key = self._MakeKey(set_contents=False) dst_key.set_contents_from_file( small_src_file, headers=headers, res_upload_handler=res_upload_handler) self.assertEqual(SMALL_KEY_SIZE, dst_key.size) self.assertEqual(small_src_file_as_string, dst_key.get_contents_as_string()) dst_key.open_read() self.assertEqual('text/plain', dst_key.content_type) self.assertTrue('abc' in dst_key.metadata) self.assertEqual('my meta', str(dst_key.metadata['abc'])) acl = dst_key.get_acl() for entry in acl.entries.entry_list: if str(entry.scope) == '': self.assertEqual('READ', str(acl.entries.entry_list[1].permission)) return self.fail('No scope found') def test_upload_with_file_size_change_between_starts(self): """ Tests resumable upload on a file that changes sizes between initial upload start and restart """ harness = CallbackTestHarness( fail_after_n_bytes=LARGE_KEY_SIZE/2) tracker_file_name = self.make_tracker_file() # Set up first process' ResumableUploadHandler not to do any # retries (initial upload request will establish expected size to # upload server). res_upload_handler = ResumableUploadHandler( tracker_file_name=tracker_file_name, num_retries=0) larger_src_file_as_string, larger_src_file = self.make_large_file() larger_src_file.seek(0) dst_key = self._MakeKey(set_contents=False) try: dst_key.set_contents_from_file( larger_src_file, cb=harness.call, res_upload_handler=res_upload_handler) self.fail('Did not get expected ResumableUploadException') except ResumableUploadException, e: # First abort (from harness-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(tracker_file_name)) # Try it again, this time with different size source file. # Wait 1 second between retry attempts, to give upload server a # chance to save state so it can respond to changed file size with # 500 response in the next attempt. time.sleep(1) try: largest_src_file = self.build_input_file(LARGEST_KEY_SIZE)[1] largest_src_file.seek(0) dst_key.set_contents_from_file( largest_src_file, res_upload_handler=res_upload_handler) self.fail('Did not get expected ResumableUploadException') except ResumableUploadException, e: # This abort should be a hard abort (file size changing during # transfer). self.assertEqual(e.disposition, ResumableTransferDisposition.ABORT) self.assertNotEqual(e.message.find('file size changed'), -1, e.message) def test_upload_with_file_size_change_during_upload(self): """ Tests resumable upload on a file that changes sizes while upload in progress """ # Create a file we can change during the upload. test_file_size = 500 * 1024 # 500 KB. test_file = self.build_input_file(test_file_size)[1] harness = CallbackTestHarness(fp_to_change=test_file, fp_change_pos=test_file_size) res_upload_handler = ResumableUploadHandler(num_retries=1) dst_key = self._MakeKey(set_contents=False) try: dst_key.set_contents_from_file( test_file, cb=harness.call, res_upload_handler=res_upload_handler) self.fail('Did not get expected ResumableUploadException') except ResumableUploadException, e: self.assertEqual(e.disposition, ResumableTransferDisposition.ABORT) self.assertNotEqual( e.message.find('File changed during upload'), -1) def test_upload_with_file_content_change_during_upload(self): """ Tests resumable upload on a file that changes one byte of content (so, size stays the same) while upload in progress. """ def Execute(): res_upload_handler = ResumableUploadHandler(num_retries=1) dst_key = self._MakeKey(set_contents=False) bucket_uri = storage_uri('gs://' + dst_key.bucket.name) dst_key_uri = bucket_uri.clone_replace_name(dst_key.name) try: dst_key.set_contents_from_file( test_file, cb=harness.call, res_upload_handler=res_upload_handler) return False except ResumableUploadException, e: self.assertEqual(e.disposition, ResumableTransferDisposition.ABORT) # Ensure the file size didn't change. test_file.seek(0, os.SEEK_END) self.assertEqual(test_file_size, test_file.tell()) self.assertNotEqual( e.message.find('md5 signature doesn\'t match etag'), -1) # Ensure the bad data wasn't left around. try: dst_key_uri.get_key() self.fail('Did not get expected InvalidUriError') except InvalidUriError, e: pass return True test_file_size = 500 * 1024 # 500 KB # The sizes of all the blocks written, except the final block, must be a # multiple of 256K bytes. We need to trigger a failure after the first # 256K bytes have been uploaded so that at least one block of data is # written on the server. # See https://developers.google.com/storage/docs/concepts-techniques#resumable # for more information about chunking of uploads. n_bytes = 300 * 1024 # 300 KB delay = 0 # First, try the test without a delay. If that fails, try it with a # 15-second delay. The first attempt may fail to recognize that the # server has a block if the server hasn't yet committed that block # when we resume the transfer. This would cause a restarted upload # instead of a resumed upload. for attempt in range(2): test_file = self.build_input_file(test_file_size)[1] harness = CallbackTestHarness( fail_after_n_bytes=n_bytes, fp_to_change=test_file, # Write to byte 1, as the CallbackTestHarness writes # 3 bytes. This will result in the data on the server # being different than the local file. fp_change_pos=1, delay_after_change=delay) if Execute(): break if (attempt == 0 and 0 in harness.transferred_seq_after_first_failure): # We can confirm the upload was restarted instead of resumed # by determining if there is an entry of 0 in the # transferred_seq_after_first_failure list. # In that case, try again with a 15 second delay. delay = 15 continue self.fail('Did not get expected ResumableUploadException') def test_upload_with_content_length_header_set(self): """ Tests resumable upload on a file when the user supplies a Content-Length header. This is used by gsutil, for example, to set the content length when gzipping a file. """ res_upload_handler = ResumableUploadHandler() small_src_file_as_string, small_src_file = self.make_small_file() small_src_file.seek(0) dst_key = self._MakeKey(set_contents=False) try: dst_key.set_contents_from_file( small_src_file, res_upload_handler=res_upload_handler, headers={'Content-Length' : SMALL_KEY_SIZE}) self.fail('Did not get expected ResumableUploadException') except ResumableUploadException, e: self.assertEqual(e.disposition, ResumableTransferDisposition.ABORT) self.assertNotEqual( e.message.find('Attempt to specify Content-Length header'), -1) def test_upload_with_syntactically_invalid_tracker_uri(self): """ Tests resumable upload with a syntactically invalid tracker URI """ tmp_dir = self._MakeTempDir() syntactically_invalid_tracker_file_name = os.path.join(tmp_dir, 'synt_invalid_uri_tracker') with open(syntactically_invalid_tracker_file_name, 'w') as f: f.write('ftp://example.com') res_upload_handler = ResumableUploadHandler( tracker_file_name=syntactically_invalid_tracker_file_name) small_src_file_as_string, small_src_file = self.make_small_file() # An error should be printed about the invalid URI, but then it # should run the update successfully. small_src_file.seek(0) dst_key = self._MakeKey(set_contents=False) dst_key.set_contents_from_file( small_src_file, res_upload_handler=res_upload_handler) self.assertEqual(SMALL_KEY_SIZE, dst_key.size) self.assertEqual(small_src_file_as_string, dst_key.get_contents_as_string()) def test_upload_with_invalid_upload_id_in_tracker_file(self): """ Tests resumable upload with invalid upload ID """ invalid_upload_id = ('http://pub.storage.googleapis.com/?upload_id=' 'AyzB2Uo74W4EYxyi5dp_-r68jz8rtbvshsv4TX7srJVkJ57CxTY5Dw2') tmpdir = self._MakeTempDir() invalid_upload_id_tracker_file_name = os.path.join(tmpdir, 'invalid_upload_id_tracker') with open(invalid_upload_id_tracker_file_name, 'w') as f: f.write(invalid_upload_id) res_upload_handler = ResumableUploadHandler( tracker_file_name=invalid_upload_id_tracker_file_name) small_src_file_as_string, small_src_file = self.make_small_file() # An error should occur, but then the tracker URI should be # regenerated and the the update should succeed. small_src_file.seek(0) dst_key = self._MakeKey(set_contents=False) dst_key.set_contents_from_file( small_src_file, res_upload_handler=res_upload_handler) self.assertEqual(SMALL_KEY_SIZE, dst_key.size) self.assertEqual(small_src_file_as_string, dst_key.get_contents_as_string()) self.assertNotEqual(invalid_upload_id, res_upload_handler.get_tracker_uri()) def test_upload_with_unwritable_tracker_file(self): """ Tests resumable upload with an unwritable tracker file """ # Make dir where tracker_file lives temporarily unwritable. tmp_dir = self._MakeTempDir() tracker_file_name = self.make_tracker_file(tmp_dir) save_mod = os.stat(tmp_dir).st_mode try: os.chmod(tmp_dir, 0) res_upload_handler = ResumableUploadHandler( tracker_file_name=tracker_file_name) except ResumableUploadException, e: self.assertEqual(e.disposition, ResumableTransferDisposition.ABORT) self.assertNotEqual( e.message.find('Couldn\'t write URI tracker file'), -1) finally: # Restore original protection of dir where tracker_file lives. os.chmod(tmp_dir, save_mod)