test: add concurrent checks

* group together fixtures for backend testing

* Add per-backend setup to allow slow initialization.
  This is specially important as kafka subscribe can return while the consumer
  is not ready. See related test setup upstream:
    - https://github.com/aio-libs/aiokafka/blob/2c54e10c57760f779961a8c2f5df8ad609ef6983/tests/test_consumer.py#L433
    - https://github.com/aio-libs/aiokafka/blob/2c54e10c57760f779961a8c2f5df8ad609ef6983/tests/_testutil.py#L376
    - https://github.com/aio-libs/aiokafka/blob/2c54e10c57760f779961a8c2f5df8ad609ef6983/tests/_testutil.py#L364

fixes #42

Author: pwoolvett

tests/conftest.py

@@ -0,0 +1,59 @@
+"""Check for #22"""
+
+import asyncio
+import functools
+
+import pytest_asyncio
+
+from broadcaster import Broadcast
+from broadcaster._backends.kafka import KafkaBackend
+
+
+async def __has_topic_now(client, topic):
+    if await client.force_metadata_update():
+        if topic in client.cluster.topics():
+            print(f'Topic "{topic}" exists')
+            return True
+    return False
+
+
+async def __wait_has_topic(client, topic, *, timeout_sec=5):
+    poll_time_sec = 1 / 10000
+    from datetime import datetime
+
+    pre = datetime.now()
+    while True:
+        if (datetime.now() - pre).total_seconds() >= timeout_sec:
+            raise ValueError(f'No topic "{topic}" exists')
+        if await __has_topic_now(client, topic):
+            return
+        await asyncio.sleep(poll_time_sec)
+
+
+def kafka_backend_setup(kafka_backend):
+    """Block until consumer client contains the topic"""
+    subscribe_impl = kafka_backend.subscribe
+
+    @functools.wraps(subscribe_impl)
+    async def subscribe(channel: str) -> None:
+        await subscribe_impl(channel)
+        await __wait_has_topic(kafka_backend._consumer._client, channel)
+
+    kafka_backend.subscribe = subscribe
+
+
+BROADCASTS_SETUP = {
+    KafkaBackend: kafka_backend_setup,
+}
+
+
+@pytest_asyncio.fixture(scope="function")
+async def setup_broadcast(request):
+    url = request.param
+    async with Broadcast(url) as broadcast:
+        backend = broadcast._backend
+        for klass, setup in BROADCASTS_SETUP.items():
+            if isinstance(backend, klass):
+                setup(backend)
+                break
+        yield broadcast

tests/test_broadcast.py

@@ -1,45 +1,23 @@
-import pytest
-
-from broadcaster import Broadcast
-
-
-@pytest.mark.asyncio
-async def test_memory():
-    async with Broadcast("memory://") as broadcast:
-        async with broadcast.subscribe("chatroom") as subscriber:
-            await broadcast.publish("chatroom", "hello")
-            event = await subscriber.get()
-            assert event.channel == "chatroom"
-            assert event.message == "hello"
-
-
-@pytest.mark.asyncio
-async def test_redis():
-    async with Broadcast("redis://localhost:6379") as broadcast:
-        async with broadcast.subscribe("chatroom") as subscriber:
-            await broadcast.publish("chatroom", "hello")
-            event = await subscriber.get()
-            assert event.channel == "chatroom"
-            assert event.message == "hello"
+from uuid import uuid4
 
+import pytest
 
-@pytest.mark.asyncio
-async def test_postgres():
-    async with Broadcast(
-        "postgres://postgres:postgres@localhost:5432/broadcaster"
-    ) as broadcast:
-        async with broadcast.subscribe("chatroom") as subscriber:
-            await broadcast.publish("chatroom", "hello")
-            event = await subscriber.get()
-            assert event.channel == "chatroom"
-            assert event.message == "hello"
+URLS = [
+    ("memory://",),
+    ("redis://localhost:6379",),
+    ("postgres://postgres:postgres@localhost:5432/broadcaster",),
+    ("kafka://localhost:9092",),
+]
 
 
 @pytest.mark.asyncio
-async def test_kafka():
-    async with Broadcast("kafka://localhost:9092") as broadcast:
-        async with broadcast.subscribe("chatroom") as subscriber:
-            await broadcast.publish("chatroom", "hello")
-            event = await subscriber.get()
-            assert event.channel == "chatroom"
-            assert event.message == "hello"
+@pytest.mark.parametrize(["setup_broadcast"], URLS, indirect=True)
+async def test_broadcast(setup_broadcast):
+    uid = uuid4()
+    channel = f"chatroom-{uid}"
+    msg = f"hello {uid}"
+    async with setup_broadcast.subscribe(channel) as subscriber:
+        await setup_broadcast.publish(channel, msg)
+        event = await subscriber.get()
+        assert event.channel == channel
+        assert event.message == msg

tests/test_concurrent.py

@@ -0,0 +1,30 @@
+"""Check for #22"""
+import asyncio
+from uuid import uuid4
+
+import pytest
+
+MESSAGES = ["hello", "goodbye"]
+
+URLS = [
+    ("memory://",),
+    ("redis://localhost:6379",),
+    ("postgres://postgres:postgres@localhost:5432/broadcaster",),
+    ("kafka://localhost:9092",),
+]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(["setup_broadcast"], URLS, indirect=True)
+async def test_broadcast(setup_broadcast):
+    uid = uuid4()
+    channel = f"chatroom-{uid}"
+    msgs = [f"{msg} {uid}" for msg in MESSAGES]
+    async with setup_broadcast.subscribe(channel) as subscriber:
+        to_publish = [setup_broadcast.publish(channel, msg) for msg in msgs]
+
+        await asyncio.gather(*to_publish)
+        for msg in msgs:
+            event = await subscriber.get()
+            assert event.channel == channel
+            assert event.message == msg