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
11 from wvtest import WVPASSEQ, wvfailure_count
13 from bup import helpers
14 from bup.compat import fsencode, str_type
15 from bup.io import byte_stream
19 def no_lingering_errors():
21 if helpers.saved_errors:
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),
31 helpers.clear_errors()
34 helpers.clear_errors()
37 # Assumes (of course) this file is at the top-level of the source tree
38 _bup_tmp = realpath(dirname(fsencode(__file__))) + b'/test/tmp'
42 if e.errno != errno.EEXIST:
47 def test_tempdir(prefix):
48 initial_failures = wvfailure_count()
49 tmpdir = tempfile.mkdtemp(dir=_bup_tmp, prefix=prefix)
51 if wvfailure_count() == initial_failures:
52 subprocess.call(['chmod', '-R', 'u+rwX', tmpdir])
53 subprocess.call(['rm', '-rf', tmpdir])
56 ex_res = namedtuple('SubprocResult', ['out', 'err', 'proc', 'rc'])
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
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'
73 (', stderr: %r' % err) if err else ''))
74 return ex_res(out=out, err=err, proc=p, rc=p.returncode)
77 s = helpers.shstr(cmd)
78 if isinstance(cmd, str_type):
79 print(s, file=sys.stderr)
81 # bytes - for now just escape it
82 print(s.decode(errors='backslashreplace'), file=sys.stderr)
84 def ex(cmd, **kwargs):
85 """Print cmd to stderr and then run it as per ex(...).
86 Print the subprocess stderr to stderr if stderr=PIPE and there's
90 result = run(cmd, **kwargs)
93 byte_stream(sys.stderr).write(result.err)
96 def exo(cmd, **kwargs):
97 """Print cmd to stderr and then run it as per ex(..., stdout=PIPE).
98 Print the subprocess stderr to stderr if stderr=PIPE and there's
102 assert 'stdout' not in kwargs
103 kwargs['stdout'] = PIPE
104 return ex(cmd, **kwargs)