| Index: Lib/test/test_threading.py |
| =================================================================== |
| --- Lib/test/test_threading.py (revision 66142) |
| +++ Lib/test/test_threading.py (working copy) |
| @@ -438,10 +438,61 @@ |
| 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__": |