Merge branch '21718-memoryview-decode' refs #21718

[arvados.git] / sdk / python / tests / test_arv_copy.py

diff --git a/sdk/python/tests/test_arv_copy.py b/sdk/python/tests/test_arv_copy.py
index 4be212c73b28dd36c75d23c2136c6ed74e9ff4e3..1af5c68e6c94934b57364b33aaacc4ee35307d2e 100644 (file)
--- a/sdk/python/tests/test_arv_copy.py
+++ b/sdk/python/tests/test_arv_copy.py
@@ -1,45 +1,89 @@
-#!/usr/bin/env python
-# -*- coding: utf-8 -*-
+# Copyright (C) The Arvados Authors. All rights reserved.
+#
+# SPDX-License-Identifier: Apache-2.0
 
-import multiprocessing
 import os
 import sys
 import tempfile
 import unittest
+import shutil
+import arvados.api
+from arvados.collection import Collection, CollectionReader
 
 import arvados.commands.arv_copy as arv_copy
+from . import arvados_testutil as tutil
+from . import run_test_server
+
+class ArvCopyVersionTestCase(run_test_server.TestCaseWithServers, tutil.VersionChecker):
+    MAIN_SERVER = {}
+    KEEP_SERVER = {}
 
-class ArvCopyTestCase(unittest.TestCase):
     def run_copy(self, args):
         sys.argv = ['arv-copy'] + args
         return arv_copy.main()
 
-    def run_copy_process(self, args):
-        _, stdout_path = tempfile.mkstemp()
-        _, stderr_path = tempfile.mkstemp()
-        def wrap():
-            def wrapper():
-                sys.argv = ['arv-copy'] + args
-                sys.stdout = open(stdout_path, 'w')
-                sys.stderr = open(stderr_path, 'w')
-                arv_copy.main()
-            return wrapper
-        p = multiprocessing.Process(target=wrap())
-        p.start()
-        p.join()
-        out = open(stdout_path, 'r').read()
-        err = open(stderr_path, 'r').read()
-        os.unlink(stdout_path)
-        os.unlink(stderr_path)
-        return p.exitcode, out, err
-
     def test_unsupported_arg(self):
         with self.assertRaises(SystemExit):
             self.run_copy(['-x=unknown'])
 
     def test_version_argument(self):
-        exitcode, out, err = self.run_copy_process(['--version'])
-        self.assertEqual(0, exitcode)
-        self.assertEqual('', out)
-        self.assertNotEqual('', err)
-        self.assertRegexpMatches(err, "[0-9]+\.[0-9]+\.[0-9]+")
+        with tutil.redirected_streams(
+                stdout=tutil.StringIO, stderr=tutil.StringIO) as (out, err):
+            with self.assertRaises(SystemExit):
+                self.run_copy(['--version'])
+        self.assertVersionOutput(out, err)
+
+    def test_copy_project(self):
+        api = arvados.api()
+        src_proj = api.groups().create(body={"group": {"name": "arv-copy project", "group_class": "project"}}).execute()["uuid"]
+
+        c = Collection()
+        with c.open('foo', 'wt') as f:
+            f.write('foo')
+        c.save_new("arv-copy foo collection", owner_uuid=src_proj)
+        coll_record = api.collections().get(uuid=c.manifest_locator()).execute()
+        assert coll_record['storage_classes_desired'] == ['default']
+
+        dest_proj = api.groups().create(body={"group": {"name": "arv-copy dest project", "group_class": "project"}}).execute()["uuid"]
+
+        tmphome = tempfile.mkdtemp()
+        home_was = os.environ['HOME']
+        os.environ['HOME'] = tmphome
+        try:
+            cfgdir = os.path.join(tmphome, ".config", "arvados")
+            os.makedirs(cfgdir)
+            with open(os.path.join(cfgdir, "zzzzz.conf"), "wt") as f:
+                f.write("ARVADOS_API_HOST=%s\n" % os.environ["ARVADOS_API_HOST"])
+                f.write("ARVADOS_API_TOKEN=%s\n" % os.environ["ARVADOS_API_TOKEN"])
+                f.write("ARVADOS_API_HOST_INSECURE=1\n")
+
+            contents = api.groups().list(filters=[["owner_uuid", "=", dest_proj]]).execute()
+            assert len(contents["items"]) == 0
+
+            with tutil.redirected_streams(
+                    stdout=tutil.StringIO, stderr=tutil.StringIO) as (out, err):
+                try:
+                    self.run_copy(["--project-uuid", dest_proj, "--storage-classes", "foo", src_proj])
+                except SystemExit as e:
+                    assert e.code == 0
+                copy_uuid_from_stdout = out.getvalue().strip()
+
+            contents = api.groups().list(filters=[["owner_uuid", "=", dest_proj]]).execute()
+            assert len(contents["items"]) == 1
+
+            assert contents["items"][0]["name"] == "arv-copy project"
+            copied_project = contents["items"][0]["uuid"]
+
+            assert copied_project == copy_uuid_from_stdout
+
+            contents = api.collections().list(filters=[["owner_uuid", "=", copied_project]]).execute()
+            assert len(contents["items"]) == 1
+
+            assert contents["items"][0]["uuid"] != c.manifest_locator()
+            assert contents["items"][0]["name"] == "arv-copy foo collection"
+            assert contents["items"][0]["portable_data_hash"] == c.portable_data_hash()
+            assert contents["items"][0]["storage_classes_desired"] == ["foo"]
+
+        finally:
+            os.environ['HOME'] = home_was
+            shutil.rmtree(tmphome)