sdk/python/tests/test_arv_copy.py

   1 # Copyright (C) The Arvados Authors. All rights reserved.
   2 #
   3 # SPDX-License-Identifier: Apache-2.0
   4
   5 from __future__ import absolute_import
   6 import os
   7 import sys
   8 import tempfile
   9 import unittest
  10 import shutil
  11 import arvados.api
  12 from arvados.collection import Collection, CollectionReader
  13
  14 import arvados.commands.arv_copy as arv_copy
  15 from . import arvados_testutil as tutil
  16 from . import run_test_server
  17
  18 class ArvCopyVersionTestCase(run_test_server.TestCaseWithServers, tutil.VersionChecker):
  19     MAIN_SERVER = {}
  20     KEEP_SERVER = {}
  21
  22     def run_copy(self, args):
  23         sys.argv = ['arv-copy'] + args
  24         return arv_copy.main()
  25
  26     def test_unsupported_arg(self):
  27         with self.assertRaises(SystemExit):
  28             self.run_copy(['-x=unknown'])
  29
  30     def test_version_argument(self):
  31         with tutil.redirected_streams(
  32                 stdout=tutil.StringIO, stderr=tutil.StringIO) as (out, err):
  33             with self.assertRaises(SystemExit):
  34                 self.run_copy(['--version'])
  35         self.assertVersionOutput(out, err)
  36
  37     def test_copy_project(self):
  38         api = arvados.api()
  39         src_proj = api.groups().create(body={"group": {"name": "arv-copy project", "group_class": "project"}}).execute()["uuid"]
  40
  41         c = Collection()
  42         with c.open('foo', 'wt') as f:
  43             f.write('foo')
  44         c.save_new("arv-copy foo collection", owner_uuid=src_proj)
  45         coll_record = api.collections().get(uuid=c.manifest_locator()).execute()
  46         assert coll_record['storage_classes_desired'] == ['default']
  47
  48         dest_proj = api.groups().create(body={"group": {"name": "arv-copy dest project", "group_class": "project"}}).execute()["uuid"]
  49
  50         tmphome = tempfile.mkdtemp()
  51         home_was = os.environ['HOME']
  52         os.environ['HOME'] = tmphome
  53         try:
  54             cfgdir = os.path.join(tmphome, ".config", "arvados")
  55             os.makedirs(cfgdir)
  56             with open(os.path.join(cfgdir, "zzzzz.conf"), "wt") as f:
  57                 f.write("ARVADOS_API_HOST=%s\n" % os.environ["ARVADOS_API_HOST"])
  58                 f.write("ARVADOS_API_TOKEN=%s\n" % os.environ["ARVADOS_API_TOKEN"])
  59                 f.write("ARVADOS_API_HOST_INSECURE=1\n")
  60
  61             contents = api.groups().list(filters=[["owner_uuid", "=", dest_proj]]).execute()
  62             assert len(contents["items"]) == 0
  63
  64             with tutil.redirected_streams(
  65                     stdout=tutil.StringIO, stderr=tutil.StringIO) as (out, err):
  66                 try:
  67                     self.run_copy(["--project-uuid", dest_proj, "--storage-classes", "foo", src_proj])
  68                 except SystemExit as e:
  69                     assert e.code == 0
  70                 copy_uuid_from_stdout = out.getvalue().strip()
  71
  72             contents = api.groups().list(filters=[["owner_uuid", "=", dest_proj]]).execute()
  73             assert len(contents["items"]) == 1
  74
  75             assert contents["items"][0]["name"] == "arv-copy project"
  76             copied_project = contents["items"][0]["uuid"]
  77
  78             assert copied_project == copy_uuid_from_stdout
  79
  80             contents = api.collections().list(filters=[["owner_uuid", "=", copied_project]]).execute()
  81             assert len(contents["items"]) == 1
  82
  83             assert contents["items"][0]["uuid"] != c.manifest_locator()
  84             assert contents["items"][0]["name"] == "arv-copy foo collection"
  85             assert contents["items"][0]["portable_data_hash"] == c.portable_data_hash()
  86             assert contents["items"][0]["storage_classes_desired"] == ["foo"]
  87
  88         finally:
  89             os.environ['HOME'] = home_was
  90             shutil.rmtree(tmphome)