From 144cb5f9efd652bf8dd4de1a0284f84e912f8cb9 Mon Sep 17 00:00:00 2001 From: William Barnhart Date: Fri, 29 Sep 2023 11:03:57 -0400 Subject: [PATCH 1/6] Update agent.py to call loop.create_task rather than asyncio.Task Credit goes to @samtx for finding this solution for https://github.com/faust-streaming/faust/issues/175#issuecomment-1177730361 --- faust/agents/agent.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/faust/agents/agent.py b/faust/agents/agent.py index 41bebfe6b..dffe068a6 100644 --- a/faust/agents/agent.py +++ b/faust/agents/agent.py @@ -661,7 +661,7 @@ async def _prepare_actor(self, aref: ActorRefT, beacon: NodeT) -> ActorRefT: else: # agent yields and is an AsyncIterator so we have to consume it. coro = self._slurp(aref, aiter(aref)) - task = asyncio.Task(self._execute_actor(coro, aref), loop=self.loop) + task = self.loop.create_task(self._execute_actor(coro, aref)) task._beacon = beacon # type: ignore aref.actor_task = task self._actors.add(aref) From 1d9400b33ff6579065a62b37d7087af3209df4bc Mon Sep 17 00:00:00 2001 From: William Barnhart Date: Fri, 29 Sep 2023 11:12:33 -0400 Subject: [PATCH 2/6] Update agent.py Co-authored-by: Sam Friedman --- faust/agents/agent.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/faust/agents/agent.py b/faust/agents/agent.py index dffe068a6..5de726f3b 100644 --- a/faust/agents/agent.py +++ b/faust/agents/agent.py @@ -661,6 +661,8 @@ async def _prepare_actor(self, aref: ActorRefT, beacon: NodeT) -> ActorRefT: else: # agent yields and is an AsyncIterator so we have to consume it. coro = self._slurp(aref, aiter(aref)) + # Calling asyncio.Task is not proper usage of asyncio, we need to create the task + # directly from the loop task = self.loop.create_task(self._execute_actor(coro, aref)) task._beacon = beacon # type: ignore aref.actor_task = task From 166cb07520ffc01d5a04913267462ebb478fc135 Mon Sep 17 00:00:00 2001 From: William Barnhart Date: Fri, 29 Sep 2023 11:16:36 -0400 Subject: [PATCH 3/6] Fix linting in agent.py --- faust/agents/agent.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/faust/agents/agent.py b/faust/agents/agent.py index 5de726f3b..a16cd9515 100644 --- a/faust/agents/agent.py +++ b/faust/agents/agent.py @@ -661,8 +661,8 @@ async def _prepare_actor(self, aref: ActorRefT, beacon: NodeT) -> ActorRefT: else: # agent yields and is an AsyncIterator so we have to consume it. coro = self._slurp(aref, aiter(aref)) - # Calling asyncio.Task is not proper usage of asyncio, we need to create the task - # directly from the loop + # Calling asyncio.Task is not proper usage of asyncio, + # we need to create the task directly from the loop task = self.loop.create_task(self._execute_actor(coro, aref)) task._beacon = beacon # type: ignore aref.actor_task = task From 9669c0fd42ef8f94d227aa6090ae1b8a7ae2ee21 Mon Sep 17 00:00:00 2001 From: Sam Friedman Date: Tue, 2 Jan 2024 12:02:53 -0600 Subject: [PATCH 4/6] Update unit test to verify slurp is awaited --- tests/unit/agents/test_agent.py | 30 +++++++++++++++--------------- 1 file changed, 15 insertions(+), 15 deletions(-) diff --git a/tests/unit/agents/test_agent.py b/tests/unit/agents/test_agent.py index bb31b75b8..5bb64b7b8 100644 --- a/tests/unit/agents/test_agent.py +++ b/tests/unit/agents/test_agent.py @@ -392,22 +392,22 @@ async def test_start_task(self, *, agent): assert ret is agent._prepare_actor.return_value @pytest.mark.asyncio - async def test_prepare_actor__AsyncIterable(self, *, agent): + async def test_prepare_actor__AsyncIterable(self, *, agent, monkeypatch): + async def mock_execute_actor(coro, aref): + await coro + mock_beacon = Mock(name="beacon", autospec=Node) + mock_slurp = AsyncMock(name='slurp') + monkeypatch.setattr(agent, '_slurp', mock_slurp) + monkeypatch.setattr(agent, '_execute_actor', mock_execute_actor) aref = agent(index=0, active_partitions=None) - with patch("asyncio.Task") as Task: - agent._slurp = Mock(name="_slurp") - agent._execute_actor = Mock(name="_execute_actor") - beacon = Mock(name="beacon", autospec=Node) - ret = await agent._prepare_actor(aref, beacon) - agent._slurp.assert_called() - coro = agent._slurp() - agent._execute_actor.assert_called_once_with(coro, aref) - Task.assert_called_once_with(agent._execute_actor(), loop=agent.loop) - task = Task() - assert task._beacon is beacon - assert aref.actor_task is task - assert aref in agent._actors - assert ret is aref + ret = await agent._prepare_actor(aref, mock_beacon) + task = aref.actor_task + await task + mock_slurp.assert_awaited() + assert mock_slurp.await_args.args[0] is aref + assert task._beacon is mock_beacon + assert aref in agent._actors + assert ret is aref @pytest.mark.asyncio async def test_prepare_actor__Awaitable(self, *, agent2): From 214c7926010013c04e19bfab458eec74d548881b Mon Sep 17 00:00:00 2001 From: Sam Friedman Date: Tue, 2 Jan 2024 13:02:12 -0600 Subject: [PATCH 5/6] Update unit test to check awaitable task creation --- tests/unit/agents/test_agent.py | 28 ++++++++++++++-------------- 1 file changed, 14 insertions(+), 14 deletions(-) diff --git a/tests/unit/agents/test_agent.py b/tests/unit/agents/test_agent.py index 5bb64b7b8..75d258ac9 100644 --- a/tests/unit/agents/test_agent.py +++ b/tests/unit/agents/test_agent.py @@ -428,22 +428,22 @@ async def test_prepare_actor__Awaitable(self, *, agent2): assert ret is aref @pytest.mark.asyncio - async def test_prepare_actor__Awaitable_with_multiple_topics(self, *, agent2): + async def test_prepare_actor__Awaitable_with_multiple_topics(self, *, agent2, monkeypatch): aref = agent2(index=0, active_partitions=None) - asyncio.ensure_future(aref.it).cancel() # silence warning agent2.channel.topics = ["foo", "bar"] - with patch("asyncio.Task") as Task: - agent2._execute_actor = Mock(name="_execute_actor") - beacon = Mock(name="beacon", autospec=Node) - ret = await agent2._prepare_actor(aref, beacon) - coro = aref - agent2._execute_actor.assert_called_once_with(coro, aref) - Task.assert_called_once_with(agent2._execute_actor(), loop=agent2.loop) - task = Task() - assert task._beacon is beacon - assert aref.actor_task is task - assert aref in agent2._actors - assert ret is aref + mock_beacon = Mock(name="beacon", autospec=Node) + mock_slurp = AsyncMock(name='slurp') + mock_execute_actor = AsyncMock(name='execute_actor') + monkeypatch.setattr(agent2, '_slurp', mock_slurp) + monkeypatch.setattr(agent2, '_execute_actor', mock_execute_actor) + ret = await agent2._prepare_actor(aref, mock_beacon) + task = aref.actor_task + mock_slurp.assert_not_called() + mock_slurp.assert_not_awaited() + mock_execute_actor.assert_called_with(aref, aref) + assert task._beacon is mock_beacon + assert aref in agent2._actors + assert ret is aref @pytest.mark.asyncio async def test_prepare_actor__Awaitable_cannot_have_sinks(self, *, agent2): From 8ee21b0f31775707056b50b85d384242263998b7 Mon Sep 17 00:00:00 2001 From: Sam Friedman Date: Tue, 2 Jan 2024 14:20:00 -0600 Subject: [PATCH 6/6] Linting and formatting --- tests/unit/agents/test_agent.py | 19 +++++++++++-------- 1 file changed, 11 insertions(+), 8 deletions(-) diff --git a/tests/unit/agents/test_agent.py b/tests/unit/agents/test_agent.py index 75d258ac9..e58d876ee 100644 --- a/tests/unit/agents/test_agent.py +++ b/tests/unit/agents/test_agent.py @@ -395,10 +395,11 @@ async def test_start_task(self, *, agent): async def test_prepare_actor__AsyncIterable(self, *, agent, monkeypatch): async def mock_execute_actor(coro, aref): await coro + mock_beacon = Mock(name="beacon", autospec=Node) - mock_slurp = AsyncMock(name='slurp') - monkeypatch.setattr(agent, '_slurp', mock_slurp) - monkeypatch.setattr(agent, '_execute_actor', mock_execute_actor) + mock_slurp = AsyncMock(name="slurp") + monkeypatch.setattr(agent, "_slurp", mock_slurp) + monkeypatch.setattr(agent, "_execute_actor", mock_execute_actor) aref = agent(index=0, active_partitions=None) ret = await agent._prepare_actor(aref, mock_beacon) task = aref.actor_task @@ -428,14 +429,16 @@ async def test_prepare_actor__Awaitable(self, *, agent2): assert ret is aref @pytest.mark.asyncio - async def test_prepare_actor__Awaitable_with_multiple_topics(self, *, agent2, monkeypatch): + async def test_prepare_actor__Awaitable_with_multiple_topics( + self, *, agent2, monkeypatch + ): aref = agent2(index=0, active_partitions=None) agent2.channel.topics = ["foo", "bar"] mock_beacon = Mock(name="beacon", autospec=Node) - mock_slurp = AsyncMock(name='slurp') - mock_execute_actor = AsyncMock(name='execute_actor') - monkeypatch.setattr(agent2, '_slurp', mock_slurp) - monkeypatch.setattr(agent2, '_execute_actor', mock_execute_actor) + mock_slurp = AsyncMock(name="slurp") + mock_execute_actor = AsyncMock(name="execute_actor") + monkeypatch.setattr(agent2, "_slurp", mock_slurp) + monkeypatch.setattr(agent2, "_execute_actor", mock_execute_actor) ret = await agent2._prepare_actor(aref, mock_beacon) task = aref.actor_task mock_slurp.assert_not_called()