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 from bup.compat import str_type
  15 from bup.io import byte_stream
  16
  17
  18 @contextmanager
  19 def no_lingering_errors():
  20     def fail_if_errors():
  21         if helpers.saved_errors:
  22             bt = extract_stack()
  23             src_file, src_line, src_func, src_txt = bt[-4]
  24             msg = 'saved_errors ' + repr(helpers.saved_errors)
  25             print('! %-70s %s' % ('%s:%-4d %s' % (basename(src_file),
  26                                                   src_line,
  27                                                   msg),
  28                                   'FAILED'))
  29             sys.stdout.flush()
  30     fail_if_errors()
  31     helpers.clear_errors()
  32     yield
  33     fail_if_errors()
  34     helpers.clear_errors()
  35
  36
  37 # Assumes (of course) this file is at the top-level of the source tree
  38 _bup_tmp = realpath(dirname(__file__.encode('iso-8859-1')) + b'/t/tmp')
  39 try:
  40     os.makedirs(_bup_tmp)
  41 except OSError as e:
  42     if e.errno != errno.EEXIST:
  43         raise
  44
  45
  46 @contextmanager
  47 def test_tempdir(prefix):
  48     initial_failures = wvfailure_count()
  49     tmpdir = tempfile.mkdtemp(dir=_bup_tmp, prefix=prefix)
  50     yield tmpdir
  51     if wvfailure_count() == initial_failures:
  52         subprocess.call(['chmod', '-R', 'u+rwX', tmpdir])
  53         subprocess.call(['rm', '-rf', tmpdir])
  54
  55
  56 ex_res = namedtuple('SubprocResult', ['out', 'err', 'proc', 'rc'])
  57
  58 def run(cmd, check=True, input=None, **kwargs):
  59     """Run a subprocess as per subprocess.Popen(cmd, **kwargs) followed by
  60     communicate(input=input).  If check is true, then throw an
  61     exception if the subprocess exits with non-zero status.  Return a
  62     SubprocResult tuple.
  63
  64     """
  65     if input:
  66         assert 'stdin' not in kwargs
  67         kwargs['stdin'] = PIPE
  68     p = Popen(cmd, **kwargs)
  69     out, err = p.communicate(input=input)
  70     if check and p.returncode != 0:
  71         raise Exception('subprocess %r failed with status %d%s'
  72                         % (cmd, p.returncode,
  73                            (', stderr: %r' % err) if err else ''))
  74     return ex_res(out=out, err=err, proc=p, rc=p.returncode)
  75
  76 def logcmd(cmd):
  77     s = helpers.shstr(cmd)
  78     if isinstance(cmd, str_type):
  79         print(s, file=sys.stderr)
  80     else:
  81         # bytes - for now just continue to pass it through given
  82         # bup-python wrapper
  83         print(s.decode('iso-8859-1'), file=sys.stderr)
  84
  85 def ex(cmd, **kwargs):
  86     """Print cmd to stderr and then run it as per ex(...).
  87     Print the subprocess stderr to stderr if stderr=PIPE and there's
  88     any data.
  89     """
  90     logcmd(cmd)
  91     result = run(cmd, **kwargs)
  92     if result.err:
  93         sys.stderr.flush()
  94         byte_stream(sys.stderr).write(result.err)
  95     return result
  96
  97 def exo(cmd, **kwargs):
  98     """Print cmd to stderr and then run it as per ex(..., stdout=PIPE).
  99     Print the subprocess stderr to stderr if stderr=PIPE and there's
 100     any data.
 101
 102     """
 103     assert 'stdout' not in kwargs
 104     kwargs['stdout'] = PIPE
 105     return ex(cmd, **kwargs)