services/fuse/tests/test_exec.py

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