buptest.py

   1
   2 from __future__ import print_function
   3 from collections import namedtuple
   4 from contextlib import contextmanager
   5 from os.path import basename, dirname, realpath
   6 from pipes import quote
   7 from subprocess import PIPE, Popen, check_call
   8 from traceback import extract_stack
   9 import subprocess, sys, tempfile
  10
  11 from wvtest import WVPASSEQ, wvfailure_count
  12
  13 from bup import helpers
  14
  15
  16 @contextmanager
  17 def no_lingering_errors():
  18     def fail_if_errors():
  19         if helpers.saved_errors:
  20             bt = extract_stack()
  21             src_file, src_line, src_func, src_txt = bt[-4]
  22             msg = 'saved_errors ' + repr(helpers.saved_errors)
  23             print('! %-70s %s' % ('%s:%-4d %s' % (basename(src_file),
  24                                                   src_line,
  25                                                   msg),
  26                                   'FAILED'))
  27             sys.stdout.flush()
  28     fail_if_errors()
  29     helpers.clear_errors()
  30     yield
  31     fail_if_errors()
  32     helpers.clear_errors()
  33
  34
  35 # Assumes (of course) this file is at the top-level of the source tree
  36 _bup_tmp = realpath(dirname(__file__) + '/t/tmp')
  37 helpers.mkdirp(_bup_tmp)
  38
  39
  40 @contextmanager
  41 def test_tempdir(prefix):
  42     initial_failures = wvfailure_count()
  43     tmpdir = tempfile.mkdtemp(dir=_bup_tmp, prefix=prefix)
  44     yield tmpdir
  45     if wvfailure_count() == initial_failures:
  46         subprocess.call(['chmod', '-R', 'u+rwX', tmpdir])
  47         subprocess.call(['rm', '-rf', tmpdir])
  48
  49
  50 def logcmd(cmd):
  51     if isinstance(cmd, basestring):
  52         print(cmd, file=sys.stderr)
  53     else:
  54         print(' '.join(map(quote, cmd)), file=sys.stderr)
  55
  56
  57 SubprocInfo = namedtuple('SubprocInfo', ('out', 'err', 'rc', 'p'))
  58
  59 def exo(cmd, input=None, stdin=None, stdout=PIPE, stderr=PIPE,
  60         shell=False, check=True):
  61     """Print cmd to stderr, run it, and return the resulting SubprocInfo.
  62     The keyword arguments are passed to Popen, and the defaults
  63     capture both stdout and stderr.
  64
  65     """
  66     logcmd(cmd)
  67     p = Popen(cmd,
  68               stdin=(PIPE if input else stdin),
  69               stdout=stdout,
  70               stderr=stderr,
  71               shell=shell)
  72     out, err = p.communicate(input=input)
  73     if check and p.returncode != 0:
  74         raise Exception('subprocess %r failed with status %d%s'
  75                         % (' '.join(map(quote, cmd)),
  76                            p.returncode,
  77                            (', stderr: %r' % err) if stderr else ''))
  78     return SubprocInfo(out=out, err=err, rc=p.returncode, p=p)
  79
  80 def exc(cmd, input=None, stdout=None, stderr=None, shell=False, check=True):
  81     """Print cmd to stderr, run it, and return the resulting SubprocInfo.
  82     The keyword arguments are passed to Popen, and the defaults
  83     allow stdout and stderr to pass through.
  84
  85     """
  86     return exo(cmd, input=input, stdout=stdout, stderr=stderr, shell=shell,
  87                check=check)