From c2541c7f6414c3d5db03b672412f3f5d92a07275 Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Wed, 16 Apr 2025 12:19:50 -1000 Subject: [PATCH] tests --- tests/components/dhcp/__init__.py | 38 ----------- tests/components/dhcp/test_init.py | 34 +++++++++- tests/components/dhcp/test_websocket_api.py | 75 +++++++++++++++++++++ 3 files changed, 106 insertions(+), 41 deletions(-) create mode 100644 tests/components/dhcp/test_websocket_api.py diff --git a/tests/components/dhcp/__init__.py b/tests/components/dhcp/__init__.py index c40b1acde53..fc58a7de903 100644 --- a/tests/components/dhcp/__init__.py +++ b/tests/components/dhcp/__init__.py @@ -1,39 +1 @@ """Tests for the dhcp integration.""" - -from __future__ import annotations - -from collections.abc import Awaitable, Callable -from typing import Any, cast -from unittest.mock import patch - -import aiodhcpwatcher - -from homeassistant.components import dhcp -from homeassistant.components.dhcp.models import DHCPData -from homeassistant.core import HomeAssistant - - -async def async_get_handle_dhcp_packet( - hass: HomeAssistant, - integration_matchers: dhcp.DhcpMatchers, - address_data: dict | None = None, -) -> Callable[[Any], Awaitable[None]]: - """Make a handler for a dhcp packet.""" - if address_data is None: - address_data = {} - dhcp_watcher = dhcp.DHCPWatcher( - hass, - DHCPData(integration_matchers, set(), address_data), - ) - with patch("aiodhcpwatcher.async_start"): - await dhcp_watcher.async_start() - - def _async_handle_dhcp_request(request: aiodhcpwatcher.DHCPRequest) -> None: - dhcp_watcher._async_process_dhcp_request(request) - - handler = aiodhcpwatcher.make_packet_handler(_async_handle_dhcp_request) - - async def _async_handle_dhcp_packet(packet): - handler(packet) - - return cast("Callable[[Any], Awaitable[None]]", _async_handle_dhcp_packet) diff --git a/tests/components/dhcp/test_init.py b/tests/components/dhcp/test_init.py index 8c515cc3038..0b704cf0369 100644 --- a/tests/components/dhcp/test_init.py +++ b/tests/components/dhcp/test_init.py @@ -1,10 +1,14 @@ """Test the DHCP discovery integration.""" +from __future__ import annotations + +from collections.abc import Awaitable, Callable import datetime import threading -from typing import Any +from typing import Any, cast from unittest.mock import patch +import aiodhcpwatcher import pytest from scapy import interfaces from scapy.error import Scapy_Exception @@ -37,8 +41,6 @@ from homeassistant.helpers.service_info.dhcp import DhcpServiceInfo from homeassistant.setup import async_setup_component from homeassistant.util import dt as dt_util -from . import async_get_handle_dhcp_packet - from tests.common import ( MockConfigEntry, MockModule, @@ -143,6 +145,32 @@ RAW_DHCP_REQUEST_WITHOUT_HOSTNAME = ( ) +async def async_get_handle_dhcp_packet( + hass: HomeAssistant, + integration_matchers: dhcp.DhcpMatchers, + address_data: dict | None = None, +) -> Callable[[Any], Awaitable[None]]: + """Make a handler for a dhcp packet.""" + if address_data is None: + address_data = {} + dhcp_watcher = dhcp.DHCPWatcher( + hass, + DHCPData(integration_matchers, set(), address_data), + ) + with patch("aiodhcpwatcher.async_start"): + await dhcp_watcher.async_start() + + def _async_handle_dhcp_request(request: aiodhcpwatcher.DHCPRequest) -> None: + dhcp_watcher._async_process_dhcp_request(request) + + handler = aiodhcpwatcher.make_packet_handler(_async_handle_dhcp_request) + + async def _async_handle_dhcp_packet(packet): + handler(packet) + + return cast("Callable[[Any], Awaitable[None]]", _async_handle_dhcp_packet) + + async def test_dhcp_match_hostname_and_macaddress(hass: HomeAssistant) -> None: """Test matching based on hostname and macaddress.""" integration_matchers = dhcp.async_index_integration_matchers( diff --git a/tests/components/dhcp/test_websocket_api.py b/tests/components/dhcp/test_websocket_api.py new file mode 100644 index 00000000000..eb008c49ab1 --- /dev/null +++ b/tests/components/dhcp/test_websocket_api.py @@ -0,0 +1,75 @@ +"""The tests for the dhcp WebSocket API.""" + +import asyncio +from collections.abc import Callable +from unittest.mock import patch + +import aiodhcpwatcher + +from homeassistant.components.dhcp import DOMAIN +from homeassistant.core import EVENT_HOMEASSISTANT_STARTED, HomeAssistant +from homeassistant.setup import async_setup_component + +from tests.typing import WebSocketGenerator + + +async def test_subscribe_discovery( + hass: HomeAssistant, + hass_ws_client: WebSocketGenerator, +) -> None: + """Test dhcp subscribe_discovery.""" + saved_callback: Callable[[aiodhcpwatcher.DHCPRequest], None] | None = None + + async def mock_start( + callback: Callable[[aiodhcpwatcher.DHCPRequest], None], + ) -> None: + """Mock start.""" + nonlocal saved_callback + saved_callback = callback + + with ( + patch("homeassistant.components.dhcp.aiodhcpwatcher.async_start", mock_start), + patch("homeassistant.components.dhcp.DiscoverHosts"), + ): + await async_setup_component(hass, DOMAIN, {}) + await hass.async_block_till_done() + hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED) + await hass.async_block_till_done() + + saved_callback(aiodhcpwatcher.DHCPRequest("4.3.2.2", "happy", "44:44:33:11:23:12")) + client = await hass_ws_client() + await client.send_json( + { + "id": 1, + "type": "dhcp/subscribe_discovery", + } + ) + async with asyncio.timeout(1): + response = await client.receive_json() + assert response["success"] + + async with asyncio.timeout(1): + response = await client.receive_json() + assert response["event"] == { + "add": [ + { + "hostname": "happy", + "ip_address": "4.3.2.2", + "mac_address": "44:44:33:11:23:12", + } + ] + } + + saved_callback(aiodhcpwatcher.DHCPRequest("4.3.2.1", "sad", "44:44:33:11:23:13")) + + async with asyncio.timeout(1): + response = await client.receive_json() + assert response["event"] == { + "add": [ + { + "hostname": "sad", + "ip_address": "4.3.2.1", + "mac_address": "44:44:33:11:23:13", + } + ] + }