services/fuse/tests/test_command_args.py

   1 import arvados
   2 import arvados_fuse
   3 import arvados_fuse.command
   4 import contextlib
   5 import functools
   6 import json
   7 import llfuse
   8 import logging
   9 import os
  10 import run_test_server
  11 import sys
  12 import tempfile
  13 import unittest
  14
  15 def noexit(func):
  16     """If argparse or arvados_fuse tries to exit, fail the test instead"""
  17     class SystemExitCaught(StandardError):
  18         pass
  19     @functools.wraps(func)
  20     def wrapper(*args, **kwargs):
  21         try:
  22             return func(*args, **kwargs)
  23         except SystemExit:
  24             raise SystemExitCaught
  25     return wrapper
  26
  27 @contextlib.contextmanager
  28 def nostderr():
  29     orig, sys.stderr = sys.stderr, open(os.devnull, 'w')
  30     try:
  31         yield
  32     finally:
  33         sys.stderr = orig
  34
  35
  36 class MountArgsTest(unittest.TestCase):
  37     def setUp(self):
  38         self.mntdir = tempfile.mkdtemp()
  39         run_test_server.authorize_with('active')
  40
  41     def tearDown(self):
  42         os.rmdir(self.mntdir)
  43
  44     def lookup(self, mnt, *path):
  45         ent = mnt.operations.inodes[llfuse.ROOT_INODE]
  46         for p in path:
  47             ent = ent[p]
  48         return ent
  49
  50     def check_ent_type(self, cls, *path):
  51         ent = self.lookup(self.mnt, *path)
  52         self.assertEqual(ent.__class__, cls)
  53         return ent
  54
  55     @noexit
  56     def test_default_all(self):
  57         args = arvados_fuse.command.ArgumentParser().parse_args([
  58             '--foreground', self.mntdir])
  59         self.assertEqual(args.mode, None)
  60         self.mnt = arvados_fuse.command.Mount(args)
  61         e = self.check_ent_type(arvados_fuse.ProjectDirectory, 'home')
  62         self.assertEqual(e.project_object['uuid'],
  63                          run_test_server.fixture('users')['active']['uuid'])
  64         e = self.check_ent_type(arvados_fuse.MagicDirectory, 'by_id')
  65
  66         e = self.check_ent_type(arvados_fuse.StringFile, 'README')
  67         readme = e.readfrom(0, -1)
  68         self.assertRegexpMatches(readme, r'active-user@arvados\.local')
  69         self.assertRegexpMatches(readme, r'\n$')
  70
  71         e = self.check_ent_type(arvados_fuse.StringFile, 'by_id', 'README')
  72         txt = e.readfrom(0, -1)
  73         self.assertRegexpMatches(txt, r'portable data hash')
  74         self.assertRegexpMatches(txt, r'\n$')
  75
  76     @noexit
  77     def test_by_id(self):
  78         args = arvados_fuse.command.ArgumentParser().parse_args([
  79             '--by-id',
  80             '--foreground', self.mntdir])
  81         self.assertEqual(args.mode, 'by_id')
  82         self.mnt = arvados_fuse.command.Mount(args)
  83         e = self.check_ent_type(arvados_fuse.MagicDirectory)
  84         self.assertEqual(e.pdh_only, False)
  85
  86     @noexit
  87     def test_by_pdh(self):
  88         args = arvados_fuse.command.ArgumentParser().parse_args([
  89             '--by-pdh',
  90             '--foreground', self.mntdir])
  91         self.assertEqual(args.mode, 'by_pdh')
  92         self.mnt = arvados_fuse.command.Mount(args)
  93         e = self.check_ent_type(arvados_fuse.MagicDirectory)
  94         self.assertEqual(e.pdh_only, True)
  95
  96     @noexit
  97     def test_by_tag(self):
  98         args = arvados_fuse.command.ArgumentParser().parse_args([
  99             '--by-tag',
 100             '--foreground', self.mntdir])
 101         self.assertEqual(args.mode, 'by_tag')
 102         self.mnt = arvados_fuse.command.Mount(args)
 103         e = self.check_ent_type(arvados_fuse.TagsDirectory)
 104
 105     @noexit
 106     def test_collection(self, id_type='uuid'):
 107         c = run_test_server.fixture('collections')['public_text_file']
 108         cid = c[id_type]
 109         args = arvados_fuse.command.ArgumentParser().parse_args([
 110             '--collection', cid,
 111             '--foreground', self.mntdir])
 112         self.mnt = arvados_fuse.command.Mount(args)
 113         e = self.check_ent_type(arvados_fuse.CollectionDirectory)
 114         self.assertEqual(e.collection_locator, cid)
 115
 116     def test_collection_pdh(self):
 117         self.test_collection('portable_data_hash')
 118
 119     @noexit
 120     def test_home(self):
 121         args = arvados_fuse.command.ArgumentParser().parse_args([
 122             '--home',
 123             '--foreground', self.mntdir])
 124         self.assertEqual(args.mode, 'home')
 125         self.mnt = arvados_fuse.command.Mount(args)
 126         e = self.check_ent_type(arvados_fuse.ProjectDirectory)
 127         self.assertEqual(e.project_object['uuid'],
 128                          run_test_server.fixture('users')['active']['uuid'])
 129
 130     def test_mutually_exclusive_args(self):
 131         cid = run_test_server.fixture('collections')['public_text_file']['uuid']
 132         gid = run_test_server.fixture('groups')['aproject']['uuid']
 133         for badargs in [
 134                 ['--mount-tmp', 'foo', '--collection', cid],
 135                 ['--mount-tmp', 'foo', '--project', gid],
 136                 ['--collection', cid, '--project', gid],
 137                 ['--by-id', '--project', gid],
 138                 ['--mount-tmp', 'foo', '--by-id'],
 139         ]:
 140             with nostderr():
 141                 with self.assertRaises(SystemExit):
 142                     args = arvados_fuse.command.ArgumentParser().parse_args(
 143                         badargs + ['--foreground', self.mntdir])
 144                     arvados_fuse.command.Mount(args)
 145     @noexit
 146     def test_project(self):
 147         uuid = run_test_server.fixture('groups')['aproject']['uuid']
 148         args = arvados_fuse.command.ArgumentParser().parse_args([
 149             '--project', uuid,
 150             '--foreground', self.mntdir])
 151         self.mnt = arvados_fuse.command.Mount(args)
 152         e = self.check_ent_type(arvados_fuse.ProjectDirectory)
 153         self.assertEqual(e.project_object['uuid'], uuid)
 154
 155     @noexit
 156     def test_shared(self):
 157         args = arvados_fuse.command.ArgumentParser().parse_args([
 158             '--shared',
 159             '--foreground', self.mntdir])
 160         self.assertEqual(args.mode, 'shared')
 161         self.mnt = arvados_fuse.command.Mount(args)
 162         e = self.check_ent_type(arvados_fuse.SharedDirectory)
 163         self.assertEqual(e.current_user['uuid'],
 164                          run_test_server.fixture('users')['active']['uuid'])
 165
 166     @noexit
 167     def test_custom(self):
 168         args = arvados_fuse.command.ArgumentParser().parse_args([
 169             '--mount-tmp', 'foo',
 170             '--mount-tmp', 'bar',
 171             '--mount-home', 'my_home',
 172             '--foreground', self.mntdir])
 173         self.assertEqual(args.mode, None)
 174         self.mnt = arvados_fuse.command.Mount(args)
 175         self.check_ent_type(arvados_fuse.Directory)
 176         self.check_ent_type(arvados_fuse.TmpCollectionDirectory, 'foo')
 177         self.check_ent_type(arvados_fuse.TmpCollectionDirectory, 'bar')
 178         e = self.check_ent_type(arvados_fuse.ProjectDirectory, 'my_home')
 179         self.assertEqual(e.project_object['uuid'],
 180                          run_test_server.fixture('users')['active']['uuid'])
 181
 182     def test_custom_unsupported_layouts(self):
 183         for name in ['.', '..', '', 'foo/bar', '/foo']:
 184             with nostderr():
 185                 with self.assertRaises(SystemExit):
 186                     args = arvados_fuse.command.ArgumentParser().parse_args([
 187                         '--mount-tmp', name,
 188                         '--foreground', self.mntdir])
 189                     arvados_fuse.command.Mount(args)