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