from __future__ import annotations from abc import abstractmethod from typing import Any from unittest import mock import pytest from queuelib.queue import ( BaseQueue, FifoDiskQueue, FifoMemoryQueue, FifoSQLiteQueue, LifoDiskQueue, LifoMemoryQueue, LifoSQLiteQueue, ) from queuelib.tests import QueuelibTestCase class DummyQueue: def __init__(self) -> None: self.q: list[Any] = [] def push(self, obj: Any) -> None: self.q.append(obj) def pop(self) -> Any | None: return self.q.pop() if self.q else None def peek(self) -> Any | None: return self.q[-1] if self.q else None def close(self) -> None: pass def __len__(self): return len(self.q) class InterfaceTest(QueuelibTestCase): def test_queue(self): queue = BaseQueue() with self.assertRaises(NotImplementedError): queue.push(b"") with self.assertRaises(NotImplementedError): queue.peek() with self.assertRaises(NotImplementedError): queue.pop() with self.assertRaises(NotImplementedError): len(queue) queue.close() def test_issubclass(self): assert not issubclass(list, BaseQueue) assert not issubclass(int, BaseQueue) assert not issubclass(QueuelibTestCase, BaseQueue) assert issubclass(DummyQueue, BaseQueue) assert issubclass(FifoMemoryQueue, BaseQueue) assert issubclass(LifoMemoryQueue, BaseQueue) assert issubclass(FifoDiskQueue, BaseQueue) assert issubclass(LifoDiskQueue, BaseQueue) assert issubclass(FifoSQLiteQueue, BaseQueue) assert issubclass(LifoSQLiteQueue, BaseQueue) def test_isinstance(self): assert not isinstance(1, BaseQueue) assert not isinstance([], BaseQueue) assert isinstance(DummyQueue(), BaseQueue) assert isinstance(FifoMemoryQueue(), BaseQueue) assert isinstance(LifoMemoryQueue(), BaseQueue) for cls in [FifoDiskQueue, LifoDiskQueue, FifoSQLiteQueue, LifoSQLiteQueue]: queue = cls(self.tempfilename()) assert isinstance(queue, BaseQueue) queue.close() class QueueTestMixin: @abstractmethod def queue(self) -> BaseQueue: raise NotImplementedError def test_empty(self): """Empty queue test""" q = self.queue() assert q.pop() is None q.close() def test_single_pushpop(self): q = self.queue() q.push(b"a") assert q.pop() == b"a" q.close() def test_binary_element(self): elem = ( b"\x80\x02}q\x01(U\x04bodyq\x02U\x00U\t_encodingq\x03U\x05utf-" b"8q\x04U\x07cookiesq\x05}q\x06U\x04metaq\x07}q\x08U\x07header" b"sq\t}U\x03urlq\nX\x15\x00\x00\x00file:///tmp/tmphDJYsgU\x0bd" b"ont_filterq\x0b\x89U\x08priorityq\x0cK\x00U\x08callbackq\rNU" b"\x06methodq\x0eU\x03GETq\x0fU\x07errbackq\x10Nu." ) q = self.queue() q.push(elem) assert q.pop() == elem q.close() def test_len(self): q = self.queue() self.assertEqual(len(q), 0) q.push(b"a") self.assertEqual(len(q), 1) q.push(b"b") q.push(b"c") self.assertEqual(len(q), 3) q.pop() q.pop() q.pop() self.assertEqual(len(q), 0) q.close() def test_peek_one_element(self): q = self.queue() self.assertIsNone(q.peek()) q.push(b"a") self.assertEqual(q.peek(), b"a") self.assertEqual(q.pop(), b"a") self.assertIsNone(q.peek()) q.close() class FifoTestMixin: def test_push_pop1(self): """Basic push/pop test""" q = self.queue() q.push(b"a") q.push(b"b") q.push(b"c") self.assertEqual(q.pop(), b"a") self.assertEqual(q.pop(), b"b") self.assertEqual(q.pop(), b"c") self.assertEqual(q.pop(), None) q.close() def test_push_pop2(self): """Test interleaved push and pops""" q = self.queue() q.push(b"a") q.push(b"b") q.push(b"c") q.push(b"d") self.assertEqual(q.pop(), b"a") self.assertEqual(q.pop(), b"b") q.push(b"e") self.assertEqual(q.pop(), b"c") self.assertEqual(q.pop(), b"d") self.assertEqual(q.pop(), b"e") q.close() def test_peek_fifo(self): q = self.queue() self.assertIsNone(q.peek()) q.push(b"a") q.push(b"b") q.push(b"c") self.assertEqual(q.peek(), b"a") self.assertEqual(q.peek(), b"a") self.assertEqual(q.pop(), b"a") self.assertEqual(q.peek(), b"b") self.assertEqual(q.peek(), b"b") self.assertEqual(q.pop(), b"b") self.assertEqual(q.peek(), b"c") self.assertEqual(q.peek(), b"c") self.assertEqual(q.pop(), b"c") self.assertIsNone(q.peek()) q.close() class LifoTestMixin: def test_push_pop1(self): """Basic push/pop test""" q = self.queue() q.push(b"a") q.push(b"b") q.push(b"c") self.assertEqual(q.pop(), b"c") self.assertEqual(q.pop(), b"b") self.assertEqual(q.pop(), b"a") self.assertEqual(q.pop(), None) q.close() def test_push_pop2(self): """Test interleaved push and pops""" q = self.queue() q.push(b"a") q.push(b"b") q.push(b"c") q.push(b"d") self.assertEqual(q.pop(), b"d") self.assertEqual(q.pop(), b"c") q.push(b"e") self.assertEqual(q.pop(), b"e") self.assertEqual(q.pop(), b"b") self.assertEqual(q.pop(), b"a") q.close() def test_peek_lifo(self): q = self.queue() self.assertIsNone(q.peek()) q.push(b"a") q.push(b"b") q.push(b"c") self.assertEqual(q.peek(), b"c") self.assertEqual(q.peek(), b"c") self.assertEqual(q.pop(), b"c") self.assertEqual(q.peek(), b"b") self.assertEqual(q.peek(), b"b") self.assertEqual(q.pop(), b"b") self.assertEqual(q.peek(), b"a") self.assertEqual(q.peek(), b"a") self.assertEqual(q.pop(), b"a") self.assertIsNone(q.peek()) q.close() class PersistentTestMixin: chunksize = 100000 @pytest.mark.xfail( reason="Reenable once Scrapy.squeues stops extending from this testsuite" ) def test_non_bytes_raises_typeerror(self): q = self.queue() self.assertRaises(TypeError, q.push, 0) self.assertRaises(TypeError, q.push, "") self.assertRaises(TypeError, q.push, None) self.assertRaises(TypeError, q.push, lambda x: x) q.close() def test_text_in_windows(self): e1 = b"\r\n" q = self.queue() q.push(e1) q.close() q = self.queue() e2 = q.pop() self.assertEqual(e1, e2) q.close() def test_close_open(self): """Test closing and re-opening keeps state""" q = self.queue() q.push(b"a") q.push(b"b") q.push(b"c") q.push(b"d") q.pop() q.pop() q.close() del q q = self.queue() self.assertEqual(len(q), 2) q.push(b"e") q.pop() q.pop() q.close() del q q = self.queue() assert q.pop() is not None self.assertEqual(len(q), 0) q.close() def test_cleanup(self): """Test queue dir is removed if queue is empty""" q = self.queue() values = [b"0", b"1", b"2", b"3", b"4"] assert self.qpath.exists() for x in values: q.push(x) for _ in values: q.pop() q.close() assert not self.qpath.exists() class FifoMemoryQueueTest(FifoTestMixin, QueueTestMixin, QueuelibTestCase): def queue(self): return FifoMemoryQueue() class LifoMemoryQueueTest(LifoTestMixin, QueueTestMixin, QueuelibTestCase): def queue(self): return LifoMemoryQueue() class FifoDiskQueueTest( FifoTestMixin, PersistentTestMixin, QueueTestMixin, QueuelibTestCase ): def queue(self): return FifoDiskQueue(self.qpath, chunksize=self.chunksize) def test_not_szhdr(self): q = self.queue() q.push(b"something") with ( self.tempfilename().open("w+") as empty_file, mock.patch.object(q, "tailf", empty_file), ): assert q.peek() is None assert q.pop() is None q.close() def test_chunks(self): """Test chunks are created and removed""" values = [b"0", b"1", b"2", b"3", b"4"] q = self.queue() for x in values: q.push(x) chunks = list(self.qpath.glob("q*")) self.assertEqual(len(chunks), 5 // self.chunksize + 1) for _ in values: q.pop() chunks = list(self.qpath.glob("q*")) self.assertEqual(len(chunks), 1) q.close() class ChunkSize1FifoDiskQueueTest(FifoDiskQueueTest): chunksize = 1 class ChunkSize2FifoDiskQueueTest(FifoDiskQueueTest): chunksize = 2 class ChunkSize3FifoDiskQueueTest(FifoDiskQueueTest): chunksize = 3 class ChunkSize4FifoDiskQueueTest(FifoDiskQueueTest): chunksize = 4 class LifoDiskQueueTest( LifoTestMixin, PersistentTestMixin, QueueTestMixin, QueuelibTestCase ): def queue(self): return LifoDiskQueue(self.qpath) def test_file_size_shrinks(self): """Test size of queue file shrinks when popping items""" q = self.queue() q.push(b"a") q.push(b"b") q.close() size = self.qpath.stat().st_size q = self.queue() q.pop() q.close() assert self.qpath.stat().st_size, size class FifoSQLiteQueueTest( FifoTestMixin, PersistentTestMixin, QueueTestMixin, QueuelibTestCase ): def queue(self): return FifoSQLiteQueue(self.qpath) class LifoSQLiteQueueTest( LifoTestMixin, PersistentTestMixin, QueueTestMixin, QueuelibTestCase ): def queue(self): return LifoSQLiteQueue(self.qpath)