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     options, args = parser.parse_args()
  69     return options, args
  70
  71 def remove_coverage_files():
  72     for filename in os.listdir('.'):
  73         if filename.startswith('.coverage'):
  74             os.remove(filename)
  75
  76 def end_coverage():
  77     coverage.the_coverage.start()
  78     coverage.the_coverage.collect()
  79     coverage.the_coverage.stop()
  80     modules = [p[:-3] for p in os.listdir('../gsh') if p.endswith('.py')]
  81     coverage.report(['../gsh/%s.py' % (m) for m in modules])
  82     remove_coverage_files()
  83     # Prevent the atexit.register(the_coverage.save) from recreating the files
  84     coverage.the_coverage.usecache = coverage.the_coverage.cache = None
  85
  86 def main():
  87     options, args = parse_cmdline()
  88     if options.coverage:
  89         remove_coverage_files()
  90     if args:
  91         import_specified_tests(args)
  92     else:
  93         import_all_tests()
  94     try:
  95         unittest.main(argv=[sys.argv[0], '-v'], defaultTest='TESTS')
  96     finally:
  97         if options.coverage:
  98             end_coverage()
  99
 100 class non_interactive_spawn(pexpect.spawn):
 101     def __init__(self, argv, input_data, *args, **kwargs):
 102         pexpect.spawn.__init__(self, None, *args, **kwargs)
 103         self.use_native_pty_fork = False
 104         self.argv = argv
 105         self.input_data = input_data
 106         self.command = argv[0]
 107         self.args = argv[1:]
 108         self._spawn(self.command, self.args)
 109
 110     def _spawn__fork_pty(self):
 111         process = subprocess.Popen(self.argv, stdin=subprocess.PIPE,
 112                                               stdout=subprocess.PIPE,
 113                                               stderr=subprocess.STDOUT)
 114
 115         fd = process.stdin.fileno()
 116         while self.input_data:
 117             written = os.write(fd, self.input_data)
 118             self.input_data = self.input_data[written:]
 119         process.stdin.close()
 120         # process will be garbage collected and process.stdout closed, so
 121         # use a dupped fd.
 122         return process.pid, os.dup(process.stdout.fileno())
 123
 124 def launch_gsh(args, input_data=None):
 125     args = ['../gsh.py'] + args
 126     options, unused_args = parse_cmdline()
 127     if options.coverage:
 128         args = ['./coverage.py', '-x', '-p'] + args
 129     if options.log:
 130         logfile = open(options.log, 'a', 0644)
 131         args += ['--debug']
 132         print >> logfile, 'Launching:', str(args)
 133     else:
 134         logfile = None
 135
 136     if input_data is None:
 137         child = pexpect.spawn(args[0], args=args[1:], logfile=logfile)
 138     else:
 139         child = non_interactive_spawn(args, input_data, logfile=logfile)
 140     return child
 141
 142 if __name__ == '__main__':
 143     main()