Merge branch 'master' into 4054-sharing-email
[arvados.git] / sdk / python / tests / test_collections.py
1 # usage example:
2 #
3 # ARVADOS_API_TOKEN=abc ARVADOS_API_HOST=arvados.local python -m unittest discover
4
5 import arvados
6 import bz2
7 import copy
8 import mock
9 import os
10 import pprint
11 import subprocess
12 import tempfile
13 import unittest
14
15 import run_test_server
16 import arvados_testutil as tutil
17
18 class TestResumableWriter(arvados.ResumableCollectionWriter):
19     KEEP_BLOCK_SIZE = 1024  # PUT to Keep every 1K.
20
21     def current_state(self):
22         return self.dump_state(copy.deepcopy)
23
24
25 class ArvadosCollectionsTest(run_test_server.TestCaseWithServers,
26                              tutil.ArvadosBaseTestCase):
27     MAIN_SERVER = {}
28
29     @classmethod
30     def setUpClass(cls):
31         super(ArvadosCollectionsTest, cls).setUpClass()
32         run_test_server.authorize_with('active')
33         cls.api_client = arvados.api('v1')
34         cls.keep_client = arvados.KeepClient(api_client=cls.api_client,
35                                              local_store=cls.local_store)
36
37     def write_foo_bar_baz(self):
38         cw = arvados.CollectionWriter(self.api_client)
39         self.assertEqual(cw.current_stream_name(), '.',
40                          'current_stream_name() should be "." now')
41         cw.set_current_file_name('foo.txt')
42         cw.write('foo')
43         self.assertEqual(cw.current_file_name(), 'foo.txt',
44                          'current_file_name() should be foo.txt now')
45         cw.start_new_file('bar.txt')
46         cw.write('bar')
47         cw.start_new_stream('baz')
48         cw.write('baz')
49         cw.set_current_file_name('baz.txt')
50         return cw.finish()
51
52     def test_keep_local_store(self):
53         self.assertEqual(self.keep_client.put('foo'), 'acbd18db4cc2f85cedef654fccc4a4d8+3', 'wrong md5 hash from Keep.put')
54         self.assertEqual(self.keep_client.get('acbd18db4cc2f85cedef654fccc4a4d8+3'), 'foo', 'wrong data from Keep.get')
55
56     def test_local_collection_writer(self):
57         self.assertEqual(self.write_foo_bar_baz(),
58                          'd6c3b8e571f1b81ebb150a45ed06c884+114',
59                          "wrong locator hash for files foo, bar, baz")
60
61     def test_local_collection_reader(self):
62         self.write_foo_bar_baz()
63         cr = arvados.CollectionReader(
64             'd6c3b8e571f1b81ebb150a45ed06c884+114+Xzizzle', self.api_client)
65         got = []
66         for s in cr.all_streams():
67             for f in s.all_files():
68                 got += [[f.size(), f.stream_name(), f.name(), f.read(2**26)]]
69         expected = [[3, '.', 'bar.txt', 'bar'],
70                     [3, '.', 'foo.txt', 'foo'],
71                     [3, './baz', 'baz.txt', 'baz']]
72         self.assertEqual(got,
73                          expected)
74         stream0 = cr.all_streams()[0]
75         self.assertEqual(stream0.readfrom(0, 0),
76                          '',
77                          'reading zero bytes should have returned empty string')
78         self.assertEqual(stream0.readfrom(0, 2**26),
79                          'foobar',
80                          'reading entire stream failed')
81         self.assertEqual(stream0.readfrom(2**26, 0),
82                          '',
83                          'reading zero bytes should have returned empty string')
84
85     def _test_subset(self, collection, expected):
86         cr = arvados.CollectionReader(collection, self.api_client)
87         for s in cr.all_streams():
88             for ex in expected:
89                 if ex[0] == s:
90                     f = s.files()[ex[2]]
91                     got = [f.size(), f.stream_name(), f.name(), "".join(f.readall(2**26))]
92                     self.assertEqual(got,
93                                      ex,
94                                      'all_files|as_manifest did not preserve manifest contents: got %s expected %s' % (got, ex))
95
96     def test_collection_manifest_subset(self):
97         self.write_foo_bar_baz()
98         self._test_subset('d6c3b8e571f1b81ebb150a45ed06c884+114',
99                           [[3, '.',     'bar.txt', 'bar'],
100                            [3, '.',     'foo.txt', 'foo'],
101                            [3, './baz', 'baz.txt', 'baz']])
102         self._test_subset((". %s %s 0:3:foo.txt 3:3:bar.txt\n" %
103                            (self.keep_client.put("foo"),
104                             self.keep_client.put("bar"))),
105                           [[3, '.', 'bar.txt', 'bar'],
106                            [3, '.', 'foo.txt', 'foo']])
107         self._test_subset((". %s %s 0:2:fo.txt 2:4:obar.txt\n" %
108                            (self.keep_client.put("foo"),
109                             self.keep_client.put("bar"))),
110                           [[2, '.', 'fo.txt', 'fo'],
111                            [4, '.', 'obar.txt', 'obar']])
112         self._test_subset((". %s %s 0:2:fo.txt 2:0:zero.txt 2:2:ob.txt 4:2:ar.txt\n" %
113                            (self.keep_client.put("foo"),
114                             self.keep_client.put("bar"))),
115                           [[2, '.', 'ar.txt', 'ar'],
116                            [2, '.', 'fo.txt', 'fo'],
117                            [2, '.', 'ob.txt', 'ob'],
118                            [0, '.', 'zero.txt', '']])
119
120     def _test_readline(self, what_in, what_out):
121         cw = arvados.CollectionWriter(self.api_client)
122         cw.start_new_file('test.txt')
123         cw.write(what_in)
124         test1 = cw.finish()
125         cr = arvados.CollectionReader(test1, self.api_client)
126         got = []
127         for x in list(cr.all_files())[0].readlines():
128             got += [x]
129         self.assertEqual(got,
130                          what_out,
131                          "readlines did not split lines correctly: %s" % got)
132
133     def test_collection_readline(self):
134         self._test_readline("\na\nbcd\n\nefg\nz",
135                             ["\n", "a\n", "bcd\n", "\n", "efg\n", "z"])
136         self._test_readline("ab\ncd\n",
137                             ["ab\n", "cd\n"])
138
139     def test_collection_empty_file(self):
140         cw = arvados.CollectionWriter(self.api_client)
141         cw.start_new_file('zero.txt')
142         cw.write('')
143
144         self.assertEqual(cw.manifest_text(), ". d41d8cd98f00b204e9800998ecf8427e+0 0:0:zero.txt\n")
145         self.check_manifest_file_sizes(cw.manifest_text(), [0])
146         cw = arvados.CollectionWriter(self.api_client)
147         cw.start_new_file('zero.txt')
148         cw.write('')
149         cw.start_new_file('one.txt')
150         cw.write('1')
151         cw.start_new_stream('foo')
152         cw.start_new_file('zero.txt')
153         cw.write('')
154         self.check_manifest_file_sizes(cw.manifest_text(), [1,0,0])
155
156     def check_manifest_file_sizes(self, manifest_text, expect_sizes):
157         cr = arvados.CollectionReader(manifest_text, self.api_client)
158         got_sizes = []
159         for f in cr.all_files():
160             got_sizes += [f.size()]
161         self.assertEqual(got_sizes, expect_sizes, "got wrong file sizes %s, expected %s" % (got_sizes, expect_sizes))
162
163     def test_collection_bz2_decompression(self):
164         n_lines_in = 2**18
165         data_in = "abc\n"
166         for x in xrange(0, 18):
167             data_in += data_in
168         compressed_data_in = bz2.compress(data_in)
169         cw = arvados.CollectionWriter(self.api_client)
170         cw.start_new_file('test.bz2')
171         cw.write(compressed_data_in)
172         bz2_manifest = cw.manifest_text()
173
174         cr = arvados.CollectionReader(bz2_manifest, self.api_client)
175
176         got = 0
177         for x in list(cr.all_files())[0].readlines():
178             self.assertEqual(x, "abc\n", "decompression returned wrong data: %s" % x)
179             got += 1
180         self.assertEqual(got,
181                          n_lines_in,
182                          "decompression returned %d lines instead of %d" % (got, n_lines_in))
183
184     def test_collection_gzip_decompression(self):
185         n_lines_in = 2**18
186         data_in = "abc\n"
187         for x in xrange(0, 18):
188             data_in += data_in
189         p = subprocess.Popen(["gzip", "-1cn"],
190                              stdout=subprocess.PIPE,
191                              stdin=subprocess.PIPE,
192                              stderr=subprocess.PIPE,
193                              shell=False, close_fds=True)
194         compressed_data_in, stderrdata = p.communicate(data_in)
195
196         cw = arvados.CollectionWriter(self.api_client)
197         cw.start_new_file('test.gz')
198         cw.write(compressed_data_in)
199         gzip_manifest = cw.manifest_text()
200
201         cr = arvados.CollectionReader(gzip_manifest, self.api_client)
202         got = 0
203         for x in list(cr.all_files())[0].readlines():
204             self.assertEqual(x, "abc\n", "decompression returned wrong data: %s" % x)
205             got += 1
206         self.assertEqual(got,
207                          n_lines_in,
208                          "decompression returned %d lines instead of %d" % (got, n_lines_in))
209
210     def test_normalized_collection(self):
211         m1 = """. 5348b82a029fd9e971a811ce1f71360b+43 0:43:md5sum.txt
212 . 085c37f02916da1cad16f93c54d899b7+41 0:41:md5sum.txt
213 . 8b22da26f9f433dea0a10e5ec66d73ba+43 0:43:md5sum.txt"""
214         self.assertEqual(arvados.CollectionReader(m1, self.api_client).manifest_text(),
215                          """. 5348b82a029fd9e971a811ce1f71360b+43 085c37f02916da1cad16f93c54d899b7+41 8b22da26f9f433dea0a10e5ec66d73ba+43 0:127:md5sum.txt
216 """)
217
218         m2 = """. 204e43b8a1185621ca55a94839582e6f+67108864 b9677abbac956bd3e86b1deb28dfac03+67108864 fc15aff2a762b13f521baf042140acec+67108864 323d2a3ce20370c4ca1d3462a344f8fd+25885655 0:227212247:var-GS000016015-ASM.tsv.bz2
219 """
220         self.assertEqual(arvados.CollectionReader(m2, self.api_client).manifest_text(), m2)
221
222         m3 = """. 5348b82a029fd9e971a811ce1f71360b+43 3:40:md5sum.txt
223 . 085c37f02916da1cad16f93c54d899b7+41 0:41:md5sum.txt
224 . 8b22da26f9f433dea0a10e5ec66d73ba+43 0:43:md5sum.txt"""
225         self.assertEqual(arvados.CollectionReader(m3, self.api_client).manifest_text(),
226                          """. 5348b82a029fd9e971a811ce1f71360b+43 085c37f02916da1cad16f93c54d899b7+41 8b22da26f9f433dea0a10e5ec66d73ba+43 3:124:md5sum.txt
227 """)
228
229         m4 = """. 204e43b8a1185621ca55a94839582e6f+67108864 0:3:foo/bar
230 ./zzz 204e43b8a1185621ca55a94839582e6f+67108864 0:999:zzz
231 ./foo 323d2a3ce20370c4ca1d3462a344f8fd+25885655 0:3:bar"""
232         self.assertEqual(arvados.CollectionReader(m4, self.api_client).manifest_text(),
233                          """./foo 204e43b8a1185621ca55a94839582e6f+67108864 323d2a3ce20370c4ca1d3462a344f8fd+25885655 0:3:bar 67108864:3:bar
234 ./zzz 204e43b8a1185621ca55a94839582e6f+67108864 0:999:zzz
235 """)
236
237         m5 = """. 204e43b8a1185621ca55a94839582e6f+67108864 0:3:foo/bar
238 ./zzz 204e43b8a1185621ca55a94839582e6f+67108864 0:999:zzz
239 ./foo 204e43b8a1185621ca55a94839582e6f+67108864 3:3:bar"""
240         self.assertEqual(arvados.CollectionReader(m5, self.api_client).manifest_text(),
241                          """./foo 204e43b8a1185621ca55a94839582e6f+67108864 0:6:bar
242 ./zzz 204e43b8a1185621ca55a94839582e6f+67108864 0:999:zzz
243 """)
244
245         with self.data_file('1000G_ref_manifest') as f6:
246             m6 = f6.read()
247             self.assertEqual(arvados.CollectionReader(m6, self.api_client).manifest_text(), m6)
248
249         with self.data_file('jlake_manifest') as f7:
250             m7 = f7.read()
251             self.assertEqual(arvados.CollectionReader(m7, self.api_client).manifest_text(), m7)
252
253         m8 = """./a\\040b\\040c 59ca0efa9f5633cb0371bbc0355478d8+13 0:13:hello\\040world.txt
254 """
255         self.assertEqual(arvados.CollectionReader(m8, self.api_client).manifest_text(), m8)
256
257     def test_locators_and_ranges(self):
258         blocks2 = [['a', 10, 0],
259                   ['b', 10, 10],
260                   ['c', 10, 20],
261                   ['d', 10, 30],
262                   ['e', 10, 40],
263                   ['f', 10, 50]]
264
265         self.assertEqual(arvados.locators_and_ranges(blocks2,  2,  2), [['a', 10, 2, 2]])
266         self.assertEqual(arvados.locators_and_ranges(blocks2, 12, 2), [['b', 10, 2, 2]])
267         self.assertEqual(arvados.locators_and_ranges(blocks2, 22, 2), [['c', 10, 2, 2]])
268         self.assertEqual(arvados.locators_and_ranges(blocks2, 32, 2), [['d', 10, 2, 2]])
269         self.assertEqual(arvados.locators_and_ranges(blocks2, 42, 2), [['e', 10, 2, 2]])
270         self.assertEqual(arvados.locators_and_ranges(blocks2, 52, 2), [['f', 10, 2, 2]])
271         self.assertEqual(arvados.locators_and_ranges(blocks2, 62, 2), [])
272         self.assertEqual(arvados.locators_and_ranges(blocks2, -2, 2), [])
273
274         self.assertEqual(arvados.locators_and_ranges(blocks2,  0,  2), [['a', 10, 0, 2]])
275         self.assertEqual(arvados.locators_and_ranges(blocks2, 10, 2), [['b', 10, 0, 2]])
276         self.assertEqual(arvados.locators_and_ranges(blocks2, 20, 2), [['c', 10, 0, 2]])
277         self.assertEqual(arvados.locators_and_ranges(blocks2, 30, 2), [['d', 10, 0, 2]])
278         self.assertEqual(arvados.locators_and_ranges(blocks2, 40, 2), [['e', 10, 0, 2]])
279         self.assertEqual(arvados.locators_and_ranges(blocks2, 50, 2), [['f', 10, 0, 2]])
280         self.assertEqual(arvados.locators_and_ranges(blocks2, 60, 2), [])
281         self.assertEqual(arvados.locators_and_ranges(blocks2, -2, 2), [])
282
283         self.assertEqual(arvados.locators_and_ranges(blocks2,  9,  2), [['a', 10, 9, 1], ['b', 10, 0, 1]])
284         self.assertEqual(arvados.locators_and_ranges(blocks2, 19, 2), [['b', 10, 9, 1], ['c', 10, 0, 1]])
285         self.assertEqual(arvados.locators_and_ranges(blocks2, 29, 2), [['c', 10, 9, 1], ['d', 10, 0, 1]])
286         self.assertEqual(arvados.locators_and_ranges(blocks2, 39, 2), [['d', 10, 9, 1], ['e', 10, 0, 1]])
287         self.assertEqual(arvados.locators_and_ranges(blocks2, 49, 2), [['e', 10, 9, 1], ['f', 10, 0, 1]])
288         self.assertEqual(arvados.locators_and_ranges(blocks2, 59, 2), [['f', 10, 9, 1]])
289
290
291         blocks3 = [['a', 10, 0],
292                   ['b', 10, 10],
293                   ['c', 10, 20],
294                   ['d', 10, 30],
295                   ['e', 10, 40],
296                   ['f', 10, 50],
297                   ['g', 10, 60]]
298
299         self.assertEqual(arvados.locators_and_ranges(blocks3,  2,  2), [['a', 10, 2, 2]])
300         self.assertEqual(arvados.locators_and_ranges(blocks3, 12, 2), [['b', 10, 2, 2]])
301         self.assertEqual(arvados.locators_and_ranges(blocks3, 22, 2), [['c', 10, 2, 2]])
302         self.assertEqual(arvados.locators_and_ranges(blocks3, 32, 2), [['d', 10, 2, 2]])
303         self.assertEqual(arvados.locators_and_ranges(blocks3, 42, 2), [['e', 10, 2, 2]])
304         self.assertEqual(arvados.locators_and_ranges(blocks3, 52, 2), [['f', 10, 2, 2]])
305         self.assertEqual(arvados.locators_and_ranges(blocks3, 62, 2), [['g', 10, 2, 2]])
306
307
308         blocks = [['a', 10, 0],
309                   ['b', 15, 10],
310                   ['c', 5, 25]]
311         self.assertEqual(arvados.locators_and_ranges(blocks, 1, 0), [])
312         self.assertEqual(arvados.locators_and_ranges(blocks, 0, 5), [['a', 10, 0, 5]])
313         self.assertEqual(arvados.locators_and_ranges(blocks, 3, 5), [['a', 10, 3, 5]])
314         self.assertEqual(arvados.locators_and_ranges(blocks, 0, 10), [['a', 10, 0, 10]])
315
316         self.assertEqual(arvados.locators_and_ranges(blocks, 0, 11), [['a', 10, 0, 10],
317                                                                       ['b', 15, 0, 1]])
318         self.assertEqual(arvados.locators_and_ranges(blocks, 1, 11), [['a', 10, 1, 9],
319                                                                       ['b', 15, 0, 2]])
320         self.assertEqual(arvados.locators_and_ranges(blocks, 0, 25), [['a', 10, 0, 10],
321                                                                       ['b', 15, 0, 15]])
322
323         self.assertEqual(arvados.locators_and_ranges(blocks, 0, 30), [['a', 10, 0, 10],
324                                                                       ['b', 15, 0, 15],
325                                                                       ['c', 5, 0, 5]])
326         self.assertEqual(arvados.locators_and_ranges(blocks, 1, 30), [['a', 10, 1, 9],
327                                                                       ['b', 15, 0, 15],
328                                                                       ['c', 5, 0, 5]])
329         self.assertEqual(arvados.locators_and_ranges(blocks, 0, 31), [['a', 10, 0, 10],
330                                                                       ['b', 15, 0, 15],
331                                                                       ['c', 5, 0, 5]])
332
333         self.assertEqual(arvados.locators_and_ranges(blocks, 15, 5), [['b', 15, 5, 5]])
334
335         self.assertEqual(arvados.locators_and_ranges(blocks, 8, 17), [['a', 10, 8, 2],
336                                                                       ['b', 15, 0, 15]])
337
338         self.assertEqual(arvados.locators_and_ranges(blocks, 8, 20), [['a', 10, 8, 2],
339                                                                       ['b', 15, 0, 15],
340                                                                       ['c', 5, 0, 3]])
341
342         self.assertEqual(arvados.locators_and_ranges(blocks, 26, 2), [['c', 5, 1, 2]])
343
344         self.assertEqual(arvados.locators_and_ranges(blocks, 9, 15), [['a', 10, 9, 1],
345                                                                       ['b', 15, 0, 14]])
346         self.assertEqual(arvados.locators_and_ranges(blocks, 10, 15), [['b', 15, 0, 15]])
347         self.assertEqual(arvados.locators_and_ranges(blocks, 11, 15), [['b', 15, 1, 14],
348                                                                        ['c', 5, 0, 1]])
349
350     class MockStreamReader(object):
351         def __init__(self, content):
352             self.content = content
353             self.num_retries = 0
354
355         def readfrom(self, start, size, num_retries=0):
356             return self.content[start:start+size]
357
358     def test_file_stream(self):
359         content = 'abcdefghijklmnopqrstuvwxyz0123456789'
360         msr = self.MockStreamReader(content)
361         segments = [[0, 10, 0],
362                     [10, 15, 10],
363                     [25, 5, 25]]
364
365         sfr = arvados.StreamFileReader(msr, segments, "test")
366
367         self.assertEqual(sfr.name(), "test")
368         self.assertEqual(sfr.size(), 30)
369
370         self.assertEqual(sfr.readfrom(0, 30), content[0:30])
371         self.assertEqual(sfr.readfrom(2, 30), content[2:30])
372
373         self.assertEqual(sfr.readfrom(2, 8), content[2:10])
374         self.assertEqual(sfr.readfrom(0, 10), content[0:10])
375
376         self.assertEqual(sfr.tell(), 0)
377         self.assertEqual(sfr.read(5), content[0:5])
378         self.assertEqual(sfr.tell(), 5)
379         self.assertEqual(sfr.read(5), content[5:10])
380         self.assertEqual(sfr.tell(), 10)
381         self.assertEqual(sfr.read(5), content[10:15])
382         self.assertEqual(sfr.tell(), 15)
383         self.assertEqual(sfr.read(5), content[15:20])
384         self.assertEqual(sfr.tell(), 20)
385         self.assertEqual(sfr.read(5), content[20:25])
386         self.assertEqual(sfr.tell(), 25)
387         self.assertEqual(sfr.read(5), content[25:30])
388         self.assertEqual(sfr.tell(), 30)
389         self.assertEqual(sfr.read(5), '')
390         self.assertEqual(sfr.tell(), 30)
391
392         segments = [[26, 10, 0],
393                     [0, 15, 10],
394                     [15, 5, 25]]
395
396         sfr = arvados.StreamFileReader(msr, segments, "test")
397
398         self.assertEqual(sfr.size(), 30)
399
400         self.assertEqual(sfr.readfrom(0, 30), content[26:36] + content[0:20])
401         self.assertEqual(sfr.readfrom(2, 30), content[28:36] + content[0:20])
402
403         self.assertEqual(sfr.readfrom(2, 8), content[28:36])
404         self.assertEqual(sfr.readfrom(0, 10), content[26:36])
405
406         self.assertEqual(sfr.tell(), 0)
407         self.assertEqual(sfr.read(5), content[26:31])
408         self.assertEqual(sfr.tell(), 5)
409         self.assertEqual(sfr.read(5), content[31:36])
410         self.assertEqual(sfr.tell(), 10)
411         self.assertEqual(sfr.read(5), content[0:5])
412         self.assertEqual(sfr.tell(), 15)
413         self.assertEqual(sfr.read(5), content[5:10])
414         self.assertEqual(sfr.tell(), 20)
415         self.assertEqual(sfr.read(5), content[10:15])
416         self.assertEqual(sfr.tell(), 25)
417         self.assertEqual(sfr.read(5), content[15:20])
418         self.assertEqual(sfr.tell(), 30)
419         self.assertEqual(sfr.read(5), '')
420         self.assertEqual(sfr.tell(), 30)
421
422
423     class MockKeep(object):
424         def __init__(self, content, num_retries=0):
425             self.content = content
426
427         def get(self, locator, num_retries=0):
428             return self.content[locator]
429
430     def test_stream_reader(self):
431         keepblocks = {'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa+10': 'abcdefghij',
432                       'bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb+15': 'klmnopqrstuvwxy',
433                       'cccccccccccccccccccccccccccccccc+5': 'z0123'}
434         mk = self.MockKeep(keepblocks)
435
436         sr = arvados.StreamReader([".", "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa+10", "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb+15", "cccccccccccccccccccccccccccccccc+5", "0:30:foo"], mk)
437
438         content = 'abcdefghijklmnopqrstuvwxyz0123456789'
439
440         self.assertEqual(sr.readfrom(0, 30), content[0:30])
441         self.assertEqual(sr.readfrom(2, 30), content[2:30])
442
443         self.assertEqual(sr.readfrom(2, 8), content[2:10])
444         self.assertEqual(sr.readfrom(0, 10), content[0:10])
445
446         self.assertEqual(sr.readfrom(0, 5), content[0:5])
447         self.assertEqual(sr.readfrom(5, 5), content[5:10])
448         self.assertEqual(sr.readfrom(10, 5), content[10:15])
449         self.assertEqual(sr.readfrom(15, 5), content[15:20])
450         self.assertEqual(sr.readfrom(20, 5), content[20:25])
451         self.assertEqual(sr.readfrom(25, 5), content[25:30])
452         self.assertEqual(sr.readfrom(30, 5), '')
453
454     def test_file_reader(self):
455         keepblocks = {'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa+10': 'abcdefghij',
456                       'bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb+15': 'klmnopqrstuvwxy',
457                       'cccccccccccccccccccccccccccccccc+5': 'z0123'}
458         mk = self.MockKeep(keepblocks)
459
460         sr = arvados.StreamReader([".", "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa+10", "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb+15", "cccccccccccccccccccccccccccccccc+5", "0:10:foo", "15:10:foo"], mk)
461
462         content = 'abcdefghijpqrstuvwxy'
463
464         f = sr.files()["foo"]
465
466         # f.read() calls will be aligned on block boundaries (as a
467         # result of ticket #3663).
468
469         f.seek(0)
470         self.assertEqual(f.read(20), content[0:10])
471
472         f.seek(0)
473         self.assertEqual(f.read(6), content[0:6])
474         self.assertEqual(f.read(6), content[6:10])
475         self.assertEqual(f.read(6), content[10:16])
476         self.assertEqual(f.read(6), content[16:20])
477
478     def test_extract_file(self):
479         m1 = """. 5348b82a029fd9e971a811ce1f71360b+43 0:43:md5sum.txt
480 . 085c37f02916da1cad16f93c54d899b7+41 0:41:md6sum.txt
481 . 8b22da26f9f433dea0a10e5ec66d73ba+43 0:43:md7sum.txt
482 . 085c37f02916da1cad16f93c54d899b7+41 5348b82a029fd9e971a811ce1f71360b+43 8b22da26f9f433dea0a10e5ec66d73ba+43 47:80:md8sum.txt
483 . 085c37f02916da1cad16f93c54d899b7+41 5348b82a029fd9e971a811ce1f71360b+43 8b22da26f9f433dea0a10e5ec66d73ba+43 40:80:md9sum.txt"""
484
485         m2 = arvados.CollectionReader(m1, self.api_client)
486
487         self.assertEqual(m2.manifest_text(),
488                          ". 5348b82a029fd9e971a811ce1f71360b+43 085c37f02916da1cad16f93c54d899b7+41 8b22da26f9f433dea0a10e5ec66d73ba+43 0:43:md5sum.txt 43:41:md6sum.txt 84:43:md7sum.txt 6:37:md8sum.txt 84:43:md8sum.txt 83:1:md9sum.txt 0:43:md9sum.txt 84:36:md9sum.txt\n")
489
490         self.assertEqual(arvados.CollectionReader(m1, self.api_client).all_streams()[0].files()['md5sum.txt'].as_manifest(),
491                          ". 5348b82a029fd9e971a811ce1f71360b+43 0:43:md5sum.txt\n")
492         self.assertEqual(arvados.CollectionReader(m1, self.api_client).all_streams()[0].files()['md6sum.txt'].as_manifest(),
493                          ". 085c37f02916da1cad16f93c54d899b7+41 0:41:md6sum.txt\n")
494         self.assertEqual(arvados.CollectionReader(m1, self.api_client).all_streams()[0].files()['md7sum.txt'].as_manifest(),
495                          ". 8b22da26f9f433dea0a10e5ec66d73ba+43 0:43:md7sum.txt\n")
496         self.assertEqual(arvados.CollectionReader(m1, self.api_client).all_streams()[0].files()['md9sum.txt'].as_manifest(),
497                          ". 085c37f02916da1cad16f93c54d899b7+41 5348b82a029fd9e971a811ce1f71360b+43 8b22da26f9f433dea0a10e5ec66d73ba+43 40:80:md9sum.txt\n")
498
499     def test_write_directory_tree(self):
500         cwriter = arvados.CollectionWriter(self.api_client)
501         cwriter.write_directory_tree(self.build_directory_tree(
502                 ['basefile', 'subdir/subfile']))
503         self.assertEqual(cwriter.manifest_text(),
504                          """. c5110c5ac93202d8e0f9e381f22bac0f+8 0:8:basefile
505 ./subdir 1ca4dec89403084bf282ad31e6cf7972+14 0:14:subfile\n""")
506
507     def test_write_named_directory_tree(self):
508         cwriter = arvados.CollectionWriter(self.api_client)
509         cwriter.write_directory_tree(self.build_directory_tree(
510                 ['basefile', 'subdir/subfile']), 'root')
511         self.assertEqual(
512             cwriter.manifest_text(),
513             """./root c5110c5ac93202d8e0f9e381f22bac0f+8 0:8:basefile
514 ./root/subdir 1ca4dec89403084bf282ad31e6cf7972+14 0:14:subfile\n""")
515
516     def test_write_directory_tree_in_one_stream(self):
517         cwriter = arvados.CollectionWriter(self.api_client)
518         cwriter.write_directory_tree(self.build_directory_tree(
519                 ['basefile', 'subdir/subfile']), max_manifest_depth=0)
520         self.assertEqual(cwriter.manifest_text(),
521                          """. 4ace875ffdc6824a04950f06858f4465+22 0:8:basefile
522 ./subdir 4ace875ffdc6824a04950f06858f4465+22 8:14:subfile\n""")
523
524     def test_write_directory_tree_with_limited_recursion(self):
525         cwriter = arvados.CollectionWriter(self.api_client)
526         cwriter.write_directory_tree(
527             self.build_directory_tree(['f1', 'd1/f2', 'd1/d2/f3']),
528             max_manifest_depth=1)
529         self.assertEqual(cwriter.manifest_text(),
530                          """. bd19836ddb62c11c55ab251ccaca5645+2 0:2:f1
531 ./d1 50170217e5b04312024aa5cd42934494+13 8:5:f2
532 ./d1/d2 50170217e5b04312024aa5cd42934494+13 0:8:f3\n""")
533
534     def test_write_one_file(self):
535         cwriter = arvados.CollectionWriter(self.api_client)
536         with self.make_test_file() as testfile:
537             cwriter.write_file(testfile.name)
538             self.assertEqual(
539                 cwriter.manifest_text(),
540                 ". 098f6bcd4621d373cade4e832627b4f6+4 0:4:{}\n".format(
541                     os.path.basename(testfile.name)))
542
543     def test_write_named_file(self):
544         cwriter = arvados.CollectionWriter(self.api_client)
545         with self.make_test_file() as testfile:
546             cwriter.write_file(testfile.name, 'foo')
547             self.assertEqual(cwriter.manifest_text(),
548                              ". 098f6bcd4621d373cade4e832627b4f6+4 0:4:foo\n")
549
550     def test_write_multiple_files(self):
551         cwriter = arvados.CollectionWriter(self.api_client)
552         for letter in 'ABC':
553             with self.make_test_file(letter) as testfile:
554                 cwriter.write_file(testfile.name, letter)
555         self.assertEqual(
556             cwriter.manifest_text(),
557             ". 902fbdd2b1df0c4f70b4a5d23525e932+3 0:1:A 1:1:B 2:1:C\n")
558
559     def test_basic_resume(self):
560         cwriter = TestResumableWriter()
561         with self.make_test_file() as testfile:
562             cwriter.write_file(testfile.name, 'test')
563             resumed = TestResumableWriter.from_state(cwriter.current_state())
564         self.assertEquals(cwriter.manifest_text(), resumed.manifest_text(),
565                           "resumed CollectionWriter had different manifest")
566
567     def test_resume_fails_when_missing_dependency(self):
568         cwriter = TestResumableWriter()
569         with self.make_test_file() as testfile:
570             cwriter.write_file(testfile.name, 'test')
571         self.assertRaises(arvados.errors.StaleWriterStateError,
572                           TestResumableWriter.from_state,
573                           cwriter.current_state())
574
575     def test_resume_fails_when_dependency_mtime_changed(self):
576         cwriter = TestResumableWriter()
577         with self.make_test_file() as testfile:
578             cwriter.write_file(testfile.name, 'test')
579             os.utime(testfile.name, (0, 0))
580             self.assertRaises(arvados.errors.StaleWriterStateError,
581                               TestResumableWriter.from_state,
582                               cwriter.current_state())
583
584     def test_resume_fails_when_dependency_is_nonfile(self):
585         cwriter = TestResumableWriter()
586         cwriter.write_file('/dev/null', 'empty')
587         self.assertRaises(arvados.errors.StaleWriterStateError,
588                           TestResumableWriter.from_state,
589                           cwriter.current_state())
590
591     def test_resume_fails_when_dependency_size_changed(self):
592         cwriter = TestResumableWriter()
593         with self.make_test_file() as testfile:
594             cwriter.write_file(testfile.name, 'test')
595             orig_mtime = os.fstat(testfile.fileno()).st_mtime
596             testfile.write('extra')
597             testfile.flush()
598             os.utime(testfile.name, (orig_mtime, orig_mtime))
599             self.assertRaises(arvados.errors.StaleWriterStateError,
600                               TestResumableWriter.from_state,
601                               cwriter.current_state())
602
603     def test_resume_fails_with_expired_locator(self):
604         cwriter = TestResumableWriter()
605         state = cwriter.current_state()
606         # Add an expired locator to the state.
607         state['_current_stream_locators'].append(''.join([
608                     'a' * 32, '+1+A', 'b' * 40, '@', '10000000']))
609         self.assertRaises(arvados.errors.StaleWriterStateError,
610                           TestResumableWriter.from_state, state)
611
612     def test_arbitrary_objects_not_resumable(self):
613         cwriter = TestResumableWriter()
614         with open('/dev/null') as badfile:
615             self.assertRaises(arvados.errors.AssertionError,
616                               cwriter.write_file, badfile)
617
618     def test_arbitrary_writes_not_resumable(self):
619         cwriter = TestResumableWriter()
620         self.assertRaises(arvados.errors.AssertionError,
621                           cwriter.write, "badtext")
622
623
624 class CollectionTestMixin(object):
625     PROXY_RESPONSE = {
626         'items_available': 1,
627         'items': [{
628                 'uuid': 'zzzzz-bi6l4-mockproxy012345',
629                 'owner_uuid': 'zzzzz-tpzed-mockowner012345',
630                 'service_host': tutil.TEST_HOST,
631                 'service_port': 65535,
632                 'service_ssl_flag': True,
633                 'service_type': 'proxy',
634                 }]}
635     API_COLLECTIONS = run_test_server.fixture('collections')
636     DEFAULT_COLLECTION = API_COLLECTIONS['foo_file']
637     DEFAULT_DATA_HASH = DEFAULT_COLLECTION['portable_data_hash']
638     DEFAULT_MANIFEST = DEFAULT_COLLECTION['manifest_text']
639     DEFAULT_UUID = DEFAULT_COLLECTION['uuid']
640
641     def _mock_api_call(self, mock_method, code, body):
642         mock_method = mock_method().execute
643         if code == 200:
644             mock_method.return_value = body
645         else:
646             mock_method.side_effect = arvados.errors.ApiError(
647                 tutil.fake_httplib2_response(code), "{}")
648
649     def mock_keep_services(self, api_mock, code, body):
650         self._mock_api_call(api_mock.keep_services().accessible, code, body)
651
652     def api_client_mock(self, code=200):
653         client = mock.MagicMock(name='api_client')
654         self.mock_keep_services(client, code, self.PROXY_RESPONSE)
655         return client
656
657
658 @tutil.skip_sleep
659 class CollectionReaderTestCase(unittest.TestCase, CollectionTestMixin):
660     def mock_get_collection(self, api_mock, code, body):
661         body = self.API_COLLECTIONS.get(body)
662         self._mock_api_call(api_mock.collections().get, code, body)
663
664     def api_client_mock(self, code=200):
665         client = super(CollectionReaderTestCase, self).api_client_mock(code)
666         self.mock_get_collection(client, code, 'foo_file')
667         return client
668
669     def test_init_no_default_retries(self):
670         client = self.api_client_mock(200)
671         reader = arvados.CollectionReader(self.DEFAULT_UUID, api_client=client)
672         reader.manifest_text()
673         client.collections().get().execute.assert_called_with(num_retries=0)
674
675     def test_uuid_init_success(self):
676         client = self.api_client_mock(200)
677         reader = arvados.CollectionReader(self.DEFAULT_UUID, api_client=client,
678                                           num_retries=3)
679         self.assertEqual(self.DEFAULT_COLLECTION['manifest_text'],
680                          reader.manifest_text())
681         client.collections().get().execute.assert_called_with(num_retries=3)
682
683     def test_uuid_init_failure_raises_api_error(self):
684         client = self.api_client_mock(500)
685         reader = arvados.CollectionReader(self.DEFAULT_UUID, api_client=client)
686         with self.assertRaises(arvados.errors.ApiError):
687             reader.manifest_text()
688
689     def test_locator_init(self):
690         client = self.api_client_mock(200)
691         # Ensure Keep will not return anything if asked.
692         with tutil.mock_responses(None, 404):
693             reader = arvados.CollectionReader(self.DEFAULT_DATA_HASH,
694                                               api_client=client)
695             self.assertEqual(self.DEFAULT_MANIFEST, reader.manifest_text())
696
697     def test_locator_init_falls_back_to_keep(self):
698         # Reading manifests from Keep is deprecated.  Feel free to
699         # remove this test when we remove the fallback.
700         client = self.api_client_mock(200)
701         self.mock_get_collection(client, 404, None)
702         with tutil.mock_responses(self.DEFAULT_MANIFEST, 200):
703             reader = arvados.CollectionReader(self.DEFAULT_DATA_HASH,
704                                               api_client=client, num_retries=3)
705             self.assertEqual(self.DEFAULT_MANIFEST, reader.manifest_text())
706
707     def test_init_num_retries_propagated(self):
708         # More of an integration test...
709         client = self.api_client_mock(200)
710         reader = arvados.CollectionReader(self.DEFAULT_UUID, api_client=client,
711                                           num_retries=3)
712         with tutil.mock_responses('foo', 500, 500, 200):
713             self.assertEqual('foo',
714                              ''.join(f.read(9) for f in reader.all_files()))
715
716
717 @tutil.skip_sleep
718 class CollectionWriterTestCase(unittest.TestCase, CollectionTestMixin):
719     def mock_keep(self, body, *codes, **headers):
720         headers.setdefault('x-keep-replicas-stored', 2)
721         return tutil.mock_responses(body, *codes, **headers)
722
723     def foo_writer(self, **kwargs):
724         api_client = self.api_client_mock()
725         writer = arvados.CollectionWriter(api_client, **kwargs)
726         writer.start_new_file('foo')
727         writer.write('foo')
728         return writer
729
730     def test_write_whole_collection(self):
731         writer = self.foo_writer()
732         with self.mock_keep(self.DEFAULT_DATA_HASH, 200, 200):
733             self.assertEqual(self.DEFAULT_DATA_HASH, writer.finish())
734
735     def test_write_no_default(self):
736         writer = self.foo_writer()
737         with self.mock_keep(None, 500):
738             with self.assertRaises(arvados.errors.KeepWriteError):
739                 writer.finish()
740
741     def test_write_whole_collection_through_retries(self):
742         writer = self.foo_writer(num_retries=2)
743         with self.mock_keep(self.DEFAULT_DATA_HASH,
744                             500, 500, 200, 500, 500, 200):
745             self.assertEqual(self.DEFAULT_DATA_HASH, writer.finish())
746
747     def test_flush_data_retries(self):
748         writer = self.foo_writer(num_retries=2)
749         foo_hash = self.DEFAULT_MANIFEST.split()[1]
750         with self.mock_keep(foo_hash, 500, 200):
751             writer.flush_data()
752         self.assertEqual(self.DEFAULT_MANIFEST, writer.manifest_text())
753
754
755 if __name__ == '__main__':
756     unittest.main()