1 # Copyright (C) The Arvados Authors. All rights reserved.
3 # SPDX-License-Identifier: Apache-2.0
13 from arvados.collection import Collection, CollectionReader
17 import arvados.commands.arv_copy as arv_copy
18 from . import arvados_testutil as tutil
19 from . import run_test_server
21 class ArvCopyVersionTestCase(run_test_server.TestCaseWithServers, tutil.VersionChecker):
25 def run_copy(self, args):
26 sys.argv = ['arv-copy'] + args
27 return arv_copy.main()
29 def test_unsupported_arg(self):
30 with self.assertRaises(SystemExit):
31 self.run_copy(['-x=unknown'])
33 def test_version_argument(self):
34 with tutil.redirected_streams(
35 stdout=tutil.StringIO, stderr=tutil.StringIO) as (out, err):
36 with self.assertRaises(SystemExit):
37 self.run_copy(['--version'])
38 self.assertVersionOutput(out, err)
40 def test_copy_project(self):
42 src_proj = api.groups().create(body={"group": {"name": "arv-copy project", "group_class": "project"}}).execute()["uuid"]
45 with c.open('foo', 'wt') as f:
47 c.save_new("arv-copy foo collection", owner_uuid=src_proj)
48 coll_record = api.collections().get(uuid=c.manifest_locator()).execute()
49 assert coll_record['storage_classes_desired'] == ['default']
51 dest_proj = api.groups().create(body={"group": {"name": "arv-copy dest project", "group_class": "project"}}).execute()["uuid"]
53 tmphome = tempfile.mkdtemp()
54 home_was = os.environ['HOME']
55 os.environ['HOME'] = tmphome
57 cfgdir = os.path.join(tmphome, ".config", "arvados")
59 with open(os.path.join(cfgdir, "zzzzz.conf"), "wt") as f:
60 f.write("ARVADOS_API_HOST=%s\n" % os.environ["ARVADOS_API_HOST"])
61 f.write("ARVADOS_API_TOKEN=%s\n" % os.environ["ARVADOS_API_TOKEN"])
62 f.write("ARVADOS_API_HOST_INSECURE=1\n")
64 contents = api.groups().list(filters=[["owner_uuid", "=", dest_proj]]).execute()
65 assert len(contents["items"]) == 0
67 with tutil.redirected_streams(
68 stdout=tutil.StringIO, stderr=tutil.StringIO) as (out, err):
70 self.run_copy(["--project-uuid", dest_proj, "--storage-classes", "foo", src_proj])
71 except SystemExit as e:
73 copy_uuid_from_stdout = out.getvalue().strip()
75 contents = api.groups().list(filters=[["owner_uuid", "=", dest_proj]]).execute()
76 assert len(contents["items"]) == 1
78 assert contents["items"][0]["name"] == "arv-copy project"
79 copied_project = contents["items"][0]["uuid"]
81 assert copied_project == copy_uuid_from_stdout
83 contents = api.collections().list(filters=[["owner_uuid", "=", copied_project]]).execute()
84 assert len(contents["items"]) == 1
86 assert contents["items"][0]["uuid"] != c.manifest_locator()
87 assert contents["items"][0]["name"] == "arv-copy foo collection"
88 assert contents["items"][0]["portable_data_hash"] == c.portable_data_hash()
89 assert contents["items"][0]["storage_classes_desired"] == ["foo"]
92 os.environ['HOME'] = home_was
93 shutil.rmtree(tmphome)
96 class TestApiForInstance:
97 _token_counter = itertools.count(1)
100 def api_config(version, **kwargs):
101 assert version == 'v1'
105 def patch_api(self, monkeypatch):
106 monkeypatch.setattr(arvados, 'api', self.api_config)
109 def config_file(self, tmp_path):
110 count = next(self._token_counter)
111 path = tmp_path / f'config{count}.conf'
112 with path.open('w') as config_file:
114 "ARVADOS_API_HOST=localhost",
115 f"ARVADOS_API_TOKEN={self.expected_token(path)}",
116 sep="\n", file=config_file,
121 def patch_search(self, tmp_path, monkeypatch):
122 def search(self, name):
123 path = tmp_path / name
126 monkeypatch.setattr(arvados.util._BaseDirectories, 'search', search)
128 def expected_token(self, path):
129 return f"v2/zzzzz-gj3su-{path.stem:>015s}/{path.stem:>050s}"
131 def test_from_environ(self, patch_api):
132 actual = arv_copy.api_for_instance('', 0)
135 def test_relative_path(self, patch_api, config_file, monkeypatch):
136 monkeypatch.chdir(config_file.parent)
137 actual = arv_copy.api_for_instance(f'./{config_file.name}', 0)
138 assert actual['host'] == 'localhost'
139 assert actual['token'] == self.expected_token(config_file)
141 def test_absolute_path(self, patch_api, config_file):
142 actual = arv_copy.api_for_instance(str(config_file), 0)
143 assert actual['host'] == 'localhost'
144 assert actual['token'] == self.expected_token(config_file)
146 def test_search_path(self, patch_api, patch_search, config_file):
147 actual = arv_copy.api_for_instance(config_file.stem, 0)
148 assert actual['host'] == 'localhost'
149 assert actual['token'] == self.expected_token(config_file)
151 def test_search_failed(self, patch_api, patch_search):
152 with pytest.raises(SystemExit) as exc_info:
153 arv_copy.api_for_instance('NotFound', 0)
154 assert exc_info.value.code > 0
156 def test_path_unreadable(self, patch_api, tmp_path):
157 with pytest.raises(SystemExit) as exc_info:
158 arv_copy.api_for_instance(str(tmp_path / 'nonexistent.conf'), 0)
159 assert exc_info.value.code > 0