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 import inspect
  22 from ipsilon.util import plugin
  23 import os
  24 import sys
  25 import subprocess
  26 import time
  27 import traceback
  28 from helpers.common import WRAP_HOSTNAME  # pylint: disable=relative-import
  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 try_wrappers(base, wrappers):
  56     if wrappers == 'no':
  57         return {}
  58
  59     pkgcfg = subprocess.Popen(['pkg-config', '--exists', 'socket_wrapper'])
  60     pkgcfg.wait()
  61     if pkgcfg.returncode != 0:
  62         if wrappers == 'auto':
  63             return {}
  64         else:
  65             raise ValueError('Socket Wrappers not available')
  66
  67     pkgcfg = subprocess.Popen(['pkg-config', '--exists', 'nss_wrapper'])
  68     pkgcfg.wait()
  69     if pkgcfg.returncode != 0:
  70         if wrappers == 'auto':
  71             return {}
  72         else:
  73             raise ValueError('Nss Wrappers not available')
  74
  75     wrapdir = os.path.join(base, 'wrapdir')
  76     os.mkdir(wrapdir)
  77
  78     hosts_file = os.path.join(base, 'hosts')
  79     with open(hosts_file, 'w+') as f:
  80         f.write('127.0.0.9 %s\n' % WRAP_HOSTNAME)
  81
  82     wenv = {'LD_PRELOAD': 'libsocket_wrapper.so libnss_wrapper.so',
  83             'SOCKET_WRAPPER_DIR': wrapdir,
  84             'SOCKET_WRAPPER_DEFAULT_IFACE': '9',
  85             'SOCKET_WRAPPER_DEBUGLEVEL': '1',
  86             'NSS_WRAPPER_HOSTNAME': WRAP_HOSTNAME,
  87             'NSS_WRAPPER_HOSTS': hosts_file}
  88
  89     return wenv
  90
  91
  92 if __name__ == '__main__':
  93
  94     args = parse_args()
  95
  96     tests = Tests()
  97     if args['test'] not in tests.plugins:
  98         print >> sys.stderr, "Unknown test [%s]" % args['test']
  99         sys.exit(1)
 100     test = tests.plugins[args['test']]
 101
 102     if not os.path.exists(args['path']):
 103         os.makedirs(args['path'])
 104
 105     test.setup_base(args['path'], test)
 106
 107     env = try_wrappers(test.testdir, args['wrappers'])
 108     env['PYTHONPATH'] = test.rootdir
 109     env['TESTDIR'] = test.testdir
 110
 111     try:
 112         test.setup_servers(env)
 113
 114         code = test.run(env)
 115         if code:
 116             sys.exit(code)
 117     except Exception, e:  # pylint: disable=broad-except
 118         print >> sys.stderr, "Error: %s" % repr(e)
 119         traceback.print_exc(None, sys.stderr)
 120         sys.exit(1)
 121     finally:
 122         test.wait()
 123
 124     # Wait until all of the sockets are closed by the OS
 125     time.sleep(0.5)
 126     print "FINISHED"