tests/gsh_tests.py

   1 #!/usr/bin/env python
   2 #
   3 # This program is free software; you can redistribute it and/or modify
   4 # it under the terms of the GNU General Public License as published by
   5 # the Free Software Foundation; either version 2 of the License, or
   6 # (at your option) any later version.
   7 #
   8 # This program is distributed in the hope that it will be useful,
   9 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  10 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  11 # GNU Library General Public License for more details.
  12 #
  13 # You should have received a copy of the GNU General Public License
  14 # along with this program; if not, write to the Free Software
  15 # Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
  16 #
  17 # See the COPYING file for license information.
  18 #
  19 # Copyright (c) 2007, 2008 Guillaume Chazarain <guichaz@gmail.com>
  20
  21 import os
  22 import unittest
  23 import sys
  24 import optparse
  25 import pexpect
  26 import subprocess
  27
  28 import coverage
  29
  30 TESTS = unittest.TestSuite()
  31
  32 def iter_over_all_tests():
  33     py_files = [p for p in os.listdir('tests') if p.endswith('.py')]
  34     tests = list(set([p[:p.index('.')] for p in py_files]))
  35     for name in tests:
  36         module = getattr(__import__('tests.' + name), name)
  37         for module_content in dir(module):
  38             candidate = getattr(module, module_content)
  39             if not isinstance(candidate, type):
  40                 continue
  41             if not issubclass(candidate, unittest.TestCase):
  42                 continue
  43             suite = unittest.defaultTestLoader.loadTestsFromTestCase(candidate)
  44             for test_method in suite:
  45                 yield test_method
  46
  47 def import_all_tests():
  48     for test in iter_over_all_tests():
  49         TESTS.addTest(test)
  50
  51 def import_specified_tests(names):
  52     for test in iter_over_all_tests():
  53         test_name = test.id().split('.')[-1]
  54         if test_name in names:
  55             names.remove(test_name)
  56             TESTS.addTest(test)
  57     if names:
  58         print 'Cannot find tests:', names
  59         sys.exit(1)
  60
  61 def parse_cmdline():
  62     usage='Usage: %s [OPTIONS...] [TESTS...]' % sys.argv[0]
  63     parser = optparse.OptionParser(usage=usage)
  64     parser.add_option('--coverage', action='store_true', dest='coverage',
  65                       default=False, help='include coverage tests')
  66     parser.add_option('--log', type='str', dest='log',
  67                       help='log all pexpect I/O and gsh debug info')
  68     parser.add_option('--python', type='str', dest='python', default='python',
  69                       help='python binary to use')
  70     options, args = parser.parse_args()
  71     return options, args
  72
  73 def remove_coverage_files():
  74     for filename in os.listdir('.'):
  75         if filename.startswith('.coverage'):
  76             os.remove(filename)
  77
  78 def end_coverage():
  79     coverage.the_coverage.start()
  80     coverage.the_coverage.collect()
  81     coverage.the_coverage.stop()
  82     modules = [p[:-3] for p in os.listdir('../gsh') if p.endswith('.py')]
  83     coverage.report(['../gsh/%s.py' % (m) for m in modules])
  84     remove_coverage_files()
  85     # Prevent the atexit.register(the_coverage.save) from recreating the files
  86     coverage.the_coverage.usecache = coverage.the_coverage.cache = None
  87
  88 def main():
  89     options, args = parse_cmdline()
  90     if options.coverage:
  91         remove_coverage_files()
  92     if args:
  93         import_specified_tests(args)
  94     else:
  95         import_all_tests()
  96     try:
  97         unittest.main(argv=[sys.argv[0], '-v'], defaultTest='TESTS')
  98     finally:
  99         if options.coverage:
 100             end_coverage()
 101
 102 class non_interactive_spawn(pexpect.spawn):
 103     def __init__(self, argv, input_data, *args, **kwargs):
 104         pexpect.spawn.__init__(self, None, *args, **kwargs)
 105         self.use_native_pty_fork = False
 106         self.argv = argv
 107         self.input_data = input_data
 108         self.command = argv[0]
 109         self.args = argv[1:]
 110         self._spawn(self.command, self.args)
 111
 112     def _spawn__fork_pty(self):
 113         process = subprocess.Popen(self.argv, stdin=subprocess.PIPE,
 114                                               stdout=subprocess.PIPE,
 115                                               stderr=subprocess.STDOUT)
 116
 117         fd = process.stdin.fileno()
 118         while self.input_data:
 119             written = os.write(fd, self.input_data)
 120             self.input_data = self.input_data[written:]
 121         process.stdin.close()
 122         # process will be garbage collected and process.stdout closed, so
 123         # use a dupped fd.
 124         return process.pid, os.dup(process.stdout.fileno())
 125
 126 def launch_gsh(args, input_data=None):
 127     args = ['../gsh.py'] + args
 128     options, unused_args = parse_cmdline()
 129     if options.coverage:
 130         args = ['./coverage.py', '-x', '-p'] + args
 131     args = [options.python] + args
 132     if options.log:
 133         logfile = open(options.log, 'a', 0644)
 134         args += ['--debug']
 135         print >> logfile, 'Launching:', str(args)
 136     else:
 137         logfile = None
 138
 139     if input_data is None:
 140         child = pexpect.spawn(args[0], args=args[1:], logfile=logfile)
 141     else:
 142         child = non_interactive_spawn(args, input_data, logfile=logfile)
 143     return child
 144
 145 if __name__ == '__main__':
 146     main()