diff --git a/backend/fluksio/flow/connector.py b/backend/fluksio/flow/connector.py index ac49e4b..18f2e77 100644 --- a/backend/fluksio/flow/connector.py +++ b/backend/fluksio/flow/connector.py @@ -161,11 +161,7 @@ class ConnectorNode(Node): return self._stop_event.set() if self._poll_task is not None: - self._poll_task.cancel() - try: - await self._poll_task - except (asyncio.CancelledError, Exception): # noqa: B014 - shutting down - pass + await self._cancel_task(self._poll_task) self._poll_task = None self._stop_event = None self._last_published = {} diff --git a/backend/tests/flow/test_node_teardown.py b/backend/tests/flow/test_node_teardown.py index 440ce9d..a1b284c 100644 --- a/backend/tests/flow/test_node_teardown.py +++ b/backend/tests/flow/test_node_teardown.py @@ -4,17 +4,19 @@ import asyncio import pytest +from fluksio.flow.connector import ConnectorNode from fluksio.flow.nodes import DelayNode -def test_stop_cron_lets_the_callers_cancellation_through(): - async def stubborn() -> None: - """A loop whose shutdown does not answer the first cancellation.""" - try: - await asyncio.sleep(3600) - except asyncio.CancelledError: - await asyncio.sleep(3600) +async def stubborn() -> None: + """A loop whose shutdown does not answer the first cancellation.""" + try: + await asyncio.sleep(3600) + except asyncio.CancelledError: + await asyncio.sleep(3600) + +def test_stop_cron_lets_the_callers_cancellation_through(): async def scenario() -> None: node = DelayNode(params={"cron": "* * * * *"}) node._stop_cron = asyncio.Event() @@ -29,3 +31,20 @@ def test_stop_cron_lets_the_callers_cancellation_through(): node._cron_task.cancel() asyncio.run(scenario()) + + +def test_connector_stop_lets_the_callers_cancellation_through(): + async def scenario() -> None: + node = ConnectorNode() + node._stop_event = asyncio.Event() + node._poll_task = asyncio.create_task(stubborn()) + + stopping = asyncio.create_task(node.stop()) + await asyncio.sleep(0.05) # let it reach the await on the poll task + stopping.cancel() + with pytest.raises(asyncio.CancelledError): + await stopping + + node._poll_task.cancel() + + asyncio.run(scenario())