aioesphomeapi/tests/test_host_resolver.py
2023-10-15 13:22:12 -10:00

190 lines
6.6 KiB
Python

import socket
from ipaddress import ip_address
import pytest
from mock import AsyncMock, MagicMock, patch
from zeroconf import DNSCache
from zeroconf.asyncio import AsyncServiceInfo, AsyncZeroconf
import aioesphomeapi.host_resolver as hr
from aioesphomeapi.core import APIConnectionError
@pytest.fixture
def async_zeroconf():
with patch("aioesphomeapi.host_resolver.AsyncZeroconf") as klass:
async_zeroconf = klass.return_value
async_zeroconf.async_close = AsyncMock()
async_zeroconf.zeroconf.cache = DNSCache()
yield async_zeroconf
@pytest.fixture
def addr_infos():
return [
hr.AddrInfo(
family=socket.AF_INET,
type=socket.SOCK_STREAM,
proto=socket.IPPROTO_TCP,
sockaddr=hr.IPv4Sockaddr(address="10.0.0.42", port=6052),
),
hr.AddrInfo(
family=socket.AF_INET6,
type=socket.SOCK_STREAM,
proto=socket.IPPROTO_TCP,
sockaddr=hr.IPv6Sockaddr(
address="2001:db8:85a3::8a2e:370:7334",
port=6052,
flowinfo=0,
scope_id=0,
),
),
]
@pytest.mark.asyncio
async def test_resolve_host_zeroconf(async_zeroconf: AsyncZeroconf, addr_infos):
info = MagicMock(auto_spec=AsyncServiceInfo)
info.ip_addresses_by_version.return_value = [
ip_address(b"\n\x00\x00*"),
ip_address(b" \x01\r\xb8\x85\xa3\x00\x00\x00\x00\x8a.\x03ps4"),
]
info.async_request = AsyncMock(return_value=True)
with patch("aioesphomeapi.host_resolver.AsyncServiceInfo", return_value=info):
ret = await hr._async_resolve_host_zeroconf("asdf", 6052)
info.async_request.assert_called_once()
async_zeroconf.async_close.assert_called_once_with()
assert ret == addr_infos
@pytest.mark.asyncio
async def test_resolve_host_passed_zeroconf_does_not_close(addr_infos):
async_zeroconf = AsyncZeroconf(zc=MagicMock())
async_zeroconf.async_close = AsyncMock()
async_zeroconf.zeroconf.cache = DNSCache()
info = MagicMock(auto_spec=AsyncServiceInfo)
info.ip_addresses_by_version.return_value = [
ip_address(b"\n\x00\x00*"),
ip_address(b" \x01\r\xb8\x85\xa3\x00\x00\x00\x00\x8a.\x03ps4"),
]
info.async_request = AsyncMock(return_value=True)
with patch("aioesphomeapi.host_resolver.AsyncServiceInfo", return_value=info):
ret = await hr._async_resolve_host_zeroconf(
"asdf", 6052, zeroconf_instance=async_zeroconf
)
info.async_request.assert_called_once()
async_zeroconf.async_close.assert_not_called()
assert ret == addr_infos
@pytest.mark.asyncio
async def test_resolve_host_zeroconf_empty(async_zeroconf: AsyncZeroconf):
with patch(
"aioesphomeapi.host_resolver.AsyncServiceInfo.async_request"
) as mock_async_request:
ret = await hr._async_resolve_host_zeroconf("asdf.local", 6052)
assert mock_async_request.call_count == 1
assert ret == []
@pytest.mark.asyncio
async def test_resolve_host_getaddrinfo(event_loop, addr_infos):
with patch.object(event_loop, "getaddrinfo") as mock:
mock.return_value = [
(
socket.AF_INET,
socket.SOCK_STREAM,
socket.IPPROTO_TCP,
"canon1",
("10.0.0.42", 6052),
),
(
socket.AF_INET6,
socket.SOCK_STREAM,
socket.IPPROTO_TCP,
"canon2",
("2001:db8:85a3::8a2e:370:7334", 6052, 0, 0),
),
(-1, socket.SOCK_STREAM, socket.IPPROTO_TCP, "canon3", ("10.0.0.42", 6052)),
]
ret = await hr._async_resolve_host_getaddrinfo("example.com", 6052)
assert ret == addr_infos
@pytest.mark.asyncio
async def test_resolve_host_getaddrinfo_oserror(event_loop):
with patch.object(event_loop, "getaddrinfo") as mock:
mock.side_effect = OSError()
with pytest.raises(APIConnectionError):
await hr._async_resolve_host_getaddrinfo("example.com", 6052)
@pytest.mark.asyncio
@patch("aioesphomeapi.host_resolver._async_resolve_host_zeroconf")
@patch("aioesphomeapi.host_resolver._async_resolve_host_getaddrinfo")
async def test_resolve_host_mdns(resolve_addr, resolve_zc, addr_infos):
resolve_zc.return_value = addr_infos
ret = await hr.async_resolve_host("example.local", 6052)
resolve_zc.assert_called_once_with("example", 6052, zeroconf_instance=None)
resolve_addr.assert_not_called()
assert ret == addr_infos[0]
@pytest.mark.asyncio
@patch("aioesphomeapi.host_resolver._async_resolve_host_zeroconf")
@patch("aioesphomeapi.host_resolver._async_resolve_host_getaddrinfo")
async def test_resolve_host_mdns_empty(resolve_addr, resolve_zc, addr_infos):
resolve_zc.return_value = []
resolve_addr.return_value = addr_infos
ret = await hr.async_resolve_host("example.local", 6052)
resolve_zc.assert_called_once_with("example", 6052, zeroconf_instance=None)
resolve_addr.assert_called_once_with("example.local", 6052)
assert ret == addr_infos[0]
@pytest.mark.asyncio
@patch("aioesphomeapi.host_resolver._async_resolve_host_zeroconf")
@patch("aioesphomeapi.host_resolver._async_resolve_host_getaddrinfo")
async def test_resolve_host_addrinfo(resolve_addr, resolve_zc, addr_infos):
resolve_addr.return_value = addr_infos
ret = await hr.async_resolve_host("example.com", 6052)
resolve_zc.assert_not_called()
resolve_addr.assert_called_once_with("example.com", 6052)
assert ret == addr_infos[0]
@pytest.mark.asyncio
@patch("aioesphomeapi.host_resolver._async_resolve_host_zeroconf")
@patch("aioesphomeapi.host_resolver._async_resolve_host_getaddrinfo")
async def test_resolve_host_addrinfo_empty(resolve_addr, resolve_zc, addr_infos):
resolve_addr.return_value = []
with pytest.raises(APIConnectionError):
await hr.async_resolve_host("example.com", 6052)
resolve_zc.assert_not_called()
resolve_addr.assert_called_once_with("example.com", 6052)
@pytest.mark.asyncio
@patch("aioesphomeapi.host_resolver._async_resolve_host_zeroconf")
@patch("aioesphomeapi.host_resolver._async_resolve_host_getaddrinfo")
async def test_resolve_host_with_address(resolve_addr, resolve_zc):
resolve_zc.return_value = []
resolve_addr.return_value = addr_infos
ret = await hr.async_resolve_host("127.0.0.1", 6052)
resolve_zc.assert_not_called()
resolve_addr.assert_not_called()
assert ret == hr.AddrInfo(
family=socket.AddressFamily.AF_INET,
type=socket.SocketKind.SOCK_STREAM,
proto=6,
sockaddr=hr.IPv4Sockaddr(address="127.0.0.1", port=6052),
)