def all_tasks()

in later/unittest/case.py [0:0]

• 14 lines of code
• 7 McCabe index (conditional complexity)

def all_tasks(loop):
    # This is a copy of asyncio.all_task but returns even done tasks
    # so we can see if they were awaited instead of ignored

    # This is copied from the guts of asyncio.all_tasks
    # Looping the WeakSet since it is possible it fails during iteration
    _all_tasks = asyncio.tasks._all_tasks
    _get_loop = asyncio.futures._get_loop
    i = 0
    while True:
        try:
            tasks = list(_all_tasks)
        except RuntimeError:  # pragma: nocover
            i += 1
            if i >= 1000:
                raise
            else:
                break
        return {t for t in tasks if _get_loop(t) is loop}