evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add tests for Shelly binary_sensor platform (#82367)

Browse source 
* Add tests for Shelly binary_sensor platform

* Rename create to register for
This commit is contained in:
Shay Levy 2022-11-19 18:07:02 +02:00 • committed by GitHub
parent e028516939
commit ea98f0e9e8
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
6 changed files with 238 additions and 44 deletions
Download patch file Download diff file Expand all files Collapse all files

1
.coveragerc
View file

@ -1109,7 +1109,6 @@ omit =
homeassistant/components/sesame/lock.py
homeassistant/components/seven_segments/image_processing.py
homeassistant/components/seventeentrack/sensor.py
homeassistant/components/shelly/binary_sensor.py
homeassistant/components/shelly/climate.py
homeassistant/components/shelly/coordinator.py
homeassistant/components/shelly/entity.py

10
homeassistant/components/shelly/binary_sensor.py
View file

@ -268,10 +268,8 @@ class RestBinarySensor(ShellyRestAttributeEntity, BinarySensorEntity):
entity_description: RestBinarySensorDescription
@property
def is_on(self) -> bool | None:
def is_on(self) -> bool:
"""Return true if REST sensor state is on."""
if self.attribute_value is None:
return None
return bool(self.attribute_value)
@ -281,10 +279,8 @@ class RpcBinarySensor(ShellyRpcAttributeEntity, BinarySensorEntity):
entity_description: RpcBinarySensorDescription
@property
def is_on(self) -> bool | None:
def is_on(self) -> bool:
"""Return true if RPC sensor state is on."""
if self.attribute_value is None:
return None
return bool(self.attribute_value)
@ -308,7 +304,7 @@ class RpcSleepingBinarySensor(ShellySleepingRpcAttributeEntity, BinarySensorEnti
entity_description: RpcBinarySensorDescription
@property
def is_on(self) -> bool | None:
def is_on(self) -> bool:
"""Return true if RPC sensor state is on."""
if self.coordinator.device.initialized:
return bool(self.attribute_value)

56
tests/components/shelly/__init__.py
View file

@ -2,16 +2,25 @@
from __future__ import annotations
from copy import deepcopy
from datetime import timedelta
from typing import Any
from unittest.mock import Mock
import pytest
from homeassistant.components.shelly.const import CONF_SLEEP_PERIOD, DOMAIN
from homeassistant.components.shelly.const import (
CONF_SLEEP_PERIOD,
DOMAIN,
REST_SENSORS_UPDATE_INTERVAL,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_HOST
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry
from homeassistant.helpers.entity_registry import async_get
from homeassistant.util import dt
from tests.common import MockConfigEntry
from tests.common import MockConfigEntry, async_fire_time_changed
MOCK_MAC = "123456789ABC"
@ -22,6 +31,7 @@ async def init_integration(
model="SHSW-25",
sleep_period=0,
options: dict[str, Any] | None = None,
skip_setup: bool = False,
) -> MockConfigEntry:
"""Set up the Shelly integration in Home Assistant."""
data = {
@ -36,6 +46,7 @@ async def init_integration(
)
entry.add_to_hass(hass)
if not skip_setup:
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
@ -63,3 +74,44 @@ def inject_rpc_device_event(
"""Inject event for rpc device."""
monkeypatch.setattr(mock_rpc_device, "event", event)
mock_rpc_device.mock_event()
async def mock_rest_update(hass: HomeAssistant):
"""Move time to create REST sensors update event."""
async_fire_time_changed(
hass, dt.utcnow() + timedelta(seconds=REST_SENSORS_UPDATE_INTERVAL)
)
await hass.async_block_till_done()
def register_entity(
hass: HomeAssistant,
domain: str,
object_id: str,
unique_id: str,
config_entry: ConfigEntry | None = None,
) -> str:
"""Register enabled entity, return entity_id."""
entity_registry = async_get(hass)
entity_registry.async_get_or_create(
domain,
DOMAIN,
f"{MOCK_MAC}-{unique_id}",
suggested_object_id=object_id,
disabled_by=None,
config_entry=config_entry,
)
return f"{domain}.{object_id}"
def register_device(device_reg, config_entry: ConfigEntry):
"""Register Shelly device."""
device_reg.async_get_or_create(
config_entry_id=config_entry.entry_id,
connections={
(
device_registry.CONNECTION_NETWORK_MAC,
device_registry.format_mac(MOCK_MAC),
)
},
)

18
tests/components/shelly/conftest.py
View file

@ -63,9 +63,11 @@ def mock_light_set_state(
MOCK_BLOCKS = [
Mock(
sensor_ids={"inputEvent": "S", "inputEventCnt": 2},
sensor_ids={"inputEvent": "S", "inputEventCnt": 2, "overpower": 0},
channel="0",
type="relay",
overpower=0,
description="relay_0",
set_state=AsyncMock(side_effect=lambda turn: {"ison": turn == "on"}),
),
Mock(
@ -88,6 +90,12 @@ MOCK_BLOCKS = [
type="light",
set_state=AsyncMock(side_effect=mock_light_set_state),
),
Mock(
sensor_ids={"motion": 0},
motion=0,
description="sensor_0",
type="sensor",
),
]
MOCK_CONFIG = {
@ -135,7 +143,13 @@ MOCK_STATUS_COAP = {
MOCK_STATUS_RPC = {
"switch:0": {"output": True},
"cover:0": {"state": "stopped", "pos_control": True, "current_pos": 50},
"cloud": {"connected": False},
"cover:0": {
"state": "stopped",
"pos_control": True,
"current_pos": 50,
"apower": 85.3,
},
"sys": {
"available_updates": {
"beta": {"version": "some_beta_version"},

155
tests/components/shelly/test_binary_sensor.py Normal file
View file

@ -0,0 +1,155 @@
"""Tests for Shelly binary sensor platform."""
from homeassistant.components.binary_sensor import DOMAIN as BINARY_SENSOR_DOMAIN
from homeassistant.const import STATE_OFF, STATE_ON
from homeassistant.core import State
from . import (
init_integration,
mock_rest_update,
mutate_rpc_device_status,
register_device,
register_entity,
)
from tests.common import mock_restore_cache
RELAY_BLOCK_ID = 0
SENSOR_BLOCK_ID = 3
async def test_block_binary_sensor(hass, mock_block_device, monkeypatch):
"""Test block binary sensor."""
entity_id = f"{BINARY_SENSOR_DOMAIN}.test_name_channel_1_overpowering"
await init_integration(hass, 1)
assert hass.states.get(entity_id).state == STATE_OFF
monkeypatch.setattr(mock_block_device.blocks[RELAY_BLOCK_ID], "overpower", 1)
mock_block_device.mock_update()
assert hass.states.get(entity_id).state == STATE_ON
async def test_block_rest_binary_sensor(hass, mock_block_device, monkeypatch):
"""Test block REST binary sensor."""
entity_id = register_entity(hass, BINARY_SENSOR_DOMAIN, "test_name_cloud", "cloud")
monkeypatch.setitem(mock_block_device.status, "cloud", {"connected": False})
await init_integration(hass, 1)
assert hass.states.get(entity_id).state == STATE_OFF
monkeypatch.setitem(mock_block_device.status["cloud"], "connected", True)
await mock_rest_update(hass)
assert hass.states.get(entity_id).state == STATE_ON
async def test_block_sleeping_binary_sensor(hass, mock_block_device, monkeypatch):
"""Test block sleeping binary sensor."""
entity_id = f"{BINARY_SENSOR_DOMAIN}.test_name_motion"
await init_integration(hass, 1, sleep_period=1000)
# Sensor should be created when device is online
assert hass.states.get(entity_id) is None
# Make device online
mock_block_device.mock_update()
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_OFF
monkeypatch.setattr(mock_block_device.blocks[SENSOR_BLOCK_ID], "motion", 1)
mock_block_device.mock_update()
assert hass.states.get(entity_id).state == STATE_ON
async def test_block_restored_sleeping_binary_sensor(
hass, mock_block_device, device_reg, monkeypatch
):
"""Test block restored sleeping binary sensor."""
entry = await init_integration(hass, 1, sleep_period=1000, skip_setup=True)
register_device(device_reg, entry)
entity_id = register_entity(
hass, BINARY_SENSOR_DOMAIN, "test_name_motion", "sensor_0-motion", entry
)
mock_restore_cache(hass, [State(entity_id, STATE_ON)])
monkeypatch.setattr(mock_block_device, "initialized", False)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_ON
# Make device online
monkeypatch.setattr(mock_block_device, "initialized", True)
mock_block_device.mock_update()
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_OFF
async def test_rpc_binary_sensor(hass, mock_rpc_device, monkeypatch) -> None:
"""Test RPC binary sensor."""
entity_id = f"{BINARY_SENSOR_DOMAIN}.test_cover_0_overpowering"
await init_integration(hass, 2)
assert hass.states.get(entity_id).state == STATE_OFF
mutate_rpc_device_status(
monkeypatch, mock_rpc_device, "cover:0", "errors", "overpower"
)
mock_rpc_device.mock_update()
assert hass.states.get(entity_id).state == STATE_ON
async def test_rpc_sleeping_binary_sensor(
hass, mock_rpc_device, device_reg, monkeypatch
) -> None:
"""Test RPC online sleeping binary sensor."""
entity_id = f"{BINARY_SENSOR_DOMAIN}.test_name_cloud"
entry = await init_integration(hass, 2, sleep_period=1000)
# Sensor should be created when device is online
assert hass.states.get(entity_id) is None
register_entity(hass, BINARY_SENSOR_DOMAIN, "test_name_cloud", "cloud-cloud", entry)
# Make device online
mock_rpc_device.mock_update()
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_OFF
mutate_rpc_device_status(monkeypatch, mock_rpc_device, "cloud", "connected", True)
mock_rpc_device.mock_update()
assert hass.states.get(entity_id).state == STATE_ON
async def test_rpc_restored_sleeping_binary_sensor(
hass, mock_rpc_device, device_reg, monkeypatch
):
"""Test RPC restored binary sensor."""
entry = await init_integration(hass, 2, sleep_period=1000, skip_setup=True)
register_device(device_reg, entry)
entity_id = register_entity(
hass, BINARY_SENSOR_DOMAIN, "test_name_cloud", "cloud-cloud", entry
)
mock_restore_cache(hass, [State(entity_id, STATE_ON)])
monkeypatch.setattr(mock_rpc_device, "initialized", False)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_ON
# Make device online
monkeypatch.setattr(mock_rpc_device, "initialized", True)
mock_rpc_device.mock_update()
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_OFF

38
tests/components/shelly/test_update.py
View file

@ -1,5 +1,4 @@
"""Tests for Shelly update platform."""
from datetime import timedelta
from unittest.mock import AsyncMock
from aioshelly.exceptions import DeviceConnectionError, InvalidAuthError, RpcCallError
@ -7,7 +6,7 @@ import pytest
from homeassistant.components.binary_sensor import DOMAIN as BINARY_SENSOR_DOMAIN
from homeassistant.components.button import DOMAIN as BUTTON_DOMAIN
from homeassistant.components.shelly.const import DOMAIN, REST_SENSORS_UPDATE_INTERVAL
from homeassistant.components.shelly.const import DOMAIN
from homeassistant.components.update import (
ATTR_IN_PROGRESS,
ATTR_INSTALLED_VERSION,
@ -20,11 +19,8 @@ from homeassistant.const import ATTR_ENTITY_ID, STATE_OFF, STATE_ON
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.entity_registry import async_get
from homeassistant.util import dt
from . import MOCK_MAC, init_integration
from tests.common import async_fire_time_changed
from . import MOCK_MAC, init_integration, mock_rest_update
@pytest.mark.parametrize(
@ -100,10 +96,7 @@ async def test_block_update(hass: HomeAssistant, mock_block_device, monkeypatch)
assert state.attributes[ATTR_IN_PROGRESS] is True
monkeypatch.setitem(mock_block_device.status["update"], "old_version", "2")
async_fire_time_changed(
hass, dt.utcnow() + timedelta(seconds=REST_SENSORS_UPDATE_INTERVAL)
)
await hass.async_block_till_done()
await mock_rest_update(hass)
state = hass.states.get("update.test_name_firmware_update")
assert state.state == STATE_OFF
@ -134,10 +127,7 @@ async def test_block_beta_update(hass: HomeAssistant, mock_block_device, monkeyp
assert state.attributes[ATTR_IN_PROGRESS] is False
monkeypatch.setitem(mock_block_device.status["update"], "beta_version", "2b")
async_fire_time_changed(
hass, dt.utcnow() + timedelta(seconds=REST_SENSORS_UPDATE_INTERVAL)
)
await hass.async_block_till_done()
await mock_rest_update(hass)
state = hass.states.get("update.test_name_beta_firmware_update")
assert state.state == STATE_ON
@ -160,10 +150,7 @@ async def test_block_beta_update(hass: HomeAssistant, mock_block_device, monkeyp
assert state.attributes[ATTR_IN_PROGRESS] is True
monkeypatch.setitem(mock_block_device.status["update"], "old_version", "2b")
async_fire_time_changed(
hass, dt.utcnow() + timedelta(seconds=REST_SENSORS_UPDATE_INTERVAL)
)
await hass.async_block_till_done()
await mock_rest_update(hass)
state = hass.states.get("update.test_name_beta_firmware_update")
assert state.state == STATE_OFF
@ -288,10 +275,7 @@ async def test_rpc_update(hass: HomeAssistant, mock_rpc_device, monkeypatch):
assert state.attributes[ATTR_IN_PROGRESS] is True
monkeypatch.setitem(mock_rpc_device.shelly, "ver", "2")
async_fire_time_changed(
hass, dt.utcnow() + timedelta(seconds=REST_SENSORS_UPDATE_INTERVAL)
)
await hass.async_block_till_done()
await mock_rest_update(hass)
state = hass.states.get("update.test_name_firmware_update")
assert state.state == STATE_OFF
@ -335,10 +319,7 @@ async def test_rpc_beta_update(hass: HomeAssistant, mock_rpc_device, monkeypatch
"beta": {"version": "2b"},
},
)
async_fire_time_changed(
hass, dt.utcnow() + timedelta(seconds=REST_SENSORS_UPDATE_INTERVAL)
)
await hass.async_block_till_done()
await mock_rest_update(hass)
state = hass.states.get("update.test_name_beta_firmware_update")
assert state.state == STATE_ON
@ -361,10 +342,7 @@ async def test_rpc_beta_update(hass: HomeAssistant, mock_rpc_device, monkeypatch
assert state.attributes[ATTR_IN_PROGRESS] is True
monkeypatch.setitem(mock_rpc_device.shelly, "ver", "2b")
async_fire_time_changed(
hass, dt.utcnow() + timedelta(seconds=REST_SENSORS_UPDATE_INTERVAL)
)
await hass.async_block_till_done()
await mock_rest_update(hass)
state = hass.states.get("update.test_name_beta_firmware_update")
assert state.state == STATE_OFF