Wait for databroker availability in integration tests

Signed-off-by: Sebastian Schildt <[email protected]>

kuksa_databroker/integration_test/helper.py

@@ -46,6 +46,12 @@ class Databroker:
     Databroker wraps collector and broker APIs of the databroker.
     """
 
+    @classmethod
+    async def ConnectedDatabroker(cls, address: str):
+        self = Databroker(address)
+        await self.await_connectivity()
+        return self
+
     def __init__(self, address: str) -> None:
 
         self._address = address
@@ -87,6 +93,23 @@ async def __get_datapoints(self, datapoints: list):
             GetDatapointsRequest(datapoints=datapoints)
         )
         return response
+
+    async def await_connectivity(self):
+        # We need to "manually" wait in a loop, as wait_for_state_change can not wait for a
+        # specific target state
+        tries = 0
+        while tries < 10:
+            state = self._channel.get_state(try_to_connect=True)
+            logger.info("GRPC State is %s", state)
+            if state == grpc.ChannelConnectivity.READY:
+                break
+            logger.info("Try %i: Waiting for GRPC connection to become READY...", tries)
+            tries = tries + 1
+            try:
+                await asyncio.wait_for(self._channel.wait_for_state_change(state), timeout=2)
+            except asyncio.exceptions.TimeoutError:
+                # We need to catch this, and will wait again until tries are used up
+                pass
 
     async def get_metadata(self, names=[]):
         """Requests metadata from databroker, allows for optional list of names

kuksa_databroker/integration_test/test_databroker.py

@@ -31,14 +31,14 @@
 @pytest.fixture
 async def setup_helper() -> Databroker:
     logger.info("Using DATABROKER_ADDRESS={}".format(DATABROKER_ADDRESS))
-    helper = Databroker(DATABROKER_ADDRESS)
+    helper = await Databroker.ConnectedDatabroker(DATABROKER_ADDRESS)
     return helper
 
 
 @pytest.mark.asyncio
 async def test_databroker_connection() -> None:
     logger.info("Connecting to VehicleDataBroker {}".format(DATABROKER_ADDRESS))
-    helper = Databroker(DATABROKER_ADDRESS)
+    helper = await Databroker.ConnectedDatabroker(DATABROKER_ADDRESS)
     await helper.get_metadata()
     logger.info("Databroker._address =  {}".format(helper._address))
     await helper.close()