'''usage: python run_tests.py [--url=<base-url>] [--fs-type=<fs-type>]
[--verbose] [--cleanup] [--enable-sasl] [--parallel]
[--http-library=<http-library>]
[--config-file=<file>]
[--server-minor-version=<version>] <abs_srcdir> <abs_builddir>
<prog ...>
The optional flags and the first two parameters are passed unchanged
to the TestHarness constructor. All other parameters are names of
test programs.
'''
import os, sys
import getopt
try:
my_getopt = getopt.gnu_getopt
except AttributeError:
my_getopt = getopt.getopt
class TestHarness:
'''Test harness for Subversion tests.
'''
def __init__(self, abs_srcdir, abs_builddir, logfile,
base_url=None, fs_type=None, http_library=None,
server_minor_version=None, verbose=None,
cleanup=None, enable_sasl=None, parallel=None, config_file=None,
fsfs_sharding=None, fsfs_packing=None,
list_tests=None, svn_bin=None):
'''Construct a TestHarness instance.
ABS_SRCDIR and ABS_BUILDDIR are the source and build directories.
LOGFILE is the name of the log file.
BASE_URL is the base url for DAV tests.
FS_TYPE is the FS type for repository creation.
HTTP_LIBRARY is the HTTP library for DAV-based communications.
SERVER_MINOR_VERSION is the minor version of the server being tested.
SVN_BIN is the path where the svn binaries are installed.
'''
self.srcdir = abs_srcdir
self.builddir = abs_builddir
self.logfile = logfile
self.base_url = base_url
self.fs_type = fs_type
self.http_library = http_library
self.server_minor_version = server_minor_version
self.verbose = verbose
self.cleanup = cleanup
self.enable_sasl = enable_sasl
self.parallel = parallel
self.fsfs_sharding = fsfs_sharding
self.fsfs_packing = fsfs_packing
if fsfs_packing is not None and fsfs_sharding is None:
raise Exception('--fsfs-packing requires --fsfs-sharding')
self.config_file = None
if config_file is not None:
self.config_file = os.path.abspath(config_file)
self.list_tests = list_tests
self.svn_bin = svn_bin
self.log = None
def run(self, list):
'Run all test programs given in LIST.'
self._open_log('w')
failed = 0
for cnt, prog in enumerate(list):
failed = self._run_test(prog, cnt, len(list)) or failed
self._open_log('r')
log_lines = self.log.readlines()
skipped = [x for x in log_lines if x[:6] == 'SKIP: ']
if skipped:
print('At least one test was SKIPPED, checking ' + self.logfile)
for x in skipped:
sys.stdout.write(x)
xfailed = [x for x in log_lines if x[:6] == 'XFAIL:']
if xfailed:
print('At least one test XFAILED, checking ' + self.logfile)
for x in xfailed:
sys.stdout.write(x)
failed_list = [x for x in log_lines if x[:6] == 'FAIL: ']
if failed_list:
print('At least one test FAILED, checking ' + self.logfile)
for x in failed_list:
sys.stdout.write(x)
xpassed = [x for x in log_lines if x[:6] == 'XPASS:']
if xpassed:
print('At least one test XPASSED, checking ' + self.logfile)
for x in xpassed:
sys.stdout.write(x)
if skipped or xfailed or failed_list or xpassed:
print('Summary of test results:')
if skipped:
print(' %d test%s SKIPPED' % (len(skipped), 's'*min(len(skipped), 1)))
if xfailed:
print(' %d test%s XFAILED' % (len(xfailed), 's'*min(len(xfailed), 1)))
if failed_list:
print(' %d test%s FAILED' % (len(failed_list),
's'*min(len(failed_list), 1)))
if xpassed:
print(' %d test%s XPASSED' % (len(xpassed), 's'*min(len(xpassed), 1)))
self._close_log()
return failed
def _open_log(self, mode):
'Open the log file with the required MODE.'
self._close_log()
self.log = open(self.logfile, mode)
def _close_log(self):
'Close the log file.'
if not self.log is None:
self.log.close()
self.log = None
def _run_test(self, prog, test_nr, total_tests):
"Run a single test. Return the test's exit code."
def quote(arg):
if sys.platform == 'win32':
return '"' + arg + '"'
else:
return arg
progdir, progbase = os.path.split(prog)
sys.stdout.write('Running all tests in %s [%d/%d]...' % (
progbase, test_nr + 1, total_tests))
self.log.write('START: %s\n' % progbase)
self.log.flush()
if progbase[-3:] == '.py':
progname = sys.executable
cmdline = [quote(progname),
quote(os.path.join(self.srcdir, prog))]
if self.base_url is not None:
cmdline.append(quote('--url=' + self.base_url))
if self.enable_sasl is not None:
cmdline.append('--enable-sasl')
if self.parallel is not None:
cmdline.append('--parallel')
if self.config_file is not None:
cmdline.append(quote('--config-file=' + self.config_file))
elif os.access(prog, os.X_OK):
progname = './' + progbase
cmdline = [quote(progname),
quote('--srcdir=' + os.path.join(self.srcdir, progdir))]
if self.config_file is not None:
cmdline.append(quote('--config-file=' + self.config_file))
else:
print('Don\'t know what to do about ' + progbase)
sys.exit(1)
if self.verbose is not None:
cmdline.append('--verbose')
if self.cleanup is not None:
cmdline.append('--cleanup')
if self.fs_type is not None:
cmdline.append(quote('--fs-type=' + self.fs_type))
if self.http_library is not None:
cmdline.append(quote('--http-library=' + self.http_library))
if self.server_minor_version is not None:
cmdline.append(quote('--server-minor-version=' + self.server_minor_version))
if self.list_tests is not None:
cmdline.append('--list')
if self.svn_bin is not None:
cmdline.append(quote('--bin=' + self.svn_bin))
if self.fsfs_sharding is not None:
cmdline.append('--fsfs-sharding=%d' % self.fsfs_sharding)
if self.fsfs_packing is not None:
cmdline.append('--fsfs-packing')
old_cwd = os.getcwd()
try:
os.chdir(progdir)
failed = self._run_prog(progname, cmdline)
except:
os.chdir(old_cwd)
raise
else:
os.chdir(old_cwd)
if failed == 1:
print('FAILURE')
elif failed:
self.log.write('FAIL: %s: Unknown test failure see tests.log.\n\n' % progbase)
self.log.flush()
print('FAILURE')
else:
print('success')
self.log.write('END: %s\n\n' % progbase)
return failed
def _run_prog(self, progname, arglist):
'''Execute the file PROGNAME in a subprocess, with ARGLIST as its
arguments (a list/tuple of arg0..argN), redirecting standard output and
error to the log file. Return the command's exit code.'''
def restore_streams(stdout, stderr):
os.dup2(stdout, 1)
os.dup2(stderr, 2)
os.close(stdout)
os.close(stderr)
sys.stdout.flush()
sys.stderr.flush()
self.log.flush()
old_stdout = os.dup(1)
old_stderr = os.dup(2)
try:
os.dup2(self.log.fileno(), 1)
os.dup2(self.log.fileno(), 2)
rv = os.spawnv(os.P_WAIT, progname, arglist)
except:
restore_streams(old_stdout, old_stderr)
raise
else:
restore_streams(old_stdout, old_stderr)
return rv
def main():
try:
opts, args = my_getopt(sys.argv[1:], 'u:f:vc',
['url=', 'fs-type=', 'verbose', 'cleanup',
'http-library=', 'server-minor-version=',
'fsfs-packing', 'fsfs-sharding=',
'enable-sasl', 'parallel', 'config-file='])
except getopt.GetoptError:
args = []
if len(args) < 3:
print(__doc__)
sys.exit(2)
base_url, fs_type, verbose, cleanup, enable_sasl, http_library, \
server_minor_version, fsfs_sharding, fsfs_packing, parallel, \
config_file = \
None, None, None, None, None, None, None, None, None, None, None
for opt, val in opts:
if opt in ['-u', '--url']:
base_url = val
elif opt in ['-f', '--fs-type']:
fs_type = val
elif opt in ['--http-library']:
http_library = val
elif opt in ['--fsfs-sharding']:
fsfs_sharding = int(val)
elif opt in ['--fsfs-packing']:
fsfs_packing = 1
elif opt in ['--server-minor-version']:
server_minor_version = val
elif opt in ['-v', '--verbose']:
verbose = 1
elif opt in ['-c', '--cleanup']:
cleanup = 1
elif opt in ['--enable-sasl']:
enable_sasl = 1
elif opt in ['--parallel']:
parallel = 1
elif opt in ['--config-file']:
config_file = val
else:
raise getopt.GetoptError
th = TestHarness(args[0], args[1],
os.path.abspath('tests.log'),
base_url, fs_type, http_library, server_minor_version,
verbose, cleanup, enable_sasl, parallel, config_file,
fsfs_sharding, fsfs_packing)
failed = th.run(args[2:])
if failed:
sys.exit(1)
if __name__ == '__main__':
main()