Package logilab :: Package common :: Package test :: Module unittest_taskqueue
[frames] | no frames]

Source Code for Module logilab.common.test.unittest_taskqueue

 1  from logilab.common.testlib import TestCase, unittest_main 
 2   
 3  from logilab.common.tasksqueue import * 
 4   
5 -class TaskTC(TestCase):
6
7 - def test_eq(self):
8 self.failIf(Task('t1') == Task('t2')) 9 self.failUnless(Task('t1') == Task('t1'))
10
11 - def test_cmp(self):
12 self.failUnless(Task('t1', LOW) < Task('t2', MEDIUM)) 13 self.failIf(Task('t1', LOW) > Task('t2', MEDIUM)) 14 self.failUnless(Task('t1', HIGH) > Task('t2', MEDIUM)) 15 self.failIf(Task('t1', HIGH) < Task('t2', MEDIUM))
16 17
18 -class PrioritizedTasksQueueTC(TestCase):
19
20 - def test_priority(self):
21 queue = PrioritizedTasksQueue() 22 queue.put(Task('t1')) 23 queue.put(Task('t2', MEDIUM)) 24 queue.put(Task('t3', HIGH)) 25 queue.put(Task('t4', LOW)) 26 self.assertEquals(queue.get().id, 't3') 27 self.assertEquals(queue.get().id, 't2') 28 self.assertEquals(queue.get().id, 't1') 29 self.assertEquals(queue.get().id, 't4')
30
31 - def test_remove_equivalent(self):
32 queue = PrioritizedTasksQueue() 33 queue.put(Task('t1')) 34 queue.put(Task('t2', MEDIUM)) 35 queue.put(Task('t1', HIGH)) 36 queue.put(Task('t3', MEDIUM)) 37 queue.put(Task('t2', MEDIUM)) 38 self.assertEquals(queue.qsize(), 3) 39 self.assertEquals(queue.get().id, 't1') 40 self.assertEquals(queue.get().id, 't2') 41 self.assertEquals(queue.get().id, 't3') 42 self.assertEquals(queue.qsize(), 0)
43
44 - def test_remove(self):
45 queue = PrioritizedTasksQueue() 46 queue.put(Task('t1')) 47 queue.put(Task('t2')) 48 queue.put(Task('t3')) 49 queue.remove('t2') 50 self.assertEquals([t.id for t in queue], ['t3', 't1']) 51 self.assertRaises(ValueError, queue.remove, 't4')
52 53 if __name__ == '__main__': 54 unittest_main() 55