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     MAIN_SERVER = {}
 856     KEEP_SERVER = {'blob_signing': True}
 857     PROJECT_UUID = run_test_server.fixture('groups')['aproject']['uuid']
 858
 859     @classmethod
 860     def setUpClass(cls):
 861         super(ArvPutIntegrationTest, cls).setUpClass()
 862         cls.ENVIRON = os.environ.copy()
 863         cls.ENVIRON['PYTHONPATH'] = ':'.join(sys.path)
 864
 865     def datetime_to_hex(self, dt):
 866         return hex(int(time.mktime(dt.timetuple())))[2:]
 867
 868     def setUp(self):
 869         super(ArvPutIntegrationTest, self).setUp()
 870         arv_put.api_client = None
 871
 872     def authorize_with(self, token_name):
 873         run_test_server.authorize_with(token_name)
 874         for v in ["ARVADOS_API_HOST",
 875                   "ARVADOS_API_HOST_INSECURE",
 876                   "ARVADOS_API_TOKEN"]:
 877             self.ENVIRON[v] = arvados.config.settings()[v]
 878         arv_put.api_client = arvados.api('v1')
 879
 880     def current_user(self):
 881         return arv_put.api_client.users().current().execute()
 882
 883     def test_check_real_project_found(self):
 884         self.authorize_with('active')
 885         self.assertTrue(arv_put.desired_project_uuid(arv_put.api_client, self.PROJECT_UUID, 0),
 886                         "did not correctly find test fixture project")
 887
 888     def test_check_error_finding_nonexistent_uuid(self):
 889         BAD_UUID = 'zzzzz-zzzzz-zzzzzzzzzzzzzzz'
 890         self.authorize_with('active')
 891         try:
 892             result = arv_put.desired_project_uuid(arv_put.api_client, BAD_UUID,
 893                                                   0)
 894         except ValueError as error:
 895             self.assertIn(BAD_UUID, str(error))
 896         else:
 897             self.assertFalse(result, "incorrectly found nonexistent project")
 898
 899     def test_check_error_finding_nonexistent_project(self):
 900         BAD_UUID = 'zzzzz-tpzed-zzzzzzzzzzzzzzz'
 901         self.authorize_with('active')
 902         with self.assertRaises(apiclient.errors.HttpError):
 903             arv_put.desired_project_uuid(arv_put.api_client, BAD_UUID,
 904                                                   0)
 905
 906     def test_short_put_from_stdin(self):
 907         # Have to run this as an integration test since arv-put can't
 908         # read from the tests' stdin.
 909         # arv-put usually can't stat(os.path.realpath('/dev/stdin')) in this
 910         # case, because the /proc entry is already gone by the time it tries.
 911         pipe = subprocess.Popen(
 912             [sys.executable, arv_put.__file__, '--stream'],
 913             stdin=subprocess.PIPE, stdout=subprocess.PIPE,
 914             stderr=subprocess.STDOUT, env=self.ENVIRON)
 915         pipe.stdin.write(b'stdin test\xa6\n')
 916         pipe.stdin.close()
 917         deadline = time.time() + 5
 918         while (pipe.poll() is None) and (time.time() < deadline):
 919             time.sleep(.1)
 920         returncode = pipe.poll()
 921         if returncode is None:
 922             pipe.terminate()
 923             self.fail("arv-put did not PUT from stdin within 5 seconds")
 924         elif returncode != 0:
 925             sys.stdout.write(pipe.stdout.read())
 926             self.fail("arv-put returned exit code {}".format(returncode))
 927         self.assertIn('1cb671b355a0c23d5d1c61d59cdb1b2b+12',
 928                       pipe.stdout.read().decode())
 929
 930     def test_sigint_logs_request_id(self):
 931         # Start arv-put, give it a chance to start up, send SIGINT,
 932         # and check that its output includes the X-Request-Id.
 933         input_stream = subprocess.Popen(
 934             ['sleep', '10'],
 935             stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
 936         pipe = subprocess.Popen(
 937             [sys.executable, arv_put.__file__, '--stream'],
 938             stdin=input_stream.stdout, stdout=subprocess.PIPE,
 939             stderr=subprocess.STDOUT, env=self.ENVIRON)
 940         # Wait for arv-put child process to print something (i.e., a
 941         # log message) so we know its signal handler is installed.
 942         select.select([pipe.stdout], [], [], 10)
 943         pipe.send_signal(signal.SIGINT)
 944         deadline = time.time() + 5
 945         while (pipe.poll() is None) and (time.time() < deadline):
 946             time.sleep(.1)
 947         returncode = pipe.poll()
 948         input_stream.terminate()
 949         if returncode is None:
 950             pipe.terminate()
 951             self.fail("arv-put did not exit within 5 seconds")
 952         self.assertRegex(pipe.stdout.read().decode(), r'\(X-Request-Id: req-[a-z0-9]{20}\)')
 953
 954     def test_ArvPutSignedManifest(self):
 955         # ArvPutSignedManifest runs "arv-put foo" and then attempts to get
 956         # the newly created manifest from the API server, testing to confirm
 957         # that the block locators in the returned manifest are signed.
 958         self.authorize_with('active')
 959
 960         # Before doing anything, demonstrate that the collection
 961         # we're about to create is not present in our test fixture.
 962         manifest_uuid = "00b4e9f40ac4dd432ef89749f1c01e74+47"
 963         with self.assertRaises(apiclient.errors.HttpError):
 964             arv_put.api_client.collections().get(
 965                 uuid=manifest_uuid).execute()
 966
 967         datadir = self.make_tmpdir()
 968         with open(os.path.join(datadir, "foo"), "w") as f:
 969             f.write("The quick brown fox jumped over the lazy dog")
 970         p = subprocess.Popen([sys.executable, arv_put.__file__,
 971                               os.path.join(datadir, 'foo')],
 972                              stdout=subprocess.PIPE,
 973                              stderr=subprocess.PIPE,
 974                              env=self.ENVIRON)
 975         (_, err) = p.communicate()
 976         self.assertRegex(err.decode(), r'INFO: Collection saved as ')
 977         self.assertEqual(p.returncode, 0)
 978
 979         # The manifest text stored in the API server under the same
 980         # manifest UUID must use signed locators.
 981         c = arv_put.api_client.collections().get(uuid=manifest_uuid).execute()
 982         self.assertRegex(
 983             c['manifest_text'],
 984             r'^\. 08a008a01d498c404b0c30852b39d3b8\+44\+A[0-9a-f]+@[0-9a-f]+ 0:44:foo\n')
 985
 986         os.remove(os.path.join(datadir, "foo"))
 987         os.rmdir(datadir)
 988
 989     def run_and_find_collection(self, text, extra_args=[]):
 990         self.authorize_with('active')
 991         pipe = subprocess.Popen(
 992             [sys.executable, arv_put.__file__] + extra_args,
 993             stdin=subprocess.PIPE, stdout=subprocess.PIPE,
 994             stderr=subprocess.PIPE, env=self.ENVIRON)
 995         stdout, stderr = pipe.communicate(text.encode())
 996         self.assertRegex(stderr.decode(), r'INFO: Collection (updated:|saved as)')
 997         search_key = ('portable_data_hash'
 998                       if '--portable-data-hash' in extra_args else 'uuid')
 999         collection_list = arvados.api('v1').collections().list(
1000             filters=[[search_key, '=', stdout.decode().strip()]]
1001         ).execute().get('items', [])
1002         self.assertEqual(1, len(collection_list))
1003         return collection_list[0]
1004
1005     def test_all_expired_signatures_invalidates_cache(self):
1006         self.authorize_with('active')
1007         tmpdir = self.make_tmpdir()
1008         with open(os.path.join(tmpdir, 'somefile.txt'), 'w') as f:
1009             f.write('foo')
1010         # Upload a directory and get the cache file name
1011         p = subprocess.Popen([sys.executable, arv_put.__file__, tmpdir],
1012                              stdout=subprocess.PIPE,
1013                              stderr=subprocess.PIPE,
1014                              env=self.ENVIRON)
1015         (_, err) = p.communicate()
1016         self.assertRegex(err.decode(), r'INFO: Creating new cache file at ')
1017         self.assertEqual(p.returncode, 0)
1018         cache_filepath = re.search(r'INFO: Creating new cache file at (.*)',
1019                                    err.decode()).groups()[0]
1020         self.assertTrue(os.path.isfile(cache_filepath))
1021         # Load the cache file contents and modify the manifest to simulate
1022         # an expired access token
1023         with open(cache_filepath, 'r') as c:
1024             cache = json.load(c)
1025         self.assertRegex(cache['manifest'], r'\+A\S+\@')
1026         a_month_ago = datetime.datetime.now() - datetime.timedelta(days=30)
1027         cache['manifest'] = re.sub(
1028             r'\@.*? ',
1029             "@{} ".format(self.datetime_to_hex(a_month_ago)),
1030             cache['manifest'])
1031         with open(cache_filepath, 'w') as c:
1032             c.write(json.dumps(cache))
1033         # Re-run the upload and expect to get an invalid cache message
1034         p = subprocess.Popen([sys.executable, arv_put.__file__, tmpdir],
1035                              stdout=subprocess.PIPE,
1036                              stderr=subprocess.PIPE,
1037                              env=self.ENVIRON)
1038         (_, err) = p.communicate()
1039         self.assertRegex(
1040             err.decode(),
1041             r'INFO: Cache expired, starting from scratch.*')
1042         self.assertEqual(p.returncode, 0)
1043
1044     def test_invalid_signature_invalidates_cache(self):
1045         self.authorize_with('active')
1046         tmpdir = self.make_tmpdir()
1047         with open(os.path.join(tmpdir, 'somefile.txt'), 'w') as f:
1048             f.write('foo')
1049         # Upload a directory and get the cache file name
1050         p = subprocess.Popen([sys.executable, arv_put.__file__, tmpdir],
1051                              stdout=subprocess.PIPE,
1052                              stderr=subprocess.PIPE,
1053                              env=self.ENVIRON)
1054         (_, err) = p.communicate()
1055         self.assertRegex(err.decode(), r'INFO: Creating new cache file at ')
1056         self.assertEqual(p.returncode, 0)
1057         cache_filepath = re.search(r'INFO: Creating new cache file at (.*)',
1058                                    err.decode()).groups()[0]
1059         self.assertTrue(os.path.isfile(cache_filepath))
1060         # Load the cache file contents and modify the manifest to simulate
1061         # an invalid access token
1062         with open(cache_filepath, 'r') as c:
1063             cache = json.load(c)
1064         self.assertRegex(cache['manifest'], r'\+A\S+\@')
1065         cache['manifest'] = re.sub(
1066             r'\+A.*\@',
1067             "+Aabcdef0123456789abcdef0123456789abcdef01@",
1068             cache['manifest'])
1069         with open(cache_filepath, 'w') as c:
1070             c.write(json.dumps(cache))
1071         # Re-run the upload and expect to get an invalid cache message
1072         p = subprocess.Popen([sys.executable, arv_put.__file__, tmpdir],
1073                              stdout=subprocess.PIPE,
1074                              stderr=subprocess.PIPE,
1075                              env=self.ENVIRON)
1076         (_, err) = p.communicate()
1077         self.assertRegex(
1078             err.decode(),
1079             r'ERROR: arv-put: Resume cache contains invalid signature.*')
1080         self.assertEqual(p.returncode, 1)
1081
1082     def test_single_expired_signature_reuploads_file(self):
1083         self.authorize_with('active')
1084         tmpdir = self.make_tmpdir()
1085         with open(os.path.join(tmpdir, 'foofile.txt'), 'w') as f:
1086             f.write('foo')
1087         # Write a second file on its own subdir to force a new stream
1088         os.mkdir(os.path.join(tmpdir, 'bar'))
1089         with open(os.path.join(tmpdir, 'bar', 'barfile.txt'), 'w') as f:
1090             f.write('bar')
1091         # Upload a directory and get the cache file name
1092         p = subprocess.Popen([sys.executable, arv_put.__file__, tmpdir],
1093                              stdout=subprocess.PIPE,
1094                              stderr=subprocess.PIPE,
1095                              env=self.ENVIRON)
1096         (_, err) = p.communicate()
1097         self.assertRegex(err.decode(), r'INFO: Creating new cache file at ')
1098         self.assertEqual(p.returncode, 0)
1099         cache_filepath = re.search(r'INFO: Creating new cache file at (.*)',
1100                                    err.decode()).groups()[0]
1101         self.assertTrue(os.path.isfile(cache_filepath))
1102         # Load the cache file contents and modify the manifest to simulate
1103         # an expired access token
1104         with open(cache_filepath, 'r') as c:
1105             cache = json.load(c)
1106         self.assertRegex(cache['manifest'], r'\+A\S+\@')
1107         a_month_ago = datetime.datetime.now() - datetime.timedelta(days=30)
1108         # Make one of the signatures appear to have expired
1109         cache['manifest'] = re.sub(
1110             r'\@.*? 3:3:barfile.txt',
1111             "@{} 3:3:barfile.txt".format(self.datetime_to_hex(a_month_ago)),
1112             cache['manifest'])
1113         with open(cache_filepath, 'w') as c:
1114             c.write(json.dumps(cache))
1115         # Re-run the upload and expect to get an invalid cache message
1116         p = subprocess.Popen([sys.executable, arv_put.__file__, tmpdir],
1117                              stdout=subprocess.PIPE,
1118                              stderr=subprocess.PIPE,
1119                              env=self.ENVIRON)
1120         (_, err) = p.communicate()
1121         self.assertRegex(
1122             err.decode(),
1123             r'WARNING: Uploaded file \'.*barfile.txt\' access token expired, will re-upload it from scratch')
1124         self.assertEqual(p.returncode, 0)
1125         # Confirm that the resulting cache is different from the last run.
1126         with open(cache_filepath, 'r') as c2:
1127             new_cache = json.load(c2)
1128         self.assertNotEqual(cache['manifest'], new_cache['manifest'])
1129
1130     def test_put_collection_with_later_update(self):
1131         tmpdir = self.make_tmpdir()
1132         with open(os.path.join(tmpdir, 'file1'), 'w') as f:
1133             f.write('Relaxing in basins at the end of inlets terminates the endless tests from the box')
1134         col = self.run_and_find_collection("", ['--no-progress', tmpdir])
1135         self.assertNotEqual(None, col['uuid'])
1136         # Add a new file to the directory
1137         with open(os.path.join(tmpdir, 'file2'), 'w') as f:
1138             f.write('The quick brown fox jumped over the lazy dog')
1139         updated_col = self.run_and_find_collection("", ['--no-progress', '--update-collection', col['uuid'], tmpdir])
1140         self.assertEqual(col['uuid'], updated_col['uuid'])
1141         # Get the manifest and check that the new file is being included
1142         c = arv_put.api_client.collections().get(uuid=updated_col['uuid']).execute()
1143         self.assertRegex(c['manifest_text'], r'^\..* .*:44:file2\n')
1144
1145     def test_put_collection_with_utc_expiring_datetime(self):
1146         tmpdir = self.make_tmpdir()
1147         trash_at = (datetime.datetime.utcnow() + datetime.timedelta(days=90)).strftime('%Y%m%dT%H%MZ')
1148         with open(os.path.join(tmpdir, 'file1'), 'w') as f:
1149             f.write('Relaxing in basins at the end of inlets terminates the endless tests from the box')
1150         col = self.run_and_find_collection(
1151             "",
1152             ['--no-progress', '--trash-at', trash_at, tmpdir])
1153         self.assertNotEqual(None, col['uuid'])
1154         c = arv_put.api_client.collections().get(uuid=col['uuid']).execute()
1155         self.assertEqual(ciso8601.parse_datetime(trash_at),
1156             ciso8601.parse_datetime(c['trash_at']))
1157
1158     def test_put_collection_with_timezone_aware_expiring_datetime(self):
1159         tmpdir = self.make_tmpdir()
1160         trash_at = (datetime.datetime.utcnow() + datetime.timedelta(days=90)).strftime('%Y%m%dT%H%M-0300')
1161         with open(os.path.join(tmpdir, 'file1'), 'w') as f:
1162             f.write('Relaxing in basins at the end of inlets terminates the endless tests from the box')
1163         col = self.run_and_find_collection(
1164             "",
1165             ['--no-progress', '--trash-at', trash_at, tmpdir])
1166         self.assertNotEqual(None, col['uuid'])
1167         c = arv_put.api_client.collections().get(uuid=col['uuid']).execute()
1168         self.assertEqual(
1169             ciso8601.parse_datetime(trash_at).replace(tzinfo=None) + datetime.timedelta(hours=3),
1170             ciso8601.parse_datetime(c['trash_at']).replace(tzinfo=None))
1171
1172     def test_put_collection_with_timezone_naive_expiring_datetime(self):
1173         tmpdir = self.make_tmpdir()
1174         trash_at = (datetime.datetime.utcnow() + datetime.timedelta(days=90)).strftime('%Y%m%dT%H%M')
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         if time.daylight:
1183             offset = datetime.timedelta(seconds=time.altzone)
1184         else:
1185             offset = datetime.timedelta(seconds=time.timezone)
1186         self.assertEqual(
1187             ciso8601.parse_datetime(trash_at) + offset,
1188             ciso8601.parse_datetime(c['trash_at']).replace(tzinfo=None))
1189
1190     def test_put_collection_with_expiring_date_only(self):
1191         tmpdir = self.make_tmpdir()
1192         trash_at = '2140-01-01'
1193         end_of_day = datetime.timedelta(hours=23, minutes=59, seconds=59)
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) + end_of_day + offset,
1207             ciso8601.parse_datetime(c['trash_at']).replace(tzinfo=None))
1208
1209     def test_put_collection_with_invalid_absolute_expiring_datetimes(self):
1210         cases = ['2100', '210010','2100-10', '2100-Oct']
1211         tmpdir = self.make_tmpdir()
1212         with open(os.path.join(tmpdir, 'file1'), 'w') as f:
1213             f.write('Relaxing in basins at the end of inlets terminates the endless tests from the box')
1214         for test_datetime in cases:
1215             with self.assertRaises(AssertionError):
1216                 self.run_and_find_collection(
1217                     "",
1218                     ['--no-progress', '--trash-at', test_datetime, tmpdir])
1219
1220     def test_put_collection_with_relative_expiring_datetime(self):
1221         expire_after = 7
1222         dt_before = datetime.datetime.utcnow() + datetime.timedelta(days=expire_after)
1223         tmpdir = self.make_tmpdir()
1224         with open(os.path.join(tmpdir, 'file1'), 'w') as f:
1225             f.write('Relaxing in basins at the end of inlets terminates the endless tests from the box')
1226         col = self.run_and_find_collection(
1227             "",
1228             ['--no-progress', '--trash-after', str(expire_after), tmpdir])
1229         self.assertNotEqual(None, col['uuid'])
1230         dt_after = datetime.datetime.utcnow() + datetime.timedelta(days=expire_after)
1231         c = arv_put.api_client.collections().get(uuid=col['uuid']).execute()
1232         trash_at = ciso8601.parse_datetime(c['trash_at']).replace(tzinfo=None)
1233         self.assertTrue(dt_before < trash_at)
1234         self.assertTrue(dt_after > trash_at)
1235
1236     def test_put_collection_with_invalid_relative_expiring_datetime(self):
1237         expire_after = 0 # Must be >= 1
1238         tmpdir = self.make_tmpdir()
1239         with open(os.path.join(tmpdir, 'file1'), 'w') as f:
1240             f.write('Relaxing in basins at the end of inlets terminates the endless tests from the box')
1241         with self.assertRaises(AssertionError):
1242             self.run_and_find_collection(
1243                 "",
1244                 ['--no-progress', '--trash-after', str(expire_after), tmpdir])
1245
1246     def test_upload_directory_reference_without_trailing_slash(self):
1247         tmpdir1 = self.make_tmpdir()
1248         tmpdir2 = self.make_tmpdir()
1249         with open(os.path.join(tmpdir1, 'foo'), 'w') as f:
1250             f.write('This is foo')
1251         with open(os.path.join(tmpdir2, 'bar'), 'w') as f:
1252             f.write('This is not foo')
1253         # Upload one directory and one file
1254         col = self.run_and_find_collection("", ['--no-progress',
1255                                                 tmpdir1,
1256                                                 os.path.join(tmpdir2, 'bar')])
1257         self.assertNotEqual(None, col['uuid'])
1258         c = arv_put.api_client.collections().get(uuid=col['uuid']).execute()
1259         # Check that 'foo' was written inside a subcollection
1260         # OTOH, 'bar' should have been directly uploaded on the root collection
1261         self.assertRegex(c['manifest_text'], r'^\. .*:15:bar\n\./.+ .*:11:foo\n')
1262
1263     def test_upload_directory_reference_with_trailing_slash(self):
1264         tmpdir1 = self.make_tmpdir()
1265         tmpdir2 = self.make_tmpdir()
1266         with open(os.path.join(tmpdir1, 'foo'), 'w') as f:
1267             f.write('This is foo')
1268         with open(os.path.join(tmpdir2, 'bar'), 'w') as f:
1269             f.write('This is not foo')
1270         # Upload one directory (with trailing slash) and one file
1271         col = self.run_and_find_collection("", ['--no-progress',
1272                                                 tmpdir1 + os.sep,
1273                                                 os.path.join(tmpdir2, 'bar')])
1274         self.assertNotEqual(None, col['uuid'])
1275         c = arv_put.api_client.collections().get(uuid=col['uuid']).execute()
1276         # Check that 'foo' and 'bar' were written at the same level
1277         self.assertRegex(c['manifest_text'], r'^\. .*:15:bar .*:11:foo\n')
1278
1279     def test_put_collection_with_high_redundancy(self):
1280         # Write empty data: we're not testing CollectionWriter, just
1281         # making sure collections.create tells the API server what our
1282         # desired replication level is.
1283         collection = self.run_and_find_collection("", ['--replication', '4'])
1284         self.assertEqual(4, collection['replication_desired'])
1285
1286     def test_put_collection_with_default_redundancy(self):
1287         collection = self.run_and_find_collection("")
1288         self.assertEqual(None, collection['replication_desired'])
1289
1290     def test_put_collection_with_unnamed_project_link(self):
1291         link = self.run_and_find_collection(
1292             "Test unnamed collection",
1293             ['--portable-data-hash', '--project-uuid', self.PROJECT_UUID])
1294         username = pwd.getpwuid(os.getuid()).pw_name
1295         self.assertRegex(
1296             link['name'],
1297             r'^Saved at .* by {}@'.format(re.escape(username)))
1298
1299     def test_put_collection_with_name_and_no_project(self):
1300         link_name = 'Test Collection Link in home project'
1301         collection = self.run_and_find_collection(
1302             "Test named collection in home project",
1303             ['--portable-data-hash', '--name', link_name])
1304         self.assertEqual(link_name, collection['name'])
1305         my_user_uuid = self.current_user()['uuid']
1306         self.assertEqual(my_user_uuid, collection['owner_uuid'])
1307
1308     def test_put_collection_with_named_project_link(self):
1309         link_name = 'Test auto Collection Link'
1310         collection = self.run_and_find_collection("Test named collection",
1311                                       ['--portable-data-hash',
1312                                        '--name', link_name,
1313                                        '--project-uuid', self.PROJECT_UUID])
1314         self.assertEqual(link_name, collection['name'])
1315
1316     def test_put_collection_with_storage_classes_specified(self):
1317         collection = self.run_and_find_collection("", ['--storage-classes', 'hot'])
1318
1319         self.assertEqual(len(collection['storage_classes_desired']), 1)
1320         self.assertEqual(collection['storage_classes_desired'][0], 'hot')
1321
1322     def test_put_collection_without_storage_classes_specified(self):
1323         collection = self.run_and_find_collection("")
1324
1325         self.assertEqual(len(collection['storage_classes_desired']), 1)
1326         self.assertEqual(collection['storage_classes_desired'][0], 'default')
1327
1328     def test_exclude_filename_pattern(self):
1329         tmpdir = self.make_tmpdir()
1330         tmpsubdir = os.path.join(tmpdir, 'subdir')
1331         os.mkdir(tmpsubdir)
1332         for fname in ['file1', 'file2', 'file3']:
1333             with open(os.path.join(tmpdir, "%s.txt" % fname), 'w') as f:
1334                 f.write("This is %s" % fname)
1335             with open(os.path.join(tmpsubdir, "%s.txt" % fname), 'w') as f:
1336                 f.write("This is %s" % fname)
1337         col = self.run_and_find_collection("", ['--no-progress',
1338                                                 '--exclude', '*2.txt',
1339                                                 '--exclude', 'file3.*',
1340                                                  tmpdir])
1341         self.assertNotEqual(None, col['uuid'])
1342         c = arv_put.api_client.collections().get(uuid=col['uuid']).execute()
1343         # None of the file2.txt & file3.txt should have been uploaded
1344         self.assertRegex(c['manifest_text'], r'^.*:file1.txt')
1345         self.assertNotRegex(c['manifest_text'], r'^.*:file2.txt')
1346         self.assertNotRegex(c['manifest_text'], r'^.*:file3.txt')
1347
1348     def test_exclude_filepath_pattern(self):
1349         tmpdir = self.make_tmpdir()
1350         tmpsubdir = os.path.join(tmpdir, 'subdir')
1351         os.mkdir(tmpsubdir)
1352         for fname in ['file1', 'file2', 'file3']:
1353             with open(os.path.join(tmpdir, "%s.txt" % fname), 'w') as f:
1354                 f.write("This is %s" % fname)
1355             with open(os.path.join(tmpsubdir, "%s.txt" % fname), 'w') as f:
1356                 f.write("This is %s" % fname)
1357         col = self.run_and_find_collection("", ['--no-progress',
1358                                                 '--exclude', 'subdir/*2.txt',
1359                                                 '--exclude', './file1.*',
1360                                                  tmpdir])
1361         self.assertNotEqual(None, col['uuid'])
1362         c = arv_put.api_client.collections().get(uuid=col['uuid']).execute()
1363         # Only tmpdir/file1.txt & tmpdir/subdir/file2.txt should have been excluded
1364         self.assertNotRegex(c['manifest_text'],
1365                             r'^\./%s.*:file1.txt' % os.path.basename(tmpdir))
1366         self.assertNotRegex(c['manifest_text'],
1367                             r'^\./%s/subdir.*:file2.txt' % os.path.basename(tmpdir))
1368         self.assertRegex(c['manifest_text'],
1369                          r'^\./%s.*:file2.txt' % os.path.basename(tmpdir))
1370         self.assertRegex(c['manifest_text'], r'^.*:file3.txt')
1371
1372     def test_unicode_on_filename(self):
1373         tmpdir = self.make_tmpdir()
1374         fname = u"i❤arvados.txt"
1375         with open(os.path.join(tmpdir, fname), 'w') as f:
1376             f.write("This is a unicode named file")
1377         col = self.run_and_find_collection("", ['--no-progress', tmpdir])
1378         self.assertNotEqual(None, col['uuid'])
1379         c = arv_put.api_client.collections().get(uuid=col['uuid']).execute()
1380         self.assertTrue(fname in c['manifest_text'], u"{} does not include {}".format(c['manifest_text'], fname))
1381
1382     def test_silent_mode_no_errors(self):
1383         self.authorize_with('active')
1384         tmpdir = self.make_tmpdir()
1385         with open(os.path.join(tmpdir, 'test.txt'), 'w') as f:
1386             f.write('hello world')
1387         pipe = subprocess.Popen(
1388             [sys.executable, arv_put.__file__] + ['--silent', tmpdir],
1389             stdin=subprocess.PIPE, stdout=subprocess.PIPE,
1390             stderr=subprocess.PIPE, env=self.ENVIRON)
1391         stdout, stderr = pipe.communicate()
1392         # No console output should occur on normal operations
1393         self.assertNotRegex(stderr.decode(), r'.+')
1394         self.assertNotRegex(stdout.decode(), r'.+')
1395
1396     def test_silent_mode_does_not_avoid_error_messages(self):
1397         self.authorize_with('active')
1398         pipe = subprocess.Popen(
1399             [sys.executable, arv_put.__file__] + ['--silent',
1400                                                   '/path/not/existant'],
1401             stdin=subprocess.PIPE, stdout=subprocess.PIPE,
1402             stderr=subprocess.PIPE, env=self.ENVIRON)
1403         stdout, stderr = pipe.communicate()
1404         # Error message should be displayed when errors happen
1405         self.assertRegex(stderr.decode(), r'.*ERROR:.*')
1406         self.assertNotRegex(stdout.decode(), r'.+')
1407
1408
1409 if __name__ == '__main__':
1410     unittest.main()