260 lines
9.6 KiB
Python
260 lines
9.6 KiB
Python
"""Tests hors ligne des capteurs Arkteos alimentés par callbacks."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
from unittest.mock import Mock
|
|
|
|
import pytest
|
|
|
|
from custom_components.arkteos.const import DOMAIN
|
|
from custom_components.arkteos.parser import filter_values, parse_frame
|
|
from custom_components.arkteos.sensor import (
|
|
FRIGO_SENSORS,
|
|
FRIGO_STATUS_OPTIONS,
|
|
PAC_STATUS_OPTIONS,
|
|
REGULATION_SENSORS,
|
|
SENSOR_DESCRIPTIONS,
|
|
ArkteosSensor,
|
|
async_setup_entry,
|
|
)
|
|
|
|
|
|
FIXTURES = Path(__file__).parent / "fixtures"
|
|
|
|
|
|
class FakeClient:
|
|
"""Client en mémoire, sans réseau, pour les capteurs."""
|
|
|
|
def __init__(self) -> None:
|
|
self.available = False
|
|
self.latest_frigo_data: dict[str, int | float | str] | None = None
|
|
self.latest_regulation_data: dict[str, int | float | str] | None = None
|
|
self.data_callbacks: list[object] = []
|
|
self.availability_callbacks: list[object] = []
|
|
|
|
def add_data_callback(self, callback) -> None:
|
|
self.data_callbacks.append(callback)
|
|
|
|
def remove_data_callback(self, callback) -> None:
|
|
self.data_callbacks.remove(callback)
|
|
|
|
def add_availability_callback(self, callback) -> None:
|
|
self.availability_callbacks.append(callback)
|
|
|
|
def remove_availability_callback(self, callback) -> None:
|
|
self.availability_callbacks.remove(callback)
|
|
|
|
def publish(
|
|
self,
|
|
frame_type: str,
|
|
data: dict[str, int | float | str],
|
|
rejected: dict[str, int | float] | None = None,
|
|
) -> None:
|
|
if frame_type == "frigo":
|
|
self.latest_frigo_data = dict(data)
|
|
elif frame_type == "regulation":
|
|
self.latest_regulation_data = dict(data)
|
|
availability_changed = not self.available
|
|
self.available = True
|
|
if availability_changed:
|
|
for callback in tuple(self.availability_callbacks):
|
|
callback(True)
|
|
for callback in tuple(self.data_callbacks):
|
|
callback(frame_type, dict(data), rejected or {})
|
|
|
|
|
|
def _frame_values(name: str) -> dict[str, int | float | str]:
|
|
filtered = filter_values(parse_frame((FIXTURES / name).read_bytes()))
|
|
return filtered.accepted
|
|
|
|
|
|
def _sensors(client: FakeClient) -> list[ArkteosSensor]:
|
|
return [ArkteosSensor(client, description) for description in SENSOR_DESCRIPTIONS]
|
|
|
|
|
|
async def _add_sensors_to_hass(sensors: list[ArkteosSensor]) -> None:
|
|
"""Exécute le cycle d'ajout qui inscrit les callbacks du client."""
|
|
|
|
for sensor in sensors:
|
|
sensor.async_write_ha_state = Mock()
|
|
await sensor.async_added_to_hass()
|
|
|
|
|
|
def _sensor(sensors: list[ArkteosSensor], key: str) -> ArkteosSensor:
|
|
return next(sensor for sensor in sensors if sensor.entity_description.key == key)
|
|
|
|
|
|
def test_total_sensor_count_and_unique_ids() -> None:
|
|
sensors = _sensors(FakeClient())
|
|
assert len(sensors) == 32
|
|
assert len({sensor.unique_id for sensor in sensors}) == len(sensors)
|
|
|
|
|
|
def test_all_node_red_keys_are_described() -> None:
|
|
assert {description.key for description in FRIGO_SENSORS} == {
|
|
"exterieur_temp",
|
|
"nb_degivrages",
|
|
"temps_compresseur",
|
|
"nb_cycles_compresseur",
|
|
"freq_comp_actuelle",
|
|
"freq_comp_cible",
|
|
"fan_speed_evaporator_1",
|
|
"dc_voltage",
|
|
"statut_frigo_s",
|
|
"statut_frigo",
|
|
"active_error_fri",
|
|
}
|
|
assert {description.key for description in REGULATION_SENSORS} == {
|
|
"puissance_inst_produite",
|
|
"puissance_inst_consommee",
|
|
"temps_mise_sous_tension",
|
|
"modele_pac_s",
|
|
"modele_pac",
|
|
"primaire_temp_eau_aller_consigne",
|
|
"primaire_temp_eau_aller",
|
|
"primaire_temp_eau_retour",
|
|
"primaire_debit_eau",
|
|
"primaire_pression",
|
|
"primaire_circulateur_consigne",
|
|
"zone1_temp_interieur",
|
|
"zone1_consigne",
|
|
"ecs_temp_eau_milieu",
|
|
"ecs_temp_eau_bas",
|
|
"ecs_consigne",
|
|
"nb_cycles_compresseur_reg",
|
|
"statut_pac_s",
|
|
"statut_pac",
|
|
"active_error_reg",
|
|
"signal_rf_sonde_1",
|
|
}
|
|
|
|
|
|
def test_initial_sensor_state_is_unavailable() -> None:
|
|
for sensor in _sensors(FakeClient()):
|
|
assert not sensor.available
|
|
assert sensor.native_value is None
|
|
|
|
|
|
def test_device_info_is_shared_by_all_sensors() -> None:
|
|
infos = [sensor.device_info for sensor in _sensors(FakeClient())]
|
|
assert all(info["identifiers"] == {(DOMAIN, "arkteos_zuran4")} for info in infos)
|
|
assert all(info["name"] == "PAC Arkteos Zuran 4" for info in infos)
|
|
assert all(info["manufacturer"] == "Arkteos" for info in infos)
|
|
|
|
|
|
async def test_frigo_frame_only_makes_frigo_sensors_available() -> None:
|
|
client = FakeClient()
|
|
sensors = _sensors(client)
|
|
await _add_sensors_to_hass(sensors)
|
|
client.publish("frigo", _frame_values("frigo_163.bin"))
|
|
assert all(sensor.available for sensor in sensors if sensor.entity_description.frame_type == "frigo")
|
|
assert not any(sensor.available for sensor in sensors if sensor.entity_description.frame_type == "regulation")
|
|
|
|
|
|
async def test_regulation_frame_keeps_frigo_available_and_enables_regulation() -> None:
|
|
client = FakeClient()
|
|
sensors = _sensors(client)
|
|
await _add_sensors_to_hass(sensors)
|
|
client.publish("frigo", _frame_values("frigo_163.bin"))
|
|
client.publish("regulation", _frame_values("regulation_227.bin"))
|
|
assert all(sensor.available for sensor in sensors)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("key", "fixture_name"),
|
|
(
|
|
("exterieur_temp", "frigo_163.bin"),
|
|
("primaire_temp_eau_aller", "regulation_227.bin"),
|
|
("primaire_temp_eau_retour", "regulation_227.bin"),
|
|
("puissance_inst_consommee", "regulation_227.bin"),
|
|
("puissance_inst_produite", "regulation_227.bin"),
|
|
("primaire_pression", "regulation_227.bin"),
|
|
("ecs_temp_eau_milieu", "regulation_227.bin"),
|
|
("ecs_temp_eau_bas", "regulation_227.bin"),
|
|
("statut_frigo_s", "frigo_163.bin"),
|
|
("statut_pac_s", "regulation_227.bin"),
|
|
),
|
|
)
|
|
async def test_fixture_values_match_parser(key: str, fixture_name: str) -> None:
|
|
client = FakeClient()
|
|
sensor = _sensor(_sensors(client), key)
|
|
await _add_sensors_to_hass([sensor])
|
|
values = _frame_values(fixture_name)
|
|
client.publish(values["frame_type"], values)
|
|
assert sensor.native_value == values[key]
|
|
|
|
|
|
def test_status_enum_options_are_exact() -> None:
|
|
descriptions = {description.key: description for description in SENSOR_DESCRIPTIONS}
|
|
assert descriptions["statut_frigo_s"].options == FRIGO_STATUS_OPTIONS
|
|
assert descriptions["statut_pac_s"].options == PAC_STATUS_OPTIONS
|
|
|
|
|
|
def test_diagnostic_and_disabled_descriptions() -> None:
|
|
descriptions = {description.key: description for description in SENSOR_DESCRIPTIONS}
|
|
for key in ("statut_frigo", "statut_pac", "modele_pac", "zone1_temp_interieur", "zone1_consigne"):
|
|
assert not descriptions[key].entity_registry_enabled_default
|
|
for key in ("active_error_fri", "active_error_reg", "signal_rf_sonde_1"):
|
|
assert descriptions[key].entity_category.value == "diagnostic"
|
|
|
|
|
|
async def test_rejected_value_does_not_replace_last_valid_value() -> None:
|
|
client = FakeClient()
|
|
sensor = _sensor(_sensors(client), "exterieur_temp")
|
|
await _add_sensors_to_hass([sensor])
|
|
valid = _frame_values("frigo_163.bin")
|
|
client.publish("frigo", valid)
|
|
previous_value = sensor.native_value
|
|
client.publish("frigo", {"frame_type": "frigo"}, {"exterieur_temp": 151})
|
|
assert sensor.native_value == previous_value
|
|
|
|
|
|
async def test_entity_added_after_frame_uses_latest_data() -> None:
|
|
client = FakeClient()
|
|
values = _frame_values("frigo_163.bin")
|
|
client.publish("frigo", values)
|
|
sensor = _sensor(_sensors(client), "exterieur_temp")
|
|
await _add_sensors_to_hass([sensor])
|
|
assert sensor.available
|
|
assert sensor.native_value == values["exterieur_temp"]
|
|
|
|
|
|
def test_callbacks_only_update_their_own_frame_type() -> None:
|
|
client = FakeClient()
|
|
sensors = _sensors(client)
|
|
frigo_sensor = _sensor(sensors, "exterieur_temp")
|
|
regulation_sensor = _sensor(sensors, "primaire_temp_eau_aller")
|
|
frigo_sensor.async_write_ha_state = Mock()
|
|
regulation_sensor.async_write_ha_state = Mock()
|
|
client.add_data_callback(frigo_sensor._handle_data)
|
|
client.add_data_callback(regulation_sensor._handle_data)
|
|
client.publish("frigo", _frame_values("frigo_163.bin"))
|
|
frigo_sensor.async_write_ha_state.assert_called_once()
|
|
regulation_sensor.async_write_ha_state.assert_not_called()
|
|
|
|
|
|
async def test_sensor_callback_is_removed_when_entity_is_removed() -> None:
|
|
client = FakeClient()
|
|
sensor = _sensor(_sensors(client), "exterieur_temp")
|
|
client.add_data_callback(sensor._handle_data)
|
|
client.add_availability_callback(sensor._handle_availability)
|
|
await sensor.async_will_remove_from_hass()
|
|
assert sensor._handle_data not in client.data_callbacks
|
|
assert sensor._handle_availability not in client.availability_callbacks
|
|
|
|
|
|
async def test_sensor_platform_creates_all_descriptions(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
import custom_components.arkteos.sensor as sensor_platform
|
|
|
|
client = FakeClient()
|
|
created: list[ArkteosSensor] = []
|
|
monkeypatch.setattr(sensor_platform, "_get_client", lambda _hass, _entry: client)
|
|
await async_setup_entry(None, object(), lambda entities: created.extend(entities))
|
|
assert len(created) == len(SENSOR_DESCRIPTIONS)
|
|
|
|
|
|
def test_sensor_entities_do_not_poll_or_use_mqtt() -> None:
|
|
assert all(sensor._attr_should_poll is False for sensor in _sensors(FakeClient()))
|
|
assert "mqtt" not in Path(__file__).parents[1].joinpath("custom_components/arkteos/sensor.py").read_text().lower()
|