git.sesse.net Git - bcachefs-tools-debian/blob - tests/util.py

   1 #!/usr/bin/python3
   2
   3 import os
   4 import re
   5 import subprocess
   6 import tempfile
   7 from pathlib import Path
   8
   9 DIR = Path('..')
  10 BCH_PATH = DIR / 'bcachefs'
  11
  12 VPAT = re.compile(r'ERROR SUMMARY: (\d+) errors from (\d+) contexts')
  13
  14 class ValgrindFailedError(Exception):
  15     def __init__(self, log):
  16         self.log = log
  17
  18 def check_valgrind(logfile):
  19     log = logfile.read().decode('utf-8')
  20     m = VPAT.search(log)
  21     assert m is not None, 'Internal error: valgrind log did not match.'
  22
  23     errors = int(m.group(1))
  24     if errors > 0:
  25         raise ValgrindFailedError(log)
  26
  27 def run(cmd, *args, valgrind=False, check=False):
  28     """Run an external program via subprocess, optionally with valgrind.
  29
  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.
  33     """
  34     cmds = [cmd] + list(args)
  35
  36     if valgrind:
  37         vout = tempfile.NamedTemporaryFile()
  38         vcmd = ['valgrind',
  39                '--leak-check=full',
  40                '--log-file={}'.format(vout.name)]
  41         cmds = vcmd + cmds
  42
  43     print("Running '{}'".format(cmds))
  44     res = subprocess.run(cmds, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
  45                          encoding='utf-8', check=check)
  46
  47     if valgrind:
  48         check_valgrind(vout)
  49
  50     return res
  51
  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)
  56
  57 def sparse_file(lpath, size):
  58     """Construct a sparse file of the specified size.
  59
  60     This is typically used to create device files for bcachefs.
  61     """
  62     path = Path(lpath)
  63     f = path.touch(mode = 0o600, exist_ok = False)
  64     os.truncate(path, size)
  65
  66     return path
  67
  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)