summaryrefslogtreecommitdiff
path: root/test/util.py
diff options
context:
space:
mode:
authorPaul Lawrence <paullawrence@google.com>2019-08-02 16:45:27 -0700
committerandroid-build-merger <android-build-merger@google.com>2019-08-02 16:45:27 -0700
commit1ca65fd045ff0aad723cfb4c740ee9250ae5c360 (patch)
treeeb415748db51aed7c3d08a5e7dae57c8576ea940 /test/util.py
parentbe83b553fef2fa13ecc8edbea974f450637cc60e (diff)
parentc77441c6bcfb18dd3c887cdf24a778960fbe0c41 (diff)
Remove GPL files am: 94aa887ced am: b4099f5e0f am: 2609a900fb
am: c77441c6bc Change-Id: I849d56bb6f9909131b449f4f472f21aabd6d2aeb
Diffstat (limited to 'test/util.py')
-rw-r--r--test/util.py171
1 files changed, 0 insertions, 171 deletions
diff --git a/test/util.py b/test/util.py
deleted file mode 100644
index 8bef265..0000000
--- a/test/util.py
+++ /dev/null
@@ -1,171 +0,0 @@
-#!/usr/bin/env python3
-import subprocess
-import pytest
-import os
-import stat
-import time
-from os.path import join as pjoin
-import sys
-import re
-import itertools
-
-basename = pjoin(os.path.dirname(__file__), '..')
-
-def test_printcap():
- cmdline = base_cmdline + [ pjoin(basename, 'example', 'printcap') ]
- proc = subprocess.Popen(cmdline, stdout=subprocess.PIPE,
- universal_newlines=True)
- (stdout, _) = proc.communicate(30)
- assert proc.returncode == 0
-
- proto = None
- caps = set()
- for line in stdout.split('\n'):
- if line.startswith('\t'):
- caps.add(line.strip())
- continue
-
- hit = re.match(r'Protocol version: (\d+)\.(\d+)$', line)
- if hit:
- proto = (int(hit.group(1)), int(hit.group(2)))
-
- return (proto, caps)
-
-
-def wait_for_mount(mount_process, mnt_dir,
- test_fn=os.path.ismount):
- elapsed = 0
- while elapsed < 30:
- if test_fn(mnt_dir):
- return True
- if mount_process.poll() is not None:
- pytest.fail('file system process terminated prematurely')
- time.sleep(0.1)
- elapsed += 0.1
- pytest.fail("mountpoint failed to come up")
-
-def cleanup(mount_process, mnt_dir):
- # Don't bother trying Valgrind if things already went wrong
-
- if 'bsd' in sys.platform or 'dragonfly' in sys.platform:
- cmd = [ 'umount', '-f', mnt_dir ]
- else:
- cmd = [pjoin(basename, 'util', 'fusermount3'),
- '-z', '-u', mnt_dir]
- subprocess.call(cmd, stdout=subprocess.DEVNULL,
- stderr=subprocess.STDOUT)
- mount_process.terminate()
- try:
- mount_process.wait(1)
- except subprocess.TimeoutExpired:
- mount_process.kill()
-
-def umount(mount_process, mnt_dir):
-
- if 'bsd' in sys.platform or 'dragonfly' in sys.platform:
- cmdline = [ 'umount', mnt_dir ]
- else:
- # fusermount3 will be setuid root, so we can only trace it with
- # valgrind if we're root
- if os.getuid() == 0:
- cmdline = base_cmdline
- else:
- cmdline = []
- cmdline = cmdline + [ pjoin(basename, 'util', 'fusermount3'),
- '-z', '-u', mnt_dir ]
-
- subprocess.check_call(cmdline)
- assert not os.path.ismount(mnt_dir)
-
- # Give mount process a little while to terminate. Popen.wait(timeout)
- # was only added in 3.3...
- elapsed = 0
- while elapsed < 30:
- code = mount_process.poll()
- if code is not None:
- if code == 0:
- return
- pytest.fail('file system process terminated with code %s' % (code,))
- time.sleep(0.1)
- elapsed += 0.1
- pytest.fail('mount process did not terminate')
-
-
-def safe_sleep(secs):
- '''Like time.sleep(), but sleep for at least *secs*
-
- `time.sleep` may sleep less than the given period if a signal is
- received. This function ensures that we sleep for at least the
- desired time.
- '''
-
- now = time.time()
- end = now + secs
- while now < end:
- time.sleep(end - now)
- now = time.time()
-
-def fuse_test_marker():
- '''Return a pytest.marker that indicates FUSE availability
-
- If system/user/environment does not support FUSE, return
- a `pytest.mark.skip` object with more details. If FUSE is
- supported, return `pytest.mark.uses_fuse()`.
- '''
-
- skip = lambda x: pytest.mark.skip(reason=x)
-
- if 'bsd' in sys.platform or 'dragonfly' in sys.platform:
- return pytest.mark.uses_fuse()
-
- with subprocess.Popen(['which', 'fusermount3'], stdout=subprocess.PIPE,
- universal_newlines=True) as which:
- fusermount_path = which.communicate()[0].strip()
-
- if not fusermount_path or which.returncode != 0:
- return skip("Can't find fusermount executable")
-
- if not os.path.exists('/dev/fuse'):
- return skip("FUSE kernel module does not seem to be loaded")
-
- if os.getuid() == 0:
- return pytest.mark.uses_fuse()
-
- mode = os.stat(fusermount_path).st_mode
- if mode & stat.S_ISUID == 0:
- return skip('fusermount executable not setuid, and we are not root.')
-
- try:
- fd = os.open('/dev/fuse', os.O_RDWR)
- except OSError as exc:
- return skip('Unable to open /dev/fuse: %s' % exc.strerror)
- else:
- os.close(fd)
-
- return pytest.mark.uses_fuse()
-
-def powerset(iterable):
- s = list(iterable)
- return itertools.chain.from_iterable(
- itertools.combinations(s, r) for r in range(len(s)+1))
-
-
-# Use valgrind if requested
-if os.environ.get('TEST_WITH_VALGRIND', 'no').lower().strip() \
- not in ('no', 'false', '0'):
- base_cmdline = [ 'valgrind', '-q', '--' ]
-else:
- base_cmdline = []
-
-# Try to use local fusermount3
-os.environ['PATH'] = '%s:%s' % (pjoin(basename, 'util'), os.environ['PATH'])
-# Put example binaries on PATH
-os.environ['PATH'] = '%s:%s' % (pjoin(basename, 'example'), os.environ['PATH'])
-
-try:
- (fuse_proto, fuse_caps) = test_printcap()
-except:
- # Rely on test to raise error
- fuse_proto = (0,0)
- fuse_caps = set()
-