"""Сущности fglair против mock_ac.py: climate/switch/select/sensor/binary.""" from __future__ import annotations import asyncio import time from helpers import DSN, KEY_ID, LANIP_KEY from homeassistant.core import HomeAssistant from homeassistant.helpers import entity_registry as er from pytest_homeassistant_custom_component.common import MockConfigEntry from custom_components.fglair.const import DOMAIN from pyfglair import State MOCK_SETTINGS = [ "--set", "operation_mode=6", "--set", "fan_speed=4", "--set", "adjust_temperature=220", "--set", "display_temperature=7000", "--set", "device_capabilities=0b1111111111111111111", "--set", "af_vertical_num_dir=5", "--set", "af_horizontal_num_dir=7", "--set", "af_vertical_direction=2", "--set", "af_horizontal_direction=3", "--set", "op_status=0b1000000010000000000000000", "--set", "economy_mode=0", "--set", "powerful_mode=0", ] async def _wait_for(predicate, timeout: float = 20.0): deadline = time.monotonic() + timeout while time.monotonic() < deadline: result = predicate() if result: return result await asyncio.sleep(0.1) raise AssertionError("условие не выполнено за отведённое время") def _entity_id(hass: HomeAssistant, platform: str, unique_id: str): return er.async_get(hass).async_get_entity_id(platform, DOMAIN, unique_id) async def _setup_entry(hass: HomeAssistant, port: int) -> MockConfigEntry: entry = MockConfigEntry( domain=DOMAIN, data={ "host": "127.0.0.1", "device_port": port, "dsn": DSN, "lanip_key": LANIP_KEY, "lanip_key_id": KEY_ID, "template": "A", "name": "Mock AC", "listen_port": 0, }, unique_id=DSN, ) entry.add_to_hass(hass) assert await hass.config_entries.async_setup(entry.entry_id) await hass.async_block_till_done() return entry async def test_entities_snapshot(hass: HomeAssistant, mock_ac, socket_enabled): proc = mock_ac(MOCK_SETTINGS) entry = await _setup_entry(hass, proc.port) runtime = entry.runtime_data await _wait_for(lambda: runtime.client.state == State.ONLINE) climate_id = _entity_id(hass, "climate", f"{DSN}_climate") await _wait_for( lambda: climate_id and hass.states.get(climate_id) is not None and hass.states.get(climate_id).state == "heat" and hass.states.get(climate_id).attributes["swing_mode"] == "off" ) climate = hass.states.get(climate_id) assert climate.attributes["current_temperature"] == 20.0 assert climate.attributes["temperature"] == 22.0 assert climate.attributes["fan_mode"] == "auto" assert climate.attributes["preset_mode"] is None assert "heat" in climate.attributes["hvac_modes"] assert "cool" in climate.attributes["hvac_modes"] assert climate.attributes["swing_mode"] == "off" room = _entity_id(hass, "sensor", f"{DSN}_display_temperature") await _wait_for( lambda: room and hass.states.get(room) is not None and hass.states.get(room).state == "20.0" ) connectivity = _entity_id(hass, "binary_sensor", f"{DSN}_connectivity") assert hass.states.get(connectivity).state == "on" defrost = _entity_id(hass, "binary_sensor", f"{DSN}_op_status_defrost") await _wait_for(lambda: hass.states.get(defrost).state == "on") maintenance = _entity_id( hass, "binary_sensor", f"{DSN}_op_status_maintenance" ) assert hass.states.get(maintenance).state == "off" vertical = _entity_id( hass, "select", f"{DSN}_af_vertical_direction" ) await _wait_for( lambda: hass.states.get(vertical).attributes["options"] == ["0", "1", "2", "3", "4"] ) vertical_state = hass.states.get(vertical) assert vertical_state.state == "2" economy = _entity_id(hass, "switch", f"{DSN}_economy_mode") assert hass.states.get(economy).state == "off" connection = _entity_id(hass, "sensor", f"{DSN}_connection_state") assert hass.states.get(connection).state == "online" assert await hass.config_entries.async_unload(entry.entry_id) async def test_entity_commands(hass: HomeAssistant, mock_ac, socket_enabled): proc = mock_ac(MOCK_SETTINGS) entry = await _setup_entry(hass, proc.port) runtime = entry.runtime_data await _wait_for(lambda: runtime.client.state == State.ONLINE) climate_id = _entity_id(hass, "climate", f"{DSN}_climate") select_id = _entity_id(hass, "select", f"{DSN}_af_vertical_direction") await _wait_for( lambda: climate_id and hass.states.get(climate_id) is not None and hass.states.get(climate_id).state == "heat" and hass.states.get(select_id).attributes["options"] == ["0", "1", "2", "3", "4"] ) await hass.services.async_call( "climate", "set_hvac_mode", {"entity_id": climate_id, "hvac_mode": "cool"}, blocking=True, ) assert proc.wait_line("CMD SET operation_mode=3", 10) await hass.services.async_call( "climate", "set_temperature", {"entity_id": climate_id, "temperature": 25.5}, blocking=True, ) assert proc.wait_line("CMD SET adjust_temperature=255", 10) await hass.services.async_call( "climate", "set_fan_mode", {"entity_id": climate_id, "fan_mode": "low"}, blocking=True, ) assert proc.wait_line("CMD SET fan_speed=1", 10) await hass.services.async_call( "climate", "set_preset_mode", {"entity_id": climate_id, "preset_mode": "eco"}, blocking=True, ) assert proc.wait_line("CMD SET economy_mode=True", 10) assert proc.wait_line("CMD SET powerful_mode=False", 10) await _wait_for( lambda: hass.states.get(climate_id).state == "cool" ) await hass.services.async_call( "select", "select_option", {"entity_id": select_id, "option": "4"}, blocking=True, ) assert proc.wait_line("CMD SET af_vertical_direction=4", 10) switch_id = _entity_id(hass, "switch", f"{DSN}_economy_mode") await hass.services.async_call( "switch", "turn_on", {"entity_id": switch_id}, blocking=True ) assert proc.wait_line("CMD SET economy_mode=True", 10) await hass.services.async_call( "climate", "turn_off", {"entity_id": climate_id}, blocking=True ) assert proc.wait_line("CMD SET operation_mode=0", 10) await hass.services.async_call( "climate", "turn_on", {"entity_id": climate_id}, blocking=True ) assert proc.wait_line("CMD SET operation_mode=1", 10) assert await hass.config_entries.async_unload(entry.entry_id) async def test_reload_online(hass: HomeAssistant, mock_ac, socket_enabled): """После unload тот же прибор снова поднимается до ONLINE.""" proc = mock_ac(MOCK_SETTINGS) entry = await _setup_entry(hass, proc.port) runtime = entry.runtime_data await _wait_for(lambda: runtime.client.state == State.ONLINE) assert await hass.config_entries.async_unload(entry.entry_id) await hass.async_block_till_done() assert await hass.config_entries.async_setup(entry.entry_id) await hass.async_block_till_done() runtime = entry.runtime_data await _wait_for(lambda: runtime.client.state == State.ONLINE) RESTRICTED_SETTINGS = [ "--set", "operation_mode=3", "--set", "fan_speed=4", "--set", "device_capabilities=0b100001", # cool (0) + fan auto (5) ] async def test_capabilities_filter( hass: HomeAssistant, mock_ac, socket_enabled ): proc = mock_ac(RESTRICTED_SETTINGS) entry = await _setup_entry(hass, proc.port) runtime = entry.runtime_data await _wait_for(lambda: runtime.client.state == State.ONLINE) climate_id = _entity_id(hass, "climate", f"{DSN}_climate") await _wait_for( lambda: climate_id and hass.states.get(climate_id) is not None and hass.states.get(climate_id).state == "cool" and "heat" not in hass.states.get(climate_id).attributes["hvac_modes"] ) climate = hass.states.get(climate_id) assert "heat" not in climate.attributes["hvac_modes"] assert "cool" in climate.attributes["hvac_modes"] assert climate.attributes["fan_modes"] == ["auto"] assert climate.attributes.get("preset_modes") is None assert await hass.config_entries.async_unload(entry.entry_id) async def test_manual_conversion_applied( hass: HomeAssistant, mock_ac, socket_enabled ): settings = [ "--set", "operation_mode=6", "--set", "adjust_temperature=220", "--set", "display_temperature=7000", ] proc = mock_ac(settings) entry = MockConfigEntry( domain=DOMAIN, data={ "host": "127.0.0.1", "device_port": proc.port, "dsn": DSN, "lanip_key": LANIP_KEY, "lanip_key_id": KEY_ID, "template": "A", "name": "Mock AC", "listen_port": 0, "temp_step": 5, "overrides": { "adjust_temperature": { "num": 1, "den": 1, "offset": 5, "min": 170, "max": 280, } }, }, unique_id=DSN, ) entry.add_to_hass(hass) assert await hass.config_entries.async_setup(entry.entry_id) await hass.async_block_till_done() runtime = entry.runtime_data await _wait_for(lambda: runtime.client.state == State.ONLINE) climate_id = _entity_id(hass, "climate", f"{DSN}_climate") await _wait_for( lambda: climate_id and hass.states.get(climate_id) is not None and hass.states.get(climate_id).state == "heat" and hass.states.get(climate_id).attributes["temperature"] == 22.5 ) assert hass.states.get(climate_id).attributes["temperature"] == 22.5 assert hass.states.get(climate_id).attributes["min_temp"] == 17.0 assert hass.states.get(climate_id).attributes["max_temp"] == 28.0 await hass.services.async_call( "climate", "set_temperature", {"entity_id": climate_id, "temperature": 25.0}, blocking=True, ) assert proc.wait_line("CMD SET adjust_temperature=245", 10) assert await hass.config_entries.async_unload(entry.entry_id)