aioesphomeapi/tests/test_connection.py

79 lines
2.2 KiB
Python
Raw Normal View History

2021-07-12 20:09:17 +02:00
import asyncio
import socket
import pytest
2021-07-29 16:09:16 +02:00
from mock import AsyncMock, MagicMock, Mock, patch
2021-07-12 20:09:17 +02:00
from aioesphomeapi.api_pb2 import ConnectResponse, HelloResponse
from aioesphomeapi.connection import APIConnection, ConnectionParams, ConnectionState
from aioesphomeapi.core import APIConnectionError, RequiresEncryptionAPIError
2021-07-12 20:09:17 +02:00
from aioesphomeapi.host_resolver import AddrInfo, IPv4Sockaddr
@pytest.fixture
def connection_params() -> ConnectionParams:
return ConnectionParams(
address="fake.address",
port=6052,
password=None,
client_info="Tests client",
keepalive=15.0,
zeroconf_instance=None,
2021-09-08 23:12:07 +02:00
noise_psk=None,
2021-07-12 20:09:17 +02:00
)
@pytest.fixture
def conn(connection_params) -> APIConnection:
async def on_stop():
pass
return APIConnection(connection_params, on_stop)
@pytest.fixture
def resolve_host():
with patch("aioesphomeapi.host_resolver.async_resolve_host") as func:
func.return_value = AddrInfo(
family=socket.AF_INET,
type=socket.SOCK_STREAM,
proto=socket.IPPROTO_TCP,
sockaddr=IPv4Sockaddr("10.0.0.512", 6052),
)
yield func
@pytest.fixture
def socket_socket():
with patch("socket.socket") as func:
yield func
@pytest.mark.asyncio
async def test_connect(conn, resolve_host, socket_socket, event_loop):
with patch.object(event_loop, "sock_connect"), patch(
"asyncio.open_connection", return_value=(None, None)
), patch.object(conn, "_read_loop"), patch.object(
conn, "_connect_start_ping"
2021-07-12 20:09:17 +02:00
), patch.object(
conn, "send_message_await_response", return_value=HelloResponse()
):
2021-10-21 19:20:05 +02:00
await conn.connect(login=False)
2021-07-12 20:09:17 +02:00
assert conn.is_connected
@pytest.mark.asyncio
async def test_requires_encryption_propagates(conn):
with patch("asyncio.open_connection") as openc:
reader = MagicMock()
writer = MagicMock()
openc.return_value = (reader, writer)
writer.drain = AsyncMock()
reader.readexactly = AsyncMock()
reader.readexactly.return_value = b"\x01"
await conn._connect_init_frame_helper()
with pytest.raises(RequiresEncryptionAPIError):
await conn._connect_hello()