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 basename, dirname, realpath
   6 from pipes import quote
   7 from subprocess import PIPE, Popen, check_call
   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 def logcmd(cmd):
  55     if isinstance(cmd, basestring):
  56         print(cmd, file=sys.stderr)
  57     else:
  58         print(' '.join(map(quote, cmd)), file=sys.stderr)
  59
  60
  61 SubprocInfo = namedtuple('SubprocInfo', ('out', 'err', 'rc', 'p'))
  62
  63 def exo(cmd, input=None, stdin=None, stdout=PIPE, stderr=PIPE,
  64         shell=False, check=True):
  65     """Print cmd to stderr, run it, and return the resulting SubprocInfo.
  66     The keyword arguments are passed to Popen, and the defaults
  67     capture both stdout and stderr.
  68
  69     """
  70     logcmd(cmd)
  71     p = Popen(cmd,
  72               stdin=(PIPE if input else stdin),
  73               stdout=stdout,
  74               stderr=stderr,
  75               shell=shell)
  76     out, err = p.communicate(input=input)
  77     if check and p.returncode != 0:
  78         raise Exception('subprocess %r failed with status %d%s'
  79                         % (' '.join(map(quote, cmd)),
  80                            p.returncode,
  81                            (', stderr: %r' % err) if stderr else ''))
  82     return SubprocInfo(out=out, err=err, rc=p.returncode, p=p)
  83
  84 def exc(cmd, input=None, stdout=None, stderr=None, shell=False, check=True):
  85     """Print cmd to stderr, run it, and return the resulting SubprocInfo.
  86     The keyword arguments are passed to Popen, and the defaults
  87     allow stdout and stderr to pass through.
  88
  89     """
  90     return exo(cmd, input=input, stdout=stdout, stderr=stderr, shell=shell,
  91                check=check)