(self)
| 377 | @isolated_context |
| 378 | @threading_helper.requires_working_threading() |
| 379 | def test_context_threads_1(self): |
| 380 | cvar = contextvars.ContextVar('cvar') |
| 381 | |
| 382 | def sub(num): |
| 383 | for i in range(10): |
| 384 | cvar.set(num + i) |
| 385 | time.sleep(random.uniform(0.001, 0.05)) |
| 386 | self.assertEqual(cvar.get(), num + i) |
| 387 | return num |
| 388 | |
| 389 | tp = concurrent.futures.ThreadPoolExecutor(max_workers=10) |
| 390 | try: |
| 391 | results = list(tp.map(sub, range(10))) |
| 392 | finally: |
| 393 | tp.shutdown() |
| 394 | self.assertEqual(results, list(range(10))) |
| 395 | |
| 396 | @isolated_context |
| 397 | @threading_helper.requires_working_threading() |
nothing calls this directly
no test coverage detected