Skip to content

Commit 0609b7a

Browse files
Add test that fails because of actor bug
When an actor throws an exception during cancellation, it restarts rather than stopping as expected. Signed-off-by: Elzbieta Kotulska <[email protected]>
1 parent 15bcff5 commit 0609b7a

File tree

1 file changed

+58
-0
lines changed

1 file changed

+58
-0
lines changed

tests/actor/test_actor.py

Lines changed: 58 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -53,6 +53,31 @@ async def _run(self) -> None:
5353
print(f"{self} done")
5454

5555

56+
class RaiseExceptionOnCancelActor(BaseTestActor):
57+
"""Actor that raises exception during stop."""
58+
59+
def __init__(
60+
self, *, name: str | None = None, recv: Receiver[int], sender: Sender[int]
61+
) -> None:
62+
"""Create an instance."""
63+
super().__init__(name=name)
64+
self._recv = recv
65+
self._sender = sender
66+
67+
async def _run(self) -> None:
68+
"""Start the actor and raise exception after receiving CancelledError."""
69+
self.inc_restart_count()
70+
if BaseTestActor.restart_count == 1:
71+
# This actor should not restart
72+
# If it does we just return to avoid infinite await on `stop`
73+
return
74+
try:
75+
async for msg in self._recv:
76+
await self._sender.send(msg)
77+
except asyncio.CancelledError as exc:
78+
raise RuntimeError("Actor should stop.") from exc
79+
80+
5681
class RaiseExceptionActor(BaseTestActor):
5782
"""A faulty actor that raises an Exception as soon as it receives a message."""
5883

@@ -333,3 +358,36 @@ async def cancel_actor() -> None:
333358
(*RUN_INFO, "Actor EchoActor[EchoActor]: Cancelled while running."),
334359
(*RUN_INFO, "All 1 actor(s) finished."),
335360
]
361+
362+
363+
async def test_actor_stop_if_error_was_raised_during_cancel(
364+
actor_auto_restart_once: None, # pylint: disable=unused-argument
365+
caplog: pytest.LogCaptureFixture,
366+
) -> None:
367+
"""If actor raises exception during cancellation it should stop.
368+
369+
And throw unhandled exception to the user..
370+
"""
371+
caplog.set_level("DEBUG", logger="frequenz.sdk.actor._actor")
372+
caplog.set_level("DEBUG", logger="frequenz.sdk.actor._run_utils")
373+
374+
input_chan: Broadcast[int] = Broadcast(name="TestChannel")
375+
376+
echo_chan: Broadcast[int] = Broadcast(name="echo output")
377+
echo_rx = echo_chan.new_receiver()
378+
379+
actor = RaiseExceptionOnCancelActor(
380+
name="test",
381+
recv=input_chan.new_receiver(),
382+
sender=echo_chan.new_sender(),
383+
)
384+
385+
# Start actor and make sure it is running
386+
actor.start()
387+
await input_chan.new_sender().send(5)
388+
msg = await echo_rx.receive()
389+
assert msg == 5
390+
assert actor.is_running is True
391+
392+
await actor.stop()
393+
assert actor.restart_count == 0

0 commit comments

Comments
 (0)