<commit_before>import asyncio
import logging
from datetime import datetime

import pytest

from lightbus.config import Config
from lightbus.message import EventMessage
from lightbus.serializers import (
    ByFieldMessageSerializer,
    ByFieldMessageDeserializer,
    BlobMessageSerializer,
    BlobMessageDeserializer,
)
from lightbus.transports.redis import RedisEventTransport, StreamUse
from lightbus.utilities.async import cancel

pytestmark = pytest.mark.unit


@pytest.mark.run_loop
async def test_connection_manager(redis_event_transport):
    """Does get_redis() provide a working redis connection"""
    connection_manager = await redis_event_transport.connection_manager()
    with await connection_manager as redis:
        assert await redis.info()


@pytest.mark.run_loop
async def test_send_event(redis_event_transport: RedisEventTransport, redis_client):
    await redis_event_transport.send_event(
        EventMessage(api_name="my.api", event_name="my_event", id="123", kwargs={"field": "value"}),
        options={},
    )
    messages = await redis_client.xrange("my.api.my_event:stream")
    assert len(messages) == 1
    assert messages[0][1] == {
        b"api_name": b"my.api",
        b"event_name": b"my_event",
        b"id": b"123",
        b"version": b"1",
        b":field": b'"value"',
    }


@pytest.mark.run_loop
async def test_send_event_per_api_stream(redis_event_transport: RedisEventTransport, redis_client):
    redis_event_transport.stream_use = StreamUse.PER_API
    await redis_event_transport.send_event(
        EventMessage(api_name="my.api", event_name="my_event", kwargs={"field": "value"}, id="123"),
        options={},
    )
    messages = await redis_client.xrange("my.api.*:stream")
    assert len(messages) == 1
    assert messages[0][1] == {
        b"api_name": b"my.api",
        b"event_name": b"my_event",
        b"id": b"123",
        b"version": b"1",
        b":field": b'"value"',
    }


@pytest.mark.run_loop
async def test_consume_events(
    loop, redis_event_transport: RedisEventTransport, redis_client, dummy_api
):

    async def co_enqeue():
        await asyncio.sleep(0.1)
        return await redis_client.xadd(
            "my.dummy.my_event:stream",
            fields={
                b"api_name": b"my.dummy",
                b"event_name": b"my_event",
                b"id": b"123",
                b"version": b"1",
                b":field": b'"value"',
            },
        )

    async def co_consume():
        async for message_ in redis_event_transport.consume([("my.dummy", "my_event")], {}, loop):
            return message_

    enqueue_result, message = await asyncio.gather(co_enqeue(), co_consume())
    assert message.api_name == "my.dummy"
    assert message.event_name == "my_event"
    assert message.kwargs == {"field": "value"}
    assert message.native_id
    assert type(message.native_id) == str


@pytest.mark.run_loop
async def test_consume_events_multiple_consumers(
    loop, redis_event_transport: RedisEventTransport, redis_client, dummy_api
):
    messages = []

    async def co_consume():
        async for message_ in redis_event_transport.consume([("my.dummy", "my_event")], {}, loop):
            messages.append(message_)

    task1 = asyncio.ensure_future(co_consume())
    task2 = asyncio.ensure_future(co_consume())

    await asyncio.sleep(0.1)
    await redis_client.xadd(
        "my.dummy.my_event:stream",
        fields={
            b"api_name": b"my.dummy",
            b"event_name": b"my_event",
            b"id": b"123",
            b"version": b"1",
            b":field": b'"value"',
        },
    )
    await asyncio.sleep(0.1)

    # Two messages, to dummy values which indicate events have been acked
    assert len(messages) == 4

    await cancel(task1, task2)


@pytest.mark.run_loop
async def test_consume_events_multiple_consumers_one_group(
    loop, redis_pool, redis_client, dummy_api
):
    messages = []

    async def co_consume(consumer_number):
        event_transport = RedisEventTransport(
            redis_pool=redis_pool,
            consumer_group_prefix="test_cg",
            consumer_name=f"test_consumer{consumer_number}",
            stream_use=StreamUse.PER_EVENT,
        )
        consumer = event_transport.consume(
            listen_for=[("my.dummy", "my_event")],
            context={},
            loop=loop,
            consumer_group="single_group",
        )
        async for message_ in consumer:<commit_msg>Increasing timeout to see if that fixes travis ci<commit_after>import asyncio
import logging
from datetime import datetime

import pytest

from lightbus.config import Config
from lightbus.message import EventMessage
from lightbus.serializers import (
    ByFieldMessageSerializer,
    ByFieldMessageDeserializer,
    BlobMessageSerializer,
    BlobMessageDeserializer,
)
from lightbus.transports.redis import RedisEventTransport, StreamUse
from lightbus.utilities.async import cancel

pytestmark = pytest.mark.unit


@pytest.mark.run_loop
async def test_connection_manager(redis_event_transport):
    """Does get_redis() provide a working redis connection"""
    connection_manager = await redis_event_transport.connection_manager()
    with await connection_manager as redis:
        assert await redis.info()


@pytest.mark.run_loop
async def test_send_event(redis_event_transport: RedisEventTransport, redis_client):
    await redis_event_transport.send_event(
        EventMessage(api_name="my.api", event_name="my_event", id="123", kwargs={"field": "value"}),
        options={},
    )
    messages = await redis_client.xrange("my.api.my_event:stream")
    assert len(messages) == 1
    assert messages[0][1] == {
        b"api_name": b"my.api",
        b"event_name": b"my_event",
        b"id": b"123",
        b"version": b"1",
        b":field": b'"value"',
    }


@pytest.mark.run_loop
async def test_send_event_per_api_stream(redis_event_transport: RedisEventTransport, redis_client):
    redis_event_transport.stream_use = StreamUse.PER_API
    await redis_event_transport.send_event(
        EventMessage(api_name="my.api", event_name="my_event", kwargs={"field": "value"}, id="123"),
        options={},
    )
    messages = await redis_client.xrange("my.api.*:stream")
    assert len(messages) == 1
    assert messages[0][1] == {
        b"api_name": b"my.api",
        b"event_name": b"my_event",
        b"id": b"123",
        b"version": b"1",
        b":field": b'"value"',
    }


@pytest.mark.run_loop
async def test_consume_events(
    loop, redis_event_transport: RedisEventTransport, redis_client, dummy_api
):

    async def co_enqeue():
        await asyncio.sleep(0.1)
        return await redis_client.xadd(
            "my.dummy.my_event:stream",
            fields={
                b"api_name": b"my.dummy",
                b"event_name": b"my_event",
                b"id": b"123",
                b"version": b"1",
                b":field": b'"value"',
            },
        )

    async def co_consume():
        async for message_ in redis_event_transport.consume([("my.dummy", "my_event")], {}, loop):
            return message_

    enqueue_result, message = await asyncio.gather(co_enqeue(), co_consume())
    assert message.api_name == "my.dummy"
    assert message.event_name == "my_event"
    assert message.kwargs == {"field": "value"}
    assert message.native_id
    assert type(message.native_id) == str


@pytest.mark.run_loop
async def test_consume_events_multiple_consumers(
    loop, redis_event_transport: RedisEventTransport, redis_client, dummy_api
):
    messages = []

    async def co_consume():
        async for message_ in redis_event_transport.consume([("my.dummy", "my_event")], {}, loop):
            messages.append(message_)

    task1 = asyncio.ensure_future(co_consume())
    task2 = asyncio.ensure_future(co_consume())

    await asyncio.sleep(0.1)
    await redis_client.xadd(
        "my.dummy.my_event:stream",
        fields={
            b"api_name": b"my.dummy",
            b"event_name": b"my_event",
            b"id": b"123",
            b"version": b"1",
            b":field": b'"value"',
        },
    )
    await asyncio.sleep(0.5)

    # Two messages, to dummy values which indicate events have been acked
    assert len(messages) == 4

    await cancel(task1, task2)


@pytest.mark.run_loop
async def test_consume_events_multiple_consumers_one_group(
    loop, redis_pool, redis_client, dummy_api
):
    messages = []

    async def co_consume(consumer_number):
        event_transport = RedisEventTransport(
            redis_pool=redis_pool,
            consumer_group_prefix="test_cg",
            consumer_name=f"test_consumer{consumer_number}",
            stream_use=StreamUse.PER_EVENT,
        )
        consumer = event_transport.consume(
            listen_for=[("my.dummy", "my_event")],
            context={},
            loop=loop,
            consumer_group="single_group",
        )
        async for message_ in consumer:
{'fraction_non_alphanumeric': 0.11737297537164411, 'fraction_numerical': 0.00865320612380741, 'mean_word_length': 3.0626408292023433, 'pattern_counts': {'":': 54, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '408132', 'n_tokens_mistral': 2909, 'n_tokens_neox': 2862, 'n_words': 557}