Skip to content

Commit

Permalink
test: add test cases to validate Queue behavior
Browse files Browse the repository at this point in the history
  • Loading branch information
imnotjames committed Nov 14, 2023
1 parent f94cfb3 commit ea4f3d1
Show file tree
Hide file tree
Showing 2 changed files with 124 additions and 0 deletions.
89 changes: 89 additions & 0 deletions tests/asyncio/asyncio_queue.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,89 @@
# 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())
35 changes: 35 additions & 0 deletions tests/asyncio/asyncio_queue.py.exp
Original file line number Diff line number Diff line change
@@ -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

0 comments on commit ea4f3d1

Please sign in to comment.