67 lines
1.8 KiB
Python
67 lines
1.8 KiB
Python
import sys
|
|
import os
|
|
import threading
|
|
from pathlib import Path
|
|
sys.path.insert(0, str(Path(__file__).parent.parent / 'src'))
|
|
from models.episode import Episode
|
|
from distill.queue import PersistenceQueue
|
|
|
|
def test_queue_basic():
|
|
q = PersistenceQueue(':memory:')
|
|
ep = Episode.create(content='test1')
|
|
assert q.enqueue(ep) == True
|
|
assert q.size() == 1
|
|
assert q.is_empty() == False
|
|
dequeued = q.dequeue()
|
|
assert dequeued is not None
|
|
assert dequeued['content'] == 'test1'
|
|
assert q.size() == 0
|
|
assert q.is_empty() == True
|
|
q.close()
|
|
|
|
def test_queue_requeue():
|
|
q = PersistenceQueue(':memory:')
|
|
ep = Episode.create(content='test2')
|
|
q.enqueue(ep)
|
|
deq = q.dequeue()
|
|
d = Episode.from_dict(deq)
|
|
assert q.requeue(d) == True
|
|
assert q.size() == 1
|
|
q.close()
|
|
|
|
def test_queue_peek():
|
|
q = PersistenceQueue(':memory:')
|
|
ep1 = Episode.create(content='first')
|
|
ep2 = Episode.create(content='second')
|
|
q.enqueue(ep1)
|
|
q.enqueue(ep2)
|
|
peeked = q.peek()
|
|
assert peeked['content'] == 'first'
|
|
assert q.size() == 2 # peek does not remove
|
|
q.close()
|
|
|
|
def test_queue_concurrent():
|
|
q = PersistenceQueue(':memory:')
|
|
errors = []
|
|
def enqueue_many(n):
|
|
try:
|
|
for i in range(n):
|
|
ep = Episode.create(content=f'thread-{i}')
|
|
q.enqueue(ep)
|
|
except Exception as e:
|
|
errors.append(str(e))
|
|
threads = [threading.Thread(target=enqueue_many, args=(20,)) for _ in range(5)]
|
|
for t in threads:
|
|
t.start()
|
|
for t in threads:
|
|
t.join()
|
|
assert len(errors) == 0, errors
|
|
assert q.size() == 100
|
|
q.close()
|
|
|
|
if __name__ == '__main__':
|
|
test_queue_basic()
|
|
test_queue_requeue()
|
|
test_queue_peek()
|
|
test_queue_concurrent()
|
|
print('All queue tests passed!') |