Change stream_* methods to sync

They just start a background task.  There's no async until we're
actually waiting for messages.

Signed-off-by: Sahas Subramanian <[email protected]>

Author: shsms

src/frequenz/client/electricity_trading/_client.py

@@ -218,7 +218,7 @@ def stub(self) -> electricity_trading_pb2_grpc.ElectricityTradingServiceAsyncStu
         # type-checker, so it can only be used for type hints.
         return self._stub  # type: ignore
 
-    async def stream_gridpool_orders(
+    def stream_gridpool_orders(
         # pylint: disable=too-many-arguments, too-many-positional-arguments
         self,
         gridpool_id: int,
@@ -285,7 +285,7 @@ async def stream_gridpool_orders(
             warn_on_overflow=warn_on_overflow, maxsize=max_size
         )
 
-    async def stream_gridpool_trades(
+    def stream_gridpool_trades(
         # pylint: disable=too-many-arguments, too-many-positional-arguments
         self,
         gridpool_id: int,
@@ -351,7 +351,7 @@ async def stream_gridpool_trades(
             warn_on_overflow=warn_on_overflow, maxsize=max_size
         )
 
-    async def stream_public_trades(
+    def stream_public_trades(
         # pylint: disable=too-many-arguments, too-many-positional-arguments
         self,
         states: list[TradeState] | None = None,

src/frequenz/client/electricity_trading/cli/etrading.py

@@ -73,7 +73,7 @@ async def list_public_trades(url: str, key: str, *, delivery_start: datetime) ->
         if delivery_start <= datetime.now(timezone.utc):
             return
 
-    stream = await client.stream_public_trades(delivery_period=delivery_period)
+    stream = client.stream_public_trades(delivery_period=delivery_period)
     async for trade in stream:
         print_public_trade(trade)
 
@@ -111,7 +111,7 @@ async def list_gridpool_trades(
     if delivery_start and delivery_start <= datetime.now(timezone.utc):
         return
 
-    stream = await client.stream_gridpool_trades(gid, delivery_period=delivery_period)
+    stream = client.stream_gridpool_trades(gid, delivery_period=delivery_period)
     async for trade in stream:
         print_trade(trade)
 
@@ -154,7 +154,7 @@ async def list_gridpool_orders(
     if delivery_start and delivery_start <= datetime.now(timezone.utc):
         return
 
-    stream = await client.stream_gridpool_orders(gid, delivery_period=delivery_period)
+    stream = client.stream_gridpool_orders(gid, delivery_period=delivery_period)
     async for order in stream:
         print_order(order)
 

tests/test_client.py

@@ -131,27 +131,23 @@ def set_up_order_detail_response(
     ).to_pb()
 
 
-def test_stream_gridpool_orders(set_up: SetupParams) -> None:
+async def test_stream_gridpool_orders(set_up: SetupParams) -> None:
     """Test the method streaming gridpool orders."""
-    set_up.loop.run_until_complete(
-        set_up.client.stream_gridpool_orders(set_up.gridpool_id)
-    )
+    set_up.client.stream_gridpool_orders(set_up.gridpool_id)
+    await asyncio.sleep(0)
 
     set_up.mock_stub.ReceiveGridpoolOrdersStream.assert_called_once()
     args, _ = set_up.mock_stub.ReceiveGridpoolOrdersStream.call_args
     assert args[0].gridpool_id == set_up.gridpool_id
 
 
-def test_stream_gridpool_orders_with_optional_inputs(set_up: SetupParams) -> None:
+async def test_stream_gridpool_orders_with_optional_inputs(set_up: SetupParams) -> None:
     """Test the method streaming gridpool orders with some fields to filter for."""
     # Fields to filter for
     order_states = [OrderState.ACTIVE]
 
-    set_up.loop.run_until_complete(
-        set_up.client.stream_gridpool_orders(
-            set_up.gridpool_id, order_states=order_states
-        )
-    )
+    set_up.client.stream_gridpool_orders(set_up.gridpool_id, order_states=order_states)
+    await asyncio.sleep(0)
 
     set_up.mock_stub.ReceiveGridpoolOrdersStream.assert_called_once()
     args, _ = set_up.mock_stub.ReceiveGridpoolOrdersStream.call_args
@@ -161,32 +157,30 @@ def test_stream_gridpool_orders_with_optional_inputs(set_up: SetupParams) -> Non
     ]
 
 
-def test_stream_gridpool_trades(
+async def test_stream_gridpool_trades(
     set_up: SetupParams,
 ) -> None:
     """Test the method streaming gridpool trades."""
-    set_up.loop.run_until_complete(
-        set_up.client.stream_gridpool_trades(
-            gridpool_id=set_up.gridpool_id, market_side=set_up.side
-        )
+    set_up.client.stream_gridpool_trades(
+        gridpool_id=set_up.gridpool_id, market_side=set_up.side
     )
+    await asyncio.sleep(0)
 
     set_up.mock_stub.ReceiveGridpoolTradesStream.assert_called_once()
     args, _ = set_up.mock_stub.ReceiveGridpoolTradesStream.call_args
     assert args[0].gridpool_id == set_up.gridpool_id
     assert args[0].filter.side == set_up.side.to_pb()
 
 
-def test_stream_public_trades(
+async def test_stream_public_trades(
     set_up: SetupParams,
 ) -> None:
     """Test the method streaming public trades."""
     # Fields to filter for
     trade_states = [TradeState.ACTIVE]
 
-    set_up.loop.run_until_complete(
-        set_up.client.stream_public_trades(states=trade_states)
-    )
+    set_up.client.stream_public_trades(states=trade_states)
+    await asyncio.sleep(0)
 
     set_up.mock_stub.ReceivePublicTradesStream.assert_called_once()
     args, _ = set_up.mock_stub.ReceivePublicTradesStream.call_args