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

Add tests coverage for Shelly entity (#82432)

Browse source 
* Add tests coverage for Shelly entity

* Make it generator expression
This commit is contained in:
Shay Levy 2022-11-20 23:59:25 +02:00 • committed by GitHub
parent 4bb69fee23
commit 32f3eb722e
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
8 changed files with 272 additions and 60 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
.coveragerc
View file

@ -1111,7 +1111,6 @@ omit =
homeassistant/components/seventeentrack/sensor.py
homeassistant/components/shelly/climate.py
homeassistant/components/shelly/coordinator.py
homeassistant/components/shelly/entity.py
homeassistant/components/shelly/number.py
homeassistant/components/shelly/utils.py
homeassistant/components/shiftr/*

63
homeassistant/components/shelly/entity.py
View file

@ -21,12 +21,7 @@ from homeassistant.helpers.typing import StateType
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import CONF_SLEEP_PERIOD, LOGGER
from .coordinator import (
ShellyBlockCoordinator,
ShellyRpcCoordinator,
ShellyRpcPollingCoordinator,
get_entry_data,
)
from .coordinator import ShellyBlockCoordinator, ShellyRpcCoordinator, get_entry_data
from .utils import (
async_remove_shelly_entity,
get_block_entity_name,
@ -269,21 +264,10 @@ def async_setup_entry_rest(
"""Set up entities for REST sensors."""
coordinator = get_entry_data(hass)[config_entry.entry_id].rest
assert coordinator
entities = []
for sensor_id in sensors:
description = sensors.get(sensor_id)
if not coordinator.device.settings.get("sleep_mode"):
entities.append((sensor_id, description))
if not entities:
return
async_add_entities(
[
sensor_class(coordinator, sensor_id, description)
for sensor_id, description in entities
]
sensor_class(coordinator, sensor_id, sensors[sensor_id])
for sensor_id in sensors
)
@ -350,10 +334,6 @@ class ShellyBlockEntity(CoordinatorEntity[ShellyBlockCoordinator]):
"""When entity is added to HASS."""
self.async_on_remove(self.coordinator.async_add_listener(self._update_callback))
async def async_update(self) -> None:
"""Update entity with latest info."""
await self.coordinator.async_request_refresh()
@callback
def _update_callback(self) -> None:
"""Handle device update."""
@ -373,16 +353,12 @@ class ShellyBlockEntity(CoordinatorEntity[ShellyBlockCoordinator]):
self.coordinator.entry.async_start_reauth(self.hass)
class ShellyRpcEntity(entity.Entity):
class ShellyRpcEntity(CoordinatorEntity[ShellyRpcCoordinator]):
"""Helper class to represent a rpc entity."""
def __init__(
self,
coordinator: ShellyRpcCoordinator | ShellyRpcPollingCoordinator,
key: str,
) -> None:
def __init__(self, coordinator: ShellyRpcCoordinator, key: str) -> None:
"""Initialize Shelly entity."""
self.coordinator = coordinator
super().__init__(coordinator)
self.key = key
self._attr_should_poll = False
self._attr_device_info = {
@ -405,10 +381,6 @@ class ShellyRpcEntity(entity.Entity):
"""When entity is added to HASS."""
self.async_on_remove(self.coordinator.async_add_listener(self._update_callback))
async def async_update(self) -> None:
"""Update entity with latest info."""
await self.coordinator.async_request_refresh()
@callback
def _update_callback(self) -> None:
"""Handle device update."""
@ -525,16 +497,6 @@ class ShellyRestAttributeEntity(CoordinatorEntity[ShellyBlockCoordinator]):
)
return self._last_value
@property
def extra_state_attributes(self) -> dict[str, Any] | None:
"""Return the state attributes."""
if self.entity_description.extra_state_attributes is None:
return None
return self.entity_description.extra_state_attributes(
self.block_coordinator.device.status
)
class ShellyRpcAttributeEntity(ShellyRpcEntity, entity.Entity):
"""Helper class to represent a rpc attribute."""
@ -586,19 +548,6 @@ class ShellyRpcAttributeEntity(ShellyRpcEntity, entity.Entity):
self.coordinator.device.status[self.key][self.entity_description.sub_key]
)
@property
def extra_state_attributes(self) -> dict[str, Any] | None:
"""Return the state attributes."""
if self.entity_description.extra_state_attributes is None:
return None
assert self.coordinator.device.shelly
return self.entity_description.extra_state_attributes(
self.coordinator.device.status[self.key][self.entity_description.sub_key],
self.coordinator.device.shelly,
)
class ShellySleepingBlockAttributeEntity(ShellyBlockAttributeEntity, RestoreEntity):
"""Represent a shelly sleeping block attribute entity."""

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

@ -12,6 +12,7 @@ from homeassistant.components.shelly.const import (
CONF_SLEEP_PERIOD,
DOMAIN,
REST_SENSORS_UPDATE_INTERVAL,
RPC_SENSORS_POLLING_INTERVAL,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_HOST
@ -84,6 +85,14 @@ async def mock_rest_update(hass: HomeAssistant):
await hass.async_block_till_done()
async def mock_polling_rpc_update(hass: HomeAssistant):
"""Move time to create polling RPC sensors update event."""
async_fire_time_changed(
hass, dt.utcnow() + timedelta(seconds=RPC_SENSORS_POLLING_INTERVAL)
)
await hass.async_block_till_done()
def register_entity(
hass: HomeAssistant,
domain: str,

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

@ -29,6 +29,7 @@ MOCK_SETTINGS = {
"fw": "20201124-092159/v1.9.0@57ac4ad8",
"relays": [{"btn_type": "momentary"}, {"btn_type": "toggle"}],
"rollers": [{"positioning": True}],
"external_power": 0,
}
@ -97,12 +98,20 @@ MOCK_BLOCKS = [
set_state=AsyncMock(side_effect=mock_light_set_state),
),
Mock(
sensor_ids={"motion": 0, "temp": 22.1},
sensor_ids={"motion": 0, "temp": 22.1, "gas": "mild"},
motion=0,
temp=22.1,
gas="mild",
description="sensor_0",
type="sensor",
),
Mock(
sensor_ids={"battery": 98},
battery=98,
cfgChanged=0,
description="device_0",
type="device",
),
]
MOCK_CONFIG = {
@ -165,6 +174,8 @@ MOCK_STATUS_RPC = {
"stable": {"version": "some_beta_version"},
}
},
"voltmeter": {"voltage": 4.3},
"wifi": {"rssi": -63},
}

35
tests/components/shelly/test_binary_sensor.py
View file

@ -4,6 +4,7 @@
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 homeassistant.helpers.entity_registry import async_get
from . import (
init_integration,
@ -32,6 +33,25 @@ async def test_block_binary_sensor(hass, mock_block_device, monkeypatch):
assert hass.states.get(entity_id).state == STATE_ON
async def test_block_binary_sensor_extra_state_attr(
hass, mock_block_device, monkeypatch
):
"""Test block binary sensor extra state attributes."""
entity_id = f"{BINARY_SENSOR_DOMAIN}.test_name_gas"
await init_integration(hass, 1)
state = hass.states.get(entity_id)
assert state.state == STATE_ON
assert state.attributes.get("detected") == "mild"
monkeypatch.setattr(mock_block_device.blocks[SENSOR_BLOCK_ID], "gas", "none")
mock_block_device.mock_update()
state = hass.states.get(entity_id)
assert state.state == STATE_OFF
assert state.attributes.get("detected") == "none"
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")
@ -105,6 +125,21 @@ async def test_rpc_binary_sensor(hass, mock_rpc_device, monkeypatch) -> None:
assert hass.states.get(entity_id).state == STATE_ON
async def test_rpc_binary_sensor_removal(hass, mock_rpc_device, monkeypatch):
"""Test RPC binary sensor is removed due to removal_condition."""
entity_registry = async_get(hass)
entity_id = register_entity(
hass, BINARY_SENSOR_DOMAIN, "test_cover_0_input", "input:0-input"
)
assert entity_registry.async_get(entity_id) is not None
monkeypatch.setattr(mock_rpc_device, "status", {"input:0": {"state": False}})
await init_integration(hass, 2)
assert entity_registry.async_get(entity_id) is None
async def test_rpc_sleeping_binary_sensor(
hass, mock_rpc_device, device_reg, monkeypatch
) -> None:

3
tests/components/shelly/test_diagnostics.py
View file

@ -70,6 +70,7 @@ async def test_rpc_config_entry_diagnostics(
"beta": {"version": "some_beta_version"},
"stable": {"version": "some_beta_version"},
}
}
},
"wifi": {"rssi": -63},
},
}

103
tests/components/shelly/test_sensor.py
View file

@ -2,10 +2,13 @@
from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN
from homeassistant.const import STATE_UNAVAILABLE, STATE_UNKNOWN
from homeassistant.core import State
from homeassistant.helpers.entity_registry import async_get
from . import (
init_integration,
mock_polling_rpc_update,
mock_rest_update,
mutate_rpc_device_status,
register_device,
@ -16,6 +19,7 @@ from tests.common import mock_restore_cache
RELAY_BLOCK_ID = 0
SENSOR_BLOCK_ID = 3
DEVICE_BLOCK_ID = 4
async def test_block_sensor(hass, mock_block_device, monkeypatch):
@ -88,6 +92,79 @@ async def test_block_restored_sleeping_sensor(
assert hass.states.get(entity_id).state == "22.1"
async def test_block_sensor_error(hass, mock_block_device, monkeypatch):
"""Test block sensor unavailable on sensor error."""
entity_id = f"{SENSOR_DOMAIN}.test_name_battery"
await init_integration(hass, 1)
assert hass.states.get(entity_id).state == "98"
monkeypatch.setattr(mock_block_device.blocks[DEVICE_BLOCK_ID], "battery", -1)
mock_block_device.mock_update()
assert hass.states.get(entity_id).state == STATE_UNAVAILABLE
async def test_block_sensor_removal(hass, mock_block_device, monkeypatch):
"""Test block sensor is removed due to removal_condition."""
entity_registry = async_get(hass)
entity_id = register_entity(
hass, SENSOR_DOMAIN, "test_name_battery", "device_0-battery"
)
assert entity_registry.async_get(entity_id) is not None
monkeypatch.setitem(mock_block_device.settings, "external_power", 1)
await init_integration(hass, 1)
assert entity_registry.async_get(entity_id) is None
async def test_block_not_matched_restored_sleeping_sensor(
hass, mock_block_device, device_reg, monkeypatch
):
"""Test block not matched to restored sleeping sensor."""
entry = await init_integration(hass, 1, sleep_period=1000, skip_setup=True)
register_device(device_reg, entry)
entity_id = register_entity(
hass, SENSOR_DOMAIN, "test_name_temperature", "sensor_0-temp", entry
)
mock_restore_cache(hass, [State(entity_id, "20.4")])
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 == "20.4"
# Make device online
monkeypatch.setattr(mock_block_device.blocks[SENSOR_BLOCK_ID], "type", "other_type")
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 == "20.4"
async def test_block_sensor_without_value(hass, mock_block_device, monkeypatch):
"""Test block sensor without value is not created."""
entity_id = f"{SENSOR_DOMAIN}.test_name_battery"
monkeypatch.setattr(mock_block_device.blocks[DEVICE_BLOCK_ID], "battery", None)
await init_integration(hass, 1)
assert hass.states.get(entity_id) is None
async def test_block_sensor_unknown_value(hass, mock_block_device, monkeypatch):
"""Test block sensor unknown value."""
entity_id = f"{SENSOR_DOMAIN}.test_name_battery"
await init_integration(hass, 1)
monkeypatch.setattr(mock_block_device.blocks[DEVICE_BLOCK_ID], "battery", None)
mock_block_device.mock_update()
assert hass.states.get(entity_id).state == STATE_UNKNOWN
async def test_rpc_sensor(hass, mock_rpc_device, monkeypatch) -> None:
"""Test RPC sensor."""
entity_id = f"{SENSOR_DOMAIN}.test_cover_0_power"
@ -101,6 +178,32 @@ async def test_rpc_sensor(hass, mock_rpc_device, monkeypatch) -> None:
assert hass.states.get(entity_id).state == "88.2"
async def test_rpc_sensor_error(hass, mock_rpc_device, monkeypatch):
"""Test RPC sensor unavailable on sensor error."""
entity_id = f"{SENSOR_DOMAIN}.test_name_voltmeter"
await init_integration(hass, 2)
assert hass.states.get(entity_id).state == "4.3"
mutate_rpc_device_status(monkeypatch, mock_rpc_device, "voltmeter", "voltage", None)
mock_rpc_device.mock_update()
assert hass.states.get(entity_id).state == STATE_UNAVAILABLE
async def test_rpc_polling_sensor(hass, mock_rpc_device, monkeypatch) -> None:
"""Test RPC polling sensor."""
entity_id = register_entity(hass, SENSOR_DOMAIN, "test_name_rssi", "wifi-rssi")
await init_integration(hass, 2)
assert hass.states.get(entity_id).state == "-63"
mutate_rpc_device_status(monkeypatch, mock_rpc_device, "wifi", "rssi", "-70")
await mock_polling_rpc_update(hass)
assert hass.states.get(entity_id).state == "-70"
async def test_rpc_sleeping_sensor(
hass, mock_rpc_device, device_reg, monkeypatch
) -> None:

105
tests/components/shelly/test_switch.py
View file

@ -1,5 +1,12 @@
"""Tests for Shelly switch platform."""
from unittest.mock import AsyncMock
from aioshelly.exceptions import DeviceConnectionError, InvalidAuthError, RpcCallError
import pytest
from homeassistant.components.shelly.const import DOMAIN
from homeassistant.components.switch import DOMAIN as SWITCH_DOMAIN
from homeassistant.config_entries import SOURCE_REAUTH, ConfigEntryState
from homeassistant.const import (
ATTR_ENTITY_ID,
SERVICE_TURN_OFF,
@ -7,6 +14,7 @@ from homeassistant.const import (
STATE_OFF,
STATE_ON,
)
from homeassistant.exceptions import HomeAssistantError
from . import init_integration
@ -34,6 +42,56 @@ async def test_block_device_services(hass, mock_block_device):
assert hass.states.get("switch.test_name_channel_1").state == STATE_OFF
async def test_block_set_state_connection_error(hass, mock_block_device, monkeypatch):
"""Test block device set state connection error."""
monkeypatch.setattr(
mock_block_device.blocks[RELAY_BLOCK_ID],
"set_state",
AsyncMock(side_effect=DeviceConnectionError),
)
await init_integration(hass, 1)
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
SWITCH_DOMAIN,
SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: "switch.test_name_channel_1"},
blocking=True,
)
async def test_block_set_state_auth_error(hass, mock_block_device, monkeypatch):
"""Test block device set state authentication error."""
monkeypatch.setattr(
mock_block_device.blocks[RELAY_BLOCK_ID],
"set_state",
AsyncMock(side_effect=InvalidAuthError),
)
entry = await init_integration(hass, 1)
assert entry.state == ConfigEntryState.LOADED
await hass.services.async_call(
SWITCH_DOMAIN,
SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: "switch.test_name_channel_1"},
blocking=True,
)
assert entry.state == ConfigEntryState.LOADED
flows = hass.config_entries.flow.async_progress()
assert len(flows) == 1
flow = flows[0]
assert flow.get("step_id") == "reauth_confirm"
assert flow.get("handler") == DOMAIN
assert "context" in flow
assert flow["context"].get("source") == SOURCE_REAUTH
assert flow["context"].get("entry_id") == entry.entry_id
async def test_block_device_update(hass, mock_block_device, monkeypatch):
"""Test block device update."""
monkeypatch.setattr(mock_block_device.blocks[RELAY_BLOCK_ID], "output", False)
@ -98,3 +156,50 @@ async def test_rpc_device_switch_type_lights_mode(hass, mock_rpc_device, monkeyp
)
await init_integration(hass, 2)
assert hass.states.get("switch.test_switch_0") is None
@pytest.mark.parametrize("exc", [DeviceConnectionError, RpcCallError(-1, "error")])
async def test_rpc_set_state_errors(hass, exc, mock_rpc_device, monkeypatch):
"""Test RPC device set state connection/call errors."""
monkeypatch.setattr(mock_rpc_device, "call_rpc", AsyncMock(side_effect=exc))
await init_integration(hass, 2)
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
SWITCH_DOMAIN,
SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: "switch.test_switch_0"},
blocking=True,
)
async def test_rpc_auth_error(hass, mock_rpc_device, monkeypatch):
"""Test RPC device set state authentication error."""
monkeypatch.setattr(
mock_rpc_device,
"call_rpc",
AsyncMock(side_effect=InvalidAuthError),
)
entry = await init_integration(hass, 2)
assert entry.state == ConfigEntryState.LOADED
await hass.services.async_call(
SWITCH_DOMAIN,
SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: "switch.test_switch_0"},
blocking=True,
)
assert entry.state == ConfigEntryState.LOADED
flows = hass.config_entries.flow.async_progress()
assert len(flows) == 1
flow = flows[0]
assert flow.get("step_id") == "reauth_confirm"
assert flow.get("handler") == DOMAIN
assert "context" in flow
assert flow["context"].get("source") == SOURCE_REAUTH
assert flow["context"].get("entry_id") == entry.entry_id