jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-31 01:07:10 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add MockDhcpServiceInfo common test helper

Browse Source
This commit is contained in:
epenet 2025-05-14 06:11:24 +00:00
parent 9729f1f38b
commit 594769aa95
5 changed files with 60 additions and 84 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
tests/common.py
View File

@ -98,6 +98,7 @@ from homeassistant.helpers.entity_platform import (
AddEntitiesCallback,
)
from homeassistant.helpers.json import JSONEncoder, _orjson_default_encoder, json_dumps
from homeassistant.helpers.service_info.dhcp import DhcpServiceInfo
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.util import dt as dt_util, ulid as ulid_util, uuid as uuid_util
from homeassistant.util.async_ import (
@ -1953,3 +1954,26 @@ def get_schema_suggested_value(schema: vol.Schema, key: str) -> Any | None:
return None
return schema_key.description["suggested_value"]
return None
class MockDhcpServiceInfo(DhcpServiceInfo):
"""Mocked DHCP service info."""
def __init__(self, ip: str, hostname: str, macaddress: str) -> None:
"""Initialize the mock service info."""
# Historically, the MAC address was formatted without colons
# and since all consumers of this data are expecting it to be
# formatted without colons we will continue to do so
super().__init__(
ip=ip,
hostname=hostname,
macaddress=dr.format_mac(macaddress).replace(":", ""),
)
async def start_discovery_flow(
self, hass: HomeAssistant, domain: str
) -> ConfigFlowResult:
"""Start a reauthentication flow."""
return await hass.config_entries.flow.async_init(
domain, context={"source": config_entries.SOURCE_DHCP}, data=self
)

17
tests/components/airthings/test_config_flow.py
View File

@ -10,9 +10,8 @@ from homeassistant.components.airthings.const import CONF_SECRET, DOMAIN
from homeassistant.const import CONF_ID
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from homeassistant.helpers.service_info.dhcp import DhcpServiceInfo
from tests.common import MockConfigEntry
from tests.common import MockConfigEntry, MockDhcpServiceInfo
TEST_DATA = {
CONF_ID: "client_id",
@ -20,17 +19,17 @@ TEST_DATA = {
}
DHCP_SERVICE_INFO = [
DhcpServiceInfo(
MockDhcpServiceInfo(
hostname="airthings-view",
ip="192.168.1.100",
macaddress="00:00:00:00:00:00",
),
DhcpServiceInfo(
MockDhcpServiceInfo(
hostname="airthings-hub",
ip="192.168.1.101",
macaddress="D0:14:11:90:00:00",
),
DhcpServiceInfo(
MockDhcpServiceInfo(
hostname="airthings-hub",
ip="192.168.1.102",
macaddress="70:B3:D5:2A:00:00",
@ -147,15 +146,11 @@ async def test_flow_entry_already_exists(hass: HomeAssistant) -> None:
@pytest.mark.parametrize("dhcp_service_info", DHCP_SERVICE_INFO)
async def test_dhcp_flow(
hass: HomeAssistant, dhcp_service_info: DhcpServiceInfo
hass: HomeAssistant, dhcp_service_info: MockDhcpServiceInfo
) -> None:
"""Test the DHCP discovery flow."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
data=dhcp_service_info,
context={"source": config_entries.SOURCE_DHCP},
)
result = await dhcp_service_info.start_discovery_flow(hass, DOMAIN)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"

33
tests/components/airzone/test_config_flow.py
View File

@ -11,15 +11,12 @@ from aioairzone.exceptions import (
SystemOutOfRange,
)
from homeassistant import config_entries
from homeassistant.components.airzone.config_flow import short_mac
from homeassistant.components.airzone.const import DOMAIN
from homeassistant.config_entries import SOURCE_USER, ConfigEntryState
from homeassistant.const import CONF_HOST, CONF_ID, CONF_PORT
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.service_info.dhcp import DhcpServiceInfo
from .util import (
CONFIG,
@ -31,12 +28,12 @@ from .util import (
USER_INPUT,
)
from tests.common import MockConfigEntry
from tests.common import MockConfigEntry, MockDhcpServiceInfo
DHCP_SERVICE_INFO = DhcpServiceInfo(
DHCP_SERVICE_INFO = MockDhcpServiceInfo(
hostname="airzone",
ip="192.168.1.100",
macaddress=dr.format_mac("E84F25000000").replace(":", ""),
macaddress="E84F25000000",
)
TEST_ID = 1
@ -204,11 +201,7 @@ async def test_dhcp_flow(hass: HomeAssistant) -> None:
"homeassistant.components.airzone.AirzoneLocalApi.get_version",
return_value=HVAC_VERSION_MOCK,
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
data=DHCP_SERVICE_INFO,
context={"source": config_entries.SOURCE_DHCP},
)
result = await DHCP_SERVICE_INFO.start_discovery_flow(hass, DOMAIN)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "discovered_connection"
@ -262,11 +255,7 @@ async def test_dhcp_flow_error(hass: HomeAssistant) -> None:
"homeassistant.components.airzone.AirzoneLocalApi.get_version",
side_effect=AirzoneError,
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
data=DHCP_SERVICE_INFO,
context={"source": config_entries.SOURCE_DHCP},
)
result = await DHCP_SERVICE_INFO.start_discovery_flow(hass, DOMAIN)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "cannot_connect"
@ -279,11 +268,7 @@ async def test_dhcp_connection_error(hass: HomeAssistant) -> None:
"homeassistant.components.airzone.AirzoneLocalApi.get_version",
return_value=HVAC_VERSION_MOCK,
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
data=DHCP_SERVICE_INFO,
context={"source": config_entries.SOURCE_DHCP},
)
result = await DHCP_SERVICE_INFO.start_discovery_flow(hass, DOMAIN)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "discovered_connection"
@ -355,11 +340,7 @@ async def test_dhcp_invalid_system_id(hass: HomeAssistant) -> None:
"homeassistant.components.airzone.AirzoneLocalApi.get_version",
return_value=HVAC_VERSION_MOCK,
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
data=DHCP_SERVICE_INFO,
context={"source": config_entries.SOURCE_DHCP},
)
result = await DHCP_SERVICE_INFO.start_discovery_flow(hass, DOMAIN)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "discovered_connection"

29
tests/components/roborock/test_config_flow.py
View File

@ -19,11 +19,16 @@ from homeassistant.components.roborock.const import CONF_ENTRY_CODE, DOMAIN, DRA
from homeassistant.const import CONF_USERNAME, Platform
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from homeassistant.helpers.service_info.dhcp import DhcpServiceInfo
from .mock_data import MOCK_CONFIG, NETWORK_INFO, ROBOROCK_RRUID, USER_DATA, USER_EMAIL
from tests.common import MockConfigEntry
from tests.common import MockConfigEntry, MockDhcpServiceInfo
DNCP_SERVICE_INFO = MockDhcpServiceInfo(
ip=NETWORK_INFO.ip,
macaddress=NETWORK_INFO.mac,
hostname="roborock-vacuum-a72",
)
@pytest.fixture
@ -346,15 +351,7 @@ async def test_discovery_not_setup(
with (
patch("homeassistant.components.roborock.async_setup_entry", return_value=True),
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_DHCP},
data=DhcpServiceInfo(
ip=NETWORK_INFO.ip,
macaddress=NETWORK_INFO.mac.replace(":", ""),
hostname="roborock-vacuum-a72",
),
)
result = await DNCP_SERVICE_INFO.start_discovery_flow(hass, DOMAIN)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
with patch(
@ -391,15 +388,7 @@ async def test_discovery_already_setup(
"""Handle aborting if the device is already setup."""
await hass.config_entries.async_setup(mock_roborock_entry.entry_id)
await hass.async_block_till_done()
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_DHCP},
data=DhcpServiceInfo(
ip=NETWORK_INFO.ip,
macaddress=NETWORK_INFO.mac.replace(":", ""),
hostname="roborock-vacuum-a72",
),
)
result = await DNCP_SERVICE_INFO.start_discovery_flow(hass, DOMAIN)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"

41
tests/components/wmspro/test_config_flow.py
View File

@ -5,15 +5,14 @@ from unittest.mock import AsyncMock, patch
import aiohttp
from homeassistant.components.wmspro.const import DOMAIN
from homeassistant.config_entries import SOURCE_DHCP, SOURCE_USER, ConfigEntryState
from homeassistant.config_entries import SOURCE_USER, ConfigEntryState
from homeassistant.const import CONF_HOST
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from homeassistant.helpers.service_info.dhcp import DhcpServiceInfo
from . import setup_config_entry
from tests.common import MockConfigEntry
from tests.common import MockConfigEntry, MockDhcpServiceInfo
async def test_config_flow(
@ -49,12 +48,10 @@ async def test_config_flow_from_dhcp(
hass: HomeAssistant, mock_setup_entry: AsyncMock, mock_hub_refresh: AsyncMock
) -> None:
"""Test we can handle DHCP discovery to create a config entry."""
info = DhcpServiceInfo(
info = MockDhcpServiceInfo(
ip="1.2.3.4", hostname="webcontrol", macaddress="00:11:22:33:44:55"
)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_DHCP}, data=info
)
result = await info.start_discovery_flow(hass, DOMAIN)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {}
@ -108,12 +105,10 @@ async def test_config_flow_from_dhcp_add_mac(
assert len(mock_setup_entry.mock_calls) == 1
assert hass.config_entries.async_entries(DOMAIN)[0].unique_id is None
info = DhcpServiceInfo(
info = MockDhcpServiceInfo(
ip="1.2.3.4", hostname="webcontrol", macaddress="00:11:22:33:44:55"
)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_DHCP}, data=info
)
result = await info.start_discovery_flow(hass, DOMAIN)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
assert hass.config_entries.async_entries(DOMAIN)[0].unique_id == "00:11:22:33:44:55"
@ -125,12 +120,10 @@ async def test_config_flow_from_dhcp_ip_update(
mock_hub_refresh: AsyncMock,
) -> None:
"""Test we can use DHCP discovery to update IP in a config entry."""
info = DhcpServiceInfo(
info = MockDhcpServiceInfo(
ip="1.2.3.4", hostname="webcontrol", macaddress="00:11:22:33:44:55"
)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_DHCP}, data=info
)
result = await info.start_discovery_flow(hass, DOMAIN)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {}
@ -153,12 +146,10 @@ async def test_config_flow_from_dhcp_ip_update(
assert len(mock_setup_entry.mock_calls) == 1
assert hass.config_entries.async_entries(DOMAIN)[0].unique_id == "00:11:22:33:44:55"
info = DhcpServiceInfo(
info = MockDhcpServiceInfo(
ip="5.6.7.8", hostname="webcontrol", macaddress="00:11:22:33:44:55"
)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_DHCP}, data=info
)
result = await info.start_discovery_flow(hass, DOMAIN)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
assert hass.config_entries.async_entries(DOMAIN)[0].unique_id == "00:11:22:33:44:55"
@ -171,12 +162,10 @@ async def test_config_flow_from_dhcp_no_update(
mock_hub_refresh: AsyncMock,
) -> None:
"""Test we do not use DHCP discovery to overwrite hostname with IP in config entry."""
info = DhcpServiceInfo(
info = MockDhcpServiceInfo(
ip="1.2.3.4", hostname="webcontrol", macaddress="00:11:22:33:44:55"
)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_DHCP}, data=info
)
result = await info.start_discovery_flow(hass, DOMAIN)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {}
@ -199,12 +188,10 @@ async def test_config_flow_from_dhcp_no_update(
assert len(mock_setup_entry.mock_calls) == 1
assert hass.config_entries.async_entries(DOMAIN)[0].unique_id == "00:11:22:33:44:55"
info = DhcpServiceInfo(
info = MockDhcpServiceInfo(
ip="5.6.7.8", hostname="webcontrol", macaddress="00:11:22:33:44:55"
)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_DHCP}, data=info
)
result = await info.start_discovery_flow(hass, DOMAIN)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
assert hass.config_entries.async_entries(DOMAIN)[0].unique_id == "00:11:22:33:44:55"