+# Copyright (C) The Arvados Authors. All rights reserved.
+#
+# SPDX-License-Identifier: AGPL-3.0
+
+from __future__ import absolute_import
+from __future__ import print_function
+from six import assertRegex
import arvados
import arvados_fuse
import arvados_fuse.command
+import contextlib
import functools
+import io
import json
import llfuse
import logging
+import mock
import os
-import run_test_server
+from . import run_test_server
+import sys
import tempfile
import unittest
def noexit(func):
"""If argparse or arvados_fuse tries to exit, fail the test instead"""
- class SystemExitCaught(StandardError):
+ class SystemExitCaught(Exception):
pass
@functools.wraps(func)
def wrapper(*args, **kwargs):
raise SystemExitCaught
return wrapper
+@contextlib.contextmanager
+def nostderr():
+ with open(os.devnull, 'w') as dn:
+ orig, sys.stderr = sys.stderr, dn
+ try:
+ yield
+ finally:
+ sys.stderr = orig
+
+
class MountArgsTest(unittest.TestCase):
def setUp(self):
self.mntdir = tempfile.mkdtemp()
ent = ent[p]
return ent
+ @contextlib.contextmanager
+ def stderrMatches(self, stderr):
+ orig, sys.stderr = sys.stderr, stderr
+ try:
+ yield
+ finally:
+ sys.stderr = orig
+
def check_ent_type(self, cls, *path):
ent = self.lookup(self.mnt, *path)
self.assertEqual(ent.__class__, cls)
e = self.check_ent_type(arvados_fuse.MagicDirectory, 'by_id')
e = self.check_ent_type(arvados_fuse.StringFile, 'README')
- readme = e.readfrom(0, -1)
- self.assertRegexpMatches(readme, r'active-user@arvados\.local')
- self.assertRegexpMatches(readme, r'\n$')
+ readme = e.readfrom(0, -1).decode()
+ assertRegex(self, readme, r'active-user@arvados\.local')
+ assertRegex(self, readme, r'\n$')
e = self.check_ent_type(arvados_fuse.StringFile, 'by_id', 'README')
- txt = e.readfrom(0, -1)
- self.assertRegexpMatches(txt, r'portable data hash')
- self.assertRegexpMatches(txt, r'\n$')
+ txt = e.readfrom(0, -1).decode()
+ assertRegex(self, txt, r'portable data hash')
+ assertRegex(self, txt, r'\n$')
@noexit
def test_by_id(self):
self.mnt = arvados_fuse.command.Mount(args)
e = self.check_ent_type(arvados_fuse.MagicDirectory)
self.assertEqual(e.pdh_only, False)
+ self.assertEqual(True, self.mnt.listen_for_events)
@noexit
def test_by_pdh(self):
self.mnt = arvados_fuse.command.Mount(args)
e = self.check_ent_type(arvados_fuse.MagicDirectory)
self.assertEqual(e.pdh_only, True)
+ self.assertEqual(False, self.mnt.listen_for_events)
@noexit
def test_by_tag(self):
self.assertEqual(args.mode, 'by_tag')
self.mnt = arvados_fuse.command.Mount(args)
e = self.check_ent_type(arvados_fuse.TagsDirectory)
+ self.assertEqual(True, self.mnt.listen_for_events)
@noexit
def test_collection(self, id_type='uuid'):
self.mnt = arvados_fuse.command.Mount(args)
e = self.check_ent_type(arvados_fuse.CollectionDirectory)
self.assertEqual(e.collection_locator, cid)
+ self.assertEqual(id_type == 'uuid', self.mnt.listen_for_events)
def test_collection_pdh(self):
self.test_collection('portable_data_hash')
e = self.check_ent_type(arvados_fuse.ProjectDirectory)
self.assertEqual(e.project_object['uuid'],
run_test_server.fixture('users')['active']['uuid'])
+ self.assertEqual(True, self.mnt.listen_for_events)
def test_mutually_exclusive_args(self):
cid = run_test_server.fixture('collections')['public_text_file']['uuid']
['--by-id', '--project', gid],
['--mount-tmp', 'foo', '--by-id'],
]:
- with self.assertRaises(SystemExit):
- args = arvados_fuse.command.ArgumentParser().parse_args(
- badargs + ['--foreground', self.mntdir])
- arvados_fuse.command.Mount(args)
+ with nostderr():
+ with self.assertRaises(SystemExit):
+ args = arvados_fuse.command.ArgumentParser().parse_args(
+ badargs + ['--foreground', self.mntdir])
+ arvados_fuse.command.Mount(args)
@noexit
def test_project(self):
uuid = run_test_server.fixture('groups')['aproject']['uuid']
e = self.check_ent_type(arvados_fuse.SharedDirectory)
self.assertEqual(e.current_user['uuid'],
run_test_server.fixture('users')['active']['uuid'])
+ self.assertEqual(True, self.mnt.listen_for_events)
+
+ def test_version_argument(self):
+ # The argparse version action prints to stderr in Python 2 and stdout
+ # in Python 3.4 and up. Write both to the same stream so the test can pass
+ # in both cases.
+ origerr = sys.stderr
+ origout = sys.stdout
+ sys.stderr = io.StringIO()
+ sys.stdout = sys.stderr
+
+ with self.assertRaises(SystemExit):
+ args = arvados_fuse.command.ArgumentParser().parse_args(['--version'])
+ assertRegex(self, sys.stdout.getvalue(), "[0-9]+\.[0-9]+\.[0-9]+")
+ sys.stderr.close()
+ sys.stderr = origerr
+ sys.stdout = origout
@noexit
- def test_custom(self):
+ @mock.patch('arvados.events.subscribe')
+ def test_disable_event_listening(self, mock_subscribe):
+ args = arvados_fuse.command.ArgumentParser().parse_args([
+ '--disable-event-listening',
+ '--by-id',
+ '--foreground', self.mntdir])
+ self.mnt = arvados_fuse.command.Mount(args)
+ self.assertEqual(True, self.mnt.listen_for_events)
+ self.assertEqual(True, self.mnt.args.disable_event_listening)
+ with self.mnt:
+ pass
+ self.assertEqual(0, mock_subscribe.call_count)
+
+ @noexit
+ @mock.patch('arvados.events.subscribe')
+ def test_custom(self, mock_subscribe):
args = arvados_fuse.command.ArgumentParser().parse_args([
'--mount-tmp', 'foo',
'--mount-tmp', 'bar',
e = self.check_ent_type(arvados_fuse.ProjectDirectory, 'my_home')
self.assertEqual(e.project_object['uuid'],
run_test_server.fixture('users')['active']['uuid'])
+ self.assertEqual(True, self.mnt.listen_for_events)
+ with self.mnt:
+ pass
+ self.assertEqual(1, mock_subscribe.call_count)
+
+ @noexit
+ @mock.patch('arvados.events.subscribe')
+ def test_custom_no_listen(self, mock_subscribe):
+ args = arvados_fuse.command.ArgumentParser().parse_args([
+ '--mount-by-pdh', 'pdh',
+ '--mount-tmp', 'foo',
+ '--mount-tmp', 'bar',
+ '--foreground', self.mntdir])
+ self.mnt = arvados_fuse.command.Mount(args)
+ self.assertEqual(False, self.mnt.listen_for_events)
+ with self.mnt:
+ pass
+ self.assertEqual(0, mock_subscribe.call_count)
def test_custom_unsupported_layouts(self):
for name in ['.', '..', '', 'foo/bar', '/foo']:
- with self.assertRaises(SystemExit):
- args = arvados_fuse.command.ArgumentParser().parse_args([
- '--mount-tmp', name,
- '--foreground', self.mntdir])
- arvados_fuse.command.Mount(args)
+ with nostderr():
+ with self.assertRaises(SystemExit):
+ args = arvados_fuse.command.ArgumentParser().parse_args([
+ '--mount-tmp', name,
+ '--foreground', self.mntdir])
+ arvados_fuse.command.Mount(args)
+
+class MountErrorTest(unittest.TestCase):
+ def setUp(self):
+ self.mntdir = tempfile.mkdtemp()
+ run_test_server.run()
+ run_test_server.authorize_with("active")
+ self.logger = logging.getLogger("null")
+ self.logger.setLevel(logging.CRITICAL+1)
+
+ def tearDown(self):
+ if os.path.exists(self.mntdir):
+ # If the directory was not unmounted, this will raise an exception.
+ os.rmdir(self.mntdir)
+ run_test_server.reset()
+
+ def test_no_token(self):
+ del arvados.config._settings["ARVADOS_API_TOKEN"]
+ arvados.config._settings = {}
+ with self.assertRaises(SystemExit) as ex:
+ args = arvados_fuse.command.ArgumentParser().parse_args([self.mntdir])
+ arvados_fuse.command.Mount(args, logger=self.logger).run()
+ self.assertEqual(1, ex.exception.code)
+
+ def test_no_host(self):
+ del arvados.config._settings["ARVADOS_API_HOST"]
+ with self.assertRaises(SystemExit) as ex:
+ args = arvados_fuse.command.ArgumentParser().parse_args([self.mntdir])
+ arvados_fuse.command.Mount(args, logger=self.logger).run()
+ self.assertEqual(1, ex.exception.code)
+
+ def test_bogus_host(self):
+ arvados.config._settings["ARVADOS_API_HOST"] = "100::"
+ with self.assertRaises(SystemExit) as ex:
+ args = arvados_fuse.command.ArgumentParser().parse_args([self.mntdir])
+ arvados_fuse.command.Mount(args, logger=self.logger).run()
+ self.assertEqual(1, ex.exception.code)
+
+ def test_bogus_token(self):
+ arvados.config._settings["ARVADOS_API_TOKEN"] = "zzzzzzzzzzzzz"
+ with self.assertRaises(SystemExit) as ex:
+ args = arvados_fuse.command.ArgumentParser().parse_args([self.mntdir])
+ arvados_fuse.command.Mount(args, logger=self.logger).run()
+ self.assertEqual(1, ex.exception.code)
+
+ def test_bogus_mount_dir(self):
+ # All FUSE errors in llfuse.init() are raised as RuntimeError
+ # An easy error to trigger is to supply a nonexistent mount point,
+ # so test that one.
+ #
+ # Other possible errors that also raise RuntimeError (but are much
+ # harder to test automatically because they depend on operating
+ # system configuration):
+ #
+ # The user doesn't have permission to use FUSE
+ # The user specified --allow-other but user_allow_other is not set
+ # in /etc/fuse.conf
+ os.rmdir(self.mntdir)
+ with self.assertRaises(SystemExit) as ex:
+ args = arvados_fuse.command.ArgumentParser().parse_args([self.mntdir])
+ arvados_fuse.command.Mount(args, logger=self.logger).run()
+ self.assertEqual(1, ex.exception.code)
+
+ def test_unreadable_collection(self):
+ with self.assertRaises(SystemExit) as ex:
+ args = arvados_fuse.command.ArgumentParser().parse_args([
+ "--collection", "zzzzz-4zz18-zzzzzzzzzzzzzzz", self.mntdir])
+ arvados_fuse.command.Mount(args, logger=self.logger).run()
+ self.assertEqual(1, ex.exception.code)
+
+ def test_unreadable_project(self):
+ with self.assertRaises(SystemExit) as ex:
+ args = arvados_fuse.command.ArgumentParser().parse_args([
+ "--project", "zzzzz-j7d0g-zzzzzzzzzzzzzzz", self.mntdir])
+ arvados_fuse.command.Mount(args, logger=self.logger).run()
+ self.assertEqual(1, ex.exception.code)
projects / arvados.git / blobdiff