1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
|
# 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 resumable downloads.
"""
import errno
import os
import re
import boto
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 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.
class ResumableDownloadTests(GSTestCase):
"""Resumable download test suite."""
def make_small_key(self):
small_src_key_as_string = os.urandom(SMALL_KEY_SIZE)
small_src_key = self._MakeKey(data=small_src_key_as_string)
return small_src_key_as_string, small_src_key
def make_tracker_file(self, tmpdir=None):
if not tmpdir:
tmpdir = self._MakeTempDir()
tracker_file = os.path.join(tmpdir, 'tracker')
return tracker_file
def make_dst_fp(self, tmpdir=None):
if not tmpdir:
tmpdir = self._MakeTempDir()
dst_file = os.path.join(tmpdir, 'dstfile')
return open(dst_file, 'w')
def test_non_resumable_download(self):
"""
Tests that non-resumable downloads work
"""
dst_fp = self.make_dst_fp()
small_src_key_as_string, small_src_key = self.make_small_key()
small_src_key.get_contents_to_file(dst_fp)
self.assertEqual(SMALL_KEY_SIZE,
get_cur_file_size(dst_fp))
self.assertEqual(small_src_key_as_string,
small_src_key.get_contents_as_string())
def test_download_without_persistent_tracker(self):
"""
Tests a single resumable download, with no tracker persistence
"""
res_download_handler = ResumableDownloadHandler()
dst_fp = self.make_dst_fp()
small_src_key_as_string, small_src_key = self.make_small_key()
small_src_key.get_contents_to_file(
dst_fp, res_download_handler=res_download_handler)
self.assertEqual(SMALL_KEY_SIZE,
get_cur_file_size(dst_fp))
self.assertEqual(small_src_key_as_string,
small_src_key.get_contents_as_string())
def test_failed_download_with_persistent_tracker(self):
"""
Tests that failed resumable download leaves a correct tracker file
"""
harness = CallbackTestHarness()
tmpdir = self._MakeTempDir()
tracker_file_name = self.make_tracker_file(tmpdir)
dst_fp = self.make_dst_fp(tmpdir)
res_download_handler = ResumableDownloadHandler(
tracker_file_name=tracker_file_name, num_retries=0)
small_src_key_as_string, small_src_key = self.make_small_key()
try:
small_src_key.get_contents_to_file(
dst_fp, cb=harness.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 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)
etag_line = f.readline()
self.assertEquals(etag_line.rstrip('\n'), small_src_key.etag.strip('"\''))
def test_retryable_exception_recovery(self):
"""
Tests handling of a retryable exception
"""
# Test one of the RETRYABLE_EXCEPTIONS.
exception = ResumableDownloadHandler.RETRYABLE_EXCEPTIONS[0]
harness = CallbackTestHarness(exception=exception)
res_download_handler = ResumableDownloadHandler(num_retries=1)
dst_fp = self.make_dst_fp()
small_src_key_as_string, small_src_key = self.make_small_key()
small_src_key.get_contents_to_file(
dst_fp, cb=harness.call,
res_download_handler=res_download_handler)
# Ensure downloaded object has correct content.
self.assertEqual(SMALL_KEY_SIZE,
get_cur_file_size(dst_fp))
self.assertEqual(small_src_key_as_string,
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")
harness = CallbackTestHarness(exception=exception)
res_download_handler = ResumableDownloadHandler(num_retries=1)
dst_fp = self.make_dst_fp()
small_src_key_as_string, small_src_key = self.make_small_key()
small_src_key.get_contents_to_file(
dst_fp, cb=harness.call,
res_download_handler=res_download_handler)
# Ensure downloaded object has correct content.
self.assertEqual(SMALL_KEY_SIZE,
get_cur_file_size(dst_fp))
self.assertEqual(small_src_key_as_string,
small_src_key.get_contents_as_string())
def test_non_retryable_exception_handling(self):
"""
Tests resumable download that fails with a non-retryable exception
"""
harness = CallbackTestHarness(
exception=OSError(errno.EACCES, 'Permission denied'))
res_download_handler = ResumableDownloadHandler(num_retries=1)
dst_fp = self.make_dst_fp()
small_src_key_as_string, small_src_key = self.make_small_key()
try:
small_src_key.get_contents_to_file(
dst_fp, cb=harness.call,
res_download_handler=res_download_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_download_with_persistent_tracker(self):
"""
Tests resumable download that fails once and then completes,
with tracker file
"""
harness = CallbackTestHarness()
tmpdir = self._MakeTempDir()
tracker_file_name = self.make_tracker_file(tmpdir)
dst_fp = self.make_dst_fp(tmpdir)
small_src_key_as_string, small_src_key = self.make_small_key()
res_download_handler = ResumableDownloadHandler(
tracker_file_name=tracker_file_name, num_retries=1)
small_src_key.get_contents_to_file(
dst_fp, cb=harness.call,
res_download_handler=res_download_handler)
# Ensure downloaded object has correct content.
self.assertEqual(SMALL_KEY_SIZE,
get_cur_file_size(dst_fp))
self.assertEqual(small_src_key_as_string,
small_src_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 download that fails twice in one process, then completes
"""
res_download_handler = ResumableDownloadHandler(num_retries=3)
dst_fp = self.make_dst_fp()
small_src_key_as_string, small_src_key = self.make_small_key()
small_src_key.get_contents_to_file(
dst_fp, res_download_handler=res_download_handler)
# Ensure downloaded object has correct content.
self.assertEqual(SMALL_KEY_SIZE,
get_cur_file_size(dst_fp))
self.assertEqual(small_src_key_as_string,
small_src_key.get_contents_as_string())
def test_multiple_in_process_failures_then_succeed_with_tracker_file(self):
"""
Tests resumable download 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
# ResumableDownloadHandler 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)
larger_src_key_as_string = os.urandom(LARGE_KEY_SIZE)
larger_src_key = self._MakeKey(data=larger_src_key_as_string)
tmpdir = self._MakeTempDir()
tracker_file_name = self.make_tracker_file(tmpdir)
dst_fp = self.make_dst_fp(tmpdir)
res_download_handler = ResumableDownloadHandler(
tracker_file_name=tracker_file_name, num_retries=0)
try:
larger_src_key.get_contents_to_file(
dst_fp, cb=harness.call,
res_download_handler=res_download_handler)
self.fail('Did not get expected ResumableDownloadException')
except ResumableDownloadException, 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_key.get_contents_to_file(
dst_fp, cb=harness.call,
res_download_handler=res_download_handler)
self.assertEqual(LARGE_KEY_SIZE,
get_cur_file_size(dst_fp))
self.assertEqual(larger_src_key_as_string,
larger_src_key.get_contents_as_string())
self.assertFalse(os.path.exists(tracker_file_name))
# Ensure some of the file was downloaded 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_download_with_inital_partial_download_before_failure(self):
"""
Tests resumable download that successfully downloads some content
before it fails, then restarts and completes
"""
# Set up harness to fail download after several hundred KB so download
# server will have saved something before we retry.
harness = CallbackTestHarness(
fail_after_n_bytes=LARGE_KEY_SIZE/2)
larger_src_key_as_string = os.urandom(LARGE_KEY_SIZE)
larger_src_key = self._MakeKey(data=larger_src_key_as_string)
res_download_handler = ResumableDownloadHandler(num_retries=1)
dst_fp = self.make_dst_fp()
larger_src_key.get_contents_to_file(
dst_fp, cb=harness.call,
res_download_handler=res_download_handler)
# Ensure downloaded object has correct content.
self.assertEqual(LARGE_KEY_SIZE,
get_cur_file_size(dst_fp))
self.assertEqual(larger_src_key_as_string,
larger_src_key.get_contents_as_string())
# Ensure some of the file was downloaded 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_zero_length_object_download(self):
"""
Tests downloading a zero-length object (exercises boundary conditions).
"""
res_download_handler = ResumableDownloadHandler()
dst_fp = self.make_dst_fp()
k = self._MakeKey()
k.get_contents_to_file(dst_fp,
res_download_handler=res_download_handler)
self.assertEqual(0, get_cur_file_size(dst_fp))
def test_download_with_invalid_tracker_etag(self):
"""
Tests resumable download with a tracker file containing an invalid etag
"""
tmp_dir = self._MakeTempDir()
dst_fp = self.make_dst_fp(tmp_dir)
small_src_key_as_string, small_src_key = self.make_small_key()
invalid_etag_tracker_file_name = os.path.join(tmp_dir,
'invalid_etag_tracker')
f = open(invalid_etag_tracker_file_name, 'w')
f.write('3.14159\n')
f.close()
res_download_handler = ResumableDownloadHandler(
tracker_file_name=invalid_etag_tracker_file_name)
# An error should be printed about the invalid tracker, but then it
# should run the update successfully.
small_src_key.get_contents_to_file(
dst_fp, res_download_handler=res_download_handler)
self.assertEqual(SMALL_KEY_SIZE, get_cur_file_size(dst_fp))
self.assertEqual(small_src_key_as_string,
small_src_key.get_contents_as_string())
def test_download_with_inconsistent_etag_in_tracker(self):
"""
Tests resumable download with an inconsistent etag in tracker file
"""
tmp_dir = self._MakeTempDir()
dst_fp = self.make_dst_fp(tmp_dir)
small_src_key_as_string, small_src_key = self.make_small_key()
inconsistent_etag_tracker_file_name = os.path.join(tmp_dir,
'inconsistent_etag_tracker')
f = open(inconsistent_etag_tracker_file_name, 'w')
good_etag = small_src_key.etag.strip('"\'')
new_val_as_list = []
for c in reversed(good_etag):
new_val_as_list.append(c)
f.write('%s\n' % ''.join(new_val_as_list))
f.close()
res_download_handler = ResumableDownloadHandler(
tracker_file_name=inconsistent_etag_tracker_file_name)
# An error should be printed about the expired tracker, but then it
# should run the update successfully.
small_src_key.get_contents_to_file(
dst_fp, res_download_handler=res_download_handler)
self.assertEqual(SMALL_KEY_SIZE,
get_cur_file_size(dst_fp))
self.assertEqual(small_src_key_as_string,
small_src_key.get_contents_as_string())
def test_download_with_unwritable_tracker_file(self):
"""
Tests resumable download with an unwritable tracker file
"""
# Make dir where tracker_file lives temporarily unwritable.
tmp_dir = self._MakeTempDir()
tracker_file_name = os.path.join(tmp_dir, 'tracker')
save_mod = os.stat(tmp_dir).st_mode
try:
os.chmod(tmp_dir, 0)
res_download_handler = ResumableDownloadHandler(
tracker_file_name=tracker_file_name)
except ResumableDownloadException, 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)
|
