twisteroidambassador / async_stagger Goto Github PK

Traceback (most recent call last):
  File "prettysocks.py", line 34, in <module>
    from typing import Callable, Awaitable, Tuple, Union
  File "/tmp/prettysocks/typing.py", line 3, in <module>
    from typing import Tuple, Union
ImportError: cannot import name 'Tuple' from 'typing' (/tmp/prettysocks/typing.py)

============================= test session starts ==============================
platform linux -- Python 3.9.0rc2, pytest-5.4.3, py-1.9.0, pluggy-0.13.1
rootdir: /build/async_stagger-0.3.0
plugins: asyncio-0.14.0, mock-3.2.0
collected 40 items

async_stagger/test_aitertools.py .......                                 [ 17%]
async_stagger/test_happy_eyeballs.py ...........                         [ 45%]
async_stagger/test_resolver.py ..................                        [ 90%]
async_stagger/test_stagger.py F.F.                                       [100%]

=================================== FAILURES ===================================
__________________________ test_stagger_random_tasks ___________________________

    @pytest.mark.asyncio
    async def test_stagger_random_tasks():
        for _ in range(10):
>           await random_tasks()

async_stagger/test_stagger.py:20:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

    async def random_tasks():
        async def one_coro(index, sleep_for, succeed):
            print('Coroutine %d starting, will sleep for %f' % (index, sleep_for))
            await asyncio.sleep(sleep_for)
            if succeed:
                print('Coroutine %d finishing' % index)
                return index
            else:
                print('Coroutine %d raising RuntimeError' % index)
                raise RuntimeError

        coro_fns = aiter_from_iter(
            partial(one_coro, i, random.random() * 2, random.random() < 0.8)
            for i in range(10))
        delay = 0.3
        winner_result, winner_idx, exc, aiter_exc = \
            await staggered_race(coro_fns, delay)
        if winner_idx is not None:
            assert winner_result == winner_idx
            for i, e in enumerate(exc):
                if i == winner_idx:
                    assert e is None
                else:
>                   assert isinstance(e, (RuntimeError, asyncio.CancelledError))
E                   AssertionError: assert False
E                    +  where False = isinstance(None, (<class 'RuntimeError'>, <class 'asyncio.exceptions.CancelledError'>))

async_stagger/test_stagger.py:46: AssertionError
----------------------------- Captured stdout call -----------------------------
Coroutine 0 starting, will sleep for 0.506632
Coroutine 1 starting, will sleep for 1.748210
Coroutine 0 finishing
________________________ test_stagger_simultaneous_done ________________________

    @pytest.mark.asyncio
    async def test_stagger_simultaneous_done():
        for _ in range(5):
>           await simultaneous_done()

async_stagger/test_stagger.py:78:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

    async def simultaneous_done():
        coro_fns = aiter_from_iter(
            partial(asyncio.sleep, i*0.2) for i in range(5, 0, -1))
        winner_result, winner_idx, exceptions, aiter_exc = \
            await staggered_race(coro_fns, 0.2)
        assert winner_idx is not None
>       assert all(isinstance(e, asyncio.CancelledError)
                   for i, e in enumerate(exceptions)
                   if i != winner_idx)
E       assert False
E        +  where False = all(<generator object simultaneous_done.<locals>.<genexpr> at 0x7ffff61e5190>)

async_stagger/test_stagger.py:87: AssertionError
=========================== short test summary info ============================
FAILED async_stagger/test_stagger.py::test_stagger_random_tasks - AssertionEr...
FAILED async_stagger/test_stagger.py::test_stagger_simultaneous_done - assert...
======================== 2 failed, 38 passed in 27.47s =========================