broadcaster/tests/test_concurrent.py at 54e70b255ffc954087af8b7ea586bdbce4338ce9 · encode/broadcaster · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
"""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


@pytest.mark.asyncio
@pytest.mark.parametrize(["setup_broadcast"], URLS, indirect=True)
async def test_sub(setup_broadcast):
    uid = uuid4()
    channel1 = f"chatroom-{uid}1"
    channel2 = f"chatroom-{uid}2"

    to_sub = [
        setup_broadcast._backend.subscribe(channel1),
        setup_broadcast._backend.subscribe(channel2),
    ]
    await asyncio.gather(*to_sub)


@pytest.mark.asyncio
@pytest.mark.parametrize(["setup_broadcast"], URLS, indirect=True)
async def test_unsub(setup_broadcast):
    uid = uuid4()
    channel1 = f"chatroom-{uid}1"
    channel2 = f"chatroom-{uid}2"

    await setup_broadcast._backend.subscribe(channel1)
    await setup_broadcast._backend.subscribe(channel2)

    to_unsub = [
        setup_broadcast._backend.unsubscribe(channel1),
        setup_broadcast._backend.unsubscribe(channel2),
    ]

    await asyncio.gather(*to_unsub)