jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-23 16:57:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add discovery support to steamist (#63707)

Browse Source
This commit is contained in:
J. Nick Koston 2022-01-09 10:34:50 -10:00 committed by GitHub
parent 0efdc7fa65
commit 96aa623d2a
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
15 changed files with 843 additions and 30 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

41
homeassistant/components/steamist/__init__.py
View File

@ -1,27 +1,60 @@
"""The Steamist integration."""
from __future__ import annotations
from datetime import timedelta
from typing import Any
from aiosteamist import Steamist
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_HOST, Platform
from homeassistant.const import CONF_HOST, CONF_NAME, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.event import async_track_time_interval
from homeassistant.helpers.typing import ConfigType
from .const import DOMAIN
from .const import DISCOVER_SCAN_TIMEOUT, DISCOVERY, DOMAIN, STARTUP_SCAN_TIMEOUT
from .coordinator import SteamistDataUpdateCoordinator
from .discovery import (
async_discover_device,
async_discover_devices,
async_get_discovery,
async_trigger_discovery,
async_update_entry_from_discovery,
)
PLATFORMS: list[str] = [Platform.SENSOR, Platform.SWITCH]
DISCOVERY_INTERVAL = timedelta(minutes=15)
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the flux_led component."""
domain_data = hass.data.setdefault(DOMAIN, {})
domain_data[DISCOVERY] = await async_discover_devices(hass, STARTUP_SCAN_TIMEOUT)
async def _async_discovery(*_: Any) -> None:
async_trigger_discovery(
hass, await async_discover_devices(hass, DISCOVER_SCAN_TIMEOUT)
)
async_trigger_discovery(hass, domain_data[DISCOVERY])
async_track_time_interval(hass, _async_discovery, DISCOVERY_INTERVAL)
return True
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up Steamist from a config entry."""
host = entry.data[CONF_HOST]
coordinator = SteamistDataUpdateCoordinator(
hass,
Steamist(entry.data[CONF_HOST], async_get_clientsession(hass)),
entry.data[CONF_HOST],
Steamist(host, async_get_clientsession(hass)),
host,
entry.data.get(CONF_NAME), # Only found from discovery
)
await coordinator.async_config_entry_first_refresh()
if not async_get_discovery(hass, host):
if discovery := await async_discover_device(hass, host):
async_update_entry_from_discovery(hass, entry, discovery)
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = coordinator
hass.config_entries.async_setup_platforms(entry, PLATFORMS)
return True

151
homeassistant/components/steamist/config_flow.py
View File

@ -5,14 +5,25 @@ import logging
from typing import Any
from aiosteamist import Steamist
from discovery30303 import Device30303, normalize_mac
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import CONF_HOST
from homeassistant.components import dhcp
from homeassistant.const import CONF_DEVICE, CONF_HOST, CONF_NAME
from homeassistant.core import callback
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.typing import DiscoveryInfoType
from .const import CONNECTION_EXCEPTIONS, DOMAIN
from .const import CONF_MODEL, CONNECTION_EXCEPTIONS, DISCOVER_SCAN_TIMEOUT, DOMAIN
from .discovery import (
async_discover_device,
async_discover_devices,
async_is_steamist_device,
async_update_entry_from_discovery,
)
_LOGGER = logging.getLogger(__name__)
@ -22,6 +33,124 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
VERSION = 1
def __init__(self) -> None:
"""Initialize the config flow."""
self._discovered_devices: dict[str, Device30303] = {}
self._discovered_device: Device30303 | None = None
async def async_step_dhcp(self, discovery_info: dhcp.DhcpServiceInfo) -> FlowResult:
"""Handle discovery via dhcp."""
self._discovered_device = Device30303(
ipaddress=discovery_info.ip,
name="",
mac=normalize_mac(discovery_info.macaddress),
hostname=discovery_info.hostname,
)
return await self._async_handle_discovery()
async def async_step_discovery(
self, discovery_info: DiscoveryInfoType
) -> FlowResult:
"""Handle discovery."""
self._discovered_device = Device30303(
ipaddress=discovery_info["ipaddress"],
name=discovery_info["name"],
mac=discovery_info["mac"],
hostname=discovery_info["hostname"],
)
return await self._async_handle_discovery()
async def _async_handle_discovery(self) -> FlowResult:
"""Handle any discovery."""
device = self._discovered_device
assert device is not None
mac_address = device.mac
mac = dr.format_mac(mac_address)
host = device.ipaddress
await self.async_set_unique_id(mac)
for entry in self._async_current_entries(include_ignore=False):
if entry.unique_id == mac or entry.data[CONF_HOST] == host:
if async_update_entry_from_discovery(self.hass, entry, device):
self.hass.async_create_task(
self.hass.config_entries.async_reload(entry.entry_id)
)
return self.async_abort(reason="already_configured")
self.context[CONF_HOST] = host
for progress in self._async_in_progress():
if progress.get("context", {}).get(CONF_HOST) == host:
return self.async_abort(reason="already_in_progress")
if not device.name:
discovery = await async_discover_device(self.hass, device.ipaddress)
if not discovery:
return self.async_abort(reason="cannot_connect")
self._discovered_device = discovery
assert self._discovered_device is not None
if not async_is_steamist_device(self._discovered_device):
return self.async_abort(reason="not_steamist_device")
return await self.async_step_discovery_confirm()
async def async_step_discovery_confirm(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Confirm discovery."""
assert self._discovered_device is not None
device = self._discovered_device
if user_input is not None:
return self._async_create_entry_from_device(self._discovered_device)
self._set_confirm_only()
placeholders = {
"name": device.name,
"ipaddress": device.ipaddress,
}
self.context["title_placeholders"] = placeholders
return self.async_show_form(
step_id="discovery_confirm", description_placeholders=placeholders
)
@callback
def _async_create_entry_from_device(self, device: Device30303) -> FlowResult:
"""Create a config entry from a device."""
self._async_abort_entries_match({CONF_HOST: device.ipaddress})
data = {CONF_HOST: device.ipaddress, CONF_NAME: device.name}
if device.hostname:
data[CONF_MODEL] = device.hostname.split("-", maxsplit=1)[0]
return self.async_create_entry(
title=device.name,
data=data,
)
async def async_step_pick_device(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle the step to pick discovered device."""
if user_input is not None:
mac = user_input[CONF_DEVICE]
await self.async_set_unique_id(mac, raise_on_progress=False)
device = self._discovered_devices[mac]
return self._async_create_entry_from_device(device)
current_unique_ids = self._async_current_ids()
current_hosts = {
entry.data[CONF_HOST]
for entry in self._async_current_entries(include_ignore=False)
}
self._discovered_devices = {
dr.format_mac(device.mac): device
for device in await async_discover_devices(self.hass, DISCOVER_SCAN_TIMEOUT)
}
devices_name = {
mac: f"{device.name} ({device.ipaddress})"
for mac, device in self._discovered_devices.items()
if mac not in current_unique_ids and device.ipaddress not in current_hosts
}
# Check if there is at least one device
if not devices_name:
return self.async_abort(reason="no_devices_found")
return self.async_show_form(
step_id="pick_device",
data_schema=vol.Schema({vol.Required(CONF_DEVICE): vol.In(devices_name)}),
)
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
@ -29,24 +158,24 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
errors = {}
if user_input is not None:
if not (host := user_input[CONF_HOST]):
return await self.async_step_pick_device()
websession = async_get_clientsession(self.hass)
try:
await Steamist(
user_input[CONF_HOST],
async_get_clientsession(self.hass),
).async_get_status()
await Steamist(host, websession).async_get_status()
except CONNECTION_EXCEPTIONS:
errors["base"] = "cannot_connect"
except Exception: # pylint: disable=broad-except
_LOGGER.exception("Unexpected exception")
errors["base"] = "unknown"
else:
self._async_abort_entries_match({CONF_HOST: user_input[CONF_HOST]})
return self.async_create_entry(
title=user_input[CONF_HOST], data=user_input
)
if discovery := await async_discover_device(self.hass, host):
return self._async_create_entry_from_device(discovery)
self._async_abort_entries_match({CONF_HOST: host})
return self.async_create_entry(title=host, data=user_input)
return self.async_show_form(
step_id="user",
data_schema=vol.Schema({vol.Required(CONF_HOST): str}),
data_schema=vol.Schema({vol.Optional(CONF_HOST, default=""): str}),
errors=errors,
)

7
homeassistant/components/steamist/const.py
View File

@ -7,3 +7,10 @@ import aiohttp
DOMAIN = "steamist"
CONNECTION_EXCEPTIONS = (asyncio.TimeoutError, aiohttp.ClientError)
CONF_MODEL = "model"
STARTUP_SCAN_TIMEOUT = 5
DISCOVER_SCAN_TIMEOUT = 10
DISCOVERY = "discovery"

2
homeassistant/components/steamist/coordinator.py
View File

@ -20,9 +20,11 @@ class SteamistDataUpdateCoordinator(DataUpdateCoordinator[SteamistStatus]):
hass: HomeAssistant,
client: Steamist,
host: str,
device_name: str | None,
) -> None:
"""Initialize DataUpdateCoordinator to gather data for specific steamist."""
self.client = client
self.device_name = device_name
super().__init__(
hass,
_LOGGER,

136
homeassistant/components/steamist/discovery.py Normal file
View File

@ -0,0 +1,136 @@
"""The Steamist integration discovery."""
from __future__ import annotations
import asyncio
import logging
from typing import Any
from discovery30303 import AIODiscovery30303, Device30303
from homeassistant import config_entries
from homeassistant.components import network
from homeassistant.const import CONF_NAME
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import device_registry as dr
from homeassistant.util.network import is_ip_address
from .const import CONF_MODEL, DISCOVER_SCAN_TIMEOUT, DISCOVERY, DOMAIN
_LOGGER = logging.getLogger(__name__)
MODEL_450_HOSTNAME_PREFIX = "MY450-"
MODEL_550_HOSTNAME_PREFIX = "MY550-"
@callback
def async_is_steamist_device(device: Device30303) -> bool:
"""Check if a 30303 discovery is a steamist device."""
return device.hostname.startswith(
MODEL_450_HOSTNAME_PREFIX
) or device.hostname.startswith(MODEL_550_HOSTNAME_PREFIX)
@callback
def async_update_entry_from_discovery(
hass: HomeAssistant,
entry: config_entries.ConfigEntry,
device: Device30303,
) -> bool:
"""Update a config entry from a discovery."""
data_updates: dict[str, Any] = {}
updates: dict[str, Any] = {}
if not entry.unique_id:
updates["unique_id"] = dr.format_mac(device.mac)
if not entry.data.get(CONF_NAME) or is_ip_address(entry.data[CONF_NAME]):
updates["title"] = data_updates[CONF_NAME] = device.name
if not entry.data.get(CONF_MODEL) and "-" in device.hostname:
data_updates[CONF_MODEL] = device.hostname.split("-", maxsplit=1)[0]
if data_updates:
updates["data"] = {**entry.data, **data_updates}
if updates:
return hass.config_entries.async_update_entry(entry, **updates)
return False
async def async_discover_devices(
hass: HomeAssistant, timeout: int, address: str | None = None
) -> list[Device30303]:
"""Discover devices."""
if address:
targets = [address]
else:
targets = [
str(address)
for address in await network.async_get_ipv4_broadcast_addresses(hass)
]
scanner = AIODiscovery30303()
for idx, discovered in enumerate(
await asyncio.gather(
*[
scanner.async_scan(timeout=timeout, address=address)
for address in targets
],
return_exceptions=True,
)
):
if isinstance(discovered, Exception):
_LOGGER.debug("Scanning %s failed with error: %s", targets[idx], discovered)
continue
_LOGGER.debug("Found devices: %s", scanner.found_devices)
if not address:
return [
device
for device in scanner.found_devices
if async_is_steamist_device(device)
]
return [device for device in scanner.found_devices if device.ipaddress == address]
@callback
def async_find_discovery_by_ip(
discoveries: list[Device30303], host: str
) -> Device30303 | None:
"""Search a list of discoveries for one with a matching ip."""
for discovery in discoveries:
if discovery.ipaddress == host:
return discovery
return None
async def async_discover_device(hass: HomeAssistant, host: str) -> Device30303 | None:
"""Direct discovery to a single ip instead of broadcast."""
return async_find_discovery_by_ip(
await async_discover_devices(hass, DISCOVER_SCAN_TIMEOUT, host), host
)
@callback
def async_get_discovery(hass: HomeAssistant, host: str) -> Device30303 | None:
"""Check if a device was already discovered via a broadcast discovery."""
discoveries: list[Device30303] = hass.data[DOMAIN][DISCOVERY]
return async_find_discovery_by_ip(discoveries, host)
@callback
def async_trigger_discovery(
hass: HomeAssistant,
discovered_devices: list[Device30303],
) -> None:
"""Trigger config flows for discovered devices."""
for device in discovered_devices:
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_DISCOVERY},
data={
"ipaddress": device.ipaddress,
"name": device.name,
"mac": device.mac,
"hostname": device.hostname,
},
)
)

2
homeassistant/components/steamist/entity.py
View File

@ -24,6 +24,8 @@ class SteamistEntity(CoordinatorEntity, Entity):
"""Initialize the entity."""
super().__init__(coordinator)
self.entity_description = description
if coordinator.device_name:
self._attr_name = f"{coordinator.device_name} {description.name}"
self._attr_unique_id = f"{entry.entry_id}_{description.key}"
@property

11
homeassistant/components/steamist/manifest.json
View File

@ -2,8 +2,15 @@
"domain": "steamist",
"name": "Steamist",
"config_flow": true,
"dependencies": ["network"],
"documentation": "https://www.home-assistant.io/integrations/steamist",
"requirements": ["aiosteamist==0.3.1"],
"requirements": ["aiosteamist==0.3.1", "discovery30303==0.2.1"],
"codeowners": ["@bdraco"],
"iot_class": "local_polling"
"iot_class": "local_polling",
"dhcp": [
{
"macaddress": "001E0C*",
"hostname": "my[45]50*"
}
]
}

20
homeassistant/components/steamist/strings.json
View File

@ -1,18 +1,32 @@
{
"config": {
"flow_title": "{name} ({ipaddress})",
"step": {
"user": {
"description": "If you leave the host empty, discovery will be used to find devices.",
"data": {
"host": "[%key:common::config_flow::data::host%]"
}
}
},
"pick_device": {
"data": {
"device": "Device"
}
},
"discovery_confirm": {
"description": "Do you want to setup {name} ({ipaddress})?"
}
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
"already_in_progress": "[%key:common::config_flow::abort::already_in_progress%]",
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"no_devices_found": "[%key:common::config_flow::abort::no_devices_found%]",
"not_steamist_device": "Not a steamist device"
}
}
}
}

18
homeassistant/components/steamist/translations/en.json
View File

@ -1,17 +1,31 @@
{
"config": {
"abort": {
"already_configured": "Device is already configured"
"already_configured": "Device is already configured",
"already_in_progress": "Configuration flow is already in progress",
"cannot_connect": "Failed to connect",
"no_devices_found": "No devices found on the network",
"not_steamist_device": "Not a steamist device"
},
"error": {
"cannot_connect": "Failed to connect",
"unknown": "Unexpected error"
},
"flow_title": "{name} ({ipaddress})",
"step": {
"discovery_confirm": {
"description": "Do you want to setup {name} ({ipaddress})?"
},
"pick_device": {
"data": {
"device": "Device"
}
},
"user": {
"data": {
"host": "Host"
}
},
"description": "If you leave the host empty, discovery will be used to find devices."
}
}
}

5
homeassistant/generated/dhcp.py
View File

@ -335,6 +335,11 @@ DHCP = [
"hostname": "squeezebox*",
"macaddress": "000420*"
},
{
"domain": "steamist",
"macaddress": "001E0C*",
"hostname": "my[45]50*"
},
{
"domain": "tado",
"hostname": "tado*"

3
requirements_all.txt
View File

@ -565,6 +565,9 @@ discogs_client==2.3.0
# homeassistant.components.discord
discord.py==1.7.3
# homeassistant.components.steamist
discovery30303==0.2.1
# homeassistant.components.digitalloggers
dlipower==0.7.165

3
requirements_test_all.txt
View File

@ -366,6 +366,9 @@ devolo-plc-api==0.6.3
# homeassistant.components.directv
directv==0.4.0
# homeassistant.components.steamist
discovery30303==0.2.1
# homeassistant.components.doorbird
doorbirdpy==2.1.0

60
tests/components/steamist/__init__.py
View File

@ -5,12 +5,14 @@ from contextlib import contextmanager
from unittest.mock import AsyncMock, MagicMock, patch
from aiosteamist import Steamist, SteamistStatus
from discovery30303 import AIODiscovery30303, Device30303
from homeassistant.components import steamist
from homeassistant.components.steamist.const import DOMAIN
from homeassistant.components.steamist.const import CONF_MODEL, DOMAIN
from homeassistant.config_entries import ConfigEntry, ConfigEntryState
from homeassistant.const import CONF_HOST
from homeassistant.const import CONF_HOST, CONF_NAME
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr
from homeassistant.setup import async_setup_component
from tests.common import MockConfigEntry
@ -21,6 +23,41 @@ MOCK_ASYNC_GET_STATUS_INACTIVE = SteamistStatus(
MOCK_ASYNC_GET_STATUS_ACTIVE = SteamistStatus(
temp=102, temp_units="F", minutes_remain=14, active=True
)
DEVICE_IP_ADDRESS = "127.0.0.1"
DEVICE_NAME = "Master Bath"
DEVICE_MAC_ADDRESS = "AA:BB:CC:DD:EE:FF"
DEVICE_HOSTNAME = "MY450-EEFF"
FORMATTED_MAC_ADDRESS = dr.format_mac(DEVICE_MAC_ADDRESS)
DEVICE_MODEL = "MY450"
DEVICE_30303 = Device30303(
ipaddress=DEVICE_IP_ADDRESS,
name=DEVICE_NAME,
mac=DEVICE_MAC_ADDRESS,
hostname=DEVICE_HOSTNAME,
)
DEVICE_30303_NOT_STEAMIST = Device30303(
ipaddress=DEVICE_IP_ADDRESS,
name=DEVICE_NAME,
mac=DEVICE_MAC_ADDRESS,
hostname="not_steamist",
)
DISCOVERY_30303 = {
"ipaddress": DEVICE_IP_ADDRESS,
"name": DEVICE_NAME,
"mac": DEVICE_MAC_ADDRESS,
"hostname": DEVICE_HOSTNAME,
}
DISCOVERY_30303_NOT_STEAMIST = {
"ipaddress": DEVICE_IP_ADDRESS,
"name": DEVICE_NAME,
"mac": DEVICE_MAC_ADDRESS,
"hostname": "not_steamist",
}
DEFAULT_ENTRY_DATA = {
CONF_HOST: DEVICE_IP_ADDRESS,
CONF_NAME: DEVICE_NAME,
CONF_MODEL: DEVICE_MODEL,
}
async def _async_setup_entry_with_status(
@ -59,3 +96,22 @@ def _patch_status(status: SteamistStatus, client: Steamist | None = None):
yield
return _patcher()
def _patch_discovery(device=None, no_device=False):
mock_aio_discovery = MagicMock(auto_spec=AIODiscovery30303)
if no_device:
mock_aio_discovery.async_scan = AsyncMock(side_effect=OSError)
else:
mock_aio_discovery.async_scan = AsyncMock()
mock_aio_discovery.found_devices = [] if no_device else [device or DEVICE_30303]
@contextmanager
def _patcher():
with patch(
"homeassistant.components.steamist.discovery.AIODiscovery30303",
return_value=mock_aio_discovery,
):
yield
return _patcher()

323
tests/components/steamist/test_config_flow.py
View File

@ -2,10 +2,43 @@
import asyncio
from unittest.mock import patch
import pytest
from homeassistant import config_entries
from homeassistant.components import dhcp
from homeassistant.components.steamist.const import DOMAIN
from homeassistant.const import CONF_DEVICE, CONF_HOST
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import RESULT_TYPE_CREATE_ENTRY, RESULT_TYPE_FORM
from homeassistant.data_entry_flow import (
RESULT_TYPE_ABORT,
RESULT_TYPE_CREATE_ENTRY,
RESULT_TYPE_FORM,
)
from . import (
DEFAULT_ENTRY_DATA,
DEVICE_30303_NOT_STEAMIST,
DEVICE_HOSTNAME,
DEVICE_IP_ADDRESS,
DEVICE_MAC_ADDRESS,
DEVICE_NAME,
DISCOVERY_30303,
FORMATTED_MAC_ADDRESS,
MOCK_ASYNC_GET_STATUS_INACTIVE,
_patch_discovery,
_patch_status,
)
from tests.common import MockConfigEntry
MODULE = "homeassistant.components.steamist"
DHCP_DISCOVERY = dhcp.DhcpServiceInfo(
hostname=DEVICE_HOSTNAME,
ip=DEVICE_IP_ADDRESS,
macaddress=DEVICE_MAC_ADDRESS,
)
async def test_form(hass: HomeAssistant) -> None:
@ -16,7 +49,7 @@ async def test_form(hass: HomeAssistant) -> None:
assert result["type"] == RESULT_TYPE_FORM
assert result["errors"] == {}
with patch(
with _patch_discovery(no_device=True), patch(
"homeassistant.components.steamist.config_flow.Steamist.async_get_status"
), patch(
"homeassistant.components.steamist.async_setup_entry",
@ -38,6 +71,34 @@ async def test_form(hass: HomeAssistant) -> None:
assert len(mock_setup_entry.mock_calls) == 1
async def test_form_with_discovery(hass: HomeAssistant) -> None:
"""Test we can also discovery the device during manual setup."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == RESULT_TYPE_FORM
assert result["errors"] == {}
with _patch_discovery(), patch(
"homeassistant.components.steamist.config_flow.Steamist.async_get_status"
), patch(
"homeassistant.components.steamist.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"host": "127.0.0.1",
},
)
await hass.async_block_till_done()
assert result2["type"] == RESULT_TYPE_CREATE_ENTRY
assert result2["title"] == DEVICE_NAME
assert result2["data"] == DEFAULT_ENTRY_DATA
assert len(mock_setup_entry.mock_calls) == 1
async def test_form_cannot_connect(hass: HomeAssistant) -> None:
"""Test we handle cannot connect error."""
result = await hass.config_entries.flow.async_init(
@ -78,3 +139,261 @@ async def test_form_unknown_exception(hass: HomeAssistant) -> None:
assert result2["type"] == RESULT_TYPE_FORM
assert result2["errors"] == {"base": "unknown"}
async def test_discovery(hass: HomeAssistant) -> None:
"""Test setting up discovery."""
with _patch_discovery(), _patch_status(MOCK_ASYNC_GET_STATUS_INACTIVE):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
await hass.async_block_till_done()
assert result["type"] == RESULT_TYPE_FORM
assert result["step_id"] == "user"
assert not result["errors"]
result2 = await hass.config_entries.flow.async_configure(result["flow_id"], {})
await hass.async_block_till_done()
assert result2["type"] == RESULT_TYPE_FORM
assert result2["step_id"] == "pick_device"
assert not result2["errors"]
# test we can try again
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == RESULT_TYPE_FORM
assert result["step_id"] == "user"
assert not result["errors"]
result2 = await hass.config_entries.flow.async_configure(result["flow_id"], {})
await hass.async_block_till_done()
assert result2["type"] == "form"
assert result2["step_id"] == "pick_device"
assert not result2["errors"]
with _patch_discovery(), _patch_status(MOCK_ASYNC_GET_STATUS_INACTIVE), patch(
f"{MODULE}.async_setup", return_value=True
) as mock_setup, patch(
f"{MODULE}.async_setup_entry", return_value=True
) as mock_setup_entry:
result3 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_DEVICE: FORMATTED_MAC_ADDRESS},
)
await hass.async_block_till_done()
assert result3["type"] == RESULT_TYPE_CREATE_ENTRY
assert result3["title"] == DEVICE_NAME
assert result3["data"] == DEFAULT_ENTRY_DATA
mock_setup.assert_called_once()
mock_setup_entry.assert_called_once()
# ignore configured devices
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == RESULT_TYPE_FORM
assert result["step_id"] == "user"
assert not result["errors"]
with _patch_discovery(), _patch_status(MOCK_ASYNC_GET_STATUS_INACTIVE):
result2 = await hass.config_entries.flow.async_configure(result["flow_id"], {})
await hass.async_block_till_done()
assert result2["type"] == RESULT_TYPE_ABORT
assert result2["reason"] == "no_devices_found"
async def test_discovered_by_discovery_and_dhcp(hass: HomeAssistant) -> None:
"""Test we get the form with discovery and abort for dhcp source when we get both."""
with _patch_discovery(), _patch_status(MOCK_ASYNC_GET_STATUS_INACTIVE):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_DISCOVERY},
data=DISCOVERY_30303,
)
await hass.async_block_till_done()
assert result["type"] == RESULT_TYPE_FORM
assert result["errors"] is None
with _patch_discovery(), _patch_status(MOCK_ASYNC_GET_STATUS_INACTIVE):
result2 = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_DHCP},
data=DHCP_DISCOVERY,
)
await hass.async_block_till_done()
assert result2["type"] == RESULT_TYPE_ABORT
assert result2["reason"] == "already_in_progress"
with _patch_discovery(), _patch_status(MOCK_ASYNC_GET_STATUS_INACTIVE):
result3 = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_DHCP},
data=dhcp.DhcpServiceInfo(
hostname="any",
ip=DEVICE_IP_ADDRESS,
macaddress="00:00:00:00:00:00",
),
)
await hass.async_block_till_done()
assert result3["type"] == RESULT_TYPE_ABORT
assert result3["reason"] == "already_in_progress"
async def test_discovered_by_discovery(hass: HomeAssistant) -> None:
"""Test we can setup when discovered from discovery."""
with _patch_discovery(), _patch_status(MOCK_ASYNC_GET_STATUS_INACTIVE):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_DISCOVERY},
data=DISCOVERY_30303,
)
await hass.async_block_till_done()
assert result["type"] == RESULT_TYPE_FORM
assert result["errors"] is None
with _patch_discovery(), _patch_status(MOCK_ASYNC_GET_STATUS_INACTIVE), patch(
f"{MODULE}.async_setup", return_value=True
) as mock_async_setup, patch(
f"{MODULE}.async_setup_entry", return_value=True
) as mock_async_setup_entry:
result2 = await hass.config_entries.flow.async_configure(result["flow_id"], {})
await hass.async_block_till_done()
assert result2["type"] == RESULT_TYPE_CREATE_ENTRY
assert result2["data"] == DEFAULT_ENTRY_DATA
assert mock_async_setup.called
assert mock_async_setup_entry.called
async def test_discovered_by_dhcp(hass: HomeAssistant) -> None:
"""Test we can setup when discovered from dhcp."""
with _patch_discovery(), _patch_status(MOCK_ASYNC_GET_STATUS_INACTIVE):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_DHCP},
data=DHCP_DISCOVERY,
)
await hass.async_block_till_done()
assert result["type"] == RESULT_TYPE_FORM
assert result["errors"] is None
with _patch_discovery(), _patch_status(MOCK_ASYNC_GET_STATUS_INACTIVE), patch(
f"{MODULE}.async_setup", return_value=True
) as mock_async_setup, patch(
f"{MODULE}.async_setup_entry", return_value=True
) as mock_async_setup_entry:
result2 = await hass.config_entries.flow.async_configure(result["flow_id"], {})
await hass.async_block_till_done()
assert result2["type"] == RESULT_TYPE_CREATE_ENTRY
assert result2["data"] == DEFAULT_ENTRY_DATA
assert mock_async_setup.called
assert mock_async_setup_entry.called
async def test_discovered_by_dhcp_discovery_fails(hass: HomeAssistant) -> None:
"""Test we can setup when discovered from dhcp but then we cannot get the device name."""
with _patch_discovery(no_device=True), _patch_status(
MOCK_ASYNC_GET_STATUS_INACTIVE
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_DHCP},
data=DHCP_DISCOVERY,
)
await hass.async_block_till_done()
assert result["type"] == RESULT_TYPE_ABORT
assert result["reason"] == "cannot_connect"
async def test_discovered_by_dhcp_discovery_finds_non_steamist_device(
hass: HomeAssistant,
) -> None:
"""Test we can setup when discovered from dhcp but its not a steamist device."""
with _patch_discovery(device=DEVICE_30303_NOT_STEAMIST), _patch_status(
MOCK_ASYNC_GET_STATUS_INACTIVE
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_DHCP},
data=DHCP_DISCOVERY,
)
await hass.async_block_till_done()
assert result["type"] == RESULT_TYPE_ABORT
assert result["reason"] == "not_steamist_device"
@pytest.mark.parametrize(
"source, data",
[
(config_entries.SOURCE_DHCP, DHCP_DISCOVERY),
(config_entries.SOURCE_DISCOVERY, DISCOVERY_30303),
],
)
async def test_discovered_by_dhcp_or_discovery_adds_missing_unique_id(
hass, source, data
):
"""Test we can setup when discovered from dhcp or discovery and add a missing unique id."""
config_entry = MockConfigEntry(domain=DOMAIN, data={CONF_HOST: DEVICE_IP_ADDRESS})
config_entry.add_to_hass(hass)
with _patch_discovery(), _patch_status(MOCK_ASYNC_GET_STATUS_INACTIVE), patch(
f"{MODULE}.async_setup", return_value=True
) as mock_setup, patch(
f"{MODULE}.async_setup_entry", return_value=True
) as mock_setup_entry:
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": source}, data=data
)
await hass.async_block_till_done()
assert result["type"] == RESULT_TYPE_ABORT
assert result["reason"] == "already_configured"
assert config_entry.unique_id == FORMATTED_MAC_ADDRESS
assert mock_setup.called
assert mock_setup_entry.called
@pytest.mark.parametrize(
"source, data",
[
(config_entries.SOURCE_DHCP, DHCP_DISCOVERY),
(config_entries.SOURCE_DISCOVERY, DISCOVERY_30303),
],
)
async def test_discovered_by_dhcp_or_discovery_existing_unique_id_does_not_reload(
hass, source, data
):
"""Test we can setup when discovered from dhcp or discovery and it does not reload."""
config_entry = MockConfigEntry(
domain=DOMAIN, data=DEFAULT_ENTRY_DATA, unique_id=FORMATTED_MAC_ADDRESS
)
config_entry.add_to_hass(hass)
with _patch_discovery(), _patch_status(MOCK_ASYNC_GET_STATUS_INACTIVE), patch(
f"{MODULE}.async_setup", return_value=True
) as mock_setup, patch(
f"{MODULE}.async_setup_entry", return_value=True
) as mock_setup_entry:
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": source}, data=data
)
await hass.async_block_till_done()
assert result["type"] == RESULT_TYPE_ABORT
assert result["reason"] == "already_configured"
assert not mock_setup.called
assert not mock_setup_entry.called

91
tests/components/steamist/test_init.py
View File

@ -2,18 +2,41 @@
from __future__ import annotations
import asyncio
from unittest.mock import patch
from unittest.mock import AsyncMock, MagicMock, patch
from discovery30303 import AIODiscovery30303
import pytest
from homeassistant.components import steamist
from homeassistant.components.steamist.const import DOMAIN
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import CONF_HOST
from homeassistant.const import CONF_HOST, CONF_NAME
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
from homeassistant.util.dt import utcnow
from . import MOCK_ASYNC_GET_STATUS_ACTIVE, _async_setup_entry_with_status
from . import (
DEFAULT_ENTRY_DATA,
DEVICE_30303,
DEVICE_IP_ADDRESS,
DEVICE_NAME,
FORMATTED_MAC_ADDRESS,
MOCK_ASYNC_GET_STATUS_ACTIVE,
_async_setup_entry_with_status,
_patch_status,
)
from tests.common import MockConfigEntry
from tests.common import MockConfigEntry, async_fire_time_changed
@pytest.fixture
def mock_single_broadcast_address():
"""Mock network's async_async_get_ipv4_broadcast_addresses."""
with patch(
"homeassistant.components.network.async_get_ipv4_broadcast_addresses",
return_value={"10.255.255.255"},
):
yield
async def test_config_entry_reload(hass: HomeAssistant) -> None:
@ -40,3 +63,63 @@ async def test_config_entry_retry_later(hass: HomeAssistant) -> None:
await async_setup_component(hass, steamist.DOMAIN, {steamist.DOMAIN: {}})
await hass.async_block_till_done()
assert config_entry.state == ConfigEntryState.SETUP_RETRY
async def test_config_entry_fills_unique_id_with_directed_discovery(
hass: HomeAssistant,
) -> None:
"""Test that the unique id is added if its missing via directed (not broadcast) discovery."""
config_entry = MockConfigEntry(
domain=DOMAIN, data={CONF_HOST: DEVICE_IP_ADDRESS}, unique_id=None
)
config_entry.add_to_hass(hass)
last_address = None
async def _async_scan(*args, address=None, **kwargs):
# Only return discovery results when doing directed discovery
nonlocal last_address
last_address = address
@property
def found_devices(self):
nonlocal last_address
return [DEVICE_30303] if last_address == DEVICE_IP_ADDRESS else []
mock_aio_discovery = MagicMock(auto_spec=AIODiscovery30303)
mock_aio_discovery.async_scan = _async_scan
type(mock_aio_discovery).found_devices = found_devices
with _patch_status(MOCK_ASYNC_GET_STATUS_ACTIVE), patch(
"homeassistant.components.steamist.discovery.AIODiscovery30303",
return_value=mock_aio_discovery,
):
await async_setup_component(hass, steamist.DOMAIN, {steamist.DOMAIN: {}})
await hass.async_block_till_done()
assert config_entry.state == ConfigEntryState.LOADED
assert config_entry.unique_id == FORMATTED_MAC_ADDRESS
assert config_entry.data[CONF_NAME] == DEVICE_NAME
assert config_entry.title == DEVICE_NAME
@pytest.mark.usefixtures("mock_single_broadcast_address")
async def test_discovery_happens_at_interval(hass: HomeAssistant) -> None:
"""Test that discovery happens at interval."""
config_entry = MockConfigEntry(
domain=DOMAIN, data=DEFAULT_ENTRY_DATA, unique_id=FORMATTED_MAC_ADDRESS
)
config_entry.add_to_hass(hass)
mock_aio_discovery = MagicMock(auto_spec=AIODiscovery30303)
mock_aio_discovery.async_scan = AsyncMock()
with patch(
"homeassistant.components.steamist.discovery.AIODiscovery30303",
return_value=mock_aio_discovery,
), _patch_status(MOCK_ASYNC_GET_STATUS_ACTIVE):
await async_setup_component(hass, steamist.DOMAIN, {steamist.DOMAIN: {}})
await hass.async_block_till_done()
assert len(mock_aio_discovery.async_scan.mock_calls) == 2
async_fire_time_changed(hass, utcnow() + steamist.DISCOVERY_INTERVAL)
await hass.async_block_till_done()
assert len(mock_aio_discovery.async_scan.mock_calls) == 3