sdk/python/tests/test_arv_put.py

   1 # -*- coding: utf-8 -*-
   2
   3 # Copyright (C) The Arvados Authors. All rights reserved.
   4 #
   5 # SPDX-License-Identifier: Apache-2.0
   6
   7 from __future__ import absolute_import
   8 from __future__ import division
   9 from future import standard_library
  10 standard_library.install_aliases()
  11 from builtins import str
  12 from builtins import range
  13 from functools import partial
  14 import apiclient
  15 import ciso8601
  16 import datetime
  17 import hashlib
  18 import json
  19 import logging
  20 import mock
  21 import os
  22 import pwd
  23 import random
  24 import re
  25 import select
  26 import shutil
  27 import signal
  28 import subprocess
  29 import sys
  30 import tempfile
  31 import time
  32 import unittest
  33 import uuid
  34 import yaml
  35
  36 import arvados
  37 import arvados.commands.put as arv_put
  38 from . import arvados_testutil as tutil
  39
  40 from .arvados_testutil import ArvadosBaseTestCase, fake_httplib2_response
  41 from . import run_test_server
  42
  43 class ArvadosPutResumeCacheTest(ArvadosBaseTestCase):
  44     CACHE_ARGSET = [
  45         [],
  46         ['/dev/null'],
  47         ['/dev/null', '--filename', 'empty'],
  48         ['/tmp']
  49         ]
  50
  51     def tearDown(self):
  52         super(ArvadosPutResumeCacheTest, self).tearDown()
  53         try:
  54             self.last_cache.destroy()
  55         except AttributeError:
  56             pass
  57
  58     def cache_path_from_arglist(self, arglist):
  59         return arv_put.ResumeCache.make_path(arv_put.parse_arguments(arglist))
  60
  61     def test_cache_names_stable(self):
  62         for argset in self.CACHE_ARGSET:
  63             self.assertEqual(self.cache_path_from_arglist(argset),
  64                               self.cache_path_from_arglist(argset),
  65                               "cache name changed for {}".format(argset))
  66
  67     def test_cache_names_unique(self):
  68         results = []
  69         for argset in self.CACHE_ARGSET:
  70             path = self.cache_path_from_arglist(argset)
  71             self.assertNotIn(path, results)
  72             results.append(path)
  73
  74     def test_cache_names_simple(self):
  75         # The goal here is to make sure the filename doesn't use characters
  76         # reserved by the filesystem.  Feel free to adjust this regexp as
  77         # long as it still does that.
  78         bad_chars = re.compile(r'[^-\.\w]')
  79         for argset in self.CACHE_ARGSET:
  80             path = self.cache_path_from_arglist(argset)
  81             self.assertFalse(bad_chars.search(os.path.basename(path)),
  82                              "path too exotic: {}".format(path))
  83
  84     def test_cache_names_ignore_argument_order(self):
  85         self.assertEqual(
  86             self.cache_path_from_arglist(['a', 'b', 'c']),
  87             self.cache_path_from_arglist(['c', 'a', 'b']))
  88         self.assertEqual(
  89             self.cache_path_from_arglist(['-', '--filename', 'stdin']),
  90             self.cache_path_from_arglist(['--filename', 'stdin', '-']))
  91
  92     def test_cache_names_differ_for_similar_paths(self):
  93         # This test needs names at / that don't exist on the real filesystem.
  94         self.assertNotEqual(
  95             self.cache_path_from_arglist(['/_arvputtest1', '/_arvputtest2']),
  96             self.cache_path_from_arglist(['/_arvputtest1/_arvputtest2']))
  97
  98     def test_cache_names_ignore_irrelevant_arguments(self):
  99         # Workaround: parse_arguments bails on --filename with a directory.
 100         path1 = self.cache_path_from_arglist(['/tmp'])
 101         args = arv_put.parse_arguments(['/tmp'])
 102         args.filename = 'tmp'
 103         path2 = arv_put.ResumeCache.make_path(args)
 104         self.assertEqual(path1, path2,
 105                          "cache path considered --filename for directory")
 106         self.assertEqual(
 107             self.cache_path_from_arglist(['-']),
 108             self.cache_path_from_arglist(['-', '--max-manifest-depth', '1']),
 109             "cache path considered --max-manifest-depth for file")
 110
 111     def test_cache_names_treat_negative_manifest_depths_identically(self):
 112         base_args = ['/tmp', '--max-manifest-depth']
 113         self.assertEqual(
 114             self.cache_path_from_arglist(base_args + ['-1']),
 115             self.cache_path_from_arglist(base_args + ['-2']))
 116
 117     def test_cache_names_treat_stdin_consistently(self):
 118         self.assertEqual(
 119             self.cache_path_from_arglist(['-', '--filename', 'test']),
 120             self.cache_path_from_arglist(['/dev/stdin', '--filename', 'test']))
 121
 122     def test_cache_names_identical_for_synonymous_names(self):
 123         self.assertEqual(
 124             self.cache_path_from_arglist(['.']),
 125             self.cache_path_from_arglist([os.path.realpath('.')]))
 126         testdir = self.make_tmpdir()
 127         looplink = os.path.join(testdir, 'loop')
 128         os.symlink(testdir, looplink)
 129         self.assertEqual(
 130             self.cache_path_from_arglist([testdir]),
 131             self.cache_path_from_arglist([looplink]))
 132
 133     def test_cache_names_different_by_api_host(self):
 134         config = arvados.config.settings()
 135         orig_host = config.get('ARVADOS_API_HOST')
 136         try:
 137             name1 = self.cache_path_from_arglist(['.'])
 138             config['ARVADOS_API_HOST'] = 'x' + (orig_host or 'localhost')
 139             self.assertNotEqual(name1, self.cache_path_from_arglist(['.']))
 140         finally:
 141             if orig_host is None:
 142                 del config['ARVADOS_API_HOST']
 143             else:
 144                 config['ARVADOS_API_HOST'] = orig_host
 145
 146     @mock.patch('arvados.keep.KeepClient.head')
 147     def test_resume_cache_with_current_stream_locators(self, keep_client_head):
 148         keep_client_head.side_effect = [True]
 149         thing = {}
 150         thing['_current_stream_locators'] = ['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(self, keep_client_head):
 160         keep_client_head.side_effect = [True]
 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
 170     @mock.patch('arvados.keep.KeepClient.head')
 171     def test_resume_cache_with_finished_streams_error_on_head(self, keep_client_head):
 172         keep_client_head.side_effect = Exception('Locator not found')
 173         thing = {}
 174         thing['_finished_streams'] = [['.', ['098f6bcd4621d373cade4e832627b4f6+4', '1f253c60a2306e0ee12fb6ce0c587904+6']]]
 175         with tempfile.NamedTemporaryFile() as cachefile:
 176             self.last_cache = arv_put.ResumeCache(cachefile.name)
 177         self.last_cache.save(thing)
 178         self.last_cache.close()
 179         resume_cache = arv_put.ResumeCache(self.last_cache.filename)
 180         self.assertNotEqual(None, resume_cache)
 181         resume_cache.check_cache()
 182
 183     def test_basic_cache_storage(self):
 184         thing = ['test', 'list']
 185         with tempfile.NamedTemporaryFile() as cachefile:
 186             self.last_cache = arv_put.ResumeCache(cachefile.name)
 187         self.last_cache.save(thing)
 188         self.assertEqual(thing, self.last_cache.load())
 189
 190     def test_empty_cache(self):
 191         with tempfile.NamedTemporaryFile() as cachefile:
 192             cache = arv_put.ResumeCache(cachefile.name)
 193         self.assertRaises(ValueError, cache.load)
 194
 195     def test_cache_persistent(self):
 196         thing = ['test', 'list']
 197         path = os.path.join(self.make_tmpdir(), 'cache')
 198         cache = arv_put.ResumeCache(path)
 199         cache.save(thing)
 200         cache.close()
 201         self.last_cache = arv_put.ResumeCache(path)
 202         self.assertEqual(thing, self.last_cache.load())
 203
 204     def test_multiple_cache_writes(self):
 205         thing = ['short', 'list']
 206         with tempfile.NamedTemporaryFile() as cachefile:
 207             self.last_cache = arv_put.ResumeCache(cachefile.name)
 208         # Start writing an object longer than the one we test, to make
 209         # sure the cache file gets truncated.
 210         self.last_cache.save(['long', 'long', 'list'])
 211         self.last_cache.save(thing)
 212         self.assertEqual(thing, self.last_cache.load())
 213
 214     def test_cache_is_locked(self):
 215         with tempfile.NamedTemporaryFile() as cachefile:
 216             _ = arv_put.ResumeCache(cachefile.name)
 217             self.assertRaises(arv_put.ResumeCacheConflict,
 218                               arv_put.ResumeCache, cachefile.name)
 219
 220     def test_cache_stays_locked(self):
 221         with tempfile.NamedTemporaryFile() as cachefile:
 222             self.last_cache = arv_put.ResumeCache(cachefile.name)
 223             path = cachefile.name
 224         self.last_cache.save('test')
 225         self.assertRaises(arv_put.ResumeCacheConflict,
 226                           arv_put.ResumeCache, path)
 227
 228     def test_destroy_cache(self):
 229         cachefile = tempfile.NamedTemporaryFile(delete=False)
 230         try:
 231             cache = arv_put.ResumeCache(cachefile.name)
 232             cache.save('test')
 233             cache.destroy()
 234             try:
 235                 arv_put.ResumeCache(cachefile.name)
 236             except arv_put.ResumeCacheConflict:
 237                 self.fail("could not load cache after destroying it")
 238             self.assertRaises(ValueError, cache.load)
 239         finally:
 240             if os.path.exists(cachefile.name):
 241                 os.unlink(cachefile.name)
 242
 243     def test_restart_cache(self):
 244         path = os.path.join(self.make_tmpdir(), 'cache')
 245         cache = arv_put.ResumeCache(path)
 246         cache.save('test')
 247         cache.restart()
 248         self.assertRaises(ValueError, cache.load)
 249         self.assertRaises(arv_put.ResumeCacheConflict,
 250                           arv_put.ResumeCache, path)
 251
 252
 253 class ArvPutUploadJobTest(run_test_server.TestCaseWithServers,
 254                           ArvadosBaseTestCase):
 255
 256     def setUp(self):
 257         super(ArvPutUploadJobTest, self).setUp()
 258         run_test_server.authorize_with('active')
 259         # Temp files creation
 260         self.tempdir = tempfile.mkdtemp()
 261         subdir = os.path.join(self.tempdir, 'subdir')
 262         os.mkdir(subdir)
 263         data = "x" * 1024 # 1 KB
 264         for i in range(1, 5):
 265             with open(os.path.join(self.tempdir, str(i)), 'w') as f:
 266                 f.write(data * i)
 267         with open(os.path.join(subdir, 'otherfile'), 'w') as f:
 268             f.write(data * 5)
 269         # Large temp file for resume test
 270         _, self.large_file_name = tempfile.mkstemp()
 271         fileobj = open(self.large_file_name, 'w')
 272         # Make sure to write just a little more than one block
 273         for _ in range((arvados.config.KEEP_BLOCK_SIZE>>20)+1):
 274             data = random.choice(['x', 'y', 'z']) * 1024 * 1024 # 1 MiB
 275             fileobj.write(data)
 276         fileobj.close()
 277         # Temp dir containing small files to be repacked
 278         self.small_files_dir = tempfile.mkdtemp()
 279         data = 'y' * 1024 * 1024 # 1 MB
 280         for i in range(1, 70):
 281             with open(os.path.join(self.small_files_dir, str(i)), 'w') as f:
 282                 f.write(data + str(i))
 283         self.arvfile_write = getattr(arvados.arvfile.ArvadosFileWriter, 'write')
 284         # Temp dir to hold a symlink to other temp dir
 285         self.tempdir_with_symlink = tempfile.mkdtemp()
 286         os.symlink(self.tempdir, os.path.join(self.tempdir_with_symlink, 'linkeddir'))
 287         os.symlink(os.path.join(self.tempdir, '1'),
 288                    os.path.join(self.tempdir_with_symlink, 'linkedfile'))
 289
 290     def tearDown(self):
 291         super(ArvPutUploadJobTest, self).tearDown()
 292         shutil.rmtree(self.tempdir)
 293         os.unlink(self.large_file_name)
 294         shutil.rmtree(self.small_files_dir)
 295         shutil.rmtree(self.tempdir_with_symlink)
 296
 297     def test_symlinks_are_followed_by_default(self):
 298         self.assertTrue(os.path.islink(os.path.join(self.tempdir_with_symlink, 'linkeddir')))
 299         self.assertTrue(os.path.islink(os.path.join(self.tempdir_with_symlink, 'linkedfile')))
 300         cwriter = arv_put.ArvPutUploadJob([self.tempdir_with_symlink])
 301         cwriter.start(save_collection=False)
 302         self.assertIn('linkeddir', cwriter.manifest_text())
 303         self.assertIn('linkedfile', cwriter.manifest_text())
 304         cwriter.destroy_cache()
 305
 306     def test_symlinks_are_not_followed_when_requested(self):
 307         self.assertTrue(os.path.islink(os.path.join(self.tempdir_with_symlink, 'linkeddir')))
 308         self.assertTrue(os.path.islink(os.path.join(self.tempdir_with_symlink, 'linkedfile')))
 309         cwriter = arv_put.ArvPutUploadJob([self.tempdir_with_symlink],
 310                                           follow_links=False)
 311         cwriter.start(save_collection=False)
 312         self.assertNotIn('linkeddir', cwriter.manifest_text())
 313         self.assertNotIn('linkedfile', cwriter.manifest_text())
 314         cwriter.destroy_cache()
 315         # Check for bug #17800: passed symlinks should also be ignored.
 316         linked_dir = os.path.join(self.tempdir_with_symlink, 'linkeddir')
 317         cwriter = arv_put.ArvPutUploadJob([linked_dir], follow_links=False)
 318         cwriter.start(save_collection=False)
 319         self.assertNotIn('linkeddir', cwriter.manifest_text())
 320         cwriter.destroy_cache()
 321
 322     def test_no_empty_collection_saved(self):
 323         self.assertTrue(os.path.islink(os.path.join(self.tempdir_with_symlink, 'linkeddir')))
 324         linked_dir = os.path.join(self.tempdir_with_symlink, 'linkeddir')
 325         cwriter = arv_put.ArvPutUploadJob([linked_dir], follow_links=False)
 326         cwriter.start(save_collection=True)
 327         self.assertIsNone(cwriter.manifest_locator())
 328         self.assertEqual('', cwriter.manifest_text())
 329         cwriter.destroy_cache()
 330
 331     def test_passing_nonexistant_path_raise_exception(self):
 332         uuid_str = str(uuid.uuid4())
 333         with self.assertRaises(arv_put.PathDoesNotExistError):
 334             arv_put.ArvPutUploadJob(["/this/path/does/not/exist/{}".format(uuid_str)])
 335
 336     def test_writer_works_without_cache(self):
 337         cwriter = arv_put.ArvPutUploadJob(['/dev/null'], resume=False)
 338         cwriter.start(save_collection=False)
 339         self.assertEqual(". d41d8cd98f00b204e9800998ecf8427e+0 0:0:null\n", cwriter.manifest_text())
 340
 341     def test_writer_works_with_cache(self):
 342         with tempfile.NamedTemporaryFile() as f:
 343             f.write(b'foo')
 344             f.flush()
 345             cwriter = arv_put.ArvPutUploadJob([f.name])
 346             cwriter.start(save_collection=False)
 347             self.assertEqual(0, cwriter.bytes_skipped)
 348             self.assertEqual(3, cwriter.bytes_written)
 349             # Don't destroy the cache, and start another upload
 350             cwriter_new = arv_put.ArvPutUploadJob([f.name])
 351             cwriter_new.start(save_collection=False)
 352             cwriter_new.destroy_cache()
 353             self.assertEqual(3, cwriter_new.bytes_skipped)
 354             self.assertEqual(3, cwriter_new.bytes_written)
 355
 356     def make_progress_tester(self):
 357         progression = []
 358         def record_func(written, expected):
 359             progression.append((written, expected))
 360         return progression, record_func
 361
 362     def test_progress_reporting(self):
 363         with tempfile.NamedTemporaryFile() as f:
 364             f.write(b'foo')
 365             f.flush()
 366             for expect_count in (None, 8):
 367                 progression, reporter = self.make_progress_tester()
 368                 cwriter = arv_put.ArvPutUploadJob([f.name],
 369                                                   reporter=reporter)
 370                 cwriter.bytes_expected = expect_count
 371                 cwriter.start(save_collection=False)
 372                 cwriter.destroy_cache()
 373                 self.assertIn((3, expect_count), progression)
 374
 375     def test_writer_upload_directory(self):
 376         cwriter = arv_put.ArvPutUploadJob([self.tempdir])
 377         cwriter.start(save_collection=False)
 378         cwriter.destroy_cache()
 379         self.assertEqual(1024*(1+2+3+4+5), cwriter.bytes_written)
 380
 381     def test_resume_large_file_upload(self):
 382         def wrapped_write(*args, **kwargs):
 383             data = args[1]
 384             # Exit only on last block
 385             if len(data) < arvados.config.KEEP_BLOCK_SIZE:
 386                 # Simulate a checkpoint before quitting. Ensure block commit.
 387                 self.writer._update(final=True)
 388                 raise SystemExit("Simulated error")
 389             return self.arvfile_write(*args, **kwargs)
 390
 391         with mock.patch('arvados.arvfile.ArvadosFileWriter.write',
 392                         autospec=True) as mocked_write:
 393             mocked_write.side_effect = wrapped_write
 394             writer = arv_put.ArvPutUploadJob([self.large_file_name],
 395                                              replication_desired=1)
 396             # We'll be accessing from inside the wrapper
 397             self.writer = writer
 398             with self.assertRaises(SystemExit):
 399                 writer.start(save_collection=False)
 400             # Confirm that the file was partially uploaded
 401             self.assertGreater(writer.bytes_written, 0)
 402             self.assertLess(writer.bytes_written,
 403                             os.path.getsize(self.large_file_name))
 404         # Retry the upload
 405         writer2 = arv_put.ArvPutUploadJob([self.large_file_name],
 406                                           replication_desired=1)
 407         writer2.start(save_collection=False)
 408         self.assertEqual(writer.bytes_written + writer2.bytes_written - writer2.bytes_skipped,
 409                          os.path.getsize(self.large_file_name))
 410         writer2.destroy_cache()
 411         del(self.writer)
 412
 413     # Test for bug #11002
 414     def test_graceful_exit_while_repacking_small_blocks(self):
 415         def wrapped_commit(*args, **kwargs):
 416             raise SystemExit("Simulated error")
 417
 418         with mock.patch('arvados.arvfile._BlockManager.commit_bufferblock',
 419                         autospec=True) as mocked_commit:
 420             mocked_commit.side_effect = wrapped_commit
 421             # Upload a little more than 1 block, wrapped_commit will make the first block
 422             # commit to fail.
 423             # arv-put should not exit with an exception by trying to commit the collection
 424             # as it's in an inconsistent state.
 425             writer = arv_put.ArvPutUploadJob([self.small_files_dir],
 426                                              replication_desired=1)
 427             try:
 428                 with self.assertRaises(SystemExit):
 429                     writer.start(save_collection=False)
 430             except arvados.arvfile.UnownedBlockError:
 431                 self.fail("arv-put command is trying to use a corrupted BlockManager. See https://dev.arvados.org/issues/11002")
 432         writer.destroy_cache()
 433
 434     def test_no_resume_when_asked(self):
 435         def wrapped_write(*args, **kwargs):
 436             data = args[1]
 437             # Exit only on last block
 438             if len(data) < arvados.config.KEEP_BLOCK_SIZE:
 439                 # Simulate a checkpoint before quitting.
 440                 self.writer._update()
 441                 raise SystemExit("Simulated error")
 442             return self.arvfile_write(*args, **kwargs)
 443
 444         with mock.patch('arvados.arvfile.ArvadosFileWriter.write',
 445                         autospec=True) as mocked_write:
 446             mocked_write.side_effect = wrapped_write
 447             writer = arv_put.ArvPutUploadJob([self.large_file_name],
 448                                              replication_desired=1)
 449             # We'll be accessing from inside the wrapper
 450             self.writer = writer
 451             with self.assertRaises(SystemExit):
 452                 writer.start(save_collection=False)
 453             # Confirm that the file was partially uploaded
 454             self.assertGreater(writer.bytes_written, 0)
 455             self.assertLess(writer.bytes_written,
 456                             os.path.getsize(self.large_file_name))
 457         # Retry the upload, this time without resume
 458         writer2 = arv_put.ArvPutUploadJob([self.large_file_name],
 459                                           replication_desired=1,
 460                                           resume=False)
 461         writer2.start(save_collection=False)
 462         self.assertEqual(writer2.bytes_skipped, 0)
 463         self.assertEqual(writer2.bytes_written,
 464                          os.path.getsize(self.large_file_name))
 465         writer2.destroy_cache()
 466         del(self.writer)
 467
 468     def test_no_resume_when_no_cache(self):
 469         def wrapped_write(*args, **kwargs):
 470             data = args[1]
 471             # Exit only on last block
 472             if len(data) < arvados.config.KEEP_BLOCK_SIZE:
 473                 # Simulate a checkpoint before quitting.
 474                 self.writer._update()
 475                 raise SystemExit("Simulated error")
 476             return self.arvfile_write(*args, **kwargs)
 477
 478         with mock.patch('arvados.arvfile.ArvadosFileWriter.write',
 479                         autospec=True) as mocked_write:
 480             mocked_write.side_effect = wrapped_write
 481             writer = arv_put.ArvPutUploadJob([self.large_file_name],
 482                                              replication_desired=1)
 483             # We'll be accessing from inside the wrapper
 484             self.writer = writer
 485             with self.assertRaises(SystemExit):
 486                 writer.start(save_collection=False)
 487             # Confirm that the file was partially uploaded
 488             self.assertGreater(writer.bytes_written, 0)
 489             self.assertLess(writer.bytes_written,
 490                             os.path.getsize(self.large_file_name))
 491         # Retry the upload, this time without cache usage
 492         writer2 = arv_put.ArvPutUploadJob([self.large_file_name],
 493                                           replication_desired=1,
 494                                           resume=False,
 495                                           use_cache=False)
 496         writer2.start(save_collection=False)
 497         self.assertEqual(writer2.bytes_skipped, 0)
 498         self.assertEqual(writer2.bytes_written,
 499                          os.path.getsize(self.large_file_name))
 500         writer2.destroy_cache()
 501         del(self.writer)
 502
 503     def test_dry_run_feature(self):
 504         def wrapped_write(*args, **kwargs):
 505             data = args[1]
 506             # Exit only on last block
 507             if len(data) < arvados.config.KEEP_BLOCK_SIZE:
 508                 # Simulate a checkpoint before quitting.
 509                 self.writer._update()
 510                 raise SystemExit("Simulated error")
 511             return self.arvfile_write(*args, **kwargs)
 512
 513         with mock.patch('arvados.arvfile.ArvadosFileWriter.write',
 514                         autospec=True) as mocked_write:
 515             mocked_write.side_effect = wrapped_write
 516             writer = arv_put.ArvPutUploadJob([self.large_file_name],
 517                                              replication_desired=1)
 518             # We'll be accessing from inside the wrapper
 519             self.writer = writer
 520             with self.assertRaises(SystemExit):
 521                 writer.start(save_collection=False)
 522             # Confirm that the file was partially uploaded
 523             self.assertGreater(writer.bytes_written, 0)
 524             self.assertLess(writer.bytes_written,
 525                             os.path.getsize(self.large_file_name))
 526         with self.assertRaises(arv_put.ArvPutUploadIsPending):
 527             # Retry the upload using dry_run to check if there is a pending upload
 528             writer2 = arv_put.ArvPutUploadJob([self.large_file_name],
 529                                               replication_desired=1,
 530                                               dry_run=True)
 531         # Complete the pending upload
 532         writer3 = arv_put.ArvPutUploadJob([self.large_file_name],
 533                                           replication_desired=1)
 534         writer3.start(save_collection=False)
 535         with self.assertRaises(arv_put.ArvPutUploadNotPending):
 536             # Confirm there's no pending upload with dry_run=True
 537             writer4 = arv_put.ArvPutUploadJob([self.large_file_name],
 538                                               replication_desired=1,
 539                                               dry_run=True)
 540         # Test obvious cases
 541         with self.assertRaises(arv_put.ArvPutUploadIsPending):
 542             arv_put.ArvPutUploadJob([self.large_file_name],
 543                                     replication_desired=1,
 544                                     dry_run=True,
 545                                     resume=False,
 546                                     use_cache=False)
 547         with self.assertRaises(arv_put.ArvPutUploadIsPending):
 548             arv_put.ArvPutUploadJob([self.large_file_name],
 549                                     replication_desired=1,
 550                                     dry_run=True,
 551                                     resume=False)
 552         del(self.writer)
 553
 554 class CachedManifestValidationTest(ArvadosBaseTestCase):
 555     class MockedPut(arv_put.ArvPutUploadJob):
 556         def __init__(self, cached_manifest=None):
 557             self._state = arv_put.ArvPutUploadJob.EMPTY_STATE
 558             self._state['manifest'] = cached_manifest
 559             self._api_client = mock.MagicMock()
 560             self.logger = mock.MagicMock()
 561             self.num_retries = 1
 562
 563     def datetime_to_hex(self, dt):
 564         return hex(int(time.mktime(dt.timetuple())))[2:]
 565
 566     def setUp(self):
 567         super(CachedManifestValidationTest, self).setUp()
 568         self.block1 = "fdba98970961edb29f88241b9d99d890" # foo
 569         self.block2 = "37b51d194a7513e45b56f6524f2d51f2" # bar
 570         self.template = ". "+self.block1+"+3+Asignature@%s "+self.block2+"+3+Anothersignature@%s 0:3:foofile.txt 3:6:barfile.txt\n"
 571
 572     def test_empty_cached_manifest_is_valid(self):
 573         put_mock = self.MockedPut()
 574         self.assertEqual(None, put_mock._state.get('manifest'))
 575         self.assertTrue(put_mock._cached_manifest_valid())
 576         put_mock._state['manifest'] = ''
 577         self.assertTrue(put_mock._cached_manifest_valid())
 578
 579     def test_signature_cases(self):
 580         now = datetime.datetime.utcnow()
 581         yesterday = now - datetime.timedelta(days=1)
 582         lastweek = now - datetime.timedelta(days=7)
 583         tomorrow = now + datetime.timedelta(days=1)
 584         nextweek = now + datetime.timedelta(days=7)
 585
 586         def mocked_head(blocks={}, loc=None):
 587             blk = loc.split('+', 1)[0]
 588             if blocks.get(blk):
 589                 return True
 590             raise arvados.errors.KeepRequestError("mocked error - block invalid")
 591
 592         # Block1_expiration, Block2_expiration, Block1_HEAD, Block2_HEAD, Expectation
 593         cases = [
 594             # All expired, reset cache - OK
 595             (yesterday, lastweek, False, False, True),
 596             (lastweek, yesterday, False, False, True),
 597             # All non-expired valid blocks - OK
 598             (tomorrow, nextweek, True, True, True),
 599             (nextweek, tomorrow, True, True, True),
 600             # All non-expired invalid blocks - Not OK
 601             (tomorrow, nextweek, False, False, False),
 602             (nextweek, tomorrow, False, False, False),
 603             # One non-expired valid block - OK
 604             (tomorrow, yesterday, True, False, True),
 605             (yesterday, tomorrow, False, True, True),
 606             # One non-expired invalid block - Not OK
 607             (tomorrow, yesterday, False, False, False),
 608             (yesterday, tomorrow, False, False, False),
 609         ]
 610         for case in cases:
 611             b1_expiration, b2_expiration, b1_valid, b2_valid, outcome = case
 612             head_responses = {
 613                 self.block1: b1_valid,
 614                 self.block2: b2_valid,
 615             }
 616             cached_manifest = self.template % (
 617                 self.datetime_to_hex(b1_expiration),
 618                 self.datetime_to_hex(b2_expiration),
 619             )
 620             arvput = self.MockedPut(cached_manifest)
 621             with mock.patch('arvados.collection.KeepClient.head') as head_mock:
 622                 head_mock.side_effect = partial(mocked_head, head_responses)
 623                 self.assertEqual(outcome, arvput._cached_manifest_valid(),
 624                     "Case '%s' should have produced outcome '%s'" % (case, outcome)
 625                 )
 626                 if b1_expiration > now or b2_expiration > now:
 627                     # A HEAD request should have been done
 628                     head_mock.assert_called_once()
 629                 else:
 630                     head_mock.assert_not_called()
 631
 632
 633 class ArvadosExpectedBytesTest(ArvadosBaseTestCase):
 634     TEST_SIZE = os.path.getsize(__file__)
 635
 636     def test_expected_bytes_for_file(self):
 637         writer = arv_put.ArvPutUploadJob([__file__])
 638         self.assertEqual(self.TEST_SIZE,
 639                          writer.bytes_expected)
 640
 641     def test_expected_bytes_for_tree(self):
 642         tree = self.make_tmpdir()
 643         shutil.copyfile(__file__, os.path.join(tree, 'one'))
 644         shutil.copyfile(__file__, os.path.join(tree, 'two'))
 645
 646         writer = arv_put.ArvPutUploadJob([tree])
 647         self.assertEqual(self.TEST_SIZE * 2,
 648                          writer.bytes_expected)
 649         writer = arv_put.ArvPutUploadJob([tree, __file__])
 650         self.assertEqual(self.TEST_SIZE * 3,
 651                          writer.bytes_expected)
 652
 653     def test_expected_bytes_for_device(self):
 654         writer = arv_put.ArvPutUploadJob(['/dev/null'], use_cache=False, resume=False)
 655         self.assertIsNone(writer.bytes_expected)
 656         writer = arv_put.ArvPutUploadJob([__file__, '/dev/null'])
 657         self.assertIsNone(writer.bytes_expected)
 658
 659
 660 class ArvadosPutReportTest(ArvadosBaseTestCase):
 661     def test_machine_progress(self):
 662         for count, total in [(0, 1), (0, None), (1, None), (235, 9283)]:
 663             expect = ": {} written {} total\n".format(
 664                 count, -1 if (total is None) else total)
 665             self.assertTrue(
 666                 arv_put.machine_progress(count, total).endswith(expect))
 667
 668     def test_known_human_progress(self):
 669         for count, total in [(0, 1), (2, 4), (45, 60)]:
 670             expect = '{:.1%}'.format(1.0*count/total)
 671             actual = arv_put.human_progress(count, total)
 672             self.assertTrue(actual.startswith('\r'))
 673             self.assertIn(expect, actual)
 674
 675     def test_unknown_human_progress(self):
 676         for count in [1, 20, 300, 4000, 50000]:
 677             self.assertTrue(re.search(r'\b{}\b'.format(count),
 678                                       arv_put.human_progress(count, None)))
 679
 680
 681 class ArvPutLogFormatterTest(ArvadosBaseTestCase):
 682     matcher = r'\(X-Request-Id: req-[a-z0-9]{20}\)'
 683
 684     def setUp(self):
 685         super(ArvPutLogFormatterTest, self).setUp()
 686         self.stderr = tutil.StringIO()
 687         self.loggingHandler = logging.StreamHandler(self.stderr)
 688         self.loggingHandler.setFormatter(
 689             arv_put.ArvPutLogFormatter(arvados.util.new_request_id()))
 690         self.logger = logging.getLogger()
 691         self.logger.addHandler(self.loggingHandler)
 692         self.logger.setLevel(logging.DEBUG)
 693
 694     def tearDown(self):
 695         self.logger.removeHandler(self.loggingHandler)
 696         self.stderr.close()
 697         self.stderr = None
 698         super(ArvPutLogFormatterTest, self).tearDown()
 699
 700     def test_request_id_logged_only_once_on_error(self):
 701         self.logger.error('Ooops, something bad happened.')
 702         self.logger.error('Another bad thing just happened.')
 703         log_lines = self.stderr.getvalue().split('\n')[:-1]
 704         self.assertEqual(2, len(log_lines))
 705         self.assertRegex(log_lines[0], self.matcher)
 706         self.assertNotRegex(log_lines[1], self.matcher)
 707
 708     def test_request_id_logged_only_once_on_debug(self):
 709         self.logger.debug('This is just a debug message.')
 710         self.logger.debug('Another message, move along.')
 711         log_lines = self.stderr.getvalue().split('\n')[:-1]
 712         self.assertEqual(2, len(log_lines))
 713         self.assertRegex(log_lines[0], self.matcher)
 714         self.assertNotRegex(log_lines[1], self.matcher)
 715
 716     def test_request_id_not_logged_on_info(self):
 717         self.logger.info('This should be a useful message')
 718         log_lines = self.stderr.getvalue().split('\n')[:-1]
 719         self.assertEqual(1, len(log_lines))
 720         self.assertNotRegex(log_lines[0], self.matcher)
 721
 722 class ArvadosPutTest(run_test_server.TestCaseWithServers,
 723                      ArvadosBaseTestCase,
 724                      tutil.VersionChecker):
 725     MAIN_SERVER = {}
 726     Z_UUID = 'zzzzz-zzzzz-zzzzzzzzzzzzzzz'
 727
 728     def call_main_with_args(self, args):
 729         self.main_stdout.seek(0, 0)
 730         self.main_stdout.truncate(0)
 731         self.main_stderr.seek(0, 0)
 732         self.main_stderr.truncate(0)
 733         return arv_put.main(args, self.main_stdout, self.main_stderr)
 734
 735     def call_main_on_test_file(self, args=[]):
 736         with self.make_test_file() as testfile:
 737             path = testfile.name
 738             self.call_main_with_args(['--stream', '--no-progress'] + args + [path])
 739         self.assertTrue(
 740             os.path.exists(os.path.join(os.environ['KEEP_LOCAL_STORE'],
 741                                         '098f6bcd4621d373cade4e832627b4f6')),
 742             "did not find file stream in Keep store")
 743
 744     def setUp(self):
 745         super(ArvadosPutTest, self).setUp()
 746         run_test_server.authorize_with('active')
 747         arv_put.api_client = None
 748         self.main_stdout = tutil.StringIO()
 749         self.main_stderr = tutil.StringIO()
 750         self.loggingHandler = logging.StreamHandler(self.main_stderr)
 751         self.loggingHandler.setFormatter(
 752             arv_put.ArvPutLogFormatter(arvados.util.new_request_id()))
 753         logging.getLogger().addHandler(self.loggingHandler)
 754
 755     def tearDown(self):
 756         logging.getLogger().removeHandler(self.loggingHandler)
 757         for outbuf in ['main_stdout', 'main_stderr']:
 758             if hasattr(self, outbuf):
 759                 getattr(self, outbuf).close()
 760                 delattr(self, outbuf)
 761         super(ArvadosPutTest, self).tearDown()
 762
 763     def test_version_argument(self):
 764         with tutil.redirected_streams(
 765                 stdout=tutil.StringIO, stderr=tutil.StringIO) as (out, err):
 766             with self.assertRaises(SystemExit):
 767                 self.call_main_with_args(['--version'])
 768         self.assertVersionOutput(out, err)
 769
 770     def test_simple_file_put(self):
 771         self.call_main_on_test_file()
 772
 773     def test_put_with_unwriteable_cache_dir(self):
 774         orig_cachedir = arv_put.ResumeCache.CACHE_DIR
 775         cachedir = self.make_tmpdir()
 776         os.chmod(cachedir, 0o0)
 777         arv_put.ResumeCache.CACHE_DIR = cachedir
 778         try:
 779             self.call_main_on_test_file()
 780         finally:
 781             arv_put.ResumeCache.CACHE_DIR = orig_cachedir
 782             os.chmod(cachedir, 0o700)
 783
 784     def test_put_with_unwritable_cache_subdir(self):
 785         orig_cachedir = arv_put.ResumeCache.CACHE_DIR
 786         cachedir = self.make_tmpdir()
 787         os.chmod(cachedir, 0o0)
 788         arv_put.ResumeCache.CACHE_DIR = os.path.join(cachedir, 'cachedir')
 789         try:
 790             self.call_main_on_test_file()
 791         finally:
 792             arv_put.ResumeCache.CACHE_DIR = orig_cachedir
 793             os.chmod(cachedir, 0o700)
 794
 795     def test_put_block_replication(self):
 796         self.call_main_on_test_file()
 797         with mock.patch('arvados.collection.KeepClient.local_store_put') as put_mock:
 798             put_mock.return_value = 'acbd18db4cc2f85cedef654fccc4a4d8+3'
 799             self.call_main_on_test_file(['--replication', '1'])
 800             self.call_main_on_test_file(['--replication', '4'])
 801             self.call_main_on_test_file(['--replication', '5'])
 802             self.assertEqual(
 803                 [x[-1].get('copies') for x in put_mock.call_args_list],
 804                 [1, 4, 5])
 805
 806     def test_normalize(self):
 807         testfile1 = self.make_test_file()
 808         testfile2 = self.make_test_file()
 809         test_paths = [testfile1.name, testfile2.name]
 810         # Reverse-sort the paths, so normalization must change their order.
 811         test_paths.sort(reverse=True)
 812         self.call_main_with_args(['--stream', '--no-progress', '--normalize'] +
 813                                  test_paths)
 814         manifest = self.main_stdout.getvalue()
 815         # Assert the second file we specified appears first in the manifest.
 816         file_indices = [manifest.find(':' + os.path.basename(path))
 817                         for path in test_paths]
 818         self.assertGreater(*file_indices)
 819
 820     def test_error_name_without_collection(self):
 821         self.assertRaises(SystemExit, self.call_main_with_args,
 822                           ['--name', 'test without Collection',
 823                            '--stream', '/dev/null'])
 824
 825     def test_error_when_project_not_found(self):
 826         self.assertRaises(SystemExit,
 827                           self.call_main_with_args,
 828                           ['--project-uuid', self.Z_UUID])
 829
 830     def test_error_bad_project_uuid(self):
 831         self.assertRaises(SystemExit,
 832                           self.call_main_with_args,
 833                           ['--project-uuid', self.Z_UUID, '--stream'])
 834
 835     def test_error_when_excluding_absolute_path(self):
 836         tmpdir = self.make_tmpdir()
 837         self.assertRaises(SystemExit,
 838                           self.call_main_with_args,
 839                           ['--exclude', '/some/absolute/path/*',
 840                            tmpdir])
 841
 842     def test_api_error_handling(self):
 843         coll_save_mock = mock.Mock(name='arv.collection.Collection().save_new()')
 844         coll_save_mock.side_effect = arvados.errors.ApiError(
 845             fake_httplib2_response(403), b'{}')
 846         with mock.patch('arvados.collection.Collection.save_new',
 847                         new=coll_save_mock):
 848             with self.assertRaises(SystemExit) as exc_test:
 849                 self.call_main_with_args(['/dev/null'])
 850             self.assertLess(0, exc_test.exception.args[0])
 851             self.assertLess(0, coll_save_mock.call_count)
 852             self.assertEqual("", self.main_stdout.getvalue())
 853
 854     def test_request_id_logging_on_error(self):
 855         matcher = r'\(X-Request-Id: req-[a-z0-9]{20}\)\n'
 856         coll_save_mock = mock.Mock(name='arv.collection.Collection().save_new()')
 857         coll_save_mock.side_effect = arvados.errors.ApiError(
 858             fake_httplib2_response(403), b'{}')
 859         with mock.patch('arvados.collection.Collection.save_new',
 860                         new=coll_save_mock):
 861             with self.assertRaises(SystemExit):
 862                 self.call_main_with_args(['/dev/null'])
 863             self.assertRegex(
 864                 self.main_stderr.getvalue(), matcher)
 865
 866
 867 class ArvPutIntegrationTest(run_test_server.TestCaseWithServers,
 868                             ArvadosBaseTestCase):
 869     MAIN_SERVER = {}
 870     KEEP_SERVER = {'blob_signing': True}
 871     PROJECT_UUID = run_test_server.fixture('groups')['aproject']['uuid']
 872
 873     @classmethod
 874     def setUpClass(cls):
 875         super(ArvPutIntegrationTest, cls).setUpClass()
 876         cls.ENVIRON = os.environ.copy()
 877         cls.ENVIRON['PYTHONPATH'] = ':'.join(sys.path)
 878
 879     def datetime_to_hex(self, dt):
 880         return hex(int(time.mktime(dt.timetuple())))[2:]
 881
 882     def setUp(self):
 883         super(ArvPutIntegrationTest, self).setUp()
 884         arv_put.api_client = None
 885
 886     def authorize_with(self, token_name):
 887         run_test_server.authorize_with(token_name)
 888         for v in ["ARVADOS_API_HOST",
 889                   "ARVADOS_API_HOST_INSECURE",
 890                   "ARVADOS_API_TOKEN"]:
 891             self.ENVIRON[v] = arvados.config.settings()[v]
 892         arv_put.api_client = arvados.api('v1')
 893
 894     def current_user(self):
 895         return arv_put.api_client.users().current().execute()
 896
 897     def test_check_real_project_found(self):
 898         self.authorize_with('active')
 899         self.assertTrue(arv_put.desired_project_uuid(arv_put.api_client, self.PROJECT_UUID, 0),
 900                         "did not correctly find test fixture project")
 901
 902     def test_check_error_finding_nonexistent_uuid(self):
 903         BAD_UUID = 'zzzzz-zzzzz-zzzzzzzzzzzzzzz'
 904         self.authorize_with('active')
 905         try:
 906             result = arv_put.desired_project_uuid(arv_put.api_client, BAD_UUID,
 907                                                   0)
 908         except ValueError as error:
 909             self.assertIn(BAD_UUID, str(error))
 910         else:
 911             self.assertFalse(result, "incorrectly found nonexistent project")
 912
 913     def test_check_error_finding_nonexistent_project(self):
 914         BAD_UUID = 'zzzzz-tpzed-zzzzzzzzzzzzzzz'
 915         self.authorize_with('active')
 916         with self.assertRaises(apiclient.errors.HttpError):
 917             arv_put.desired_project_uuid(arv_put.api_client, BAD_UUID,
 918                                                   0)
 919
 920     def test_short_put_from_stdin(self):
 921         # Have to run this as an integration test since arv-put can't
 922         # read from the tests' stdin.
 923         # arv-put usually can't stat(os.path.realpath('/dev/stdin')) in this
 924         # case, because the /proc entry is already gone by the time it tries.
 925         pipe = subprocess.Popen(
 926             [sys.executable, arv_put.__file__, '--stream'],
 927             stdin=subprocess.PIPE, stdout=subprocess.PIPE,
 928             stderr=subprocess.STDOUT, env=self.ENVIRON)
 929         pipe.stdin.write(b'stdin test\xa6\n')
 930         pipe.stdin.close()
 931         deadline = time.time() + 5
 932         while (pipe.poll() is None) and (time.time() < deadline):
 933             time.sleep(.1)
 934         returncode = pipe.poll()
 935         if returncode is None:
 936             pipe.terminate()
 937             self.fail("arv-put did not PUT from stdin within 5 seconds")
 938         elif returncode != 0:
 939             sys.stdout.write(pipe.stdout.read())
 940             self.fail("arv-put returned exit code {}".format(returncode))
 941         self.assertIn('1cb671b355a0c23d5d1c61d59cdb1b2b+12',
 942                       pipe.stdout.read().decode())
 943
 944     def test_sigint_logs_request_id(self):
 945         # Start arv-put, give it a chance to start up, send SIGINT,
 946         # and check that its output includes the X-Request-Id.
 947         input_stream = subprocess.Popen(
 948             ['sleep', '10'],
 949             stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
 950         pipe = subprocess.Popen(
 951             [sys.executable, arv_put.__file__, '--stream'],
 952             stdin=input_stream.stdout, stdout=subprocess.PIPE,
 953             stderr=subprocess.STDOUT, env=self.ENVIRON)
 954         # Wait for arv-put child process to print something (i.e., a
 955         # log message) so we know its signal handler is installed.
 956         select.select([pipe.stdout], [], [], 10)
 957         pipe.send_signal(signal.SIGINT)
 958         deadline = time.time() + 5
 959         while (pipe.poll() is None) and (time.time() < deadline):
 960             time.sleep(.1)
 961         returncode = pipe.poll()
 962         input_stream.terminate()
 963         if returncode is None:
 964             pipe.terminate()
 965             self.fail("arv-put did not exit within 5 seconds")
 966         self.assertRegex(pipe.stdout.read().decode(), r'\(X-Request-Id: req-[a-z0-9]{20}\)')
 967
 968     def test_ArvPutSignedManifest(self):
 969         # ArvPutSignedManifest runs "arv-put foo" and then attempts to get
 970         # the newly created manifest from the API server, testing to confirm
 971         # that the block locators in the returned manifest are signed.
 972         self.authorize_with('active')
 973
 974         # Before doing anything, demonstrate that the collection
 975         # we're about to create is not present in our test fixture.
 976         manifest_uuid = "00b4e9f40ac4dd432ef89749f1c01e74+47"
 977         with self.assertRaises(apiclient.errors.HttpError):
 978             arv_put.api_client.collections().get(
 979                 uuid=manifest_uuid).execute()
 980
 981         datadir = self.make_tmpdir()
 982         with open(os.path.join(datadir, "foo"), "w") as f:
 983             f.write("The quick brown fox jumped over the lazy dog")
 984         p = subprocess.Popen([sys.executable, arv_put.__file__,
 985                               os.path.join(datadir, 'foo')],
 986                              stdout=subprocess.PIPE,
 987                              stderr=subprocess.PIPE,
 988                              env=self.ENVIRON)
 989         (_, err) = p.communicate()
 990         self.assertRegex(err.decode(), r'INFO: Collection saved as ')
 991         self.assertEqual(p.returncode, 0)
 992
 993         # The manifest text stored in the API server under the same
 994         # manifest UUID must use signed locators.
 995         c = arv_put.api_client.collections().get(uuid=manifest_uuid).execute()
 996         self.assertRegex(
 997             c['manifest_text'],
 998             r'^\. 08a008a01d498c404b0c30852b39d3b8\+44\+A[0-9a-f]+@[0-9a-f]+ 0:44:foo\n')
 999
1000         os.remove(os.path.join(datadir, "foo"))
1001         os.rmdir(datadir)
1002
1003     def run_and_find_collection(self, text, extra_args=[]):
1004         self.authorize_with('active')
1005         pipe = subprocess.Popen(
1006             [sys.executable, arv_put.__file__] + extra_args,
1007             stdin=subprocess.PIPE, stdout=subprocess.PIPE,
1008             stderr=subprocess.PIPE, env=self.ENVIRON)
1009         stdout, stderr = pipe.communicate(text.encode())
1010         self.assertRegex(stderr.decode(), r'INFO: Collection (updated:|saved as)')
1011         search_key = ('portable_data_hash'
1012                       if '--portable-data-hash' in extra_args else 'uuid')
1013         collection_list = arvados.api('v1').collections().list(
1014             filters=[[search_key, '=', stdout.decode().strip()]]
1015         ).execute().get('items', [])
1016         self.assertEqual(1, len(collection_list))
1017         return collection_list[0]
1018
1019     def test_all_expired_signatures_invalidates_cache(self):
1020         self.authorize_with('active')
1021         tmpdir = self.make_tmpdir()
1022         with open(os.path.join(tmpdir, 'somefile.txt'), 'w') as f:
1023             f.write('foo')
1024         # Upload a directory and get the cache file name
1025         p = subprocess.Popen([sys.executable, arv_put.__file__, tmpdir],
1026                              stdout=subprocess.PIPE,
1027                              stderr=subprocess.PIPE,
1028                              env=self.ENVIRON)
1029         (_, err) = p.communicate()
1030         self.assertRegex(err.decode(), r'INFO: Creating new cache file at ')
1031         self.assertEqual(p.returncode, 0)
1032         cache_filepath = re.search(r'INFO: Creating new cache file at (.*)',
1033                                    err.decode()).groups()[0]
1034         self.assertTrue(os.path.isfile(cache_filepath))
1035         # Load the cache file contents and modify the manifest to simulate
1036         # an expired access token
1037         with open(cache_filepath, 'r') as c:
1038             cache = json.load(c)
1039         self.assertRegex(cache['manifest'], r'\+A\S+\@')
1040         a_month_ago = datetime.datetime.now() - datetime.timedelta(days=30)
1041         cache['manifest'] = re.sub(
1042             r'\@.*? ',
1043             "@{} ".format(self.datetime_to_hex(a_month_ago)),
1044             cache['manifest'])
1045         with open(cache_filepath, 'w') as c:
1046             c.write(json.dumps(cache))
1047         # Re-run the upload and expect to get an invalid cache message
1048         p = subprocess.Popen([sys.executable, arv_put.__file__, tmpdir],
1049                              stdout=subprocess.PIPE,
1050                              stderr=subprocess.PIPE,
1051                              env=self.ENVIRON)
1052         (_, err) = p.communicate()
1053         self.assertRegex(
1054             err.decode(),
1055             r'INFO: Cache expired, starting from scratch.*')
1056         self.assertEqual(p.returncode, 0)
1057
1058     def test_invalid_signature_invalidates_cache(self):
1059         self.authorize_with('active')
1060         tmpdir = self.make_tmpdir()
1061         with open(os.path.join(tmpdir, 'somefile.txt'), 'w') as f:
1062             f.write('foo')
1063         # Upload a directory and get the cache file name
1064         p = subprocess.Popen([sys.executable, arv_put.__file__, tmpdir],
1065                              stdout=subprocess.PIPE,
1066                              stderr=subprocess.PIPE,
1067                              env=self.ENVIRON)
1068         (_, err) = p.communicate()
1069         self.assertRegex(err.decode(), r'INFO: Creating new cache file at ')
1070         self.assertEqual(p.returncode, 0)
1071         cache_filepath = re.search(r'INFO: Creating new cache file at (.*)',
1072                                    err.decode()).groups()[0]
1073         self.assertTrue(os.path.isfile(cache_filepath))
1074         # Load the cache file contents and modify the manifest to simulate
1075         # an invalid access token
1076         with open(cache_filepath, 'r') as c:
1077             cache = json.load(c)
1078         self.assertRegex(cache['manifest'], r'\+A\S+\@')
1079         cache['manifest'] = re.sub(
1080             r'\+A.*\@',
1081             "+Aabcdef0123456789abcdef0123456789abcdef01@",
1082             cache['manifest'])
1083         with open(cache_filepath, 'w') as c:
1084             c.write(json.dumps(cache))
1085         # Re-run the upload and expect to get an invalid cache message
1086         p = subprocess.Popen([sys.executable, arv_put.__file__, tmpdir],
1087                              stdout=subprocess.PIPE,
1088                              stderr=subprocess.PIPE,
1089                              env=self.ENVIRON)
1090         (_, err) = p.communicate()
1091         self.assertRegex(
1092             err.decode(),
1093             r'ERROR: arv-put: Resume cache contains invalid signature.*')
1094         self.assertEqual(p.returncode, 1)
1095
1096     def test_single_expired_signature_reuploads_file(self):
1097         self.authorize_with('active')
1098         tmpdir = self.make_tmpdir()
1099         with open(os.path.join(tmpdir, 'foofile.txt'), 'w') as f:
1100             f.write('foo')
1101         # Write a second file on its own subdir to force a new stream
1102         os.mkdir(os.path.join(tmpdir, 'bar'))
1103         with open(os.path.join(tmpdir, 'bar', 'barfile.txt'), 'w') as f:
1104             f.write('bar')
1105         # Upload a directory and get the cache file name
1106         p = subprocess.Popen([sys.executable, arv_put.__file__, tmpdir],
1107                              stdout=subprocess.PIPE,
1108                              stderr=subprocess.PIPE,
1109                              env=self.ENVIRON)
1110         (_, err) = p.communicate()
1111         self.assertRegex(err.decode(), r'INFO: Creating new cache file at ')
1112         self.assertEqual(p.returncode, 0)
1113         cache_filepath = re.search(r'INFO: Creating new cache file at (.*)',
1114                                    err.decode()).groups()[0]
1115         self.assertTrue(os.path.isfile(cache_filepath))
1116         # Load the cache file contents and modify the manifest to simulate
1117         # an expired access token
1118         with open(cache_filepath, 'r') as c:
1119             cache = json.load(c)
1120         self.assertRegex(cache['manifest'], r'\+A\S+\@')
1121         a_month_ago = datetime.datetime.now() - datetime.timedelta(days=30)
1122         # Make one of the signatures appear to have expired
1123         cache['manifest'] = re.sub(
1124             r'\@.*? 3:3:barfile.txt',
1125             "@{} 3:3:barfile.txt".format(self.datetime_to_hex(a_month_ago)),
1126             cache['manifest'])
1127         with open(cache_filepath, 'w') as c:
1128             c.write(json.dumps(cache))
1129         # Re-run the upload and expect to get an invalid cache message
1130         p = subprocess.Popen([sys.executable, arv_put.__file__, tmpdir],
1131                              stdout=subprocess.PIPE,
1132                              stderr=subprocess.PIPE,
1133                              env=self.ENVIRON)
1134         (_, err) = p.communicate()
1135         self.assertRegex(
1136             err.decode(),
1137             r'WARNING: Uploaded file \'.*barfile.txt\' access token expired, will re-upload it from scratch')
1138         self.assertEqual(p.returncode, 0)
1139         # Confirm that the resulting cache is different from the last run.
1140         with open(cache_filepath, 'r') as c2:
1141             new_cache = json.load(c2)
1142         self.assertNotEqual(cache['manifest'], new_cache['manifest'])
1143
1144     def test_put_collection_with_later_update(self):
1145         tmpdir = self.make_tmpdir()
1146         with open(os.path.join(tmpdir, 'file1'), 'w') as f:
1147             f.write('Relaxing in basins at the end of inlets terminates the endless tests from the box')
1148         col = self.run_and_find_collection("", ['--no-progress', tmpdir])
1149         self.assertNotEqual(None, col['uuid'])
1150         # Add a new file to the directory
1151         with open(os.path.join(tmpdir, 'file2'), 'w') as f:
1152             f.write('The quick brown fox jumped over the lazy dog')
1153         updated_col = self.run_and_find_collection("", ['--no-progress', '--update-collection', col['uuid'], tmpdir])
1154         self.assertEqual(col['uuid'], updated_col['uuid'])
1155         # Get the manifest and check that the new file is being included
1156         c = arv_put.api_client.collections().get(uuid=updated_col['uuid']).execute()
1157         self.assertRegex(c['manifest_text'], r'^\..* .*:44:file2\n')
1158
1159     def test_put_collection_with_utc_expiring_datetime(self):
1160         tmpdir = self.make_tmpdir()
1161         trash_at = (datetime.datetime.utcnow() + datetime.timedelta(days=90)).strftime('%Y%m%dT%H%MZ')
1162         with open(os.path.join(tmpdir, 'file1'), 'w') as f:
1163             f.write('Relaxing in basins at the end of inlets terminates the endless tests from the box')
1164         col = self.run_and_find_collection(
1165             "",
1166             ['--no-progress', '--trash-at', trash_at, tmpdir])
1167         self.assertNotEqual(None, col['uuid'])
1168         c = arv_put.api_client.collections().get(uuid=col['uuid']).execute()
1169         self.assertEqual(ciso8601.parse_datetime(trash_at),
1170             ciso8601.parse_datetime(c['trash_at']))
1171
1172     def test_put_collection_with_timezone_aware_expiring_datetime(self):
1173         tmpdir = self.make_tmpdir()
1174         trash_at = (datetime.datetime.utcnow() + datetime.timedelta(days=90)).strftime('%Y%m%dT%H%M-0300')
1175         with open(os.path.join(tmpdir, 'file1'), 'w') as f:
1176             f.write('Relaxing in basins at the end of inlets terminates the endless tests from the box')
1177         col = self.run_and_find_collection(
1178             "",
1179             ['--no-progress', '--trash-at', trash_at, tmpdir])
1180         self.assertNotEqual(None, col['uuid'])
1181         c = arv_put.api_client.collections().get(uuid=col['uuid']).execute()
1182         self.assertEqual(
1183             ciso8601.parse_datetime(trash_at).replace(tzinfo=None) + datetime.timedelta(hours=3),
1184             ciso8601.parse_datetime(c['trash_at']).replace(tzinfo=None))
1185
1186     def test_put_collection_with_timezone_naive_expiring_datetime(self):
1187         tmpdir = self.make_tmpdir()
1188         trash_at = (datetime.datetime.utcnow() + datetime.timedelta(days=90)).strftime('%Y%m%dT%H%M')
1189         with open(os.path.join(tmpdir, 'file1'), 'w') as f:
1190             f.write('Relaxing in basins at the end of inlets terminates the endless tests from the box')
1191         col = self.run_and_find_collection(
1192             "",
1193             ['--no-progress', '--trash-at', trash_at, tmpdir])
1194         self.assertNotEqual(None, col['uuid'])
1195         c = arv_put.api_client.collections().get(uuid=col['uuid']).execute()
1196         if time.daylight:
1197             offset = datetime.timedelta(seconds=time.altzone)
1198         else:
1199             offset = datetime.timedelta(seconds=time.timezone)
1200         self.assertEqual(
1201             ciso8601.parse_datetime(trash_at) + offset,
1202             ciso8601.parse_datetime(c['trash_at']).replace(tzinfo=None))
1203
1204     def test_put_collection_with_expiring_date_only(self):
1205         tmpdir = self.make_tmpdir()
1206         trash_at = '2140-01-01'
1207         end_of_day = datetime.timedelta(hours=23, minutes=59, seconds=59)
1208         with open(os.path.join(tmpdir, 'file1'), 'w') as f:
1209             f.write('Relaxing in basins at the end of inlets terminates the endless tests from the box')
1210         col = self.run_and_find_collection(
1211             "",
1212             ['--no-progress', '--trash-at', trash_at, tmpdir])
1213         self.assertNotEqual(None, col['uuid'])
1214         c = arv_put.api_client.collections().get(uuid=col['uuid']).execute()
1215         if time.daylight:
1216             offset = datetime.timedelta(seconds=time.altzone)
1217         else:
1218             offset = datetime.timedelta(seconds=time.timezone)
1219         self.assertEqual(
1220             ciso8601.parse_datetime(trash_at) + end_of_day + offset,
1221             ciso8601.parse_datetime(c['trash_at']).replace(tzinfo=None))
1222
1223     def test_put_collection_with_invalid_absolute_expiring_datetimes(self):
1224         cases = ['2100', '210010','2100-10', '2100-Oct']
1225         tmpdir = self.make_tmpdir()
1226         with open(os.path.join(tmpdir, 'file1'), 'w') as f:
1227             f.write('Relaxing in basins at the end of inlets terminates the endless tests from the box')
1228         for test_datetime in cases:
1229             with self.assertRaises(AssertionError):
1230                 self.run_and_find_collection(
1231                     "",
1232                     ['--no-progress', '--trash-at', test_datetime, tmpdir])
1233
1234     def test_put_collection_with_relative_expiring_datetime(self):
1235         expire_after = 7
1236         dt_before = datetime.datetime.utcnow() + datetime.timedelta(days=expire_after)
1237         tmpdir = self.make_tmpdir()
1238         with open(os.path.join(tmpdir, 'file1'), 'w') as f:
1239             f.write('Relaxing in basins at the end of inlets terminates the endless tests from the box')
1240         col = self.run_and_find_collection(
1241             "",
1242             ['--no-progress', '--trash-after', str(expire_after), tmpdir])
1243         self.assertNotEqual(None, col['uuid'])
1244         dt_after = datetime.datetime.utcnow() + datetime.timedelta(days=expire_after)
1245         c = arv_put.api_client.collections().get(uuid=col['uuid']).execute()
1246         trash_at = ciso8601.parse_datetime(c['trash_at']).replace(tzinfo=None)
1247         self.assertTrue(dt_before < trash_at)
1248         self.assertTrue(dt_after > trash_at)
1249
1250     def test_put_collection_with_invalid_relative_expiring_datetime(self):
1251         expire_after = 0 # Must be >= 1
1252         tmpdir = self.make_tmpdir()
1253         with open(os.path.join(tmpdir, 'file1'), 'w') as f:
1254             f.write('Relaxing in basins at the end of inlets terminates the endless tests from the box')
1255         with self.assertRaises(AssertionError):
1256             self.run_and_find_collection(
1257                 "",
1258                 ['--no-progress', '--trash-after', str(expire_after), tmpdir])
1259
1260     def test_upload_directory_reference_without_trailing_slash(self):
1261         tmpdir1 = self.make_tmpdir()
1262         tmpdir2 = self.make_tmpdir()
1263         with open(os.path.join(tmpdir1, 'foo'), 'w') as f:
1264             f.write('This is foo')
1265         with open(os.path.join(tmpdir2, 'bar'), 'w') as f:
1266             f.write('This is not foo')
1267         # Upload one directory and one file
1268         col = self.run_and_find_collection("", ['--no-progress',
1269                                                 tmpdir1,
1270                                                 os.path.join(tmpdir2, 'bar')])
1271         self.assertNotEqual(None, col['uuid'])
1272         c = arv_put.api_client.collections().get(uuid=col['uuid']).execute()
1273         # Check that 'foo' was written inside a subcollection
1274         # OTOH, 'bar' should have been directly uploaded on the root collection
1275         self.assertRegex(c['manifest_text'], r'^\. .*:15:bar\n\./.+ .*:11:foo\n')
1276
1277     def test_upload_directory_reference_with_trailing_slash(self):
1278         tmpdir1 = self.make_tmpdir()
1279         tmpdir2 = self.make_tmpdir()
1280         with open(os.path.join(tmpdir1, 'foo'), 'w') as f:
1281             f.write('This is foo')
1282         with open(os.path.join(tmpdir2, 'bar'), 'w') as f:
1283             f.write('This is not foo')
1284         # Upload one directory (with trailing slash) and one file
1285         col = self.run_and_find_collection("", ['--no-progress',
1286                                                 tmpdir1 + os.sep,
1287                                                 os.path.join(tmpdir2, 'bar')])
1288         self.assertNotEqual(None, col['uuid'])
1289         c = arv_put.api_client.collections().get(uuid=col['uuid']).execute()
1290         # Check that 'foo' and 'bar' were written at the same level
1291         self.assertRegex(c['manifest_text'], r'^\. .*:15:bar .*:11:foo\n')
1292
1293     def test_put_collection_with_high_redundancy(self):
1294         # Write empty data: we're not testing CollectionWriter, just
1295         # making sure collections.create tells the API server what our
1296         # desired replication level is.
1297         collection = self.run_and_find_collection("", ['--replication', '4'])
1298         self.assertEqual(4, collection['replication_desired'])
1299
1300     def test_put_collection_with_default_redundancy(self):
1301         collection = self.run_and_find_collection("")
1302         self.assertEqual(None, collection['replication_desired'])
1303
1304     def test_put_collection_with_unnamed_project_link(self):
1305         link = self.run_and_find_collection(
1306             "Test unnamed collection",
1307             ['--portable-data-hash', '--project-uuid', self.PROJECT_UUID])
1308         username = pwd.getpwuid(os.getuid()).pw_name
1309         self.assertRegex(
1310             link['name'],
1311             r'^Saved at .* by {}@'.format(re.escape(username)))
1312
1313     def test_put_collection_with_name_and_no_project(self):
1314         link_name = 'Test Collection Link in home project'
1315         collection = self.run_and_find_collection(
1316             "Test named collection in home project",
1317             ['--portable-data-hash', '--name', link_name])
1318         self.assertEqual(link_name, collection['name'])
1319         my_user_uuid = self.current_user()['uuid']
1320         self.assertEqual(my_user_uuid, collection['owner_uuid'])
1321
1322     def test_put_collection_with_named_project_link(self):
1323         link_name = 'Test auto Collection Link'
1324         collection = self.run_and_find_collection("Test named collection",
1325                                       ['--portable-data-hash',
1326                                        '--name', link_name,
1327                                        '--project-uuid', self.PROJECT_UUID])
1328         self.assertEqual(link_name, collection['name'])
1329
1330     def test_put_collection_with_storage_classes_specified(self):
1331         collection = self.run_and_find_collection("", ['--storage-classes', 'hot'])
1332         self.assertEqual(len(collection['storage_classes_desired']), 1)
1333         self.assertEqual(collection['storage_classes_desired'][0], 'hot')
1334
1335     def test_put_collection_with_multiple_storage_classes_specified(self):
1336         collection = self.run_and_find_collection("", ['--storage-classes', ' foo, bar  ,baz'])
1337         self.assertEqual(len(collection['storage_classes_desired']), 3)
1338         self.assertEqual(collection['storage_classes_desired'], ['foo', 'bar', 'baz'])
1339
1340     def test_put_collection_without_storage_classes_specified(self):
1341         collection = self.run_and_find_collection("")
1342         self.assertEqual(len(collection['storage_classes_desired']), 1)
1343         self.assertEqual(collection['storage_classes_desired'][0], 'default')
1344
1345     def test_exclude_filename_pattern(self):
1346         tmpdir = self.make_tmpdir()
1347         tmpsubdir = os.path.join(tmpdir, 'subdir')
1348         os.mkdir(tmpsubdir)
1349         for fname in ['file1', 'file2', 'file3']:
1350             with open(os.path.join(tmpdir, "%s.txt" % fname), 'w') as f:
1351                 f.write("This is %s" % fname)
1352             with open(os.path.join(tmpsubdir, "%s.txt" % fname), 'w') as f:
1353                 f.write("This is %s" % fname)
1354         col = self.run_and_find_collection("", ['--no-progress',
1355                                                 '--exclude', '*2.txt',
1356                                                 '--exclude', 'file3.*',
1357                                                  tmpdir])
1358         self.assertNotEqual(None, col['uuid'])
1359         c = arv_put.api_client.collections().get(uuid=col['uuid']).execute()
1360         # None of the file2.txt & file3.txt should have been uploaded
1361         self.assertRegex(c['manifest_text'], r'^.*:file1.txt')
1362         self.assertNotRegex(c['manifest_text'], r'^.*:file2.txt')
1363         self.assertNotRegex(c['manifest_text'], r'^.*:file3.txt')
1364
1365     def test_exclude_filepath_pattern(self):
1366         tmpdir = self.make_tmpdir()
1367         tmpsubdir = os.path.join(tmpdir, 'subdir')
1368         os.mkdir(tmpsubdir)
1369         for fname in ['file1', 'file2', 'file3']:
1370             with open(os.path.join(tmpdir, "%s.txt" % fname), 'w') as f:
1371                 f.write("This is %s" % fname)
1372             with open(os.path.join(tmpsubdir, "%s.txt" % fname), 'w') as f:
1373                 f.write("This is %s" % fname)
1374         col = self.run_and_find_collection("", ['--no-progress',
1375                                                 '--exclude', 'subdir/*2.txt',
1376                                                 '--exclude', './file1.*',
1377                                                  tmpdir])
1378         self.assertNotEqual(None, col['uuid'])
1379         c = arv_put.api_client.collections().get(uuid=col['uuid']).execute()
1380         # Only tmpdir/file1.txt & tmpdir/subdir/file2.txt should have been excluded
1381         self.assertNotRegex(c['manifest_text'],
1382                             r'^\./%s.*:file1.txt' % os.path.basename(tmpdir))
1383         self.assertNotRegex(c['manifest_text'],
1384                             r'^\./%s/subdir.*:file2.txt' % os.path.basename(tmpdir))
1385         self.assertRegex(c['manifest_text'],
1386                          r'^\./%s.*:file2.txt' % os.path.basename(tmpdir))
1387         self.assertRegex(c['manifest_text'], r'^.*:file3.txt')
1388
1389     def test_unicode_on_filename(self):
1390         tmpdir = self.make_tmpdir()
1391         fname = u"i❤arvados.txt"
1392         with open(os.path.join(tmpdir, fname), 'w') as f:
1393             f.write("This is a unicode named file")
1394         col = self.run_and_find_collection("", ['--no-progress', tmpdir])
1395         self.assertNotEqual(None, col['uuid'])
1396         c = arv_put.api_client.collections().get(uuid=col['uuid']).execute()
1397         self.assertTrue(fname in c['manifest_text'], u"{} does not include {}".format(c['manifest_text'], fname))
1398
1399     def test_silent_mode_no_errors(self):
1400         self.authorize_with('active')
1401         tmpdir = self.make_tmpdir()
1402         with open(os.path.join(tmpdir, 'test.txt'), 'w') as f:
1403             f.write('hello world')
1404         pipe = subprocess.Popen(
1405             [sys.executable, arv_put.__file__] + ['--silent', tmpdir],
1406             stdin=subprocess.PIPE, stdout=subprocess.PIPE,
1407             stderr=subprocess.PIPE, env=self.ENVIRON)
1408         stdout, stderr = pipe.communicate()
1409         # No console output should occur on normal operations
1410         self.assertNotRegex(stderr.decode(), r'.+')
1411         self.assertNotRegex(stdout.decode(), r'.+')
1412
1413     def test_silent_mode_does_not_avoid_error_messages(self):
1414         self.authorize_with('active')
1415         pipe = subprocess.Popen(
1416             [sys.executable, arv_put.__file__] + ['--silent',
1417                                                   '/path/not/existant'],
1418             stdin=subprocess.PIPE, stdout=subprocess.PIPE,
1419             stderr=subprocess.PIPE, env=self.ENVIRON)
1420         stdout, stderr = pipe.communicate()
1421         # Error message should be displayed when errors happen
1422         self.assertRegex(stderr.decode(), r'.*ERROR:.*')
1423         self.assertNotRegex(stdout.decode(), r'.+')
1424
1425
1426 if __name__ == '__main__':
1427     unittest.main()