[tests] adding tests for UniqPriorityQueue and UniqFifoQueue

Author: mutax

networking_nsxv3/tests/unit/realization/test_coordination.py

@@ -18,7 +18,6 @@ def setUp(self):
         set_logging_levels()
         logging.setup(cfg.CONF, "demo")
 
-
     def test_rerunner_simple(self):
         from networking_nsxv3.common.synchronization import JobRerunner, Runnable
 
@@ -43,7 +42,6 @@ def nop(item):
         # no job was added twice, so no job should be re-executed
         self.assertEqual(rerunner.pop(), None)
 
-
     def test_rerunner_rerun(self):
         from networking_nsxv3.common.synchronization import JobRerunner, Runnable
 
@@ -84,14 +82,73 @@ def nop(item):
         # no job should have the same hash
         self.assertEqual(len(set(alljobs)), len(alljobs))
 
-
-
-
-
-
-
-
-
-
-
-
+    def test_fifoqueue(self):
+        from networking_nsxv3.common.synchronization import UniqFiFoQueue
+        from eventlet.queue import Empty
+        queue = UniqFiFoQueue()
+
+        for i in range(10):
+            queue.put(i)
+            queue.put(i)
+        for i in range(20, 10, -1):
+            queue.put(i)
+
+        try:
+            ret = []
+            for i in range(200):
+                item = queue.get(block=False)
+                ret.append(item)
+        except Empty:
+            pass
+
+        expected = [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 20, 19, 18, 17, 16, 15, 14, 13, 12, 11]
+        self.assertEqual(expected, ret)
+
+    def test_prioqueue(self):
+        from networking_nsxv3.common.synchronization import UniqPriorityQueue
+        from eventlet.queue import Empty
+        queue = UniqPriorityQueue()
+
+        class PrioItem:
+
+            def __init__(self, uid, prio):
+                self.priority = prio
+                self.uid = uid
+
+            def __lt__(self, other):
+                if self.priority < other.priority:
+                    return -1
+                elif self.priority > other.priority:
+                    return 1
+                return 0
+
+            def __eq__(self, other):
+                return self.uid == other.uid
+
+            def __repr__(self):
+                return f"PrioItem({self.uid},{self.priority})"
+
+        for i in range(10):
+            queue.put(PrioItem(i, 5))
+            queue.put(PrioItem(i, 1))
+        for i in range(15, 5, -1):
+            queue.put(PrioItem(i, 4))
+
+        try:
+            ret = []
+            for i in range(200):
+                item = queue.get(block=False)
+                ret.append((item.uid, item.priority))
+        except Empty:
+            pass
+
+        # the first 6 items should have the lowest prio,
+        # then 10 with the highest, prio 5 should never be seen
+        self.assertEqual([4] * 6, [prio for uid, prio in ret[:6]])
+        self.assertEqual([1] * 10, [prio for uid, prio in ret[6:]])
+
+        # each item should be present only once.
+        # because sorting is not stable, we need to sort the uids
+        # to compare
+        ret = sorted([uid for uid, prio in ret])
+        self.assertEqual(list(range(16)), ret)