Commits

Anonymous committed b980a5f

Further bugfixes of run_tests.py, fixed test in mixer_test.py.

Comments (0)

Files changed (6)

 
 ################################################################################
 
-# import signal
-
-# cludge: try to mimic the standard library kill
-# signal.SIGKILL = 120
-
-# def kill(pid, sig):
-#     cludge: try to mimic the standard library kill
-#     assert sig == signal.SIGKILL 
-#     handle = win32api.OpenProcess(win32con.PROCESS_TERMINATE, 0, pid)
-#     win32api.TerminateProcess(handle, 1) # exit status
-
-# def waitpid(pid):
-#     handle = win32api.OpenProcess(win32con.SYNCHRONIZE|win32con .PROCESS_QUERY_INFORMATION , 0, pid)
-#     win32event.WaitForSingleObject(handle, win32event.INFINITE)
-#     exitCode = win32process.GetExitCodeProcess(handle)
-
-# return pid, exitCode
-
-################################################################################
-
 class Popen(subprocess.Popen):
     def recv(self, maxsize=None):
         return self._recv('stdout', maxsize)
             #http://aspn.activestate.com/ASPN/Cookbook/Python/Recipe/347462
             
             """kill function for Win32"""
-            # try:
-                # This works as well
             
             win32api.TerminateProcess(int(self._handle), 0) # returns None
                                            # handle,  # exit code
 
-            #     handle = win32api.OpenProcess(1, 0, self.pid)
-            #     try:
-            #         win32api.TerminateProcess(handle, 0)
-            #     finally:
-            #         win32api.CloseHandle(handle)
-            # except win32api.error:
-            #     return False
-            # return True
-
         def send(self, input):
             if not self.stdin:
                 return None
 
     else:
         def kill(self):
-            # TODO make return val consistent with windows
-            
-            # waitpid
-             
-            # and return a tuple containing its pid and exit status
-            # indication: a 16-bit number, whose low byte is the
-            # signal number that killed the process, and whose high
-            # byte is the exit status (if the signal number is
-            # zero); the high bit of the low byte is set if a core
-            # file was produced. Availability: Macintosh, Unix.
-
-            # try:
-
             for i, sig in enumerate([SIGTERM, SIGKILL] * 2):
                 if i % 2 == 0: os.kill(self.pid, sig)
                 time.sleep((i * (i % 2) / 5.0)  + 0.01)
 
                 killed_pid, stat = os.waitpid(self.pid, os.WNOHANG)
-                # print (i, killed_pid, stat)
-
-                if killed_pid != 0: return #True  # ???
-
-            # except OSError:
-            #     pass
-
-            # return False
-
+                if killed_pid != 0: return
+                
         def send(self, input):
             if not self.stdin:
                 return None
 
     ret_code = None
     response = []
-    # err = []
 
     t = time.time()
     while ret_code is None and ((time.time() -t) < time_out):
         ret_code = proc.poll()
         response += [proc.read_async(wait=0.1, e=0)]
-        # err      += [proc.read_async(wait=0.1, e=0, stderr=1)]
 
     if ret_code is None:
         ret_code = '"Process timed out (time_out = %s secs) ' % time_out
         except Exception, e:
             ret_code += 'and termination failed (exception: %s)"' % e
 
-    return ret_code, ''.join(response) #+ ''.join(err)
+    return ret_code, ''.join(response)
 
 ################################################################################
 
 ################################################################################
     
 if __name__ == '__main__':
-    if 1:
-        unittest.main()
-    else:
-        _example()
+    if 1: unittest.main()
+    else: _example()
 import sys, os, re, unittest, subprocess, time, optparse
 import pygame.threads 
 
-from test_runner import run_test, TEST_RESULTS_RE
+from test_runner import run_test, TEST_RESULTS_RE, TEST_RESULTS_START
 from pprint import pformat
 
 # async_sub imported if needed when run in subprocess mode
 )
 
 ################################################################################
+# Human readable output
+#
 
 COMPLETE_FAILURE_TEMPLATE = """
 ======================================================================
 ################################################################################
 # Set the command line options
 #
+# Defined in test_runner.py as it shares options, added to here
 
-USEAGE = """
+from test_runner import opt_parser
+
+opt_parser.set_usage("""
 
 Runs all the test/xxxx_test.py tests.
 
-"""
+""")
 
-opt_parser = optparse.OptionParser(USEAGE)
-
-opt_parser.add_option (
-     "-i",  "--incomplete", action = 'store_true',
-     help   = "fail incomplete tests (only single process mode)" )
-
-opt_parser.add_option (
-     "-s",  "--subprocess", action = 'store_true',
-     help   = "run test suites in subprocesses (default: same process)" )
-
-opt_parser.add_option (
-     "-d",  "--dump", action = 'store_true',
-     help   = "dump results as dict ready to eval" )
-
-opt_parser.add_option (
-     "-m",  "--multi_thread", metavar = 'THREADS', type = 'int',
-     help   = "run subprocessed tests in x THREADS" )
-
-opt_parser.add_option (
-     "-t",  "--time_out", metavar = 'SECONDS', type = 'int', default = TIME_OUT,
-     help   = "kill stalled subprocessed tests after SECONDS" )
-
-opt_parser.add_option (
-     "-f",  "--fake", metavar = "DIR",
-     help   = "run fake tests in %s%s$DIR"  % (fake_test_subdir, os.path.sep) )
-
-opt_parser.add_option (
-     "-p",  "--python", metavar = "PYTHON", default = sys.executable,
-     help   = "path to python excutable to run subproccesed tests\n"
-              "default (sys.executable): %s" % sys.executable)
+opt_parser.set_defaults (
+    python = sys.executable,
+    time_out = TIME_OUT,
+)
 
 options, args = opt_parser.parse_args()
 
 test_modules = []
 for f in sorted(os.listdir(test_subdir)):
     for match in TEST_MODULE_RE.findall(f):
+        if ((options.subprocess and match in SUBPROCESS_IGNORE) 
+             or match in IGNORE): continue
         test_modules.append(match)
+if args:    
+    test_modules = [
+        m.endswith('_test') and m or ('%s_test' % m) for m in args
+    ]    
 
 ################################################################################
 # Single process mode
 #
 
-if not options.subprocess:
-    test_utils.fail_incomplete_tests = options.incomplete
-    single_results = run_test([m for m in test_modules if m not in IGNORE])
-    if options.dump: print pformat(single_results)
+if not options.subprocess:    
+    single_results = run_test (
+        test_modules,
+        options = options
+    )
+    if options.dump: print pformat(single_results)    #TODO
     else: print single_results['output']
 
 ################################################################################
 # Subprocess mode
 #
 
-def count(results, *args):
-    for arg in args:
-        all_of = [a for a in [v.get(arg) for v in results.values()] if a]
-        if not all_of: yield 0
-        else:
-            yield sum (
-            isinstance(all_of[0], int) and all_of or (len(v) for v in all_of)
-        )
-
 def combine_results(all_results, t):
     """
 
         )
 
         if not output or (return_code and RAN_TESTS_DIV not in output):
+            # would this effect the original? TODO
             results['raw_return'] = ''.join(raw_return.splitlines(1)[:5])
             failures.append( COMPLETE_FAILURE_TEMPLATE % results )
+            all_dots += 'E'
             continue
 
         dots = DOTS.search(output).group(1)
 
 ################################################################################
 
+def count(results, *args):
+    for arg in args:
+        all_of = [a for a in [v.get(arg) for v in results.values()] if a]
+        if not all_of: yield 0
+        else:
+            yield sum (
+            isinstance(all_of[0], int) and all_of or (len(v) for v in all_of)
+        )
+
+def test_failures(results):
+    total,   = count(results, 'num_tests')
+    errors = {}
+
+    for module, result in results.items():
+        for breaker in ['errors', 'failures', 'return_code']:
+            if breaker not in result or result[breaker]:
+                if breaker not in result: total += 1
+                errors.update({module:result})
+                break
+
+    return total, errors
+
+################################################################################
+
 if options.subprocess:
     from async_sub import proc_in_time_or_kill
 
     def sub_test(module):
         print 'loading', module
-        
-        cmd = [options.python, test_runner_py, module ]
+        cmd = [options.python, test_runner_py, module ] + sys.argv[1:]
 
         return module, (cmd, test_env, working_dir), proc_in_time_or_kill (
             cmd,
             env = test_env,
             wd = working_dir,
         )
-    
+
     if options.multi_thread:
         def tmap(f, args):
             return pygame.threads.tmap (
                 num_workers = options.multi_thread
             )
     else: tmap = map
-        
 
-    test_modules = (m for m in test_modules if m not in SUBPROCESS_IGNORE)
     results = {}
 
     t = time.time()
 
-    for module, proc, (return_code, raw_return) in tmap(sub_test, test_modules):
-        cmd, test_env, working_dir = proc
+    for module, cmd, (return_code, raw_return) in tmap(sub_test, test_modules):
+        cmd, test_env, working_dir = cmd
 
         test_results = TEST_RESULTS_RE.search(raw_return)
         if test_results: 
         )
 
     untrusty_total, combined = combine_results(results, time.time() -t)
-    errors, failures, total  = count(results, 'errors', 'failures', 'num_tests')
+    total, fails = test_failures(results)
 
-    if not options.dump and untrusty_total == total:
+    if not options.dump or (options.human and untrusty_total == total):
         print combined
     else:
-        for module, result in results.items():
-            for breaker in ['errors', 'return_code', 'failures']:
-                if breaker not in result or result[breaker]:
-                    print pformat(result)
-
-        print "Tests:%s Errors:%s Failures:%s"% (total, errors, failures)
+        print TEST_RESULTS_START
+        print pformat(fails)
 
 ################################################################################

run_tests_old.py

-#!/usr/bin/env python
-import sys, os, re, unittest
-
-main_dir = os.path.split(os.path.abspath(sys.argv[0]))[0]
-test_subdir = 'test'
-
-# Make sure we're in the correct directory
-os.chdir( main_dir )
-
-# Add the modules directory to the python path    
-sys.path.insert( 0, test_subdir )
-
-# Load test util functions
-import test_utils
-
-# Load all the tests
-suite = unittest.TestSuite()
-test_module_re = re.compile('^(.+_test)\.py$')
-for file in os.listdir(test_subdir):
-    for module in test_module_re.findall(file):
-        if module == "scrap_test":
-            continue
-        print 'loading ' + module
-        __import__( module )
-        test = unittest.defaultTestLoader.loadTestsFromName( module )
-        suite.addTest( test )
-
-# Parse command line options
-if "--incomplete" in sys.argv or "-i" in sys.argv:
-    test_utils.fail_incomplete_tests = 1
-
-verbose = "--verbose" in sys.argv or "-v" in sys.argv
-
-# Run the tests
-runner = unittest.TextTestRunner()
-
-if verbose: runner.verbosity = 2
-runner.run( suite )

test/mixer_test.py

             self.assert_(mixer.get_num_channels() == i)
 
         mixer.quit()
-    
+
     def test_quit(self):
+        """ get_num_channels() Should throw pygame.error if uninitialized
+        after mixer.quit() """
+
         mixer.init()
         mixer.quit()
 
-        # assertRaises does not work here
-        # self.assertRaises(pygame.error, mixer.get_num_channels())
-
-        try:
-            chans = mixer.get_num_channels()
-        except pygame.error:
-            pass
-        except Exception:
-            self.fail()
-        else:
-            self.assert_( chans is
-                'get_num_channels() Should throw pygame.error if uninitialized '
-                'after mixer.quit()' )
+        self.assertRaises (
+            pygame.error, mixer.get_num_channels,
+        )
 
     def test_pre_init(self):
     

test/run_tests__tests/run_tests__test.py

     else:
         passes += 1
         print '%s suite comparison OK' % suite
-
+    
     if verbose or failed:
         print "difflib.Differ().compare(single, suprocessed):\n"
         print ''.join ( list(
 
 print "infinite_loop suite (subprocess mode timeout)",
 loop_test = call_proc(time_out_cmd, trunk_dir)
-assert "ERROR: all_tests_for" in loop_test
+assert "successfully terminated" in loop_test
 passes += 1
 print "OK"
 
-import sys, os, re, unittest, StringIO, time
+import sys, os, re, unittest, StringIO, time, optparse
 from pprint import pformat
 
+################################################################################
+# Set the command line options
+#
+# options are shared with run_tests.py so make sure not to conflict
+# in time more will be added here
+
+opt_parser = optparse.OptionParser()
+
+opt_parser.add_option (
+     "-i",  "--incomplete", action = 'store_true',
+     help   = "fail incomplete tests" )
+
+opt_parser.add_option (
+     "-s",  "--subprocess", action = 'store_true',
+     help   = "run test suites in subprocesses (default: same process)" )
+
+opt_parser.add_option (
+     "-d",  "--dump", action = 'store_true',
+     help   = "dump results as dict ready to eval" )
+
+opt_parser.add_option (
+     "-H",  "--human", action = 'store_true',
+     help   = "dump results as dict ready to eval if unsure" 
+              " (subprocess mode)" ) # TODO
+
+opt_parser.add_option (
+     "-m",  "--multi_thread", metavar = 'THREADS', type = 'int',
+     help   = "run subprocessed tests in x THREADS" )
+
+opt_parser.add_option (
+     "-t",  "--time_out", metavar = 'SECONDS', type = 'int',
+     help   = "kill stalled subprocessed tests after SECONDS" )
+
+opt_parser.add_option (
+     "-f",  "--fake", metavar = "DIR",
+     help   = "run fake tests in run_tests__tests/$DIR" )
+
+opt_parser.add_option (
+     "-p",  "--python", metavar = "PYTHON",
+     help   = "path to python excutable to run subproccesed tests\n"
+              "default (sys.executable): %s" % sys.executable)
+
+################################################################################
+
 TEST_RESULTS_START = "<--!! TEST RESULTS START HERE !!-->"
 TEST_RESULTS_RE = re.compile('%s\n(.*)' % TEST_RESULTS_START, re.DOTALL | re.M)
 
     io.seek(0)
     return io.read()
     
-unittest._TextTestResult.monkey = lambda self, errors: [ 
+unittest._TextTestResult.monkey = lambda self, errors: [
     (self.getDescription(e[0]), e[1]) for e in errors
 ]
 
-def run_test(module, sub_process_mode=False):
+def run_test(modules, options):
+    if isinstance(modules, str): modules = [modules]
     suite = unittest.TestSuite()
-    if not isinstance(module, list): module = [module]
 
-    for modules in module:   
-        __import__(modules)
-        print 'loading', modules
-        test = unittest.defaultTestLoader.loadTestsFromName(modules)
+    if not options.fake:
+        import test_utils
+        test_utils.fail_incomplete_tests = options.incomplete
+
+    for module in modules:
+        __import__(module)
+        print 'loading', module
+
+        # filter test by tags based on options
+        test = unittest.defaultTestLoader.loadTestsFromName(module)
         suite.addTest(test)
-    
+
     (realerr, realout), (err, out) =  redirect_output()
-    # restore_output(realerr, realout)   DEBUG
-    
+    # restore_output(realerr, realout)   #DEBUG
+
     captured = StringIO.StringIO()
     runner = unittest.TextTestRunner(stream = captured)
-    results = runner.run( suite )
+    results = runner.run(suite)
 
     captured, err, out = map(StringIOContents, (captured, err, out))
     restore_output(realerr, realout)
 
     results = (
         {
-            len(module) == 1 and module[0] or 'all_tests':
+            options.subprocess and modules[0] or 'all_tests':
             {
                 'num_tests' : results.testsRun,
                 'failures'  : results.monkey(results.failures),
         }
     )
 
-    if sub_process_mode:
+    if options.subprocess:
         print TEST_RESULTS_START
         print pformat(results)
     else:
         return results['all_tests']
 
 if __name__ == '__main__':
-    run_test(sys.argv[1], 1)
+    options, args = opt_parser.parse_args()
+    if not args: sys.exit('Called from run_tests.py, use that')
+    run_test(args[0], options)
Tip: Filter by directory path e.g. /media app.js to search for public/media/app.js.
Tip: Use camelCasing e.g. ProjME to search for ProjectModifiedEvent.java.
Tip: Filter by extension type e.g. /repo .js to search for all .js files in the /repo directory.
Tip: Separate your search with spaces e.g. /ssh pom.xml to search for src/ssh/pom.xml.
Tip: Use ↑ and ↓ arrow keys to navigate and return to view the file.
Tip: You can also navigate files with Ctrl+j (next) and Ctrl+k (previous) and view the file with Ctrl+o.
Tip: You can also navigate files with Alt+j (next) and Alt+k (previous) and view the file with Alt+o.