Refactor tradfri sensor tests (#88026)

This commit is contained in:
Martin Hjelmare 2023-02-14 13:35:27 +01:00 committed by GitHub
parent d7861e79c5
commit e1a5d5a749
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 199 additions and 190 deletions

View File

@ -9,6 +9,7 @@ from unittest.mock import MagicMock, Mock, patch
import pytest import pytest
from pytradfri.device import Device from pytradfri.device import Device
from pytradfri.device.air_purifier import AirPurifier from pytradfri.device.air_purifier import AirPurifier
from pytradfri.device.blind import Blind
from homeassistant.components.tradfri.const import DOMAIN from homeassistant.components.tradfri.const import DOMAIN
@ -102,3 +103,18 @@ def air_purifier(air_purifier_response: dict[str, Any]) -> AirPurifier:
air_purifier_control = device.air_purifier_control air_purifier_control = device.air_purifier_control
assert air_purifier_control assert air_purifier_control
return air_purifier_control.air_purifiers[0] return air_purifier_control.air_purifiers[0]
@pytest.fixture(scope="session")
def blind_response() -> dict[str, Any]:
"""Return a blind response."""
return json.loads(load_fixture("blind.json", DOMAIN))
@pytest.fixture
def blind(blind_response: dict[str, Any]) -> Blind:
"""Return blind."""
device = Device(blind_response)
blind_control = device.blind_control
assert blind_control
return blind_control.blinds[0]

View File

@ -0,0 +1,17 @@
{
"3": {
"0": "IKEA of Sweden",
"1": "TRADFRI remote control",
"2": "",
"3": "1.2.214",
"6": 3,
"9": 87
},
"5750": 0,
"9001": "Test",
"9002": 1509923521,
"9003": 65536,
"9019": 1,
"9020": 1510010209,
"9054": 0
}

View File

@ -1,39 +1,19 @@
"""Tradfri cover (recognised as blinds in the IKEA ecosystem) platform tests.""" """Tradfri cover (recognised as blinds in the IKEA ecosystem) platform tests."""
from __future__ import annotations from __future__ import annotations
import json
from typing import Any from typing import Any
from unittest.mock import MagicMock, Mock from unittest.mock import MagicMock, Mock
import pytest import pytest
from pytradfri.const import ATTR_REACHABLE_STATE from pytradfri.const import ATTR_REACHABLE_STATE
from pytradfri.device import Device
from pytradfri.device.blind import Blind from pytradfri.device.blind import Blind
from homeassistant.components.cover import ATTR_CURRENT_POSITION, DOMAIN as COVER_DOMAIN from homeassistant.components.cover import ATTR_CURRENT_POSITION, DOMAIN as COVER_DOMAIN
from homeassistant.components.tradfri.const import DOMAIN
from homeassistant.const import STATE_CLOSED, STATE_OPEN, STATE_UNAVAILABLE from homeassistant.const import STATE_CLOSED, STATE_OPEN, STATE_UNAVAILABLE
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from .common import setup_integration, trigger_observe_callback from .common import setup_integration, trigger_observe_callback
from tests.common import load_fixture
@pytest.fixture(scope="module")
def blind_response() -> dict[str, Any]:
"""Return a blind response."""
return json.loads(load_fixture("blind.json", DOMAIN))
@pytest.fixture
def blind(blind_response: dict[str, Any]) -> Blind:
"""Return blind."""
device = Device(blind_response)
blind_control = device.blind_control
assert blind_control
return blind_control.blinds[0]
async def test_cover_available( async def test_cover_available(
hass: HomeAssistant, hass: HomeAssistant,

View File

@ -1,203 +1,201 @@
"""Tradfri sensor platform tests.""" """Tradfri sensor platform tests."""
from __future__ import annotations from __future__ import annotations
from unittest.mock import MagicMock, Mock, PropertyMock, patch import json
from typing import Any
from unittest.mock import MagicMock, Mock
import pytest import pytest
from pytradfri.device.air_purifier_control import AirPurifierControl from pytradfri.const import (
ATTR_AIR_PURIFIER_AIR_QUALITY,
ATTR_DEVICE_BATTERY,
ATTR_DEVICE_INFO,
ATTR_REACHABLE_STATE,
ROOT_AIR_PURIFIER,
)
from pytradfri.device import Device
from pytradfri.device.air_purifier import AirPurifier
from pytradfri.device.blind import Blind
from homeassistant.components import tradfri from homeassistant.components.sensor import (
ATTR_STATE_CLASS,
DOMAIN as SENSOR_DOMAIN,
SensorDeviceClass,
SensorStateClass,
)
from homeassistant.components.tradfri.const import DOMAIN
from homeassistant.const import (
ATTR_DEVICE_CLASS,
ATTR_UNIT_OF_MEASUREMENT,
CONCENTRATION_MICROGRAMS_PER_CUBIC_METER,
PERCENTAGE,
STATE_UNAVAILABLE,
STATE_UNKNOWN,
UnitOfTime,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er from homeassistant.helpers import entity_registry as er
from . import GATEWAY_ID from . import GATEWAY_ID
from .common import setup_integration from .common import setup_integration, trigger_observe_callback
from tests.common import MockConfigEntry from tests.common import MockConfigEntry, load_fixture
@pytest.fixture(autouse=True) @pytest.fixture(scope="module")
def setup(request): def remote_control_response() -> dict[str, Any]:
"""Set up patches for pytradfri methods for the fan platform. """Return a remote control response."""
return json.loads(load_fixture("remote_control.json", DOMAIN))
This is used in test_fan as well as in test_sensor.
"""
with patch(
"pytradfri.device.AirPurifierControl.raw",
new_callable=PropertyMock,
return_value=[{"mock": "mock"}],
), patch(
"pytradfri.device.AirPurifierControl.air_purifiers",
):
yield
def mock_fan(test_features=None, test_state=None, device_number=0): @pytest.fixture
"""Mock a tradfri fan/air purifier.""" def remote_control(remote_control_response: dict[str, Any]) -> Device:
if test_features is None: """Return remote control."""
test_features = {} return Device(remote_control_response)
if test_state is None:
test_state = {}
mock_fan_data = Mock(**test_state)
dev_info_mock = MagicMock()
dev_info_mock.manufacturer = "manufacturer"
dev_info_mock.model_number = "model"
dev_info_mock.firmware_version = "1.2.3"
_mock_fan = Mock(
id=f"mock-fan-id-{device_number}",
reachable=True,
observe=Mock(),
device_info=dev_info_mock,
has_light_control=False,
has_socket_control=False,
has_blind_control=False,
has_signal_repeater_control=False,
has_air_purifier_control=True,
)
_mock_fan.name = f"tradfri_fan_{device_number}"
air_purifier_control = AirPurifierControl(_mock_fan)
# Store the initial state.
setattr(air_purifier_control, "air_purifiers", [mock_fan_data])
_mock_fan.air_purifier_control = air_purifier_control
return _mock_fan
def mock_sensor(test_state: list, device_number=0): async def test_battery_sensor(
"""Mock a tradfri sensor.""" hass: HomeAssistant,
dev_info_mock = MagicMock() mock_gateway: Mock,
dev_info_mock.manufacturer = "manufacturer" mock_api_factory: MagicMock,
dev_info_mock.model_number = "model" remote_control: Device,
dev_info_mock.firmware_version = "1.2.3" ) -> None:
_mock_sensor = Mock(
id=f"mock-sensor-id-{device_number}",
reachable=True,
observe=Mock(),
device_info=dev_info_mock,
has_light_control=False,
has_socket_control=False,
has_blind_control=False,
has_signal_repeater_control=False,
has_air_purifier_control=False,
)
# Set state value, eg battery_level = 50, or has_air_purifier_control=True
for state in test_state:
setattr(dev_info_mock, state["attribute"], state["value"])
_mock_sensor.name = f"tradfri_sensor_{device_number}"
return _mock_sensor
async def test_battery_sensor(hass, mock_gateway, mock_api_factory):
"""Test that a battery sensor is correctly added.""" """Test that a battery sensor is correctly added."""
mock_gateway.mock_devices.append( entity_id = "sensor.test"
mock_sensor(test_state=[{"attribute": "battery_level", "value": 60}]) device = remote_control
) mock_gateway.mock_devices.append(device)
await setup_integration(hass) await setup_integration(hass)
sensor_1 = hass.states.get("sensor.tradfri_sensor_0") state = hass.states.get(entity_id)
assert sensor_1 is not None assert state
assert sensor_1.state == "60" assert state.state == "87"
assert sensor_1.attributes["unit_of_measurement"] == "%" assert state.attributes[ATTR_UNIT_OF_MEASUREMENT] == PERCENTAGE
assert sensor_1.attributes["device_class"] == "battery" assert state.attributes[ATTR_DEVICE_CLASS] == SensorDeviceClass.BATTERY
assert state.attributes[ATTR_STATE_CLASS] == SensorStateClass.MEASUREMENT
await trigger_observe_callback(
hass, mock_gateway, device, {ATTR_DEVICE_INFO: {ATTR_DEVICE_BATTERY: 60}}
)
state = hass.states.get(entity_id)
assert state
assert state.state == "60"
assert state.attributes[ATTR_UNIT_OF_MEASUREMENT] == PERCENTAGE
assert state.attributes[ATTR_DEVICE_CLASS] == SensorDeviceClass.BATTERY
assert state.attributes[ATTR_STATE_CLASS] == SensorStateClass.MEASUREMENT
async def test_cover_battery_sensor(hass, mock_gateway, mock_api_factory): async def test_cover_battery_sensor(
hass: HomeAssistant,
mock_gateway: Mock,
mock_api_factory: MagicMock,
blind: Blind,
) -> None:
"""Test that a battery sensor is correctly added for a cover (blind).""" """Test that a battery sensor is correctly added for a cover (blind)."""
mock_gateway.mock_devices.append( entity_id = "sensor.test"
mock_sensor( device = blind.device
test_state=[ mock_gateway.mock_devices.append(device)
{"attribute": "battery_level", "value": 42, "has_blind_control": True}
]
)
)
await setup_integration(hass) await setup_integration(hass)
sensor_1 = hass.states.get("sensor.tradfri_sensor_0") state = hass.states.get(entity_id)
assert sensor_1 is not None assert state
assert sensor_1.state == "42" assert state.state == "77"
assert sensor_1.attributes["unit_of_measurement"] == "%" assert state.attributes[ATTR_UNIT_OF_MEASUREMENT] == PERCENTAGE
assert sensor_1.attributes["device_class"] == "battery" assert state.attributes[ATTR_DEVICE_CLASS] == SensorDeviceClass.BATTERY
assert sensor_1.attributes["state_class"] == "measurement" assert state.attributes[ATTR_STATE_CLASS] == SensorStateClass.MEASUREMENT
async def test_air_quality_sensor(hass, mock_gateway, mock_api_factory): async def test_air_quality_sensor(
hass: HomeAssistant,
mock_gateway: Mock,
mock_api_factory: MagicMock,
air_purifier: AirPurifier,
) -> None:
"""Test that a battery sensor is correctly added.""" """Test that a battery sensor is correctly added."""
mock_gateway.mock_devices.append( entity_id = "sensor.test_air_quality"
mock_fan( device = air_purifier.device
test_state={ mock_gateway.mock_devices.append(device)
"fan_speed": 10,
"air_quality": 42,
"filter_lifetime_remaining": 120,
}
)
)
await setup_integration(hass) await setup_integration(hass)
sensor_1 = hass.states.get("sensor.tradfri_fan_0_air_quality") state = hass.states.get(entity_id)
assert sensor_1 is not None assert state
assert sensor_1.state == "42" assert state.state == "5"
assert sensor_1.attributes["unit_of_measurement"] == "µg/m³" assert (
assert sensor_1.attributes["state_class"] == "measurement" state.attributes[ATTR_UNIT_OF_MEASUREMENT]
assert "device_class" not in sensor_1.attributes == CONCENTRATION_MICROGRAMS_PER_CUBIC_METER
)
assert state.attributes[ATTR_STATE_CLASS] == SensorStateClass.MEASUREMENT
assert ATTR_DEVICE_CLASS not in state.attributes
# The sensor returns 65535 if the fan is turned off
await trigger_observe_callback(
hass,
mock_gateway,
device,
{ROOT_AIR_PURIFIER: [{ATTR_AIR_PURIFIER_AIR_QUALITY: 65535}]},
)
state = hass.states.get(entity_id)
assert state
assert state.state == STATE_UNKNOWN
async def test_filter_time_left_sensor(hass, mock_gateway, mock_api_factory): async def test_filter_time_left_sensor(
hass: HomeAssistant,
mock_gateway: Mock,
mock_api_factory: MagicMock,
air_purifier: AirPurifier,
) -> None:
"""Test that a battery sensor is correctly added.""" """Test that a battery sensor is correctly added."""
mock_gateway.mock_devices.append( entity_id = "sensor.test_filter_time_left"
mock_fan( device = air_purifier.device
test_state={ mock_gateway.mock_devices.append(device)
"fan_speed": 10,
"air_quality": 42,
"filter_lifetime_remaining": 120,
}
)
)
await setup_integration(hass) await setup_integration(hass)
sensor_1 = hass.states.get("sensor.tradfri_fan_0_filter_time_left") state = hass.states.get(entity_id)
assert sensor_1 is not None
assert sensor_1.state == "2" assert state
assert sensor_1.attributes["unit_of_measurement"] == "h" assert state.state == "4320"
assert state.attributes[ATTR_UNIT_OF_MEASUREMENT] == UnitOfTime.HOURS
assert state.attributes[ATTR_STATE_CLASS] == SensorStateClass.MEASUREMENT
async def test_sensor_observed(hass, mock_gateway, mock_api_factory): async def test_sensor_available(
"""Test that sensors are correctly observed.""" hass: HomeAssistant,
sensor = mock_sensor(test_state=[{"attribute": "battery_level", "value": 60}]) mock_gateway: Mock,
mock_gateway.mock_devices.append(sensor) mock_api_factory: MagicMock,
await setup_integration(hass) air_purifier: AirPurifier,
assert len(sensor.observe.mock_calls) > 0 ) -> None:
async def test_sensor_available(hass, mock_gateway, mock_api_factory):
"""Test sensor available property.""" """Test sensor available property."""
sensor = mock_sensor( entity_id = "sensor.test_filter_time_left"
test_state=[{"attribute": "battery_level", "value": 60}], device_number=1 device = air_purifier.device
) mock_gateway.mock_devices.append(device)
sensor.reachable = True
sensor2 = mock_sensor(
test_state=[{"attribute": "battery_level", "value": 60}], device_number=2
)
sensor2.reachable = False
mock_gateway.mock_devices.append(sensor)
mock_gateway.mock_devices.append(sensor2)
await setup_integration(hass) await setup_integration(hass)
assert hass.states.get("sensor.tradfri_sensor_1").state == "60" state = hass.states.get(entity_id)
assert hass.states.get("sensor.tradfri_sensor_2").state == "unavailable" assert state
assert state.state == "4320"
await trigger_observe_callback(
hass, mock_gateway, device, {ATTR_REACHABLE_STATE: 0}
)
state = hass.states.get(entity_id)
assert state
assert state.state == STATE_UNAVAILABLE
async def test_unique_id_migration(hass, mock_gateway, mock_api_factory): async def test_unique_id_migration(
hass: HomeAssistant,
mock_gateway: Mock,
mock_api_factory: MagicMock,
remote_control: Device,
) -> None:
"""Test unique ID is migrated from old format to new.""" """Test unique ID is migrated from old format to new."""
ent_reg = er.async_get(hass) ent_reg = er.async_get(hass)
old_unique_id = f"{GATEWAY_ID}-mock-sensor-id-0" old_unique_id = f"{GATEWAY_ID}-65536"
entry = MockConfigEntry( entry = MockConfigEntry(
domain=tradfri.DOMAIN, domain=DOMAIN,
data={ data={
"host": "mock-host", "host": "mock-host",
"identity": "mock-identity", "identity": "mock-identity",
@ -208,32 +206,30 @@ async def test_unique_id_migration(hass, mock_gateway, mock_api_factory):
entry.add_to_hass(hass) entry.add_to_hass(hass)
# Version 1 # Version 1
sensor_name = "sensor.tradfri_sensor_0" entity_id = "sensor.test"
entity_name = sensor_name.split(".")[1] entity_name = entity_id.split(".")[1]
entity_entry = ent_reg.async_get_or_create( entity_entry = ent_reg.async_get_or_create(
"sensor", SENSOR_DOMAIN,
tradfri.DOMAIN, DOMAIN,
old_unique_id, old_unique_id,
suggested_object_id=entity_name, suggested_object_id=entity_name,
config_entry=entry, config_entry=entry,
original_name=entity_name, original_name=entity_name,
) )
assert entity_entry.entity_id == sensor_name assert entity_entry.entity_id == entity_id
assert entity_entry.unique_id == old_unique_id assert entity_entry.unique_id == old_unique_id
# Add a sensor to the gateway so that it populates coordinator list # Add a sensor to the gateway so that it populates coordinator list
sensor = mock_sensor( device = remote_control
test_state=[{"attribute": "battery_level", "value": 60}], mock_gateway.mock_devices.append(device)
)
mock_gateway.mock_devices.append(sensor)
await hass.config_entries.async_setup(entry.entry_id) await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done() await hass.async_block_till_done()
# Check that new RegistryEntry is using new unique ID format # Check that new RegistryEntry is using new unique ID format
entity_entry = ent_reg.async_get(sensor_name) new_unique_id = f"{old_unique_id}-battery_level"
new_unique_id = f"{GATEWAY_ID}-mock-sensor-id-0-battery_level" migrated_entity_entry = ent_reg.async_get(entity_id)
assert entity_entry.unique_id == new_unique_id assert migrated_entity_entry is not None
assert ent_reg.async_get_entity_id("sensor", tradfri.DOMAIN, old_unique_id) is None assert migrated_entity_entry.unique_id == new_unique_id
assert ent_reg.async_get_entity_id(SENSOR_DOMAIN, DOMAIN, old_unique_id) is None