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 check Asyncio and yield from interaction

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

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

@asyncio.coroutine
def greet_every_two_seconds() -> 'Generator[Any, None, None]':
    n = 0
    while n < 5:
        print('Prev', n)
        yield from asyncio.sleep(0.1)
        print('After', n)
        n += 1

loop = asyncio.get_event_loop()
try:
    loop.run_until_complete(greet_every_two_seconds())
finally:
    loop.close()
[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 Generator, Any
import asyncio
from asyncio import Future

@asyncio.coroutine
def compute(x: int, y: int) -> 'Generator[Any, None, int]':
    print("Compute %s + %s ..." % (x, y))
    yield from asyncio.sleep(0.1)
    return x + y   # Here the int is wrapped in Future[int]

@asyncio.coroutine
def print_sum(x: int, y: int) -> 'Generator[Any, None, None]':
    result = yield from compute(x, y)  # The type of result will be int (is extracted from Future[int]
    print("%s + %s = %s" % (x, y, result))

loop = asyncio.get_event_loop()
loop.run_until_complete(print_sum(1, 2))
loop.close()
[out]
Compute 1 + 2 ...
1 + 2 = 3

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

@asyncio.coroutine
def slow_operation(future: 'Future[str]') -> 'Generator[Any, None, None]':
    yield from asyncio.sleep(0.1)
    future.set_result('Future is done!')

loop = asyncio.get_event_loop()
future = asyncio.Future()  # type: Future[str]
asyncio.Task(slow_operation(future))
loop.run_until_complete(future)
print(future.result())
loop.close()
[out]
Future is done!

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

@asyncio.coroutine
def slow_operation(future: 'Future[str]') -> 'Generator[Any, None, None]':
    yield from asyncio.sleep(1)
    future.set_result('Callback works!')

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

loop = asyncio.get_event_loop() # type: AbstractEventLoop
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
try:
    loop.run_forever()
finally:
    loop.close()
[out]
Callback works!

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

loop = asyncio.get_event_loop()
tasks = [
    asyncio.Task(factorial("A", 2)),
    asyncio.Task(factorial("B", 3)),
    asyncio.Task(factorial("C", 4))]
loop.run_until_complete(asyncio.wait(tasks))
loop.close()
[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 Generator, Any
import asyncio
from asyncio import Future

@asyncio.coroutine
def h4() -> 'Generator[Any, None, int]':
    x = yield from future
    return x

@asyncio.coroutine
def h3() -> 'Generator[Any, None, int]':
    x = yield from h4()
    print("h3: %s" % x)
    return x

@asyncio.coroutine
def h2() -> 'Generator[Any, None, int]':
    x = yield from h3()
    print("h2: %s" % x)
    return x

@asyncio.coroutine
def h() -> 'Generator[Any, None, None]':
    x = yield from h2()
    print("h: %s" % x)

loop = asyncio.get_event_loop()
future = asyncio.Future()  # type: Future[int]
future.set_result(42)
loop.run_until_complete(h())
print("Outside %s" % future.result())
loop.close()
[out]
h3: 42
h2: 42
h: 42
Outside 42

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

@asyncio.coroutine
def h4() -> 'Generator[Any, None, Future[int]]':
    yield from asyncio.sleep(0.1)
    f = asyncio.Future() #type: Future[int]
    return f

@asyncio.coroutine
def h3() -> 'Generator[Any, None, Future[Future[int]]]':
    x = yield from h4()
    x.set_result(42)
    f = asyncio.Future() #type: Future[Future[int]]
    f.set_result(x)
    return f

@asyncio.coroutine
def h() -> 'Generator[Any, None, None]':
    print("Before")
    x = yield from h3()
    y = yield from x
    z = yield from 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))

loop = asyncio.get_event_loop()
loop.run_until_complete(h())
loop.close()
[out]
Before
42
Future<result=42>
Future<result=Future<result=42>>


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

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

@asyncio.coroutine
def h() -> 'Generator[Any, None, None]':
    x = yield from future
    print("h: %s" % x.x)

loop = asyncio.get_event_loop()
future = asyncio.Future()  # type: Future[A]
future.set_result(A(42))
loop.run_until_complete(h())
print("Outside %s" % future.result().x)
loop.close()
[out]
h: 42
Outside 42


-- Errors

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

@asyncio.coroutine
def greet() -> 'Generator[Any, None, None]':
    yield from asyncio.sleep(0.2)
    print('Hello World')

@asyncio.coroutine
def test() -> 'Generator[Any, None, None]':
    yield from greet()
    x = yield from greet()  # Error

loop = asyncio.get_event_loop()
try:
    loop.run_until_complete(test())
finally:
    loop.close()
[out]
_program.py:13: error: Function does not return a value

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

@asyncio.coroutine
def compute(x: int, y: int) -> 'Generator[Any, None, int]':
    print("Compute %s + %s ..." % (x, y))
    yield from asyncio.sleep(0.1)
    return str(x + y)   # Error

@asyncio.coroutine
def print_sum(x: int, y: int) -> 'Generator[Any, None, None]':
    result = yield from compute(x, y)
    print("%s + %s = %s" % (x, y, result))

loop = asyncio.get_event_loop()
loop.run_until_complete(print_sum(1, 2))
loop.close()

[out]
_program.py:9: error: Incompatible return value type (got "str", expected "int")

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

@asyncio.coroutine
def slow_operation(future: 'Future[str]') -> 'Generator[Any, None, None]':
    yield from asyncio.sleep(1)
    future.set_result(42)  # Error

loop = asyncio.get_event_loop()
future = asyncio.Future()  # type: Future[str]
asyncio.Task(slow_operation(future))
loop.run_until_complete(future)
print(future.result())
loop.close()
[out]
_program.py:8: error: Argument 1 to "set_result" of "Future" has incompatible type "int"; expected "str"


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

@asyncio.coroutine
def slow_operation(future: 'Future[int]') -> 'Generator[Any, None, None]':
    yield from asyncio.sleep(1)
    future.set_result(42)

loop = asyncio.get_event_loop()
future = asyncio.Future()  # type: Future[str]
asyncio.Task(slow_operation(future))  # Error
loop.run_until_complete(future)
print(future.result())
loop.close()
[out]
_program.py:12: error: Argument 1 to "slow_operation" has incompatible type "Future[str]"; expected "Future[int]"

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

@asyncio.coroutine
def slow_operation(future: 'Future[int]') -> 'Generator[Any, None, None]':
    yield from asyncio.sleep(1)
    future.set_result('42')  #Try to set an str as result to a Future[int]

loop = asyncio.get_event_loop()
future = asyncio.Future()  # type: Future[str]
asyncio.Task(slow_operation(future))  # Error
loop.run_until_complete(future)
print(future.result())
loop.close()
[out]
_program.py:8: error: Argument 1 to "set_result" of "Future" has incompatible type "str"; expected "int"
_program.py:12: error: Argument 1 to "slow_operation" has incompatible type "Future[str]"; expected "Future[int]"

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

@asyncio.coroutine
def slow_operation(future: 'Future[str]') -> 'Generator[Any, None, None]':
    yield from asyncio.sleep(1)
    future.set_result('Future is done!')

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

loop = asyncio.get_event_loop() # type: AbstractEventLoop
future = asyncio.Future()  # type: Future[str]
asyncio.Task(slow_operation(future))
future.add_done_callback(got_result)  # Error

try:
    loop.run_forever()
finally:
    loop.close()
[out]
_program.py:18: error: Argument 1 to "add_done_callback" of "Future" has incompatible type "Callable[[Future[int]], None]"; expected "Callable[[Future[str]], Any]"

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

@asyncio.coroutine
def h4() -> 'Generator[Any, None, Future[int]]':
    yield from asyncio.sleep(1)
    f = asyncio.Future() #type: Future[int]
    return f

@asyncio.coroutine
def h3() -> 'Generator[Any, None, Future[Future[Future[int]]]]':
    x = yield from h4()
    x.set_result(42)
    f = asyncio.Future() #type: Future[Future[int]]
    f.set_result(x)
    return f

@asyncio.coroutine
def h() -> 'Generator[Any, None, None]':
    print("Before")
    x = yield from h3()
    y = yield from x
    z = yield from y
    print(z)
    print(y)
    print(x)

loop = asyncio.get_event_loop()
loop.run_until_complete(h())
loop.close()
[out]
_program.py:18: error: Incompatible return value type (got "Future[Future[int]]", expected "Future[Future[Future[int]]]")

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

@asyncio.coroutine
def h4() -> 'Generator[Any, None, Future[int]]':
    yield from asyncio.sleep(1)
    f = asyncio.Future() #type: Future[int]
    return f

@asyncio.coroutine
def h3() -> 'Generator[Any, None, Future[int]]':
    x = yield from h4()
    x.set_result(42)
    f = asyncio.Future() #type: Future[Future[int]]
    f.set_result(x)
    return f

@asyncio.coroutine
def h() -> 'Generator[Any, None, None]':
    print("Before")
    x = yield from h3()
    y = yield from x
    print(y)
    print(x)

loop = asyncio.get_event_loop()
loop.run_until_complete(h())
loop.close()
[out]
_program.py:18: error: Incompatible return value type (got "Future[Future[int]]", expected "Future[int]")

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

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

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

@asyncio.coroutine
def h() -> 'Generator[Any, None, None]':
    x = yield from future # type: B # Error
    print("h: %s" % x.x)

loop = asyncio.get_event_loop()
future = asyncio.Future()  # type: Future[A]
future.set_result(A(42))
loop.run_until_complete(h())
loop.close()
[out]
_program.py:16: 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)

@asyncio.coroutine
def bad(arg: P) -> T:
    pass
[out]
_program.py:8: note: Revealed type is "def [T] (arg: P?) -> T`-1"
_program.py:12: error: Variable "_testForwardRefToBadAsyncShouldNotCrash_newsemanal.P" is not valid as a type
_program.py:12: note: See https://mypy.readthedocs.io/en/stable/common_issues.html#variables-vs-type-aliases