[Jython-checkins] SF.net SVN: jython:[6452] trunk/jython/Lib/test/test_list_jy.py

[<zy...@us...>]

Revision: 6452
          http://jython.svn.sourceforge.net/jython/?rev=6452&view=rev
Author:   zyasoft
Date:     2009-06-05 15:50:54 +0000 (Fri, 05 Jun 2009)

Log Message:
-----------
Minor enhancement of thread safety test case

Modified Paths:
--------------
    trunk/jython/Lib/test/test_list_jy.py

Modified: trunk/jython/Lib/test/test_list_jy.py
===================================================================
--- trunk/jython/Lib/test/test_list_jy.py	2009-06-05 13:13:00 UTC (rev 6451)
+++ trunk/jython/Lib/test/test_list_jy.py	2009-06-05 15:50:54 UTC (rev 6452)
@@ -1,4 +1,5 @@
 import unittest
+import random
 import threading
 import time
 from test import test_support
@@ -53,30 +54,54 @@
 
 class ThreadSafetyTestCase(unittest.TestCase):
     
-    def setUp(self):
-        self.globalList = []
-        self.threads = []
+    def run_threads(self, f, num=10):
+        threads = []
+        for i in xrange(num):
+            t = threading.Thread(target=f)
+            t.start()
+            threads.append(t)
+        for t in threads:
+            t.join(1.)
+        for t in threads:
+            self.assertFalse(t.isAlive())
 
     def test_append_remove(self):
         # derived from Itamar Shtull-Trauring's test for issue 521701
+        lst = []
         def tester():
             ct = threading.currentThread()
             for i in range(1000):
-                self.globalList.append(ct)
+                lst.append(ct)
                 time.sleep(0.0001)
-                self.globalList.remove(ct)
-        for i in range(10):
-            t = threading.Thread(target=tester)
-            t.start()
-            self.threads.append(t)
-        
-        for t in self.threads:
-            t.join(1.)
-        for t in self.threads:
-            self.assertFalse(t.isAlive())
-        self.assertEqual(self.globalList, [])
+                lst.remove(ct)
+        self.run_threads(tester)
+        self.assertEqual(lst, [])
 
+    def test_sort(self):
+        lst = []
+        def tester():
+            ct = threading.currentThread()
+            for i in range(1000):
+                lst.append(ct)
+                lst.sort()
+                lst.remove(ct)
+                time.sleep(0.0001)
+        self.run_threads(tester)
+        self.assertEqual(lst, [])
 
+    def test_count_reverse(self):
+        lst = [0,1,2,3,4,5,6,7,8,9,10,0]
+        def tester():
+            ct = threading.currentThread()
+            for i in range(1000):
+                self.assertEqual(lst[0], 0)
+                if random.random() > 0.5:
+                    time.sleep(0.0001)
+                lst.reverse()
+                self.assertEqual(lst.count(0), 2)
+                self.assert_(lst[1] in (1,10))
+        self.run_threads(tester)
+
 def test_main():
     test_support.run_unittest(ListTestCase, ThreadSafetyTestCase)
 


This was sent by the SourceForge.net collaborative development platform, the world's largest Open Source development site.