Might have found the issue

.github/workflows/ci-build.yml

@@ -36,4 +36,4 @@ jobs:
 
       - name: Run unit tests
         run: |
-          pytest -s -vvv tests/rsa/test_etcd3_discovery.py
+          pytest -s -vvv tests/rsa/test_etcd3_discovery.py::EtcdDiscoveryListenerTest

tests-infra/run_all.sh

@@ -0,0 +1,3 @@
+#!/bin/bash
+
+find -name "compose.yaml" -execdir docker-compose -f {} up -d \;

tests/rsa/test_etcd3_discovery.py

@@ -63,7 +63,7 @@
 # ------------------------------------------------------------------------------
 
 
-def start_framework_for_advertise(state_queue: Queue, debug_queue: Queue, start:float):
+def start_framework_for_advertise(state_queue: Queue, order_queue: Queue, debug_queue: Queue, start: float):
     """
     Starts a Pelix framework to advertise (via etcd) a helloimpl_xmlrpc
     remote service instance.  The tests can/will then discover this
@@ -129,14 +129,9 @@ def start_framework_for_advertise(state_queue: Queue, debug_queue: Queue, start:
         try:
             # Send that we are now ready
             state_queue.put("ready")
-            # Loop until ready processed
+            # Loop until we receive the order to quit
             while True:
-                if state_queue.empty():
-                    break
-            # Loop until we receive done message
-            while True:
-                state = state_queue.get(timeout=60)
-                if state is None:
+                if order_queue.get(timeout=60) == "quit":
                     break
         finally:
             # stop the framework gracefully
@@ -154,6 +149,7 @@ def setUp(self):
         TestEndpointEventListener
         """
         self.status_queue = Queue()
+        self.order_queue = Queue()
         self.debug_queue = Queue()
 
         # start a local framework
@@ -188,7 +184,8 @@ def prepareFrameworkProcess(self):
         """
         # start external framework that publishes remote service
         self.publisher_process = WrappedProcess(
-            target=start_framework_for_advertise, args=[self.status_queue, self.debug_queue, time.time()]
+            target=start_framework_for_advertise,
+            args=[self.status_queue, self.order_queue, self.debug_queue, time.time()],
         )
         self.publisher_process.start()
         state = self.status_queue.get(timeout=20)
@@ -209,6 +206,7 @@ def tearDown(self):
             print("END OF DEBUG")
 
             self.status_queue.close()
+            self.order_queue.close()
             self.debug_queue.close()
             self.publisher = None
 
@@ -261,7 +259,7 @@ def test_handler_2(endpoint_event, matched_filter):
             try:
                 if endpoint_event.get_type() == EndpointEvent.ADDED:
                     # send shutdown to trigger the removal
-                    self.status_queue.put(None)
+                    self.order_queue.put("quit")
                 elif endpoint_event.get_type() == EndpointEvent.REMOVED:
                     # do tests
                     self.assertTrue(matched_filter, ENDPOINT_LISTENER_SCOPE)