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
|
import asyncio
import pytest
from yarl import URL
from aio_pika import connect
VARIANTS = (
(dict(url="amqp://localhost/"), "amqp://localhost/"),
(dict(url="amqp://localhost"), "amqp://localhost/"),
(dict(url="amqp://localhost:5674"), "amqp://localhost:5674/"),
(dict(url="amqp://localhost:5674//"), "amqp://localhost:5674//"),
(dict(url="amqp://localhost:5674/"), "amqp://localhost:5674/"),
(dict(host="localhost", port=8888), "amqp://guest:guest@localhost:8888//"),
(
dict(host="localhost", port=8888, virtualhost="foo"),
"amqp://guest:guest@localhost:8888/foo",
),
(
dict(host="localhost", port=8888, virtualhost="/foo"),
"amqp://guest:guest@localhost:8888//foo",
),
)
class FakeConnection:
def __init__(self, url, **kwargs):
self.url = URL(url)
self.kwargs = kwargs
async def connect(self, timeout=None, **kwargs):
return
@pytest.mark.parametrize("kwargs,expected", VARIANTS)
def test_simple(kwargs, expected):
loop = asyncio.get_event_loop()
conn: FakeConnection = loop.run_until_complete(
connect(connection_class=FakeConnection, **kwargs), # type: ignore
)
assert conn.url == URL(expected)
|
