text: 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
  import tempfile
  import unittest
  
  def noexit(func):
      """If argparse or arvados_fuse tries to exit, fail the test instead"""
      class SystemExitCaught(StandardError):
          pass
      @functools.wraps(func)
      def wrapper(*args, **kwargs):
          try:
              return func(*args, **kwargs)
          except SystemExit:
              raise SystemExitCaught
      return wrapper
  
  @contextlib.contextmanager
  def nostderr():
      orig, sys.stderr = sys.stderr, open(os.devnull, 'w')
      try:
          yield
      finally:
          sys.stderr = orig
  
  
  class MountArgsTest(unittest.TestCase):
      def setUp(self):
          self.mntdir = tempfile.mkdtemp()
          run_test_server.authorize_with('active')
  
      def tearDown(self):
          os.rmdir(self.mntdir)
  
      def lookup(self, mnt, *path):
          ent = mnt.operations.inodes[llfuse.ROOT_INODE]
          for p in path:
              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)
          return ent
  
      @noexit
      def test_default_all(self):
          args = arvados_fuse.command.ArgumentParser().parse_args([
              '--foreground', self.mntdir])
          self.assertEqual(args.mode, None)
          self.mnt = arvados_fuse.command.Mount(args)
          e = self.check_ent_type(arvados_fuse.ProjectDirectory, 'home')
          self.assertEqual(e.project_object['uuid'],
                           run_test_server.fixture('users')['active']['uuid'])
          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$')
  
          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$')
  
      @noexit
      def test_by_id(self):
          args = arvados_fuse.command.ArgumentParser().parse_args([
              '--by-id',
              '--foreground', self.mntdir])
          self.assertEqual(args.mode, 'by_id')
          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):
          args = arvados_fuse.command.ArgumentParser().parse_args([
              '--by-pdh',
              '--foreground', self.mntdir])
          self.assertEqual(args.mode, 'by_pdh')
          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):
          args = arvados_fuse.command.ArgumentParser().parse_args([
              '--by-tag',
              '--foreground', self.mntdir])
          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'):
          c = run_test_server.fixture('collections')['public_text_file']
          cid = c[id_type]
          args = arvados_fuse.command.ArgumentParser().parse_args([
              '--collection', cid,
              '--foreground', self.mntdir])
          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')
  
      @noexit
      def test_home(self):
          args = arvados_fuse.command.ArgumentParser().parse_args([
              '--home',
              '--foreground', self.mntdir])
          self.assertEqual(args.mode, 'home')
          self.mnt = arvados_fuse.command.Mount(args)
          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']
          gid = run_test_server.fixture('groups')['aproject']['uuid']
          for badargs in [
                  ['--mount-tmp', 'foo', '--collection', cid],
                  ['--mount-tmp', 'foo', '--project', gid],
                  ['--collection', cid, '--project', gid],
                  ['--by-id', '--project', gid],
                  ['--mount-tmp', 'foo', '--by-id'],
          ]:
              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']
          args = arvados_fuse.command.ArgumentParser().parse_args([
              '--project', uuid,
              '--foreground', self.mntdir])
          self.mnt = arvados_fuse.command.Mount(args)
          e = self.check_ent_type(arvados_fuse.ProjectDirectory)
          self.assertEqual(e.project_object['uuid'], uuid)
  
      @noexit
      def test_shared(self):
          args = arvados_fuse.command.ArgumentParser().parse_args([
              '--shared',
              '--foreground', self.mntdir])
          self.assertEqual(args.mode, 'shared')
          self.mnt = arvados_fuse.command.Mount(args)
          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):
          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
      @mock.patch('arvados.events.subscribe')
      def test_custom(self, mock_subscribe):
          args = arvados_fuse.command.ArgumentParser().parse_args([
              '--mount-tmp', 'foo',
              '--mount-tmp', 'bar',
              '--mount-home', 'my_home',
              '--foreground', self.mntdir])
          self.assertEqual(args.mode, None)
          self.mnt = arvados_fuse.command.Mount(args)
          self.check_ent_type(arvados_fuse.Directory)
          self.check_ent_type(arvados_fuse.TmpCollectionDirectory, 'foo')
          self.check_ent_type(arvados_fuse.TmpCollectionDirectory, '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 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)
  