Add IntelliClima integration and tests (#157363)

Co-authored-by: Joostlek <joostlek@outlook.com>
This commit is contained in:
dvdinth
2026-02-11 11:18:26 +01:00
committed by GitHub
parent eab80f78d9
commit 3f9e7d1dba
21 changed files with 1177 additions and 0 deletions
+1
View File
@@ -287,6 +287,7 @@ homeassistant.components.input_button.*
homeassistant.components.input_select.*
homeassistant.components.input_text.*
homeassistant.components.integration.*
homeassistant.components.intelliclima.*
homeassistant.components.intent.*
homeassistant.components.intent_script.*
homeassistant.components.ios.*
Generated
+2
View File
@@ -804,6 +804,8 @@ build.json @home-assistant/supervisor
/tests/components/insteon/ @teharris1
/homeassistant/components/integration/ @dgomes
/tests/components/integration/ @dgomes
/homeassistant/components/intelliclima/ @dvdinth
/tests/components/intelliclima/ @dvdinth
/homeassistant/components/intellifire/ @jeeftor
/tests/components/intellifire/ @jeeftor
/homeassistant/components/intent/ @home-assistant/core @synesthesiam @arturpragacz
@@ -0,0 +1,51 @@
"""The IntelliClima VMC integration."""
from pyintelliclima.api import IntelliClimaAPI
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .const import LOGGER
from .coordinator import IntelliClimaConfigEntry, IntelliClimaCoordinator
PLATFORMS = [Platform.FAN]
async def async_setup_entry(
hass: HomeAssistant, entry: IntelliClimaConfigEntry
) -> bool:
"""Set up IntelliClima VMC from a config entry."""
# Create API client
session = async_get_clientsession(hass)
api = IntelliClimaAPI(
session,
entry.data[CONF_USERNAME],
entry.data[CONF_PASSWORD],
)
# Create coordinator
coordinator = IntelliClimaCoordinator(hass, entry, api)
# Fetch initial data
await coordinator.async_config_entry_first_refresh()
LOGGER.debug(
"Discovered %d IntelliClima VMC device(s)",
len(coordinator.data.ecocomfort2_devices),
)
# Store coordinator
entry.runtime_data = coordinator
# Set up platforms
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True
async def async_unload_entry(
hass: HomeAssistant, entry: IntelliClimaConfigEntry
) -> bool:
"""Unload a config entry."""
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
@@ -0,0 +1,71 @@
"""Config flow for IntelliClima integration."""
from typing import Any
from pyintelliclima import IntelliClimaAPI, IntelliClimaAPIError, IntelliClimaAuthError
import voluptuous as vol
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .const import DOMAIN, LOGGER
STEP_USER_DATA_SCHEMA = vol.Schema(
{
vol.Required(CONF_USERNAME): str,
vol.Required(CONF_PASSWORD): str,
}
)
class IntelliClimaConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a config flow for IntelliClima VMC."""
VERSION = 1
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle the initial step."""
errors = {}
if user_input is not None:
self._async_abort_entries_match({CONF_USERNAME: user_input[CONF_USERNAME]})
# Validate credentials
session = async_get_clientsession(self.hass)
api = IntelliClimaAPI(
session,
user_input[CONF_USERNAME],
user_input[CONF_PASSWORD],
)
try:
# Test authentication
await api.authenticate()
# Get devices to ensure we can communicate with API
devices = await api.get_all_device_status()
except IntelliClimaAuthError:
errors["base"] = "invalid_auth"
except IntelliClimaAPIError:
errors["base"] = "cannot_connect"
except Exception: # noqa: BLE001
LOGGER.exception("Unexpected exception")
errors["base"] = "unknown"
else:
if devices.num_devices == 0:
errors["base"] = "no_devices"
else:
return self.async_create_entry(
title=f"IntelliClima ({user_input[CONF_USERNAME]})",
data=user_input,
)
return self.async_show_form(
step_id="user",
data_schema=STEP_USER_DATA_SCHEMA,
errors=errors,
)
@@ -0,0 +1,11 @@
"""Constants for the IntelliClima integration."""
from datetime import timedelta
import logging
LOGGER = logging.getLogger(__package__)
DOMAIN = "intelliclima"
# Update interval
DEFAULT_SCAN_INTERVAL = timedelta(minutes=1)
@@ -0,0 +1,45 @@
"""DataUpdateCoordinator for IntelliClima."""
from pyintelliclima import IntelliClimaAPI, IntelliClimaAPIError, IntelliClimaDevices
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import DEFAULT_SCAN_INTERVAL, DOMAIN, LOGGER
type IntelliClimaConfigEntry = ConfigEntry[IntelliClimaCoordinator]
class IntelliClimaCoordinator(DataUpdateCoordinator[IntelliClimaDevices]):
"""Coordinator to manage fetching IntelliClima data."""
def __init__(
self, hass: HomeAssistant, entry: IntelliClimaConfigEntry, api: IntelliClimaAPI
) -> None:
"""Initialize the coordinator."""
super().__init__(
hass,
LOGGER,
name=DOMAIN,
update_interval=DEFAULT_SCAN_INTERVAL,
config_entry=entry,
)
self.api = api
async def _async_setup(self) -> None:
"""Set up the coordinator - called once during first refresh."""
# Authenticate and get initial device list
try:
await self.api.authenticate()
except IntelliClimaAPIError as err:
raise UpdateFailed(f"Failed to set up IntelliClima: {err}") from err
async def _async_update_data(self) -> IntelliClimaDevices:
"""Fetch data from API."""
try:
# Poll status for all devices
return await self.api.get_all_device_status()
except IntelliClimaAPIError as err:
raise UpdateFailed(f"Failed to update data: {err}") from err
@@ -0,0 +1,74 @@
"""Platform for shared base classes for sensors."""
from pyintelliclima.intelliclima_types import IntelliClimaC800, IntelliClimaECO
from homeassistant.const import ATTR_CONNECTIONS, ATTR_MODEL, ATTR_SW_VERSION
from homeassistant.helpers.device_registry import (
CONNECTION_BLUETOOTH,
CONNECTION_NETWORK_MAC,
DeviceInfo,
)
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN
from .coordinator import IntelliClimaCoordinator
class IntelliClimaEntity(CoordinatorEntity[IntelliClimaCoordinator]):
"""Define a generic class for IntelliClima entities."""
_attr_has_entity_name = True
def __init__(
self,
coordinator: IntelliClimaCoordinator,
device: IntelliClimaECO | IntelliClimaC800,
) -> None:
"""Class initializer."""
super().__init__(coordinator=coordinator)
self._attr_unique_id = device.id
# Make this HA "device" use the IntelliClima device name.
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, device.id)},
manufacturer="Fantini Cosmi",
name=device.name,
serial_number=device.crono_sn,
)
self._device_id = device.id
self._device_sn = device.crono_sn
class IntelliClimaECOEntity(IntelliClimaEntity):
"""Specific entity for the ECOCOMFORT 2.0."""
def __init__(
self,
coordinator: IntelliClimaCoordinator,
device: IntelliClimaECO,
) -> None:
"""Class initializer."""
super().__init__(coordinator, device)
self._attr_device_info: DeviceInfo = self.device_info or DeviceInfo()
self._attr_device_info[ATTR_MODEL] = "ECOCOMFORT 2.0"
self._attr_device_info[ATTR_SW_VERSION] = device.fw
self._attr_device_info[ATTR_CONNECTIONS] = {
(CONNECTION_BLUETOOTH, device.mac),
(CONNECTION_NETWORK_MAC, device.macwifi),
}
@property
def _device_data(self) -> IntelliClimaECO:
return self.coordinator.data.ecocomfort2_devices[self._device_id]
@property
def available(self) -> bool:
"""Return if entity is available."""
return (
super().available
and self._device_id in self.coordinator.data.ecocomfort2_devices
)
@@ -0,0 +1,173 @@
"""Fan platform for IntelliClima VMC."""
import math
from typing import Any
from pyintelliclima.const import FanMode, FanSpeed
from pyintelliclima.intelliclima_types import IntelliClimaECO
from homeassistant.components.fan import FanEntity, FanEntityFeature
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.util.percentage import (
percentage_to_ranged_value,
ranged_value_to_percentage,
)
from homeassistant.util.scaling import int_states_in_range
from .coordinator import IntelliClimaConfigEntry, IntelliClimaCoordinator
from .entity import IntelliClimaECOEntity
# Coordinator is used to centralize the data updates
PARALLEL_UPDATES = 0
async def async_setup_entry(
hass: HomeAssistant,
entry: IntelliClimaConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up IntelliClima VMC fans."""
coordinator = entry.runtime_data
entities: list[IntelliClimaVMCFan] = [
IntelliClimaVMCFan(
coordinator=coordinator,
device=ecocomfort2,
)
for ecocomfort2 in coordinator.data.ecocomfort2_devices.values()
]
async_add_entities(entities)
class IntelliClimaVMCFan(IntelliClimaECOEntity, FanEntity):
"""Representation of an IntelliClima VMC fan."""
_attr_name = None
_attr_supported_features = (
FanEntityFeature.PRESET_MODE
| FanEntityFeature.SET_SPEED
| FanEntityFeature.TURN_OFF
| FanEntityFeature.TURN_ON
)
_attr_preset_modes = ["auto"]
def __init__(
self,
coordinator: IntelliClimaCoordinator,
device: IntelliClimaECO,
) -> None:
"""Class initializer."""
super().__init__(coordinator, device)
self._speed_range = (int(FanSpeed.sleep), int(FanSpeed.high))
@property
def is_on(self) -> bool:
"""Return true if fan is on."""
return bool(self._device_data.mode_set != FanMode.off)
@property
def percentage(self) -> int | None:
"""Return the current speed percentage."""
device_data = self._device_data
if device_data.speed_set == FanSpeed.auto:
return None
return ranged_value_to_percentage(self._speed_range, int(device_data.speed_set))
@property
def speed_count(self) -> int:
"""Return the number of speeds the fan supports."""
return int_states_in_range(self._speed_range)
@property
def preset_mode(self) -> str | None:
"""Return the current preset mode."""
device_data = self._device_data
if device_data.mode_set == FanMode.off:
return None
if (
device_data.speed_set == FanSpeed.auto
and device_data.mode_set == FanMode.sensor
):
return "auto"
return None
async def async_turn_on(
self,
percentage: int | None = None,
preset_mode: str | None = None,
**kwargs: Any,
) -> None:
"""Turn on the fan.
Defaults back to 25% if percentage argument is 0 to prevent loop of turning off/on
infinitely.
"""
percentage = 25 if percentage == 0 else percentage
await self.async_set_mode_speed(fan_mode=preset_mode, percentage=percentage)
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn off the fan."""
await self.coordinator.api.ecocomfort.turn_off(self._device_sn)
await self.coordinator.async_request_refresh()
async def async_set_percentage(self, percentage: int) -> None:
"""Set the speed percentage."""
await self.async_set_mode_speed(percentage=percentage)
async def async_set_preset_mode(self, preset_mode: str) -> None:
"""Set preset mode."""
await self.async_set_mode_speed(fan_mode=preset_mode)
async def async_set_mode_speed(
self, fan_mode: str | None = None, percentage: int | None = None
) -> None:
"""Set mode and speed.
If percentage is None, it first defaults to the respective property.
If that is also None, then percentage defaults to 25 (sleep)
"""
percentage = self.percentage if percentage is None else percentage
percentage = 25 if percentage is None else percentage
if fan_mode == "auto":
# auto is a special case with special mode and speed setting
await self.coordinator.api.ecocomfort.set_mode_speed_auto(self._device_sn)
await self.coordinator.async_request_refresh()
return
if percentage == 0:
# Setting fan speed to zero turns off the fan
await self.async_turn_off()
return
# Determine the fan mode
if fan_mode is not None:
# Set to requested fan_mode
mode = fan_mode
elif not self.is_on:
# Default to alternate fan mode if not turned on
mode = FanMode.alternate
else:
# Maintain current mode
mode = self._device_data.mode_set
speed = str(
math.ceil(
percentage_to_ranged_value(
self._speed_range,
percentage,
)
)
)
speed = FanSpeed.sleep if speed == FanSpeed.off else speed
await self.coordinator.api.ecocomfort.set_mode_speed(
self._device_sn, mode, speed
)
await self.coordinator.async_request_refresh()
@@ -0,0 +1,11 @@
{
"domain": "intelliclima",
"name": "IntelliClima",
"codeowners": ["@dvdinth"],
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/intelliclima",
"integration_type": "device",
"iot_class": "cloud_polling",
"quality_scale": "bronze",
"requirements": ["pyintelliclima==0.2.2"]
}
@@ -0,0 +1,75 @@
rules:
# Bronze
action-setup: done
appropriate-polling: done
brands: done
common-modules: done
config-flow-test-coverage: done
config-flow: done
dependency-transparency: done
docs-actions: done
docs-high-level-description: done
docs-installation-instructions: done
docs-removal-instructions: done
entity-event-setup: done
entity-unique-id: done
has-entity-name: done
runtime-data: done
test-before-configure: done
test-before-setup: done
unique-config-entry: done
# Silver
action-exceptions: todo
config-entry-unloading: done
docs-configuration-parameters:
status: exempt
comment: |
No configuration parameters, so nothing to document.
docs-installation-parameters: done
entity-unavailable: done
integration-owner: done
log-when-unavailable: done
parallel-updates: done
reauthentication-flow: todo
test-coverage:
status: todo
comment: |
Currently 92% average, with minimum module at 80% coverage.
# Gold
devices: done
diagnostics: todo
discovery-update-info:
status: todo
comment: |
Unclear if discovery is possible.
discovery:
status: todo
comment: |
Unclear if discovery is possible.
docs-data-update: done
docs-examples: todo
docs-known-limitations: done
docs-supported-devices: done
docs-supported-functions: done
docs-troubleshooting: done
docs-use-cases: done
dynamic-devices: todo
entity-category: done
entity-device-class: done
entity-disabled-by-default: done
entity-translations: done
exception-translations: done
icon-translations: done
reconfiguration-flow: todo
repair-issues: done
stale-devices: todo
# Platinum
async-dependency: done
inject-websession: done
strict-typing:
status: todo
comment: |
External pyintelliclima module does not fully conform to PEP 561 yet.
@@ -0,0 +1,26 @@
{
"config": {
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
"no_devices": "No IntelliClima devices found in your account",
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"step": {
"user": {
"data": {
"password": "[%key:common::config_flow::data::password%]",
"username": "[%key:common::config_flow::data::email%]"
},
"data_description": {
"password": "Your IntelliClima app password",
"username": "Your IntelliClima app username"
},
"description": "Authenticate against IntelliClima cloud"
}
}
}
}
+1
View File
@@ -329,6 +329,7 @@ FLOWS = {
"inels",
"inkbird",
"insteon",
"intelliclima",
"intellifire",
"iometer",
"ios",
@@ -3141,6 +3141,12 @@
"iot_class": "local_push",
"single_config_entry": true
},
"intelliclima": {
"name": "IntelliClima",
"integration_type": "device",
"config_flow": true,
"iot_class": "cloud_polling"
},
"intellifire": {
"name": "IntelliFire",
"integration_type": "device",
Generated
+10
View File
@@ -2626,6 +2626,16 @@ disallow_untyped_defs = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.intelliclima.*]
check_untyped_defs = true
disallow_incomplete_defs = true
disallow_subclassing_any = true
disallow_untyped_calls = true
disallow_untyped_decorators = true
disallow_untyped_defs = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.intent.*]
check_untyped_defs = true
disallow_incomplete_defs = true
+3
View File
@@ -2118,6 +2118,9 @@ pyicloud==2.3.0
# homeassistant.components.insteon
pyinsteon==1.6.4
# homeassistant.components.intelliclima
pyintelliclima==0.2.2
# homeassistant.components.intesishome
pyintesishome==1.8.0
+3
View File
@@ -1801,6 +1801,9 @@ pyicloud==2.3.0
# homeassistant.components.insteon
pyinsteon==1.6.4
# homeassistant.components.intelliclima
pyintelliclima==0.2.2
# homeassistant.components.ipma
pyipma==3.0.9
+13
View File
@@ -0,0 +1,13 @@
"""Tests for the IntelliClima integration."""
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
async def setup_integration(hass: HomeAssistant, config_entry: MockConfigEntry) -> None:
"""Fixture for setting up the component."""
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
+145
View File
@@ -0,0 +1,145 @@
"""Fixtures for IntelliClima integration tests."""
from collections.abc import Generator
from types import SimpleNamespace
from unittest.mock import AsyncMock, patch
from pyintelliclima.intelliclima_types import (
IntelliClimaDevices,
IntelliClimaECO,
IntelliClimaModelType,
)
import pytest
from homeassistant.components.intelliclima.const import DOMAIN
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
from tests.common import MockConfigEntry
@pytest.fixture
def mock_setup_entry() -> Generator[AsyncMock]:
"""Mock setting up a config entry."""
with patch(
"homeassistant.components.intellifire.async_setup_entry", return_value=True
) as mock_setup:
yield mock_setup
@pytest.fixture
def mock_config_entry() -> MockConfigEntry:
"""Return a mock config entry."""
return MockConfigEntry(
domain=DOMAIN,
data={
CONF_USERNAME: "SuperUser",
CONF_PASSWORD: "hunter2",
},
)
@pytest.fixture
def single_eco_device() -> IntelliClimaDevices:
"""Create IntelliClimaDevices with one ECOCOMFORT 2.0 and no C800."""
eco = IntelliClimaECO(
id="56789",
crono_sn="11223344",
status="OK",
online="OK",
command="OK",
model=IntelliClimaModelType(modello="ECO", tipo="wifi"),
name="Test VMC",
houses_id="12345",
mode_set="1",
mode_state="1",
speed_set="3",
speed_state="3",
last_online="2025-11-18 10:22:51",
creation_date="2025-11-18 10:22:51",
fw="0.6.8",
mac="00:11:22:33:44:55",
macwifi="00:11:22:33:44:55",
conn_num="1",
conn_state="0",
role="1",
rh_thrs="2",
lux_thrs="1",
voc_thrs="1",
slv_rot="0",
slv_addr="00:11:22:33:44:55",
offset_temp="0",
offset_hum="0",
year="25",
month="11",
day="10",
dow="0",
hour="22",
minute="41",
second="35",
dst="1",
mode_prev="4",
dir_state="2",
auto_cycle="194",
tamb="16.2",
rh="65",
voc_state="89",
plun="",
pmar="",
pmer="",
pgio="",
pven="",
psab="",
pdom="",
pcustom=None,
sfondo="img/backgrounds/shutterstock_2.jpg0",
tperc=None,
fcool="0",
ws="1",
filter_from="2025-11-18 10:22:51",
filter_active="1",
timezone=None,
co2=None,
sanification=None,
rssi=None,
aqi=None,
co2_thrs=None,
dev_state=None,
online_status=True,
online_status_debug="mock",
)
return IntelliClimaDevices(ecocomfort2_devices={eco.id: eco}, c800_devices={})
@pytest.fixture
def mock_cloud_interface(single_eco_device) -> Generator[AsyncMock]:
"""Mock IntelliClimaAPI for tests."""
with (
patch(
"homeassistant.components.intelliclima.IntelliClimaAPI",
autospec=True,
) as mock_client,
patch(
"homeassistant.components.intelliclima.config_flow.IntelliClimaAPI",
new=mock_client,
),
):
# Mock async context manager
mock_client = mock_client.return_value
# Mock other async methods if needed
mock_client.authenticate.return_value = True
mock_client.get_all_device_status.return_value = single_eco_device
# Sub-API used by the fan entity
mock_client.ecocomfort = SimpleNamespace(
turn_off=AsyncMock(return_value=True),
set_mode_speed=AsyncMock(return_value=True),
set_mode_speed_auto=AsyncMock(return_value=True),
)
mock_client.auth_token = "fake-token"
mock_client.user_id = "fake-user-id"
mock_client.house_id = "fake-house-id"
yield mock_client # Yielding to the test
@@ -0,0 +1,100 @@
# serializer version: 1
# name: test_all_fan_entities.2
DeviceRegistryEntrySnapshot({
'area_id': None,
'config_entries': <ANY>,
'config_entries_subentries': <ANY>,
'configuration_url': None,
'connections': set({
tuple(
'bluetooth',
'00:11:22:33:44:55',
),
tuple(
'mac',
'00:11:22:33:44:55',
),
}),
'disabled_by': None,
'entry_type': None,
'hw_version': None,
'id': <ANY>,
'identifiers': set({
tuple(
'intelliclima',
'56789',
),
}),
'labels': set({
}),
'manufacturer': 'Fantini Cosmi',
'model': 'ECOCOMFORT 2.0',
'model_id': None,
'name': 'Test VMC',
'name_by_user': None,
'primary_config_entry': <ANY>,
'serial_number': '11223344',
'sw_version': '0.6.8',
'via_device_id': None,
})
# ---
# name: test_all_fan_entities[fan.test_vmc-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'preset_modes': list([
'auto',
]),
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'fan',
'entity_category': None,
'entity_id': 'fan.test_vmc',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': None,
'platform': 'intelliclima',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': <FanEntityFeature: 57>,
'translation_key': None,
'unique_id': '56789',
'unit_of_measurement': None,
})
# ---
# name: test_all_fan_entities[fan.test_vmc-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Test VMC',
'percentage': 75,
'percentage_step': 25.0,
'preset_mode': None,
'preset_modes': list([
'auto',
]),
'supported_features': <FanEntityFeature: 57>,
}),
'context': <ANY>,
'entity_id': 'fan.test_vmc',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'on',
})
# ---
@@ -0,0 +1,143 @@
"""Test the IntelliClima config flow."""
from unittest.mock import AsyncMock
from pyintelliclima.api import (
IntelliClimaAPIError,
IntelliClimaAuthError,
IntelliClimaDevices,
)
import pytest
from homeassistant.components.intelliclima.const import DOMAIN
from homeassistant.config_entries import SOURCE_USER
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from tests.common import MockConfigEntry
DATA_CONFIG = {
CONF_USERNAME: "SuperUser",
CONF_PASSWORD: "hunter2",
}
async def test_user_flow(
hass: HomeAssistant, mock_setup_entry: AsyncMock, mock_cloud_interface
) -> None:
"""Test the full config flow."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {}
result = await hass.config_entries.flow.async_configure(
result["flow_id"], DATA_CONFIG
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "IntelliClima (SuperUser)"
assert result["data"] == DATA_CONFIG
@pytest.mark.parametrize(
("side_effect", "error"),
[
# invalid_auth
(IntelliClimaAuthError, "invalid_auth"),
# cannot_connect
(IntelliClimaAPIError, "cannot_connect"),
# unknown
(RuntimeError("Unexpected error"), "unknown"),
],
)
async def test_form_auth_errors(
hass: HomeAssistant,
mock_setup_entry: AsyncMock,
mock_cloud_interface: AsyncMock,
side_effect: Exception,
error: str,
) -> None:
"""Test we handle authentication-related errors and recover."""
mock_cloud_interface.authenticate.side_effect = side_effect
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], DATA_CONFIG
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": error}
# Recover: clear side effect and complete flow successfully
mock_cloud_interface.authenticate.side_effect = None
result = await hass.config_entries.flow.async_configure(
result["flow_id"], DATA_CONFIG
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "IntelliClima (SuperUser)"
assert result["data"] == DATA_CONFIG
async def test_form_no_devices(
hass: HomeAssistant,
mock_setup_entry: AsyncMock,
mock_cloud_interface: AsyncMock,
single_eco_device: IntelliClimaDevices,
) -> None:
"""Test we handle no devices found error."""
# Return empty devices list
mock_cloud_interface.get_all_device_status.return_value = IntelliClimaDevices(
ecocomfort2_devices={}, c800_devices={}
)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], DATA_CONFIG
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": "no_devices"}
# Reset the return_value to its default state
mock_cloud_interface.get_all_device_status.return_value = single_eco_device
result = await hass.config_entries.flow.async_configure(
result["flow_id"], DATA_CONFIG
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "IntelliClima (SuperUser)"
assert result["data"] == DATA_CONFIG
async def test_form_already_configured(
hass: HomeAssistant,
mock_setup_entry: AsyncMock,
mock_cloud_interface: AsyncMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test creating a second config for the same account aborts."""
mock_config_entry.add_to_hass(hass)
# Second attempt with the same account
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], DATA_CONFIG
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
+213
View File
@@ -0,0 +1,213 @@
"""Test IntelliClima Fans."""
from collections.abc import AsyncGenerator
from unittest.mock import AsyncMock, patch
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.fan import (
ATTR_PERCENTAGE,
ATTR_PRESET_MODE,
DOMAIN as FAN_DOMAIN,
SERVICE_SET_PERCENTAGE,
SERVICE_SET_PRESET_MODE,
)
from homeassistant.const import (
ATTR_ENTITY_ID,
SERVICE_TURN_OFF,
SERVICE_TURN_ON,
Platform,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr, entity_registry as er
from . import setup_integration
from tests.common import MockConfigEntry, snapshot_platform
FAN_ENTITY_ID = "fan.test_vmc"
@pytest.fixture(autouse=True)
async def setup_intelliclima_fan_only(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_cloud_interface: AsyncMock,
) -> AsyncGenerator[None]:
"""Set up IntelliClima integration with only the fan platform."""
with patch("homeassistant.components.intelliclima.PLATFORMS", [Platform.FAN]):
await setup_integration(hass, mock_config_entry)
# Let tests run against this initialized state
yield
async def test_all_fan_entities(
hass: HomeAssistant,
snapshot: SnapshotAssertion,
mock_config_entry: MockConfigEntry,
entity_registry: er.EntityRegistry,
device_registry: dr.DeviceRegistry,
mock_cloud_interface: AsyncMock,
) -> None:
"""Test all entities."""
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
# There should be exactly one fan entity
fan_entries = [
entry
for entry in entity_registry.entities.values()
if entry.platform == "intelliclima" and entry.domain == FAN_DOMAIN
]
assert len(fan_entries) == 1
entity_entry = fan_entries[0]
# Device should exist and match snapshot
assert entity_entry.device_id
assert (device_entry := device_registry.async_get(entity_entry.device_id))
assert device_entry == snapshot
async def test_fan_turn_off_service_calls_api(
hass: HomeAssistant,
mock_cloud_interface: AsyncMock,
) -> None:
"""fan.turn_off should call ecocomfort.turn_off and refresh."""
await hass.services.async_call(
FAN_DOMAIN,
SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: FAN_ENTITY_ID},
blocking=True,
)
# Device serial from single_eco_device.crono_sn
mock_cloud_interface.ecocomfort.turn_off.assert_awaited_once_with("11223344")
mock_cloud_interface.ecocomfort.set_mode_speed.assert_not_awaited()
async def test_fan_turn_on_service_calls_api(
hass: HomeAssistant,
mock_cloud_interface: AsyncMock,
) -> None:
"""fan.turn_on should call ecocomfort.turn_on and refresh."""
await hass.services.async_call(
FAN_DOMAIN,
SERVICE_TURN_ON,
{
ATTR_ENTITY_ID: FAN_ENTITY_ID,
ATTR_PERCENTAGE: 30,
},
blocking=True,
)
# Device serial from single_eco_device.crono_sn
mock_cloud_interface.ecocomfort.set_mode_speed.assert_awaited_once_with(
"11223344", "1", "2"
)
async def test_fan_set_percentage_maps_to_speed(
hass: HomeAssistant,
mock_cloud_interface: AsyncMock,
) -> None:
"""fan.set_percentage maps to closest IntelliClima speed via set_mode_speed."""
# 15% is closest to 25% (sleep).
await hass.services.async_call(
FAN_DOMAIN,
SERVICE_SET_PERCENTAGE,
{ATTR_ENTITY_ID: FAN_ENTITY_ID, ATTR_PERCENTAGE: 15},
blocking=True,
)
# Initial mode_set="1" (forward) from single_eco_device.
# Sleep speed is "1" (25%).
mock_cloud_interface.ecocomfort.set_mode_speed.assert_awaited_once_with(
"11223344", "1", "1"
)
async def test_fan_set_preset_mode_service(
hass: HomeAssistant,
mock_cloud_interface: AsyncMock,
) -> None:
"""Tests whether the set preset mode service is called and correct api call is followed."""
await hass.services.async_call(
FAN_DOMAIN,
SERVICE_SET_PRESET_MODE,
{ATTR_ENTITY_ID: FAN_ENTITY_ID, ATTR_PRESET_MODE: "auto"},
blocking=True,
)
mock_cloud_interface.ecocomfort.set_mode_speed_auto.assert_awaited_once_with(
"11223344"
)
mock_cloud_interface.ecocomfort.turn_off.assert_not_awaited()
async def test_fan_set_percentage_zero_turns_off(
hass: HomeAssistant,
mock_cloud_interface: AsyncMock,
) -> None:
"""Setting percentage to 0 should call turn_off, not set_mode_speed."""
await hass.services.async_call(
FAN_DOMAIN,
SERVICE_SET_PERCENTAGE,
{ATTR_ENTITY_ID: FAN_ENTITY_ID, ATTR_PERCENTAGE: 0},
blocking=True,
)
mock_cloud_interface.ecocomfort.turn_off.assert_awaited_once_with("11223344")
mock_cloud_interface.ecocomfort.set_mode_speed.assert_not_awaited()
@pytest.mark.parametrize(
("service_data", "expected_mode", "expected_speed"),
[
# percentage=None, preset_mode=None -> defaults to previous speed > 75% (medium),
# previous mode > "inward"
({}, "1", "3"),
# percentage=0, preset_mode=None -> default 25% (sleep), previous mode (inward)
({ATTR_PERCENTAGE: 0}, "1", "1"),
],
)
async def test_fan_turn_on_defaulting_behavior(
hass: HomeAssistant,
mock_cloud_interface: AsyncMock,
service_data: dict,
expected_mode: str,
expected_speed: str,
) -> None:
"""turn_on defaults percentage/preset as expected."""
data = {ATTR_ENTITY_ID: FAN_ENTITY_ID} | service_data
await hass.services.async_call(
FAN_DOMAIN,
SERVICE_TURN_ON,
data,
blocking=True,
)
mock_cloud_interface.ecocomfort.set_mode_speed.assert_awaited_once_with(
"11223344", expected_mode, expected_speed
)
mock_cloud_interface.ecocomfort.turn_off.assert_not_awaited()
async def test_fan_turn_on_defaulting_behavior_auto_preset(
hass: HomeAssistant,
mock_cloud_interface: AsyncMock,
) -> None:
"""turn_on with auto preset mode calls auto request."""
await hass.services.async_call(
FAN_DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: FAN_ENTITY_ID, ATTR_PRESET_MODE: "auto"},
blocking=True,
)
mock_cloud_interface.ecocomfort.set_mode_speed_auto.assert_awaited_once_with(
"11223344"
)
mock_cloud_interface.ecocomfort.turn_off.assert_not_awaited()