+# Copyright (C) The Arvados Authors. All rights reserved.
+#
+# SPDX-License-Identifier: AGPL-3.0
+
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
import sys
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)
run_test_server.fixture('users')['active']['uuid'])
self.assertEqual(True, self.mnt.listen_for_events)
+ def test_version_argument(self):
+ orig, sys.stderr = sys.stderr, io.BytesIO()
+ with self.assertRaises(SystemExit):
+ args = arvados_fuse.command.ArgumentParser().parse_args(['--version'])
+ self.assertRegexpMatches(sys.stderr.getvalue(), "[0-9]+\.[0-9]+\.[0-9]+")
+ sys.stderr = orig
+
+ @noexit
+ @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
- def test_custom(self):
+ @mock.patch('arvados.events.subscribe')
+ def test_custom(self, mock_subscribe):
args = arvados_fuse.command.ArgumentParser().parse_args([
'--mount-tmp', 'foo',
'--mount-tmp', 'bar',
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
- def test_custom_no_listen(self):
+ @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']: