3499: Reconcile Python tests to new arv-put "home project" behavior.
[arvados.git] / sdk / python / tests / test_arv_put.py
1 #!/usr/bin/env python
2 # -*- coding: utf-8 -*-
3
4 import apiclient
5 import os
6 import pwd
7 import re
8 import shutil
9 import subprocess
10 import sys
11 import tempfile
12 import time
13 import unittest
14 import yaml
15
16 from cStringIO import StringIO
17
18 import arvados
19 import arvados.commands.put as arv_put
20
21 from arvados_testutil import ArvadosBaseTestCase, ArvadosKeepLocalStoreTestCase
22 import run_test_server
23
24 class ArvadosPutResumeCacheTest(ArvadosBaseTestCase):
25     CACHE_ARGSET = [
26         [],
27         ['/dev/null'],
28         ['/dev/null', '--filename', 'empty'],
29         ['/tmp'],
30         ['/tmp', '--max-manifest-depth', '0'],
31         ['/tmp', '--max-manifest-depth', '1']
32         ]
33
34     def tearDown(self):
35         super(ArvadosPutResumeCacheTest, self).tearDown()
36         try:
37             self.last_cache.destroy()
38         except AttributeError:
39             pass
40
41     def cache_path_from_arglist(self, arglist):
42         return arv_put.ResumeCache.make_path(arv_put.parse_arguments(arglist))
43
44     def test_cache_names_stable(self):
45         for argset in self.CACHE_ARGSET:
46             self.assertEquals(self.cache_path_from_arglist(argset),
47                               self.cache_path_from_arglist(argset),
48                               "cache name changed for {}".format(argset))
49
50     def test_cache_names_unique(self):
51         results = []
52         for argset in self.CACHE_ARGSET:
53             path = self.cache_path_from_arglist(argset)
54             self.assertNotIn(path, results)
55             results.append(path)
56
57     def test_cache_names_simple(self):
58         # The goal here is to make sure the filename doesn't use characters
59         # reserved by the filesystem.  Feel free to adjust this regexp as
60         # long as it still does that.
61         bad_chars = re.compile(r'[^-\.\w]')
62         for argset in self.CACHE_ARGSET:
63             path = self.cache_path_from_arglist(argset)
64             self.assertFalse(bad_chars.search(os.path.basename(path)),
65                              "path too exotic: {}".format(path))
66
67     def test_cache_names_ignore_argument_order(self):
68         self.assertEquals(
69             self.cache_path_from_arglist(['a', 'b', 'c']),
70             self.cache_path_from_arglist(['c', 'a', 'b']))
71         self.assertEquals(
72             self.cache_path_from_arglist(['-', '--filename', 'stdin']),
73             self.cache_path_from_arglist(['--filename', 'stdin', '-']))
74
75     def test_cache_names_differ_for_similar_paths(self):
76         # This test needs names at / that don't exist on the real filesystem.
77         self.assertNotEqual(
78             self.cache_path_from_arglist(['/_arvputtest1', '/_arvputtest2']),
79             self.cache_path_from_arglist(['/_arvputtest1/_arvputtest2']))
80
81     def test_cache_names_ignore_irrelevant_arguments(self):
82         # Workaround: parse_arguments bails on --filename with a directory.
83         path1 = self.cache_path_from_arglist(['/tmp'])
84         args = arv_put.parse_arguments(['/tmp'])
85         args.filename = 'tmp'
86         path2 = arv_put.ResumeCache.make_path(args)
87         self.assertEquals(path1, path2,
88                          "cache path considered --filename for directory")
89         self.assertEquals(
90             self.cache_path_from_arglist(['-']),
91             self.cache_path_from_arglist(['-', '--max-manifest-depth', '1']),
92             "cache path considered --max-manifest-depth for file")
93
94     def test_cache_names_treat_negative_manifest_depths_identically(self):
95         base_args = ['/tmp', '--max-manifest-depth']
96         self.assertEquals(
97             self.cache_path_from_arglist(base_args + ['-1']),
98             self.cache_path_from_arglist(base_args + ['-2']))
99
100     def test_cache_names_treat_stdin_consistently(self):
101         self.assertEquals(
102             self.cache_path_from_arglist(['-', '--filename', 'test']),
103             self.cache_path_from_arglist(['/dev/stdin', '--filename', 'test']))
104
105     def test_cache_names_identical_for_synonymous_names(self):
106         self.assertEquals(
107             self.cache_path_from_arglist(['.']),
108             self.cache_path_from_arglist([os.path.realpath('.')]))
109         testdir = self.make_tmpdir()
110         looplink = os.path.join(testdir, 'loop')
111         os.symlink(testdir, looplink)
112         self.assertEquals(
113             self.cache_path_from_arglist([testdir]),
114             self.cache_path_from_arglist([looplink]))
115
116     def test_cache_names_different_by_api_host(self):
117         config = arvados.config.settings()
118         orig_host = config.get('ARVADOS_API_HOST')
119         try:
120             name1 = self.cache_path_from_arglist(['.'])
121             config['ARVADOS_API_HOST'] = 'x' + (orig_host or 'localhost')
122             self.assertNotEqual(name1, self.cache_path_from_arglist(['.']))
123         finally:
124             if orig_host is None:
125                 del config['ARVADOS_API_HOST']
126             else:
127                 config['ARVADOS_API_HOST'] = orig_host
128
129     def test_basic_cache_storage(self):
130         thing = ['test', 'list']
131         with tempfile.NamedTemporaryFile() as cachefile:
132             self.last_cache = arv_put.ResumeCache(cachefile.name)
133         self.last_cache.save(thing)
134         self.assertEquals(thing, self.last_cache.load())
135
136     def test_empty_cache(self):
137         with tempfile.NamedTemporaryFile() as cachefile:
138             cache = arv_put.ResumeCache(cachefile.name)
139         self.assertRaises(ValueError, cache.load)
140
141     def test_cache_persistent(self):
142         thing = ['test', 'list']
143         path = os.path.join(self.make_tmpdir(), 'cache')
144         cache = arv_put.ResumeCache(path)
145         cache.save(thing)
146         cache.close()
147         self.last_cache = arv_put.ResumeCache(path)
148         self.assertEquals(thing, self.last_cache.load())
149
150     def test_multiple_cache_writes(self):
151         thing = ['short', 'list']
152         with tempfile.NamedTemporaryFile() as cachefile:
153             self.last_cache = arv_put.ResumeCache(cachefile.name)
154         # Start writing an object longer than the one we test, to make
155         # sure the cache file gets truncated.
156         self.last_cache.save(['long', 'long', 'list'])
157         self.last_cache.save(thing)
158         self.assertEquals(thing, self.last_cache.load())
159
160     def test_cache_is_locked(self):
161         with tempfile.NamedTemporaryFile() as cachefile:
162             cache = arv_put.ResumeCache(cachefile.name)
163             self.assertRaises(arv_put.ResumeCacheConflict,
164                               arv_put.ResumeCache, cachefile.name)
165
166     def test_cache_stays_locked(self):
167         with tempfile.NamedTemporaryFile() as cachefile:
168             self.last_cache = arv_put.ResumeCache(cachefile.name)
169             path = cachefile.name
170         self.last_cache.save('test')
171         self.assertRaises(arv_put.ResumeCacheConflict,
172                           arv_put.ResumeCache, path)
173
174     def test_destroy_cache(self):
175         cachefile = tempfile.NamedTemporaryFile(delete=False)
176         try:
177             cache = arv_put.ResumeCache(cachefile.name)
178             cache.save('test')
179             cache.destroy()
180             try:
181                 arv_put.ResumeCache(cachefile.name)
182             except arv_put.ResumeCacheConflict:
183                 self.fail("could not load cache after destroying it")
184             self.assertRaises(ValueError, cache.load)
185         finally:
186             if os.path.exists(cachefile.name):
187                 os.unlink(cachefile.name)
188
189     def test_restart_cache(self):
190         path = os.path.join(self.make_tmpdir(), 'cache')
191         cache = arv_put.ResumeCache(path)
192         cache.save('test')
193         cache.restart()
194         self.assertRaises(ValueError, cache.load)
195         self.assertRaises(arv_put.ResumeCacheConflict,
196                           arv_put.ResumeCache, path)
197
198
199 class ArvadosPutCollectionWriterTest(ArvadosKeepLocalStoreTestCase):
200     def setUp(self):
201         super(ArvadosPutCollectionWriterTest, self).setUp()
202         with tempfile.NamedTemporaryFile(delete=False) as cachefile:
203             self.cache = arv_put.ResumeCache(cachefile.name)
204             self.cache_filename = cachefile.name
205
206     def tearDown(self):
207         super(ArvadosPutCollectionWriterTest, self).tearDown()
208         if os.path.exists(self.cache_filename):
209             self.cache.destroy()
210         self.cache.close()
211
212     def test_writer_caches(self):
213         cwriter = arv_put.ArvPutCollectionWriter(self.cache)
214         cwriter.write_file('/dev/null')
215         cwriter.cache_state()
216         self.assertTrue(self.cache.load())
217         self.assertEquals(". d41d8cd98f00b204e9800998ecf8427e+0 0:0:null\n", cwriter.manifest_text())
218
219     def test_writer_works_without_cache(self):
220         cwriter = arv_put.ArvPutCollectionWriter()
221         cwriter.write_file('/dev/null')
222         self.assertEquals(". d41d8cd98f00b204e9800998ecf8427e+0 0:0:null\n", cwriter.manifest_text())
223
224     def test_writer_resumes_from_cache(self):
225         cwriter = arv_put.ArvPutCollectionWriter(self.cache)
226         with self.make_test_file() as testfile:
227             cwriter.write_file(testfile.name, 'test')
228             cwriter.cache_state()
229             new_writer = arv_put.ArvPutCollectionWriter.from_cache(
230                 self.cache)
231             self.assertEquals(
232                 ". 098f6bcd4621d373cade4e832627b4f6+4 0:4:test\n",
233                 new_writer.manifest_text())
234
235     def test_new_writer_from_stale_cache(self):
236         cwriter = arv_put.ArvPutCollectionWriter(self.cache)
237         with self.make_test_file() as testfile:
238             cwriter.write_file(testfile.name, 'test')
239         new_writer = arv_put.ArvPutCollectionWriter.from_cache(self.cache)
240         new_writer.write_file('/dev/null')
241         self.assertEquals(". d41d8cd98f00b204e9800998ecf8427e+0 0:0:null\n", new_writer.manifest_text())
242
243     def test_new_writer_from_empty_cache(self):
244         cwriter = arv_put.ArvPutCollectionWriter.from_cache(self.cache)
245         cwriter.write_file('/dev/null')
246         self.assertEquals(". d41d8cd98f00b204e9800998ecf8427e+0 0:0:null\n", cwriter.manifest_text())
247
248     def test_writer_resumable_after_arbitrary_bytes(self):
249         cwriter = arv_put.ArvPutCollectionWriter(self.cache)
250         # These bytes are intentionally not valid UTF-8.
251         with self.make_test_file('\x00\x07\xe2') as testfile:
252             cwriter.write_file(testfile.name, 'test')
253             cwriter.cache_state()
254             new_writer = arv_put.ArvPutCollectionWriter.from_cache(
255                 self.cache)
256         self.assertEquals(cwriter.manifest_text(), new_writer.manifest_text())
257
258     def make_progress_tester(self):
259         progression = []
260         def record_func(written, expected):
261             progression.append((written, expected))
262         return progression, record_func
263
264     def test_progress_reporting(self):
265         for expect_count in (None, 8):
266             progression, reporter = self.make_progress_tester()
267             cwriter = arv_put.ArvPutCollectionWriter(
268                 reporter=reporter, bytes_expected=expect_count)
269             with self.make_test_file() as testfile:
270                 cwriter.write_file(testfile.name, 'test')
271             cwriter.finish_current_stream()
272             self.assertIn((4, expect_count), progression)
273
274     def test_resume_progress(self):
275         cwriter = arv_put.ArvPutCollectionWriter(self.cache, bytes_expected=4)
276         with self.make_test_file() as testfile:
277             # Set up a writer with some flushed bytes.
278             cwriter.write_file(testfile.name, 'test')
279             cwriter.finish_current_stream()
280             cwriter.cache_state()
281             new_writer = arv_put.ArvPutCollectionWriter.from_cache(self.cache)
282             self.assertEqual(new_writer.bytes_written, 4)
283
284
285 class ArvadosExpectedBytesTest(ArvadosBaseTestCase):
286     TEST_SIZE = os.path.getsize(__file__)
287
288     def test_expected_bytes_for_file(self):
289         self.assertEquals(self.TEST_SIZE,
290                           arv_put.expected_bytes_for([__file__]))
291
292     def test_expected_bytes_for_tree(self):
293         tree = self.make_tmpdir()
294         shutil.copyfile(__file__, os.path.join(tree, 'one'))
295         shutil.copyfile(__file__, os.path.join(tree, 'two'))
296         self.assertEquals(self.TEST_SIZE * 2,
297                           arv_put.expected_bytes_for([tree]))
298         self.assertEquals(self.TEST_SIZE * 3,
299                           arv_put.expected_bytes_for([tree, __file__]))
300
301     def test_expected_bytes_for_device(self):
302         self.assertIsNone(arv_put.expected_bytes_for(['/dev/null']))
303         self.assertIsNone(arv_put.expected_bytes_for([__file__, '/dev/null']))
304
305
306 class ArvadosPutReportTest(ArvadosBaseTestCase):
307     def test_machine_progress(self):
308         for count, total in [(0, 1), (0, None), (1, None), (235, 9283)]:
309             expect = ": {} written {} total\n".format(
310                 count, -1 if (total is None) else total)
311             self.assertTrue(
312                 arv_put.machine_progress(count, total).endswith(expect))
313
314     def test_known_human_progress(self):
315         for count, total in [(0, 1), (2, 4), (45, 60)]:
316             expect = '{:.1%}'.format(float(count) / total)
317             actual = arv_put.human_progress(count, total)
318             self.assertTrue(actual.startswith('\r'))
319             self.assertIn(expect, actual)
320
321     def test_unknown_human_progress(self):
322         for count in [1, 20, 300, 4000, 50000]:
323             self.assertTrue(re.search(r'\b{}\b'.format(count),
324                                       arv_put.human_progress(count, None)))
325
326
327 class ArvadosPutProjectLinkTest(ArvadosBaseTestCase):
328     Z_UUID = 'zzzzz-zzzzz-zzzzzzzzzzzzzzz'
329
330     def setUp(self):
331         self.stderr = StringIO()
332         super(ArvadosPutProjectLinkTest, self).setUp()
333
334     def tearDown(self):
335         self.stderr.close()
336         super(ArvadosPutProjectLinkTest, self).tearDown()
337
338     def prep_link_from_arguments(self, args, uuid_found=True):
339         try:
340             link = arv_put.prep_project_link(arv_put.parse_arguments(args),
341                                              self.stderr,
342                                              lambda uuid: uuid_found)
343         finally:
344             self.stderr.seek(0)
345         return link
346
347     def check_link(self, link, project_uuid, link_name=None):
348         self.assertEqual(project_uuid, link.get('tail_uuid'))
349         self.assertEqual(project_uuid, link.get('owner_uuid'))
350         self.assertEqual('name', link.get('link_class'))
351         if link_name is None:
352             self.assertNotIn('name', link)
353         else:
354             self.assertEqual(link_name, link.get('name'))
355         self.assertNotIn('head_uuid', link)
356
357     def check_stderr_empty(self):
358         self.assertEqual('', self.stderr.getvalue())
359
360     def test_project_link_with_name(self):
361         link = self.prep_link_from_arguments(['--project-uuid', self.Z_UUID,
362                                               '--name', 'test link AAA'])
363         self.check_link(link, self.Z_UUID, 'test link AAA')
364         self.check_stderr_empty()
365
366     def test_project_link_without_name(self):
367         username = pwd.getpwuid(os.getuid()).pw_name
368         link = self.prep_link_from_arguments(['--project-uuid', self.Z_UUID])
369         self.assertIsNotNone(link.get('name', None))
370         self.assertRegexpMatches(
371             link['name'],
372             r'^Saved at .* by {}@'.format(re.escape(username)))
373         self.check_link(link, self.Z_UUID, link.get('name', None))
374         for line in self.stderr:
375             if "No --name specified" in line:
376                 break
377         else:
378             self.fail("no warning emitted about the lack of collection name")
379
380     @unittest.skip("prep_project_link needs an API lookup for this case")
381     def test_collection_without_project_defaults_to_home(self):
382         link = self.prep_link_from_arguments(['--name', 'test link BBB'])
383         self.check_link(link, self.Z_UUID)
384         self.check_stderr_empty()
385
386     def test_no_link_or_warning_with_no_collection(self):
387         self.assertIsNone(self.prep_link_from_arguments(['--raw']))
388         self.check_stderr_empty()
389
390     def test_error_when_project_not_found(self):
391         self.assertRaises(ValueError,
392                           self.prep_link_from_arguments,
393                           ['--project-uuid', self.Z_UUID], False)
394
395     def test_link_without_collection_is_error(self):
396         self.assertRaises(ValueError,
397                           self.prep_link_from_arguments,
398                           ['--project-uuid', self.Z_UUID, '--stream'])
399
400
401 class ArvadosPutTest(ArvadosKeepLocalStoreTestCase):
402     def call_main_with_args(self, args):
403         self.main_stdout = StringIO()
404         self.main_stderr = StringIO()
405         return arv_put.main(args, self.main_stdout, self.main_stderr)
406
407     def call_main_on_test_file(self):
408         with self.make_test_file() as testfile:
409             path = testfile.name
410             self.call_main_with_args(['--stream', '--no-progress', path])
411         self.assertTrue(
412             os.path.exists(os.path.join(os.environ['KEEP_LOCAL_STORE'],
413                                         '098f6bcd4621d373cade4e832627b4f6')),
414             "did not find file stream in Keep store")
415
416     def tearDown(self):
417         for outbuf in ['main_stdout', 'main_stderr']:
418             if hasattr(self, outbuf):
419                 getattr(self, outbuf).close()
420                 delattr(self, outbuf)
421         super(ArvadosPutTest, self).tearDown()
422
423     def test_simple_file_put(self):
424         self.call_main_on_test_file()
425
426     def test_put_with_unwriteable_cache_dir(self):
427         orig_cachedir = arv_put.ResumeCache.CACHE_DIR
428         cachedir = self.make_tmpdir()
429         os.chmod(cachedir, 0o0)
430         arv_put.ResumeCache.CACHE_DIR = cachedir
431         try:
432             self.call_main_on_test_file()
433         finally:
434             arv_put.ResumeCache.CACHE_DIR = orig_cachedir
435             os.chmod(cachedir, 0o700)
436
437     def test_put_with_unwritable_cache_subdir(self):
438         orig_cachedir = arv_put.ResumeCache.CACHE_DIR
439         cachedir = self.make_tmpdir()
440         os.chmod(cachedir, 0o0)
441         arv_put.ResumeCache.CACHE_DIR = os.path.join(cachedir, 'cachedir')
442         try:
443             self.call_main_on_test_file()
444         finally:
445             arv_put.ResumeCache.CACHE_DIR = orig_cachedir
446             os.chmod(cachedir, 0o700)
447
448     def test_link_without_collection_aborts(self):
449         self.assertRaises(SystemExit, self.call_main_with_args,
450                           ['--name', 'test without Collection',
451                            '--stream', '/dev/null'])
452
453 class ArvPutIntegrationTest(unittest.TestCase):
454     PROJECT_UUID = run_test_server.fixture('groups')['aproject']['uuid']
455     ENVIRON = os.environ
456     ENVIRON['PYTHONPATH'] = ':'.join(sys.path)
457
458     @classmethod
459     def setUpClass(cls):
460         try:
461             del os.environ['KEEP_LOCAL_STORE']
462         except KeyError:
463             pass
464
465         # Use the blob_signing_key from the Rails "test" configuration
466         # to provision the Keep server.
467         config_blob_signing_key = None
468         for config_file in ['application.yml', 'application.default.yml']:
469             with open(os.path.join(os.path.dirname(__file__),
470                                    run_test_server.ARV_API_SERVER_DIR,
471                                    "config",
472                                    config_file)) as f:
473                 rails_config = yaml.load(f.read())
474                 for config_section in ['test', 'common']:
475                     try:
476                         config_blob_signing_key = rails_config[config_section]["blob_signing_key"]
477                         break
478                     except KeyError, AttributeError:
479                         pass
480             if config_blob_signing_key != None:
481                 break
482         run_test_server.run()
483         run_test_server.run_keep(blob_signing_key=config_blob_signing_key,
484                                  enforce_permissions=(config_blob_signing_key != None))
485
486     @classmethod
487     def tearDownClass(cls):
488         run_test_server.stop()
489         run_test_server.stop_keep()
490
491     def authorize_with(self, token_name):
492         run_test_server.authorize_with(token_name)
493         for v in ["ARVADOS_API_HOST",
494                   "ARVADOS_API_HOST_INSECURE",
495                   "ARVADOS_API_TOKEN"]:
496             os.environ[v] = arvados.config.settings()[v]
497
498     def current_user(self):
499         return arvados.api('v1').users().current().execute()
500
501     def test_check_real_project_found(self):
502         self.assertTrue(arv_put.check_project_exists(self.PROJECT_UUID),
503                         "did not correctly find test fixture project")
504
505     def test_check_error_finding_nonexistent_project(self):
506         BAD_UUID = 'zzzzz-zzzzz-zzzzzzzzzzzzzzz'
507         try:
508             result = arv_put.check_project_exists(BAD_UUID)
509         except ValueError as error:
510             self.assertIn(BAD_UUID, error.message)
511         else:
512             self.assertFalse(result, "incorrectly found nonexistent project")
513
514     def test_short_put_from_stdin(self):
515         # Have to run this as an integration test since arv-put can't
516         # read from the tests' stdin.
517         # arv-put usually can't stat(os.path.realpath('/dev/stdin')) in this
518         # case, because the /proc entry is already gone by the time it tries.
519         pipe = subprocess.Popen(
520             [sys.executable, arv_put.__file__, '--stream'],
521             stdin=subprocess.PIPE, stdout=subprocess.PIPE,
522             stderr=subprocess.STDOUT, env=self.ENVIRON)
523         pipe.stdin.write('stdin test\n')
524         pipe.stdin.close()
525         deadline = time.time() + 5
526         while (pipe.poll() is None) and (time.time() < deadline):
527             time.sleep(.1)
528         returncode = pipe.poll()
529         if returncode is None:
530             pipe.terminate()
531             self.fail("arv-put did not PUT from stdin within 5 seconds")
532         elif returncode != 0:
533             sys.stdout.write(pipe.stdout.read())
534             self.fail("arv-put returned exit code {}".format(returncode))
535         self.assertIn('4a9c8b735dce4b5fa3acf221a0b13628+11', pipe.stdout.read())
536
537     def test_ArvPutSignedManifest(self):
538         # ArvPutSignedManifest runs "arv-put foo" and then attempts to get
539         # the newly created manifest from the API server, testing to confirm
540         # that the block locators in the returned manifest are signed.
541         self.authorize_with('active')
542
543         # Before doing anything, demonstrate that the collection
544         # we're about to create is not present in our test fixture.
545         api = arvados.api('v1', cache=False)
546         manifest_uuid = "00b4e9f40ac4dd432ef89749f1c01e74+47"
547         with self.assertRaises(apiclient.errors.HttpError):
548             notfound = api.collections().get(uuid=manifest_uuid).execute()
549
550         datadir = tempfile.mkdtemp()
551         with open(os.path.join(datadir, "foo"), "w") as f:
552             f.write("The quick brown fox jumped over the lazy dog")
553         p = subprocess.Popen([sys.executable, arv_put.__file__, datadir],
554                              stdout=subprocess.PIPE, env=self.ENVIRON)
555         (arvout, arverr) = p.communicate()
556         self.assertEqual(p.returncode, 0)
557         self.assertEqual(arverr, None)
558         self.assertEqual(arvout.strip(), manifest_uuid)
559
560         # The manifest text stored in the API server under the same
561         # manifest UUID must use signed locators.
562         c = api.collections().get(uuid=manifest_uuid).execute()
563         self.assertRegexpMatches(
564             c['manifest_text'],
565             r'^\. 08a008a01d498c404b0c30852b39d3b8\+44\+A[0-9a-f]+@[0-9a-f]+ 0:44:foo\n')
566
567         os.remove(os.path.join(datadir, "foo"))
568         os.rmdir(datadir)
569
570     def run_and_find_link(self, text, extra_args=[]):
571         self.authorize_with('active')
572         pipe = subprocess.Popen(
573             [sys.executable, arv_put.__file__] + extra_args,
574             stdin=subprocess.PIPE, stdout=subprocess.PIPE,
575             stderr=subprocess.PIPE, env=self.ENVIRON)
576         stdout, stderr = pipe.communicate(text)
577         link_list = arvados.api('v1', cache=False).links().list(
578             filters=[['head_uuid', '=', stdout.strip()],
579                      ['link_class', '=', 'name']]).execute().get('items', [])
580         self.assertEqual(1, len(link_list))
581         return link_list[0]
582
583     def test_put_collection_with_unnamed_project_link(self):
584         link = self.run_and_find_link("Test unnamed collection",
585                                       ['--project-uuid', self.PROJECT_UUID])
586         username = pwd.getpwuid(os.getuid()).pw_name
587         self.assertRegexpMatches(
588             link['name'],
589             r'^Saved at .* by {}@'.format(re.escape(username)))
590
591     def test_put_collection_with_name_and_no_project(self):
592         link_name = 'Test Collection Link in home project'
593         link = self.run_and_find_link("Test named collection in home project",
594                                       ['--name', link_name])
595         self.assertEqual(link_name, link['name'])
596         my_user_uuid = self.current_user()['uuid']
597         self.assertEqual(my_user_uuid, link['tail_uuid'])
598         self.assertEqual(my_user_uuid, link['owner_uuid'])
599
600     def test_put_collection_with_named_project_link(self):
601         link_name = 'Test auto Collection Link'
602         link = self.run_and_find_link("Test named collection",
603                                       ['--name', link_name,
604                                        '--project-uuid', self.PROJECT_UUID])
605         self.assertEqual(link_name, link['name'])
606
607
608 if __name__ == '__main__':
609     unittest.main()