[tests] adding tests for JobRerunner

Author: mutax

networking_nsxv3/tests/unit/realization/test_coordination.py

@@ -0,0 +1,97 @@
+from oslo_log import log as logging
+from oslo_config import cfg
+from neutron.tests import base
+
+LOG: logging.KeywordArgumentAdapter = logging.getLogger(__name__)
+
+
+def set_logging_levels():
+    cfg.CONF.set_override("default_log_levels", [
+        'networking_nsxv3.common.synchronization=WARNING',
+    ])
+
+
+class TestAgentRealizerScheduling(base.BaseTestCase):
+    def setUp(self):
+        super().setUp()
+
+        set_logging_levels()
+        logging.setup(cfg.CONF, "demo")
+
+
+    def test_rerunner_simple(self):
+        from networking_nsxv3.common.synchronization import JobRerunner, Runnable
+
+        rerunner = JobRerunner()
+
+        def nop(item):
+            LOG.debug("nop(%s) called", item)
+
+        # add some jobs
+        for some_id in range(100):
+            job = Runnable(str(some_id), nop)
+            ret = rerunner.add_job(job)
+            self.assertTrue(ret)
+
+        # mark them all done -- note that because
+        # if and fn-name are the same, this should match,
+        # although creating new jobs
+        for some_id in range(100):
+            job = Runnable(str(some_id), nop)
+            rerunner.job_done(job)
+
+        # 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
+
+        rerunner = JobRerunner()
+
+        def nop(item):
+            # should never happen
+            LOG.error("nop(%s) called", item)
+
+        # add some jobs
+        for some_id in range(100):
+            job = Runnable(str(some_id), nop)
+            ret = rerunner.add_job(job)
+            self.assertTrue(ret)
+
+        # add them again, twice
+        for some_id in range(100):
+            job = Runnable(str(some_id), nop)
+            ret = rerunner.add_job(job)
+            self.assertFalse(ret)
+            ret = rerunner.add_job(job)
+            self.assertFalse(ret)
+
+        # mark them all done, then expect them
+        # all to be returned once in pop,
+        # although added twice again above
+        for some_id in range(100):
+            job = Runnable(str(some_id), nop)
+            rerunner.job_done(job)
+
+        alljobs = []
+        while job := rerunner.pop():
+            alljobs.append(job)
+
+        # each job should be returned once
+        self.assertEqual(len(alljobs), 100)
+
+        # no job should have the same hash
+        self.assertEqual(len(set(alljobs)), len(alljobs))
+
+
+
+
+
+
+
+
+
+
+
+