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