"Fossies" - the Fresh Open Source Software Archive

Member "fuse-3.2.3/test/util.py" (11 May 2018, 4673 Bytes) of package /linux/misc/fuse-3.2.3.tar.xz:


As a special service "Fossies" has tried to format the requested source page into HTML format using (guessed) Python source code syntax highlighting (style: standard) with prefixed line numbers. Alternatively you can here view or download the uninterpreted source code file. See also the last Fossies "Diffs" side-by-side code changes report for "util.py": 3.2.1_vs_3.2.2.

    1 #!/usr/bin/env python3
    2 import subprocess
    3 import pytest
    4 import os
    5 import stat
    6 import time
    7 from os.path import join as pjoin
    8 import sys
    9 import re
   10 
   11 basename = pjoin(os.path.dirname(__file__), '..')
   12 
   13 def test_printcap():
   14     cmdline = base_cmdline + [ pjoin(basename, 'example', 'printcap') ]
   15     proc = subprocess.Popen(cmdline, stdout=subprocess.PIPE,
   16                             universal_newlines=True)
   17     (stdout, _) = proc.communicate(30)
   18     assert proc.returncode == 0
   19 
   20     proto = None
   21     caps = set()
   22     for line in stdout.split('\n'):
   23         if line.startswith('\t'):
   24             caps.add(line.strip())
   25             continue
   26 
   27         hit = re.match(r'Protocol version: (\d+)\.(\d+)$', line) 
   28         if hit:
   29             proto = (int(hit.group(1)), int(hit.group(2)))
   30 
   31     return (proto, caps)
   32 
   33 
   34 def wait_for_mount(mount_process, mnt_dir,
   35                    test_fn=os.path.ismount):
   36     elapsed = 0
   37     while elapsed < 30:
   38         if test_fn(mnt_dir):
   39             return True
   40         if mount_process.poll() is not None:
   41             pytest.fail('file system process terminated prematurely')
   42         time.sleep(0.1)
   43         elapsed += 0.1
   44     pytest.fail("mountpoint failed to come up")
   45 
   46 def cleanup(mnt_dir):
   47     # Don't bother trying Valgrind if things already went wrong
   48 
   49     if 'bsd' in sys.platform or 'dragonfly' in sys.platform:
   50         cmd = [ 'umount', '-f', mnt_dir ]
   51     else:
   52         cmd = [pjoin(basename, 'util', 'fusermount3'),
   53                          '-z', '-u', mnt_dir]
   54     subprocess.call(cmd, stdout=subprocess.DEVNULL,
   55                     stderr=subprocess.STDOUT)
   56 
   57 def umount(mount_process, mnt_dir):
   58 
   59     if 'bsd' in sys.platform or 'dragonfly' in sys.platform:
   60         cmdline = [ 'umount', mnt_dir ]
   61     else:
   62         # fusermount3 will be setuid root, so we can only trace it with
   63         # valgrind if we're root
   64         if os.getuid() == 0:
   65             cmdline = base_cmdline
   66         else:
   67             cmdline = []
   68         cmdline = cmdline + [ pjoin(basename, 'util', 'fusermount3'),
   69                               '-z', '-u', mnt_dir ]
   70 
   71     subprocess.check_call(cmdline)
   72     assert not os.path.ismount(mnt_dir)
   73 
   74     # Give mount process a little while to terminate. Popen.wait(timeout)
   75     # was only added in 3.3...
   76     elapsed = 0
   77     while elapsed < 30:
   78         code = mount_process.poll()
   79         if code is not None:
   80             if code == 0:
   81                 return
   82             pytest.fail('file system process terminated with code %s' % (code,))
   83         time.sleep(0.1)
   84         elapsed += 0.1
   85     pytest.fail('mount process did not terminate')
   86 
   87 
   88 def safe_sleep(secs):
   89     '''Like time.sleep(), but sleep for at least *secs*
   90 
   91     `time.sleep` may sleep less than the given period if a signal is
   92     received. This function ensures that we sleep for at least the
   93     desired time.
   94     '''
   95 
   96     now = time.time()
   97     end = now + secs
   98     while now < end:
   99         time.sleep(end - now)
  100         now = time.time()
  101 
  102 def fuse_test_marker():
  103     '''Return a pytest.marker that indicates FUSE availability
  104 
  105     If system/user/environment does not support FUSE, return
  106     a `pytest.mark.skip` object with more details. If FUSE is
  107     supported, return `pytest.mark.uses_fuse()`.
  108     '''
  109 
  110     skip = lambda x: pytest.mark.skip(reason=x)
  111 
  112     if 'bsd' in sys.platform or 'dragonfly' in sys.platform:
  113         return pytest.mark.uses_fuse()
  114 
  115     with subprocess.Popen(['which', 'fusermount3'], stdout=subprocess.PIPE,
  116                           universal_newlines=True) as which:
  117         fusermount_path = which.communicate()[0].strip()
  118 
  119     if not fusermount_path or which.returncode != 0:
  120         return skip("Can't find fusermount executable")
  121 
  122     if not os.path.exists('/dev/fuse'):
  123         return skip("FUSE kernel module does not seem to be loaded")
  124 
  125     if os.getuid() == 0:
  126         return pytest.mark.uses_fuse()
  127 
  128     mode = os.stat(fusermount_path).st_mode
  129     if mode & stat.S_ISUID == 0:
  130         return skip('fusermount executable not setuid, and we are not root.')
  131 
  132     try:
  133         fd = os.open('/dev/fuse', os.O_RDWR)
  134     except OSError as exc:
  135         return skip('Unable to open /dev/fuse: %s' % exc.strerror)
  136     else:
  137         os.close(fd)
  138 
  139     return pytest.mark.uses_fuse()
  140 
  141 # Use valgrind if requested
  142 if os.environ.get('TEST_WITH_VALGRIND', 'no').lower().strip() \
  143    not in ('no', 'false', '0'):
  144     base_cmdline = [ 'valgrind', '-q', '--' ]
  145 else:
  146     base_cmdline = []
  147 
  148 # Try to use local fusermount3
  149 os.environ['PATH'] = '%s:%s' % (pjoin(basename, 'util'), os.environ['PATH'])
  150 
  151 try:
  152     (fuse_proto, fuse_caps) = test_printcap()
  153 except:
  154     # Rely on test to raise error
  155     fuse_proto = (0,0)
  156     fuse_caps = set()
  157