bunny-storm/tests/test_async_connection.py at 1159fc6b83f161a3daea341b22dcf37adcd86ca9 · PythonCloudFrameworks/bunny-storm · 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
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
import asyncio
import logging

import pytest
from aio_pika import RobustConnection

from bunny_storm import RabbitMQConnectionData, AsyncConnection


@pytest.mark.asyncio
async def test_async_connection_get_connection(event_loop: asyncio.AbstractEventLoop,
                                               logger: logging.Logger,
                                               rabbitmq_connection_data: RabbitMQConnectionData) -> None:
    # Arrange
    async_connection = AsyncConnection(rabbitmq_connection_data, logger, event_loop)

    # Act
    connection = await async_connection.get_connection()

    # Assert
    assert isinstance(connection, RobustConnection) and async_connection.is_connected()

@pytest.mark.asyncio
async def test_async_connection_get_ssl_connection(event_loop: asyncio.AbstractEventLoop,
                                               logger: logging.Logger,
                                               rabbitmq_ssl_connection_data: RabbitMQConnectionData) -> None:
    # Arrange
    async_connection = AsyncConnection(rabbitmq_ssl_connection_data, logger, event_loop)

    # Act
    connection = await async_connection.get_connection()

    # Assert
    assert isinstance(connection, RobustConnection) and async_connection.is_connected()

@pytest.mark.asyncio
async def test_async_connection_connection_failure(event_loop: asyncio.AbstractEventLoop,
                                                   logger: logging.Logger,
                                                   rabbitmq_connection_data: RabbitMQConnectionData) -> None:
    # Arrange
    rabbitmq_connection_data.host = "1.1.1.1"
    async_connection = AsyncConnection(rabbitmq_connection_data,
                                       logger,
                                       event_loop,
                                       connection_attempts=2,
                                       timeout=0.1,
                                       attempt_backoff=1)

    # Act
    with pytest.raises(ConnectionError):
        await async_connection.get_connection()


@pytest.mark.asyncio
async def test_async_connection_close_connection(event_loop: asyncio.AbstractEventLoop,
                                                 logger: logging.Logger,
                                                 rabbitmq_connection_data: RabbitMQConnectionData) -> None:
    # Arrange
    async_connection = AsyncConnection(rabbitmq_connection_data, logger, event_loop)
    connection = await async_connection.get_connection()

    # Act
    await async_connection.close()

    # Assert
    assert not async_connection.is_connected()

    # Act
    await async_connection.close()

    # Assert
    assert not async_connection.is_connected()

    # Act
    new_connection = await async_connection.get_connection()

    # Assert
    assert isinstance(new_connection, RobustConnection) and async_connection.is_connected()
    assert async_connection.is_connected() and connection != new_connection