path: root/tests/util.py

#!/usr/bin/python3

import errno
import os
import re
import subprocess
import tempfile
import threading
import time

from pathlib import Path

BASE_PATH= os.path.dirname(__file__)
BCH_PATH = os.path.abspath(os.path.join(BASE_PATH, '../target/release', 'bcachefs'))
VALGRIND_PATH= os.path.abspath(os.path.join(BASE_PATH,
    'valgrind-suppressions.txt'))

VPAT = re.compile(r'ERROR SUMMARY: (\d+) errors from (\d+) contexts')

ENABLE_VALGRIND = os.getenv('BCACHEFS_TEST_USE_VALGRIND', 'no') == 'yes'

class ValgrindFailedError(Exception):
    def __init__(self, log):
        self.log = log

def check_valgrind(log):
    m = VPAT.search(log)
    if m is None:
        print('Internal error: valgrind log did not match.')
        print('-- valgrind log:')
        print(log)
        print('-- end log --')
        assert False

    errors = int(m.group(1))
    if errors > 0:
        raise ValgrindFailedError(log)

def run(cmd, *args, valgrind=False, check=False):
    """Run an external program via subprocess, optionally with valgrind.

    This subprocess wrapper will capture the stdout and stderr. If valgrind is
    requested, it will be checked for errors and raise a
    ValgrindFailedError if there's a problem.
    """
    cmds = [cmd] + list(args)
    valgrind = valgrind and ENABLE_VALGRIND

    print("Running '{}'".format(cmds))
    if valgrind:
        vout = tempfile.NamedTemporaryFile()
        vcmd = ['valgrind',
               '--leak-check=full',
               '--gen-suppressions=all',
               '--suppressions={}'.format(VALGRIND_PATH),
               '--log-file={}'.format(vout.name)]
        cmds = vcmd + cmds

        res = subprocess.run(cmds, stdout=subprocess.PIPE,
                stderr=subprocess.PIPE, encoding='utf-8', check=check)
        check_valgrind(vout.read().decode('utf-8'))
    else:
        res = subprocess.run(cmds, stdout=subprocess.PIPE,
                stderr=subprocess.PIPE, encoding='utf-8', check=check)

    return res

def run_bch(*args, **kwargs):
    """Wrapper to run the bcachefs binary specifically."""
    cmds = [BCH_PATH] + list(args)
    return run(*cmds, **kwargs)

def sparse_file(lpath, size):
    """Construct a sparse file of the specified size.

    This is typically used to create device files for bcachefs.
    """
    path = Path(lpath)
    path.touch(mode = 0o600, exist_ok = False)
    os.truncate(path, size)

    return path

def device_1g(tmpdir):
    """Default 1g sparse file for use with bcachefs."""
    path = tmpdir / 'dev-1g'
    return sparse_file(path, 1024**3)

def format_1g(tmpdir):
    """Format a default filesystem on a 1g device."""
    dev = device_1g(tmpdir)
    run_bch('format', dev, check=True)
    return dev

def mountpoint(tmpdir):
    """Construct a mountpoint "mnt" for tests."""
    path = Path(tmpdir) / 'mnt'
    path.mkdir(mode = 0o700)
    return path

class Timestamp:
    '''Context manager to assist in verifying timestamps.

    Records the range of times which would be valid for an encoded operation to
    use.

    FIXME: The kernel code is currently using CLOCK_REALTIME_COARSE, but python
    didn't expose this time API (yet).  Probably the kernel shouldn't be using
    _COARSE anyway, but this might lead to occasional errors.

    To make sure this doesn't happen, we sleep a fraction of a second in an
    attempt to guarantee containment.

    N.B. this might be better tested by overriding the clock used in bcachefs.

    '''
    def __init__(self):
        self.start = None
        self.end = None

    def __enter__(self):
        self.start = time.clock_gettime(time.CLOCK_REALTIME)
        time.sleep(0.1)
        return self

    def __exit__(self, type, value, traceback):
        time.sleep(0.1)
        self.end = time.clock_gettime(time.CLOCK_REALTIME)

    def contains(self, test):
        '''True iff the test time is within the range.'''
        return self.start <= test <= self.end

class FuseError(Exception):
    def __init__(self, msg):
        self.msg = msg

class BFuse:
    '''bcachefs fuse runner.

    This class runs bcachefs in fusemount mode, and waits until the mount has
    reached a point suitable for testing the filesystem.

    bcachefs is run under valgrind by default, and is checked for errors.
    '''

    def __init__(self, dev, mnt):
        self.thread = None
        self.dev = dev
        self.mnt = mnt
        self.ready = threading.Event()
        self.proc = None
        self.returncode = None
        self.stdout = None
        self.stderr = None
        self.vout = None

    def run(self):
        """Background thread which runs "bcachefs fusemount" under valgrind"""

        vlog = None
        cmd = []

        if ENABLE_VALGRIND:
            vlog = tempfile.NamedTemporaryFile()
            cmd += [ 'valgrind',
                     '--leak-check=full',
                     '--gen-suppressions=all',
                     '--suppressions=valgrind-suppressions.txt',
                     '--log-file={}'.format(vlog.name) ]

        cmd += [ BCH_PATH,
                 'fusemount', '-f', self.dev, self.mnt]

        print("Running {}".format(cmd))

        err = tempfile.TemporaryFile()
        self.proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=err,
                                     encoding='utf-8')

        out1 = self.expect(self.proc.stdout, r'^Fuse mount initialized.$')
        self.ready.set()

        print("Waiting for process.")
        (out2, _) = self.proc.communicate()
        print("Process exited.")

        self.returncode = self.proc.returncode
        if self.returncode == 0:
            errors = [ 'btree iterators leaked!',
                       'emergency read only!' ]
            for e in errors:
                if e in out2:
                    print('Debug error found in output: "{}"'.format(e))
                    self.returncode = errno.ENOMSG

        self.stdout = out1 + out2
        self.stderr = err.read()
        if vlog:
            self.vout = vlog.read().decode('utf-8')

    def expect(self, pipe, regex):
        """Wait for the child process to mount."""

        c = re.compile(regex)

        out = ""
        for line in pipe:
            print('Expect line "{}"'.format(line.rstrip()))
            out += line
            if c.match(line):
                print("Matched.")
                return out

        raise FuseError('stdout did not contain regex "{}"'.format(regex))

    def mount(self):
        print("Starting fuse thread.")

        assert not self.thread
        self.thread = threading.Thread(target=self.run)
        self.thread.start()

        self.ready.wait()
        print("Fuse is mounted.")

    def unmount(self, timeout=None):
        print("Unmounting fuse.")
        run("fusermount3", "-zu", self.mnt)

        if self.thread:
            print("Waiting for thread to exit.")
            self.thread.join(timeout)
            if self.thread.is_alive():
                if self.proc:
                    self.proc.kill()
                self.thread.join()
        else:
            print("Thread was already done.")

        self.thread = None
        self.ready.clear()

        if self.vout:
            check_valgrind(self.vout)

    def verify(self):
        # avoid throwing exception in assertion
        assert self.stdout is not None
        assert self.stderr is not None
        assert self.returncode == 0
        assert len(self.stdout) > 0
        assert len(self.stderr) == 0

def have_fuse():
    res = run(BCH_PATH, 'fusemount', valgrind=False)
    return "Please supply a mountpoint." in res.stdout