diff options
author | Sylvain Thenault <sylvain.thenault@logilab.fr> | 2008-12-01 11:36:51 +0100 |
---|---|---|
committer | Sylvain Thenault <sylvain.thenault@logilab.fr> | 2008-12-01 11:36:51 +0100 |
commit | 33239a3d2d6c35ec56a506afb47991afcf65e54a (patch) | |
tree | ec87751ba9f22df21f71e222c190b3c24b8b16bd /test/unittest_taskqueue.py | |
parent | 65c478e1645131acc92cf160df7712146ce043f4 (diff) | |
download | logilab-common-33239a3d2d6c35ec56a506afb47991afcf65e54a.tar.gz |
new module containing a class to handle prioritized tasks queue
Diffstat (limited to 'test/unittest_taskqueue.py')
-rw-r--r-- | test/unittest_taskqueue.py | 54 |
1 files changed, 54 insertions, 0 deletions
diff --git a/test/unittest_taskqueue.py b/test/unittest_taskqueue.py new file mode 100644 index 0000000..a723d1c --- /dev/null +++ b/test/unittest_taskqueue.py @@ -0,0 +1,54 @@ +from logilab.common.testlib import TestCase, unittest_main + +from logilab.common.tasksqueue import * + +class TaskTC(TestCase): + + def test_eq(self): + self.failIf(Task('t1') == Task('t2')) + self.failUnless(Task('t1') == Task('t1')) + + def test_cmp(self): + self.failUnless(Task('t1', LOW) < Task('t2', MEDIUM)) + self.failIf(Task('t1', LOW) > Task('t2', MEDIUM)) + self.failUnless(Task('t1', HIGH) > Task('t2', MEDIUM)) + self.failIf(Task('t1', HIGH) < Task('t2', MEDIUM)) + + +class PrioritizedTasksQueueTC(TestCase): + + def test_priority(self): + queue = PrioritizedTasksQueue() + queue.put(Task('t1')) + queue.put(Task('t2', MEDIUM)) + queue.put(Task('t3', HIGH)) + queue.put(Task('t4', LOW)) + self.assertEquals(queue.get().id, 't3') + self.assertEquals(queue.get().id, 't2') + self.assertEquals(queue.get().id, 't1') + self.assertEquals(queue.get().id, 't4') + + def test_remove_equivalent(self): + queue = PrioritizedTasksQueue() + queue.put(Task('t1')) + queue.put(Task('t2', MEDIUM)) + queue.put(Task('t1', HIGH)) + queue.put(Task('t3', MEDIUM)) + queue.put(Task('t2', MEDIUM)) + self.assertEquals(queue.qsize(), 3) + self.assertEquals(queue.get().id, 't1') + self.assertEquals(queue.get().id, 't2') + self.assertEquals(queue.get().id, 't3') + self.assertEquals(queue.qsize(), 0) + + def test_remove(self): + queue = PrioritizedTasksQueue() + queue.put(Task('t1')) + queue.put(Task('t2')) + queue.put(Task('t3')) + queue.remove('t2') + self.assertEquals([t.id for t in queue], ['t3', 't1']) + self.assertRaises(ValueError, queue.remove, 't4') + +if __name__ == '__main__': + unittest_main() |