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
|
import ssl
from functools import partial
import pytest
import aio_pika
from tests import test_amqp as amqp
@pytest.fixture(
scope="module", params=[aio_pika.connect, aio_pika.connect_robust],
)
def connection_fabric(request):
return request.param
@pytest.fixture
def create_connection(connection_fabric, event_loop, rabbitmq_container):
ssl_context = ssl.create_default_context()
ssl_context.check_hostname = False
ssl_context.verify_mode = ssl.VerifyMode.CERT_NONE
return partial(
connection_fabric,
rabbitmq_container.get_amqps_url(),
loop=event_loop,
ssl_context=ssl_context,
)
async def test_default_context(connection_fabric, amqp_url):
with pytest.raises(ConnectionError):
await connection_fabric(
amqp_url.with_scheme("amqps").with_port(5671),
ssl_context=None,
)
ssl_context = ssl.create_default_context()
with pytest.raises(ConnectionError):
await connection_fabric(
amqp_url.with_scheme("amqps").with_port(5671),
ssl_context=ssl_context,
)
class TestCaseAMQPS(amqp.TestCaseAmqp):
pass
|
