Insert a small timeout before reporting the test successful
[cascardo/ipsilon.git] / 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
29
30 logger = None
31
32
33 class Tests(object):
34
35     def __init__(self):
36         p = plugin.Plugins()
37         (pathname, dummy) = os.path.split(inspect.getfile(Tests))
38         self.plugins = p.get_plugins(pathname, 'IpsilonTest')
39
40
41 def parse_args():
42     parser = argparse.ArgumentParser(description='Ipsilon Tests Environment')
43     parser.add_argument('--path', default='%s/testdir' % os.getcwd(),
44                         help="Directory in which tests are run")
45     parser.add_argument('--test', default='test1',
46                         help="The test to run")
47     parser.add_argument('--wrappers', default='auto',
48                         choices=['yes', 'no', 'auto'],
49                         help="Run the tests with socket wrappers")
50
51     return vars(parser.parse_args())
52
53
54 def try_wrappers(base, wrappers):
55     if wrappers == 'no':
56         return {}
57
58     pkgcfg = subprocess.Popen(['pkg-config', '--exists', 'socket_wrapper'])
59     pkgcfg.wait()
60     if pkgcfg.returncode != 0:
61         if wrappers == 'auto':
62             return {}
63         else:
64             raise ValueError('Socket Wrappers not available')
65
66     wrapdir = os.path.join(base, 'wrapdir')
67     os.mkdir(wrapdir)
68
69     wenv = {'LD_PRELOAD': 'libsocket_wrapper.so',
70             'SOCKET_WRAPPER_DIR': wrapdir,
71             'SOCKET_WRAPPER_DEFAULT_IFACE': '9'}
72
73     return wenv
74
75
76 if __name__ == '__main__':
77
78     args = parse_args()
79
80     tests = Tests()
81     if args['test'] not in tests.plugins:
82         print >> sys.stderr, "Unknown test [%s]" % args['test']
83         sys.exit(1)
84     test = tests.plugins[args['test']]
85
86     if not os.path.exists(args['path']):
87         os.makedirs(args['path'])
88
89     test.setup_base(args['path'], test)
90
91     env = try_wrappers(test.testdir, args['wrappers'])
92     env['PYTHONPATH'] = test.rootdir
93
94     try:
95         test.setup_servers(env)
96
97         code = test.run(env)
98         if code:
99             sys.exit(code)
100     except Exception, e:  # pylint: disable=broad-except
101         print >> sys.stderr, "Error: %s" % repr(e)
102         traceback.print_exc(None, sys.stderr)
103         sys.exit(1)
104     finally:
105         test.wait()
106
107     # Wait until all of the sockets are closed by the OS
108     time.sleep(0.5)
109     print "FINISHED"