3.3.2 test regression on Python 3.11: _python_utils_tests/test_generators.py::test_abatcher - TypeError: Passing coroutines is forbidden, use tasks explicitly.

[mgorny]

With Python 3.11.0b1:

Test log

========================================================= test session starts =========================================================
platform linux -- Python 3.11.0b1, pytest-7.1.2, pluggy-1.0.0
rootdir: /tmp/python-utils, configfile: pytest.ini
plugins: mypy-0.9.1, asyncio-0.18.3, cov-3.0.0
asyncio: mode=Mode.STRICT
collected 25 items                                                                                                                    

_python_utils_tests/test_decorators.py ..                                                                                       [  8%]
_python_utils_tests/test_generators.py FFF.                                                                                     [ 24%]
_python_utils_tests/test_import.py ......                                                                                       [ 48%]
_python_utils_tests/test_logger.py .                                                                                            [ 52%]
_python_utils_tests/test_python_utils.py .                                                                                      [ 56%]
_python_utils_tests/test_time.py ...........                                                                                    [100%]

============================================================== FAILURES ===============================================================
____________________________________________________________ test_abatcher ____________________________________________________________

    @pytest.mark.asyncio
    async def test_abatcher():
>       async for batch in python_utils.abatcher(python_utils.acount(stop=9), 3):

_python_utils_tests/test_generators.py:10: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
python_utils/generators.py:35: in abatcher
    done, pending = await asyncio.wait(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

fs = {<async_generator_asend object at 0x7febd2f789c0>}

    async def wait(fs, *, timeout=None, return_when=ALL_COMPLETED):
        """Wait for the Futures or Tasks given by fs to complete.
    
        The fs iterable must not be empty.
    
        Coroutines will be wrapped in Tasks.
    
        Returns two sets of Future: (done, pending).
    
        Usage:
    
            done, pending = await asyncio.wait(fs)
    
        Note: This does not raise TimeoutError! Futures that aren't done
        when the timeout occurs are returned in the second set.
        """
        if futures.isfuture(fs) or coroutines.iscoroutine(fs):
            raise TypeError(f"expect a list of futures, not {type(fs).__name__}")
        if not fs:
            raise ValueError('Set of Tasks/Futures is empty.')
        if return_when not in (FIRST_COMPLETED, FIRST_EXCEPTION, ALL_COMPLETED):
            raise ValueError(f'Invalid return_when value: {return_when}')
    
        fs = set(fs)
    
        if any(coroutines.iscoroutine(f) for f in fs):
>           raise TypeError("Passing coroutines is forbidden, use tasks explicitly.")
E           TypeError: Passing coroutines is forbidden, use tasks explicitly.

/usr/lib/python3.11/asyncio/tasks.py:424: TypeError
_________________________________________________________ test_abatcher_timed _________________________________________________________

    @pytest.mark.asyncio
    async def test_abatcher_timed():
        batches = []
>       async for batch in python_utils.abatcher(
            python_utils.acount(stop=10, delay=0.08), interval=0.1
        ):

_python_utils_tests/test_generators.py:20: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
python_utils/generators.py:35: in abatcher
    done, pending = await asyncio.wait(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

fs = {<async_generator_asend object at 0x7febd2cd5200>}

    async def wait(fs, *, timeout=None, return_when=ALL_COMPLETED):
        """Wait for the Futures or Tasks given by fs to complete.
    
        The fs iterable must not be empty.
    
        Coroutines will be wrapped in Tasks.
    
        Returns two sets of Future: (done, pending).
    
        Usage:
    
            done, pending = await asyncio.wait(fs)
    
        Note: This does not raise TimeoutError! Futures that aren't done
        when the timeout occurs are returned in the second set.
        """
        if futures.isfuture(fs) or coroutines.iscoroutine(fs):
            raise TypeError(f"expect a list of futures, not {type(fs).__name__}")
        if not fs:
            raise ValueError('Set of Tasks/Futures is empty.')
        if return_when not in (FIRST_COMPLETED, FIRST_EXCEPTION, ALL_COMPLETED):
            raise ValueError(f'Invalid return_when value: {return_when}')
    
        fs = set(fs)
    
        if any(coroutines.iscoroutine(f) for f in fs):
>           raise TypeError("Passing coroutines is forbidden, use tasks explicitly.")
E           TypeError: Passing coroutines is forbidden, use tasks explicitly.

/usr/lib/python3.11/asyncio/tasks.py:424: TypeError
__________________________________________________ test_abatcher_timed_with_timeout ___________________________________________________

    @pytest.mark.asyncio
    async def test_abatcher_timed_with_timeout():
        async def generator():
            # Test if the timeout is respected
            yield 0
            yield 1
            await asyncio.sleep(0.11)
    
            # Test if the timeout is respected
            yield 2
            yield 3
            await asyncio.sleep(0.11)
    
            # Test if exceptions are handled correctly
            await asyncio.wait_for(asyncio.sleep(1), timeout=0.05)
    
            # Test if StopAsyncIteration is handled correctly
            yield 4
    
        batcher = python_utils.abatcher(generator(), interval=0.1)
>       assert await batcher.__anext__() == [0, 1]

_python_utils_tests/test_generators.py:49: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
python_utils/generators.py:35: in abatcher
    done, pending = await asyncio.wait(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

fs = {<async_generator_asend object at 0x7febd2f6e240>}

    async def wait(fs, *, timeout=None, return_when=ALL_COMPLETED):
        """Wait for the Futures or Tasks given by fs to complete.
    
        The fs iterable must not be empty.
    
        Coroutines will be wrapped in Tasks.
    
        Returns two sets of Future: (done, pending).
    
        Usage:
    
            done, pending = await asyncio.wait(fs)
    
        Note: This does not raise TimeoutError! Futures that aren't done
        when the timeout occurs are returned in the second set.
        """
        if futures.isfuture(fs) or coroutines.iscoroutine(fs):
            raise TypeError(f"expect a list of futures, not {type(fs).__name__}")
        if not fs:
            raise ValueError('Set of Tasks/Futures is empty.')
        if return_when not in (FIRST_COMPLETED, FIRST_EXCEPTION, ALL_COMPLETED):
            raise ValueError(f'Invalid return_when value: {return_when}')
    
        fs = set(fs)
    
        if any(coroutines.iscoroutine(f) for f in fs):
>           raise TypeError("Passing coroutines is forbidden, use tasks explicitly.")
E           TypeError: Passing coroutines is forbidden, use tasks explicitly.

/usr/lib/python3.11/asyncio/tasks.py:424: TypeError
======================================================= short test summary info =======================================================
FAILED _python_utils_tests/test_generators.py::test_abatcher - TypeError: Passing coroutines is forbidden, use tasks explicitly.
FAILED _python_utils_tests/test_generators.py::test_abatcher_timed - TypeError: Passing coroutines is forbidden, use tasks explicitly.
FAILED _python_utils_tests/test_generators.py::test_abatcher_timed_with_timeout - TypeError: Passing coroutines is forbidden, use ta...
==================================================== 3 failed, 22 passed in 2.83s =====================================================

A quick bisect blames the following commit:

commit c9d4cd8b091c6aafa7e35eb38746d860dfcb9af1
Author: Rick van Hattem <Wolph@wol.ph>
Date:   Tue May 31 00:10:23 2022 +0200

    Fixed bug with batcher skipping items in the case of timeouts thanks to @jorenham

 _python_utils_tests/test_generators.py | 38 +++++++++++++++++++++++++++++++---
 python_utils/generators.py             | 32 +++++++++++++++++-----------
 2 files changed, 55 insertions(+), 15 deletions(-)

CC @jorenham

[wolph]

That's indeed my mistake. I skipped the create_task here because mypy was getting angry about it. In all fairness, mypy is right, the Python type hints appear to be wrong here...

python_utils/generators.py:36: error: Argument 1 to "create_task" has incompatible type "Awaitable[Any]"; expected "Union[Generator[Any, None, <nothing>], Coroutine[Any, Any, <nothing>]]"
Found 1 error in 1 file (checked 1 source file)

I'm simply ignoring the error for now.

[mgorny]

Thanks!