Added various functionality to the containers and a small test suite

This commit is contained in:
nocturn9x 2020-07-06 08:57:16 +00:00
parent d3bb3a2ca0
commit bf3c2ee6b8
3 changed files with 123 additions and 2 deletions

1
tests/..7 Normal file
View File

@ -0,0 +1 @@
/tt

88
tests/ttl.py Normal file
View File

@ -0,0 +1,88 @@
import time
from ttlcollections import TTLQueue, TTLHeap, TTLStack
from termcolor import cprint
def test_queue(queue: TTLQueue) -> bool:
"""Tests the TTLQueue class"""
for x in range(queue.qsize - 1):
queue.put(x)
time.sleep(queue.ttl)
1 in queue
if queue:
cprint("Error: Queue not empty after sleep", "red")
return False
queue.put(1, ttl=5)
time.sleep(5)
1 in queue # This triggers the expire function
if queue:
cprint("Error: Queue not empty after custom TTL sleep", "red")
return False
cprint("Queue test passed", "green")
return True
def test_heap(heap: TTLHeap) -> bool:
"""Tests the TTLHeap class"""
for x in range(heap.qsize - 1):
heap.put(x)
time.sleep(heap.ttl)
1 in heap
if heap:
cprint("Error: Heap not empty after sleep", "red")
return False
heap.put(1, ttl=5)
time.sleep(5)
1 in heap # This triggers the expire function
if heap:
cprint("Error: Heap not empty after custom TTL sleep", "red")
return False
cprint("Heap test passed", "green")
return True
def test_stack(stack: TTLStack) -> bool:
"""Tests the TTLStack class"""
for x in range(stack.size - 1):
stack.push(x)
time.sleep(stack.ttl)
1 in stack
if stack:
cprint("Error: Stack not empty after sleep", "red")
return False
stack.push(1, ttl=5)
time.sleep(5)
1 in stack # This triggers the expire function
if stack:
cprint("Error: Stack not empty after custom TTL sleep", "red")
return False
cprint("Stack test passed", "green")
return True
def main():
"""Test entry point"""
start = time.monotonic()
results = {}
results["stack"] = test_stack(TTLStack(10, 10))
results["queue"] = test_queue(TTLQueue(10, 10))
results["heap"] = test_heap(TTLHeap(10, 10))
end = time.monotonic()
values = results.values()
took = round(end - start, 2)
if not all(values):
failed = list(values).count(False)
cprint(f"{failed} tests failed, {len(values) - failed} passed in {took} seconds", "red")
else:
cprint(f"0 tests failed, {len(values)} passed in {took} seconds", "green")
if __name__ == "__main__":
try:
main()
except KeyboardInterrupt:
cprint("Operation cancelled by user", "red")

View File

@ -113,7 +113,18 @@ class TTLQueue:
"""Implements item in self"""
self.expire(self.timer())
return self._queue.__contains__(TTLitem(item, None))
return self._queue.__contains__(TTLItem(item, None))
def __bool__(self):
"""Implement bool(self)"""
return bool(self._queue)
def __eq__(self, other: object):
"""Implements self == other"""
queue = [item.obj for item in self._queue]
return queue == other
class TTLStack:
@ -220,6 +231,17 @@ class TTLStack:
self.expire(self.timer())
return self._stack.__contains__(TTLItem(item, None))
def __bool__(self):
"""Implement bool(self)"""
return bool(self._stack)
def __eq__(self, other: object):
"""Implements self == other"""
stack = [item.obj for item in self._stack]
return stack == other
class TTLHeap(TTLQueue):
"""A heap queue with per-item time to live (TTL)
@ -269,7 +291,7 @@ class TTLHeap(TTLQueue):
def __contains__(self, item):
"""Implements item in self"""
super().__contains__(item)
super().__contains__(item)
def put(self, element, ttl: int = 0):
"""Puts an item onto the queue
@ -313,3 +335,13 @@ class TTLHeap(TTLQueue):
"""
super().expire(when)
def __bool__(self):
"""Implement bool(self)"""
return super().__bool__()
def __eq__(self, other: object):
"""Implements self == other"""
return super().__eq__(other)