WIP-test

Author: llucax

tests/streaming/test_grpc_stream_broadcaster.py

@@ -344,3 +344,43 @@ async def test_messages_on_retry(
         StreamStarted(),
         StreamFatalError(error),
     ]
+
+
+@mock.patch(
+    "frequenz.client.base.streaming.asyncio.sleep", autospec=True, wraps=asyncio.sleep
+)
+async def test_retry_reset(
+    mock_sleep: mock.MagicMock,
+    receiver_ready_event: asyncio.Event,  # pylint: disable=redefined-outer-name
+) -> None:
+    """Test that retry strategy resets after a successful start."""
+    helper = streaming.GrpcStreamBroadcaster(
+        stream_name="test_helper",
+        stream_method=lambda: _ErroringAsyncIter(
+            mock_error(),
+            receiver_ready_event,
+        ),
+        transform=_transformer,
+        retry_strategy=retry.LinearBackoff(
+            limit=1,
+            interval=0.01,
+            jitter=0.0,
+        ),
+        retry_on_exhausted_stream=True,
+    )
+
+    items: list[str] = []
+    events: list[StreamEvent] = []
+    async with AsyncExitStack() as stack:
+        stack.push_async_callback(helper.stop)
+
+        receiver = helper.new_receiver()
+        receiver_ready_event.set()
+        _ = await _split_message(receiver)
+
+    mock_sleep.assert_has_calls(
+        [
+            mock.call(0.01),  # First retry after the first error
+            mock.call(0.01),  # Second retry after the second error
+        ]
+    )