7 from pathlib import Path
10 BCH_PATH = DIR / 'bcachefs'
12 VPAT = re.compile(r'ERROR SUMMARY: (\d+) errors from (\d+) contexts')
14 class ValgrindFailedError(Exception):
15 def __init__(self, log):
18 def check_valgrind(logfile):
19 log = logfile.read().decode('utf-8')
21 assert m is not None, 'Internal error: valgrind log did not match.'
23 errors = int(m.group(1))
25 raise ValgrindFailedError(log)
27 def run(cmd, *args, valgrind=False, check=False):
28 """Run an external program via subprocess, optionally with valgrind.
30 This subprocess wrapper will capture the stdout and stderr. If valgrind is
31 requested, it will be checked for errors and raise a
32 ValgrindFailedError if there's a problem.
34 cmds = [cmd] + list(args)
37 vout = tempfile.NamedTemporaryFile()
40 '--log-file={}'.format(vout.name)]
43 print("Running '{}'".format(cmds))
44 res = subprocess.run(cmds, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
45 encoding='utf-8', check=check)
52 def run_bch(*args, **kwargs):
53 """Wrapper to run the bcachefs binary specifically."""
54 cmds = [BCH_PATH] + list(args)
55 return run(*cmds, **kwargs)
57 def sparse_file(lpath, size):
58 """Construct a sparse file of the specified size.
60 This is typically used to create device files for bcachefs.
63 f = path.touch(mode = 0o600, exist_ok = False)
64 os.truncate(path, size)
68 def device_1g(tmpdir):
69 """Default 1g sparse file for use with bcachefs."""
70 path = tmpdir / 'dev-1g'
71 return sparse_file(path, 1024**3)