pytest-thread-autojump.py

@pytest.mark.trio
async def test_worker_thread_autojump(nursery, autojump_clock):
    def slowish_task():
        for i in range(1000000):
            pass
        return 'foo'

    async def async_run_task(queue):
        result = await trio.run_sync_in_worker_thread(slowish_task)
        await queue.put(result)

    queue = trio.Queue(1)
    nursery.start_soon(async_run_task, queue)
    with trio.fail_after(300):
        assert await queue.get() == 'foo'