services/fuse/tests/test_exec.py

   1 # Copyright (C) The Arvados Authors. All rights reserved.
   2 #
   3 # SPDX-License-Identifier: AGPL-3.0
   4
   5 import arvados_fuse.command
   6 import json
   7 import multiprocessing
   8 import os
   9 import run_test_server
  10 import tempfile
  11 import unittest
  12
  13 try:
  14     from shlex import quote
  15 except:
  16     from pipes import quote
  17
  18 def try_exec(mnt, cmd):
  19     try:
  20         arvados_fuse.command.Mount(
  21             arvados_fuse.command.ArgumentParser().parse_args([
  22                 '--read-write',
  23                 '--mount-tmp=zzz',
  24                 '--unmount-timeout=0.1',
  25                 mnt,
  26                 '--exec'] + cmd)).run()
  27     except SystemExit:
  28         pass
  29     else:
  30         raise AssertionError('should have exited')
  31
  32
  33 class ExecMode(unittest.TestCase):
  34     @classmethod
  35     def setUpClass(cls):
  36         run_test_server.run()
  37         run_test_server.run_keep(enforce_permissions=True, num_servers=2)
  38         run_test_server.authorize_with('active')
  39
  40     @classmethod
  41     def tearDownClass(cls):
  42         run_test_server.stop_keep(num_servers=2)
  43
  44     def setUp(self):
  45         self.mnt = tempfile.mkdtemp()
  46         _, self.okfile = tempfile.mkstemp()
  47         self.pool = multiprocessing.Pool(1)
  48
  49     def tearDown(self):
  50         self.pool.terminate()
  51         self.pool.join()
  52         os.rmdir(self.mnt)
  53         os.unlink(self.okfile)
  54
  55     def test_exec(self):
  56         self.pool.apply(try_exec, (self.mnt, [
  57             'sh', '-c',
  58             'echo -n foo >{}; cp {} {}'.format(
  59                 quote(os.path.join(self.mnt, 'zzz', 'foo.txt')),
  60                 quote(os.path.join(self.mnt, 'zzz', '.arvados#collection')),
  61                 quote(os.path.join(self.okfile)))]))
  62         self.assertRegexpMatches(
  63             json.load(open(self.okfile))['manifest_text'],
  64             r' 0:3:foo.txt\n')