sdk/python/tests/test_arv_put.py

   1 from __future__ import absolute_import
   2 from __future__ import division
   3 from future import standard_library
   4 standard_library.install_aliases()
   5 from builtins import str
   6 from builtins import range
   7 import apiclient
   8 import mock
   9 import os
  10 import pwd
  11 import re
  12 import shutil
  13 import subprocess
  14 import sys
  15 import tempfile
  16 import time
  17 import unittest
  18 import yaml
  19 import threading
  20 import hashlib
  21 import random
  22 import uuid
  23
  24 import arvados
  25 import arvados.commands.put as arv_put
  26 from . import arvados_testutil as tutil
  27
  28 from .arvados_testutil import ArvadosBaseTestCase, fake_httplib2_response
  29 from . import run_test_server
  30
  31 class ArvadosPutResumeCacheTest(ArvadosBaseTestCase):
  32     CACHE_ARGSET = [
  33         [],
  34         ['/dev/null'],
  35         ['/dev/null', '--filename', 'empty'],
  36         ['/tmp']
  37         ]
  38
  39     def tearDown(self):
  40         super(ArvadosPutResumeCacheTest, self).tearDown()
  41         try:
  42             self.last_cache.destroy()
  43         except AttributeError:
  44             pass
  45
  46     def cache_path_from_arglist(self, arglist):
  47         return arv_put.ResumeCache.make_path(arv_put.parse_arguments(arglist))
  48
  49     def test_cache_names_stable(self):
  50         for argset in self.CACHE_ARGSET:
  51             self.assertEqual(self.cache_path_from_arglist(argset),
  52                               self.cache_path_from_arglist(argset),
  53                               "cache name changed for {}".format(argset))
  54
  55     def test_cache_names_unique(self):
  56         results = []
  57         for argset in self.CACHE_ARGSET:
  58             path = self.cache_path_from_arglist(argset)
  59             self.assertNotIn(path, results)
  60             results.append(path)
  61
  62     def test_cache_names_simple(self):
  63         # The goal here is to make sure the filename doesn't use characters
  64         # reserved by the filesystem.  Feel free to adjust this regexp as
  65         # long as it still does that.
  66         bad_chars = re.compile(r'[^-\.\w]')
  67         for argset in self.CACHE_ARGSET:
  68             path = self.cache_path_from_arglist(argset)
  69             self.assertFalse(bad_chars.search(os.path.basename(path)),
  70                              "path too exotic: {}".format(path))
  71
  72     def test_cache_names_ignore_argument_order(self):
  73         self.assertEqual(
  74             self.cache_path_from_arglist(['a', 'b', 'c']),
  75             self.cache_path_from_arglist(['c', 'a', 'b']))
  76         self.assertEqual(
  77             self.cache_path_from_arglist(['-', '--filename', 'stdin']),
  78             self.cache_path_from_arglist(['--filename', 'stdin', '-']))
  79
  80     def test_cache_names_differ_for_similar_paths(self):
  81         # This test needs names at / that don't exist on the real filesystem.
  82         self.assertNotEqual(
  83             self.cache_path_from_arglist(['/_arvputtest1', '/_arvputtest2']),
  84             self.cache_path_from_arglist(['/_arvputtest1/_arvputtest2']))
  85
  86     def test_cache_names_ignore_irrelevant_arguments(self):
  87         # Workaround: parse_arguments bails on --filename with a directory.
  88         path1 = self.cache_path_from_arglist(['/tmp'])
  89         args = arv_put.parse_arguments(['/tmp'])
  90         args.filename = 'tmp'
  91         path2 = arv_put.ResumeCache.make_path(args)
  92         self.assertEqual(path1, path2,
  93                          "cache path considered --filename for directory")
  94         self.assertEqual(
  95             self.cache_path_from_arglist(['-']),
  96             self.cache_path_from_arglist(['-', '--max-manifest-depth', '1']),
  97             "cache path considered --max-manifest-depth for file")
  98
  99     def test_cache_names_treat_negative_manifest_depths_identically(self):
 100         base_args = ['/tmp', '--max-manifest-depth']
 101         self.assertEqual(
 102             self.cache_path_from_arglist(base_args + ['-1']),
 103             self.cache_path_from_arglist(base_args + ['-2']))
 104
 105     def test_cache_names_treat_stdin_consistently(self):
 106         self.assertEqual(
 107             self.cache_path_from_arglist(['-', '--filename', 'test']),
 108             self.cache_path_from_arglist(['/dev/stdin', '--filename', 'test']))
 109
 110     def test_cache_names_identical_for_synonymous_names(self):
 111         self.assertEqual(
 112             self.cache_path_from_arglist(['.']),
 113             self.cache_path_from_arglist([os.path.realpath('.')]))
 114         testdir = self.make_tmpdir()
 115         looplink = os.path.join(testdir, 'loop')
 116         os.symlink(testdir, looplink)
 117         self.assertEqual(
 118             self.cache_path_from_arglist([testdir]),
 119             self.cache_path_from_arglist([looplink]))
 120
 121     def test_cache_names_different_by_api_host(self):
 122         config = arvados.config.settings()
 123         orig_host = config.get('ARVADOS_API_HOST')
 124         try:
 125             name1 = self.cache_path_from_arglist(['.'])
 126             config['ARVADOS_API_HOST'] = 'x' + (orig_host or 'localhost')
 127             self.assertNotEqual(name1, self.cache_path_from_arglist(['.']))
 128         finally:
 129             if orig_host is None:
 130                 del config['ARVADOS_API_HOST']
 131             else:
 132                 config['ARVADOS_API_HOST'] = orig_host
 133
 134     @mock.patch('arvados.keep.KeepClient.head')
 135     def test_resume_cache_with_current_stream_locators(self, keep_client_head):
 136         keep_client_head.side_effect = [True]
 137         thing = {}
 138         thing['_current_stream_locators'] = ['098f6bcd4621d373cade4e832627b4f6+4', '1f253c60a2306e0ee12fb6ce0c587904+6']
 139         with tempfile.NamedTemporaryFile() as cachefile:
 140             self.last_cache = arv_put.ResumeCache(cachefile.name)
 141         self.last_cache.save(thing)
 142         self.last_cache.close()
 143         resume_cache = arv_put.ResumeCache(self.last_cache.filename)
 144         self.assertNotEqual(None, resume_cache)
 145
 146     @mock.patch('arvados.keep.KeepClient.head')
 147     def test_resume_cache_with_finished_streams(self, keep_client_head):
 148         keep_client_head.side_effect = [True]
 149         thing = {}
 150         thing['_finished_streams'] = [['.', ['098f6bcd4621d373cade4e832627b4f6+4', '1f253c60a2306e0ee12fb6ce0c587904+6']]]
 151         with tempfile.NamedTemporaryFile() as cachefile:
 152             self.last_cache = arv_put.ResumeCache(cachefile.name)
 153         self.last_cache.save(thing)
 154         self.last_cache.close()
 155         resume_cache = arv_put.ResumeCache(self.last_cache.filename)
 156         self.assertNotEqual(None, resume_cache)
 157
 158     @mock.patch('arvados.keep.KeepClient.head')
 159     def test_resume_cache_with_finished_streams_error_on_head(self, keep_client_head):
 160         keep_client_head.side_effect = Exception('Locator not found')
 161         thing = {}
 162         thing['_finished_streams'] = [['.', ['098f6bcd4621d373cade4e832627b4f6+4', '1f253c60a2306e0ee12fb6ce0c587904+6']]]
 163         with tempfile.NamedTemporaryFile() as cachefile:
 164             self.last_cache = arv_put.ResumeCache(cachefile.name)
 165         self.last_cache.save(thing)
 166         self.last_cache.close()
 167         resume_cache = arv_put.ResumeCache(self.last_cache.filename)
 168         self.assertNotEqual(None, resume_cache)
 169         self.assertRaises(None, resume_cache.check_cache())
 170
 171     def test_basic_cache_storage(self):
 172         thing = ['test', 'list']
 173         with tempfile.NamedTemporaryFile() as cachefile:
 174             self.last_cache = arv_put.ResumeCache(cachefile.name)
 175         self.last_cache.save(thing)
 176         self.assertEqual(thing, self.last_cache.load())
 177
 178     def test_empty_cache(self):
 179         with tempfile.NamedTemporaryFile() as cachefile:
 180             cache = arv_put.ResumeCache(cachefile.name)
 181         self.assertRaises(ValueError, cache.load)
 182
 183     def test_cache_persistent(self):
 184         thing = ['test', 'list']
 185         path = os.path.join(self.make_tmpdir(), 'cache')
 186         cache = arv_put.ResumeCache(path)
 187         cache.save(thing)
 188         cache.close()
 189         self.last_cache = arv_put.ResumeCache(path)
 190         self.assertEqual(thing, self.last_cache.load())
 191
 192     def test_multiple_cache_writes(self):
 193         thing = ['short', 'list']
 194         with tempfile.NamedTemporaryFile() as cachefile:
 195             self.last_cache = arv_put.ResumeCache(cachefile.name)
 196         # Start writing an object longer than the one we test, to make
 197         # sure the cache file gets truncated.
 198         self.last_cache.save(['long', 'long', 'list'])
 199         self.last_cache.save(thing)
 200         self.assertEqual(thing, self.last_cache.load())
 201
 202     def test_cache_is_locked(self):
 203         with tempfile.NamedTemporaryFile() as cachefile:
 204             cache = arv_put.ResumeCache(cachefile.name)
 205             self.assertRaises(arv_put.ResumeCacheConflict,
 206                               arv_put.ResumeCache, cachefile.name)
 207
 208     def test_cache_stays_locked(self):
 209         with tempfile.NamedTemporaryFile() as cachefile:
 210             self.last_cache = arv_put.ResumeCache(cachefile.name)
 211             path = cachefile.name
 212         self.last_cache.save('test')
 213         self.assertRaises(arv_put.ResumeCacheConflict,
 214                           arv_put.ResumeCache, path)
 215
 216     def test_destroy_cache(self):
 217         cachefile = tempfile.NamedTemporaryFile(delete=False)
 218         try:
 219             cache = arv_put.ResumeCache(cachefile.name)
 220             cache.save('test')
 221             cache.destroy()
 222             try:
 223                 arv_put.ResumeCache(cachefile.name)
 224             except arv_put.ResumeCacheConflict:
 225                 self.fail("could not load cache after destroying it")
 226             self.assertRaises(ValueError, cache.load)
 227         finally:
 228             if os.path.exists(cachefile.name):
 229                 os.unlink(cachefile.name)
 230
 231     def test_restart_cache(self):
 232         path = os.path.join(self.make_tmpdir(), 'cache')
 233         cache = arv_put.ResumeCache(path)
 234         cache.save('test')
 235         cache.restart()
 236         self.assertRaises(ValueError, cache.load)
 237         self.assertRaises(arv_put.ResumeCacheConflict,
 238                           arv_put.ResumeCache, path)
 239
 240
 241 class ArvPutUploadJobTest(run_test_server.TestCaseWithServers,
 242                           ArvadosBaseTestCase):
 243
 244     def setUp(self):
 245         super(ArvPutUploadJobTest, self).setUp()
 246         run_test_server.authorize_with('active')
 247         # Temp files creation
 248         self.tempdir = tempfile.mkdtemp()
 249         subdir = os.path.join(self.tempdir, 'subdir')
 250         os.mkdir(subdir)
 251         data = "x" * 1024 # 1 KB
 252         for i in range(1, 5):
 253             with open(os.path.join(self.tempdir, str(i)), 'w') as f:
 254                 f.write(data * i)
 255         with open(os.path.join(subdir, 'otherfile'), 'w') as f:
 256             f.write(data * 5)
 257         # Large temp file for resume test
 258         _, self.large_file_name = tempfile.mkstemp()
 259         fileobj = open(self.large_file_name, 'w')
 260         # Make sure to write just a little more than one block
 261         for _ in range((arvados.config.KEEP_BLOCK_SIZE>>20)+1):
 262             data = random.choice(['x', 'y', 'z']) * 1024 * 1024 # 1 MiB
 263             fileobj.write(data)
 264         fileobj.close()
 265         # Temp dir containing small files to be repacked
 266         self.small_files_dir = tempfile.mkdtemp()
 267         data = 'y' * 1024 * 1024 # 1 MB
 268         for i in range(1, 70):
 269             with open(os.path.join(self.small_files_dir, str(i)), 'w') as f:
 270                 f.write(data + str(i))
 271         self.arvfile_write = getattr(arvados.arvfile.ArvadosFileWriter, 'write')
 272         # Temp dir to hold a symlink to other temp dir
 273         self.tempdir_with_symlink = tempfile.mkdtemp()
 274         os.symlink(self.tempdir, os.path.join(self.tempdir_with_symlink, 'linkeddir'))
 275         os.symlink(os.path.join(self.tempdir, '1'),
 276                    os.path.join(self.tempdir_with_symlink, 'linkedfile'))
 277
 278     def tearDown(self):
 279         super(ArvPutUploadJobTest, self).tearDown()
 280         shutil.rmtree(self.tempdir)
 281         os.unlink(self.large_file_name)
 282         shutil.rmtree(self.small_files_dir)
 283         shutil.rmtree(self.tempdir_with_symlink)
 284
 285     def test_symlinks_are_followed_by_default(self):
 286         cwriter = arv_put.ArvPutUploadJob([self.tempdir_with_symlink])
 287         cwriter.start(save_collection=False)
 288         self.assertIn('linkeddir', cwriter.manifest_text())
 289         self.assertIn('linkedfile', cwriter.manifest_text())
 290         cwriter.destroy_cache()
 291
 292     def test_symlinks_are_not_followed_when_requested(self):
 293         cwriter = arv_put.ArvPutUploadJob([self.tempdir_with_symlink],
 294                                           follow_links=False)
 295         cwriter.start(save_collection=False)
 296         self.assertNotIn('linkeddir', cwriter.manifest_text())
 297         self.assertNotIn('linkedfile', cwriter.manifest_text())
 298         cwriter.destroy_cache()
 299
 300     def test_passing_nonexistant_path_raise_exception(self):
 301         uuid_str = str(uuid.uuid4())
 302         with self.assertRaises(arv_put.PathDoesNotExistError):
 303             cwriter = arv_put.ArvPutUploadJob(["/this/path/does/not/exist/{}".format(uuid_str)])
 304
 305     def test_writer_works_without_cache(self):
 306         cwriter = arv_put.ArvPutUploadJob(['/dev/null'], resume=False)
 307         cwriter.start(save_collection=False)
 308         self.assertEqual(". d41d8cd98f00b204e9800998ecf8427e+0 0:0:null\n", cwriter.manifest_text())
 309
 310     def test_writer_works_with_cache(self):
 311         with tempfile.NamedTemporaryFile() as f:
 312             f.write(b'foo')
 313             f.flush()
 314             cwriter = arv_put.ArvPutUploadJob([f.name])
 315             cwriter.start(save_collection=False)
 316             self.assertEqual(0, cwriter.bytes_skipped)
 317             self.assertEqual(3, cwriter.bytes_written)
 318             # Don't destroy the cache, and start another upload
 319             cwriter_new = arv_put.ArvPutUploadJob([f.name])
 320             cwriter_new.start(save_collection=False)
 321             cwriter_new.destroy_cache()
 322             self.assertEqual(3, cwriter_new.bytes_skipped)
 323             self.assertEqual(3, cwriter_new.bytes_written)
 324
 325     def make_progress_tester(self):
 326         progression = []
 327         def record_func(written, expected):
 328             progression.append((written, expected))
 329         return progression, record_func
 330
 331     def test_progress_reporting(self):
 332         with tempfile.NamedTemporaryFile() as f:
 333             f.write(b'foo')
 334             f.flush()
 335             for expect_count in (None, 8):
 336                 progression, reporter = self.make_progress_tester()
 337                 cwriter = arv_put.ArvPutUploadJob([f.name],
 338                                                   reporter=reporter)
 339                 cwriter.bytes_expected = expect_count
 340                 cwriter.start(save_collection=False)
 341                 cwriter.destroy_cache()
 342                 self.assertIn((3, expect_count), progression)
 343
 344     def test_writer_upload_directory(self):
 345         cwriter = arv_put.ArvPutUploadJob([self.tempdir])
 346         cwriter.start(save_collection=False)
 347         cwriter.destroy_cache()
 348         self.assertEqual(1024*(1+2+3+4+5), cwriter.bytes_written)
 349
 350     def test_resume_large_file_upload(self):
 351         def wrapped_write(*args, **kwargs):
 352             data = args[1]
 353             # Exit only on last block
 354             if len(data) < arvados.config.KEEP_BLOCK_SIZE:
 355                 # Simulate a checkpoint before quitting. Ensure block commit.
 356                 self.writer._update(final=True)
 357                 raise SystemExit("Simulated error")
 358             return self.arvfile_write(*args, **kwargs)
 359
 360         with mock.patch('arvados.arvfile.ArvadosFileWriter.write',
 361                         autospec=True) as mocked_write:
 362             mocked_write.side_effect = wrapped_write
 363             writer = arv_put.ArvPutUploadJob([self.large_file_name],
 364                                              replication_desired=1)
 365             # We'll be accessing from inside the wrapper
 366             self.writer = writer
 367             with self.assertRaises(SystemExit):
 368                 writer.start(save_collection=False)
 369             # Confirm that the file was partially uploaded
 370             self.assertGreater(writer.bytes_written, 0)
 371             self.assertLess(writer.bytes_written,
 372                             os.path.getsize(self.large_file_name))
 373         # Retry the upload
 374         writer2 = arv_put.ArvPutUploadJob([self.large_file_name],
 375                                           replication_desired=1)
 376         writer2.start(save_collection=False)
 377         self.assertEqual(writer.bytes_written + writer2.bytes_written - writer2.bytes_skipped,
 378                          os.path.getsize(self.large_file_name))
 379         writer2.destroy_cache()
 380         del(self.writer)
 381
 382     # Test for bug #11002
 383     def test_graceful_exit_while_repacking_small_blocks(self):
 384         def wrapped_commit(*args, **kwargs):
 385             raise SystemExit("Simulated error")
 386
 387         with mock.patch('arvados.arvfile._BlockManager.commit_bufferblock',
 388                         autospec=True) as mocked_commit:
 389             mocked_commit.side_effect = wrapped_commit
 390             # Upload a little more than 1 block, wrapped_commit will make the first block
 391             # commit to fail.
 392             # arv-put should not exit with an exception by trying to commit the collection
 393             # as it's in an inconsistent state.
 394             writer = arv_put.ArvPutUploadJob([self.small_files_dir],
 395                                              replication_desired=1)
 396             try:
 397                 with self.assertRaises(SystemExit):
 398                     writer.start(save_collection=False)
 399             except arvados.arvfile.UnownedBlockError:
 400                 self.fail("arv-put command is trying to use a corrupted BlockManager. See https://dev.arvados.org/issues/11002")
 401         writer.destroy_cache()
 402
 403     def test_no_resume_when_asked(self):
 404         def wrapped_write(*args, **kwargs):
 405             data = args[1]
 406             # Exit only on last block
 407             if len(data) < arvados.config.KEEP_BLOCK_SIZE:
 408                 # Simulate a checkpoint before quitting.
 409                 self.writer._update()
 410                 raise SystemExit("Simulated error")
 411             return self.arvfile_write(*args, **kwargs)
 412
 413         with mock.patch('arvados.arvfile.ArvadosFileWriter.write',
 414                         autospec=True) as mocked_write:
 415             mocked_write.side_effect = wrapped_write
 416             writer = arv_put.ArvPutUploadJob([self.large_file_name],
 417                                              replication_desired=1)
 418             # We'll be accessing from inside the wrapper
 419             self.writer = writer
 420             with self.assertRaises(SystemExit):
 421                 writer.start(save_collection=False)
 422             # Confirm that the file was partially uploaded
 423             self.assertGreater(writer.bytes_written, 0)
 424             self.assertLess(writer.bytes_written,
 425                             os.path.getsize(self.large_file_name))
 426         # Retry the upload, this time without resume
 427         writer2 = arv_put.ArvPutUploadJob([self.large_file_name],
 428                                           replication_desired=1,
 429                                           resume=False)
 430         writer2.start(save_collection=False)
 431         self.assertEqual(writer2.bytes_skipped, 0)
 432         self.assertEqual(writer2.bytes_written,
 433                          os.path.getsize(self.large_file_name))
 434         writer2.destroy_cache()
 435         del(self.writer)
 436
 437     def test_no_resume_when_no_cache(self):
 438         def wrapped_write(*args, **kwargs):
 439             data = args[1]
 440             # Exit only on last block
 441             if len(data) < arvados.config.KEEP_BLOCK_SIZE:
 442                 # Simulate a checkpoint before quitting.
 443                 self.writer._update()
 444                 raise SystemExit("Simulated error")
 445             return self.arvfile_write(*args, **kwargs)
 446
 447         with mock.patch('arvados.arvfile.ArvadosFileWriter.write',
 448                         autospec=True) as mocked_write:
 449             mocked_write.side_effect = wrapped_write
 450             writer = arv_put.ArvPutUploadJob([self.large_file_name],
 451                                              replication_desired=1)
 452             # We'll be accessing from inside the wrapper
 453             self.writer = writer
 454             with self.assertRaises(SystemExit):
 455                 writer.start(save_collection=False)
 456             # Confirm that the file was partially uploaded
 457             self.assertGreater(writer.bytes_written, 0)
 458             self.assertLess(writer.bytes_written,
 459                             os.path.getsize(self.large_file_name))
 460         # Retry the upload, this time without cache usage
 461         writer2 = arv_put.ArvPutUploadJob([self.large_file_name],
 462                                           replication_desired=1,
 463                                           resume=False,
 464                                           use_cache=False)
 465         writer2.start(save_collection=False)
 466         self.assertEqual(writer2.bytes_skipped, 0)
 467         self.assertEqual(writer2.bytes_written,
 468                          os.path.getsize(self.large_file_name))
 469         writer2.destroy_cache()
 470         del(self.writer)
 471
 472     def test_dry_run_feature(self):
 473         def wrapped_write(*args, **kwargs):
 474             data = args[1]
 475             # Exit only on last block
 476             if len(data) < arvados.config.KEEP_BLOCK_SIZE:
 477                 # Simulate a checkpoint before quitting.
 478                 self.writer._update()
 479                 raise SystemExit("Simulated error")
 480             return self.arvfile_write(*args, **kwargs)
 481
 482         with mock.patch('arvados.arvfile.ArvadosFileWriter.write',
 483                         autospec=True) as mocked_write:
 484             mocked_write.side_effect = wrapped_write
 485             writer = arv_put.ArvPutUploadJob([self.large_file_name],
 486                                              replication_desired=1)
 487             # We'll be accessing from inside the wrapper
 488             self.writer = writer
 489             with self.assertRaises(SystemExit):
 490                 writer.start(save_collection=False)
 491             # Confirm that the file was partially uploaded
 492             self.assertGreater(writer.bytes_written, 0)
 493             self.assertLess(writer.bytes_written,
 494                             os.path.getsize(self.large_file_name))
 495         with self.assertRaises(arv_put.ArvPutUploadIsPending):
 496             # Retry the upload using dry_run to check if there is a pending upload
 497             writer2 = arv_put.ArvPutUploadJob([self.large_file_name],
 498                                               replication_desired=1,
 499                                               dry_run=True)
 500         # Complete the pending upload
 501         writer3 = arv_put.ArvPutUploadJob([self.large_file_name],
 502                                           replication_desired=1)
 503         writer3.start(save_collection=False)
 504         with self.assertRaises(arv_put.ArvPutUploadNotPending):
 505             # Confirm there's no pending upload with dry_run=True
 506             writer4 = arv_put.ArvPutUploadJob([self.large_file_name],
 507                                               replication_desired=1,
 508                                               dry_run=True)
 509         # Test obvious cases
 510         with self.assertRaises(arv_put.ArvPutUploadIsPending):
 511             arv_put.ArvPutUploadJob([self.large_file_name],
 512                                     replication_desired=1,
 513                                     dry_run=True,
 514                                     resume=False,
 515                                     use_cache=False)
 516         with self.assertRaises(arv_put.ArvPutUploadIsPending):
 517             arv_put.ArvPutUploadJob([self.large_file_name],
 518                                     replication_desired=1,
 519                                     dry_run=True,
 520                                     resume=False)
 521         del(self.writer)
 522
 523 class ArvadosExpectedBytesTest(ArvadosBaseTestCase):
 524     TEST_SIZE = os.path.getsize(__file__)
 525
 526     def test_expected_bytes_for_file(self):
 527         writer = arv_put.ArvPutUploadJob([__file__])
 528         self.assertEqual(self.TEST_SIZE,
 529                          writer.bytes_expected)
 530
 531     def test_expected_bytes_for_tree(self):
 532         tree = self.make_tmpdir()
 533         shutil.copyfile(__file__, os.path.join(tree, 'one'))
 534         shutil.copyfile(__file__, os.path.join(tree, 'two'))
 535
 536         writer = arv_put.ArvPutUploadJob([tree])
 537         self.assertEqual(self.TEST_SIZE * 2,
 538                          writer.bytes_expected)
 539         writer = arv_put.ArvPutUploadJob([tree, __file__])
 540         self.assertEqual(self.TEST_SIZE * 3,
 541                          writer.bytes_expected)
 542
 543     def test_expected_bytes_for_device(self):
 544         writer = arv_put.ArvPutUploadJob(['/dev/null'])
 545         self.assertIsNone(writer.bytes_expected)
 546         writer = arv_put.ArvPutUploadJob([__file__, '/dev/null'])
 547         self.assertIsNone(writer.bytes_expected)
 548
 549
 550 class ArvadosPutReportTest(ArvadosBaseTestCase):
 551     def test_machine_progress(self):
 552         for count, total in [(0, 1), (0, None), (1, None), (235, 9283)]:
 553             expect = ": {} written {} total\n".format(
 554                 count, -1 if (total is None) else total)
 555             self.assertTrue(
 556                 arv_put.machine_progress(count, total).endswith(expect))
 557
 558     def test_known_human_progress(self):
 559         for count, total in [(0, 1), (2, 4), (45, 60)]:
 560             expect = '{:.1%}'.format(1.0*count/total)
 561             actual = arv_put.human_progress(count, total)
 562             self.assertTrue(actual.startswith('\r'))
 563             self.assertIn(expect, actual)
 564
 565     def test_unknown_human_progress(self):
 566         for count in [1, 20, 300, 4000, 50000]:
 567             self.assertTrue(re.search(r'\b{}\b'.format(count),
 568                                       arv_put.human_progress(count, None)))
 569
 570
 571 class ArvadosPutTest(run_test_server.TestCaseWithServers,
 572                      ArvadosBaseTestCase,
 573                      tutil.VersionChecker):
 574     MAIN_SERVER = {}
 575     Z_UUID = 'zzzzz-zzzzz-zzzzzzzzzzzzzzz'
 576
 577     def call_main_with_args(self, args):
 578         self.main_stdout = tutil.StringIO()
 579         self.main_stderr = tutil.StringIO()
 580         return arv_put.main(args, self.main_stdout, self.main_stderr)
 581
 582     def call_main_on_test_file(self, args=[]):
 583         with self.make_test_file() as testfile:
 584             path = testfile.name
 585             self.call_main_with_args(['--stream', '--no-progress'] + args + [path])
 586         self.assertTrue(
 587             os.path.exists(os.path.join(os.environ['KEEP_LOCAL_STORE'],
 588                                         '098f6bcd4621d373cade4e832627b4f6')),
 589             "did not find file stream in Keep store")
 590
 591     def setUp(self):
 592         super(ArvadosPutTest, self).setUp()
 593         run_test_server.authorize_with('active')
 594         arv_put.api_client = None
 595
 596     def tearDown(self):
 597         for outbuf in ['main_stdout', 'main_stderr']:
 598             if hasattr(self, outbuf):
 599                 getattr(self, outbuf).close()
 600                 delattr(self, outbuf)
 601         super(ArvadosPutTest, self).tearDown()
 602
 603     def test_version_argument(self):
 604         with tutil.redirected_streams(
 605                 stdout=tutil.StringIO, stderr=tutil.StringIO) as (out, err):
 606             with self.assertRaises(SystemExit):
 607                 self.call_main_with_args(['--version'])
 608         self.assertVersionOutput(out, err)
 609
 610     def test_simple_file_put(self):
 611         self.call_main_on_test_file()
 612
 613     def test_put_with_unwriteable_cache_dir(self):
 614         orig_cachedir = arv_put.ResumeCache.CACHE_DIR
 615         cachedir = self.make_tmpdir()
 616         os.chmod(cachedir, 0o0)
 617         arv_put.ResumeCache.CACHE_DIR = cachedir
 618         try:
 619             self.call_main_on_test_file()
 620         finally:
 621             arv_put.ResumeCache.CACHE_DIR = orig_cachedir
 622             os.chmod(cachedir, 0o700)
 623
 624     def test_put_with_unwritable_cache_subdir(self):
 625         orig_cachedir = arv_put.ResumeCache.CACHE_DIR
 626         cachedir = self.make_tmpdir()
 627         os.chmod(cachedir, 0o0)
 628         arv_put.ResumeCache.CACHE_DIR = os.path.join(cachedir, 'cachedir')
 629         try:
 630             self.call_main_on_test_file()
 631         finally:
 632             arv_put.ResumeCache.CACHE_DIR = orig_cachedir
 633             os.chmod(cachedir, 0o700)
 634
 635     def test_put_block_replication(self):
 636         self.call_main_on_test_file()
 637         with mock.patch('arvados.collection.KeepClient.local_store_put') as put_mock:
 638             put_mock.return_value = 'acbd18db4cc2f85cedef654fccc4a4d8+3'
 639             self.call_main_on_test_file(['--replication', '1'])
 640             self.call_main_on_test_file(['--replication', '4'])
 641             self.call_main_on_test_file(['--replication', '5'])
 642             self.assertEqual(
 643                 [x[-1].get('copies') for x in put_mock.call_args_list],
 644                 [1, 4, 5])
 645
 646     def test_normalize(self):
 647         testfile1 = self.make_test_file()
 648         testfile2 = self.make_test_file()
 649         test_paths = [testfile1.name, testfile2.name]
 650         # Reverse-sort the paths, so normalization must change their order.
 651         test_paths.sort(reverse=True)
 652         self.call_main_with_args(['--stream', '--no-progress', '--normalize'] +
 653                                  test_paths)
 654         manifest = self.main_stdout.getvalue()
 655         # Assert the second file we specified appears first in the manifest.
 656         file_indices = [manifest.find(':' + os.path.basename(path))
 657                         for path in test_paths]
 658         self.assertGreater(*file_indices)
 659
 660     def test_error_name_without_collection(self):
 661         self.assertRaises(SystemExit, self.call_main_with_args,
 662                           ['--name', 'test without Collection',
 663                            '--stream', '/dev/null'])
 664
 665     def test_error_when_project_not_found(self):
 666         self.assertRaises(SystemExit,
 667                           self.call_main_with_args,
 668                           ['--project-uuid', self.Z_UUID])
 669
 670     def test_error_bad_project_uuid(self):
 671         self.assertRaises(SystemExit,
 672                           self.call_main_with_args,
 673                           ['--project-uuid', self.Z_UUID, '--stream'])
 674
 675     def test_api_error_handling(self):
 676         coll_save_mock = mock.Mock(name='arv.collection.Collection().save_new()')
 677         coll_save_mock.side_effect = arvados.errors.ApiError(
 678             fake_httplib2_response(403), b'{}')
 679         with mock.patch('arvados.collection.Collection.save_new',
 680                         new=coll_save_mock):
 681             with self.assertRaises(SystemExit) as exc_test:
 682                 self.call_main_with_args(['/dev/null'])
 683             self.assertLess(0, exc_test.exception.args[0])
 684             self.assertLess(0, coll_save_mock.call_count)
 685             self.assertEqual("", self.main_stdout.getvalue())
 686
 687
 688 class ArvPutIntegrationTest(run_test_server.TestCaseWithServers,
 689                             ArvadosBaseTestCase):
 690     def _getKeepServerConfig():
 691         for config_file, mandatory in [
 692                 ['application.yml', False], ['application.default.yml', True]]:
 693             path = os.path.join(run_test_server.SERVICES_SRC_DIR,
 694                                 "api", "config", config_file)
 695             if not mandatory and not os.path.exists(path):
 696                 continue
 697             with open(path) as f:
 698                 rails_config = yaml.load(f.read())
 699                 for config_section in ['test', 'common']:
 700                     try:
 701                         key = rails_config[config_section]["blob_signing_key"]
 702                     except (KeyError, TypeError):
 703                         pass
 704                     else:
 705                         return {'blob_signing_key': key,
 706                                 'enforce_permissions': True}
 707         return {'blog_signing_key': None, 'enforce_permissions': False}
 708
 709     MAIN_SERVER = {}
 710     KEEP_SERVER = _getKeepServerConfig()
 711     PROJECT_UUID = run_test_server.fixture('groups')['aproject']['uuid']
 712
 713     @classmethod
 714     def setUpClass(cls):
 715         super(ArvPutIntegrationTest, cls).setUpClass()
 716         cls.ENVIRON = os.environ.copy()
 717         cls.ENVIRON['PYTHONPATH'] = ':'.join(sys.path)
 718
 719     def setUp(self):
 720         super(ArvPutIntegrationTest, self).setUp()
 721         arv_put.api_client = None
 722
 723     def authorize_with(self, token_name):
 724         run_test_server.authorize_with(token_name)
 725         for v in ["ARVADOS_API_HOST",
 726                   "ARVADOS_API_HOST_INSECURE",
 727                   "ARVADOS_API_TOKEN"]:
 728             self.ENVIRON[v] = arvados.config.settings()[v]
 729         arv_put.api_client = arvados.api('v1')
 730
 731     def current_user(self):
 732         return arv_put.api_client.users().current().execute()
 733
 734     def test_check_real_project_found(self):
 735         self.authorize_with('active')
 736         self.assertTrue(arv_put.desired_project_uuid(arv_put.api_client, self.PROJECT_UUID, 0),
 737                         "did not correctly find test fixture project")
 738
 739     def test_check_error_finding_nonexistent_uuid(self):
 740         BAD_UUID = 'zzzzz-zzzzz-zzzzzzzzzzzzzzz'
 741         self.authorize_with('active')
 742         try:
 743             result = arv_put.desired_project_uuid(arv_put.api_client, BAD_UUID,
 744                                                   0)
 745         except ValueError as error:
 746             self.assertIn(BAD_UUID, str(error))
 747         else:
 748             self.assertFalse(result, "incorrectly found nonexistent project")
 749
 750     def test_check_error_finding_nonexistent_project(self):
 751         BAD_UUID = 'zzzzz-tpzed-zzzzzzzzzzzzzzz'
 752         self.authorize_with('active')
 753         with self.assertRaises(apiclient.errors.HttpError):
 754             result = arv_put.desired_project_uuid(arv_put.api_client, BAD_UUID,
 755                                                   0)
 756
 757     def test_short_put_from_stdin(self):
 758         # Have to run this as an integration test since arv-put can't
 759         # read from the tests' stdin.
 760         # arv-put usually can't stat(os.path.realpath('/dev/stdin')) in this
 761         # case, because the /proc entry is already gone by the time it tries.
 762         pipe = subprocess.Popen(
 763             [sys.executable, arv_put.__file__, '--stream'],
 764             stdin=subprocess.PIPE, stdout=subprocess.PIPE,
 765             stderr=subprocess.STDOUT, env=self.ENVIRON)
 766         pipe.stdin.write(b'stdin test\n')
 767         pipe.stdin.close()
 768         deadline = time.time() + 5
 769         while (pipe.poll() is None) and (time.time() < deadline):
 770             time.sleep(.1)
 771         returncode = pipe.poll()
 772         if returncode is None:
 773             pipe.terminate()
 774             self.fail("arv-put did not PUT from stdin within 5 seconds")
 775         elif returncode != 0:
 776             sys.stdout.write(pipe.stdout.read())
 777             self.fail("arv-put returned exit code {}".format(returncode))
 778         self.assertIn('4a9c8b735dce4b5fa3acf221a0b13628+11',
 779                       pipe.stdout.read().decode())
 780
 781     def test_ArvPutSignedManifest(self):
 782         # ArvPutSignedManifest runs "arv-put foo" and then attempts to get
 783         # the newly created manifest from the API server, testing to confirm
 784         # that the block locators in the returned manifest are signed.
 785         self.authorize_with('active')
 786
 787         # Before doing anything, demonstrate that the collection
 788         # we're about to create is not present in our test fixture.
 789         manifest_uuid = "00b4e9f40ac4dd432ef89749f1c01e74+47"
 790         with self.assertRaises(apiclient.errors.HttpError):
 791             notfound = arv_put.api_client.collections().get(
 792                 uuid=manifest_uuid).execute()
 793
 794         datadir = self.make_tmpdir()
 795         with open(os.path.join(datadir, "foo"), "w") as f:
 796             f.write("The quick brown fox jumped over the lazy dog")
 797         p = subprocess.Popen([sys.executable, arv_put.__file__,
 798                               os.path.join(datadir, 'foo')],
 799                              stdout=subprocess.PIPE,
 800                              stderr=subprocess.PIPE,
 801                              env=self.ENVIRON)
 802         (out, err) = p.communicate()
 803         self.assertRegex(err.decode(), r'INFO: Collection saved as ')
 804         self.assertEqual(p.returncode, 0)
 805
 806         # The manifest text stored in the API server under the same
 807         # manifest UUID must use signed locators.
 808         c = arv_put.api_client.collections().get(uuid=manifest_uuid).execute()
 809         self.assertRegex(
 810             c['manifest_text'],
 811             r'^\. 08a008a01d498c404b0c30852b39d3b8\+44\+A[0-9a-f]+@[0-9a-f]+ 0:44:foo\n')
 812
 813         os.remove(os.path.join(datadir, "foo"))
 814         os.rmdir(datadir)
 815
 816     def run_and_find_collection(self, text, extra_args=[]):
 817         self.authorize_with('active')
 818         pipe = subprocess.Popen(
 819             [sys.executable, arv_put.__file__] + extra_args,
 820             stdin=subprocess.PIPE, stdout=subprocess.PIPE,
 821             stderr=subprocess.PIPE, env=self.ENVIRON)
 822         stdout, stderr = pipe.communicate(text.encode())
 823         self.assertRegex(stderr.decode(), r'INFO: Collection (updated:|saved as)')
 824         search_key = ('portable_data_hash'
 825                       if '--portable-data-hash' in extra_args else 'uuid')
 826         collection_list = arvados.api('v1').collections().list(
 827             filters=[[search_key, '=', stdout.decode().strip()]]
 828         ).execute().get('items', [])
 829         self.assertEqual(1, len(collection_list))
 830         return collection_list[0]
 831
 832     def test_put_collection_with_later_update(self):
 833         tmpdir = self.make_tmpdir()
 834         with open(os.path.join(tmpdir, 'file1'), 'w') as f:
 835             f.write('Relaxing in basins at the end of inlets terminates the endless tests from the box')
 836         col = self.run_and_find_collection("", ['--no-progress', tmpdir])
 837         self.assertNotEqual(None, col['uuid'])
 838         # Add a new file to the directory
 839         with open(os.path.join(tmpdir, 'file2'), 'w') as f:
 840             f.write('The quick brown fox jumped over the lazy dog')
 841         updated_col = self.run_and_find_collection("", ['--no-progress', '--update-collection', col['uuid'], tmpdir])
 842         self.assertEqual(col['uuid'], updated_col['uuid'])
 843         # Get the manifest and check that the new file is being included
 844         c = arv_put.api_client.collections().get(uuid=updated_col['uuid']).execute()
 845         self.assertRegex(c['manifest_text'], r'^\..* .*:44:file2\n')
 846
 847     def test_upload_directory_reference_without_trailing_slash(self):
 848         tmpdir1 = self.make_tmpdir()
 849         tmpdir2 = self.make_tmpdir()
 850         with open(os.path.join(tmpdir1, 'foo'), 'w') as f:
 851             f.write('This is foo')
 852         with open(os.path.join(tmpdir2, 'bar'), 'w') as f:
 853             f.write('This is not foo')
 854         # Upload one directory and one file
 855         col = self.run_and_find_collection("", ['--no-progress',
 856                                                 tmpdir1,
 857                                                 os.path.join(tmpdir2, 'bar')])
 858         self.assertNotEqual(None, col['uuid'])
 859         c = arv_put.api_client.collections().get(uuid=col['uuid']).execute()
 860         # Check that 'foo' was written inside a subcollection
 861         # OTOH, 'bar' should have been directly uploaded on the root collection
 862         self.assertRegex(c['manifest_text'], r'^\. .*:15:bar\n\./.+ .*:11:foo\n')
 863
 864     def test_upload_directory_reference_with_trailing_slash(self):
 865         tmpdir1 = self.make_tmpdir()
 866         tmpdir2 = self.make_tmpdir()
 867         with open(os.path.join(tmpdir1, 'foo'), 'w') as f:
 868             f.write('This is foo')
 869         with open(os.path.join(tmpdir2, 'bar'), 'w') as f:
 870             f.write('This is not foo')
 871         # Upload one directory (with trailing slash) and one file
 872         col = self.run_and_find_collection("", ['--no-progress',
 873                                                 tmpdir1 + os.sep,
 874                                                 os.path.join(tmpdir2, 'bar')])
 875         self.assertNotEqual(None, col['uuid'])
 876         c = arv_put.api_client.collections().get(uuid=col['uuid']).execute()
 877         # Check that 'foo' and 'bar' were written at the same level
 878         self.assertRegex(c['manifest_text'], r'^\. .*:15:bar .*:11:foo\n')
 879
 880     def test_put_collection_with_high_redundancy(self):
 881         # Write empty data: we're not testing CollectionWriter, just
 882         # making sure collections.create tells the API server what our
 883         # desired replication level is.
 884         collection = self.run_and_find_collection("", ['--replication', '4'])
 885         self.assertEqual(4, collection['replication_desired'])
 886
 887     def test_put_collection_with_default_redundancy(self):
 888         collection = self.run_and_find_collection("")
 889         self.assertEqual(None, collection['replication_desired'])
 890
 891     def test_put_collection_with_unnamed_project_link(self):
 892         link = self.run_and_find_collection(
 893             "Test unnamed collection",
 894             ['--portable-data-hash', '--project-uuid', self.PROJECT_UUID])
 895         username = pwd.getpwuid(os.getuid()).pw_name
 896         self.assertRegex(
 897             link['name'],
 898             r'^Saved at .* by {}@'.format(re.escape(username)))
 899
 900     def test_put_collection_with_name_and_no_project(self):
 901         link_name = 'Test Collection Link in home project'
 902         collection = self.run_and_find_collection(
 903             "Test named collection in home project",
 904             ['--portable-data-hash', '--name', link_name])
 905         self.assertEqual(link_name, collection['name'])
 906         my_user_uuid = self.current_user()['uuid']
 907         self.assertEqual(my_user_uuid, collection['owner_uuid'])
 908
 909     def test_put_collection_with_named_project_link(self):
 910         link_name = 'Test auto Collection Link'
 911         collection = self.run_and_find_collection("Test named collection",
 912                                       ['--portable-data-hash',
 913                                        '--name', link_name,
 914                                        '--project-uuid', self.PROJECT_UUID])
 915         self.assertEqual(link_name, collection['name'])
 916
 917
 918 if __name__ == '__main__':
 919     unittest.main()