Skip to content

Instantly share code, notes, and snippets.

@shamshirz

shamshirz/test_concurrency.py

Created February 4, 2022 20:44
Show Gist options
  • Save shamshirz/b6c57c9bdeb91cb899290b9919b07f0a to your computer and use it in GitHub Desktop.
Save shamshirz/b6c57c9bdeb91cb899290b9919b07f0a to your computer and use it in GitHub Desktop.
Download ZIP

Revisions

  1. shamshirz created this gist Feb 4, 2022.
    48 changes: 48 additions & 0 deletions test_concurrency.py
    View file
    Original file line number Diff line number Diff line change
    @@ -0,0 +1,48 @@
    import asyncio
    import time


    async def await_sleep():
    await asyncio.create_task(sleep())


    async def await_multiple():
    task1 = asyncio.create_task(sleep())
    task2 = asyncio.create_task(sleep())
    task3 = asyncio.create_task(sleep())
    task4 = asyncio.create_task(sleep())
    task5 = asyncio.create_task(sleep())
    task6 = asyncio.create_task(sleep())
    task7 = asyncio.create_task(sleep())

    await task1
    await task2
    await task3
    await task4
    await task5
    await task6
    await task7


    # This needs to be awaiting an asyncio sleep. time.sleep will not work
    async def sleep():
    await asyncio.sleep(0.2)
    return "ok"


    def test_await():
    start = time.time()
    # Runs and "awaitable"
    # awaitable = async def that includes "await"
    asyncio.run(await_sleep())
    end = time.time()
    assert (end - start) < 1


    def test_await_concurrently():
    start = time.time()
    # Runs and "awaitable"
    # awaitable = async def that includes "await"
    asyncio.run(await_multiple())
    end = time.time()
    assert (end - start) < 1