test: add test cases to validate Queue behavior

Author: imnotjames

tests/asyncio/asyncio_queue.py

@@ -0,0 +1,85 @@
+# SPDX-FileCopyrightText: 2019 Damien P. George
+#
+# SPDX-License-Identifier: MIT
+#
+# MicroPython uasyncio module
+# MIT license; Copyright (c) 2019 Damien P. George
+#
+# pylint: skip-file
+
+import asyncio
+
+def print_queue_info(queue: asyncio.Queue):
+    print("Size:", queue.qsize())
+    print("Empty:", queue.empty())
+    print("Full:", queue.full())
+
+async def put_queue(queue: asyncio.Queue, value):
+    await queue.put(value)
+
+async def get_queue(queue: asyncio.Queue):
+    print(await queue.get())
+
+async def main():
+    # Can put and retrieve items in the right order
+    print("=" * 10)
+    queue = asyncio.Queue()
+    await queue.put("a")
+    await queue.put("b")
+
+    print(await queue.get())
+    print(await queue.get())
+
+    # Waits for item to be put if get is called on empty queue
+    print("=" * 10)
+    queue = asyncio.Queue()
+
+    task = asyncio.create_task(get_queue(queue))
+    await asyncio.sleep(0.01)
+    print("putting on the queue")
+    await queue.put("example")
+    await task
+
+    # Waits for item to be taken off the queue if max size is specified
+    print("=" * 10)
+    queue = asyncio.Queue(1)
+    await queue.put("hello world")
+    task = asyncio.create_task(put_queue(queue, "example"))
+    await asyncio.sleep(0.01)
+    print_queue_info(queue)
+    print(await queue.get())
+    await task
+    print_queue_info(queue)
+    print(await queue.get())
+
+    # Raises an error if not waiting
+    print("=" * 10)
+    queue = asyncio.Queue(1)
+    try:
+        queue.get_nowait()
+    except Exception as e:
+        print(repr(e))
+
+    await queue.put("hello world")
+
+    try:
+        queue.put_nowait("example")
+    except Exception as e:
+        print(repr(e))
+
+    # Sets the size, empty, and full values as expected when no max size is set
+    print("=" * 10)
+    queue = asyncio.Queue()
+    print_queue_info(queue)
+    await queue.put("example")
+    print_queue_info(queue)
+
+    # Sets the size, empty, and full values as expected when max size is set
+    print("=" * 10)
+    queue = asyncio.Queue(1)
+    print_queue_info(queue)
+    await queue.put("example")
+    print_queue_info(queue)
+
+
+asyncio.run(main())

tests/asyncio/asyncio_queue.py.exp

@@ -0,0 +1,35 @@
+# SPDX-FileCopyrightText: 2019 Damien P. George
+#
+# SPDX-License-Identifier: MIT
+==========
+a
+b
+==========
+putting on the queue
+example
+==========
+Size: 1
+Empty: False
+Full: True
+hello world
+Size: 1
+Empty: False
+Full: True
+example
+==========
+QueueEmpty()
+QueueFull()
+==========
+Size: 0
+Empty: True
+Full: False
+Size: 1
+Empty: False
+Full: False
+==========
+Size: 0
+Empty: True
+Full: False
+Size: 1
+Empty: False
+Full: True