[test][bcp] Add max iteration count as watchdog system

Without that some tests might stale indefinitely.

beat/core/test/test_bcp.py

@@ -128,6 +128,12 @@ CONFIGURATION2 = {
 # ----------------------------------------------------------
 
 
+DEFAULT_MAX_ITERATION_COUNT = 30
+
+
+# ----------------------------------------------------------
+
+
 class ZMQBrokerProcess(multiprocessing.Process):
     def __init__(self, port, verbose, callbacks=None):
         super(ZMQBrokerProcess, self).__init__()
@@ -166,6 +172,11 @@ class ZMQWorkerProcess(multiprocessing.Process):
 
 
 class ExcecutionTestCase(unittest.TestCase):
+    def setUp(self):
+        self.MAX_ITERATION_COUNT = int(
+            os.environ.get("BPC_MAX_ITERATION_COUNT", DEFAULT_MAX_ITERATION_COUNT)
+        )
+
     def prepare_databases(self, configuration):
         for _, input_cfg in configuration["inputs"].items():
             database = Database(prefix, input_cfg["database"])
@@ -224,6 +235,8 @@ class TestBCP(ExcecutionTestCase):
     docker_images_cache = None
 
     def setUp(self):
+        super().setUp()
+
         self.worker_name = b"test_worker"
 
         port = find_free_port()
@@ -255,12 +268,16 @@ class TestBCP(ExcecutionTestCase):
         self.client.send(self.worker_name, request)
 
         reply = None
-        while reply is None:
+        iterations = 0
+        while reply is None and iterations < self.MAX_ITERATION_COUNT:
             try:
                 reply = self.client.recv()
             except KeyboardInterrupt:
                 break
+            else:
+                iterations += 1
 
+        self.assertTrue(iterations < self.MAX_ITERATION_COUNT)
         self.assertEqual(reply[1], BCP.BCPP_ERROR)
         self.assertEqual(reply[2], b"Unknown job: 1")
 
@@ -275,7 +292,8 @@ class TestBCP(ExcecutionTestCase):
         self.client.send(self.worker_name, request)
 
         messages = []
-        while len(messages) < 3:
+        iterations = 0
+        while len(messages) < 3 and iterations < self.MAX_ITERATION_COUNT:
             try:
                 reply = self.client.recv()
             except KeyboardInterrupt:
@@ -283,7 +301,9 @@ class TestBCP(ExcecutionTestCase):
             else:
                 if reply:
                     messages.append(reply)
+                iterations += 1
 
+        self.assertTrue(iterations < self.MAX_ITERATION_COUNT)
         self.assertEqual(messages[0][1], BCP.BCPP_JOB_RECEIVED)
         self.assertEqual(messages[0][2], job_id)
         self.assertEqual(messages[1][1], BCP.BCPP_JOB_STARTED)
@@ -299,7 +319,8 @@ class TestBCP(ExcecutionTestCase):
         self.client.send(self.worker_name, request)
 
         messages = []
-        while len(messages) < 3:
+        iterations = 0
+        while len(messages) < 3 and iterations < self.MAX_ITERATION_COUNT:
             try:
                 reply = self.client.recv()
             except KeyboardInterrupt:
@@ -307,7 +328,9 @@ class TestBCP(ExcecutionTestCase):
             else:
                 if reply:
                     messages.append(reply)
+                iterations += 1
 
+        self.assertTrue(iterations < self.MAX_ITERATION_COUNT)
         self.assertEqual(messages[0][1], BCP.BCPP_JOB_RECEIVED)
         self.assertEqual(messages[0][2], job_id)
         self.assertEqual(messages[1][1], BCP.BCPP_JOB_STARTED)
@@ -350,10 +373,12 @@ class TestExcecutionProcess(ExcecutionTestCase):
         poller.register(socket, zmq.POLLIN)
 
         process = self.setup_process()
-        done = False
 
+        done = False
+        iterations = 0
         messages = []
-        while True:
+
+        while True and iterations < self.MAX_ITERATION_COUNT:
             try:
                 items = poller.poll(1000)
             except KeyboardInterrupt:
@@ -369,11 +394,13 @@ class TestExcecutionProcess(ExcecutionTestCase):
                     break
                 elif result in [BCP.BCPP_JOB_ERROR, BCP.BCPP_ERROR]:
                     break
+            iterations += 1
 
         process.terminate()
         process.join()
         ctx.destroy()
 
+        self.assertTrue(iterations < self.MAX_ITERATION_COUNT)
         self.assertTrue(done)
         self.assertEqual(process.queue.get(), "started")
         self.assertEqual(messages[0][1], BCP.BCPP_JOB_DONE)
@@ -385,9 +412,11 @@ class TestExcecutionProcess(ExcecutionTestCase):
         process = self.setup_process()
 
         done = False
+        iterations = 0
 
         messages = []
-        while True:
+
+        while True and iterations < self.MAX_ITERATION_COUNT:
             try:
                 items = poller.poll(1000)
             except KeyboardInterrupt:
@@ -403,10 +432,12 @@ class TestExcecutionProcess(ExcecutionTestCase):
                     break
                 elif result in [BCP.BCPP_JOB_ERROR, BCP.BCPP_ERROR]:
                     break
+            iterations += 1
 
         process.terminate()
         process.join()
 
+        self.assertTrue(iterations < self.MAX_ITERATION_COUNT)
         self.assertTrue(done)
         self.assertEqual(process.queue.get(), "started")
         self.assertEqual(messages[0][1], BCP.BCPP_JOB_DONE)