Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
async def test_auth_plain(amqp_connection, event_loop):
auth = PlainAuth(amqp_connection).marshal()
assert auth == PlainAuth(amqp_connection).marshal()
auth_parts = auth.split(b"\x00")
assert auth_parts == [b"", b"guest", b"guest"]
connection = aiormq.Connection(
amqp_connection.url.with_user("foo").with_password("bar"),
loop=event_loop,
)
auth = PlainAuth(connection).marshal()
auth_parts = auth.split(b"\x00")
assert auth_parts == [b"", b"foo", b"bar"]
auth = PlainAuth(connection)
auth.value = b"boo"
assert auth.marshal() == b"boo"
async def test_conncetion_reject(event_loop):
with pytest.raises(ConnectionError):
await aiormq.connect(
"amqp://guest:guest@127.0.0.1:59999/", loop=event_loop
)
connection = aiormq.Connection(
"amqp://guest:guest@127.0.0.1:59999/", loop=event_loop
)
with pytest.raises(ConnectionError):
await event_loop.create_task(connection.connect())
async def test_bad_credentials(amqp_connection, event_loop):
connection = aiormq.Connection(
amqp_connection.url.with_password(uuid.uuid4().hex), loop=event_loop
)
with pytest.raises(aiormq.exceptions.ProbableAuthenticationError):
await connection.connect()
async def test_connection_urls_vhosts(url, vhost, event_loop):
assert aiormq.Connection(url, loop=event_loop).vhost == vhost
async def _make_connection(self, **kwargs) -> aiormq.Connection:
connection = await aiormq.connect(self.url, **kwargs)
connection.closing.add_done_callback(
partial(self._on_connection_close, self.connection)
)
return connection