fix: TestStreamClient should not wait for topics that are empty (#93)

Co-authored-by: Marcos Schroh <[email protected]>

kstreams/test_utils/topics.py

@@ -115,9 +115,12 @@ def all_messages_consumed(cls) -> bool:
     @classmethod
     async def join(cls) -> None:
         """
-        Wait for all topic messages to be processed
+        Wait for all topic messages to be processed.
+        Only topics that have a consumer assigned should be awaited.
         """
-        await asyncio.gather(*[topic.join() for topic in cls.topics.values()])
+        await asyncio.gather(
+            *[topic.join() for topic in cls.topics.values() if not topic.consumed]
+        )
 
     @classmethod
     def clean(cls) -> None:

tests/test_client.py

@@ -79,6 +79,31 @@ async def consume(stream):
     save_to_db.assert_called_once_with(event)
 
 
+@pytest.mark.asyncio
+async def test_only_consume_topics_with_streams(stream_engine: StreamEngine):
+    """
+    The test creates a stream but no events are send to it,
+    it means that the `TestStreamClient` should not wait for the topic to be consumed
+    even thought the topic is exist.
+    """
+    client = TestStreamClient(stream_engine)
+    topic = "local--kstreams"
+
+    @stream_engine.stream("a-different-topic", name="my-stream")
+    async def consume(stream):
+        async for cr in stream:
+            ...
+
+    async with client:
+        metadata = await client.send(
+            topic, value=b'{"message": "Hello world!"}', key="1"
+        )
+
+        assert metadata.topic == topic
+        assert metadata.partition == 0
+        assert metadata.offset == 1
+
+
 @pytest.mark.asyncio
 async def test_topic_created(stream_engine: StreamEngine):
     topic_name = "local--kstreams"