| 125 | self.assertRaises(TypeError, self._local, 1) |
| 126 | |
| 127 | def _test_one_class(self, c): |
| 128 | self._failed = "No error message set or cleared." |
| 129 | obj = c() |
| 130 | e1 = threading.Event() |
| 131 | e2 = threading.Event() |
| 132 | |
| 133 | def f1(): |
| 134 | obj.x = 'foo' |
| 135 | obj.y = 'bar' |
| 136 | del obj.y |
| 137 | e1.set() |
| 138 | e2.wait() |
| 139 | |
| 140 | def f2(): |
| 141 | try: |
| 142 | foo = obj.x |
| 143 | except AttributeError: |
| 144 | # This is expected -- we haven't set obj.x in this thread yet! |
| 145 | self._failed = "" # passed |
| 146 | else: |
| 147 | self._failed = ('Incorrectly got value %r from class %r\n' % |
| 148 | (foo, c)) |
| 149 | sys.stderr.write(self._failed) |
| 150 | |
| 151 | t1 = threading.Thread(target=f1) |
| 152 | t1.start() |
| 153 | e1.wait() |
| 154 | t2 = threading.Thread(target=f2) |
| 155 | t2.start() |
| 156 | t2.join() |
| 157 | # The test is done; just let t1 know it can exit, and wait for it. |
| 158 | e2.set() |
| 159 | t1.join() |
| 160 | |
| 161 | self.assertFalse(self._failed, self._failed) |
| 162 | |
| 163 | def test_threading_local(self): |
| 164 | self._test_one_class(self._local) |