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 import itertools
   6 import os
   7 import sys
   8 import tempfile
   9 import unittest
  10 import shutil
  11 import arvados.api
  12 import arvados.util
  13 from arvados.collection import Collection, CollectionReader
  14
  15 import pytest
  16
  17 import arvados.commands.arv_copy as arv_copy
  18 from arvados._internal import basedirs
  19 from . import arvados_testutil as tutil
  20 from . import run_test_server
  21
  22 class ArvCopyVersionTestCase(run_test_server.TestCaseWithServers, tutil.VersionChecker):
  23     MAIN_SERVER = {}
  24     KEEP_SERVER = {}
  25
  26     def run_copy(self, args):
  27         sys.argv = ['arv-copy'] + args
  28         return arv_copy.main()
  29
  30     def test_unsupported_arg(self):
  31         with self.assertRaises(SystemExit):
  32             self.run_copy(['-x=unknown'])
  33
  34     def test_version_argument(self):
  35         with tutil.redirected_streams(
  36                 stdout=tutil.StringIO, stderr=tutil.StringIO) as (out, err):
  37             with self.assertRaises(SystemExit):
  38                 self.run_copy(['--version'])
  39         self.assertVersionOutput(out, err)
  40
  41     def test_copy_project(self):
  42         api = arvados.api()
  43         src_proj = api.groups().create(body={"group": {"name": "arv-copy project", "group_class": "project"}}).execute()["uuid"]
  44
  45         c = Collection()
  46         with c.open('foo', 'wt') as f:
  47             f.write('foo')
  48         c.save_new("arv-copy foo collection", owner_uuid=src_proj)
  49         coll_record = api.collections().get(uuid=c.manifest_locator()).execute()
  50         assert coll_record['storage_classes_desired'] == ['default']
  51
  52         dest_proj = api.groups().create(body={"group": {"name": "arv-copy dest project", "group_class": "project"}}).execute()["uuid"]
  53
  54         tmphome = tempfile.mkdtemp()
  55         home_was = os.environ['HOME']
  56         os.environ['HOME'] = tmphome
  57         try:
  58             cfgdir = os.path.join(tmphome, ".config", "arvados")
  59             os.makedirs(cfgdir)
  60             with open(os.path.join(cfgdir, "zzzzz.conf"), "wt") as f:
  61                 f.write("ARVADOS_API_HOST=%s\n" % os.environ["ARVADOS_API_HOST"])
  62                 f.write("ARVADOS_API_TOKEN=%s\n" % os.environ["ARVADOS_API_TOKEN"])
  63                 f.write("ARVADOS_API_HOST_INSECURE=1\n")
  64
  65             contents = api.groups().list(filters=[["owner_uuid", "=", dest_proj]]).execute()
  66             assert len(contents["items"]) == 0
  67
  68             with tutil.redirected_streams(
  69                     stdout=tutil.StringIO, stderr=tutil.StringIO) as (out, err):
  70                 try:
  71                     self.run_copy(["--project-uuid", dest_proj, "--storage-classes", "foo", src_proj])
  72                 except SystemExit as e:
  73                     assert e.code == 0
  74                 copy_uuid_from_stdout = out.getvalue().strip()
  75
  76             contents = api.groups().list(filters=[["owner_uuid", "=", dest_proj]]).execute()
  77             assert len(contents["items"]) == 1
  78
  79             assert contents["items"][0]["name"] == "arv-copy project"
  80             copied_project = contents["items"][0]["uuid"]
  81
  82             assert copied_project == copy_uuid_from_stdout
  83
  84             contents = api.collections().list(filters=[["owner_uuid", "=", copied_project]]).execute()
  85             assert len(contents["items"]) == 1
  86
  87             assert contents["items"][0]["uuid"] != c.manifest_locator()
  88             assert contents["items"][0]["name"] == "arv-copy foo collection"
  89             assert contents["items"][0]["portable_data_hash"] == c.portable_data_hash()
  90             assert contents["items"][0]["storage_classes_desired"] == ["foo"]
  91
  92         finally:
  93             os.environ['HOME'] = home_was
  94             shutil.rmtree(tmphome)
  95
  96
  97 class TestApiForInstance:
  98     _token_counter = itertools.count(1)
  99
 100     @staticmethod
 101     def api_config(version, **kwargs):
 102         assert version == 'v1'
 103         return kwargs
 104
 105     @pytest.fixture
 106     def patch_api(self, monkeypatch):
 107         monkeypatch.setattr(arvados, 'api', self.api_config)
 108
 109     @pytest.fixture
 110     def config_file(self, tmp_path):
 111         count = next(self._token_counter)
 112         path = tmp_path / f'config{count}.conf'
 113         with path.open('w') as config_file:
 114             print(
 115                 "ARVADOS_API_HOST=localhost",
 116                 f"ARVADOS_API_TOKEN={self.expected_token(path)}",
 117                 sep="\n", file=config_file,
 118             )
 119         return path
 120
 121     @pytest.fixture
 122     def patch_search(self, tmp_path, monkeypatch):
 123         def search(self, name):
 124             path = tmp_path / name
 125             if path.exists():
 126                 yield path
 127         monkeypatch.setattr(basedirs.BaseDirectories, 'search', search)
 128
 129     def expected_token(self, path):
 130         return f"v2/zzzzz-gj3su-{path.stem:>015s}/{path.stem:>050s}"
 131
 132     def test_from_environ(self, patch_api):
 133         actual = arv_copy.api_for_instance('', 0)
 134         assert actual == {}
 135
 136     def test_relative_path(self, patch_api, config_file, monkeypatch):
 137         monkeypatch.chdir(config_file.parent)
 138         actual = arv_copy.api_for_instance(f'./{config_file.name}', 0)
 139         assert actual['host'] == 'localhost'
 140         assert actual['token'] == self.expected_token(config_file)
 141
 142     def test_absolute_path(self, patch_api, config_file):
 143         actual = arv_copy.api_for_instance(str(config_file), 0)
 144         assert actual['host'] == 'localhost'
 145         assert actual['token'] == self.expected_token(config_file)
 146
 147     def test_search_path(self, patch_api, patch_search, config_file):
 148         actual = arv_copy.api_for_instance(config_file.stem, 0)
 149         assert actual['host'] == 'localhost'
 150         assert actual['token'] == self.expected_token(config_file)
 151
 152     def test_search_failed(self, patch_api, patch_search):
 153         with pytest.raises(SystemExit) as exc_info:
 154             arv_copy.api_for_instance('NotFound', 0)
 155         assert exc_info.value.code > 0
 156
 157     def test_path_unreadable(self, patch_api, tmp_path):
 158         with pytest.raises(SystemExit) as exc_info:
 159             arv_copy.api_for_instance(str(tmp_path / 'nonexistent.conf'), 0)
 160         assert exc_info.value.code > 0