evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
hass-core/tests/components/dhcp/test_init.py
J. Nick Koston 958c4764f7
Handle routers that convert hostnames to lowercase in dhcp () 
Some routers will lowercase all the hostnames. Since we
already lowercase hostnames for matching purposes, we now
pass the lowercased hostname to the integration.

Currently only roomba cared about this, and has been adjusted.
2021-03-28 11:23:06 +02:00

563 lines
18 KiB
Python
Raw Blame History

"""Test the DHCP discovery integration."""
import threading
from unittest.mock import patch
from scapy.error import Scapy_Exception
from scapy.layers.dhcp import DHCP
from scapy.layers.l2 import Ether
from homeassistant.components import dhcp
from homeassistant.components.device_tracker.const import (
ATTR_HOST_NAME,
ATTR_IP,
ATTR_MAC,
ATTR_SOURCE_TYPE,
SOURCE_TYPE_ROUTER,
)
from homeassistant.const import (
EVENT_HOMEASSISTANT_STARTED,
EVENT_HOMEASSISTANT_STOP,
STATE_HOME,
STATE_NOT_HOME,
)
from homeassistant.setup import async_setup_component
from tests.common import mock_coro
# connect b8:b7:f1:6d:b5:33 192.168.210.56
RAW_DHCP_REQUEST = (
b"\xff\xff\xff\xff\xff\xff\xb8\xb7\xf1m\xb53\x08\x00E\x00\x01P\x06E"
b"\x00\x00\xff\x11\xb4X\x00\x00\x00\x00\xff\xff\xff\xff\x00D\x00C\x01<"
b"\x0b\x14\x01\x01\x06\x00jmjV\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"
b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\xb8\xb7\xf1m\xb53\x00\x00\x00\x00"
b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"
b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"
b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"
b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"
b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"
b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"
b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"
b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"
b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"
b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"
b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"
b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"
b"\x00\x00\x00\x00\x00\x00c\x82Sc5\x01\x039\x02\x05\xdc2\x04\xc0\xa8\xd286"
b"\x04\xc0\xa8\xd0\x017\x04\x01\x03\x1c\x06\x0c\x07connect\xff\x00\x00\x00"
b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"
b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"
)
async def test_dhcp_match_hostname_and_macaddress(hass):
"""Test matching based on hostname and macaddress."""
dhcp_watcher = dhcp.DHCPWatcher(
hass,
{},
[{"domain": "mock-domain", "hostname": "connect", "macaddress": "B8B7F1*"}],
)
packet = Ether(RAW_DHCP_REQUEST)
with patch.object(
hass.config_entries.flow, "async_init", return_value=mock_coro()
) as mock_init:
dhcp_watcher.handle_dhcp_packet(packet)
# Ensure no change is ignored
dhcp_watcher.handle_dhcp_packet(packet)
assert len(mock_init.mock_calls) == 1
assert mock_init.mock_calls[0][1][0] == "mock-domain"
assert mock_init.mock_calls[0][2]["context"] == {"source": "dhcp"}
assert mock_init.mock_calls[0][2]["data"] == {
dhcp.IP_ADDRESS: "192.168.210.56",
dhcp.HOSTNAME: "connect",
dhcp.MAC_ADDRESS: "b8b7f16db533",
}
async def test_dhcp_match_hostname(hass):
"""Test matching based on hostname only."""
dhcp_watcher = dhcp.DHCPWatcher(
hass, {}, [{"domain": "mock-domain", "hostname": "connect"}]
)
packet = Ether(RAW_DHCP_REQUEST)
with patch.object(
hass.config_entries.flow, "async_init", return_value=mock_coro()
) as mock_init:
dhcp_watcher.handle_dhcp_packet(packet)
assert len(mock_init.mock_calls) == 1
assert mock_init.mock_calls[0][1][0] == "mock-domain"
assert mock_init.mock_calls[0][2]["context"] == {"source": "dhcp"}
assert mock_init.mock_calls[0][2]["data"] == {
dhcp.IP_ADDRESS: "192.168.210.56",
dhcp.HOSTNAME: "connect",
dhcp.MAC_ADDRESS: "b8b7f16db533",
}
async def test_dhcp_match_macaddress(hass):
"""Test matching based on macaddress only."""
dhcp_watcher = dhcp.DHCPWatcher(
hass, {}, [{"domain": "mock-domain", "macaddress": "B8B7F1*"}]
)
packet = Ether(RAW_DHCP_REQUEST)
with patch.object(
hass.config_entries.flow, "async_init", return_value=mock_coro()
) as mock_init:
dhcp_watcher.handle_dhcp_packet(packet)
assert len(mock_init.mock_calls) == 1
assert mock_init.mock_calls[0][1][0] == "mock-domain"
assert mock_init.mock_calls[0][2]["context"] == {"source": "dhcp"}
assert mock_init.mock_calls[0][2]["data"] == {
dhcp.IP_ADDRESS: "192.168.210.56",
dhcp.HOSTNAME: "connect",
dhcp.MAC_ADDRESS: "b8b7f16db533",
}
async def test_dhcp_nomatch(hass):
"""Test not matching based on macaddress only."""
dhcp_watcher = dhcp.DHCPWatcher(
hass, {}, [{"domain": "mock-domain", "macaddress": "ABC123*"}]
)
packet = Ether(RAW_DHCP_REQUEST)
with patch.object(
hass.config_entries.flow, "async_init", return_value=mock_coro()
) as mock_init:
dhcp_watcher.handle_dhcp_packet(packet)
assert len(mock_init.mock_calls) == 0
async def test_dhcp_nomatch_hostname(hass):
"""Test not matching based on hostname only."""
dhcp_watcher = dhcp.DHCPWatcher(
hass, {}, [{"domain": "mock-domain", "hostname": "nomatch*"}]
)
packet = Ether(RAW_DHCP_REQUEST)
with patch.object(
hass.config_entries.flow, "async_init", return_value=mock_coro()
) as mock_init:
dhcp_watcher.handle_dhcp_packet(packet)
assert len(mock_init.mock_calls) == 0
async def test_dhcp_nomatch_non_dhcp_packet(hass):
"""Test matching does not throw on a non-dhcp packet."""
dhcp_watcher = dhcp.DHCPWatcher(
hass, {}, [{"domain": "mock-domain", "hostname": "nomatch*"}]
)
packet = Ether(b"")
with patch.object(
hass.config_entries.flow, "async_init", return_value=mock_coro()
) as mock_init:
dhcp_watcher.handle_dhcp_packet(packet)
assert len(mock_init.mock_calls) == 0
async def test_dhcp_nomatch_non_dhcp_request_packet(hass):
"""Test nothing happens with the wrong message-type."""
dhcp_watcher = dhcp.DHCPWatcher(
hass, {}, [{"domain": "mock-domain", "hostname": "nomatch*"}]
)
packet = Ether(RAW_DHCP_REQUEST)
packet[DHCP].options = [
("message-type", 4),
("max_dhcp_size", 1500),
("requested_addr", "192.168.210.56"),
("server_id", "192.168.208.1"),
("param_req_list", [1, 3, 28, 6]),
("hostname", b"connect"),
]
with patch.object(
hass.config_entries.flow, "async_init", return_value=mock_coro()
) as mock_init:
dhcp_watcher.handle_dhcp_packet(packet)
assert len(mock_init.mock_calls) == 0
async def test_dhcp_invalid_hostname(hass):
"""Test we ignore invalid hostnames."""
dhcp_watcher = dhcp.DHCPWatcher(
hass, {}, [{"domain": "mock-domain", "hostname": "nomatch*"}]
)
packet = Ether(RAW_DHCP_REQUEST)
packet[DHCP].options = [
("message-type", 3),
("max_dhcp_size", 1500),
("requested_addr", "192.168.210.56"),
("server_id", "192.168.208.1"),
("param_req_list", [1, 3, 28, 6]),
("hostname", "connect"),
]
with patch.object(
hass.config_entries.flow, "async_init", return_value=mock_coro()
) as mock_init:
dhcp_watcher.handle_dhcp_packet(packet)
assert len(mock_init.mock_calls) == 0
async def test_dhcp_missing_hostname(hass):
"""Test we ignore missing hostnames."""
dhcp_watcher = dhcp.DHCPWatcher(
hass, {}, [{"domain": "mock-domain", "hostname": "nomatch*"}]
)
packet = Ether(RAW_DHCP_REQUEST)
packet[DHCP].options = [
("message-type", 3),
("max_dhcp_size", 1500),
("requested_addr", "192.168.210.56"),
("server_id", "192.168.208.1"),
("param_req_list", [1, 3, 28, 6]),
("hostname", None),
]
with patch.object(
hass.config_entries.flow, "async_init", return_value=mock_coro()
) as mock_init:
dhcp_watcher.handle_dhcp_packet(packet)
assert len(mock_init.mock_calls) == 0
async def test_dhcp_invalid_option(hass):
"""Test we ignore invalid hostname option."""
dhcp_watcher = dhcp.DHCPWatcher(
hass, {}, [{"domain": "mock-domain", "hostname": "nomatch*"}]
)
packet = Ether(RAW_DHCP_REQUEST)
packet[DHCP].options = [
("message-type", 3),
("max_dhcp_size", 1500),
("requested_addr", "192.168.208.55"),
("server_id", "192.168.208.1"),
("param_req_list", [1, 3, 28, 6]),
("hostname"),
]
with patch.object(
hass.config_entries.flow, "async_init", return_value=mock_coro()
) as mock_init:
dhcp_watcher.handle_dhcp_packet(packet)
assert len(mock_init.mock_calls) == 0
async def test_setup_and_stop(hass):
"""Test we can setup and stop."""
assert await async_setup_component(
hass,
dhcp.DOMAIN,
{},
)
await hass.async_block_till_done()
with patch("homeassistant.components.dhcp.AsyncSniffer.start") as start_call, patch(
"homeassistant.components.dhcp._verify_l2socket_setup",
), patch(
"homeassistant.components.dhcp.compile_filter",
):
hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED)
await hass.async_block_till_done()
hass.bus.async_fire(EVENT_HOMEASSISTANT_STOP)
await hass.async_block_till_done()
start_call.assert_called_once()
async def test_setup_fails_as_root(hass, caplog):
"""Test we handle sniff setup failing as root."""
assert await async_setup_component(
hass,
dhcp.DOMAIN,
{},
)
await hass.async_block_till_done()
wait_event = threading.Event()
with patch("os.geteuid", return_value=0), patch(
"homeassistant.components.dhcp._verify_l2socket_setup",
side_effect=Scapy_Exception,
):
hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED)
await hass.async_block_till_done()
hass.bus.async_fire(EVENT_HOMEASSISTANT_STOP)
await hass.async_block_till_done()
wait_event.set()
assert "Cannot watch for dhcp packets" in caplog.text
async def test_setup_fails_non_root(hass, caplog):
"""Test we handle sniff setup failing as non-root."""
assert await async_setup_component(
hass,
dhcp.DOMAIN,
{},
)
await hass.async_block_till_done()
with patch("os.geteuid", return_value=10), patch(
"homeassistant.components.dhcp._verify_l2socket_setup",
side_effect=Scapy_Exception,
):
hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED)
await hass.async_block_till_done()
hass.bus.async_fire(EVENT_HOMEASSISTANT_STOP)
await hass.async_block_till_done()
assert "Cannot watch for dhcp packets without root or CAP_NET_RAW" in caplog.text
async def test_setup_fails_with_broken_libpcap(hass, caplog):
"""Test we abort if libpcap is missing or broken."""
assert await async_setup_component(
hass,
dhcp.DOMAIN,
{},
)
await hass.async_block_till_done()
with patch("homeassistant.components.dhcp._verify_l2socket_setup",), patch(
"homeassistant.components.dhcp.compile_filter",
side_effect=ImportError,
) as compile_filter, patch(
"homeassistant.components.dhcp.AsyncSniffer",
) as async_sniffer:
hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED)
await hass.async_block_till_done()
hass.bus.async_fire(EVENT_HOMEASSISTANT_STOP)
await hass.async_block_till_done()
assert compile_filter.called
assert not async_sniffer.called
assert (
"Cannot watch for dhcp packets without a functional packet filter"
in caplog.text
)
async def test_device_tracker_hostname_and_macaddress_exists_before_start(hass):
"""Test matching based on hostname and macaddress before start."""
hass.states.async_set(
"device_tracker.august_connect",
STATE_HOME,
{
ATTR_HOST_NAME: "Connect",
ATTR_IP: "192.168.210.56",
ATTR_SOURCE_TYPE: SOURCE_TYPE_ROUTER,
ATTR_MAC: "B8:B7:F1:6D:B5:33",
},
)
with patch.object(
hass.config_entries.flow, "async_init", return_value=mock_coro()
) as mock_init:
device_tracker_watcher = dhcp.DeviceTrackerWatcher(
hass,
{},
[{"domain": "mock-domain", "hostname": "connect", "macaddress": "B8B7F1*"}],
)
await device_tracker_watcher.async_start()
await hass.async_block_till_done()
await device_tracker_watcher.async_stop()
await hass.async_block_till_done()
assert len(mock_init.mock_calls) == 1
assert mock_init.mock_calls[0][1][0] == "mock-domain"
assert mock_init.mock_calls[0][2]["context"] == {"source": "dhcp"}
assert mock_init.mock_calls[0][2]["data"] == {
dhcp.IP_ADDRESS: "192.168.210.56",
dhcp.HOSTNAME: "connect",
dhcp.MAC_ADDRESS: "b8b7f16db533",
}
async def test_device_tracker_hostname_and_macaddress_after_start(hass):
"""Test matching based on hostname and macaddress after start."""
with patch.object(
hass.config_entries.flow, "async_init", return_value=mock_coro()
) as mock_init:
device_tracker_watcher = dhcp.DeviceTrackerWatcher(
hass,
{},
[{"domain": "mock-domain", "hostname": "connect", "macaddress": "B8B7F1*"}],
)
await device_tracker_watcher.async_start()
await hass.async_block_till_done()
hass.states.async_set(
"device_tracker.august_connect",
STATE_HOME,
{
ATTR_HOST_NAME: "Connect",
ATTR_IP: "192.168.210.56",
ATTR_SOURCE_TYPE: SOURCE_TYPE_ROUTER,
ATTR_MAC: "B8:B7:F1:6D:B5:33",
},
)
await hass.async_block_till_done()
await device_tracker_watcher.async_stop()
await hass.async_block_till_done()
assert len(mock_init.mock_calls) == 1
assert mock_init.mock_calls[0][1][0] == "mock-domain"
assert mock_init.mock_calls[0][2]["context"] == {"source": "dhcp"}
assert mock_init.mock_calls[0][2]["data"] == {
dhcp.IP_ADDRESS: "192.168.210.56",
dhcp.HOSTNAME: "connect",
dhcp.MAC_ADDRESS: "b8b7f16db533",
}
async def test_device_tracker_hostname_and_macaddress_after_start_not_home(hass):
"""Test matching based on hostname and macaddress after start but not home."""
with patch.object(
hass.config_entries.flow, "async_init", return_value=mock_coro()
) as mock_init:
device_tracker_watcher = dhcp.DeviceTrackerWatcher(
hass,
{},
[{"domain": "mock-domain", "hostname": "connect", "macaddress": "B8B7F1*"}],
)
await device_tracker_watcher.async_start()
await hass.async_block_till_done()
hass.states.async_set(
"device_tracker.august_connect",
STATE_NOT_HOME,
{
ATTR_HOST_NAME: "connect",
ATTR_IP: "192.168.210.56",
ATTR_SOURCE_TYPE: SOURCE_TYPE_ROUTER,
ATTR_MAC: "B8:B7:F1:6D:B5:33",
},
)
await hass.async_block_till_done()
await device_tracker_watcher.async_stop()
await hass.async_block_till_done()
assert len(mock_init.mock_calls) == 0
async def test_device_tracker_hostname_and_macaddress_after_start_not_router(hass):
"""Test matching based on hostname and macaddress after start but not router."""
with patch.object(
hass.config_entries.flow, "async_init", return_value=mock_coro()
) as mock_init:
device_tracker_watcher = dhcp.DeviceTrackerWatcher(
hass,
{},
[{"domain": "mock-domain", "hostname": "connect", "macaddress": "B8B7F1*"}],
)
await device_tracker_watcher.async_start()
await hass.async_block_till_done()
hass.states.async_set(
"device_tracker.august_connect",
STATE_HOME,
{
ATTR_HOST_NAME: "connect",
ATTR_IP: "192.168.210.56",
ATTR_SOURCE_TYPE: "something_else",
ATTR_MAC: "B8:B7:F1:6D:B5:33",
},
)
await hass.async_block_till_done()
await device_tracker_watcher.async_stop()
await hass.async_block_till_done()
assert len(mock_init.mock_calls) == 0
async def test_device_tracker_hostname_and_macaddress_after_start_hostname_missing(
hass,
):
"""Test matching based on hostname and macaddress after start but missing hostname."""
with patch.object(
hass.config_entries.flow, "async_init", return_value=mock_coro()
) as mock_init:
device_tracker_watcher = dhcp.DeviceTrackerWatcher(
hass,
{},
[{"domain": "mock-domain", "hostname": "connect", "macaddress": "B8B7F1*"}],
)
await device_tracker_watcher.async_start()
await hass.async_block_till_done()
hass.states.async_set(
"device_tracker.august_connect",
STATE_HOME,
{
ATTR_IP: "192.168.210.56",
ATTR_SOURCE_TYPE: SOURCE_TYPE_ROUTER,
ATTR_MAC: "B8:B7:F1:6D:B5:33",
},
)
await hass.async_block_till_done()
await device_tracker_watcher.async_stop()
await hass.async_block_till_done()
assert len(mock_init.mock_calls) == 0
async def test_device_tracker_ignore_self_assigned_ips_before_start(hass):
"""Test matching ignores self assigned ip address."""
hass.states.async_set(
"device_tracker.august_connect",
STATE_HOME,
{
ATTR_HOST_NAME: "connect",
ATTR_IP: "169.254.210.56",
ATTR_SOURCE_TYPE: SOURCE_TYPE_ROUTER,
ATTR_MAC: "B8:B7:F1:6D:B5:33",
},
)
with patch.object(
hass.config_entries.flow, "async_init", return_value=mock_coro()
) as mock_init:
device_tracker_watcher = dhcp.DeviceTrackerWatcher(
hass,
{},
[{"domain": "mock-domain", "hostname": "connect", "macaddress": "B8B7F1*"}],
)
await device_tracker_watcher.async_start()
await hass.async_block_till_done()
await device_tracker_watcher.async_stop()
await hass.async_block_till_done()
assert len(mock_init.mock_calls) == 0
Reference in a new issue View git blame Copy permalink