Method test_context_threads_1

Lib/test/test_context.py:359–374

(self)

    @threading_helper.requires_working_threading()
    @unittest.skipIf(sys.platform == 'darwin', 'TODO: RUSTPYTHON; Flaky on Mac, self.assertEqual(cvar.get(), num + i) AssertionError: 8 != 12')
    def test_context_threads_1(self):
        cvar = contextvars.ContextVar('cvar')

        def sub(num):
            for i in range(10):
                cvar.set(num + i)
                time.sleep(random.uniform(0.001, 0.05))
                self.assertEqual(cvar.get(), num + i)
            return num

        tp = concurrent.futures.ThreadPoolExecutor(max_workers=10)
        try:
            results = list(tp.map(sub, range(10)))
        finally:
            tp.shutdown()
        self.assertEqual(results, list(range(10)))


# HAMT Tests