1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50
|
from boltons.queueutils import SortedPriorityQueue, HeapPriorityQueue
def _test_priority_queue(queue_type):
pq = queue_type()
item1 = 'a'
item2 = 'b'
item3 = 'c'
pq.add(item1)
pq.remove(item1)
# integer priorities
pq.add(item1, 2)
pq.add(item2, 9)
pq.add(item3, 7)
assert len(pq) == 3
assert item2 == pq.pop()
assert len(pq) == 2
assert item3 == pq.pop()
assert len(pq) == 1
assert item1 == pq.pop()
assert len(pq) == 0
# float priorities
pq.add(item1, 0.2)
pq.add(item2, 0.9)
pq.add(item3, 0.7)
assert len(pq) == 3
assert item2 == pq.pop()
assert len(pq) == 2
assert item3 == pq.pop()
assert len(pq) == 1
assert item1 == pq.pop()
assert len(pq) == 0
try:
pq.pop()
except IndexError:
pass
else:
assert False, 'priority queue should be empty'
return
def test_heap_queue():
_test_priority_queue(HeapPriorityQueue)
def test_sorted_queue():
_test_priority_queue(SortedPriorityQueue)
|