mirror of
https://github.com/home-assistant/core.git
synced 2025-07-23 05:07:41 +00:00
tests
This commit is contained in:
parent
1d1a3ec830
commit
c2541c7f64
@ -1,39 +1 @@
|
|||||||
"""Tests for the dhcp integration."""
|
"""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)
|
|
||||||
|
@ -1,10 +1,14 @@
|
|||||||
"""Test the DHCP discovery integration."""
|
"""Test the DHCP discovery integration."""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from collections.abc import Awaitable, Callable
|
||||||
import datetime
|
import datetime
|
||||||
import threading
|
import threading
|
||||||
from typing import Any
|
from typing import Any, cast
|
||||||
from unittest.mock import patch
|
from unittest.mock import patch
|
||||||
|
|
||||||
|
import aiodhcpwatcher
|
||||||
import pytest
|
import pytest
|
||||||
from scapy import interfaces
|
from scapy import interfaces
|
||||||
from scapy.error import Scapy_Exception
|
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.setup import async_setup_component
|
||||||
from homeassistant.util import dt as dt_util
|
from homeassistant.util import dt as dt_util
|
||||||
|
|
||||||
from . import async_get_handle_dhcp_packet
|
|
||||||
|
|
||||||
from tests.common import (
|
from tests.common import (
|
||||||
MockConfigEntry,
|
MockConfigEntry,
|
||||||
MockModule,
|
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:
|
async def test_dhcp_match_hostname_and_macaddress(hass: HomeAssistant) -> None:
|
||||||
"""Test matching based on hostname and macaddress."""
|
"""Test matching based on hostname and macaddress."""
|
||||||
integration_matchers = dhcp.async_index_integration_matchers(
|
integration_matchers = dhcp.async_index_integration_matchers(
|
||||||
|
75
tests/components/dhcp/test_websocket_api.py
Normal file
75
tests/components/dhcp/test_websocket_api.py
Normal file
@ -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",
|
||||||
|
}
|
||||||
|
]
|
||||||
|
}
|
Loading…
x
Reference in New Issue
Block a user