(self)
| 40 | longMessage = True |
| 41 | |
| 42 | def test_thread_local(self): |
| 43 | q1 = Queue() |
| 44 | q2 = Queue() |
| 45 | thread1 = TestThread(2, q1) |
| 46 | thread2 = TestThread(40, q2) |
| 47 | |
| 48 | thread1.start() |
| 49 | thread2.start() |
| 50 | |
| 51 | result1 = q1.get(True, 1) |
| 52 | result2 = q2.get(True, 1) |
| 53 | |
| 54 | thread1.join() |
| 55 | thread2.join() |
| 56 | |
| 57 | print(result1) |
| 58 | print(result2) |
| 59 | |
| 60 | # Result 1 and 2 shoudl be different |
| 61 | self.assertNotEqual(result1, result2) |
| 62 | |
| 63 | # This thread should have no cache |
| 64 | self.assertIsNone(_get_cache()) |
| 65 | |
| 66 | |
| 67 | if __name__ == "__main__": |
nothing calls this directly
no test coverage detected