diff --git a/tests/..7 b/tests/..7 new file mode 100644 index 0000000..ac8cb60 --- /dev/null +++ b/tests/..7 @@ -0,0 +1 @@ +/tt diff --git a/tests/ttl.py b/tests/ttl.py new file mode 100644 index 0000000..e6d2b67 --- /dev/null +++ b/tests/ttl.py @@ -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") diff --git a/ttlcollections/structures.py b/ttlcollections/structures.py index 13da91e..263bccf 100644 --- a/ttlcollections/structures.py +++ b/ttlcollections/structures.py @@ -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)