Anonymous avatar Anonymous committed c308d41

some tests for pygame.threads

Comments (0)

Files changed (5)

lib/threads/__init__.py

 """
+* Experimental *
+
 Like the map function, but can use a pool of threads.
 
 Really easy to use threads.  eg.  tmap(f, alist)
 if (sys.version_info[0] == 2 and sys.version_info[1] < 5):
     from Py25Queue import Queue
     from Py25Queue import Empty
-else:   # use up to date version
+else:
+    # use up to date version
     from Queue import Queue
     from Queue import Empty
-
+    
 import threading
 Thread = threading.Thread
-#from threading import Thread
+
 STOP = object()
 FINISH = object()
-DONE_ONE = object()
-DONE_TWO = object()
 
+# DONE_ONE = object()
+# DONE_TWO = object()
 
 # a default worker queue.
 _wq = None
             a_thread.start()
 
 
-    def do(self, f, args, kwArgs):
+    def do(self, f, *args, **kwArgs):
         """ puts a function on a queue for running later.
         """
         self.queue.put((f, args, kwArgs))
             thread.join()
 
 
-    def threadloop(self, finish = False):
+    def threadloop(self): #, finish = False):
         """ Loops until all of the tasks are finished.
         """
         while True:
         """
         self.queue.join()
 
-    #def __del__(self):
-    #    self.stop()
-
-
 class FuncResult:
     """ Used for wrapping up a function call so that the results are stored
          inside the instances result attribute.
     results = []
     for sa in seq_args:
         results.append(FuncResult(f))
-        wq.do(results[-1], [sa], {})
+        wq.do(results[-1], sa)
 
 
     #wq.stop()
 
 Option to run tests in subprocesses using subprocess and async_sub. Will poll 
 tests for return code and if tests don't return after TIME_OUT, will kill 
-process with os.kill.
-
-os.kill is defined on win32 platform using win32api.TerminateProcess
+process with os.kill. On win32 platform win32api.TerminateProcess is used.
 
 Dependencies:
     async_sub.py:
 
 COMPLETE_FAILURE_TEMPLATE = """
 ======================================================================
-ERROR: all_tests_for (%s.AllTestCases)
+ERROR: all_tests_for (%(module)s.AllTestCases)
 ----------------------------------------------------------------------
 Traceback (most recent call last):
-  File "test\%s.py", line 1, in all_tests_for
+  File "test\%(module)s.py", line 1, in all_tests_for
 
-subprocess completely failed with return code of %s
+subprocess completely failed with return code of %(ret_code)s
 
-cmd: %s
+cmd: %(cmd)s
 
 return (abbrv):
-%s
+%(ret)s
 
 """  # Leave that last empty line else build page regex won't match
 
 RAN_TESTS_DIV = (70 * "-") + "\nRan"
 
-DOTS = re.compile("^([FE.]+)$", re.MULTILINE)
+DOTS = re.compile("^([FE.]*)$", re.MULTILINE)
 
 TEST_MODULE_RE = re.compile('^(.+_test)\.py$')
 
 
 all_dots = ''
 failures = []
-complete_failures = 0
 
 for cmd, module, ret_code, ret in test_results:
     if ret_code and RAN_TESTS_DIV not in ret:
         ret = ''.join(ret.splitlines(1)[:5])
-
-        failures.append (
-            COMPLETE_FAILURE_TEMPLATE % (module, module, ret_code, cmd, ret)
-        )
-        complete_failures += 1
+        failures.append( COMPLETE_FAILURE_TEMPLATE % locals() )
         continue
 
-    dots = DOTS.search(ret)
-    if not dots: continue                        # in case of empty xxxx_test.py
-    else: dots = dots.group(1)
-
+    dots = DOTS.search(ret).group(1)
     all_dots += dots
 
     if 'E' in dots or 'F' in dots:
             failure.replace( "(__main__.", "(%s." % module)
         )
 
-total_fails, total_errors = all_dots.count('F'), all_dots.count('E')
+total_fails, total_errors = map(all_dots.count, 'FE')
 total_tests = len(all_dots)
 
 print all_dots
 from test_utils import test_not_implemented
 
 import pygame
-import pygame.display
 
 class KeyModuleTest(unittest.TestCase):
     def test_import(self):

test/run_tests__tests/run_tests__test.py

 
 def call_proc(cmd):
     proc = subprocess.Popen (
-        cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, shell = 1,
+        cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
     )
     assert not proc.wait()
     return proc.stdout.read()
 
     failed = normed_single != normed_subs
     if failed:
-        print '%s suite FAILED\n' % suite    
+        print '%s suite comparison FAILED\n' % suite    
     else:
         passes += 1
-        print '%s suite OK' % suite
+        print '%s suite comparison OK' % suite
     
     if verbose or failed:
         print "difflib.Differ().compare(single, suprocessed):\n"
             ))
         )
 
-print "\n%s/%s passes" % (passes, len(test_suite_dirs))
+
+print "infinite_loop suite (subprocess mode timeout)",
+loop_cmd = subprocess_cmd + ' -t 2'
+loop_test = call_proc(loop_cmd % (sys.executable, "infinite_loop"))
+passes += 1
+print "OK"
+
+print "\n%s/%s passes" % (passes, len(test_suite_dirs) + 1)
+
 print "\n-h for help"
 
 ################################################################################

test/threads_test.py

 #################################### IMPORTS ###################################
 
-import test_utils, unittest
+import test_utils, unittest, time
 from test_utils import test_not_implemented
 
-from pygame.threads import FuncResult, tmap, WorkerQueue
+from pygame.threads import FuncResult, tmap, WorkerQueue, Empty, STOP
+
+import pygame.threads as threads
 
 ################################################################################
 
         wq = WorkerQueue()
         fr = FuncResult(f)
         fr2 = FuncResult(f2)
-        wq.do(fr, [1], {})
-        wq.do(fr2, [1], {})
+        wq.do(fr, 1)
+        wq.do(fr2, 1)
         wq.wait()
         wq.stop()
 
         # __doc__ (as of 2008-06-28) for pygame.threads.WorkerQueue.do:
 
           # puts a function on a queue for running later.
-          #         
-
-        self.assert_(test_not_implemented()) 
+          #
+        return
 
     def test_stop(self):
 
 
           # Stops the WorkerQueue, waits for all of the threads to finish up.
           #         
-
-        self.assert_(test_not_implemented()) 
+        
+        wq = WorkerQueue()
+        
+        self.assert_(len(wq.pool) > 0)
+        for t in wq.pool: self.assert_(t.isAlive())
+        
+        for i in xrange(200): wq.do(lambda x: x+1, i)
+        
+        wq.stop()
+        for t in wq.pool: self.assert_(not t.isAlive())
+        
+        self.assert_(wq.queue.get() is STOP)
 
     def test_threadloop(self):
 
           # Loops until all of the tasks are finished.
           #         
 
-        self.assert_(test_not_implemented()) 
+        self.assert_(test_not_implemented())
 
     def test_wait(self):
 
           # waits until all tasks are complete.
           #         
 
-        self.assert_(test_not_implemented()) 
+        wq = WorkerQueue()
+        
+        for i in xrange(2000): wq.do(lambda x: x+1, i)
+        wq.wait()
+
+        self.assertRaises(Empty, wq.queue.get_nowait)
+
+        wq.stop()
 
 class ThreadsModuleTest(unittest.TestCase):
     def test_benchmark_workers(self):
+        "tags:long_running"
 
         # __doc__ (as of 2008-06-28) for pygame.threads.benchmark_workers:
 
           # a_bench_func - f(data)
           # the_data - data to work on.
 
-        self.assert_(test_not_implemented()) 
+        self.assert_(test_not_implemented())
 
     def test_init(self):
 
 
           # Does a little test to see if threading is worth it.
           #   Sets up a global worker queue if it's worth it.
-          # 
+          #
           # Calling init() is not required, but is generally better to do.
 
-        self.assert_(test_not_implemented()) 
+        threads.init(8)
+
+        self.assert_(isinstance(threads._wq, WorkerQueue))
+
+        threads.quit()
 
     def test_quit(self):
 
         # __doc__ (as of 2008-06-28) for pygame.threads.quit:
 
           # cleans up everything.
-          #     
+          #
 
-        self.assert_(test_not_implemented()) 
+        threads.init(8)
+
+        threads.quit()
+
+        self.assert_(threads._wq is None)
 
     def test_tmap(self):
         # __doc__ (as of 2008-06-28) for pygame.threads.tmap:
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.