thread._local, threading.local Python issue 1868 & 3710

Lib/test/test_threading.py

 # Very rudimentary test of threading module
 
 import test.test_support
 from test.test_support import verbose
 import random
 import re
 import sys
 import threading
 import thread
 import time
@@ skipping 418 matching lines @@
         self.assertRaises(RuntimeError, current_thread.join);
 
     def test_joining_inactive_thread(self):
         thread = threading.Thread()
         self.assertRaises(RuntimeError, thread.join)
 
     def test_daemonize_active_thread(self):
         thread = threading.Thread()
         thread.start()
         self.assertRaises(RuntimeError, setattr, thread, "daemon", True)
+
+
+# This test came from Ben Cottrell in http://bugs.python.org/issue1868
+class TestThreadingLocal(unittest.TestCase):
+    def setUp(self):
+        self._failed = "No error message set or cleared."
+
+    def _test_one_class(self, c):
+        obj = c()
+        e1 = threading.Event()
+        e2 = threading.Event()
+
+        def f1():
+            obj.x = 'foo'
+            obj.y = 'bar'
+            del obj.y
+            e1.set()
+            e2.wait()
+
+        def f2():
+            try:
+                foo = obj.x
+            except AttributeError:
+                # This is expected -- we haven't set obj.x in this thread yet!
+                self._failed = ""  # passed
+            else:
+                self._failed = ('Incorrectly got value %r from class %r\n' %
+                                (foo, c))
+                sys.stderr.write(self._failed)
+
+        t1 = threading.Thread(target=f1)
+        t1.start()
+        e1.wait()
+        t2 = threading.Thread(target=f2)
+        t2.start()
+        t2.join()
+
+        # The test is done; just let t1 know it can exit, and wait for it.
+        e2.set()
+        t1.join()
+
+    def test_threading_local(self):
+        self._test_one_class(threading.local)
+        self.assert_(not self._failed, self._failed)
+
+    def test_threading_local_subclass(self):
+        class LocalSubclass(threading.local):
+            """To test that subclasses behave properly."""
+        self._test_one_class(LocalSubclass)
+        self.assert_(not self._failed, self._failed)
 
 
 def test_main():
     test.test_support.run_unittest(ThreadTests,
                                    ThreadJoinOnShutdown,
                                    ThreadingExceptionTests,
+                                   TestThreadingLocal,
                                    )
 
 if __name__ == "__main__":
     test_main()