test-data/unit/pythoneval-asyncio.test

-- Test cases for type checking mypy programs using full stubs and running
-- using CPython.
--
-- These are mostly regression tests -- no attempt is made to make these
-- complete.
--
-- This test file checks Asyncio and await interaction

[case testImportAsyncio]
import asyncio
print('Imported')
[out]
Imported

[case testSimpleCoroutineSleep]
from typing import Any, Generator
import asyncio
from asyncio import Future

async def greet_every_two_seconds() -> None:
    n = 0
    while n < 5:
        print('Prev', n)
        await asyncio.sleep(0.01)
        print('After', n)
        n += 1

asyncio.run(greet_every_two_seconds())
[out]
Prev 0
After 0
Prev 1
After 1
Prev 2
After 2
Prev 3
After 3
Prev 4
After 4

[case testCoroutineCallingOtherCoroutine]
from typing import Any
import asyncio
from asyncio import Future

async def compute(x: int, y: int) -> int:
    print("Compute %s + %s ..." % (x, y))
    await asyncio.sleep(0.01)
    return x + y   # Here the int is wrapped in Future[int]

async def print_sum(x: int, y: int) -> None:
    result = await compute(x, y)  # The type of result will be int (is extracted from Future[int]
    print("%s + %s = %s" % (x, y, result))

asyncio.run(print_sum(1, 2))
[out]
Compute 1 + 2 ...
1 + 2 = 3

[case testCoroutineChangingFuture]
from typing import Any
import asyncio
from asyncio import Future

async def slow_operation(future: 'Future[str]') -> None:
    await asyncio.sleep(0.01)
    future.set_result('Future is done!')

async def main() -> None:
    future = asyncio.Future()  # type: Future[str]
    asyncio.Task(slow_operation(future))
    await future
    print(future.result())

asyncio.run(main())
[out]
Future is done!

[case testFunctionAssignedAsCallback]
import typing
from typing import Any
import asyncio
from asyncio import Future, AbstractEventLoop

async def slow_operation(future: 'Future[str]') -> None:
    await asyncio.sleep(1)
    future.set_result('Callback works!')

def got_result(future: 'Future[str]') -> None:
    print(future.result())
    loop.stop()

async def main() -> None:
    future = asyncio.Future()  # type: Future[str]
    asyncio.Task(slow_operation(future))  # Here create a task with the function. (The Task need a Future[T] as first argument)
    future.add_done_callback(got_result)  # and assign the callback to the future

loop = asyncio.new_event_loop()  # type: AbstractEventLoop
loop.run_until_complete(main())
try:
    loop.run_forever()
finally:
    loop.close()
[out]
Callback works!

[case testMultipleTasks]
import typing
from typing import Any
import asyncio
from asyncio import Task, Future
async def factorial(name, number) -> None:
    f = 1
    for i in range(2, number+1):
        print("Task %s: Compute factorial(%s)..." % (name, i))
        await asyncio.sleep(0.01)
        f *= i
    print("Task %s: factorial(%s) = %s" % (name, number, f))

async def main() -> None:
    tasks = [
        asyncio.Task(factorial("A", 2)),
        asyncio.Task(factorial("B", 3)),
        asyncio.Task(factorial("C", 4))]
    await asyncio.wait(tasks)

asyncio.run(main())
[out]
Task A: Compute factorial(2)...
Task B: Compute factorial(2)...
Task C: Compute factorial(2)...
Task A: factorial(2) = 2
Task B: Compute factorial(3)...
Task C: Compute factorial(3)...
Task B: factorial(3) = 6
Task C: Compute factorial(4)...
Task C: factorial(4) = 24


[case testConcatenatedCoroutines]
import typing
from typing import Any
import asyncio
from asyncio import Future

future: Future[int]

async def h4() -> int:
    x = await future
    return x

async def h3() -> int:
    x = await h4()
    print("h3: %s" % x)
    return x

async def h2() -> int:
    x = await h3()
    print("h2: %s" % x)
    return x

async def h() -> None:
    x = await h2()
    print("h: %s" % x)

async def main() -> None:
    global future
    future = asyncio.Future()
    future.set_result(42)
    await h()
    print("Outside %s" % future.result())

asyncio.run(main())
[out]
h3: 42
h2: 42
h: 42
Outside 42

[case testConcatenatedCoroutinesReturningFutures]
import typing
from typing import Any
import asyncio
from asyncio import Future

async def h4() -> "Future[int]":
    await asyncio.sleep(0.01)
    f = asyncio.Future()  # type: Future[int]
    return f

async def h3() -> "Future[Future[int]]":
    x = await h4()
    x.set_result(42)
    f = asyncio.Future()  # type: Future[Future[int]]
    f.set_result(x)
    return f

async def h() -> None:
    print("Before")
    x = await h3()
    y = await x
    z = await y
    print(z)
    def normalize(future):
        # The str conversion seems inconsistent; not sure exactly why. Normalize
        # the result.
        return str(future).replace('<Future finished ', 'Future<')
    print(normalize(y))
    print(normalize(x))

asyncio.run(h())
[out]
Before
42
Future<result=42>
Future<result=Future<result=42>>


[case testCoroutineWithOwnClass]
import typing
from typing import Any
import asyncio
from asyncio import Future

future: Future["A"]

class A:
    def __init__(self, x: int) -> None:
        self.x = x

async def h() -> None:
    x = await future
    print("h: %s" % x.x)

async def main() -> None:
    global future
    future = asyncio.Future()
    future.set_result(A(42))
    await h()
    print("Outside %s" % future.result().x)

asyncio.run(main())
[out]
h: 42
Outside 42


-- Errors

[case testErrorAssigningCoroutineThatDontReturn]
from typing import Any
import asyncio
from asyncio import Future

async def greet() -> None:
    await asyncio.sleep(0.2)
    print('Hello World')

async def test() -> None:
    await greet()
    x = await greet()  # Error

asyncio.run(test())
[out]
_program.py:11: error: Function does not return a value (it only ever returns None)

[case testErrorReturnIsNotTheSameType]
from typing import Any
import asyncio
from asyncio import Future

async def compute(x: int, y: int) -> int:
    print("Compute %s + %s ..." % (x, y))
    await asyncio.sleep(0.01)
    return str(x + y)   # Error

async def print_sum(x: int, y: int) -> None:
    result = await compute(x, y)
    print("%s + %s = %s" % (x, y, result))

asyncio.run(print_sum(1, 2))
[out]
_program.py:8: error: Incompatible return value type (got "str", expected "int")

[case testErrorSetFutureDifferentInternalType]
from typing import Any
import asyncio
from asyncio import Future

async def slow_operation(future: 'Future[str]') -> None:
    await asyncio.sleep(1)
    future.set_result(42)  # Error

async def main() -> None:
    future = asyncio.Future()  # type: Future[str]
    asyncio.Task(slow_operation(future))
    await future
    print(future.result())

asyncio.run(main())
[out]
_program.py:7: error: Argument 1 to "set_result" of "Future" has incompatible type "int"; expected "str"


[case testErrorUsingDifferentFutureType]
from typing import Any
import asyncio
from asyncio import Future

async def slow_operation(future: 'Future[int]') -> None:
    await asyncio.sleep(1)
    future.set_result(42)

async def main() -> None:
    future = asyncio.Future()  # type: Future[str]
    asyncio.Task(slow_operation(future))  # Error
    await future
    print(future.result())

asyncio.run(main())
[out]
_program.py:11: error: Argument 1 to "slow_operation" has incompatible type "Future[str]"; expected "Future[int]"

[case testErrorUsingDifferentFutureTypeAndSetFutureDifferentInternalType]
from typing import Any
import asyncio
from asyncio import Future

async def slow_operation(future: 'Future[int]') -> None:
    await asyncio.sleep(1)
    future.set_result('42')  # Try to set an str as result to a Future[int]

async def main() -> None:
    future = asyncio.Future()  # type: Future[str]
    asyncio.Task(slow_operation(future))  # Error
    await future
    print(future.result())

asyncio.run(main())
[out]
_program.py:7: error: Argument 1 to "set_result" of "Future" has incompatible type "str"; expected "int"
_program.py:11: error: Argument 1 to "slow_operation" has incompatible type "Future[str]"; expected "Future[int]"

[case testErrorSettingCallbackWithDifferentFutureType]
import typing
from typing import Any
import asyncio
from asyncio import Future, AbstractEventLoop

async def slow_operation(future: 'Future[str]') -> None:
    await asyncio.sleep(1)
    future.set_result('Future is done!')

def got_result(future: 'Future[int]') -> None:
    print(future.result())
    loop.stop()

async def main() -> None:
    future = asyncio.Future()  # type: Future[str]
    asyncio.Task(slow_operation(future))
    future.add_done_callback(got_result)  # Error

loop = asyncio.new_event_loop()
loop.run_until_complete(main())
try:
    loop.run_forever()
finally:
    loop.close()
[out]
_program.py:17: error: Argument 1 to "add_done_callback" of "Future" has incompatible type "Callable[[Future[int]], None]"; expected "Callable[[Future[str]], object]"

[case testErrorOneMoreFutureInReturnType]
import typing
from typing import Any, Generator
import asyncio
from asyncio import Future

async def h4() -> Future[int]:
    await asyncio.sleep(1)
    f = asyncio.Future()  # type: Future[int]
    return f

async def h3() -> Future[Future[Future[int]]]:
    x = await h4()
    x.set_result(42)
    f = asyncio.Future()  # type: Future[Future[int]]
    f.set_result(x)
    return f

async def h() -> None:
    print("Before")
    x = await h3()
    y = await x
    z = await y
    print(z)
    print(y)
    print(x)

asyncio.run(h())
[out]
_program.py:16: error: Incompatible return value type (got "Future[Future[int]]", expected "Future[Future[Future[int]]]")

[case testErrorOneLessFutureInReturnType]
import typing
from typing import Any
import asyncio
from asyncio import Future

async def h4() -> Future[int]:
    await asyncio.sleep(1)
    f = asyncio.Future()  # type: Future[int]
    return f

async def h3() -> Future[int]:
    x = await h4()
    x.set_result(42)
    f = asyncio.Future()  # type: Future[Future[int]]
    f.set_result(x)
    return f

async def h() -> None:
    print("Before")
    x = await h3()
    y = await x
    print(y)
    print(x)

asyncio.run(h())
[out]
_program.py:16: error: Incompatible return value type (got "Future[Future[int]]", expected "Future[int]")
_program.py:16: note: Maybe you forgot to use "await"?

[case testErrorAssignmentDifferentType]
import typing
from typing import Any
import asyncio
from asyncio import Future

future: Future["A"]

class A:
    def __init__(self, x: int) -> None:
        self.x = x

class B:
    def __init__(self, x: int) -> None:
        self.x = x

async def h() -> None:
    x = await future  # type: B  # Error
    print("h: %s" % x.x)

async def main() -> None:
    global future
    future = asyncio.Future()
    future.set_result(A(42))
    await h()

asyncio.run(main())
[out]
_program.py:17: error: Incompatible types in assignment (expression has type "A", variable has type "B")

[case testForwardRefToBadAsyncShouldNotCrash_newsemanal]
from typing import TypeVar
import asyncio

T = TypeVar('T')
P = whatever  # type: ignore

def test() -> None:
    reveal_type(bad)
    bad(0)

async def bad(arg: P) -> T:
    pass
[out]
_program.py:8: note: Revealed type is "def [T] (arg: P?) -> typing.Coroutine[Any, Any, T]"
_program.py:9: error: Value of type "Coroutine[Any, Any, Never]" must be used
_program.py:9: note: Are you missing an await?
_program.py:11: error: Variable "_testForwardRefToBadAsyncShouldNotCrash_newsemanal.P" is not valid as a type
_program.py:11: note: See https://mypy.readthedocs.io/en/stable/common_issues.html#variables-vs-type-aliases