tests/tests.py

   1 #!/usr/bin/python
   2 #
   3 # Copyright (C) 2014  Simo Sorce <simo@redhat.com>
   4 #
   5 # see file 'COPYING' for use and warranty information
   6 #
   7 # This program is free software; you can redistribute it and/or modify
   8 # it under the terms of the GNU General Public License as published by
   9 # the Free Software Foundation, either version 3 of the License, or
  10 # (at your option) any later version.
  11 #
  12 # This program is distributed in the hope that it will be useful,
  13 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  14 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  15 # GNU General Public License for more details.
  16 #
  17 # You should have received a copy of the GNU General Public License
  18 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
  19
  20 import argparse
  21 from datetime import datetime
  22 import inspect
  23 from ipsilon.util import plugin
  24 import logging
  25 import os
  26 import sys
  27 import subprocess
  28 import traceback
  29
  30
  31 logger = None
  32
  33
  34 class Tests(object):
  35
  36     def __init__(self):
  37         p = plugin.Plugins()
  38         (pathname, dummy) = os.path.split(inspect.getfile(Tests))
  39         self.plugins = p.get_plugins(pathname, 'IpsilonTest')
  40
  41
  42 def parse_args():
  43     parser = argparse.ArgumentParser(description='Ipsilon Tests Environment')
  44     parser.add_argument('--path', default='%s/testdir' % os.getcwd(),
  45                         help="Directory in which tests are run")
  46     parser.add_argument('--test', default='test1',
  47                         help="The test to run")
  48     parser.add_argument('--wrappers', default='auto',
  49                         choices=['yes', 'no', 'auto'],
  50                         help="Run the tests with socket wrappers")
  51
  52     return vars(parser.parse_args())
  53
  54
  55 def openlogs(path, name):
  56     global logger  # pylint: disable=W0603
  57     logger = logging.getLogger()
  58     try:
  59         datestr = datetime.now().strftime("%Y-%m-%d_%H:%M:%S")
  60         filename = '%s/test-%s-%s.log' % (path, name, datestr)
  61         lh = logging.FileHandler(filename)
  62     except IOError, e:
  63         print >> sys.stderr, 'Unable to open %s (%s)' % (filename, str(e))
  64         lh = logging.StreamHandler(sys.stderr)
  65     formatter = logging.Formatter('[%(asctime)s] %(message)s')
  66     lh.setFormatter(formatter)
  67     logger.addHandler(lh)
  68     logger.setLevel(logging.DEBUG)
  69
  70
  71 def try_wrappers(base, wrappers):
  72     if wrappers == 'no':
  73         return {}
  74
  75     pkgcfg = subprocess.Popen(['pkg-config', '--exists', 'socket_wrapper'])
  76     pkgcfg.wait()
  77     if pkgcfg.returncode != 0:
  78         if wrappers == 'auto':
  79             return {}
  80         else:
  81             raise ValueError('Socket Wrappers not available')
  82
  83     wrapdir = os.path.join(base, 'wrapdir')
  84     os.mkdir(wrapdir)
  85
  86     wenv = {'LD_PRELOAD': 'libsocket_wrapper.so',
  87             'SOCKET_WRAPPER_DIR': wrapdir,
  88             'SOCKET_WRAPPER_DEFAULT_IFACE': '9'}
  89
  90     return wenv
  91
  92
  93 if __name__ == '__main__':
  94
  95     args = parse_args()
  96
  97     tests = Tests()
  98     if args['test'] not in tests.plugins:
  99         print >> sys.stderr, "Unknown test [%s]" % args['test']
 100         sys.exit(1)
 101     test = tests.plugins[args['test']]
 102
 103     if not os.path.exists(args['path']):
 104         os.makedirs(args['path'])
 105
 106     openlogs(args['path'], args['test'])
 107
 108     test.setup_base(args['path'], test)
 109
 110     env = try_wrappers(test.testdir, args['wrappers'])
 111     env['PYTHONPATH'] = test.rootdir
 112
 113     try:
 114         test.setup_servers(env)
 115
 116         code = test.run(env)
 117         if code:
 118             sys.exit(code)
 119     except Exception, e:  # pylint: disable=broad-except
 120         print >> sys.stderr, "Error: %s" % repr(e)
 121         traceback.print_exc(None, sys.stderr)
 122         sys.exit(1)
 123     finally:
 124         test.wait()
 125
 126     print "FINISHED"