Fix data flow in test that uses formula and mock_resampler

Creating formulas is not async.
So the flow was:
create formula -> send data -> subscribe for mock_resampler.
It worked because mock_resampler sent latest (old) messages to the
subscribed channel and in all tests, formulas were created
on startup.

In new test (next commit) we have to subscribe for formula
during runtime. And we don't expect old messages on the
newly created receiver.

In this commit we add await after creating formulas,
so the flow is:
create formula -> subscribe for mock_resampler -> send data

And we remove old message when creating new mock_resampler
receiver, so the newly created receiver is empty (as in true
runtime).

Signed-off-by: Elzbieta Kotulska <[email protected]>

tests/microgrid/test_grid.py

@@ -3,6 +3,7 @@
 
 """Tests for the `Grid` module."""
 
+import asyncio
 from contextlib import AsyncExitStack
 
 import frequenz.client.microgrid as client
@@ -109,6 +110,8 @@ async def test_grid_power_1(mocker: MockerFixture) -> None:
         stack.push_async_callback(grid.stop)
 
         grid_power_recv = grid.power.new_receiver()
+        # Wait for mock resampler to create forward stream
+        await asyncio.sleep(0.01)
 
         grid_meter_recv = get_resampled_stream(
             grid._formula_pool._namespace,  # pylint: disable=protected-access
@@ -152,6 +155,8 @@ async def test_grid_power_2(mocker: MockerFixture) -> None:
         stack.push_async_callback(grid.stop)
 
         grid_power_recv = grid.power.new_receiver()
+        # Wait for mock resampler to create forward stream
+        await asyncio.sleep(0.01)
 
         component_receivers = [
             get_resampled_stream(
@@ -203,6 +208,8 @@ async def test_grid_power_3_phase_side_meter(mocker: MockerFixture) -> None:
         grid_power_per_phase_recv = (
             grid._power_per_phase.new_receiver()  # pylint: disable=protected-access
         )
+        # Wait for mock resampler to create forward stream
+        await asyncio.sleep(0.01)
 
         for count in range(10):
             watts_delta = 1 if count % 2 == 0 else -1
@@ -237,6 +244,8 @@ async def test_grid_power_3_phase_none_values(mocker: MockerFixture) -> None:
         grid_power_per_phase_recv = (
             grid._power_per_phase.new_receiver()  # pylint: disable=protected-access
         )
+        # Wait for mock resampler to create forward stream
+        await asyncio.sleep(0.01)
 
         for count in range(10):
             watts_delta = 1 if count % 2 == 0 else -1
@@ -273,6 +282,8 @@ async def test_grid_production_consumption_power_consumer_meter(
         stack.push_async_callback(grid.stop)
 
         grid_recv = grid.power.new_receiver()
+        # Wait for mock resampler to create forward stream
+        await asyncio.sleep(0.01)
 
         await mockgrid.mock_resampler.send_meter_power([1.0, 2.0, 3.0, 4.0])
         assert (await grid_recv.receive()).value == Power.from_watts(10.0)
@@ -295,6 +306,8 @@ async def test_grid_production_consumption_power_no_grid_meter(
         stack.push_async_callback(grid.stop)
 
         grid_recv = grid.power.new_receiver()
+        # Wait for mock resampler to create forward stream
+        await asyncio.sleep(0.01)
 
         await mockgrid.mock_resampler.send_meter_power([2.5, 3.5, 4.0])
         assert (await grid_recv.receive()).value == Power.from_watts(10.0)
@@ -315,6 +328,8 @@ async def test_consumer_power_2_grid_meters(mocker: MockerFixture) -> None:
         stack.push_async_callback(grid.stop)
 
         grid_recv = grid.power.new_receiver()
+        # Wait for mock resampler to create forward stream
+        await asyncio.sleep(0.01)
 
         await mockgrid.mock_resampler.send_meter_power([1.0, 2.0])
         assert (await grid_recv.receive()).value == Power.from_watts(3.0)

tests/timeseries/_battery_pool/test_battery_pool.py

@@ -545,6 +545,8 @@ async def test_batter_pool_power_no_batteries(mocker: MockerFixture) -> None:
     await mockgrid.start(mocker)
     battery_pool = microgrid.new_battery_pool(priority=5)
     power_receiver = battery_pool.power.new_receiver()
+    # Wait for mock resampler to create forward stream
+    await asyncio.sleep(0.01)
 
     await mockgrid.mock_resampler.send_non_existing_component_value()
     assert (await power_receiver.receive()).value == Power.from_watts(0)
@@ -594,6 +596,8 @@ async def _test_battery_pool_power(mockgrid: MockMicrogrid) -> None:
     async with mockgrid:
         battery_pool = microgrid.new_battery_pool(priority=5)
         power_receiver = battery_pool.power.new_receiver()
+        # Wait for mock resampler to create forward stream
+        await asyncio.sleep(0.01)
 
         await mockgrid.mock_resampler.send_bat_inverter_power([2.0, 3.0])
         assert (await power_receiver.receive()).value == Power.from_watts(5.0)

tests/timeseries/_ev_charger_pool/test_ev_charger_pool.py

@@ -3,6 +3,7 @@
 
 """Tests for the `EVChargerPool`."""
 
+import asyncio
 
 from pytest_mock import MockerFixture
 
@@ -25,6 +26,8 @@ async def test_ev_power(  # pylint: disable=too-many-locals
         async with mockgrid:
             ev_pool = microgrid.new_ev_charger_pool(priority=5)
             power_receiver = ev_pool.power.new_receiver()
+            # Wait for mock resampler to create forward stream
+            await asyncio.sleep(0.01)
 
             await mockgrid.mock_resampler.send_evc_power([2.0, 4.0, 10.0])
             assert (await power_receiver.receive()).value == Power.from_watts(16.0)

tests/timeseries/_formula_engine/test_formula_composition.py

@@ -3,7 +3,7 @@
 
 """Tests for formula composition."""
 
-
+import asyncio
 import math
 from contextlib import AsyncExitStack
 
@@ -53,6 +53,8 @@ async def test_formula_composition(  # pylint: disable=too-many-locals
             grid_power_recv = grid.power.new_receiver()
             battery_power_recv = battery_pool.power.new_receiver()
             pv_power_recv = pv_pool.power.new_receiver()
+            # Wait for mock resampler to create forward stream
+            await asyncio.sleep(0.01)
 
             engine = (pv_pool.power + battery_pool.power).build("inv_power")
             stack.push_async_callback(engine._stop)  # pylint: disable=protected-access
@@ -125,8 +127,10 @@ async def test_formula_composition_missing_pv(self, mocker: MockerFixture) -> No
             battery_power_recv = battery_pool.power.new_receiver()
             pv_power_recv = pv_pool.power.new_receiver()
             engine = (pv_pool.power + battery_pool.power).build("inv_power")
-            stack.push_async_callback(engine._stop)  # pylint: disable=protected-access
+            # Wait for mock resampler to create forward stream
+            await asyncio.sleep(0.01)
 
+            stack.push_async_callback(engine._stop)  # pylint: disable=protected-access
             inv_calc_recv = engine.new_receiver()
 
             for _ in range(10):
@@ -167,8 +171,10 @@ async def test_formula_composition_missing_bat(self, mocker: MockerFixture) -> N
             battery_power_recv = battery_pool.power.new_receiver()
             pv_power_recv = pv_pool.power.new_receiver()
             engine = (pv_pool.power + battery_pool.power).build("inv_power")
-            stack.push_async_callback(engine._stop)  # pylint: disable=protected-access
+            # Wait for mock resampler to create forward stream
+            await asyncio.sleep(0.01)
 
+            stack.push_async_callback(engine._stop)  # pylint: disable=protected-access
             inv_calc_recv = engine.new_receiver()
 
             for _ in range(10):
@@ -212,6 +218,8 @@ async def test_formula_composition_min_max(self, mocker: MockerFixture) -> None:
                 engine_max._stop  # pylint: disable=protected-access
             )
             engine_max_rx = engine_max.new_receiver()
+            # Wait for mock resampler to create forward stream
+            await asyncio.sleep(0.01)
 
             await mockgrid.mock_resampler.send_meter_power([100.0, 200.0])
 
@@ -275,6 +283,9 @@ async def test_formula_composition_min_max_const(
             )
             engine_max_rx = engine_max.new_receiver()
 
+            # Wait for mock resampler to create forward stream
+            await asyncio.sleep(0.01)
+
             await mockgrid.mock_resampler.send_meter_power([100.0])
 
             # Test min
@@ -346,6 +357,9 @@ async def test_formula_composition_constant(  # pylint: disable=too-many-locals
                 * 2.0
             ).build("grid_power_composite")
 
+            # Wait for mock resampler to create forward stream
+            await asyncio.sleep(0.01)
+
             await mockgrid.mock_resampler.send_meter_power([100.0])
 
             # Test addition
@@ -437,6 +451,9 @@ async def test_3_phase_formulas(self, mocker: MockerFixture) -> None:
             stack.push_async_callback(engine._stop)  # pylint: disable=protected-access
             net_current_recv = engine.new_receiver()
 
+            # Wait for mock resampler to create forward stream
+            await asyncio.sleep(0.01)
+
             for _ in range(10):
                 await mockgrid.mock_resampler.send_meter_current(
                     [

tests/timeseries/mock_resampler.py

@@ -216,6 +216,14 @@ def _resampling_request_sender(self) -> Sender[ComponentMetricRequest]:
     async def _channel_forward_messages(
         self, receiver: Receiver[Sample[Quantity]], sender: Sender[Sample[Quantity]]
     ) -> None:
+        # In normal run, new created receiver is empty.
+        # Here receiver might have last message.
+        # Consume it, so that it imitates normal run.
+        try:
+            sample = receiver.consume()
+        except AssertionError:
+            pass
+
         async for sample in receiver:
             await sender.send(sample)
 

tests/timeseries/test_consumer.py

@@ -3,6 +3,7 @@
 
 """Test the logical component for calculating high level consumer metrics."""
 
+import asyncio
 from contextlib import AsyncExitStack
 
 from pytest_mock import MockerFixture
@@ -26,6 +27,8 @@ async def test_consumer_power_grid_meter(self, mocker: MockerFixture) -> None:
             consumer = microgrid.consumer()
             stack.push_async_callback(consumer.stop)
             consumer_power_receiver = consumer.power.new_receiver()
+            # Wait for mock resampler to create forward stream
+            await asyncio.sleep(0.01)
 
             await mockgrid.mock_resampler.send_meter_power([20.0, 2.0, 3.0, 4.0, 5.0])
             assert (await consumer_power_receiver.receive()).value == Power.from_watts(
@@ -43,6 +46,8 @@ async def test_consumer_power_no_grid_meter(self, mocker: MockerFixture) -> None
             consumer = microgrid.consumer()
             stack.push_async_callback(consumer.stop)
             consumer_power_receiver = consumer.power.new_receiver()
+            # Wait for mock resampler to create forward stream
+            await asyncio.sleep(0.01)
 
             await mockgrid.mock_resampler.send_meter_power([20.0, 2.0, 3.0, 4.0, 5.0])
             assert (await consumer_power_receiver.receive()).value == Power.from_watts(
@@ -61,6 +66,8 @@ async def test_consumer_power_no_grid_meter_no_consumer_meter(
             consumer = microgrid.consumer()
             stack.push_async_callback(consumer.stop)
             consumer_power_receiver = consumer.power.new_receiver()
+            # Wait for mock resampler to create forward stream
+            await asyncio.sleep(0.01)
 
             await mockgrid.mock_resampler.send_non_existing_component_value()
             assert (await consumer_power_receiver.receive()).value == Power.from_watts(

tests/timeseries/test_logical_meter.py

@@ -3,7 +3,7 @@
 
 """Tests for the logical meter."""
 
-
+import asyncio
 from contextlib import AsyncExitStack
 
 from pytest_mock import MockerFixture
@@ -31,6 +31,8 @@ async def test_chp_power(self, mocker: MockerFixture) -> None:
             stack.push_async_callback(logical_meter.stop)
 
             chp_power_receiver = logical_meter.chp_power.new_receiver()
+            # Wait for mock resampler to create forward stream
+            await asyncio.sleep(0.01)
 
             await mockgrid.mock_resampler.send_meter_power([2.0, 3.0, 4.0])
             assert (await chp_power_receiver.receive()).value == Power.from_watts(2.0)
@@ -47,6 +49,8 @@ async def test_pv_power(self, mocker: MockerFixture) -> None:
             pv_pool = microgrid.new_pv_pool(priority=5)
             stack.push_async_callback(pv_pool.stop)
             pv_power_receiver = pv_pool.power.new_receiver()
+            # Wait for mock resampler to create forward stream
+            await asyncio.sleep(0.01)
 
             await mockgrid.mock_resampler.send_meter_power([-1.0, -2.0])
             await mockgrid.mock_resampler.send_pv_inverter_power([-10.0, -20.0])
@@ -61,6 +65,8 @@ async def test_pv_power_no_meter(self, mocker: MockerFixture) -> None:
             pv_pool = microgrid.new_pv_pool(priority=5)
             stack.push_async_callback(pv_pool.stop)
             pv_power_receiver = pv_pool.power.new_receiver()
+            # Wait for mock resampler to create forward stream
+            await asyncio.sleep(0.01)
 
             await mockgrid.mock_resampler.send_pv_inverter_power([-1.0, -2.0])
             assert (await pv_power_receiver.receive()).value == Power.from_watts(-3.0)
@@ -74,6 +80,8 @@ async def test_pv_power_no_pv_components(self, mocker: MockerFixture) -> None:
             pv_pool = microgrid.new_pv_pool(priority=5)
             stack.push_async_callback(pv_pool.stop)
             pv_power_receiver = pv_pool.power.new_receiver()
+            # Wait for mock resampler to create forward stream
+            await asyncio.sleep(0.01)
 
             await mockgrid.mock_resampler.send_non_existing_component_value()
             assert (await pv_power_receiver.receive()).value == Power.zero()

tests/timeseries/test_producer.py

@@ -3,6 +3,7 @@
 
 """Test the logical component for calculating high level producer metrics."""
 
+import asyncio
 from contextlib import AsyncExitStack
 
 from pytest_mock import MockerFixture
@@ -26,6 +27,8 @@ async def test_producer_power(self, mocker: MockerFixture) -> None:
             producer = microgrid.producer()
             stack.push_async_callback(producer.stop)
             producer_power_receiver = producer.power.new_receiver()
+            # Wait for mock resampler to create forward stream
+            await asyncio.sleep(0.01)
 
             await mockgrid.mock_resampler.send_meter_power([2.0, 3.0, 4.0, 5.0])
             assert (await producer_power_receiver.receive()).value == Power.from_watts(
@@ -41,6 +44,8 @@ async def test_producer_power_no_chp(self, mocker: MockerFixture) -> None:
             producer = microgrid.producer()
             stack.push_async_callback(producer.stop)
             producer_power_receiver = producer.power.new_receiver()
+            # Wait for mock resampler to create forward stream
+            await asyncio.sleep(0.01)
 
             await mockgrid.mock_resampler.send_meter_power([2.0, 3.0])
             assert (await producer_power_receiver.receive()).value == Power.from_watts(
@@ -58,6 +63,8 @@ async def test_producer_power_no_pv_no_consumer_meter(
             producer = microgrid.producer()
             stack.push_async_callback(producer.stop)
             producer_power_receiver = producer.power.new_receiver()
+            # Wait for mock resampler to create forward stream
+            await asyncio.sleep(0.01)
 
             await mockgrid.mock_resampler.send_chp_power([2.0])
             assert (await producer_power_receiver.receive()).value == Power.from_watts(
@@ -74,6 +81,8 @@ async def test_producer_power_no_pv(self, mocker: MockerFixture) -> None:
             producer = microgrid.producer()
             stack.push_async_callback(producer.stop)
             producer_power_receiver = producer.power.new_receiver()
+            # Wait for mock resampler to create forward stream
+            await asyncio.sleep(0.01)
 
             await mockgrid.mock_resampler.send_meter_power([20.0, 2.0])
             assert (await producer_power_receiver.receive()).value == Power.from_watts(
@@ -89,6 +98,8 @@ async def test_no_producer_power(self, mocker: MockerFixture) -> None:
             producer = microgrid.producer()
             stack.push_async_callback(producer.stop)
             producer_power_receiver = producer.power.new_receiver()
+            # Wait for mock resampler to create forward stream
+            await asyncio.sleep(0.01)
 
             await mockgrid.mock_resampler.send_non_existing_component_value()
             assert (await producer_power_receiver.receive()).value == Power.from_watts(