buptest.py

   1
   2 from __future__ import absolute_import, print_function
   3 from collections import namedtuple
   4 from contextlib import contextmanager
   5 from os.path import abspath, basename, dirname, realpath
   6 from pipes import quote
   7 from subprocess import PIPE, Popen
   8 from traceback import extract_stack
   9 import errno, os, 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 try:
  38     os.makedirs(_bup_tmp)
  39 except OSError as e:
  40     if e.errno != errno.EEXIST:
  41         raise
  42
  43
  44 @contextmanager
  45 def test_tempdir(prefix):
  46     initial_failures = wvfailure_count()
  47     tmpdir = tempfile.mkdtemp(dir=_bup_tmp, prefix=prefix)
  48     yield tmpdir
  49     if wvfailure_count() == initial_failures:
  50         subprocess.call(['chmod', '-R', 'u+rwX', tmpdir])
  51         subprocess.call(['rm', '-rf', tmpdir])
  52
  53
  54 ex_res = namedtuple('SubprocResult', ['out', 'err', 'proc', 'rc'])
  55
  56 def run(cmd, check=True, input=None, **kwargs):
  57     """Run a subprocess as per subprocess.Popen(cmd, **kwargs) followed by
  58     communicate(input=input).  If check is true, then throw an
  59     exception if the subprocess exits with non-zero status.  Return a
  60     SubprocResult tuple.
  61
  62     """
  63     if input:
  64         assert 'stdin' not in kwargs
  65         kwargs['stdin'] = PIPE
  66     p = Popen(cmd, **kwargs)
  67     out, err = p.communicate(input=input)
  68     if check and p.returncode != 0:
  69         raise Exception('subprocess %r failed with status %d%s'
  70                         % (' '.join(map(quote, cmd)), p.returncode,
  71                            (', stderr: %r' % err) if err else ''))
  72     return ex_res(out=out, err=err, proc=p, rc=p.returncode)
  73
  74 def logcmd(cmd):
  75     if isinstance(cmd, basestring):
  76         print(cmd, file=sys.stderr)
  77     else:
  78         print(' '.join(map(quote, cmd)), file=sys.stderr)
  79
  80 def ex(cmd, **kwargs):
  81     """Print cmd to stderr and then run it as per ex(...).
  82     Print the subprocess stderr to stderr if stderr=PIPE and there's
  83     any data.
  84     """
  85     logcmd(cmd)
  86     result = run(cmd, **kwargs)
  87     if result.err:
  88         sys.stderr.write(result.err)
  89     return result
  90
  91 def exo(cmd, **kwargs):
  92     """Print cmd to stderr and then run it as per ex(..., stdout=PIPE).
  93     Print the subprocess stderr to stderr if stderr=PIPE and there's
  94     any data.
  95
  96     """
  97     assert 'stdout' not in kwargs
  98     kwargs['stdout'] = PIPE
  99     return ex(cmd, **kwargs)