diff options
| author | Damien George <damien@micropython.org> | 2020-11-30 17:38:30 +1100 |
|---|---|---|
| committer | Damien George <damien@micropython.org> | 2020-12-02 12:07:06 +1100 |
| commit | 309dfe39e07d3c3c8ba873ed09116cea9f6f52c0 (patch) | |
| tree | b9546223768ef0c32ee54ff890ba4be96097f66c /tests/extmod/uasyncio_task_done.py | |
| parent | ca40eb0fdadd5a963b9a065999b092f029a598d5 (diff) | |
extmod/uasyncio: Add Task.done() method.
This is added because task.coro==None is no longer the way to detect if a
task is finished. Providing a (CPython compatible) function for this
allows the implementation to be abstracted away.
Signed-off-by: Damien George <damien@micropython.org>
Diffstat (limited to 'tests/extmod/uasyncio_task_done.py')
| -rw-r--r-- | tests/extmod/uasyncio_task_done.py | 66 |
1 files changed, 66 insertions, 0 deletions
diff --git a/tests/extmod/uasyncio_task_done.py b/tests/extmod/uasyncio_task_done.py new file mode 100644 index 000000000..2700da8c3 --- /dev/null +++ b/tests/extmod/uasyncio_task_done.py @@ -0,0 +1,66 @@ +# Test the Task.done() method + +try: + import uasyncio as asyncio +except ImportError: + try: + import asyncio + except ImportError: + print("SKIP") + raise SystemExit + + +async def task(t, exc=None): + print("task start") + if t >= 0: + await asyncio.sleep(t) + if exc: + raise exc + print("task done") + + +async def main(): + # Task that finishes immediately. + print("=" * 10) + t = asyncio.create_task(task(-1)) + print(t.done()) + await asyncio.sleep(0) + print(t.done()) + await t + print(t.done()) + + # Task that starts, runs and finishes. + print("=" * 10) + t = asyncio.create_task(task(0.01)) + print(t.done()) + await asyncio.sleep(0) + print(t.done()) + await t + print(t.done()) + + # Task that raises immediately. + print("=" * 10) + t = asyncio.create_task(task(-1, ValueError)) + print(t.done()) + await asyncio.sleep(0) + print(t.done()) + try: + await t + except ValueError as er: + print(repr(er)) + print(t.done()) + + # Task that raises after a delay. + print("=" * 10) + t = asyncio.create_task(task(0.01, ValueError)) + print(t.done()) + await asyncio.sleep(0) + print(t.done()) + try: + await t + except ValueError as er: + print(repr(er)) + print(t.done()) + + +asyncio.run(main()) |
