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
|
from unittest import mock
from amqp.exceptions import AccessRefused
from health_check.contrib.rabbitmq.backends import RabbitMQHealthCheck
class TestRabbitMQHealthCheck:
"""Test RabbitMQ health check."""
@mock.patch("health_check.contrib.rabbitmq.backends.getattr")
@mock.patch("health_check.contrib.rabbitmq.backends.Connection")
def test_broker_refused_connection(self, mocked_connection, mocked_getattr):
"""Test when the connection to RabbitMQ is refused."""
mocked_getattr.return_value = "broker_url"
conn_exception = ConnectionRefusedError("Refused connection")
# mock returns
mocked_conn = mock.MagicMock()
mocked_connection.return_value.__enter__.return_value = mocked_conn
mocked_conn.connect.side_effect = conn_exception
# instantiates the class
rabbitmq_healthchecker = RabbitMQHealthCheck()
# invokes the method check_status()
rabbitmq_healthchecker.check_status()
assert len(rabbitmq_healthchecker.errors), 1
# mock assertions
mocked_connection.assert_called_once_with("broker_url")
@mock.patch("health_check.contrib.rabbitmq.backends.getattr")
@mock.patch("health_check.contrib.rabbitmq.backends.Connection")
def test_broker_auth_error(self, mocked_connection, mocked_getattr):
"""Test that the connection to RabbitMQ has an authentication error."""
mocked_getattr.return_value = "broker_url"
conn_exception = AccessRefused("Refused connection")
# mock returns
mocked_conn = mock.MagicMock()
mocked_connection.return_value.__enter__.return_value = mocked_conn
mocked_conn.connect.side_effect = conn_exception
# instantiates the class
rabbitmq_healthchecker = RabbitMQHealthCheck()
# invokes the method check_status()
rabbitmq_healthchecker.check_status()
assert len(rabbitmq_healthchecker.errors), 1
# mock assertions
mocked_connection.assert_called_once_with("broker_url")
@mock.patch("health_check.contrib.rabbitmq.backends.getattr")
@mock.patch("health_check.contrib.rabbitmq.backends.Connection")
def test_broker_connection_upon_none_url(self, mocked_connection, mocked_getattr):
"""Thest when the connection to RabbitMQ has no ``broker_url``."""
mocked_getattr.return_value = None
# if the variable BROKER_URL is not set, AccessRefused exception is raised
conn_exception = AccessRefused("Refused connection")
# mock returns
mocked_conn = mock.MagicMock()
mocked_connection.return_value.__enter__.return_value = mocked_conn
mocked_conn.connect.side_effect = conn_exception
# instantiates the class
rabbitmq_healthchecker = RabbitMQHealthCheck()
# invokes the method check_status()
rabbitmq_healthchecker.check_status()
assert len(rabbitmq_healthchecker.errors), 1
# mock assertions
mocked_connection.assert_called_once_with(None)
|
