Add support for USB dongles to the hardware integration (#76795)
* Add support for USB dongles to the hardware integration * Update hardware integrations * Adjust tests * Add USB discovery for SkyConnect 1.0 * Improve test coverage * Apply suggestions from code review Co-authored-by: Paulus Schoutsen <balloob@gmail.com> * Fix frozen dataclass shizzle * Adjust test Co-authored-by: Paulus Schoutsen <balloob@gmail.com>
This commit is contained in:
Erik Montnemery
GitHub
parent
fb5a67fb1f
commit
bb74730e96
23 changed files with 581 additions and 28 deletions
|
|
@ -467,6 +467,8 @@ build.json @home-assistant/supervisor
|
|||
/tests/components/homeassistant/ @home-assistant/core
|
||||
/homeassistant/components/homeassistant_alerts/ @home-assistant/core
|
||||
/tests/components/homeassistant_alerts/ @home-assistant/core
|
||||
/homeassistant/components/homeassistant_sky_connect/ @home-assistant/core
|
||||
/tests/components/homeassistant_sky_connect/ @home-assistant/core
|
||||
/homeassistant/components/homeassistant_yellow/ @home-assistant/core
|
||||
/tests/components/homeassistant_yellow/ @home-assistant/core
|
||||
/homeassistant/components/homekit/ @bdraco
|
||||
|
|
|
|||
|
|
@ -34,6 +34,7 @@ def async_info(hass: HomeAssistant) -> HardwareInfo:
|
|||
model=board,
|
||||
revision=None,
|
||||
),
|
||||
dongles=None,
|
||||
name=BOARD_NAMES.get(board, f"Unknown hardkernel Odroid model '{board}'"),
|
||||
url=None,
|
||||
)
|
||||
|
|
|
|||
|
|
@ -17,12 +17,24 @@ class BoardInfo:
|
|||
revision: str | None
|
||||
|
||||
|
||||
@dataclass
|
||||
@dataclass(frozen=True)
|
||||
class USBInfo:
|
||||
"""USB info type."""
|
||||
|
||||
vid: str
|
||||
pid: str
|
||||
serial_number: str | None
|
||||
manufacturer: str | None
|
||||
description: str | None
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class HardwareInfo:
|
||||
"""Hardware info type."""
|
||||
|
||||
name: str | None
|
||||
board: BoardInfo | None
|
||||
dongles: list[USBInfo] | None
|
||||
url: str | None
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1,35 @@
|
|||
"""The Home Assistant Sky Connect integration."""
|
||||
from __future__ import annotations
|
||||
|
||||
from homeassistant.components import usb
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.exceptions import ConfigEntryNotReady
|
||||
|
||||
|
||||
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||
"""Set up a Home Assistant Sky Connect config entry."""
|
||||
usb_info = usb.UsbServiceInfo(
|
||||
device=entry.data["device"],
|
||||
vid=entry.data["vid"],
|
||||
pid=entry.data["pid"],
|
||||
serial_number=entry.data["serial_number"],
|
||||
manufacturer=entry.data["manufacturer"],
|
||||
description=entry.data["description"],
|
||||
)
|
||||
if not usb.async_is_plugged_in(hass, entry.data):
|
||||
# The USB dongle is not plugged in
|
||||
raise ConfigEntryNotReady
|
||||
|
||||
await hass.config_entries.flow.async_init(
|
||||
"zha",
|
||||
context={"source": "usb"},
|
||||
data=usb_info,
|
||||
)
|
||||
|
||||
return True
|
||||
|
||||
|
||||
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||
"""Unload a config entry."""
|
||||
return True
|
||||
|
|
@ -0,0 +1,37 @@
|
|||
"""Config flow for the Home Assistant Sky Connect integration."""
|
||||
from __future__ import annotations
|
||||
|
||||
from homeassistant.components import usb
|
||||
from homeassistant.config_entries import ConfigFlow
|
||||
from homeassistant.data_entry_flow import FlowResult
|
||||
|
||||
from .const import DOMAIN
|
||||
|
||||
|
||||
class HomeAssistantSkyConnectConfigFlow(ConfigFlow, domain=DOMAIN):
|
||||
"""Handle a config flow for Home Assistant Sky Connect."""
|
||||
|
||||
VERSION = 1
|
||||
|
||||
async def async_step_usb(self, discovery_info: usb.UsbServiceInfo) -> FlowResult:
|
||||
"""Handle usb discovery."""
|
||||
device = discovery_info.device
|
||||
vid = discovery_info.vid
|
||||
pid = discovery_info.pid
|
||||
serial_number = discovery_info.serial_number
|
||||
manufacturer = discovery_info.manufacturer
|
||||
description = discovery_info.description
|
||||
unique_id = f"{vid}:{pid}_{serial_number}_{manufacturer}_{description}"
|
||||
if await self.async_set_unique_id(unique_id):
|
||||
self._abort_if_unique_id_configured(updates={"device": device})
|
||||
return self.async_create_entry(
|
||||
title="Home Assistant Sky Connect",
|
||||
data={
|
||||
"device": device,
|
||||
"vid": vid,
|
||||
"pid": pid,
|
||||
"serial_number": serial_number,
|
||||
"manufacturer": manufacturer,
|
||||
"description": description,
|
||||
},
|
||||
)
|
||||
|
|
@ -0,0 +1,3 @@
|
|||
"""Constants for the Home Assistant Sky Connect integration."""
|
||||
|
||||
DOMAIN = "homeassistant_sky_connect"
|
||||
|
|
@ -0,0 +1,33 @@
|
|||
"""The Home Assistant Sky Connect hardware platform."""
|
||||
from __future__ import annotations
|
||||
|
||||
from homeassistant.components.hardware.models import HardwareInfo, USBInfo
|
||||
from homeassistant.core import HomeAssistant, callback
|
||||
|
||||
from .const import DOMAIN
|
||||
|
||||
DONGLE_NAME = "Home Assistant Sky Connect"
|
||||
|
||||
|
||||
@callback
|
||||
def async_info(hass: HomeAssistant) -> HardwareInfo:
|
||||
"""Return board info."""
|
||||
entries = hass.config_entries.async_entries(DOMAIN)
|
||||
|
||||
dongles = [
|
||||
USBInfo(
|
||||
vid=entry.data["vid"],
|
||||
pid=entry.data["pid"],
|
||||
serial_number=entry.data["serial_number"],
|
||||
manufacturer=entry.data["manufacturer"],
|
||||
description=entry.data["description"],
|
||||
)
|
||||
for entry in entries
|
||||
]
|
||||
|
||||
return HardwareInfo(
|
||||
board=None,
|
||||
dongles=dongles,
|
||||
name=DONGLE_NAME,
|
||||
url=None,
|
||||
)
|
||||
|
|
@ -0,0 +1,17 @@
|
|||
{
|
||||
"domain": "homeassistant_sky_connect",
|
||||
"name": "Home Assistant Sky Connect",
|
||||
"config_flow": false,
|
||||
"documentation": "https://www.home-assistant.io/integrations/homeassistant_sky_connect",
|
||||
"dependencies": ["hardware", "usb"],
|
||||
"codeowners": ["@home-assistant/core"],
|
||||
"integration_type": "hardware",
|
||||
"usb": [
|
||||
{
|
||||
"vid": "10C4",
|
||||
"pid": "EA60",
|
||||
"description": "*skyconnect v1.0*",
|
||||
"known_devices": ["SkyConnect v1.0"]
|
||||
}
|
||||
]
|
||||
}
|
||||
|
|
@ -29,6 +29,7 @@ def async_info(hass: HomeAssistant) -> HardwareInfo:
|
|||
model=MODEL,
|
||||
revision=None,
|
||||
),
|
||||
dongles=None,
|
||||
name=BOARD_NAME,
|
||||
url=None,
|
||||
)
|
||||
|
|
|
|||
|
|
@ -49,6 +49,7 @@ def async_info(hass: HomeAssistant) -> HardwareInfo:
|
|||
model=MODELS.get(board),
|
||||
revision=None,
|
||||
),
|
||||
dongles=None,
|
||||
name=BOARD_NAMES.get(board, f"Unknown Raspberry Pi model '{board}'"),
|
||||
url=None,
|
||||
)
|
||||
|
|
|
|||
|
|
@ -1,7 +1,7 @@
|
|||
"""The USB Discovery integration."""
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Coroutine
|
||||
from collections.abc import Coroutine, Mapping
|
||||
import dataclasses
|
||||
import fnmatch
|
||||
import logging
|
||||
|
|
@ -97,6 +97,27 @@ def _fnmatch_lower(name: str | None, pattern: str) -> bool:
|
|||
return fnmatch.fnmatch(name.lower(), pattern)
|
||||
|
||||
|
||||
def _is_matching(device: USBDevice, matcher: Mapping[str, str]) -> bool:
|
||||
"""Return True if a device matches."""
|
||||
if "vid" in matcher and device.vid != matcher["vid"]:
|
||||
return False
|
||||
if "pid" in matcher and device.pid != matcher["pid"]:
|
||||
return False
|
||||
if "serial_number" in matcher and not _fnmatch_lower(
|
||||
device.serial_number, matcher["serial_number"]
|
||||
):
|
||||
return False
|
||||
if "manufacturer" in matcher and not _fnmatch_lower(
|
||||
device.manufacturer, matcher["manufacturer"]
|
||||
):
|
||||
return False
|
||||
if "description" in matcher and not _fnmatch_lower(
|
||||
device.description, matcher["description"]
|
||||
):
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
class USBDiscovery:
|
||||
"""Manage USB Discovery."""
|
||||
|
||||
|
|
@ -179,23 +200,8 @@ class USBDiscovery:
|
|||
self.seen.add(device_tuple)
|
||||
matched = []
|
||||
for matcher in self.usb:
|
||||
if "vid" in matcher and device.vid != matcher["vid"]:
|
||||
continue
|
||||
if "pid" in matcher and device.pid != matcher["pid"]:
|
||||
continue
|
||||
if "serial_number" in matcher and not _fnmatch_lower(
|
||||
device.serial_number, matcher["serial_number"]
|
||||
):
|
||||
continue
|
||||
if "manufacturer" in matcher and not _fnmatch_lower(
|
||||
device.manufacturer, matcher["manufacturer"]
|
||||
):
|
||||
continue
|
||||
if "description" in matcher and not _fnmatch_lower(
|
||||
device.description, matcher["description"]
|
||||
):
|
||||
continue
|
||||
matched.append(matcher)
|
||||
if _is_matching(device, matcher):
|
||||
matched.append(matcher)
|
||||
|
||||
if not matched:
|
||||
return
|
||||
|
|
@ -265,3 +271,14 @@ async def websocket_usb_scan(
|
|||
if not usb_discovery.observer_active:
|
||||
await usb_discovery.async_request_scan_serial()
|
||||
connection.send_result(msg["id"])
|
||||
|
||||
|
||||
@callback
|
||||
def async_is_plugged_in(hass: HomeAssistant, matcher: Mapping) -> bool:
|
||||
"""Return True is a USB device is present."""
|
||||
usb_discovery: USBDiscovery = hass.data[DOMAIN]
|
||||
for device_tuple in usb_discovery.seen:
|
||||
device = USBDevice(*device_tuple)
|
||||
if _is_matching(device, matcher):
|
||||
return True
|
||||
return False
|
||||
|
|
|
|||
|
|
@ -138,11 +138,11 @@ class ZhaFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
|
|||
)
|
||||
self._set_confirm_only()
|
||||
self.context["title_placeholders"] = {CONF_NAME: self._title}
|
||||
return await self.async_step_confirm()
|
||||
return await self.async_step_confirm_usb()
|
||||
|
||||
async def async_step_confirm(self, user_input=None):
|
||||
"""Confirm a discovery."""
|
||||
if user_input is not None:
|
||||
async def async_step_confirm_usb(self, user_input=None):
|
||||
"""Confirm a USB discovery."""
|
||||
if user_input is not None or not onboarding.async_is_onboarded(self.hass):
|
||||
auto_detected_data = await detect_radios(self._device_path)
|
||||
if auto_detected_data is None:
|
||||
# This path probably will not happen now that we have
|
||||
|
|
@ -155,7 +155,7 @@ class ZhaFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
|
|||
)
|
||||
|
||||
return self.async_show_form(
|
||||
step_id="confirm",
|
||||
step_id="confirm_usb",
|
||||
description_placeholders={CONF_NAME: self._title},
|
||||
)
|
||||
|
||||
|
|
|
|||
|
|
@ -58,6 +58,7 @@ NO_IOT_CLASS = [
|
|||
"history",
|
||||
"homeassistant",
|
||||
"homeassistant_alerts",
|
||||
"homeassistant_sky_connect",
|
||||
"homeassistant_yellow",
|
||||
"image",
|
||||
"input_boolean",
|
||||
|
|
|
|||
|
|
@ -48,6 +48,7 @@ async def test_hardware_info(hass: HomeAssistant, hass_ws_client) -> None:
|
|||
"model": "odroid-n2",
|
||||
"revision": None,
|
||||
},
|
||||
"dongles": None,
|
||||
"name": "Home Assistant Blue / Hardkernel Odroid-N2",
|
||||
"url": None,
|
||||
}
|
||||
|
|
|
|||
1
tests/components/homeassistant_sky_connect/__init__.py
Normal file
1
tests/components/homeassistant_sky_connect/__init__.py
Normal file
|
|
@ -0,0 +1 @@
|
|||
"""Tests for the Home Assistant Sky Connect integration."""
|
||||
14
tests/components/homeassistant_sky_connect/conftest.py
Normal file
14
tests/components/homeassistant_sky_connect/conftest.py
Normal file
|
|
@ -0,0 +1,14 @@
|
|||
"""Test fixtures for the Home Assistant Sky Connect integration."""
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def mock_zha():
|
||||
"""Mock the zha integration."""
|
||||
with patch(
|
||||
"homeassistant.components.zha.async_setup_entry",
|
||||
return_value=True,
|
||||
):
|
||||
yield
|
||||
147
tests/components/homeassistant_sky_connect/test_config_flow.py
Normal file
147
tests/components/homeassistant_sky_connect/test_config_flow.py
Normal file
|
|
@ -0,0 +1,147 @@
|
|||
"""Test the Home Assistant Sky Connect config flow."""
|
||||
import copy
|
||||
from unittest.mock import patch
|
||||
|
||||
from homeassistant.components import homeassistant_sky_connect, usb
|
||||
from homeassistant.components.homeassistant_sky_connect.const import DOMAIN
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.data_entry_flow import FlowResultType
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
USB_DATA = usb.UsbServiceInfo(
|
||||
device="bla_device",
|
||||
vid="bla_vid",
|
||||
pid="bla_pid",
|
||||
serial_number="bla_serial_number",
|
||||
manufacturer="bla_manufacturer",
|
||||
description="bla_description",
|
||||
)
|
||||
|
||||
|
||||
async def test_config_flow(hass: HomeAssistant) -> None:
|
||||
"""Test the config flow."""
|
||||
# mock_integration(hass, MockModule("hassio"))
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.homeassistant_sky_connect.async_setup_entry",
|
||||
return_value=True,
|
||||
) as mock_setup_entry:
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": "usb"}, data=USB_DATA
|
||||
)
|
||||
|
||||
expected_data = {
|
||||
"device": USB_DATA.device,
|
||||
"vid": USB_DATA.vid,
|
||||
"pid": USB_DATA.pid,
|
||||
"serial_number": USB_DATA.serial_number,
|
||||
"manufacturer": USB_DATA.manufacturer,
|
||||
"description": USB_DATA.description,
|
||||
}
|
||||
|
||||
assert result["type"] == FlowResultType.CREATE_ENTRY
|
||||
assert result["title"] == "Home Assistant Sky Connect"
|
||||
assert result["data"] == expected_data
|
||||
assert result["options"] == {}
|
||||
assert len(mock_setup_entry.mock_calls) == 1
|
||||
|
||||
config_entry = hass.config_entries.async_entries(DOMAIN)[0]
|
||||
assert config_entry.data == expected_data
|
||||
assert config_entry.options == {}
|
||||
assert config_entry.title == "Home Assistant Sky Connect"
|
||||
assert (
|
||||
config_entry.unique_id
|
||||
== f"{USB_DATA.vid}:{USB_DATA.pid}_{USB_DATA.serial_number}_{USB_DATA.manufacturer}_{USB_DATA.description}"
|
||||
)
|
||||
|
||||
|
||||
async def test_config_flow_unique_id(hass: HomeAssistant) -> None:
|
||||
"""Test only a single entry is allowed for a dongle."""
|
||||
# Setup an existing config entry
|
||||
config_entry = MockConfigEntry(
|
||||
data={},
|
||||
domain=DOMAIN,
|
||||
options={},
|
||||
title="Home Assistant Sky Connect",
|
||||
unique_id=f"{USB_DATA.vid}:{USB_DATA.pid}_{USB_DATA.serial_number}_{USB_DATA.manufacturer}_{USB_DATA.description}",
|
||||
)
|
||||
config_entry.add_to_hass(hass)
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.homeassistant_sky_connect.async_setup_entry",
|
||||
return_value=True,
|
||||
) as mock_setup_entry:
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": "usb"}, data=USB_DATA
|
||||
)
|
||||
|
||||
assert result["type"] == FlowResultType.ABORT
|
||||
assert result["reason"] == "already_configured"
|
||||
mock_setup_entry.assert_not_called()
|
||||
|
||||
|
||||
async def test_config_flow_multiple_entries(hass: HomeAssistant) -> None:
|
||||
"""Test multiple entries are allowed."""
|
||||
# Setup an existing config entry
|
||||
config_entry = MockConfigEntry(
|
||||
data={},
|
||||
domain=DOMAIN,
|
||||
options={},
|
||||
title="Home Assistant Sky Connect",
|
||||
unique_id=f"{USB_DATA.vid}:{USB_DATA.pid}_{USB_DATA.serial_number}_{USB_DATA.manufacturer}_{USB_DATA.description}",
|
||||
)
|
||||
config_entry.add_to_hass(hass)
|
||||
|
||||
usb_data = copy.copy(USB_DATA)
|
||||
usb_data.serial_number = "bla_serial_number_2"
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.homeassistant_sky_connect.async_setup_entry",
|
||||
return_value=True,
|
||||
):
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": "usb"}, data=usb_data
|
||||
)
|
||||
|
||||
assert result["type"] == FlowResultType.CREATE_ENTRY
|
||||
|
||||
|
||||
async def test_config_flow_update_device(hass: HomeAssistant) -> None:
|
||||
"""Test updating device path."""
|
||||
# Setup an existing config entry
|
||||
config_entry = MockConfigEntry(
|
||||
data={},
|
||||
domain=DOMAIN,
|
||||
options={},
|
||||
title="Home Assistant Sky Connect",
|
||||
unique_id=f"{USB_DATA.vid}:{USB_DATA.pid}_{USB_DATA.serial_number}_{USB_DATA.manufacturer}_{USB_DATA.description}",
|
||||
)
|
||||
config_entry.add_to_hass(hass)
|
||||
|
||||
usb_data = copy.copy(USB_DATA)
|
||||
usb_data.device = "bla_device_2"
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.homeassistant_sky_connect.async_setup_entry",
|
||||
return_value=True,
|
||||
) as mock_setup_entry:
|
||||
assert await hass.config_entries.async_setup(config_entry.entry_id)
|
||||
assert len(mock_setup_entry.mock_calls) == 1
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.homeassistant_sky_connect.async_setup_entry",
|
||||
return_value=True,
|
||||
) as mock_setup_entry, patch(
|
||||
"homeassistant.components.homeassistant_sky_connect.async_unload_entry",
|
||||
wraps=homeassistant_sky_connect.async_unload_entry,
|
||||
) as mock_unload_entry:
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": "usb"}, data=usb_data
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert result["type"] == FlowResultType.ABORT
|
||||
assert result["reason"] == "already_configured"
|
||||
assert len(mock_setup_entry.mock_calls) == 1
|
||||
assert len(mock_unload_entry.mock_calls) == 1
|
||||
85
tests/components/homeassistant_sky_connect/test_hardware.py
Normal file
85
tests/components/homeassistant_sky_connect/test_hardware.py
Normal file
|
|
@ -0,0 +1,85 @@
|
|||
"""Test the Home Assistant Sky Connect hardware platform."""
|
||||
from unittest.mock import patch
|
||||
|
||||
from homeassistant.components.homeassistant_sky_connect.const import DOMAIN
|
||||
from homeassistant.core import HomeAssistant
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
CONFIG_ENTRY_DATA = {
|
||||
"device": "bla_device",
|
||||
"vid": "bla_vid",
|
||||
"pid": "bla_pid",
|
||||
"serial_number": "bla_serial_number",
|
||||
"manufacturer": "bla_manufacturer",
|
||||
"description": "bla_description",
|
||||
}
|
||||
|
||||
CONFIG_ENTRY_DATA_2 = {
|
||||
"device": "bla_device_2",
|
||||
"vid": "bla_vid_2",
|
||||
"pid": "bla_pid_2",
|
||||
"serial_number": "bla_serial_number_2",
|
||||
"manufacturer": "bla_manufacturer_2",
|
||||
"description": "bla_description_2",
|
||||
}
|
||||
|
||||
|
||||
async def test_hardware_info(hass: HomeAssistant, hass_ws_client) -> None:
|
||||
"""Test we can get the board info."""
|
||||
# Setup the config entry
|
||||
config_entry = MockConfigEntry(
|
||||
data=CONFIG_ENTRY_DATA,
|
||||
domain=DOMAIN,
|
||||
options={},
|
||||
title="Home Assistant Sky Connect",
|
||||
unique_id="unique_1",
|
||||
)
|
||||
config_entry.add_to_hass(hass)
|
||||
config_entry_2 = MockConfigEntry(
|
||||
data=CONFIG_ENTRY_DATA_2,
|
||||
domain=DOMAIN,
|
||||
options={},
|
||||
title="Home Assistant Sky Connect",
|
||||
unique_id="unique_2",
|
||||
)
|
||||
config_entry_2.add_to_hass(hass)
|
||||
with patch(
|
||||
"homeassistant.components.homeassistant_sky_connect.usb.async_is_plugged_in",
|
||||
return_value=True,
|
||||
):
|
||||
assert await hass.config_entries.async_setup(config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
client = await hass_ws_client(hass)
|
||||
|
||||
await client.send_json({"id": 1, "type": "hardware/info"})
|
||||
msg = await client.receive_json()
|
||||
|
||||
assert msg["id"] == 1
|
||||
assert msg["success"]
|
||||
assert msg["result"] == {
|
||||
"hardware": [
|
||||
{
|
||||
"board": None,
|
||||
"dongles": [
|
||||
{
|
||||
"vid": "bla_vid",
|
||||
"pid": "bla_pid",
|
||||
"serial_number": "bla_serial_number",
|
||||
"manufacturer": "bla_manufacturer",
|
||||
"description": "bla_description",
|
||||
},
|
||||
{
|
||||
"vid": "bla_vid_2",
|
||||
"pid": "bla_pid_2",
|
||||
"serial_number": "bla_serial_number_2",
|
||||
"manufacturer": "bla_manufacturer_2",
|
||||
"description": "bla_description_2",
|
||||
},
|
||||
],
|
||||
"name": "Home Assistant Sky Connect",
|
||||
"url": None,
|
||||
}
|
||||
]
|
||||
}
|
||||
101
tests/components/homeassistant_sky_connect/test_init.py
Normal file
101
tests/components/homeassistant_sky_connect/test_init.py
Normal file
|
|
@ -0,0 +1,101 @@
|
|||
"""Test the Home Assistant Sky Connect integration."""
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
from homeassistant.components.homeassistant_sky_connect.const import DOMAIN
|
||||
from homeassistant.config_entries import ConfigEntryState
|
||||
from homeassistant.core import HomeAssistant
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
CONFIG_ENTRY_DATA = {
|
||||
"device": "bla_device",
|
||||
"vid": "bla_vid",
|
||||
"pid": "bla_pid",
|
||||
"serial_number": "bla_serial_number",
|
||||
"manufacturer": "bla_manufacturer",
|
||||
"description": "bla_description",
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"onboarded, num_entries, num_flows", ((False, 1, 0), (True, 0, 1))
|
||||
)
|
||||
async def test_setup_entry(
|
||||
hass: HomeAssistant, onboarded, num_entries, num_flows
|
||||
) -> None:
|
||||
"""Test setup of a config entry, including setup of zha."""
|
||||
# Setup the config entry
|
||||
config_entry = MockConfigEntry(
|
||||
data=CONFIG_ENTRY_DATA,
|
||||
domain=DOMAIN,
|
||||
options={},
|
||||
title="Home Assistant Sky Connect",
|
||||
)
|
||||
config_entry.add_to_hass(hass)
|
||||
with patch(
|
||||
"homeassistant.components.homeassistant_sky_connect.usb.async_is_plugged_in",
|
||||
return_value=True,
|
||||
) as mock_is_plugged_in, patch(
|
||||
"homeassistant.components.onboarding.async_is_onboarded", return_value=onboarded
|
||||
), patch(
|
||||
"zigpy_znp.zigbee.application.ControllerApplication.probe", return_value=True
|
||||
):
|
||||
assert await hass.config_entries.async_setup(config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
assert len(mock_is_plugged_in.mock_calls) == 1
|
||||
|
||||
assert len(hass.config_entries.async_entries("zha")) == num_entries
|
||||
assert len(hass.config_entries.flow.async_progress_by_handler("zha")) == num_flows
|
||||
|
||||
|
||||
async def test_setup_zha(hass: HomeAssistant) -> None:
|
||||
"""Test zha gets the right config."""
|
||||
# Setup the config entry
|
||||
config_entry = MockConfigEntry(
|
||||
data=CONFIG_ENTRY_DATA,
|
||||
domain=DOMAIN,
|
||||
options={},
|
||||
title="Home Assistant Sky Connect",
|
||||
)
|
||||
config_entry.add_to_hass(hass)
|
||||
with patch(
|
||||
"homeassistant.components.homeassistant_sky_connect.usb.async_is_plugged_in",
|
||||
return_value=True,
|
||||
) as mock_is_plugged_in, patch(
|
||||
"homeassistant.components.onboarding.async_is_onboarded", return_value=False
|
||||
), patch(
|
||||
"zigpy_znp.zigbee.application.ControllerApplication.probe", return_value=True
|
||||
):
|
||||
assert await hass.config_entries.async_setup(config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
assert len(mock_is_plugged_in.mock_calls) == 1
|
||||
|
||||
config_entry = hass.config_entries.async_entries("zha")[0]
|
||||
assert config_entry.data == {
|
||||
"device": {"baudrate": 115200, "flow_control": None, "path": "bla_device"},
|
||||
"radio_type": "znp",
|
||||
}
|
||||
assert config_entry.options == {}
|
||||
assert config_entry.title == "bla_description"
|
||||
|
||||
|
||||
async def test_setup_entry_wait_usb(hass: HomeAssistant) -> None:
|
||||
"""Test setup of a config entry when the dongle is not plugged in."""
|
||||
# Setup the config entry
|
||||
config_entry = MockConfigEntry(
|
||||
data=CONFIG_ENTRY_DATA,
|
||||
domain=DOMAIN,
|
||||
options={},
|
||||
title="Home Assistant Sky Connect",
|
||||
)
|
||||
config_entry.add_to_hass(hass)
|
||||
with patch(
|
||||
"homeassistant.components.homeassistant_sky_connect.usb.async_is_plugged_in",
|
||||
return_value=False,
|
||||
) as mock_is_plugged_in:
|
||||
assert not await hass.config_entries.async_setup(config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
assert len(mock_is_plugged_in.mock_calls) == 1
|
||||
assert config_entry.state == ConfigEntryState.SETUP_RETRY
|
||||
|
|
@ -48,6 +48,7 @@ async def test_hardware_info(hass: HomeAssistant, hass_ws_client) -> None:
|
|||
"model": "yellow",
|
||||
"revision": None,
|
||||
},
|
||||
"dongles": None,
|
||||
"name": "Home Assistant Yellow",
|
||||
"url": None,
|
||||
}
|
||||
|
|
|
|||
|
|
@ -48,6 +48,7 @@ async def test_hardware_info(hass: HomeAssistant, hass_ws_client) -> None:
|
|||
"model": "1",
|
||||
"revision": None,
|
||||
},
|
||||
"dongles": None,
|
||||
"name": "Raspberry Pi",
|
||||
"url": None,
|
||||
}
|
||||
|
|
|
|||
|
|
@ -833,3 +833,45 @@ def test_human_readable_device_name():
|
|||
assert "Silicon Labs" in name
|
||||
assert "10C4" in name
|
||||
assert "8A2A" in name
|
||||
|
||||
|
||||
async def test_async_is_plugged_in(hass, hass_ws_client):
|
||||
"""Test async_is_plugged_in."""
|
||||
new_usb = [{"domain": "test1", "vid": "3039", "pid": "3039"}]
|
||||
|
||||
mock_comports = [
|
||||
MagicMock(
|
||||
device=slae_sh_device.device,
|
||||
vid=12345,
|
||||
pid=12345,
|
||||
serial_number=slae_sh_device.serial_number,
|
||||
manufacturer=slae_sh_device.manufacturer,
|
||||
description=slae_sh_device.description,
|
||||
)
|
||||
]
|
||||
|
||||
matcher = {
|
||||
"vid": "3039",
|
||||
"pid": "3039",
|
||||
}
|
||||
|
||||
with patch("pyudev.Context", side_effect=ImportError), patch(
|
||||
"homeassistant.components.usb.async_get_usb", return_value=new_usb
|
||||
), patch("homeassistant.components.usb.comports", return_value=[]), patch.object(
|
||||
hass.config_entries.flow, "async_init"
|
||||
):
|
||||
assert await async_setup_component(hass, "usb", {"usb": {}})
|
||||
await hass.async_block_till_done()
|
||||
hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED)
|
||||
await hass.async_block_till_done()
|
||||
assert not usb.async_is_plugged_in(hass, matcher)
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.usb.comports", return_value=mock_comports
|
||||
), patch.object(hass.config_entries.flow, "async_init"):
|
||||
ws_client = await hass_ws_client(hass)
|
||||
await ws_client.send_json({"id": 1, "type": "usb/scan"})
|
||||
response = await ws_client.receive_json()
|
||||
assert response["success"]
|
||||
await hass.async_block_till_done()
|
||||
assert usb.async_is_plugged_in(hass, matcher)
|
||||
|
|
|
|||
|
|
@ -228,7 +228,7 @@ async def test_discovery_via_usb(detect_mock, hass):
|
|||
)
|
||||
await hass.async_block_till_done()
|
||||
assert result["type"] == FlowResultType.FORM
|
||||
assert result["step_id"] == "confirm"
|
||||
assert result["step_id"] == "confirm_usb"
|
||||
|
||||
with patch("homeassistant.components.zha.async_setup_entry"):
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
|
|
@ -264,7 +264,7 @@ async def test_zigate_discovery_via_usb(detect_mock, hass):
|
|||
)
|
||||
await hass.async_block_till_done()
|
||||
assert result["type"] == FlowResultType.FORM
|
||||
assert result["step_id"] == "confirm"
|
||||
assert result["step_id"] == "confirm_usb"
|
||||
|
||||
with patch("homeassistant.components.zha.async_setup_entry"):
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
|
|
@ -298,7 +298,7 @@ async def test_discovery_via_usb_no_radio(detect_mock, hass):
|
|||
)
|
||||
await hass.async_block_till_done()
|
||||
assert result["type"] == FlowResultType.FORM
|
||||
assert result["step_id"] == "confirm"
|
||||
assert result["step_id"] == "confirm_usb"
|
||||
|
||||
with patch("homeassistant.components.zha.async_setup_entry"):
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
|
|
@ -451,7 +451,7 @@ async def test_discovery_via_usb_deconz_ignored(detect_mock, hass):
|
|||
await hass.async_block_till_done()
|
||||
|
||||
assert result["type"] == FlowResultType.FORM
|
||||
assert result["step_id"] == "confirm"
|
||||
assert result["step_id"] == "confirm_usb"
|
||||
|
||||
|
||||
@patch("zigpy_znp.zigbee.application.ControllerApplication.probe", return_value=True)
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue