Merge branch 'master' into upstream

faust/agents/replies.py

@@ -175,6 +175,7 @@ async def _start_fetcher(self, topic_name: str) -> None:
             # declare the topic
             topic = self._reply_topic(topic_name)
             await topic.maybe_declare()
+            self.app.topics.add(topic)
             await self.sleep(3.0)
             # then create the future
             self._fetchers[topic_name] = self.add_future(self._drain_replies(topic))

faust/transport/conductor.py

@@ -413,7 +413,7 @@ def add(self, topic: TopicT) -> None:
 
     def _topic_contain_unsubscribed_topics(self, topic: TopicT) -> bool:
         index = self._topic_name_index
-        return bool(index and any(t not in index for t in topic.topics))
+        return bool(any(t not in index for t in topic.topics))
 
     def discard(self, topic: Any) -> None:
         """Unregister topic from conductor."""

tests/unit/agents/test_replies.py

@@ -1,6 +1,6 @@
 import asyncio
 import json
-from unittest.mock import Mock
+from unittest.mock import MagicMock, Mock
 
 import pytest
 
@@ -192,7 +192,7 @@ async def test_add(self, *, c):
     async def test_start_fetcher(self, *, c):
         c._drain_replies = Mock()
         c._reply_topic = Mock(
-            return_value=Mock(
+            return_value=MagicMock(
                 maybe_declare=AsyncMock(),
             ),
         )