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