home-assistant
/
core
mirror of https://github.com/home-assistant/core
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity
core/tests/components/google_assistant/test_trait.py

4159 lines
128 KiB
Python
Raw Permalink Blame History

"""Tests for the Google Assistant traits."""
from datetime import datetime, timedelta
from typing import Any
from unittest.mock import ANY, patch
from freezegun.api import FrozenDateTimeFactory
import pytest
from homeassistant.components import (
alarm_control_panel,
binary_sensor,
button,
camera,
climate,
cover,
event,
fan,
group,
humidifier,
input_boolean,
input_button,
input_select,
light,
lock,
media_player,
scene,
script,
select,
sensor,
switch,
vacuum,
valve,
water_heater,
)
from homeassistant.components.alarm_control_panel import (
AlarmControlPanelEntityFeature,
AlarmControlPanelState,
)
from homeassistant.components.camera import CameraEntityFeature
from homeassistant.components.climate import ClimateEntityFeature
from homeassistant.components.cover import CoverEntityFeature
from homeassistant.components.fan import FanEntityFeature
from homeassistant.components.google_assistant import const, error, helpers, trait
from homeassistant.components.google_assistant.error import SmartHomeError
from homeassistant.components.humidifier import HumidifierEntityFeature
from homeassistant.components.light import LightEntityFeature
from homeassistant.components.lock import LockEntityFeature
from homeassistant.components.media_player import (
SERVICE_PLAY_MEDIA,
MediaPlayerEntityFeature,
MediaType,
)
from homeassistant.components.vacuum import VacuumEntityFeature
from homeassistant.components.valve import ValveEntityFeature
from homeassistant.components.water_heater import WaterHeaterEntityFeature
from homeassistant.const import (
ATTR_ASSUMED_STATE,
ATTR_BATTERY_LEVEL,
ATTR_DEVICE_CLASS,
ATTR_ENTITY_ID,
ATTR_MODE,
ATTR_SUPPORTED_FEATURES,
ATTR_TEMPERATURE,
EVENT_CALL_SERVICE,
SERVICE_TURN_OFF,
SERVICE_TURN_ON,
STATE_IDLE,
STATE_OFF,
STATE_ON,
STATE_PAUSED,
STATE_PLAYING,
STATE_STANDBY,
STATE_UNAVAILABLE,
STATE_UNKNOWN,
UnitOfTemperature,
)
from homeassistant.core import DOMAIN as HOMEASSISTANT_DOMAIN, HomeAssistant, State
from homeassistant.core_config import async_process_ha_core_config
from homeassistant.util import color, dt as dt_util
from homeassistant.util.unit_conversion import TemperatureConverter
from . import BASIC_CONFIG, MockConfig
from tests.common import async_capture_events, async_mock_service
REQ_ID = "ff36a3cc-ec34-11e6-b1a0-64510650abcf"
BASIC_DATA = helpers.RequestData(
BASIC_CONFIG, "test-agent", const.SOURCE_CLOUD, REQ_ID, None
)
PIN_CONFIG = MockConfig(secure_devices_pin="1234")
PIN_DATA = helpers.RequestData(
PIN_CONFIG, "test-agent", const.SOURCE_CLOUD, REQ_ID, None
)
@pytest.mark.parametrize(
"supported_color_modes", [["brightness"], ["hs"], ["color_temp"]]
)
async def test_brightness_light(hass: HomeAssistant, supported_color_modes) -> None:
"""Test brightness trait support for light domain."""
assert helpers.get_google_type(light.DOMAIN, None) is not None
assert trait.BrightnessTrait.supported(
light.DOMAIN, 0, None, {"supported_color_modes": supported_color_modes}
)
trt = trait.BrightnessTrait(
hass,
State("light.bla", light.STATE_ON, {light.ATTR_BRIGHTNESS: 243}),
BASIC_CONFIG,
)
assert trt.sync_attributes() == {}
assert trt.query_attributes() == {"brightness": 95}
events = async_capture_events(hass, EVENT_CALL_SERVICE)
calls = async_mock_service(hass, light.DOMAIN, light.SERVICE_TURN_ON)
await trt.execute(
trait.COMMAND_BRIGHTNESS_ABSOLUTE, BASIC_DATA, {"brightness": 50}, {}
)
await hass.async_block_till_done()
assert len(calls) == 1
assert calls[0].data == {ATTR_ENTITY_ID: "light.bla", light.ATTR_BRIGHTNESS_PCT: 50}
assert len(events) == 1
assert events[0].data == {
"domain": "light",
"service": "turn_on",
"service_data": {"brightness_pct": 50, "entity_id": "light.bla"},
}
async def test_camera_stream(hass: HomeAssistant) -> None:
"""Test camera stream trait support for camera domain."""
await async_process_ha_core_config(
hass,
{"external_url": "https://example.com"},
)
assert helpers.get_google_type(camera.DOMAIN, None) is not None
assert trait.CameraStreamTrait.supported(
camera.DOMAIN, CameraEntityFeature.STREAM, None, None
)
trt = trait.CameraStreamTrait(
hass, State("camera.bla", camera.STATE_IDLE, {}), BASIC_CONFIG
)
assert trt.sync_attributes() == {
"cameraStreamSupportedProtocols": ["hls"],
"cameraStreamNeedAuthToken": False,
"cameraStreamNeedDrmEncryption": False,
}
assert trt.query_attributes() == {}
with patch(
"homeassistant.components.camera.async_request_stream",
return_value="/api/streams/bla",
):
await trt.execute(trait.COMMAND_GET_CAMERA_STREAM, BASIC_DATA, {}, {})
assert trt.query_attributes() == {
"cameraStreamAccessUrl": "https://example.com/api/streams/bla",
"cameraStreamReceiverAppId": "B45F4572",
}
async def test_onoff_group(hass: HomeAssistant) -> None:
"""Test OnOff trait support for group domain."""
assert helpers.get_google_type(group.DOMAIN, None) is not None
assert trait.OnOffTrait.supported(group.DOMAIN, 0, None, None)
trt_on = trait.OnOffTrait(hass, State("group.bla", STATE_ON), BASIC_CONFIG)
assert trt_on.sync_attributes() == {}
assert trt_on.query_attributes() == {"on": True}
trt_off = trait.OnOffTrait(hass, State("group.bla", STATE_OFF), BASIC_CONFIG)
assert trt_off.query_attributes() == {"on": False}
on_calls = async_mock_service(hass, HOMEASSISTANT_DOMAIN, SERVICE_TURN_ON)
await trt_on.execute(trait.COMMAND_ON_OFF, BASIC_DATA, {"on": True}, {})
assert len(on_calls) == 1
assert on_calls[0].data == {ATTR_ENTITY_ID: "group.bla"}
off_calls = async_mock_service(hass, HOMEASSISTANT_DOMAIN, SERVICE_TURN_OFF)
await trt_on.execute(trait.COMMAND_ON_OFF, BASIC_DATA, {"on": False}, {})
assert len(off_calls) == 1
assert off_calls[0].data == {ATTR_ENTITY_ID: "group.bla"}
async def test_onoff_input_boolean(hass: HomeAssistant) -> None:
"""Test OnOff trait support for input_boolean domain."""
assert helpers.get_google_type(input_boolean.DOMAIN, None) is not None
assert trait.OnOffTrait.supported(input_boolean.DOMAIN, 0, None, None)
trt_on = trait.OnOffTrait(hass, State("input_boolean.bla", STATE_ON), BASIC_CONFIG)
assert trt_on.sync_attributes() == {}
assert trt_on.query_attributes() == {"on": True}
trt_off = trait.OnOffTrait(
hass, State("input_boolean.bla", STATE_OFF), BASIC_CONFIG
)
assert trt_off.query_attributes() == {"on": False}
on_calls = async_mock_service(hass, input_boolean.DOMAIN, SERVICE_TURN_ON)
await trt_on.execute(trait.COMMAND_ON_OFF, BASIC_DATA, {"on": True}, {})
assert len(on_calls) == 1
assert on_calls[0].data == {ATTR_ENTITY_ID: "input_boolean.bla"}
off_calls = async_mock_service(hass, input_boolean.DOMAIN, SERVICE_TURN_OFF)
await trt_on.execute(trait.COMMAND_ON_OFF, BASIC_DATA, {"on": False}, {})
assert len(off_calls) == 1
assert off_calls[0].data == {ATTR_ENTITY_ID: "input_boolean.bla"}
@pytest.mark.freeze_time("2023-08-01T00:02:57+00:00")
async def test_doorbell_event(hass: HomeAssistant) -> None:
"""Test doorbell event trait support for event domain."""
assert trait.ObjectDetection.supported(event.DOMAIN, 0, "doorbell", None)
state = State(
"event.bla",
"2023-08-01T00:02:57+00:00",
attributes={"device_class": "doorbell"},
)
trt_od = trait.ObjectDetection(hass, state, BASIC_CONFIG)
assert not trt_od.sync_attributes()
assert trt_od.sync_options() == {"notificationSupportedByAgent": True}
assert not trt_od.query_attributes()
time_stamp = datetime.fromisoformat(state.state)
assert trt_od.query_notifications() == {
"ObjectDetection": {
"objects": {
"unclassified": 1,
},
"priority": 0,
"detectionTimestamp": int(time_stamp.timestamp() * 1000),
}
}
# Test that stale notifications (older than 30 s) are dropped
state = State(
"event.bla",
"2023-08-01T00:02:22+00:00",
attributes={"device_class": "doorbell"},
)
trt_od = trait.ObjectDetection(hass, state, BASIC_CONFIG)
assert trt_od.query_notifications() is None
async def test_onoff_switch(hass: HomeAssistant) -> None:
"""Test OnOff trait support for switch domain."""
assert helpers.get_google_type(switch.DOMAIN, None) is not None
assert trait.OnOffTrait.supported(switch.DOMAIN, 0, None, None)
trt_on = trait.OnOffTrait(hass, State("switch.bla", STATE_ON), BASIC_CONFIG)
assert trt_on.sync_attributes() == {}
assert trt_on.query_attributes() == {"on": True}
trt_off = trait.OnOffTrait(hass, State("switch.bla", STATE_OFF), BASIC_CONFIG)
assert trt_off.query_attributes() == {"on": False}
trt_assumed = trait.OnOffTrait(
hass, State("switch.bla", STATE_OFF, {"assumed_state": True}), BASIC_CONFIG
)
assert trt_assumed.sync_attributes() == {"commandOnlyOnOff": True}
on_calls = async_mock_service(hass, switch.DOMAIN, SERVICE_TURN_ON)
await trt_on.execute(trait.COMMAND_ON_OFF, BASIC_DATA, {"on": True}, {})
assert len(on_calls) == 1
assert on_calls[0].data == {ATTR_ENTITY_ID: "switch.bla"}
off_calls = async_mock_service(hass, switch.DOMAIN, SERVICE_TURN_OFF)
await trt_on.execute(trait.COMMAND_ON_OFF, BASIC_DATA, {"on": False}, {})
assert len(off_calls) == 1
assert off_calls[0].data == {ATTR_ENTITY_ID: "switch.bla"}
async def test_onoff_fan(hass: HomeAssistant) -> None:
"""Test OnOff trait support for fan domain."""
assert helpers.get_google_type(fan.DOMAIN, None) is not None
assert trait.OnOffTrait.supported(fan.DOMAIN, 0, None, None)
trt_on = trait.OnOffTrait(hass, State("fan.bla", STATE_ON), BASIC_CONFIG)
assert trt_on.sync_attributes() == {}
assert trt_on.query_attributes() == {"on": True}
trt_off = trait.OnOffTrait(hass, State("fan.bla", STATE_OFF), BASIC_CONFIG)
assert trt_off.query_attributes() == {"on": False}
on_calls = async_mock_service(hass, fan.DOMAIN, SERVICE_TURN_ON)
await trt_on.execute(trait.COMMAND_ON_OFF, BASIC_DATA, {"on": True}, {})
assert len(on_calls) == 1
assert on_calls[0].data == {ATTR_ENTITY_ID: "fan.bla"}
off_calls = async_mock_service(hass, fan.DOMAIN, SERVICE_TURN_OFF)
await trt_on.execute(trait.COMMAND_ON_OFF, BASIC_DATA, {"on": False}, {})
assert len(off_calls) == 1
assert off_calls[0].data == {ATTR_ENTITY_ID: "fan.bla"}
async def test_onoff_light(hass: HomeAssistant) -> None:
"""Test OnOff trait support for light domain."""
assert helpers.get_google_type(light.DOMAIN, None) is not None
assert trait.OnOffTrait.supported(light.DOMAIN, 0, None, None)
trt_on = trait.OnOffTrait(hass, State("light.bla", STATE_ON), BASIC_CONFIG)
assert trt_on.sync_attributes() == {}
assert trt_on.query_attributes() == {"on": True}
trt_off = trait.OnOffTrait(hass, State("light.bla", STATE_OFF), BASIC_CONFIG)
assert trt_off.query_attributes() == {"on": False}
on_calls = async_mock_service(hass, light.DOMAIN, SERVICE_TURN_ON)
await trt_on.execute(trait.COMMAND_ON_OFF, BASIC_DATA, {"on": True}, {})
assert len(on_calls) == 1
assert on_calls[0].data == {ATTR_ENTITY_ID: "light.bla"}
off_calls = async_mock_service(hass, light.DOMAIN, SERVICE_TURN_OFF)
await trt_on.execute(trait.COMMAND_ON_OFF, BASIC_DATA, {"on": False}, {})
assert len(off_calls) == 1
assert off_calls[0].data == {ATTR_ENTITY_ID: "light.bla"}
async def test_onoff_media_player(hass: HomeAssistant) -> None:
"""Test OnOff trait support for media_player domain."""
assert helpers.get_google_type(media_player.DOMAIN, None) is not None
assert trait.OnOffTrait.supported(media_player.DOMAIN, 0, None, None)
trt_on = trait.OnOffTrait(hass, State("media_player.bla", STATE_ON), BASIC_CONFIG)
assert trt_on.sync_attributes() == {}
assert trt_on.query_attributes() == {"on": True}
trt_off = trait.OnOffTrait(hass, State("media_player.bla", STATE_OFF), BASIC_CONFIG)
assert trt_off.query_attributes() == {"on": False}
on_calls = async_mock_service(hass, media_player.DOMAIN, SERVICE_TURN_ON)
await trt_on.execute(trait.COMMAND_ON_OFF, BASIC_DATA, {"on": True}, {})
assert len(on_calls) == 1
assert on_calls[0].data == {ATTR_ENTITY_ID: "media_player.bla"}
off_calls = async_mock_service(hass, media_player.DOMAIN, SERVICE_TURN_OFF)
await trt_on.execute(trait.COMMAND_ON_OFF, BASIC_DATA, {"on": False}, {})
assert len(off_calls) == 1
assert off_calls[0].data == {ATTR_ENTITY_ID: "media_player.bla"}
async def test_onoff_humidifier(hass: HomeAssistant) -> None:
"""Test OnOff trait support for humidifier domain."""
assert helpers.get_google_type(humidifier.DOMAIN, None) is not None
assert trait.OnOffTrait.supported(humidifier.DOMAIN, 0, None, None)
trt_on = trait.OnOffTrait(hass, State("humidifier.bla", STATE_ON), BASIC_CONFIG)
assert trt_on.sync_attributes() == {}
assert trt_on.query_attributes() == {"on": True}
trt_off = trait.OnOffTrait(hass, State("humidifier.bla", STATE_OFF), BASIC_CONFIG)
assert trt_off.query_attributes() == {"on": False}
on_calls = async_mock_service(hass, humidifier.DOMAIN, SERVICE_TURN_ON)
await trt_on.execute(trait.COMMAND_ON_OFF, BASIC_DATA, {"on": True}, {})
assert len(on_calls) == 1
assert on_calls[0].data == {ATTR_ENTITY_ID: "humidifier.bla"}
off_calls = async_mock_service(hass, humidifier.DOMAIN, SERVICE_TURN_OFF)
await trt_on.execute(trait.COMMAND_ON_OFF, BASIC_DATA, {"on": False}, {})
assert len(off_calls) == 1
assert off_calls[0].data == {ATTR_ENTITY_ID: "humidifier.bla"}
async def test_onoff_water_heater(hass: HomeAssistant) -> None:
"""Test OnOff trait support for water_heater domain."""
assert helpers.get_google_type(water_heater.DOMAIN, None) is not None
assert trait.OnOffTrait.supported(
water_heater.DOMAIN, WaterHeaterEntityFeature.ON_OFF, None, None
)
trt_on = trait.OnOffTrait(hass, State("water_heater.bla", STATE_ON), BASIC_CONFIG)
assert trt_on.sync_attributes() == {}
assert trt_on.query_attributes() == {"on": True}
trt_off = trait.OnOffTrait(hass, State("water_heater.bla", STATE_OFF), BASIC_CONFIG)
assert trt_off.query_attributes() == {"on": False}
on_calls = async_mock_service(hass, water_heater.DOMAIN, SERVICE_TURN_ON)
await trt_on.execute(trait.COMMAND_ON_OFF, BASIC_DATA, {"on": True}, {})
assert len(on_calls) == 1
assert on_calls[0].data == {ATTR_ENTITY_ID: "water_heater.bla"}
off_calls = async_mock_service(hass, water_heater.DOMAIN, SERVICE_TURN_OFF)
await trt_on.execute(trait.COMMAND_ON_OFF, BASIC_DATA, {"on": False}, {})
assert len(off_calls) == 1
assert off_calls[0].data == {ATTR_ENTITY_ID: "water_heater.bla"}
async def test_dock_vacuum(hass: HomeAssistant) -> None:
"""Test dock trait support for vacuum domain."""
assert helpers.get_google_type(vacuum.DOMAIN, None) is not None
assert trait.DockTrait.supported(vacuum.DOMAIN, 0, None, None)
trt = trait.DockTrait(hass, State("vacuum.bla", vacuum.STATE_IDLE), BASIC_CONFIG)
assert trt.sync_attributes() == {}
assert trt.query_attributes() == {"isDocked": False}
calls = async_mock_service(hass, vacuum.DOMAIN, vacuum.SERVICE_RETURN_TO_BASE)
await trt.execute(trait.COMMAND_DOCK, BASIC_DATA, {}, {})
assert len(calls) == 1
assert calls[0].data == {ATTR_ENTITY_ID: "vacuum.bla"}
async def test_locate_vacuum(hass: HomeAssistant) -> None:
"""Test locate trait support for vacuum domain."""
assert helpers.get_google_type(vacuum.DOMAIN, None) is not None
assert trait.LocatorTrait.supported(
vacuum.DOMAIN, VacuumEntityFeature.LOCATE, None, None
)
trt = trait.LocatorTrait(
hass,
State(
"vacuum.bla",
vacuum.STATE_IDLE,
{ATTR_SUPPORTED_FEATURES: VacuumEntityFeature.LOCATE},
),
BASIC_CONFIG,
)
assert trt.sync_attributes() == {}
assert trt.query_attributes() == {}
calls = async_mock_service(hass, vacuum.DOMAIN, vacuum.SERVICE_LOCATE)
await trt.execute(trait.COMMAND_LOCATE, BASIC_DATA, {"silence": False}, {})
assert len(calls) == 1
assert calls[0].data == {ATTR_ENTITY_ID: "vacuum.bla"}
with pytest.raises(helpers.SmartHomeError) as err:
await trt.execute(trait.COMMAND_LOCATE, BASIC_DATA, {"silence": True}, {})
assert err.value.code == const.ERR_FUNCTION_NOT_SUPPORTED
async def test_energystorage_vacuum(hass: HomeAssistant) -> None:
"""Test EnergyStorage trait support for vacuum domain."""
assert helpers.get_google_type(vacuum.DOMAIN, None) is not None
assert trait.EnergyStorageTrait.supported(
vacuum.DOMAIN, VacuumEntityFeature.BATTERY, None, None
)
trt = trait.EnergyStorageTrait(
hass,
State(
"vacuum.bla",
vacuum.STATE_DOCKED,
{
ATTR_SUPPORTED_FEATURES: VacuumEntityFeature.BATTERY,
ATTR_BATTERY_LEVEL: 100,
},
),
BASIC_CONFIG,
)
assert trt.sync_attributes() == {
"isRechargeable": True,
"queryOnlyEnergyStorage": True,
}
assert trt.query_attributes() == {
"descriptiveCapacityRemaining": "FULL",
"capacityRemaining": [{"rawValue": 100, "unit": "PERCENTAGE"}],
"capacityUntilFull": [{"rawValue": 0, "unit": "PERCENTAGE"}],
"isCharging": True,
"isPluggedIn": True,
}
trt = trait.EnergyStorageTrait(
hass,
State(
"vacuum.bla",
vacuum.STATE_CLEANING,
{
ATTR_SUPPORTED_FEATURES: VacuumEntityFeature.BATTERY,
ATTR_BATTERY_LEVEL: 20,
},
),
BASIC_CONFIG,
)
assert trt.sync_attributes() == {
"isRechargeable": True,
"queryOnlyEnergyStorage": True,
}
assert trt.query_attributes() == {
"descriptiveCapacityRemaining": "CRITICALLY_LOW",
"capacityRemaining": [{"rawValue": 20, "unit": "PERCENTAGE"}],
"capacityUntilFull": [{"rawValue": 80, "unit": "PERCENTAGE"}],
"isCharging": False,
"isPluggedIn": False,
}
with pytest.raises(helpers.SmartHomeError) as err:
await trt.execute(trait.COMMAND_CHARGE, BASIC_DATA, {"charge": True}, {})
assert err.value.code == const.ERR_FUNCTION_NOT_SUPPORTED
with pytest.raises(helpers.SmartHomeError) as err:
await trt.execute(trait.COMMAND_CHARGE, BASIC_DATA, {"charge": False}, {})
assert err.value.code == const.ERR_FUNCTION_NOT_SUPPORTED
async def test_startstop_vacuum(hass: HomeAssistant) -> None:
"""Test startStop trait support for vacuum domain."""
assert helpers.get_google_type(vacuum.DOMAIN, None) is not None
assert trait.StartStopTrait.supported(vacuum.DOMAIN, 0, None, None)
trt = trait.StartStopTrait(
hass,
State(
"vacuum.bla",
vacuum.STATE_PAUSED,
{ATTR_SUPPORTED_FEATURES: VacuumEntityFeature.PAUSE},
),
BASIC_CONFIG,
)
assert trt.sync_attributes() == {"pausable": True}
assert trt.query_attributes() == {"isRunning": False, "isPaused": True}
start_calls = async_mock_service(hass, vacuum.DOMAIN, vacuum.SERVICE_START)
await trt.execute(trait.COMMAND_START_STOP, BASIC_DATA, {"start": True}, {})
assert len(start_calls) == 1
assert start_calls[0].data == {ATTR_ENTITY_ID: "vacuum.bla"}
stop_calls = async_mock_service(hass, vacuum.DOMAIN, vacuum.SERVICE_STOP)
await trt.execute(trait.COMMAND_START_STOP, BASIC_DATA, {"start": False}, {})
assert len(stop_calls) == 1
assert stop_calls[0].data == {ATTR_ENTITY_ID: "vacuum.bla"}
pause_calls = async_mock_service(hass, vacuum.DOMAIN, vacuum.SERVICE_PAUSE)
await trt.execute(trait.COMMAND_PAUSE_UNPAUSE, BASIC_DATA, {"pause": True}, {})
assert len(pause_calls) == 1
assert pause_calls[0].data == {ATTR_ENTITY_ID: "vacuum.bla"}
unpause_calls = async_mock_service(hass, vacuum.DOMAIN, vacuum.SERVICE_START)
await trt.execute(trait.COMMAND_PAUSE_UNPAUSE, BASIC_DATA, {"pause": False}, {})
assert len(unpause_calls) == 1
assert unpause_calls[0].data == {ATTR_ENTITY_ID: "vacuum.bla"}
@pytest.mark.parametrize(
(
"domain",
"state_open",
"state_closed",
"state_opening",
"state_closing",
"supported_features",
"service_close",
"service_open",
"service_stop",
"service_toggle",
),
[
(
cover.DOMAIN,
cover.STATE_OPEN,
cover.STATE_CLOSED,
cover.STATE_OPENING,
cover.STATE_CLOSING,
CoverEntityFeature.STOP
| CoverEntityFeature.OPEN
| CoverEntityFeature.CLOSE,
cover.SERVICE_OPEN_COVER,
cover.SERVICE_CLOSE_COVER,
cover.SERVICE_STOP_COVER,
cover.SERVICE_TOGGLE,
),
(
valve.DOMAIN,
valve.ValveState.OPEN,
valve.ValveState.CLOSED,
valve.ValveState.OPENING,
valve.ValveState.CLOSING,
ValveEntityFeature.STOP
| ValveEntityFeature.OPEN
| ValveEntityFeature.CLOSE,
valve.SERVICE_OPEN_VALVE,
valve.SERVICE_CLOSE_VALVE,
valve.SERVICE_STOP_VALVE,
cover.SERVICE_TOGGLE,
),
],
)
async def test_startstop_cover_valve(
hass: HomeAssistant,
domain: str,
state_open: str,
state_closed: str,
state_opening: str,
state_closing: str,
supported_features: str,
service_open: str,
service_close: str,
service_stop: str,
service_toggle: str,
) -> None:
"""Test startStop trait support."""
assert helpers.get_google_type(domain, None) is not None
assert trait.StartStopTrait.supported(domain, supported_features, None, None)
state = State(
f"{domain}.bla",
state_closed,
{ATTR_SUPPORTED_FEATURES: supported_features},
)
trt = trait.StartStopTrait(
hass,
state,
BASIC_CONFIG,
)
assert trt.sync_attributes() == {}
for state_value in (state_closing, state_opening):
state.state = state_value
assert trt.query_attributes() == {"isRunning": True}
stop_calls = async_mock_service(hass, domain, service_stop)
open_calls = async_mock_service(hass, domain, service_open)
close_calls = async_mock_service(hass, domain, service_close)
toggle_calls = async_mock_service(hass, domain, service_toggle)
await trt.execute(trait.COMMAND_START_STOP, BASIC_DATA, {"start": False}, {})
assert len(stop_calls) == 1
assert stop_calls[0].data == {ATTR_ENTITY_ID: f"{domain}.bla"}
for state_value in (state_closed, state_open):
state.state = state_value
assert trt.query_attributes() == {"isRunning": False}
for state_value in (state_closing, state_opening):
state.state = state_value
assert trt.query_attributes() == {"isRunning": True}
state.state = state_open
with pytest.raises(
SmartHomeError, match=f"{domain.capitalize()} is already stopped"
):
await trt.execute(trait.COMMAND_START_STOP, BASIC_DATA, {"start": False}, {})
# Start triggers toggle open
state.state = state_closed
await trt.execute(trait.COMMAND_START_STOP, BASIC_DATA, {"start": True}, {})
assert len(open_calls) == 0
assert len(close_calls) == 0
assert len(toggle_calls) == 1
assert toggle_calls[0].data == {ATTR_ENTITY_ID: f"{domain}.bla"}
# Second start triggers toggle close
state.state = state_open
await trt.execute(trait.COMMAND_START_STOP, BASIC_DATA, {"start": True}, {})
assert len(open_calls) == 0
assert len(close_calls) == 0
assert len(toggle_calls) == 2
assert toggle_calls[1].data == {ATTR_ENTITY_ID: f"{domain}.bla"}
state.state = state_closed
with pytest.raises(
SmartHomeError,
match="Command action.devices.commands.PauseUnpause is not supported",
):
await trt.execute(trait.COMMAND_PAUSE_UNPAUSE, BASIC_DATA, {"start": True}, {})
@pytest.mark.parametrize(
(
"domain",
"state_open",
"state_closed",
"state_opening",
"state_closing",
"supported_features",
"service_close",
"service_open",
"service_stop",
"service_toggle",
),
[
(
cover.DOMAIN,
cover.STATE_OPEN,
cover.STATE_CLOSED,
cover.STATE_OPENING,
cover.STATE_CLOSING,
CoverEntityFeature.STOP
| CoverEntityFeature.OPEN
| CoverEntityFeature.CLOSE,
cover.SERVICE_OPEN_COVER,
cover.SERVICE_CLOSE_COVER,
cover.SERVICE_STOP_COVER,
cover.SERVICE_TOGGLE,
),
(
valve.DOMAIN,
valve.ValveState.OPEN,
valve.ValveState.CLOSED,
valve.ValveState.OPENING,
valve.ValveState.CLOSING,
ValveEntityFeature.STOP
| ValveEntityFeature.OPEN
| ValveEntityFeature.CLOSE,
valve.SERVICE_OPEN_VALVE,
valve.SERVICE_CLOSE_VALVE,
valve.SERVICE_STOP_VALVE,
cover.SERVICE_TOGGLE,
),
],
)
async def test_startstop_cover_valve_assumed(
hass: HomeAssistant,
domain: str,
state_open: str,
state_closed: str,
state_opening: str,
state_closing: str,
supported_features: str,
service_open: str,
service_close: str,
service_stop: str,
service_toggle: str,
) -> None:
"""Test startStop trait support for cover domain of assumed state."""
trt = trait.StartStopTrait(
hass,
State(
f"{domain}.bla",
state_closed,
{
ATTR_SUPPORTED_FEATURES: supported_features,
ATTR_ASSUMED_STATE: True,
},
),
BASIC_CONFIG,
)
stop_calls = async_mock_service(hass, domain, service_stop)
toggle_calls = async_mock_service(hass, domain, service_toggle)
await trt.execute(trait.COMMAND_START_STOP, BASIC_DATA, {"start": False}, {})
assert len(stop_calls) == 1
assert len(toggle_calls) == 0
assert stop_calls[0].data == {ATTR_ENTITY_ID: f"{domain}.bla"}
stop_calls.clear()
await trt.execute(trait.COMMAND_START_STOP, BASIC_DATA, {"start": True}, {})
assert len(stop_calls) == 0
assert len(toggle_calls) == 1
assert toggle_calls[0].data == {ATTR_ENTITY_ID: f"{domain}.bla"}
@pytest.mark.parametrize("supported_color_modes", [["hs"], ["rgb"], ["xy"]])
async def test_color_setting_color_light(
hass: HomeAssistant, supported_color_modes
) -> None:
"""Test ColorSpectrum trait support for light domain."""
assert helpers.get_google_type(light.DOMAIN, None) is not None
assert not trait.ColorSettingTrait.supported(light.DOMAIN, 0, None, {})
assert trait.ColorSettingTrait.supported(
light.DOMAIN, 0, None, {"supported_color_modes": supported_color_modes}
)
trt = trait.ColorSettingTrait(
hass,
State(
"light.bla",
STATE_ON,
{
light.ATTR_HS_COLOR: (20, 94),
light.ATTR_BRIGHTNESS: 200,
light.ATTR_COLOR_MODE: "hs",
"supported_color_modes": supported_color_modes,
},
),
BASIC_CONFIG,
)
assert trt.sync_attributes() == {"colorModel": "hsv"}
assert trt.query_attributes() == {
"color": {"spectrumHsv": {"hue": 20, "saturation": 0.94, "value": 200 / 255}}
}
assert trt.can_execute(
trait.COMMAND_COLOR_ABSOLUTE, {"color": {"spectrumRGB": 16715792}}
)
calls = async_mock_service(hass, light.DOMAIN, SERVICE_TURN_ON)
await trt.execute(
trait.COMMAND_COLOR_ABSOLUTE,
BASIC_DATA,
{"color": {"spectrumRGB": 1052927}},
{},
)
assert len(calls) == 1
assert calls[0].data == {
ATTR_ENTITY_ID: "light.bla",
light.ATTR_HS_COLOR: (240, 93.725),
}
await trt.execute(
trait.COMMAND_COLOR_ABSOLUTE,
BASIC_DATA,
{"color": {"spectrumHSV": {"hue": 100, "saturation": 0.50, "value": 0.20}}},
{},
)
assert len(calls) == 2
assert calls[1].data == {
ATTR_ENTITY_ID: "light.bla",
light.ATTR_HS_COLOR: [100, 50],
light.ATTR_BRIGHTNESS: 0.2 * 255,
}
async def test_color_setting_temperature_light(hass: HomeAssistant) -> None:
"""Test ColorTemperature trait support for light domain."""
assert helpers.get_google_type(light.DOMAIN, None) is not None
assert not trait.ColorSettingTrait.supported(light.DOMAIN, 0, None, {})
assert trait.ColorSettingTrait.supported(
light.DOMAIN, 0, None, {"supported_color_modes": ["color_temp"]}
)
trt = trait.ColorSettingTrait(
hass,
State(
"light.bla",
STATE_ON,
{
light.ATTR_MIN_MIREDS: 200,
light.ATTR_COLOR_MODE: "color_temp",
light.ATTR_COLOR_TEMP: 300,
light.ATTR_MAX_MIREDS: 500,
"supported_color_modes": ["color_temp"],
},
),
BASIC_CONFIG,
)
assert trt.sync_attributes() == {
"colorTemperatureRange": {"temperatureMinK": 2000, "temperatureMaxK": 5000}
}
assert trt.query_attributes() == {"color": {"temperatureK": 3333}}
assert trt.can_execute(
trait.COMMAND_COLOR_ABSOLUTE, {"color": {"temperature": 400}}
)
calls = async_mock_service(hass, light.DOMAIN, SERVICE_TURN_ON)
with pytest.raises(helpers.SmartHomeError) as err:
await trt.execute(
trait.COMMAND_COLOR_ABSOLUTE,
BASIC_DATA,
{"color": {"temperature": 5555}},
{},
)
assert err.value.code == const.ERR_VALUE_OUT_OF_RANGE
await trt.execute(
trait.COMMAND_COLOR_ABSOLUTE, BASIC_DATA, {"color": {"temperature": 2857}}, {}
)
assert len(calls) == 1
assert calls[0].data == {
ATTR_ENTITY_ID: "light.bla",
light.ATTR_COLOR_TEMP: color.color_temperature_kelvin_to_mired(2857),
}
async def test_color_light_temperature_light_bad_temp(hass: HomeAssistant) -> None:
"""Test ColorTemperature trait support for light domain."""
assert helpers.get_google_type(light.DOMAIN, None) is not None
assert not trait.ColorSettingTrait.supported(light.DOMAIN, 0, None, {})
assert trait.ColorSettingTrait.supported(
light.DOMAIN, 0, None, {"supported_color_modes": ["color_temp"]}
)
trt = trait.ColorSettingTrait(
hass,
State(
"light.bla",
STATE_ON,
{
light.ATTR_MIN_MIREDS: 200,
light.ATTR_COLOR_TEMP: 0,
light.ATTR_MAX_MIREDS: 500,
},
),
BASIC_CONFIG,
)
assert trt.query_attributes() == {}
async def test_light_modes(hass: HomeAssistant) -> None:
"""Test Light Mode trait."""
assert helpers.get_google_type(light.DOMAIN, None) is not None
assert trait.ModesTrait.supported(
light.DOMAIN, LightEntityFeature.EFFECT, None, None
)
trt = trait.ModesTrait(
hass,
State(
"light.living_room",
light.STATE_ON,
attributes={
light.ATTR_EFFECT_LIST: ["random", "colorloop"],
light.ATTR_EFFECT: "random",
},
),
BASIC_CONFIG,
)
attribs = trt.sync_attributes()
assert attribs == {
"availableModes": [
{
"name": "effect",
"name_values": [{"name_synonym": ["effect"], "lang": "en"}],
"settings": [
{
"setting_name": "random",
"setting_values": [
{"setting_synonym": ["random"], "lang": "en"}
],
},
{
"setting_name": "colorloop",
"setting_values": [
{"setting_synonym": ["colorloop"], "lang": "en"}
],
},
],
"ordered": False,
}
]
}
assert trt.query_attributes() == {
"currentModeSettings": {"effect": "random"},
"on": True,
}
assert trt.can_execute(
trait.COMMAND_SET_MODES,
params={"updateModeSettings": {"effect": "colorloop"}},
)
calls = async_mock_service(hass, light.DOMAIN, SERVICE_TURN_ON)
await trt.execute(
trait.COMMAND_SET_MODES,
BASIC_DATA,
{"updateModeSettings": {"effect": "colorloop"}},
{},
)
assert len(calls) == 1
assert calls[0].data == {
"entity_id": "light.living_room",
"effect": "colorloop",
}
@pytest.mark.parametrize(
"component",
[button, input_button],
)
async def test_scene_button(hass: HomeAssistant, component) -> None:
"""Test Scene trait support for the (input) button domain."""
assert helpers.get_google_type(component.DOMAIN, None) is not None
assert trait.SceneTrait.supported(component.DOMAIN, 0, None, None)
trt = trait.SceneTrait(
hass, State(f"{component.DOMAIN}.bla", STATE_UNKNOWN), BASIC_CONFIG
)
assert trt.sync_attributes() == {}
assert trt.query_attributes() == {}
assert trt.can_execute(trait.COMMAND_ACTIVATE_SCENE, {})
calls = async_mock_service(hass, component.DOMAIN, component.SERVICE_PRESS)
await trt.execute(trait.COMMAND_ACTIVATE_SCENE, BASIC_DATA, {}, {})
# We don't wait till button press is done.
await hass.async_block_till_done()
assert len(calls) == 1
assert calls[0].data == {ATTR_ENTITY_ID: f"{component.DOMAIN}.bla"}
async def test_scene_scene(hass: HomeAssistant) -> None:
"""Test Scene trait support for scene domain."""
assert helpers.get_google_type(scene.DOMAIN, None) is not None
assert trait.SceneTrait.supported(scene.DOMAIN, 0, None, None)
trt = trait.SceneTrait(hass, State("scene.bla", STATE_UNKNOWN), BASIC_CONFIG)
assert trt.sync_attributes() == {}
assert trt.query_attributes() == {}
assert trt.can_execute(trait.COMMAND_ACTIVATE_SCENE, {})
calls = async_mock_service(hass, scene.DOMAIN, SERVICE_TURN_ON)
await trt.execute(trait.COMMAND_ACTIVATE_SCENE, BASIC_DATA, {}, {})
assert len(calls) == 1
assert calls[0].data == {ATTR_ENTITY_ID: "scene.bla"}
async def test_scene_script(hass: HomeAssistant) -> None:
"""Test Scene trait support for script domain."""
assert helpers.get_google_type(script.DOMAIN, None) is not None
assert trait.SceneTrait.supported(script.DOMAIN, 0, None, None)
trt = trait.SceneTrait(hass, State("script.bla", STATE_OFF), BASIC_CONFIG)
assert trt.sync_attributes() == {}
assert trt.query_attributes() == {}
assert trt.can_execute(trait.COMMAND_ACTIVATE_SCENE, {})
calls = async_mock_service(hass, script.DOMAIN, SERVICE_TURN_ON)
await trt.execute(trait.COMMAND_ACTIVATE_SCENE, BASIC_DATA, {}, {})
# We don't wait till script execution is done.
await hass.async_block_till_done()
assert len(calls) == 1
assert calls[0].data == {ATTR_ENTITY_ID: "script.bla"}
async def test_temperature_setting_climate_onoff(hass: HomeAssistant) -> None:
"""Test TemperatureSetting trait support for climate domain - range."""
assert helpers.get_google_type(climate.DOMAIN, None) is not None
assert trait.TemperatureSettingTrait.supported(climate.DOMAIN, 0, None, None)
hass.config.units.temperature_unit = UnitOfTemperature.FAHRENHEIT
trt = trait.TemperatureSettingTrait(
hass,
State(
"climate.bla",
climate.HVACMode.AUTO,
{
ATTR_SUPPORTED_FEATURES: ClimateEntityFeature.TARGET_TEMPERATURE_RANGE
| ClimateEntityFeature.TURN_ON
| ClimateEntityFeature.TURN_OFF,
climate.ATTR_HVAC_MODES: [
climate.HVACMode.OFF,
climate.HVACMode.COOL,
climate.HVACMode.HEAT,
climate.HVACMode.HEAT_COOL,
],
climate.ATTR_MIN_TEMP: 45,
climate.ATTR_MAX_TEMP: 95,
},
),
BASIC_CONFIG,
)
assert trt.sync_attributes() == {
"availableThermostatModes": ["off", "cool", "heat", "heatcool", "on"],
"thermostatTemperatureRange": {
"minThresholdCelsius": 7,
"maxThresholdCelsius": 35,
},
"thermostatTemperatureUnit": "F",
}
assert trt.can_execute(trait.COMMAND_THERMOSTAT_SET_MODE, {})
calls = async_mock_service(hass, climate.DOMAIN, SERVICE_TURN_ON)
await trt.execute(
trait.COMMAND_THERMOSTAT_SET_MODE, BASIC_DATA, {"thermostatMode": "on"}, {}
)
assert len(calls) == 1
calls = async_mock_service(hass, climate.DOMAIN, SERVICE_TURN_OFF)
await trt.execute(
trait.COMMAND_THERMOSTAT_SET_MODE, BASIC_DATA, {"thermostatMode": "off"}, {}
)
assert len(calls) == 1
async def test_temperature_setting_climate_no_modes(hass: HomeAssistant) -> None:
"""Test TemperatureSetting trait support for climate domain not supporting any modes."""
assert helpers.get_google_type(climate.DOMAIN, None) is not None
assert trait.TemperatureSettingTrait.supported(climate.DOMAIN, 0, None, None)
hass.config.units.temperature_unit = UnitOfTemperature.CELSIUS
trt = trait.TemperatureSettingTrait(
hass,
State(
"climate.bla",
climate.HVACMode.AUTO,
{
climate.ATTR_HVAC_MODES: [],
climate.ATTR_MIN_TEMP: climate.DEFAULT_MIN_TEMP,
climate.ATTR_MAX_TEMP: climate.DEFAULT_MAX_TEMP,
},
),
BASIC_CONFIG,
)
assert trt.sync_attributes() == {
"availableThermostatModes": ["heat"],
"thermostatTemperatureRange": {
"minThresholdCelsius": climate.DEFAULT_MIN_TEMP,
"maxThresholdCelsius": climate.DEFAULT_MAX_TEMP,
},
"thermostatTemperatureUnit": "C",
}
async def test_temperature_setting_climate_range(hass: HomeAssistant) -> None:
"""Test TemperatureSetting trait support for climate domain - range."""
assert helpers.get_google_type(climate.DOMAIN, None) is not None
assert trait.TemperatureSettingTrait.supported(climate.DOMAIN, 0, None, None)
hass.config.units.temperature_unit = UnitOfTemperature.FAHRENHEIT
trt = trait.TemperatureSettingTrait(
hass,
State(
"climate.bla",
climate.HVACMode.AUTO,
{
climate.ATTR_CURRENT_TEMPERATURE: 70,
climate.ATTR_CURRENT_HUMIDITY: 25,
ATTR_SUPPORTED_FEATURES: ClimateEntityFeature.TARGET_TEMPERATURE_RANGE
| ClimateEntityFeature.TURN_ON
| ClimateEntityFeature.TURN_OFF,
climate.ATTR_HVAC_MODES: [
STATE_OFF,
climate.HVACMode.COOL,
climate.HVACMode.HEAT,
climate.HVACMode.AUTO,
],
climate.ATTR_TARGET_TEMP_HIGH: 75,
climate.ATTR_TARGET_TEMP_LOW: 65,
climate.ATTR_MIN_TEMP: 50,
climate.ATTR_MAX_TEMP: 80,
},
),
BASIC_CONFIG,
)
assert trt.sync_attributes() == {
"availableThermostatModes": ["off", "cool", "heat", "auto", "on"],
"thermostatTemperatureRange": {
"minThresholdCelsius": 10,
"maxThresholdCelsius": 27,
},
"thermostatTemperatureUnit": "F",
}
assert trt.query_attributes() == {
"thermostatMode": "auto",
"thermostatTemperatureAmbient": 21.1,
"thermostatHumidityAmbient": 25,
"thermostatTemperatureSetpointLow": 18.3,
"thermostatTemperatureSetpointHigh": 23.9,
}
assert trt.can_execute(trait.COMMAND_THERMOSTAT_TEMPERATURE_SET_RANGE, {})
assert trt.can_execute(trait.COMMAND_THERMOSTAT_SET_MODE, {})
calls = async_mock_service(hass, climate.DOMAIN, climate.SERVICE_SET_TEMPERATURE)
await trt.execute(
trait.COMMAND_THERMOSTAT_TEMPERATURE_SET_RANGE,
BASIC_DATA,
{
"thermostatTemperatureSetpointHigh": 25,
"thermostatTemperatureSetpointLow": 20,
},
{},
)
assert len(calls) == 1
assert calls[0].data == {
ATTR_ENTITY_ID: "climate.bla",
climate.ATTR_TARGET_TEMP_HIGH: 77,
climate.ATTR_TARGET_TEMP_LOW: 68,
}
calls = async_mock_service(hass, climate.DOMAIN, climate.SERVICE_SET_HVAC_MODE)
await trt.execute(
trait.COMMAND_THERMOSTAT_SET_MODE, BASIC_DATA, {"thermostatMode": "cool"}, {}
)
assert len(calls) == 1
assert calls[0].data == {
ATTR_ENTITY_ID: "climate.bla",
climate.ATTR_HVAC_MODE: climate.HVACMode.COOL,
}
with pytest.raises(helpers.SmartHomeError) as err:
await trt.execute(
trait.COMMAND_THERMOSTAT_TEMPERATURE_SET_RANGE,
BASIC_DATA,
{
"thermostatTemperatureSetpointHigh": 26,
"thermostatTemperatureSetpointLow": -100,
},
{},
)
assert err.value.code == const.ERR_VALUE_OUT_OF_RANGE
with pytest.raises(helpers.SmartHomeError) as err:
await trt.execute(
trait.COMMAND_THERMOSTAT_TEMPERATURE_SET_RANGE,
BASIC_DATA,
{
"thermostatTemperatureSetpointHigh": 100,
"thermostatTemperatureSetpointLow": 18,
},
{},
)
assert err.value.code == const.ERR_VALUE_OUT_OF_RANGE
calls = async_mock_service(hass, climate.DOMAIN, climate.SERVICE_SET_TEMPERATURE)
await trt.execute(
trait.COMMAND_THERMOSTAT_TEMPERATURE_SETPOINT,
BASIC_DATA,
{"thermostatTemperatureSetpoint": 23.9},
{},
)
assert len(calls) == 1
assert calls[0].data == {
ATTR_ENTITY_ID: "climate.bla",
climate.ATTR_TEMPERATURE: 75,
}
hass.config.units.temperature_unit = UnitOfTemperature.CELSIUS
async def test_temperature_setting_climate_setpoint(hass: HomeAssistant) -> None:
"""Test TemperatureSetting trait support for climate domain - setpoint."""
assert helpers.get_google_type(climate.DOMAIN, None) is not None
assert trait.TemperatureSettingTrait.supported(climate.DOMAIN, 0, None, None)
hass.config.units.temperature_unit = UnitOfTemperature.CELSIUS
trt = trait.TemperatureSettingTrait(
hass,
State(
"climate.bla",
climate.HVACMode.COOL,
{
ATTR_SUPPORTED_FEATURES: ClimateEntityFeature.TARGET_TEMPERATURE
| ClimateEntityFeature.TURN_ON
| ClimateEntityFeature.TURN_OFF,
climate.ATTR_HVAC_MODES: [STATE_OFF, climate.HVACMode.COOL],
climate.ATTR_MIN_TEMP: 10,
climate.ATTR_MAX_TEMP: 30,
climate.ATTR_PRESET_MODE: climate.PRESET_ECO,
ATTR_TEMPERATURE: 18,
climate.ATTR_CURRENT_TEMPERATURE: 20,
},
),
BASIC_CONFIG,
)
assert trt.sync_attributes() == {
"availableThermostatModes": ["off", "cool", "on"],
"thermostatTemperatureRange": {
"minThresholdCelsius": 10,
"maxThresholdCelsius": 30,
},
"thermostatTemperatureUnit": "C",
}
assert trt.query_attributes() == {
"thermostatMode": "eco",
"thermostatTemperatureAmbient": 20,
"thermostatTemperatureSetpoint": 18,
}
assert trt.can_execute(trait.COMMAND_THERMOSTAT_TEMPERATURE_SETPOINT, {})
assert trt.can_execute(trait.COMMAND_THERMOSTAT_SET_MODE, {})
calls = async_mock_service(hass, climate.DOMAIN, climate.SERVICE_SET_TEMPERATURE)
with pytest.raises(helpers.SmartHomeError):
await trt.execute(
trait.COMMAND_THERMOSTAT_TEMPERATURE_SETPOINT,
BASIC_DATA,
{"thermostatTemperatureSetpoint": -100},
{},
)
await trt.execute(
trait.COMMAND_THERMOSTAT_TEMPERATURE_SETPOINT,
BASIC_DATA,
{"thermostatTemperatureSetpoint": 19},
{},
)
assert len(calls) == 1
assert calls[0].data == {ATTR_ENTITY_ID: "climate.bla", ATTR_TEMPERATURE: 19}
calls = async_mock_service(hass, climate.DOMAIN, climate.SERVICE_SET_PRESET_MODE)
await trt.execute(
trait.COMMAND_THERMOSTAT_SET_MODE,
BASIC_DATA,
{"thermostatMode": "eco"},
{},
)
assert len(calls) == 1
assert calls[0].data == {
ATTR_ENTITY_ID: "climate.bla",
climate.ATTR_PRESET_MODE: "eco",
}
calls = async_mock_service(hass, climate.DOMAIN, climate.SERVICE_SET_TEMPERATURE)
await trt.execute(
trait.COMMAND_THERMOSTAT_TEMPERATURE_SET_RANGE,
BASIC_DATA,
{
"thermostatTemperatureSetpointHigh": 15,
"thermostatTemperatureSetpointLow": 22,
},
{},
)
assert len(calls) == 1
assert calls[0].data == {ATTR_ENTITY_ID: "climate.bla", ATTR_TEMPERATURE: 18.5}
async def test_temperature_setting_climate_setpoint_auto(hass: HomeAssistant) -> None:
"""Test TemperatureSetting trait support for climate domain.
Setpoint in auto mode.
"""
hass.config.units.temperature_unit = UnitOfTemperature.CELSIUS
trt = trait.TemperatureSettingTrait(
hass,
State(
"climate.bla",
climate.HVACMode.HEAT_COOL,
{
climate.ATTR_HVAC_MODES: [
climate.HVACMode.OFF,
climate.HVACMode.HEAT_COOL,
],
climate.ATTR_MIN_TEMP: 10,
climate.ATTR_MAX_TEMP: 30,
ATTR_TEMPERATURE: 18,
climate.ATTR_CURRENT_TEMPERATURE: 20,
},
),
BASIC_CONFIG,
)
assert trt.sync_attributes() == {
"availableThermostatModes": ["off", "heatcool", "on"],
"thermostatTemperatureRange": {
"minThresholdCelsius": 10,
"maxThresholdCelsius": 30,
},
"thermostatTemperatureUnit": "C",
}
assert trt.query_attributes() == {
"thermostatMode": "heatcool",
"thermostatTemperatureAmbient": 20,
"thermostatTemperatureSetpointHigh": 18,
"thermostatTemperatureSetpointLow": 18,
}
assert trt.can_execute(trait.COMMAND_THERMOSTAT_TEMPERATURE_SETPOINT, {})
assert trt.can_execute(trait.COMMAND_THERMOSTAT_SET_MODE, {})
calls = async_mock_service(hass, climate.DOMAIN, climate.SERVICE_SET_TEMPERATURE)
await trt.execute(
trait.COMMAND_THERMOSTAT_TEMPERATURE_SETPOINT,
BASIC_DATA,
{"thermostatTemperatureSetpoint": 19},
{},
)
assert len(calls) == 1
assert calls[0].data == {ATTR_ENTITY_ID: "climate.bla", ATTR_TEMPERATURE: 19}
async def test_temperature_control(hass: HomeAssistant) -> None:
"""Test TemperatureControl trait support for sensor domain."""
hass.config.units.temperature_unit = UnitOfTemperature.CELSIUS
trt = trait.TemperatureControlTrait(
hass,
State("sensor.temp", 18),
BASIC_CONFIG,
)
assert trt.sync_attributes() == {
"queryOnlyTemperatureControl": True,
"temperatureUnitForUX": "C",
"temperatureRange": {"maxThresholdCelsius": 100, "minThresholdCelsius": -100},
}
assert trt.query_attributes() == {
"temperatureSetpointCelsius": 18,
"temperatureAmbientCelsius": 18,
}
with pytest.raises(helpers.SmartHomeError) as err:
await trt.execute(trait.COMMAND_ON_OFF, BASIC_DATA, {"on": False}, {})
assert err.value.code == const.ERR_NOT_SUPPORTED
@pytest.mark.parametrize(
("unit_in", "unit_out", "temp_in", "temp_out", "current_in", "current_out"),
[
(UnitOfTemperature.CELSIUS, "C", "120", 120, "130", 130),
(UnitOfTemperature.FAHRENHEIT, "F", "248", 120, "266", 130),
],
)
async def test_temperature_control_water_heater(
hass: HomeAssistant,
unit_in: UnitOfTemperature,
unit_out: str,
temp_in: str,
temp_out: float,
current_in: str,
current_out: float,
) -> None:
"""Test TemperatureControl trait support for water heater domain."""
hass.config.units.temperature_unit = unit_in
min_temp = TemperatureConverter.convert(
water_heater.DEFAULT_MIN_TEMP,
UnitOfTemperature.CELSIUS,
unit_in,
)
max_temp = TemperatureConverter.convert(
water_heater.DEFAULT_MAX_TEMP,
UnitOfTemperature.CELSIUS,
unit_in,
)
trt = trait.TemperatureControlTrait(
hass,
State(
"water_heater.bla",
"attributes",
{
"min_temp": min_temp,
"max_temp": max_temp,
"temperature": temp_in,
"current_temperature": current_in,
},
),
BASIC_CONFIG,
)
assert trt.sync_attributes() == {
"temperatureUnitForUX": unit_out,
"temperatureRange": {
"maxThresholdCelsius": water_heater.DEFAULT_MAX_TEMP,
"minThresholdCelsius": water_heater.DEFAULT_MIN_TEMP,
},
}
assert trt.query_attributes() == {
"temperatureSetpointCelsius": temp_out,
"temperatureAmbientCelsius": current_out,
}
@pytest.mark.parametrize(
("unit", "temp_init", "temp_in", "temp_out", "current_init"),
[
(UnitOfTemperature.CELSIUS, "180", 220, 220, "180"),
(UnitOfTemperature.FAHRENHEIT, "356", 220, 428, "356"),
],
)
async def test_temperature_control_water_heater_set_temperature(
hass: HomeAssistant,
unit: UnitOfTemperature,
temp_init: str,
temp_in: float,
temp_out: float,
current_init: str,
) -> None:
"""Test TemperatureControl trait support for water heater domain - SetTemperature."""
hass.config.units.temperature_unit = unit
min_temp = TemperatureConverter.convert(
40,
UnitOfTemperature.CELSIUS,
unit,
)
max_temp = TemperatureConverter.convert(
230,
UnitOfTemperature.CELSIUS,
unit,
)
trt = trait.TemperatureControlTrait(
hass,
State(
"water_heater.bla",
"attributes",
{
"min_temp": min_temp,
"max_temp": max_temp,
"temperature": temp_init,
"current_temperature": current_init,
},
),
BASIC_CONFIG,
)
assert trt.can_execute(trait.COMMAND_SET_TEMPERATURE, {})
calls = async_mock_service(
hass, water_heater.DOMAIN, water_heater.SERVICE_SET_TEMPERATURE
)
with pytest.raises(helpers.SmartHomeError):
await trt.execute(
trait.COMMAND_SET_TEMPERATURE,
BASIC_DATA,
{"temperature": -100},
{},
)
await trt.execute(
trait.COMMAND_SET_TEMPERATURE,
BASIC_DATA,
{"temperature": temp_in},
{},
)
assert len(calls) == 1
assert calls[0].data == {
ATTR_ENTITY_ID: "water_heater.bla",
ATTR_TEMPERATURE: temp_out,
}
async def test_humidity_setting_humidifier_setpoint(hass: HomeAssistant) -> None:
"""Test HumiditySetting trait support for humidifier domain - setpoint."""
assert helpers.get_google_type(humidifier.DOMAIN, None) is not None
assert trait.HumiditySettingTrait.supported(humidifier.DOMAIN, 0, None, None)
trt = trait.HumiditySettingTrait(
hass,
State(
"humidifier.bla",
STATE_ON,
{
humidifier.ATTR_MIN_HUMIDITY: 20,
humidifier.ATTR_MAX_HUMIDITY: 90,
humidifier.ATTR_HUMIDITY: 38,
humidifier.ATTR_CURRENT_HUMIDITY: 30,
},
),
BASIC_CONFIG,
)
assert trt.sync_attributes() == {
"humiditySetpointRange": {"minPercent": 20, "maxPercent": 90}
}
assert trt.query_attributes() == {
"humiditySetpointPercent": 38,
"humidityAmbientPercent": 30,
}
assert trt.can_execute(trait.COMMAND_SET_HUMIDITY, {})
calls = async_mock_service(hass, humidifier.DOMAIN, humidifier.SERVICE_SET_HUMIDITY)
await trt.execute(trait.COMMAND_SET_HUMIDITY, BASIC_DATA, {"humidity": 32}, {})
assert len(calls) == 1
assert calls[0].data == {
ATTR_ENTITY_ID: "humidifier.bla",
humidifier.ATTR_HUMIDITY: 32,
}
async def test_lock_unlock_lock(hass: HomeAssistant) -> None:
"""Test LockUnlock trait locking support for lock domain."""
assert helpers.get_google_type(lock.DOMAIN, None) is not None
assert trait.LockUnlockTrait.supported(
lock.DOMAIN, LockEntityFeature.OPEN, None, None
)
assert trait.LockUnlockTrait.might_2fa(lock.DOMAIN, LockEntityFeature.OPEN, None)
trt = trait.LockUnlockTrait(
hass, State("lock.front_door", lock.LockState.LOCKED), PIN_CONFIG
)
assert trt.sync_attributes() == {}
assert trt.query_attributes() == {"isLocked": True}
assert trt.can_execute(trait.COMMAND_LOCK_UNLOCK, {"lock": True})
calls = async_mock_service(hass, lock.DOMAIN, lock.SERVICE_LOCK)
await trt.execute(trait.COMMAND_LOCK_UNLOCK, PIN_DATA, {"lock": True}, {})
assert len(calls) == 1
assert calls[0].data == {ATTR_ENTITY_ID: "lock.front_door"}
async def test_lock_unlock_unlocking(hass: HomeAssistant) -> None:
"""Test LockUnlock trait locking support for lock domain."""
assert helpers.get_google_type(lock.DOMAIN, None) is not None
assert trait.LockUnlockTrait.supported(
lock.DOMAIN, LockEntityFeature.OPEN, None, None
)
assert trait.LockUnlockTrait.might_2fa(lock.DOMAIN, LockEntityFeature.OPEN, None)
trt = trait.LockUnlockTrait(
hass, State("lock.front_door", lock.LockState.UNLOCKING), PIN_CONFIG
)
assert trt.sync_attributes() == {}
assert trt.query_attributes() == {"isLocked": True}
async def test_lock_unlock_lock_jammed(hass: HomeAssistant) -> None:
"""Test LockUnlock trait locking support for lock domain that jams."""
assert helpers.get_google_type(lock.DOMAIN, None) is not None
assert trait.LockUnlockTrait.supported(
lock.DOMAIN, LockEntityFeature.OPEN, None, None
)
assert trait.LockUnlockTrait.might_2fa(lock.DOMAIN, LockEntityFeature.OPEN, None)
trt = trait.LockUnlockTrait(
hass, State("lock.front_door", lock.LockState.JAMMED), PIN_CONFIG
)
assert trt.sync_attributes() == {}
assert trt.query_attributes() == {"isJammed": True}
assert trt.can_execute(trait.COMMAND_LOCK_UNLOCK, {"lock": True})
calls = async_mock_service(hass, lock.DOMAIN, lock.SERVICE_LOCK)
await trt.execute(trait.COMMAND_LOCK_UNLOCK, PIN_DATA, {"lock": True}, {})
assert len(calls) == 1
assert calls[0].data == {ATTR_ENTITY_ID: "lock.front_door"}
async def test_lock_unlock_unlock(hass: HomeAssistant) -> None:
"""Test LockUnlock trait unlocking support for lock domain."""
assert helpers.get_google_type(lock.DOMAIN, None) is not None
assert trait.LockUnlockTrait.supported(
lock.DOMAIN, LockEntityFeature.OPEN, None, None
)
trt = trait.LockUnlockTrait(
hass, State("lock.front_door", lock.LockState.LOCKED), PIN_CONFIG
)
assert trt.sync_attributes() == {}
assert trt.query_attributes() == {"isLocked": True}
assert trt.can_execute(trait.COMMAND_LOCK_UNLOCK, {"lock": False})
calls = async_mock_service(hass, lock.DOMAIN, lock.SERVICE_UNLOCK)
# No challenge data
with pytest.raises(error.ChallengeNeeded) as err:
await trt.execute(trait.COMMAND_LOCK_UNLOCK, PIN_DATA, {"lock": False}, {})
assert len(calls) == 0
assert err.value.code == const.ERR_CHALLENGE_NEEDED
assert err.value.challenge_type == const.CHALLENGE_PIN_NEEDED
# invalid pin
with pytest.raises(error.ChallengeNeeded) as err:
await trt.execute(
trait.COMMAND_LOCK_UNLOCK, PIN_DATA, {"lock": False}, {"pin": 9999}
)
assert len(calls) == 0
assert err.value.code == const.ERR_CHALLENGE_NEEDED
assert err.value.challenge_type == const.CHALLENGE_FAILED_PIN_NEEDED
await trt.execute(
trait.COMMAND_LOCK_UNLOCK, PIN_DATA, {"lock": False}, {"pin": "1234"}
)
assert len(calls) == 1
assert calls[0].data == {ATTR_ENTITY_ID: "lock.front_door"}
# Test without pin
trt = trait.LockUnlockTrait(
hass, State("lock.front_door", lock.LockState.LOCKED), BASIC_CONFIG
)
with pytest.raises(error.SmartHomeError) as err:
await trt.execute(trait.COMMAND_LOCK_UNLOCK, BASIC_DATA, {"lock": False}, {})
assert len(calls) == 1
assert err.value.code == const.ERR_CHALLENGE_NOT_SETUP
# Test with 2FA override
with patch.object(
BASIC_CONFIG,
"should_2fa",
return_value=False,
):
await trt.execute(trait.COMMAND_LOCK_UNLOCK, BASIC_DATA, {"lock": False}, {})
assert len(calls) == 2
async def test_arm_disarm_arm_away(hass: HomeAssistant) -> None:
"""Test ArmDisarm trait Arming support for alarm_control_panel domain."""
assert helpers.get_google_type(alarm_control_panel.DOMAIN, None) is not None
assert trait.ArmDisArmTrait.supported(alarm_control_panel.DOMAIN, 0, None, None)
assert trait.ArmDisArmTrait.might_2fa(alarm_control_panel.DOMAIN, 0, None)
trt = trait.ArmDisArmTrait(
hass,
State(
"alarm_control_panel.alarm",
AlarmControlPanelState.ARMED_AWAY,
{
alarm_control_panel.ATTR_CODE_ARM_REQUIRED: True,
ATTR_SUPPORTED_FEATURES: AlarmControlPanelEntityFeature.ARM_HOME
| AlarmControlPanelEntityFeature.ARM_AWAY,
},
),
PIN_CONFIG,
)
assert trt.sync_attributes() == {
"availableArmLevels": {
"levels": [
{
"level_name": "armed_home",
"level_values": [
{"level_synonym": ["armed home", "home"], "lang": "en"}
],
},
{
"level_name": "armed_away",
"level_values": [
{"level_synonym": ["armed away", "away"], "lang": "en"}
],
},
],
"ordered": True,
}
}
assert trt.query_attributes() == {
"isArmed": True,
"currentArmLevel": AlarmControlPanelState.ARMED_AWAY,
}
assert trt.can_execute(
trait.COMMAND_ARM_DISARM,
{"arm": True, "armLevel": AlarmControlPanelState.ARMED_AWAY},
)
calls = async_mock_service(
hass, alarm_control_panel.DOMAIN, alarm_control_panel.SERVICE_ALARM_ARM_AWAY
)
# Test with no secure_pin configured
trt = trait.ArmDisArmTrait(
hass,
State(
"alarm_control_panel.alarm",
AlarmControlPanelState.DISARMED,
{alarm_control_panel.ATTR_CODE_ARM_REQUIRED: True},
),
BASIC_CONFIG,
)
with pytest.raises(error.SmartHomeError) as err:
await trt.execute(
trait.COMMAND_ARM_DISARM,
BASIC_DATA,
{"arm": True, "armLevel": AlarmControlPanelState.ARMED_AWAY},
{},
)
assert len(calls) == 0
assert err.value.code == const.ERR_CHALLENGE_NOT_SETUP
trt = trait.ArmDisArmTrait(
hass,
State(
"alarm_control_panel.alarm",
AlarmControlPanelState.DISARMED,
{alarm_control_panel.ATTR_CODE_ARM_REQUIRED: True},
),
PIN_CONFIG,
)
# No challenge data
with pytest.raises(error.ChallengeNeeded) as err:
await trt.execute(
trait.COMMAND_ARM_DISARM,
PIN_DATA,
{"arm": True, "armLevel": AlarmControlPanelState.ARMED_AWAY},
{},
)
assert len(calls) == 0
assert err.value.code == const.ERR_CHALLENGE_NEEDED
assert err.value.challenge_type == const.CHALLENGE_PIN_NEEDED
# invalid pin
with pytest.raises(error.ChallengeNeeded) as err:
await trt.execute(
trait.COMMAND_ARM_DISARM,
PIN_DATA,
{"arm": True, "armLevel": AlarmControlPanelState.ARMED_AWAY},
{"pin": 9999},
)
assert len(calls) == 0
assert err.value.code == const.ERR_CHALLENGE_NEEDED
assert err.value.challenge_type == const.CHALLENGE_FAILED_PIN_NEEDED
# correct pin
await trt.execute(
trait.COMMAND_ARM_DISARM,
PIN_DATA,
{"arm": True, "armLevel": AlarmControlPanelState.ARMED_AWAY},
{"pin": "1234"},
)
assert len(calls) == 1
# Test already armed
trt = trait.ArmDisArmTrait(
hass,
State(
"alarm_control_panel.alarm",
AlarmControlPanelState.ARMED_AWAY,
{alarm_control_panel.ATTR_CODE_ARM_REQUIRED: True},
),
PIN_CONFIG,
)
with pytest.raises(error.SmartHomeError) as err:
await trt.execute(
trait.COMMAND_ARM_DISARM,
PIN_DATA,
{"arm": True, "armLevel": AlarmControlPanelState.ARMED_AWAY},
{},
)
assert len(calls) == 1
assert err.value.code == const.ERR_ALREADY_ARMED
# Test with code_arm_required False
trt = trait.ArmDisArmTrait(
hass,
State(
"alarm_control_panel.alarm",
AlarmControlPanelState.DISARMED,
{alarm_control_panel.ATTR_CODE_ARM_REQUIRED: False},
),
PIN_CONFIG,
)
await trt.execute(
trait.COMMAND_ARM_DISARM,
PIN_DATA,
{"arm": True, "armLevel": AlarmControlPanelState.ARMED_AWAY},
{},
)
assert len(calls) == 2
with pytest.raises(error.SmartHomeError) as err:
await trt.execute(
trait.COMMAND_ARM_DISARM,
PIN_DATA,
{"arm": True},
{},
)
async def test_arm_disarm_disarm(hass: HomeAssistant) -> None:
"""Test ArmDisarm trait Disarming support for alarm_control_panel domain."""
assert helpers.get_google_type(alarm_control_panel.DOMAIN, None) is not None
assert trait.ArmDisArmTrait.supported(alarm_control_panel.DOMAIN, 0, None, None)
assert trait.ArmDisArmTrait.might_2fa(alarm_control_panel.DOMAIN, 0, None)
trt = trait.ArmDisArmTrait(
hass,
State(
"alarm_control_panel.alarm",
AlarmControlPanelState.DISARMED,
{
alarm_control_panel.ATTR_CODE_ARM_REQUIRED: True,
ATTR_SUPPORTED_FEATURES: AlarmControlPanelEntityFeature.TRIGGER
| AlarmControlPanelEntityFeature.ARM_HOME
| AlarmControlPanelEntityFeature.ARM_AWAY,
},
),
PIN_CONFIG,
)
assert trt.sync_attributes() == {
"availableArmLevels": {
"levels": [
{
"level_name": "armed_home",
"level_values": [
{
"level_synonym": ["armed home", "home"],
"lang": "en",
}
],
},
{
"level_name": "armed_away",
"level_values": [
{
"level_synonym": ["armed away", "away"],
"lang": "en",
}
],
},
{
"level_name": "triggered",
"level_values": [{"level_synonym": ["triggered"], "lang": "en"}],
},
],
"ordered": True,
}
}
assert trt.query_attributes() == {
"currentArmLevel": "armed_home",
"isArmed": False,
}
assert trt.can_execute(trait.COMMAND_ARM_DISARM, {"arm": False})
calls = async_mock_service(
hass, alarm_control_panel.DOMAIN, alarm_control_panel.SERVICE_ALARM_DISARM
)
# Test without secure_pin configured
trt = trait.ArmDisArmTrait(
hass,
State(
"alarm_control_panel.alarm",
AlarmControlPanelState.ARMED_AWAY,
{alarm_control_panel.ATTR_CODE_ARM_REQUIRED: True},
),
BASIC_CONFIG,
)
with pytest.raises(error.SmartHomeError) as err:
await trt.execute(trait.COMMAND_ARM_DISARM, BASIC_DATA, {"arm": False}, {})
assert len(calls) == 0
assert err.value.code == const.ERR_CHALLENGE_NOT_SETUP
trt = trait.ArmDisArmTrait(
hass,
State(
"alarm_control_panel.alarm",
AlarmControlPanelState.ARMED_AWAY,
{alarm_control_panel.ATTR_CODE_ARM_REQUIRED: True},
),
PIN_CONFIG,
)
# No challenge data
with pytest.raises(error.ChallengeNeeded) as err:
await trt.execute(trait.COMMAND_ARM_DISARM, PIN_DATA, {"arm": False}, {})
assert len(calls) == 0
assert err.value.code == const.ERR_CHALLENGE_NEEDED
assert err.value.challenge_type == const.CHALLENGE_PIN_NEEDED
# invalid pin
with pytest.raises(error.ChallengeNeeded) as err:
await trt.execute(
trait.COMMAND_ARM_DISARM, PIN_DATA, {"arm": False}, {"pin": 9999}
)
assert len(calls) == 0
assert err.value.code == const.ERR_CHALLENGE_NEEDED
assert err.value.challenge_type == const.CHALLENGE_FAILED_PIN_NEEDED
# correct pin
await trt.execute(
trait.COMMAND_ARM_DISARM, PIN_DATA, {"arm": False}, {"pin": "1234"}
)
assert len(calls) == 1
# Test already disarmed
trt = trait.ArmDisArmTrait(
hass,
State(
"alarm_control_panel.alarm",
AlarmControlPanelState.DISARMED,
{alarm_control_panel.ATTR_CODE_ARM_REQUIRED: True},
),
PIN_CONFIG,
)
with pytest.raises(error.SmartHomeError) as err:
await trt.execute(trait.COMMAND_ARM_DISARM, PIN_DATA, {"arm": False}, {})
assert len(calls) == 1
assert err.value.code == const.ERR_ALREADY_DISARMED
trt = trait.ArmDisArmTrait(
hass,
State(
"alarm_control_panel.alarm",
AlarmControlPanelState.ARMED_AWAY,
{alarm_control_panel.ATTR_CODE_ARM_REQUIRED: False},
),
PIN_CONFIG,
)
# Cancel arming after already armed will require pin
with pytest.raises(error.SmartHomeError) as err:
await trt.execute(
trait.COMMAND_ARM_DISARM, PIN_DATA, {"arm": True, "cancel": True}, {}
)
assert len(calls) == 1
assert err.value.code == const.ERR_CHALLENGE_NEEDED
assert err.value.challenge_type == const.CHALLENGE_PIN_NEEDED
# Cancel arming while pending to arm doesn't require pin
trt = trait.ArmDisArmTrait(
hass,
State(
"alarm_control_panel.alarm",
AlarmControlPanelState.PENDING,
{alarm_control_panel.ATTR_CODE_ARM_REQUIRED: False},
),
PIN_CONFIG,
)
await trt.execute(
trait.COMMAND_ARM_DISARM, PIN_DATA, {"arm": True, "cancel": True}, {}
)
assert len(calls) == 2
async def test_fan_speed(hass: HomeAssistant) -> None:
"""Test FanSpeed trait speed control support for fan domain."""
assert helpers.get_google_type(fan.DOMAIN, None) is not None
assert trait.FanSpeedTrait.supported(
fan.DOMAIN, FanEntityFeature.SET_SPEED, None, None
)
trt = trait.FanSpeedTrait(
hass,
State(
"fan.living_room_fan",
STATE_ON,
attributes={
"percentage": 33,
"percentage_step": 1.0,
},
),
BASIC_CONFIG,
)
assert trt.sync_attributes() == {
"reversible": False,
"supportsFanSpeedPercent": True,
"availableFanSpeeds": ANY,
}
assert trt.query_attributes() == {
"currentFanSpeedPercent": 33,
"currentFanSpeedSetting": ANY,
}
assert trt.can_execute(trait.COMMAND_SET_FAN_SPEED, params={"fanSpeedPercent": 10})
calls = async_mock_service(hass, fan.DOMAIN, fan.SERVICE_SET_PERCENTAGE)
await trt.execute(
trait.COMMAND_SET_FAN_SPEED, BASIC_DATA, {"fanSpeedPercent": 10}, {}
)
assert len(calls) == 1
assert calls[0].data == {"entity_id": "fan.living_room_fan", "percentage": 10}
async def test_fan_speed_without_percentage_step(hass: HomeAssistant) -> None:
"""Test FanSpeed trait speed control percentage step for fan domain."""
assert helpers.get_google_type(fan.DOMAIN, None) is not None
assert trait.FanSpeedTrait.supported(
fan.DOMAIN, FanEntityFeature.SET_SPEED, None, None
)
trt = trait.FanSpeedTrait(
hass,
State(
"fan.living_room_fan",
STATE_ON,
),
BASIC_CONFIG,
)
assert trt.sync_attributes() == {
"reversible": False,
"supportsFanSpeedPercent": True,
"availableFanSpeeds": ANY,
}
# If a fan state has (temporary) no percentage_step attribute return 1 available
assert trt.query_attributes() == {
"currentFanSpeedPercent": 0,
"currentFanSpeedSetting": "1/5",
}
@pytest.mark.parametrize(
("percentage", "percentage_step", "speed", "speeds", "percentage_result"),
[
(
33,
1.0,
"2/5",
[
["Low", "Min", "Slow", "1"],
["Medium Low", "2"],
["Medium", "3"],
["Medium High", "4"],
["High", "Max", "Fast", "5"],
],
40,
),
(
40,
1.0,
"2/5",
[
["Low", "Min", "Slow", "1"],
["Medium Low", "2"],
["Medium", "3"],
["Medium High", "4"],
["High", "Max", "Fast", "5"],
],
40,
),
(
33,
100 / 3,
"1/3",
[
["Low", "Min", "Slow", "1"],
["Medium", "2"],
["High", "Max", "Fast", "3"],
],
33,
),
(
20,
100 / 4,
"1/4",
[
["Low", "Min", "Slow", "1"],
["Medium Low", "2"],
["Medium High", "3"],
["High", "Max", "Fast", "4"],
],
25,
),
],
)
async def test_fan_speed_ordered(
hass: HomeAssistant,
percentage: int,
percentage_step: float,
speed: str,
speeds: list[list[str]],
percentage_result: int,
) -> None:
"""Test FanSpeed trait speed control support for fan domain."""
assert helpers.get_google_type(fan.DOMAIN, None) is not None
assert trait.FanSpeedTrait.supported(
fan.DOMAIN, FanEntityFeature.SET_SPEED, None, None
)
trt = trait.FanSpeedTrait(
hass,
State(
"fan.living_room_fan",
STATE_ON,
attributes={
"percentage": percentage,
"percentage_step": percentage_step,
},
),
BASIC_CONFIG,
)
assert trt.sync_attributes() == {
"reversible": False,
"supportsFanSpeedPercent": True,
"availableFanSpeeds": {
"ordered": True,
"speeds": [
{
"speed_name": f"{idx+1}/{len(speeds)}",
"speed_values": [{"lang": "en", "speed_synonym": x}],
}
for idx, x in enumerate(speeds)
],
},
}
assert trt.query_attributes() == {
"currentFanSpeedPercent": percentage,
"currentFanSpeedSetting": speed,
}
assert trt.can_execute(trait.COMMAND_SET_FAN_SPEED, params={"fanSpeed": speed})
calls = async_mock_service(hass, fan.DOMAIN, fan.SERVICE_SET_PERCENTAGE)
await trt.execute(trait.COMMAND_SET_FAN_SPEED, BASIC_DATA, {"fanSpeed": speed}, {})
assert len(calls) == 1
assert calls[0].data == {
"entity_id": "fan.living_room_fan",
"percentage": percentage_result,
}
@pytest.mark.parametrize(
("direction_state", "direction_call"),
[
(fan.DIRECTION_FORWARD, fan.DIRECTION_REVERSE),
(fan.DIRECTION_REVERSE, fan.DIRECTION_FORWARD),
(None, fan.DIRECTION_FORWARD),
],
)
async def test_fan_reverse(
hass: HomeAssistant, direction_state, direction_call
) -> None:
"""Test FanSpeed trait speed control support for fan domain."""
calls = async_mock_service(hass, fan.DOMAIN, fan.SERVICE_SET_DIRECTION)
trt = trait.FanSpeedTrait(
hass,
State(
"fan.living_room_fan",
STATE_ON,
attributes={
"percentage": 33,
"percentage_step": 1.0,
"direction": direction_state,
"supported_features": FanEntityFeature.DIRECTION,
},
),
BASIC_CONFIG,
)
assert trt.sync_attributes() == {
"reversible": True,
"supportsFanSpeedPercent": True,
"availableFanSpeeds": ANY,
}
assert trt.query_attributes() == {
"currentFanSpeedPercent": 33,
"currentFanSpeedSetting": ANY,
}
assert trt.can_execute(trait.COMMAND_REVERSE, params={})
await trt.execute(trait.COMMAND_REVERSE, BASIC_DATA, {}, {})
assert len(calls) == 1
assert calls[0].data == {
"entity_id": "fan.living_room_fan",
"direction": direction_call,
}
async def test_climate_fan_speed(hass: HomeAssistant) -> None:
"""Test FanSpeed trait speed control support for climate domain."""
assert helpers.get_google_type(climate.DOMAIN, None) is not None
assert trait.FanSpeedTrait.supported(
climate.DOMAIN, ClimateEntityFeature.FAN_MODE, None, None
)
trt = trait.FanSpeedTrait(
hass,
State(
"climate.living_room_ac",
"on",
attributes={
"fan_modes": ["auto", "low", "medium", "high"],
"fan_mode": "low",
},
),
BASIC_CONFIG,
)
assert trt.sync_attributes() == {
"availableFanSpeeds": {
"ordered": True,
"speeds": [
{
"speed_name": "auto",
"speed_values": [{"speed_synonym": ["auto"], "lang": "en"}],
},
{
"speed_name": "low",
"speed_values": [{"speed_synonym": ["low"], "lang": "en"}],
},
{
"speed_name": "medium",
"speed_values": [{"speed_synonym": ["medium"], "lang": "en"}],
},
{
"speed_name": "high",
"speed_values": [{"speed_synonym": ["high"], "lang": "en"}],
},
],
},
"reversible": False,
}
assert trt.query_attributes() == {
"currentFanSpeedSetting": "low",
}
assert trt.can_execute(trait.COMMAND_SET_FAN_SPEED, params={"fanSpeed": "medium"})
calls = async_mock_service(hass, climate.DOMAIN, climate.SERVICE_SET_FAN_MODE)
await trt.execute(
trait.COMMAND_SET_FAN_SPEED, BASIC_DATA, {"fanSpeed": "medium"}, {}
)
assert len(calls) == 1
assert calls[0].data == {
"entity_id": "climate.living_room_ac",
"fan_mode": "medium",
}
async def test_inputselector(hass: HomeAssistant) -> None:
"""Test input selector trait."""
assert helpers.get_google_type(media_player.DOMAIN, None) is not None
assert trait.InputSelectorTrait.supported(
media_player.DOMAIN,
MediaPlayerEntityFeature.SELECT_SOURCE,
None,
None,
)
trt = trait.InputSelectorTrait(
hass,
State(
"media_player.living_room",
media_player.STATE_PLAYING,
attributes={
media_player.ATTR_INPUT_SOURCE_LIST: [
"media",
"game",
"chromecast",
"plex",
],
media_player.ATTR_INPUT_SOURCE: "game",
},
),
BASIC_CONFIG,
)
attribs = trt.sync_attributes()
assert attribs == {
"availableInputs": [
{"key": "media", "names": [{"name_synonym": ["media"], "lang": "en"}]},
{"key": "game", "names": [{"name_synonym": ["game"], "lang": "en"}]},
{
"key": "chromecast",
"names": [{"name_synonym": ["chromecast"], "lang": "en"}],
},
{"key": "plex", "names": [{"name_synonym": ["plex"], "lang": "en"}]},
],
"orderedInputs": True,
}
assert trt.query_attributes() == {
"currentInput": "game",
}
assert trt.can_execute(
trait.COMMAND_SET_INPUT,
params={"newInput": "media"},
)
calls = async_mock_service(
hass, media_player.DOMAIN, media_player.SERVICE_SELECT_SOURCE
)
await trt.execute(
trait.COMMAND_SET_INPUT,
BASIC_DATA,
{"newInput": "media"},
{},
)
assert len(calls) == 1
assert calls[0].data == {"entity_id": "media_player.living_room", "source": "media"}
@pytest.mark.parametrize(
("sources", "source", "source_next", "source_prev"),
[
(["a"], "a", "a", "a"),
(["a", "b"], "a", "b", "b"),
(["a", "b", "c"], "a", "b", "c"),
],
)
async def test_inputselector_nextprev(
hass: HomeAssistant, sources, source, source_next, source_prev
) -> None:
"""Test input selector trait."""
trt = trait.InputSelectorTrait(
hass,
State(
"media_player.living_room",
media_player.STATE_PLAYING,
attributes={
media_player.ATTR_INPUT_SOURCE_LIST: sources,
media_player.ATTR_INPUT_SOURCE: source,
},
),
BASIC_CONFIG,
)
assert trt.can_execute("action.devices.commands.NextInput", params={})
assert trt.can_execute("action.devices.commands.PreviousInput", params={})
calls = async_mock_service(
hass, media_player.DOMAIN, media_player.SERVICE_SELECT_SOURCE
)
await trt.execute(
"action.devices.commands.NextInput",
BASIC_DATA,
{},
{},
)
await trt.execute(
"action.devices.commands.PreviousInput",
BASIC_DATA,
{},
{},
)
assert len(calls) == 2
assert calls[0].data == {
"entity_id": "media_player.living_room",
"source": source_next,
}
assert calls[1].data == {
"entity_id": "media_player.living_room",
"source": source_prev,
}
@pytest.mark.parametrize(
("sources", "source"), [(None, "a"), (["a", "b"], None), (["a", "b"], "c")]
)
async def test_inputselector_nextprev_invalid(
hass: HomeAssistant, sources, source
) -> None:
"""Test input selector trait."""
trt = trait.InputSelectorTrait(
hass,
State(
"media_player.living_room",
media_player.STATE_PLAYING,
attributes={
media_player.ATTR_INPUT_SOURCE_LIST: sources,
media_player.ATTR_INPUT_SOURCE: source,
},
),
BASIC_CONFIG,
)
with pytest.raises(SmartHomeError):
await trt.execute(
"action.devices.commands.NextInput",
BASIC_DATA,
{},
{},
)
with pytest.raises(SmartHomeError):
await trt.execute(
"action.devices.commands.PreviousInput",
BASIC_DATA,
{},
{},
)
with pytest.raises(SmartHomeError):
await trt.execute(
"action.devices.commands.InvalidCommand",
BASIC_DATA,
{},
{},
)
async def test_modes_input_select(hass: HomeAssistant) -> None:
"""Test Input Select Mode trait."""
assert helpers.get_google_type(input_select.DOMAIN, None) is not None
assert trait.ModesTrait.supported(input_select.DOMAIN, None, None, None)
trt = trait.ModesTrait(
hass,
State("input_select.bla", "unavailable"),
BASIC_CONFIG,
)
assert trt.sync_attributes() == {"availableModes": []}
trt = trait.ModesTrait(
hass,
State(
"input_select.bla",
"abc",
attributes={input_select.ATTR_OPTIONS: ["abc", "123", "xyz"]},
),
BASIC_CONFIG,
)
attribs = trt.sync_attributes()
assert attribs == {
"availableModes": [
{
"name": "option",
"name_values": [
{
"name_synonym": ["option", "setting", "mode", "value"],
"lang": "en",
}
],
"settings": [
{
"setting_name": "abc",
"setting_values": [{"setting_synonym": ["abc"], "lang": "en"}],
},
{
"setting_name": "123",
"setting_values": [{"setting_synonym": ["123"], "lang": "en"}],
},
{
"setting_name": "xyz",
"setting_values": [{"setting_synonym": ["xyz"], "lang": "en"}],
},
],
"ordered": False,
}
]
}
assert trt.query_attributes() == {
"currentModeSettings": {"option": "abc"},
"on": True,
}
assert trt.can_execute(
trait.COMMAND_SET_MODES,
params={"updateModeSettings": {"option": "xyz"}},
)
calls = async_mock_service(
hass, input_select.DOMAIN, input_select.SERVICE_SELECT_OPTION
)
await trt.execute(
trait.COMMAND_SET_MODES,
BASIC_DATA,
{"updateModeSettings": {"option": "xyz"}},
{},
)
assert len(calls) == 1
assert calls[0].data == {"entity_id": "input_select.bla", "option": "xyz"}
async def test_modes_select(hass: HomeAssistant) -> None:
"""Test Select Mode trait."""
assert helpers.get_google_type(select.DOMAIN, None) is not None
assert trait.ModesTrait.supported(select.DOMAIN, None, None, None)
trt = trait.ModesTrait(
hass,
State("select.bla", "unavailable"),
BASIC_CONFIG,
)
assert trt.sync_attributes() == {"availableModes": []}
trt = trait.ModesTrait(
hass,
State(
"select.bla",
"abc",
attributes={select.ATTR_OPTIONS: ["abc", "123", "xyz"]},
),
BASIC_CONFIG,
)
attribs = trt.sync_attributes()
assert attribs == {
"availableModes": [
{
"name": "option",
"name_values": [
{
"name_synonym": ["option", "setting", "mode", "value"],
"lang": "en",
}
],
"settings": [
{
"setting_name": "abc",
"setting_values": [{"setting_synonym": ["abc"], "lang": "en"}],
},
{
"setting_name": "123",
"setting_values": [{"setting_synonym": ["123"], "lang": "en"}],
},
{
"setting_name": "xyz",
"setting_values": [{"setting_synonym": ["xyz"], "lang": "en"}],
},
],
"ordered": False,
}
]
}
assert trt.query_attributes() == {
"currentModeSettings": {"option": "abc"},
"on": True,
}
assert trt.can_execute(
trait.COMMAND_SET_MODES,
params={"updateModeSettings": {"option": "xyz"}},
)
calls = async_mock_service(hass, select.DOMAIN, select.SERVICE_SELECT_OPTION)
await trt.execute(
trait.COMMAND_SET_MODES,
BASIC_DATA,
{"updateModeSettings": {"option": "xyz"}},
{},
)
assert len(calls) == 1
assert calls[0].data == {"entity_id": "select.bla", "option": "xyz"}
async def test_modes_humidifier(hass: HomeAssistant) -> None:
"""Test Humidifier Mode trait."""
assert helpers.get_google_type(humidifier.DOMAIN, None) is not None
assert trait.ModesTrait.supported(
humidifier.DOMAIN, HumidifierEntityFeature.MODES, None, None
)
trt = trait.ModesTrait(
hass,
State(
"humidifier.humidifier",
STATE_OFF,
attributes={
humidifier.ATTR_AVAILABLE_MODES: [
humidifier.MODE_NORMAL,
humidifier.MODE_AUTO,
humidifier.MODE_AWAY,
],
ATTR_SUPPORTED_FEATURES: HumidifierEntityFeature.MODES,
humidifier.ATTR_MIN_HUMIDITY: 30,
humidifier.ATTR_MAX_HUMIDITY: 99,
humidifier.ATTR_HUMIDITY: 50,
ATTR_MODE: humidifier.MODE_AUTO,
},
),
BASIC_CONFIG,
)
attribs = trt.sync_attributes()
assert attribs == {
"availableModes": [
{
"name": "mode",
"name_values": [{"name_synonym": ["mode"], "lang": "en"}],
"settings": [
{
"setting_name": "normal",
"setting_values": [
{"setting_synonym": ["normal"], "lang": "en"}
],
},
{
"setting_name": "auto",
"setting_values": [{"setting_synonym": ["auto"], "lang": "en"}],
},
{
"setting_name": "away",
"setting_values": [{"setting_synonym": ["away"], "lang": "en"}],
},
],
"ordered": False,
},
]
}
assert trt.query_attributes() == {
"currentModeSettings": {"mode": "auto"},
"on": False,
}
assert trt.can_execute(
trait.COMMAND_SET_MODES, params={"updateModeSettings": {"mode": "away"}}
)
calls = async_mock_service(hass, humidifier.DOMAIN, humidifier.SERVICE_SET_MODE)
await trt.execute(
trait.COMMAND_SET_MODES,
BASIC_DATA,
{"updateModeSettings": {"mode": "away"}},
{},
)
assert len(calls) == 1
assert calls[0].data == {
"entity_id": "humidifier.humidifier",
"mode": "away",
}
async def test_modes_water_heater(hass: HomeAssistant) -> None:
"""Test Humidifier Mode trait."""
assert helpers.get_google_type(water_heater.DOMAIN, None) is not None
assert trait.ModesTrait.supported(
water_heater.DOMAIN, WaterHeaterEntityFeature.OPERATION_MODE, None, None
)
trt = trait.ModesTrait(
hass,
State(
"water_heater.water_heater",
STATE_OFF,
attributes={
water_heater.ATTR_OPERATION_LIST: [
water_heater.STATE_ECO,
water_heater.STATE_HEAT_PUMP,
water_heater.STATE_GAS,
],
ATTR_SUPPORTED_FEATURES: WaterHeaterEntityFeature.OPERATION_MODE,
water_heater.ATTR_OPERATION_MODE: water_heater.STATE_HEAT_PUMP,
},
),
BASIC_CONFIG,
)
attribs = trt.sync_attributes()
assert attribs == {
"availableModes": [
{
"name": "operation mode",
"name_values": [{"name_synonym": ["operation mode"], "lang": "en"}],
"settings": [
{
"setting_name": "eco",
"setting_values": [{"setting_synonym": ["eco"], "lang": "en"}],
},
{
"setting_name": "heat_pump",
"setting_values": [
{"setting_synonym": ["heat_pump"], "lang": "en"}
],
},
{
"setting_name": "gas",
"setting_values": [{"setting_synonym": ["gas"], "lang": "en"}],
},
],
"ordered": False,
},
]
}
assert trt.query_attributes() == {
"currentModeSettings": {"operation mode": "heat_pump"},
"on": False,
}
assert trt.can_execute(
trait.COMMAND_SET_MODES,
params={"updateModeSettings": {"operation mode": "gas"}},
)
calls = async_mock_service(
hass, water_heater.DOMAIN, water_heater.SERVICE_SET_OPERATION_MODE
)
await trt.execute(
trait.COMMAND_SET_MODES,
BASIC_DATA,
{"updateModeSettings": {"operation mode": "gas"}},
{},
)
assert len(calls) == 1
assert calls[0].data == {
"entity_id": "water_heater.water_heater",
"operation_mode": "gas",
}
async def test_sound_modes(hass: HomeAssistant) -> None:
"""Test Mode trait."""
assert helpers.get_google_type(media_player.DOMAIN, None) is not None
assert trait.ModesTrait.supported(
media_player.DOMAIN,
MediaPlayerEntityFeature.SELECT_SOUND_MODE,
None,
None,
)
trt = trait.ModesTrait(
hass,
State(
"media_player.living_room",
media_player.STATE_PLAYING,
attributes={
media_player.ATTR_SOUND_MODE_LIST: ["stereo", "prologic"],
media_player.ATTR_SOUND_MODE: "stereo",
},
),
BASIC_CONFIG,
)
attribs = trt.sync_attributes()
assert attribs == {
"availableModes": [
{
"name": "sound mode",
"name_values": [
{"name_synonym": ["sound mode", "effects"], "lang": "en"}
],
"settings": [
{
"setting_name": "stereo",
"setting_values": [
{"setting_synonym": ["stereo"], "lang": "en"}
],
},
{
"setting_name": "prologic",
"setting_values": [
{"setting_synonym": ["prologic"], "lang": "en"}
],
},
],
"ordered": False,
}
]
}
assert trt.query_attributes() == {
"currentModeSettings": {"sound mode": "stereo"},
"on": True,
}
assert trt.can_execute(
trait.COMMAND_SET_MODES,
params={"updateModeSettings": {"sound mode": "stereo"}},
)
calls = async_mock_service(
hass, media_player.DOMAIN, media_player.SERVICE_SELECT_SOUND_MODE
)
await trt.execute(
trait.COMMAND_SET_MODES,
BASIC_DATA,
{"updateModeSettings": {"sound mode": "stereo"}},
{},
)
assert len(calls) == 1
assert calls[0].data == {
"entity_id": "media_player.living_room",
"sound_mode": "stereo",
}
async def test_preset_modes(hass: HomeAssistant) -> None:
"""Test Mode trait for fan preset modes."""
assert helpers.get_google_type(fan.DOMAIN, None) is not None
assert trait.ModesTrait.supported(
fan.DOMAIN, FanEntityFeature.PRESET_MODE, None, None
)
trt = trait.ModesTrait(
hass,
State(
"fan.living_room",
STATE_ON,
attributes={
fan.ATTR_PRESET_MODES: ["auto", "whoosh"],
fan.ATTR_PRESET_MODE: "auto",
ATTR_SUPPORTED_FEATURES: FanEntityFeature.PRESET_MODE,
},
),
BASIC_CONFIG,
)
attribs = trt.sync_attributes()
assert attribs == {
"availableModes": [
{
"name": "preset mode",
"name_values": [
{"name_synonym": ["preset mode", "mode", "preset"], "lang": "en"}
],
"settings": [
{
"setting_name": "auto",
"setting_values": [{"setting_synonym": ["auto"], "lang": "en"}],
},
{
"setting_name": "whoosh",
"setting_values": [
{"setting_synonym": ["whoosh"], "lang": "en"}
],
},
],
"ordered": False,
}
]
}
assert trt.query_attributes() == {
"currentModeSettings": {"preset mode": "auto"},
"on": True,
}
assert trt.can_execute(
trait.COMMAND_SET_MODES,
params={"updateModeSettings": {"preset mode": "auto"}},
)
calls = async_mock_service(hass, fan.DOMAIN, fan.SERVICE_SET_PRESET_MODE)
await trt.execute(
trait.COMMAND_SET_MODES,
BASIC_DATA,
{"updateModeSettings": {"preset mode": "auto"}},
{},
)
assert len(calls) == 1
assert calls[0].data == {
"entity_id": "fan.living_room",
"preset_mode": "auto",
}
async def test_traits_unknown_domains(
hass: HomeAssistant, caplog: pytest.LogCaptureFixture
) -> None:
"""Test Mode trait for unsupported domain."""
trt = trait.ModesTrait(
hass,
State(
"switch.living_room",
STATE_ON,
),
BASIC_CONFIG,
)
assert trt.supported("not_supported_domain", False, None, None) is False
await trt.execute(
trait.COMMAND_SET_MODES,
BASIC_DATA,
{"updateModeSettings": {}},
{},
)
assert "Received an Options command for unrecognised domain" in caplog.text
caplog.clear()
@pytest.mark.parametrize(
(
"domain",
"set_position_service",
"close_service",
"open_service",
"set_position_feature",
"attr_position",
"attr_current_position",
),
[
(
cover.DOMAIN,
cover.SERVICE_SET_COVER_POSITION,
cover.SERVICE_CLOSE_COVER,
cover.SERVICE_OPEN_COVER,
CoverEntityFeature.SET_POSITION,
cover.ATTR_POSITION,
cover.ATTR_CURRENT_POSITION,
),
(
valve.DOMAIN,
valve.SERVICE_SET_VALVE_POSITION,
valve.SERVICE_CLOSE_VALVE,
valve.SERVICE_OPEN_VALVE,
ValveEntityFeature.SET_POSITION,
valve.ATTR_POSITION,
valve.ATTR_CURRENT_POSITION,
),
],
)
async def test_openclose_cover_valve(
hass: HomeAssistant,
domain: str,
set_position_service: str,
close_service: str,
open_service: str,
set_position_feature: int,
attr_position: str,
attr_current_position: str,
) -> None:
"""Test OpenClose trait support."""
assert helpers.get_google_type(domain, None) is not None
assert trait.OpenCloseTrait.supported(domain, set_position_service, None, None)
trt = trait.OpenCloseTrait(
hass,
State(
f"{domain}.bla",
"open",
{
attr_current_position: 75,
ATTR_SUPPORTED_FEATURES: set_position_feature,
},
),
BASIC_CONFIG,
)
assert trt.sync_attributes() == {}
assert trt.query_attributes() == {"openPercent": 75}
calls_set = async_mock_service(hass, domain, set_position_service)
calls_open = async_mock_service(hass, domain, open_service)
calls_close = async_mock_service(hass, domain, close_service)
await trt.execute(trait.COMMAND_OPEN_CLOSE, BASIC_DATA, {"openPercent": 50}, {})
await trt.execute(
trait.COMMAND_OPEN_CLOSE_RELATIVE, BASIC_DATA, {"openRelativePercent": 50}, {}
)
assert len(calls_set) == 1
assert calls_set[0].data == {
ATTR_ENTITY_ID: f"{domain}.bla",
attr_position: 50,
}
calls_set.pop(0)
assert len(calls_open) == 1
assert calls_open[0].data == {ATTR_ENTITY_ID: f"{domain}.bla"}
calls_open.pop(0)
assert len(calls_close) == 0
await trt.execute(trait.COMMAND_OPEN_CLOSE, BASIC_DATA, {"openPercent": 0}, {})
await trt.execute(
trait.COMMAND_OPEN_CLOSE_RELATIVE, BASIC_DATA, {"openRelativePercent": 0}, {}
)
assert len(calls_set) == 1
assert len(calls_close) == 1
assert calls_close[0].data == {ATTR_ENTITY_ID: f"{domain}.bla"}
assert len(calls_open) == 0
@pytest.mark.parametrize(
("domain", "open_service", "set_position_feature", "open_feature"),
[
(
cover.DOMAIN,
cover.SERVICE_OPEN_COVER,
CoverEntityFeature.SET_POSITION,
CoverEntityFeature.OPEN,
),
(
valve.DOMAIN,
valve.SERVICE_OPEN_VALVE,
ValveEntityFeature.SET_POSITION,
ValveEntityFeature.OPEN,
),
],
)
async def test_openclose_cover_valve_unknown_state(
hass: HomeAssistant,
open_service: str,
domain: str,
set_position_feature: int,
open_feature: int,
) -> None:
"""Test OpenClose trait support with unknown state."""
assert helpers.get_google_type(domain, None) is not None
assert trait.OpenCloseTrait.supported(
cover.DOMAIN, set_position_feature, None, None
)
# No state
trt = trait.OpenCloseTrait(
hass,
State(
f"{domain}.bla",
STATE_UNKNOWN,
{ATTR_SUPPORTED_FEATURES: open_feature},
),
BASIC_CONFIG,
)
assert trt.sync_attributes() == {"discreteOnlyOpenClose": True}
with pytest.raises(helpers.SmartHomeError):
trt.query_attributes()
calls = async_mock_service(hass, domain, open_service)
await trt.execute(trait.COMMAND_OPEN_CLOSE, BASIC_DATA, {"openPercent": 100}, {})
assert len(calls) == 1
assert calls[0].data == {ATTR_ENTITY_ID: f"{domain}.bla"}
with pytest.raises(helpers.SmartHomeError):
trt.query_attributes()
@pytest.mark.parametrize(
("domain", "set_position_service", "set_position_feature", "state_open"),
[
(
cover.DOMAIN,
cover.SERVICE_SET_COVER_POSITION,
CoverEntityFeature.SET_POSITION,
cover.STATE_OPEN,
),
(
valve.DOMAIN,
valve.SERVICE_SET_VALVE_POSITION,
ValveEntityFeature.SET_POSITION,
valve.ValveState.OPEN,
),
],
)
async def test_openclose_cover_valve_assumed_state(
hass: HomeAssistant,
domain: str,
set_position_service: str,
set_position_feature: int,
state_open: str,
) -> None:
"""Test OpenClose trait support."""
assert helpers.get_google_type(domain, None) is not None
assert trait.OpenCloseTrait.supported(domain, set_position_feature, None, None)
trt = trait.OpenCloseTrait(
hass,
State(
f"{domain}.bla",
state_open,
{
ATTR_ASSUMED_STATE: True,
ATTR_SUPPORTED_FEATURES: set_position_feature,
},
),
BASIC_CONFIG,
)
assert trt.sync_attributes() == {"commandOnlyOpenClose": True}
assert trt.query_attributes() == {}
calls = async_mock_service(hass, domain, set_position_service)
await trt.execute(trait.COMMAND_OPEN_CLOSE, BASIC_DATA, {"openPercent": 40}, {})
assert len(calls) == 1
assert calls[0].data == {ATTR_ENTITY_ID: f"{domain}.bla", cover.ATTR_POSITION: 40}
@pytest.mark.parametrize(
("domain", "state_open"),
[
(
cover.DOMAIN,
cover.STATE_OPEN,
),
(
valve.DOMAIN,
valve.ValveState.OPEN,
),
],
)
async def test_openclose_cover_valve_query_only(
hass: HomeAssistant,
domain: str,
state_open: str,
) -> None:
"""Test OpenClose trait support."""
assert helpers.get_google_type(domain, None) is not None
assert trait.OpenCloseTrait.supported(domain, 0, None, None)
state = State(
f"{domain}.bla",
state_open,
)
trt = trait.OpenCloseTrait(
hass,
state,
BASIC_CONFIG,
)
assert trt.sync_attributes() == {
"discreteOnlyOpenClose": True,
"queryOnlyOpenClose": True,
}
assert trt.query_attributes() == {"openPercent": 100}
@pytest.mark.parametrize(
(
"domain",
"state_open",
"state_closed",
"supported_features",
"open_service",
"close_service",
),
[
(
cover.DOMAIN,
cover.STATE_OPEN,
cover.STATE_CLOSED,
CoverEntityFeature.OPEN | CoverEntityFeature.CLOSE,
cover.SERVICE_OPEN_COVER,
cover.SERVICE_CLOSE_COVER,
),
(
valve.DOMAIN,
valve.ValveState.OPEN,
valve.ValveState.CLOSED,
ValveEntityFeature.OPEN | ValveEntityFeature.CLOSE,
valve.SERVICE_OPEN_VALVE,
valve.SERVICE_CLOSE_VALVE,
),
],
)
async def test_openclose_cover_valve_no_position(
hass: HomeAssistant,
domain: str,
state_open: str,
state_closed: str,
supported_features: int,
open_service: str,
close_service: str,
) -> None:
"""Test OpenClose trait support."""
assert helpers.get_google_type(domain, None) is not None
assert trait.OpenCloseTrait.supported(
domain,
supported_features,
None,
None,
)
state = State(
f"{domain}.bla",
state_open,
{
ATTR_SUPPORTED_FEATURES: supported_features,
},
)
trt = trait.OpenCloseTrait(
hass,
state,
BASIC_CONFIG,
)
assert trt.sync_attributes() == {"discreteOnlyOpenClose": True}
assert trt.query_attributes() == {"openPercent": 100}
state.state = state_closed
assert trt.sync_attributes() == {"discreteOnlyOpenClose": True}
assert trt.query_attributes() == {"openPercent": 0}
calls = async_mock_service(hass, domain, close_service)
await trt.execute(trait.COMMAND_OPEN_CLOSE, BASIC_DATA, {"openPercent": 0}, {})
assert len(calls) == 1
assert calls[0].data == {ATTR_ENTITY_ID: f"{domain}.bla"}
calls = async_mock_service(hass, domain, open_service)
await trt.execute(trait.COMMAND_OPEN_CLOSE, BASIC_DATA, {"openPercent": 100}, {})
assert len(calls) == 1
assert calls[0].data == {ATTR_ENTITY_ID: f"{domain}.bla"}
with pytest.raises(
SmartHomeError, match=r"Current position not know for relative command"
):
await trt.execute(
trait.COMMAND_OPEN_CLOSE_RELATIVE,
BASIC_DATA,
{"openRelativePercent": 100},
{},
)
with pytest.raises(SmartHomeError, match=r"No support for partial open close"):
await trt.execute(trait.COMMAND_OPEN_CLOSE, BASIC_DATA, {"openPercent": 50}, {})
@pytest.mark.parametrize(
"device_class",
[
cover.CoverDeviceClass.DOOR,
cover.CoverDeviceClass.GARAGE,
cover.CoverDeviceClass.GATE,
],
)
async def test_openclose_cover_secure(hass: HomeAssistant, device_class) -> None:
"""Test OpenClose trait support for cover domain."""
assert helpers.get_google_type(cover.DOMAIN, device_class) is not None
assert trait.OpenCloseTrait.supported(
cover.DOMAIN, CoverEntityFeature.SET_POSITION, device_class, None
)
assert trait.OpenCloseTrait.might_2fa(
cover.DOMAIN, CoverEntityFeature.SET_POSITION, device_class
)
trt = trait.OpenCloseTrait(
hass,
State(
"cover.bla",
cover.STATE_OPEN,
{
ATTR_DEVICE_CLASS: device_class,
ATTR_SUPPORTED_FEATURES: CoverEntityFeature.SET_POSITION,
cover.ATTR_CURRENT_POSITION: 75,
},
),
PIN_CONFIG,
)
assert trt.sync_attributes() == {}
assert trt.query_attributes() == {"openPercent": 75}
calls = async_mock_service(hass, cover.DOMAIN, cover.SERVICE_SET_COVER_POSITION)
calls_close = async_mock_service(hass, cover.DOMAIN, cover.SERVICE_CLOSE_COVER)
# No challenge data
with pytest.raises(error.ChallengeNeeded) as err:
await trt.execute(trait.COMMAND_OPEN_CLOSE, PIN_DATA, {"openPercent": 50}, {})
assert len(calls) == 0
assert err.value.code == const.ERR_CHALLENGE_NEEDED
assert err.value.challenge_type == const.CHALLENGE_PIN_NEEDED
# invalid pin
with pytest.raises(error.ChallengeNeeded) as err:
await trt.execute(
trait.COMMAND_OPEN_CLOSE, PIN_DATA, {"openPercent": 50}, {"pin": "9999"}
)
assert len(calls) == 0
assert err.value.code == const.ERR_CHALLENGE_NEEDED
assert err.value.challenge_type == const.CHALLENGE_FAILED_PIN_NEEDED
await trt.execute(
trait.COMMAND_OPEN_CLOSE, PIN_DATA, {"openPercent": 50}, {"pin": "1234"}
)
assert len(calls) == 1
assert calls[0].data == {ATTR_ENTITY_ID: "cover.bla", cover.ATTR_POSITION: 50}
# no challenge on close
await trt.execute(trait.COMMAND_OPEN_CLOSE, PIN_DATA, {"openPercent": 0}, {})
assert len(calls_close) == 1
assert calls_close[0].data == {ATTR_ENTITY_ID: "cover.bla"}
@pytest.mark.parametrize(
"device_class",
[
binary_sensor.BinarySensorDeviceClass.DOOR,
binary_sensor.BinarySensorDeviceClass.GARAGE_DOOR,
binary_sensor.BinarySensorDeviceClass.LOCK,
binary_sensor.BinarySensorDeviceClass.OPENING,
binary_sensor.BinarySensorDeviceClass.WINDOW,
],
)
async def test_openclose_binary_sensor(hass: HomeAssistant, device_class) -> None:
"""Test OpenClose trait support for binary_sensor domain."""
assert helpers.get_google_type(binary_sensor.DOMAIN, device_class) is not None
assert trait.OpenCloseTrait.supported(binary_sensor.DOMAIN, 0, device_class, None)
trt = trait.OpenCloseTrait(
hass,
State("binary_sensor.test", STATE_ON, {ATTR_DEVICE_CLASS: device_class}),
BASIC_CONFIG,
)
assert trt.sync_attributes() == {
"queryOnlyOpenClose": True,
"discreteOnlyOpenClose": True,
}
assert trt.query_attributes() == {"openPercent": 100}
trt = trait.OpenCloseTrait(
hass,
State("binary_sensor.test", STATE_OFF, {ATTR_DEVICE_CLASS: device_class}),
BASIC_CONFIG,
)
assert trt.sync_attributes() == {
"queryOnlyOpenClose": True,
"discreteOnlyOpenClose": True,
}
assert trt.query_attributes() == {"openPercent": 0}
async def test_volume_media_player(hass: HomeAssistant) -> None:
"""Test volume trait support for media player domain."""
assert helpers.get_google_type(media_player.DOMAIN, None) is not None
assert trait.VolumeTrait.supported(
media_player.DOMAIN,
MediaPlayerEntityFeature.VOLUME_SET,
None,
None,
)
trt = trait.VolumeTrait(
hass,
State(
"media_player.bla",
media_player.STATE_PLAYING,
{
ATTR_SUPPORTED_FEATURES: MediaPlayerEntityFeature.VOLUME_SET,
media_player.ATTR_MEDIA_VOLUME_LEVEL: 0.3,
},
),
BASIC_CONFIG,
)
assert trt.sync_attributes() == {
"volumeMaxLevel": 100,
"levelStepSize": 10,
"volumeCanMuteAndUnmute": False,
"commandOnlyVolume": False,
}
assert trt.query_attributes() == {"currentVolume": 30}
calls = async_mock_service(
hass, media_player.DOMAIN, media_player.SERVICE_VOLUME_SET
)
await trt.execute(trait.COMMAND_SET_VOLUME, BASIC_DATA, {"volumeLevel": 60}, {})
assert len(calls) == 1
assert calls[0].data == {
ATTR_ENTITY_ID: "media_player.bla",
media_player.ATTR_MEDIA_VOLUME_LEVEL: 0.6,
}
calls = async_mock_service(
hass, media_player.DOMAIN, media_player.SERVICE_VOLUME_SET
)
await trt.execute(
trait.COMMAND_VOLUME_RELATIVE, BASIC_DATA, {"relativeSteps": 10}, {}
)
assert len(calls) == 1
assert calls[0].data == {
ATTR_ENTITY_ID: "media_player.bla",
media_player.ATTR_MEDIA_VOLUME_LEVEL: 0.4,
}
async def test_volume_media_player_relative(hass: HomeAssistant) -> None:
"""Test volume trait support for relative-volume-only media players."""
assert trait.VolumeTrait.supported(
media_player.DOMAIN,
MediaPlayerEntityFeature.VOLUME_STEP,
None,
None,
)
trt = trait.VolumeTrait(
hass,
State(
"media_player.bla",
media_player.STATE_PLAYING,
{
ATTR_ASSUMED_STATE: True,
ATTR_SUPPORTED_FEATURES: MediaPlayerEntityFeature.VOLUME_STEP,
},
),
BASIC_CONFIG,
)
assert trt.sync_attributes() == {
"volumeMaxLevel": 100,
"levelStepSize": 10,
"volumeCanMuteAndUnmute": False,
"commandOnlyVolume": True,
}
assert trt.query_attributes() == {}
calls = async_mock_service(
hass, media_player.DOMAIN, media_player.SERVICE_VOLUME_UP
)
await trt.execute(
trait.COMMAND_VOLUME_RELATIVE,
BASIC_DATA,
{"relativeSteps": 10},
{},
)
assert len(calls) == 10
for call in calls:
assert call.data == {
ATTR_ENTITY_ID: "media_player.bla",
}
calls = async_mock_service(
hass, media_player.DOMAIN, media_player.SERVICE_VOLUME_DOWN
)
await trt.execute(
trait.COMMAND_VOLUME_RELATIVE,
BASIC_DATA,
{"relativeSteps": -10},
{},
)
assert len(calls) == 10
for call in calls:
assert call.data == {
ATTR_ENTITY_ID: "media_player.bla",
}
with pytest.raises(SmartHomeError):
await trt.execute(trait.COMMAND_SET_VOLUME, BASIC_DATA, {"volumeLevel": 42}, {})
with pytest.raises(SmartHomeError):
await trt.execute(trait.COMMAND_MUTE, BASIC_DATA, {"mute": True}, {})
async def test_media_player_mute(hass: HomeAssistant) -> None:
"""Test volume trait support for muting."""
assert trait.VolumeTrait.supported(
media_player.DOMAIN,
MediaPlayerEntityFeature.VOLUME_STEP | MediaPlayerEntityFeature.VOLUME_MUTE,
None,
None,
)
trt = trait.VolumeTrait(
hass,
State(
"media_player.bla",
media_player.STATE_PLAYING,
{
ATTR_SUPPORTED_FEATURES: (
MediaPlayerEntityFeature.VOLUME_STEP
| MediaPlayerEntityFeature.VOLUME_MUTE
),
media_player.ATTR_MEDIA_VOLUME_MUTED: False,
},
),
BASIC_CONFIG,
)
assert trt.sync_attributes() == {
"volumeMaxLevel": 100,
"levelStepSize": 10,
"volumeCanMuteAndUnmute": True,
"commandOnlyVolume": False,
}
assert trt.query_attributes() == {"isMuted": False}
mute_calls = async_mock_service(
hass, media_player.DOMAIN, media_player.SERVICE_VOLUME_MUTE
)
await trt.execute(
trait.COMMAND_MUTE,
BASIC_DATA,
{"mute": True},
{},
)
assert len(mute_calls) == 1
assert mute_calls[0].data == {
ATTR_ENTITY_ID: "media_player.bla",
media_player.ATTR_MEDIA_VOLUME_MUTED: True,
}
unmute_calls = async_mock_service(
hass, media_player.DOMAIN, media_player.SERVICE_VOLUME_MUTE
)
await trt.execute(
trait.COMMAND_MUTE,
BASIC_DATA,
{"mute": False},
{},
)
assert len(unmute_calls) == 1
assert unmute_calls[0].data == {
ATTR_ENTITY_ID: "media_player.bla",
media_player.ATTR_MEDIA_VOLUME_MUTED: False,
}
async def test_temperature_control_sensor(hass: HomeAssistant) -> None:
"""Test TemperatureControl trait support for temperature sensor."""
assert (
helpers.get_google_type(sensor.DOMAIN, sensor.SensorDeviceClass.TEMPERATURE)
is not None
)
assert not trait.TemperatureControlTrait.supported(
sensor.DOMAIN, 0, sensor.SensorDeviceClass.HUMIDITY, None
)
assert trait.TemperatureControlTrait.supported(
sensor.DOMAIN, 0, sensor.SensorDeviceClass.TEMPERATURE, None
)
@pytest.mark.parametrize(
("unit_in", "unit_out", "state", "ambient"),
[
(UnitOfTemperature.FAHRENHEIT, "F", "70", 21.1),
(UnitOfTemperature.CELSIUS, "C", "21.1", 21.1),
(UnitOfTemperature.FAHRENHEIT, "F", "unavailable", None),
(UnitOfTemperature.FAHRENHEIT, "F", "unknown", None),
],
)
async def test_temperature_control_sensor_data(
hass: HomeAssistant, unit_in, unit_out, state, ambient
) -> None:
"""Test TemperatureControl trait support for temperature sensor."""
hass.config.units.temperature_unit = unit_in
trt = trait.TemperatureControlTrait(
hass,
State(
"sensor.test",
state,
{ATTR_DEVICE_CLASS: sensor.SensorDeviceClass.TEMPERATURE},
),
BASIC_CONFIG,
)
assert trt.sync_attributes() == {
"queryOnlyTemperatureControl": True,
"temperatureUnitForUX": unit_out,
"temperatureRange": {"maxThresholdCelsius": 100, "minThresholdCelsius": -100},
}
if ambient:
assert trt.query_attributes() == {
"temperatureAmbientCelsius": ambient,
"temperatureSetpointCelsius": ambient,
}
else:
assert trt.query_attributes() == {}
hass.config.units.temperature_unit = UnitOfTemperature.CELSIUS
async def test_humidity_setting_sensor(hass: HomeAssistant) -> None:
"""Test HumiditySetting trait support for humidity sensor."""
assert (
helpers.get_google_type(sensor.DOMAIN, sensor.SensorDeviceClass.HUMIDITY)
is not None
)
assert not trait.HumiditySettingTrait.supported(
sensor.DOMAIN, 0, sensor.SensorDeviceClass.TEMPERATURE, None
)
assert trait.HumiditySettingTrait.supported(
sensor.DOMAIN, 0, sensor.SensorDeviceClass.HUMIDITY, None
)
@pytest.mark.parametrize(
("state", "ambient"), [("70", 70), ("unavailable", None), ("unknown", None)]
)
async def test_humidity_setting_sensor_data(
hass: HomeAssistant, state, ambient
) -> None:
"""Test HumiditySetting trait support for humidity sensor."""
trt = trait.HumiditySettingTrait(
hass,
State(
"sensor.test", state, {ATTR_DEVICE_CLASS: sensor.SensorDeviceClass.HUMIDITY}
),
BASIC_CONFIG,
)
assert trt.sync_attributes() == {"queryOnlyHumiditySetting": True}
if ambient:
assert trt.query_attributes() == {"humidityAmbientPercent": ambient}
else:
assert trt.query_attributes() == {}
with pytest.raises(helpers.SmartHomeError) as err:
await trt.execute(trait.COMMAND_ON_OFF, BASIC_DATA, {"on": False}, {})
assert err.value.code == const.ERR_NOT_SUPPORTED
async def test_transport_control(
hass: HomeAssistant, freezer: FrozenDateTimeFactory
) -> None:
"""Test the TransportControlTrait."""
assert helpers.get_google_type(media_player.DOMAIN, None) is not None
for feature in trait.MEDIA_COMMAND_SUPPORT_MAPPING.values():
assert trait.TransportControlTrait.supported(
media_player.DOMAIN, feature, None, None
)
now = datetime(2020, 1, 1, tzinfo=dt_util.UTC)
trt = trait.TransportControlTrait(
hass,
State(
"media_player.bla",
media_player.STATE_PLAYING,
{
media_player.ATTR_MEDIA_POSITION: 100,
media_player.ATTR_MEDIA_DURATION: 200,
media_player.ATTR_MEDIA_POSITION_UPDATED_AT: now
- timedelta(seconds=10),
media_player.ATTR_MEDIA_VOLUME_LEVEL: 0.5,
ATTR_SUPPORTED_FEATURES: MediaPlayerEntityFeature.PLAY
| MediaPlayerEntityFeature.STOP,
},
),
BASIC_CONFIG,
)
assert trt.sync_attributes() == {
"transportControlSupportedCommands": ["RESUME", "STOP"]
}
assert trt.query_attributes() == {}
# COMMAND_MEDIA_SEEK_RELATIVE
calls = async_mock_service(
hass, media_player.DOMAIN, media_player.SERVICE_MEDIA_SEEK
)
# Patch to avoid time ticking over during the command failing the test
freezer.move_to(now)
await trt.execute(
trait.COMMAND_MEDIA_SEEK_RELATIVE,
BASIC_DATA,
{"relativePositionMs": 10000},
{},
)
assert len(calls) == 1
assert calls[0].data == {
ATTR_ENTITY_ID: "media_player.bla",
# 100s (current position) + 10s (from command) + 10s (from updated_at)
media_player.ATTR_MEDIA_SEEK_POSITION: 120,
}
# COMMAND_MEDIA_SEEK_TO_POSITION
calls = async_mock_service(
hass, media_player.DOMAIN, media_player.SERVICE_MEDIA_SEEK
)
await trt.execute(
trait.COMMAND_MEDIA_SEEK_TO_POSITION, BASIC_DATA, {"absPositionMs": 50000}, {}
)
assert len(calls) == 1
assert calls[0].data == {
ATTR_ENTITY_ID: "media_player.bla",
media_player.ATTR_MEDIA_SEEK_POSITION: 50,
}
# COMMAND_MEDIA_NEXT
calls = async_mock_service(
hass, media_player.DOMAIN, media_player.SERVICE_MEDIA_NEXT_TRACK
)
await trt.execute(trait.COMMAND_MEDIA_NEXT, BASIC_DATA, {}, {})
assert len(calls) == 1
assert calls[0].data == {ATTR_ENTITY_ID: "media_player.bla"}
# COMMAND_MEDIA_PAUSE
calls = async_mock_service(
hass, media_player.DOMAIN, media_player.SERVICE_MEDIA_PAUSE
)
await trt.execute(trait.COMMAND_MEDIA_PAUSE, BASIC_DATA, {}, {})
assert len(calls) == 1
assert calls[0].data == {ATTR_ENTITY_ID: "media_player.bla"}
# COMMAND_MEDIA_PREVIOUS
calls = async_mock_service(
hass, media_player.DOMAIN, media_player.SERVICE_MEDIA_PREVIOUS_TRACK
)
await trt.execute(trait.COMMAND_MEDIA_PREVIOUS, BASIC_DATA, {}, {})
assert len(calls) == 1
assert calls[0].data == {ATTR_ENTITY_ID: "media_player.bla"}
# COMMAND_MEDIA_RESUME
calls = async_mock_service(
hass, media_player.DOMAIN, media_player.SERVICE_MEDIA_PLAY
)
await trt.execute(trait.COMMAND_MEDIA_RESUME, BASIC_DATA, {}, {})
assert len(calls) == 1
assert calls[0].data == {ATTR_ENTITY_ID: "media_player.bla"}
# COMMAND_MEDIA_SHUFFLE
calls = async_mock_service(
hass, media_player.DOMAIN, media_player.SERVICE_SHUFFLE_SET
)
await trt.execute(trait.COMMAND_MEDIA_SHUFFLE, BASIC_DATA, {}, {})
assert len(calls) == 1
assert calls[0].data == {
ATTR_ENTITY_ID: "media_player.bla",
media_player.ATTR_MEDIA_SHUFFLE: True,
}
# COMMAND_MEDIA_STOP
calls = async_mock_service(
hass, media_player.DOMAIN, media_player.SERVICE_MEDIA_STOP
)
await trt.execute(trait.COMMAND_MEDIA_STOP, BASIC_DATA, {}, {})
assert len(calls) == 1
assert calls[0].data == {ATTR_ENTITY_ID: "media_player.bla"}
@pytest.mark.parametrize(
"state",
[
STATE_OFF,
STATE_IDLE,
STATE_PLAYING,
STATE_ON,
STATE_PAUSED,
STATE_STANDBY,
STATE_UNAVAILABLE,
STATE_UNKNOWN,
],
)
async def test_media_state(hass: HomeAssistant, state) -> None:
"""Test the MediaStateTrait."""
assert helpers.get_google_type(media_player.DOMAIN, None) is not None
assert trait.TransportControlTrait.supported(
media_player.DOMAIN, MediaPlayerEntityFeature.PLAY, None, None
)
trt = trait.MediaStateTrait(
hass,
State(
"media_player.bla",
state,
{
media_player.ATTR_MEDIA_POSITION: 100,
media_player.ATTR_MEDIA_DURATION: 200,
media_player.ATTR_MEDIA_VOLUME_LEVEL: 0.5,
ATTR_SUPPORTED_FEATURES: MediaPlayerEntityFeature.PLAY
| MediaPlayerEntityFeature.STOP,
},
),
BASIC_CONFIG,
)
assert trt.sync_attributes() == {
"supportActivityState": True,
"supportPlaybackState": True,
}
assert trt.query_attributes() == {
"activityState": trt.activity_lookup.get(state),
"playbackState": trt.playback_lookup.get(state),
}
async def test_channel(hass: HomeAssistant) -> None:
"""Test Channel trait support."""
assert helpers.get_google_type(media_player.DOMAIN, None) is not None
assert trait.ChannelTrait.supported(
media_player.DOMAIN,
MediaPlayerEntityFeature.PLAY_MEDIA,
media_player.MediaPlayerDeviceClass.TV,
None,
)
assert (
trait.ChannelTrait.supported(
media_player.DOMAIN,
MediaPlayerEntityFeature.PLAY_MEDIA,
None,
None,
)
is False
)
assert trait.ChannelTrait.supported(media_player.DOMAIN, 0, None, None) is False
trt = trait.ChannelTrait(hass, State("media_player.demo", STATE_ON), BASIC_CONFIG)
assert trt.sync_attributes() == {
"availableChannels": [],
"commandOnlyChannels": True,
}
assert trt.query_attributes() == {}
media_player_calls = async_mock_service(
hass, media_player.DOMAIN, SERVICE_PLAY_MEDIA
)
await trt.execute(
trait.COMMAND_SELECT_CHANNEL, BASIC_DATA, {"channelNumber": "1"}, {}
)
assert len(media_player_calls) == 1
assert media_player_calls[0].data == {
ATTR_ENTITY_ID: "media_player.demo",
media_player.ATTR_MEDIA_CONTENT_ID: "1",
media_player.ATTR_MEDIA_CONTENT_TYPE: MediaType.CHANNEL,
}
with pytest.raises(SmartHomeError, match="Channel is not available"):
await trt.execute(
trait.COMMAND_SELECT_CHANNEL, BASIC_DATA, {"channelCode": "Channel 3"}, {}
)
assert len(media_player_calls) == 1
with pytest.raises(SmartHomeError, match="Unsupported command"):
await trt.execute("Unknown command", BASIC_DATA, {"channelNumber": "1"}, {})
assert len(media_player_calls) == 1
async def test_air_quality_description_for_aqi(hass: HomeAssistant) -> None:
"""Test air quality description for a given AQI value."""
trt = trait.SensorStateTrait(
hass,
State(
"sensor.test",
100.0,
{
"device_class": sensor.SensorDeviceClass.AQI,
},
),
BASIC_CONFIG,
)
assert trt._air_quality_description_for_aqi(0) == "healthy"
assert trt._air_quality_description_for_aqi(75) == "moderate"
assert (
trt._air_quality_description_for_aqi(125.0) == "unhealthy for sensitive groups"
)
assert trt._air_quality_description_for_aqi(175) == "unhealthy"
assert trt._air_quality_description_for_aqi(250) == "very unhealthy"
assert trt._air_quality_description_for_aqi(350) == "hazardous"
assert trt._air_quality_description_for_aqi(-1) == "unknown"
async def test_null_device_class(hass: HomeAssistant) -> None:
"""Test handling a null device_class in sync_attributes and query_attributes."""
trt = trait.SensorStateTrait(
hass,
State(
"sensor.test",
100.0,
{
"device_class": None,
},
),
BASIC_CONFIG,
)
assert trt.sync_attributes() == {}
assert trt.query_attributes() == {}
@pytest.mark.parametrize(
("value", "published", "aqi"),
[
(100.0, 100.0, "moderate"),
(10.0, 10.0, "healthy"),
(0, 0.0, "healthy"),
("", None, "unknown"),
("unknown", None, "unknown"),
],
)
async def test_sensorstate(
hass: HomeAssistant, value: Any, published: Any, aqi: Any
) -> None:
"""Test SensorState trait support for sensor domain."""
sensor_types = {
sensor.SensorDeviceClass.AQI: ("AirQuality", "AQI"),
sensor.SensorDeviceClass.CO: ("CarbonMonoxideLevel", "PARTS_PER_MILLION"),
sensor.SensorDeviceClass.CO2: ("CarbonDioxideLevel", "PARTS_PER_MILLION"),
sensor.SensorDeviceClass.PM25: ("PM2.5", "MICROGRAMS_PER_CUBIC_METER"),
sensor.SensorDeviceClass.PM10: ("PM10", "MICROGRAMS_PER_CUBIC_METER"),
sensor.SensorDeviceClass.VOLATILE_ORGANIC_COMPOUNDS: (
"VolatileOrganicCompounds",
"PARTS_PER_MILLION",
),
}
for sensor_type, item in sensor_types.items():
assert helpers.get_google_type(sensor.DOMAIN, None) is not None
assert trait.SensorStateTrait.supported(sensor.DOMAIN, None, sensor_type, None)
trt = trait.SensorStateTrait(
hass,
State(
"sensor.test",
value,
{
"device_class": sensor_type,
},
),
BASIC_CONFIG,
)
name = item[0]
unit = item[1]
if sensor_type == sensor.SensorDeviceClass.AQI:
assert trt.sync_attributes() == {
"sensorStatesSupported": [
{
"name": name,
"numericCapabilities": {"rawValueUnit": unit},
"descriptiveCapabilities": {
"availableStates": [
"healthy",
"moderate",
"unhealthy for sensitive groups",
"unhealthy",
"very unhealthy",
"hazardous",
"unknown",
],
},
}
]
}
else:
assert trt.sync_attributes() == {
"sensorStatesSupported": [
{
"name": name,
"numericCapabilities": {"rawValueUnit": unit},
}
]
}
if sensor_type == sensor.SensorDeviceClass.AQI:
assert trt.query_attributes() == {
"currentSensorStateData": [
{
"name": name,
"currentSensorState": aqi,
"rawValue": published,
},
]
}
else:
assert trt.query_attributes() == {
"currentSensorStateData": [{"name": name, "rawValue": published}]
}
assert helpers.get_google_type(sensor.DOMAIN, None) is not None
assert (
trait.SensorStateTrait.supported(
sensor.DOMAIN, None, sensor.SensorDeviceClass.MONETARY, None
)
is False
)
@pytest.mark.parametrize(
("state", "identifier"),
[
(STATE_ON, 0),
(STATE_OFF, 1),
(STATE_UNKNOWN, 2),
],
)
@pytest.mark.parametrize(
("device_class", "name", "states"),
[
(
binary_sensor.BinarySensorDeviceClass.CO,
"CarbonMonoxideLevel",
["carbon monoxide detected", "no carbon monoxide detected", "unknown"],
),
(
binary_sensor.BinarySensorDeviceClass.SMOKE,
"SmokeLevel",
["smoke detected", "no smoke detected", "unknown"],
),
(
binary_sensor.BinarySensorDeviceClass.MOISTURE,
"WaterLeak",
["leak", "no leak", "unknown"],
),
],
)
async def test_binary_sensorstate(
hass: HomeAssistant,
state: str,
identifier: int,
device_class: binary_sensor.BinarySensorDeviceClass,
name: str,
states: list[str],
) -> None:
"""Test SensorState trait support for binary sensor domain."""
assert helpers.get_google_type(binary_sensor.DOMAIN, None) is not None
assert trait.SensorStateTrait.supported(
binary_sensor.DOMAIN, None, device_class, None
)
trt = trait.SensorStateTrait(
hass,
State(
"binary_sensor.test",
state,
{
"device_class": device_class,
},
),
BASIC_CONFIG,
)
assert trt.sync_attributes() == {
"sensorStatesSupported": [
{
"name": name,
"descriptiveCapabilities": {
"availableStates": states,
},
}
]
}
assert trt.query_attributes() == {
"currentSensorStateData": [
{
"name": name,
"currentSensorState": states[identifier],
"rawValue": None,
},
]
}
assert helpers.get_google_type(binary_sensor.DOMAIN, None) is not None
assert (
trait.SensorStateTrait.supported(
binary_sensor.DOMAIN,
None,
binary_sensor.BinarySensorDeviceClass.TAMPER,
None,
)
is False
)
Reference in New Issue View Git Blame Copy Permalink