| 205 | " and kwargs failed") |
| 206 | |
| 207 | def test_multi_thread_creation(self): |
| 208 | def queue_mark(queue, delay): |
| 209 | time.sleep(delay) |
| 210 | queue.put(_thread.get_ident()) |
| 211 | |
| 212 | thread_count = 5 |
| 213 | testing_queue = queue.Queue(thread_count) |
| 214 | |
| 215 | if support.verbose: |
| 216 | print() |
| 217 | print("*** Testing multiple thread creation " |
| 218 | "(will take approx. %s to %s sec.) ***" % ( |
| 219 | DELAY, thread_count)) |
| 220 | |
| 221 | for count in range(thread_count): |
| 222 | if DELAY: |
| 223 | local_delay = round(random.random(), 1) |
| 224 | else: |
| 225 | local_delay = 0 |
| 226 | _thread.start_new_thread(queue_mark, |
| 227 | (testing_queue, local_delay)) |
| 228 | time.sleep(DELAY) |
| 229 | if support.verbose: |
| 230 | print('done') |
| 231 | self.assertEqual(testing_queue.qsize(), thread_count, |
| 232 | "Not all %s threads executed properly " |
| 233 | "after %s sec." % (thread_count, DELAY)) |
| 234 | |
| 235 | def test_args_not_tuple(self): |
| 236 | """ |