2 # -*- coding: utf-8 -*-
11 import arvados.collection as collection
12 import arvados.commands.get as arv_get
13 import run_test_server
15 from arvados_testutil import redirected_streams
17 class ArvadosGetTestCase(run_test_server.TestCaseWithServers):
22 super(ArvadosGetTestCase, self).setUp()
23 self.tempdir = tempfile.mkdtemp()
24 self.col_loc, self.col_pdh, self.col_manifest = self.write_test_collection()
27 super(ArvadosGetTestCase, self).tearDown()
28 shutil.rmtree(self.tempdir)
30 def write_test_collection(self,
35 'subdir/baz.txt' : 'baz',
37 c = collection.Collection()
38 for path, data in contents.items():
39 with c.open(path, 'w') as f:
42 return (c.manifest_locator(),
43 c.portable_data_hash(),
44 c.manifest_text(strip=strip_manifest))
46 def run_get(self, args):
47 self.stdout = io.BytesIO()
48 self.stderr = io.BytesIO()
49 return arv_get.main(args, self.stdout, self.stderr)
51 def test_version_argument(self):
54 with redirected_streams(stdout=out, stderr=err):
55 with self.assertRaises(SystemExit):
56 self.run_get(['--version'])
57 self.assertEqual(out.getvalue(), '')
58 self.assertRegexpMatches(err.getvalue(), "[0-9]+\.[0-9]+\.[0-9]+")
60 def test_get_single_file(self):
61 # Get the file using the collection's locator
62 r = self.run_get(["{}/subdir/baz.txt".format(self.col_loc), '-'])
63 self.assertEqual(0, r)
64 self.assertEqual('baz', self.stdout.getvalue())
66 r = self.run_get(["{}/subdir/baz.txt".format(self.col_pdh), '-'])
67 self.assertEqual(0, r)
68 self.assertEqual('baz', self.stdout.getvalue())
70 def test_get_multiple_files(self):
71 # Download the entire collection to the temp directory
72 r = self.run_get(["{}/".format(self.col_loc), self.tempdir])
73 self.assertEqual(0, r)
74 with open(os.path.join(self.tempdir, "foo.txt"), "r") as f:
75 self.assertEqual("foo", f.read())
76 with open(os.path.join(self.tempdir, "bar.txt"), "r") as f:
77 self.assertEqual("bar", f.read())
78 with open(os.path.join(self.tempdir, "subdir", "baz.txt"), "r") as f:
79 self.assertEqual("baz", f.read())
81 def test_get_collection_unstripped_manifest(self):
82 dummy_token = "+Axxxxxxx"
83 # Get the collection manifest by UUID
84 r = self.run_get([self.col_loc, self.tempdir])
85 self.assertEqual(0, r)
86 m_from_collection = re.sub(r"\+A[0-9a-f@]+", dummy_token, self.col_manifest)
87 with open(os.path.join(self.tempdir, self.col_loc), "r") as f:
88 # Replace manifest tokens before comparison to avoid races
89 m_from_file = re.sub(r"\+A[0-9a-f@]+", dummy_token, f.read())
90 self.assertEqual(m_from_collection, m_from_file)
91 # Get the collection manifest by PDH
92 r = self.run_get([self.col_pdh, self.tempdir])
93 self.assertEqual(0, r)
94 with open(os.path.join(self.tempdir, self.col_pdh), "r") as f:
95 # Replace manifest tokens before comparison to avoid races
96 m_from_file = re.sub(r"\+A[0-9a-f@]+", dummy_token, f.read())
97 self.assertEqual(m_from_collection, m_from_file)
99 def test_get_collection_stripped_manifest(self):
100 col_loc, col_pdh, col_manifest = self.write_test_collection(strip_manifest=True)
101 # Get the collection manifest by UUID
102 r = self.run_get(['--strip-manifest', col_loc, self.tempdir])
103 self.assertEqual(0, r)
104 with open(os.path.join(self.tempdir, col_loc), "r") as f:
105 self.assertEqual(col_manifest, f.read())
106 # Get the collection manifest by PDH
107 r = self.run_get(['--strip-manifest', col_pdh, self.tempdir])
108 self.assertEqual(0, r)
109 with open(os.path.join(self.tempdir, col_pdh), "r") as f:
110 self.assertEqual(col_manifest, f.read())
112 def test_invalid_collection(self):
113 # Asking for an invalid collection should generate an error.
114 r = self.run_get(['this-uuid-seems-to-be-fake', self.tempdir])
115 self.assertNotEqual(0, r)
117 def test_invalid_file_request(self):
118 # Asking for an inexistant file within a collection should generate an error.
119 r = self.run_get(["{}/im-not-here.txt".format(self.col_loc), self.tempdir])
120 self.assertNotEqual(0, r)
122 def test_invalid_destination(self):
123 # Asking to place the collection's files on a non existant directory
124 # should generate an error.
125 r = self.run_get([self.col_loc, "/fake/subdir/"])
126 self.assertNotEqual(0, r)
128 def test_preexistent_destination(self):
129 # Asking to place a file with the same path as a local one should
130 # generate an error and avoid overwrites.
131 with open(os.path.join(self.tempdir, "foo.txt"), "w") as f:
132 f.write("another foo")
133 r = self.run_get(["{}/foo.txt".format(self.col_loc), self.tempdir])
134 self.assertNotEqual(0, r)
135 with open(os.path.join(self.tempdir, "foo.txt"), "r") as f:
136 self.assertEqual("another foo", f.read())