Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add shelly ble scanner support (#82007)

pull/82151/head
J. Nick Koston 2022-11-15 12:34:45 -06:00 committed by GitHub
parent 7932864e00
commit 435fc23737
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
19 changed files with 694 additions and 49 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
CODEOWNERS
View File

@ -1013,8 +1013,8 @@ build.json @home-assistant/supervisor
/tests/components/sharkiq/ @JeffResc @funkybunch @AritroSaha10
/homeassistant/components/shell_command/ @home-assistant/core
/tests/components/shell_command/ @home-assistant/core
/homeassistant/components/shelly/ @balloob @bieniu @thecode @chemelli74
/tests/components/shelly/ @balloob @bieniu @thecode @chemelli74
/homeassistant/components/shelly/ @balloob @bieniu @thecode @chemelli74 @bdraco
/tests/components/shelly/ @balloob @bieniu @thecode @chemelli74 @bdraco
/homeassistant/components/shodan/ @fabaff
/homeassistant/components/sia/ @eavanvalkenburg
/tests/components/sia/ @eavanvalkenburg

4
homeassistant/components/shelly/__init__.py
View File

@ -6,7 +6,7 @@ from typing import Any, Final
import aioshelly
from aioshelly.block_device import BlockDevice
from aioshelly.exceptions import DeviceConnectionError, InvalidAuthError
from aioshelly.rpc_device import RpcDevice
from aioshelly.rpc_device import RpcDevice, UpdateType
import voluptuous as vol
from homeassistant.config_entries import ConfigEntry
@ -252,7 +252,7 @@ async def _async_setup_rpc_entry(hass: HomeAssistant, entry: ConfigEntry) -> boo
hass.config_entries.async_setup_platforms(entry, platforms)
@callback
def _async_device_online(_: Any) -> None:
def _async_device_online(_: Any, update_type: UpdateType) -> None:
LOGGER.debug("Device %s is online, resuming setup", entry.title)
shelly_entry_data.device = None

66
homeassistant/components/shelly/bluetooth/__init__.py Normal file
View File

@ -0,0 +1,66 @@
"""Bluetooth support for shelly."""
from __future__ import annotations
from typing import TYPE_CHECKING
from aioshelly.ble import async_start_scanner
from aioshelly.ble.const import (
BLE_SCAN_RESULT_EVENT,
BLE_SCAN_RESULT_VERSION,
DEFAULT_DURATION_MS,
DEFAULT_INTERVAL_MS,
DEFAULT_WINDOW_MS,
)
from homeassistant.components.bluetooth import (
HaBluetoothConnector,
async_get_advertisement_callback,
async_register_scanner,
)
from homeassistant.core import CALLBACK_TYPE, HomeAssistant, callback as hass_callback
from homeassistant.helpers.device_registry import format_mac
from ..const import BLEScannerMode
from .scanner import ShellyBLEScanner
if TYPE_CHECKING:
from ..coordinator import ShellyRpcCoordinator
async def async_connect_scanner(
hass: HomeAssistant,
coordinator: ShellyRpcCoordinator,
scanner_mode: BLEScannerMode,
) -> CALLBACK_TYPE:
"""Connect scanner."""
device = coordinator.device
source = format_mac(coordinator.mac).upper()
new_info_callback = async_get_advertisement_callback(hass)
connector = HaBluetoothConnector(
# no active connections to shelly yet
client=None, # type: ignore[arg-type]
source=source,
can_connect=lambda: False,
)
scanner = ShellyBLEScanner(hass, source, new_info_callback, connector, False)
unload_callbacks = [
async_register_scanner(hass, scanner, False),
scanner.async_setup(),
coordinator.async_subscribe_events(scanner.async_on_event),
]
await async_start_scanner(
device=device,
active=scanner_mode == BLEScannerMode.ACTIVE,
event_type=BLE_SCAN_RESULT_EVENT,
data_version=BLE_SCAN_RESULT_VERSION,
interval_ms=DEFAULT_INTERVAL_MS,
window_ms=DEFAULT_WINDOW_MS,
duration_ms=DEFAULT_DURATION_MS,
)
@hass_callback
def _async_unload() -> None:
for callback in unload_callbacks:
callback()
return _async_unload

47
homeassistant/components/shelly/bluetooth/scanner.py Normal file
View File

@ -0,0 +1,47 @@
"""Bluetooth scanner for shelly."""
from __future__ import annotations
import logging
from typing import Any
from aioshelly.ble import parse_ble_scan_result_event
from aioshelly.ble.const import BLE_SCAN_RESULT_EVENT, BLE_SCAN_RESULT_VERSION
from homeassistant.components.bluetooth import BaseHaRemoteScanner
from homeassistant.core import callback
_LOGGER = logging.getLogger(__name__)
class ShellyBLEScanner(BaseHaRemoteScanner):
"""Scanner for shelly."""
@callback
def async_on_event(self, event: dict[str, Any]) -> None:
"""Process an event from the shelly and ignore if its not a ble.scan_result."""
if event.get("event") != BLE_SCAN_RESULT_EVENT:
return
data = event["data"]
if data[0] != BLE_SCAN_RESULT_VERSION:
_LOGGER.warning("Unsupported BLE scan result version: %s", data[0])
return
try:
address, rssi, parsed = parse_ble_scan_result_event(data)
except Exception as err: # pylint: disable=broad-except
# Broad exception catch because we have no
# control over the data that is coming in.
_LOGGER.error("Failed to parse BLE event: %s", err, exc_info=True)
return
self._async_on_advertisement(
address,
rssi,
parsed.local_name,
parsed.service_uuids,
parsed.service_data,
parsed.manufacturer_data,
parsed.tx_power,
)

78
homeassistant/components/shelly/config_flow.py
View File

@ -12,16 +12,25 @@ from aioshelly.exceptions import (
InvalidAuthError,
)
from aioshelly.rpc_device import RpcDevice
from awesomeversion import AwesomeVersion
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.components import zeroconf
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import HomeAssistant
from homeassistant.core import HomeAssistant, callback
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers import aiohttp_client
from homeassistant.helpers import aiohttp_client, selector
from .const import CONF_SLEEP_PERIOD, DOMAIN, LOGGER
from .const import (
BLE_MIN_VERSION,
CONF_BLE_SCANNER_MODE,
CONF_SLEEP_PERIOD,
DOMAIN,
LOGGER,
BLEScannerMode,
)
from .coordinator import get_entry_data
from .utils import (
get_block_device_name,
get_block_device_sleep_period,
@ -37,6 +46,13 @@ from .utils import (
HOST_SCHEMA: Final = vol.Schema({vol.Required(CONF_HOST): str})
BLE_SCANNER_OPTIONS = [
selector.SelectOptionDict(value=BLEScannerMode.DISABLED, label="Disabled"),
selector.SelectOptionDict(value=BLEScannerMode.ACTIVE, label="Active"),
selector.SelectOptionDict(value=BLEScannerMode.PASSIVE, label="Passive"),
]
async def validate_input(
hass: HomeAssistant,
host: str,
@ -310,3 +326,59 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
return await aioshelly.common.get_info(
aiohttp_client.async_get_clientsession(self.hass), host
)
@staticmethod
@callback
def async_get_options_flow(
config_entry: config_entries.ConfigEntry,
) -> OptionsFlowHandler:
"""Get the options flow for this handler."""
return OptionsFlowHandler(config_entry)
@classmethod
@callback
def async_supports_options_flow(
cls, config_entry: config_entries.ConfigEntry
) -> bool:
"""Return options flow support for this handler."""
return config_entry.data.get("gen") == 2
class OptionsFlowHandler(config_entries.OptionsFlow):
"""Handle the option flow for shelly."""
def __init__(self, config_entry: config_entries.ConfigEntry) -> None:
"""Initialize options flow."""
self.config_entry = config_entry
async def async_step_init(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle options flow."""
if user_input is not None:
entry_data = get_entry_data(self.hass)[self.config_entry.entry_id]
if user_input[CONF_BLE_SCANNER_MODE] != BLEScannerMode.DISABLED and (
not entry_data.rpc
or AwesomeVersion(entry_data.rpc.device.version) < BLE_MIN_VERSION
):
return self.async_abort(
reason="ble_unsupported",
description_placeholders={"ble_min_version": BLE_MIN_VERSION},
)
return self.async_create_entry(title="", data=user_input)
return self.async_show_form(
step_id="init",
data_schema=vol.Schema(
{
vol.Required(
CONF_BLE_SCANNER_MODE,
default=self.config_entry.options.get(
CONF_BLE_SCANNER_MODE, BLEScannerMode.DISABLED
),
): selector.SelectSelector(
selector.SelectSelectorConfig(options=BLE_SCANNER_OPTIONS),
),
}
),
)

16
homeassistant/components/shelly/const.py
View File

@ -5,6 +5,10 @@ from logging import Logger, getLogger
import re
from typing import Final
from awesomeversion import AwesomeVersion
from homeassistant.backports.enum import StrEnum
DOMAIN: Final = "shelly"
LOGGER: Logger = getLogger(__package__)
@ -156,3 +160,15 @@ UPTIME_DEVIATION: Final = 5
ENTRY_RELOAD_COOLDOWN = 60
SHELLY_GAS_MODELS = ["SHGS-1"]
BLE_MIN_VERSION = AwesomeVersion("0.12.0-beta2")
CONF_BLE_SCANNER_MODE = "ble_scanner_mode"
class BLEScannerMode(StrEnum):
"""BLE scanner mode."""
DISABLED = "disabled"
ACTIVE = "active"
PASSIVE = "passive"

109
homeassistant/components/shelly/coordinator.py
View File

@ -1,6 +1,7 @@
"""Coordinators for the Shelly integration."""
from __future__ import annotations
import asyncio
from collections.abc import Callable, Coroutine
from dataclasses import dataclass
from datetime import timedelta
@ -9,7 +10,8 @@ from typing import Any, cast
import aioshelly
from aioshelly.block_device import BlockDevice
from aioshelly.exceptions import DeviceConnectionError, InvalidAuthError, RpcCallError
from aioshelly.rpc_device import RpcDevice
from aioshelly.rpc_device import RpcDevice, UpdateType
from awesomeversion import AwesomeVersion
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import ATTR_DEVICE_ID, CONF_HOST, EVENT_HOMEASSISTANT_STOP
@ -18,12 +20,15 @@ from homeassistant.helpers import device_registry
from homeassistant.helpers.debounce import Debouncer
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .bluetooth import async_connect_scanner
from .const import (
ATTR_CHANNEL,
ATTR_CLICK_TYPE,
ATTR_DEVICE,
ATTR_GENERATION,
BATTERY_DEVICES_WITH_PERMANENT_CONNECTION,
BLE_MIN_VERSION,
CONF_BLE_SCANNER_MODE,
CONF_SLEEP_PERIOD,
DATA_CONFIG_ENTRY,
DOMAIN,
@ -40,6 +45,7 @@ from .const import (
SHBTN_MODELS,
SLEEP_PERIOD_MULTIPLIER,
UPDATE_PERIOD_MULTIPLIER,
BLEScannerMode,
)
from .utils import (
device_update_info,
@ -336,7 +342,10 @@ class ShellyRpcCoordinator(DataUpdateCoordinator):
)
self.entry = entry
self.device = device
self.connected = False
self._disconnected_callbacks: list[CALLBACK_TYPE] = []
self._connection_lock = asyncio.Lock()
self._event_listeners: list[Callable[[dict[str, Any]], None]] = []
self._debounced_reload: Debouncer[Coroutine[Any, Any, None]] = Debouncer(
hass,
@ -346,16 +355,14 @@ class ShellyRpcCoordinator(DataUpdateCoordinator):
function=self._async_reload_entry,
)
entry.async_on_unload(self._debounced_reload.async_cancel)
self._last_event: dict[str, Any] | None = None
self._last_status: dict[str, Any] | None = None
entry.async_on_unload(
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, self._handle_ha_stop)
)
entry.async_on_unload(entry.add_update_listener(self._async_update_listener))
async def _async_reload_entry(self) -> None:
"""Reload entry."""
self._debounced_reload.async_cancel()
LOGGER.debug("Reloading entry %s", self.name)
await self.hass.config_entries.async_reload(self.entry.entry_id)
@ -390,12 +397,19 @@ class ShellyRpcCoordinator(DataUpdateCoordinator):
return _unsubscribe
async def _async_update_listener(
self, hass: HomeAssistant, entry: ConfigEntry
) -> None:
"""Reconfigure on update."""
async with self._connection_lock:
if self.connected:
self._async_run_disconnected_events()
await self._async_run_connected_events()
@callback
def _async_device_event_handler(self, event_data: dict[str, Any]) -> None:
"""Handle device events."""
self.update_sleep_period()
events: list[dict[str, Any]] = event_data["events"]
for event in events:
event_type = event.get("event")
if event_type is None:
@ -405,6 +419,7 @@ class ShellyRpcCoordinator(DataUpdateCoordinator):
event_callback(event)
if event_type == "config_changed":
self.update_sleep_period()
LOGGER.info(
"Config for %s changed, reloading entry in %s seconds",
self.name,
@ -460,21 +475,71 @@ class ShellyRpcCoordinator(DataUpdateCoordinator):
"""Firmware version of the device."""
return self.device.firmware_version if self.device.initialized else ""
@callback
def _async_handle_update(self, device_: RpcDevice) -> None:
"""Handle device update."""
device = self.device
if not device.initialized:
return
event = device.event
status = device.status
async def _async_disconnected(self) -> None:
"""Handle device disconnected."""
async with self._connection_lock:
if not self.connected: # Already disconnected
return
self.connected = False
self._async_run_disconnected_events()
if event and event != self._last_event:
self._last_event = event
@callback
def _async_run_disconnected_events(self) -> None:
"""Run disconnected events.
This will be executed on disconnect or when the config entry
is updated.
"""
for disconnected_callback in self._disconnected_callbacks:
disconnected_callback()
self._disconnected_callbacks.clear()
async def _async_connected(self) -> None:
"""Handle device connected."""
async with self._connection_lock:
if self.connected: # Already connected
return
self.connected = True
await self._async_run_connected_events()
async def _async_run_connected_events(self) -> None:
"""Run connected events.
This will be executed on connect or when the config entry
is updated.
"""
await self._async_connect_ble_scanner()
async def _async_connect_ble_scanner(self) -> None:
"""Connect BLE scanner."""
ble_scanner_mode = self.entry.options.get(
CONF_BLE_SCANNER_MODE, BLEScannerMode.DISABLED
)
if ble_scanner_mode == BLEScannerMode.DISABLED:
return
if AwesomeVersion(self.device.version) < BLE_MIN_VERSION:
LOGGER.error(
"BLE not supported on device %s with firmware %s; upgrade to %s",
self.name,
self.device.version,
BLE_MIN_VERSION,
)
return
self._disconnected_callbacks.append(
await async_connect_scanner(self.hass, self, ble_scanner_mode)
)
@callback
def _async_handle_update(self, device_: RpcDevice, update_type: UpdateType) -> None:
"""Handle device update."""
if update_type is UpdateType.INITIALIZED:
self.hass.async_create_task(self._async_connected())
elif update_type is UpdateType.DISCONNECTED:
self.hass.async_create_task(self._async_disconnected())
elif update_type is UpdateType.STATUS:
self.async_set_updated_data(self.device)
elif update_type is UpdateType.EVENT and (event := self.device.event):
self._async_device_event_handler(event)
if status and status != self._last_status:
self._last_status = status
self.async_set_updated_data(device)
def async_setup(self) -> None:
"""Set up the coordinator."""
@ -491,10 +556,14 @@ class ShellyRpcCoordinator(DataUpdateCoordinator):
)
self.device_id = entry.id
self.device.subscribe_updates(self._async_handle_update)
if self.device.initialized:
# If we are already initialized, we are connected
self.hass.async_create_task(self._async_connected())
async def shutdown(self) -> None:
"""Shutdown the coordinator."""
await self.device.shutdown()
await self._async_disconnected()
async def _handle_ha_stop(self, _event: Event) -> None:
"""Handle Home Assistant stopping."""

6
homeassistant/components/shelly/manifest.json
View File

@ -3,15 +3,15 @@
"name": "Shelly",
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/shelly",
"requirements": ["aioshelly==4.1.2"],
"dependencies": ["http"],
"requirements": ["aioshelly==5.0.0"],
"dependencies": ["bluetooth", "http"],
"zeroconf": [
{
"type": "_http._tcp.local.",
"name": "shelly*"
}
],
"codeowners": ["@balloob", "@bieniu", "@thecode", "@chemelli74"],
"codeowners": ["@balloob", "@bieniu", "@thecode", "@chemelli74", "@bdraco"],
"iot_class": "local_push",
"loggers": ["aioshelly"],
"integration_type": "device"

13
homeassistant/components/shelly/strings.json
View File

@ -58,5 +58,18 @@
"double_push": "{subtype} double push",
"long_push": "{subtype} long push"
}
},
"options": {
"step": {
"init": {
"description": "Bluetooth scanning can be active or passive. With active, the Shelly requests data from nearby devices; with passive, the Shelly receives unsolicited data from nearby devices.",
"data": {
"ble_scanner_mode": "Bluetooth scanner mode"
}
}
},
"abort": {
"ble_unsupported": "Bluetooth support requires firmware version {ble_min_version} or newer."
}
}
}

13
homeassistant/components/shelly/translations/en.json
View File

@ -58,5 +58,18 @@
"single_push": "{subtype} single push",
"triple": "{subtype} triple clicked"
}
},
"options": {
"abort": {
"ble_unsupported": "Bluetooth support requires firmware version {ble_min_version} or newer."
},
"step": {
"init": {
"data": {
"ble_scanner_mode": "Bluetooth scanner mode"
},
"description": "Bluetooth scanning can be active or passive. With active, the Shelly requests data from nearby devices; with passive, the Shelly receives unsolicited data from nearby devices."
}
}
}
}

2
requirements_all.txt
View File

@ -261,7 +261,7 @@ aiosenseme==0.6.1
aiosenz==1.0.0
# homeassistant.components.shelly
aioshelly==4.1.2
aioshelly==5.0.0
# homeassistant.components.skybell
aioskybell==22.7.0

2
requirements_test_all.txt
View File

@ -236,7 +236,7 @@ aiosenseme==0.6.1
aiosenz==1.0.0
# homeassistant.components.shelly
aioshelly==4.1.2
aioshelly==5.0.0
# homeassistant.components.skybell
aioskybell==22.7.0

22
tests/components/shelly/__init__.py
View File

@ -1,4 +1,6 @@
"""Tests for the Shelly integration."""
from __future__ import annotations
from copy import deepcopy
from typing import Any
from unittest.mock import Mock
@ -15,7 +17,11 @@ MOCK_MAC = "123456789ABC"
async def init_integration(
hass: HomeAssistant, gen: int, model="SHSW-25", sleep_period=0
hass: HomeAssistant,
gen: int,
model="SHSW-25",
sleep_period=0,
options: dict[str, Any] | None = None,
) -> MockConfigEntry:
"""Set up the Shelly integration in Home Assistant."""
data = {
@ -25,7 +31,9 @@ async def init_integration(
"gen": gen,
}
entry = MockConfigEntry(domain=DOMAIN, data=data, unique_id=MOCK_MAC)
entry = MockConfigEntry(
domain=DOMAIN, data=data, unique_id=MOCK_MAC, options=options
)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
@ -45,3 +53,13 @@ def mutate_rpc_device_status(
new_status = deepcopy(mock_rpc_device.status)
new_status[top_level_key][key] = value
monkeypatch.setattr(mock_rpc_device, "status", new_status)
def inject_rpc_device_event(
monkeypatch: pytest.MonkeyPatch,
mock_rpc_device: Mock,
event: dict[str, dict[str, Any]],
) -> None:
"""Inject event for rpc device."""
monkeypatch.setattr(mock_rpc_device, "event", event)
mock_rpc_device.mock_event()

2
tests/components/shelly/bluetooth/__init__.py Normal file
View File

@ -0,0 +1,2 @@
"""Bluetooth tests for Shelly integration."""
from __future__ import annotations

133
tests/components/shelly/bluetooth/test_scanner.py Normal file
View File

@ -0,0 +1,133 @@
"""Test the shelly bluetooth scanner."""
from __future__ import annotations
from aioshelly.ble.const import BLE_SCAN_RESULT_EVENT
from homeassistant.components import bluetooth
from homeassistant.components.shelly.const import CONF_BLE_SCANNER_MODE, BLEScannerMode
from .. import init_integration, inject_rpc_device_event
async def test_scanner(hass, mock_rpc_device, monkeypatch):
"""Test injecting data into the scanner."""
await init_integration(
hass, 2, options={CONF_BLE_SCANNER_MODE: BLEScannerMode.ACTIVE}
)
assert mock_rpc_device.initialized is True
inject_rpc_device_event(
monkeypatch,
mock_rpc_device,
{
"events": [
{
"component": "script:1",
"data": [
1,
"aa:bb:cc:dd:ee:ff",
-62,
"AgEGCf9ZANH7O3TIkA==",
"EQcbxdWlAgC4n+YRTSIADaLLBhYADUgQYQ==",
],
"event": BLE_SCAN_RESULT_EVENT,
"id": 1,
"ts": 1668522399.2,
}
],
"ts": 1668522399.2,
},
)
ble_device = bluetooth.async_ble_device_from_address(
hass, "AA:BB:CC:DD:EE:FF", connectable=False
)
assert ble_device is not None
ble_device = bluetooth.async_ble_device_from_address(
hass, "AA:BB:CC:DD:EE:FF", connectable=True
)
assert ble_device is None
async def test_scanner_ignores_non_ble_events(hass, mock_rpc_device, monkeypatch):
"""Test injecting non ble data into the scanner."""
await init_integration(
hass, 2, options={CONF_BLE_SCANNER_MODE: BLEScannerMode.ACTIVE}
)
assert mock_rpc_device.initialized is True
inject_rpc_device_event(
monkeypatch,
mock_rpc_device,
{
"events": [
{
"component": "script:1",
"data": [],
"event": "not_ble_scan_result",
"id": 1,
"ts": 1668522399.2,
}
],
"ts": 1668522399.2,
},
)
async def test_scanner_ignores_wrong_version_and_logs(
hass, mock_rpc_device, monkeypatch, caplog
):
"""Test injecting wrong version of ble data into the scanner."""
await init_integration(
hass, 2, options={CONF_BLE_SCANNER_MODE: BLEScannerMode.ACTIVE}
)
assert mock_rpc_device.initialized is True
inject_rpc_device_event(
monkeypatch,
mock_rpc_device,
{
"events": [
{
"component": "script:1",
"data": [
0,
"aa:bb:cc:dd:ee:ff",
-62,
"AgEGCf9ZANH7O3TIkA==",
"EQcbxdWlAgC4n+YRTSIADaLLBhYADUgQYQ==",
],
"event": BLE_SCAN_RESULT_EVENT,
"id": 1,
"ts": 1668522399.2,
}
],
"ts": 1668522399.2,
},
)
assert "Unsupported BLE scan result version: 0" in caplog.text
async def test_scanner_warns_on_corrupt_event(
hass, mock_rpc_device, monkeypatch, caplog
):
"""Test injecting garbage ble data into the scanner."""
await init_integration(
hass, 2, options={CONF_BLE_SCANNER_MODE: BLEScannerMode.ACTIVE}
)
assert mock_rpc_device.initialized is True
inject_rpc_device_event(
monkeypatch,
mock_rpc_device,
{
"events": [
{
"component": "script:1",
"data": [
1,
],
"event": BLE_SCAN_RESULT_EVENT,
"id": 1,
"ts": 1668522399.2,
}
],
"ts": 1668522399.2,
},
)
assert "Failed to parse BLE event" in caplog.text

68
tests/components/shelly/conftest.py
View File

@ -1,8 +1,10 @@
"""Test configuration for Shelly."""
from __future__ import annotations
from unittest.mock import AsyncMock, Mock, patch
from aioshelly.block_device import BlockDevice
from aioshelly.rpc_device import RpcDevice
from aioshelly.rpc_device import RpcDevice, UpdateType
import pytest
from homeassistant.components.shelly.const import (
@ -194,6 +196,7 @@ async def mock_block_device():
blocks=MOCK_BLOCKS,
settings=MOCK_SETTINGS,
shelly=MOCK_SHELLY_COAP,
version="0.10.0",
status=MOCK_STATUS_COAP,
firmware_version="some fw string",
initialized=True,
@ -204,25 +207,62 @@ async def mock_block_device():
yield block_device_mock.return_value
@pytest.fixture
async def mock_rpc_device():
def _mock_rpc_device(version: str | None = None):
"""Mock rpc (Gen2, Websocket) device."""
return Mock(
spec=RpcDevice,
config=MOCK_CONFIG,
event={},
shelly=MOCK_SHELLY_RPC,
version=version or "0.12.0",
status=MOCK_STATUS_RPC,
firmware_version="some fw string",
initialized=True,
)
@pytest.fixture
async def mock_pre_ble_rpc_device():
"""Mock rpc (Gen2, Websocket) device pre BLE."""
with patch("aioshelly.rpc_device.RpcDevice.create") as rpc_device_mock:
def update():
rpc_device_mock.return_value.subscribe_updates.call_args[0][0]({})
device = Mock(
spec=RpcDevice,
config=MOCK_CONFIG,
event={},
shelly=MOCK_SHELLY_RPC,
status=MOCK_STATUS_RPC,
firmware_version="some fw string",
initialized=True,
)
rpc_device_mock.return_value.subscribe_updates.call_args[0][0](
{}, UpdateType.STATUS
)
device = _mock_rpc_device("0.11.0")
rpc_device_mock.return_value = device
rpc_device_mock.return_value.mock_update = Mock(side_effect=update)
yield rpc_device_mock.return_value
@pytest.fixture
async def mock_rpc_device():
"""Mock rpc (Gen2, Websocket) device with BLE support."""
with patch("aioshelly.rpc_device.RpcDevice.create") as rpc_device_mock, patch(
"homeassistant.components.shelly.bluetooth.async_start_scanner"
):
def update():
rpc_device_mock.return_value.subscribe_updates.call_args[0][0](
{}, UpdateType.STATUS
)
def event():
rpc_device_mock.return_value.subscribe_updates.call_args[0][0](
{}, UpdateType.EVENT
)
device = _mock_rpc_device("0.12.0")
rpc_device_mock.return_value = device
rpc_device_mock.return_value.mock_update = Mock(side_effect=update)
rpc_device_mock.return_value.mock_event = Mock(side_effect=event)
yield rpc_device_mock.return_value
@pytest.fixture(autouse=True)
def mock_bluetooth(enable_bluetooth):
"""Auto mock bluetooth."""

2
tests/components/shelly/test_button.py
View File

@ -1,4 +1,6 @@
"""Tests for Shelly button platform."""
from __future__ import annotations
from homeassistant.components.button import DOMAIN as BUTTON_DOMAIN, SERVICE_PRESS
from homeassistant.const import ATTR_ENTITY_ID, STATE_UNKNOWN
from homeassistant.core import HomeAssistant

155
tests/components/shelly/test_config_flow.py
View File

@ -10,8 +10,15 @@ import pytest
from homeassistant import config_entries, data_entry_flow
from homeassistant.components import zeroconf
from homeassistant.components.shelly.const import DOMAIN
from homeassistant.components.shelly.const import (
CONF_BLE_SCANNER_MODE,
DOMAIN,
BLEScannerMode,
)
from homeassistant.config_entries import SOURCE_REAUTH
from homeassistant.setup import async_setup_component
from . import init_integration
from tests.common import MockConfigEntry
@ -880,3 +887,149 @@ async def test_reauth_get_info_error(hass, error):
assert result["type"] == data_entry_flow.FlowResultType.ABORT
assert result["reason"] == "reauth_unsuccessful"
async def test_options_flow_disabled_gen_1(hass, mock_block_device, hass_ws_client):
"""Test options are disabled for gen1 devices."""
await async_setup_component(hass, "config", {})
entry = await init_integration(hass, 1)
ws_client = await hass_ws_client(hass)
await ws_client.send_json(
{
"id": 5,
"type": "config_entries/get",
"domain": "shelly",
}
)
response = await ws_client.receive_json()
assert response["result"][0]["supports_options"] is False
await hass.config_entries.async_unload(entry.entry_id)
async def test_options_flow_enabled_gen_2(hass, mock_rpc_device, hass_ws_client):
"""Test options are enabled for gen2 devices."""
await async_setup_component(hass, "config", {})
entry = await init_integration(hass, 2)
ws_client = await hass_ws_client(hass)
await ws_client.send_json(
{
"id": 5,
"type": "config_entries/get",
"domain": "shelly",
}
)
response = await ws_client.receive_json()
assert response["result"][0]["supports_options"] is True
await hass.config_entries.async_unload(entry.entry_id)
async def test_options_flow_ble(hass, mock_rpc_device):
"""Test setting ble options for gen2 devices."""
entry = await init_integration(hass, 2)
result = await hass.config_entries.options.async_init(entry.entry_id)
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "init"
assert result["errors"] is None
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={
CONF_BLE_SCANNER_MODE: BLEScannerMode.DISABLED,
},
)
await hass.async_block_till_done()
assert result["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
assert result["data"][CONF_BLE_SCANNER_MODE] == BLEScannerMode.DISABLED
result = await hass.config_entries.options.async_init(entry.entry_id)
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "init"
assert result["errors"] is None
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={
CONF_BLE_SCANNER_MODE: BLEScannerMode.ACTIVE,
},
)
await hass.async_block_till_done()
assert result["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
assert result["data"][CONF_BLE_SCANNER_MODE] == BLEScannerMode.ACTIVE
result = await hass.config_entries.options.async_init(entry.entry_id)
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "init"
assert result["errors"] is None
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={
CONF_BLE_SCANNER_MODE: BLEScannerMode.PASSIVE,
},
)
await hass.async_block_till_done()
assert result["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
assert result["data"][CONF_BLE_SCANNER_MODE] == BLEScannerMode.PASSIVE
await hass.config_entries.async_unload(entry.entry_id)
async def test_options_flow_pre_ble_device(hass, mock_pre_ble_rpc_device):
"""Test setting ble options for gen2 devices with pre ble firmware."""
entry = await init_integration(hass, 2)
result = await hass.config_entries.options.async_init(entry.entry_id)
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "init"
assert result["errors"] is None
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={
CONF_BLE_SCANNER_MODE: BLEScannerMode.DISABLED,
},
)
await hass.async_block_till_done()
assert result["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
assert result["data"][CONF_BLE_SCANNER_MODE] == BLEScannerMode.DISABLED
result = await hass.config_entries.options.async_init(entry.entry_id)
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "init"
assert result["errors"] is None
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={
CONF_BLE_SCANNER_MODE: BLEScannerMode.ACTIVE,
},
)
await hass.async_block_till_done()
assert result["type"] == data_entry_flow.FlowResultType.ABORT
assert result["reason"] == "ble_unsupported"
result = await hass.config_entries.options.async_init(entry.entry_id)
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "init"
assert result["errors"] is None
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={
CONF_BLE_SCANNER_MODE: BLEScannerMode.PASSIVE,
},
)
await hass.async_block_till_done()
assert result["type"] == data_entry_flow.FlowResultType.ABORT
assert result["reason"] == "ble_unsupported"
await hass.config_entries.async_unload(entry.entry_id)

1
tests/components/shelly/test_init.py
View File

@ -1,4 +1,5 @@
"""Test cases for the Shelly component."""
from __future__ import annotations
from unittest.mock import AsyncMock