(self)
| 172 | """Test thread creation.""" |
| 173 | |
| 174 | def test_arg_passing(self): |
| 175 | #Make sure that parameter passing works. |
| 176 | def arg_tester(queue, arg1=False, arg2=False): |
| 177 | """Use to test _thread.start_new_thread() passes args properly.""" |
| 178 | queue.put((arg1, arg2)) |
| 179 | |
| 180 | testing_queue = queue.Queue(1) |
| 181 | _thread.start_new_thread(arg_tester, (testing_queue, True, True)) |
| 182 | result = testing_queue.get() |
| 183 | self.assertTrue(result[0] and result[1], |
| 184 | "Argument passing for thread creation " |
| 185 | "using tuple failed") |
| 186 | |
| 187 | _thread.start_new_thread( |
| 188 | arg_tester, |
| 189 | tuple(), |
| 190 | {'queue':testing_queue, 'arg1':True, 'arg2':True}) |
| 191 | |
| 192 | result = testing_queue.get() |
| 193 | self.assertTrue(result[0] and result[1], |
| 194 | "Argument passing for thread creation " |
| 195 | "using kwargs failed") |
| 196 | |
| 197 | _thread.start_new_thread( |
| 198 | arg_tester, |
| 199 | (testing_queue, True), |
| 200 | {'arg2':True}) |
| 201 | |
| 202 | result = testing_queue.get() |
| 203 | self.assertTrue(result[0] and result[1], |
| 204 | "Argument passing for thread creation using both tuple" |
| 205 | " and kwargs failed") |
| 206 | |
| 207 | def test_multi_thread_creation(self): |
| 208 | def queue_mark(queue, delay): |
nothing calls this directly
no test coverage detected