home-assistant/core
1
0
Fork 1
mirror of https://github.com/home-assistant/core.git synced 2025-09-03 11:51:40 +02:00
Code Issues Packages Projects Releases Wiki Activity

Add Seko PoolDose integration (#146972)

Browse Source 
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
Co-authored-by: Norbert Rittel <norbert@rittel.de>
Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>
This commit is contained in:
Lukas
2025-08-27 01:48:50 +02:00
committed by GitHub
parent c8964494a2
commit 51c7bafb41
24 changed files with 2472 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
CODEOWNERS generated
View File

@@ -1183,6 +1183,8 @@ build.json @home-assistant/supervisor
/tests/components/plum_lightpad/ @ColinHarrington @prystupa
/homeassistant/components/point/ @fredrike
/tests/components/point/ @fredrike
/homeassistant/components/pooldose/ @lmaertin
/tests/components/pooldose/ @lmaertin
/homeassistant/components/poolsense/ @haemishkyd
/tests/components/poolsense/ @haemishkyd
/homeassistant/components/powerfox/ @klaasnicolaas

58
homeassistant/components/pooldose/__init__.py Normal file
View File

@@ -0,0 +1,58 @@
"""The Seko Pooldose integration."""
from __future__ import annotations
import logging
from pooldose.client import PooldoseClient
from pooldose.request_status import RequestStatus
from homeassistant.const import CONF_HOST, Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from .coordinator import PooldoseConfigEntry, PooldoseCoordinator
_LOGGER = logging.getLogger(__name__)
PLATFORMS: list[Platform] = [Platform.SENSOR]
async def async_setup_entry(hass: HomeAssistant, entry: PooldoseConfigEntry) -> bool:
"""Set up Seko PoolDose from a config entry."""
# Get host from config entry data (connection-critical configuration)
host = entry.data[CONF_HOST]
# Create the PoolDose API client and connect
client = PooldoseClient(host)
try:
client_status = await client.connect()
except TimeoutError as err:
raise ConfigEntryNotReady(
f"Timeout connecting to PoolDose device: {err}"
) from err
except (ConnectionError, OSError) as err:
raise ConfigEntryNotReady(
f"Failed to connect to PoolDose device: {err}"
) from err
if client_status != RequestStatus.SUCCESS:
raise ConfigEntryNotReady(
f"Failed to create PoolDose client while initialization: {client_status}"
)
# Create coordinator and perform first refresh
coordinator = PooldoseCoordinator(hass, client, entry)
await coordinator.async_config_entry_first_refresh()
# Store runtime data
entry.runtime_data = coordinator
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True
async def async_unload_entry(hass: HomeAssistant, entry: PooldoseConfigEntry) -> bool:
"""Unload the Seko PoolDose entry."""
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)

99
homeassistant/components/pooldose/config_flow.py Normal file
View File

@@ -0,0 +1,99 @@
"""Config flow for the Seko PoolDose integration."""
from __future__ import annotations
import logging
from typing import Any
from pooldose.client import PooldoseClient
from pooldose.request_status import RequestStatus
import voluptuous as vol
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_HOST
from homeassistant.helpers import config_validation as cv
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
SCHEMA_DEVICE = vol.Schema(
{
vol.Required(CONF_HOST): cv.string,
}
)
class PooldoseConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Seko Pooldose."""
VERSION = 1
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle the initial step."""
if not user_input:
return self.async_show_form(
step_id="user",
data_schema=SCHEMA_DEVICE,
)
host = user_input[CONF_HOST]
client = PooldoseClient(host)
client_status = await client.connect()
if client_status == RequestStatus.HOST_UNREACHABLE:
return self.async_show_form(
step_id="user",
data_schema=SCHEMA_DEVICE,
errors={"base": "cannot_connect"},
)
if client_status == RequestStatus.PARAMS_FETCH_FAILED:
return self.async_show_form(
step_id="user",
data_schema=SCHEMA_DEVICE,
errors={"base": "params_fetch_failed"},
)
if client_status != RequestStatus.SUCCESS:
return self.async_show_form(
step_id="user",
data_schema=SCHEMA_DEVICE,
errors={"base": "cannot_connect"},
)
api_status, api_versions = client.check_apiversion_supported()
if api_status == RequestStatus.NO_DATA:
return self.async_show_form(
step_id="user",
data_schema=SCHEMA_DEVICE,
errors={"base": "api_not_set"},
)
if api_status == RequestStatus.API_VERSION_UNSUPPORTED:
return self.async_show_form(
step_id="user",
data_schema=SCHEMA_DEVICE,
errors={"base": "api_not_supported"},
description_placeholders=api_versions,
)
device_info = client.device_info
if not device_info:
return self.async_show_form(
step_id="user",
data_schema=SCHEMA_DEVICE,
errors={"base": "no_device_info"},
)
serial_number = device_info.get("SERIAL_NUMBER")
if not serial_number:
return self.async_show_form(
step_id="user",
data_schema=SCHEMA_DEVICE,
errors={"base": "no_serial_number"},
)
await self.async_set_unique_id(serial_number)
self._abort_if_unique_id_configured()
return self.async_create_entry(
title=f"PoolDose {serial_number}",
data={CONF_HOST: host},
)

6
homeassistant/components/pooldose/const.py Normal file
View File

@@ -0,0 +1,6 @@
"""Constants for the Seko Pooldose integration."""
from __future__ import annotations
DOMAIN = "pooldose"
MANUFACTURER = "SEKO"

68
homeassistant/components/pooldose/coordinator.py Normal file
View File

@@ -0,0 +1,68 @@
"""Data update coordinator for the PoolDose integration."""
from __future__ import annotations
from datetime import timedelta
import logging
from typing import Any
from pooldose.client import PooldoseClient
from pooldose.request_status import RequestStatus
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
_LOGGER = logging.getLogger(__name__)
type PooldoseConfigEntry = ConfigEntry[PooldoseCoordinator]
class PooldoseCoordinator(DataUpdateCoordinator[dict[str, Any]]):
"""Coordinator for PoolDose integration."""
device_info: dict[str, Any]
def __init__(
self,
hass: HomeAssistant,
client: PooldoseClient,
config_entry: PooldoseConfigEntry,
) -> None:
"""Initialize the coordinator."""
super().__init__(
hass,
_LOGGER,
name="Pooldose",
update_interval=timedelta(seconds=600), # Default update interval
config_entry=config_entry,
)
self.client = client
async def _async_setup(self) -> None:
"""Set up the coordinator."""
# Update device info after successful connection
self.device_info = self.client.device_info
_LOGGER.debug("Device info: %s", self.device_info)
async def _async_update_data(self) -> dict[str, Any]:
"""Fetch data from the PoolDose API."""
try:
status, instant_values = await self.client.instant_values_structured()
except TimeoutError as err:
raise UpdateFailed(
f"Timeout fetching data from PoolDose device: {err}"
) from err
except (ConnectionError, OSError) as err:
raise UpdateFailed(
f"Failed to connect to PoolDose device while fetching data: {err}"
) from err
if status != RequestStatus.SUCCESS:
raise UpdateFailed(f"API returned status: {status}")
if instant_values is None:
raise UpdateFailed("No data received from API")
_LOGGER.debug("Instant values structured: %s", instant_values)
return instant_values

78
homeassistant/components/pooldose/entity.py Normal file
View File

@@ -0,0 +1,78 @@
"""Base entity for Seko Pooldose integration."""
from __future__ import annotations
from typing import Any
from homeassistant.helpers.device_registry import CONNECTION_NETWORK_MAC, DeviceInfo
from homeassistant.helpers.entity import EntityDescription
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN, MANUFACTURER
from .coordinator import PooldoseCoordinator
def device_info(info: dict | None, unique_id: str) -> DeviceInfo:
"""Create device info for PoolDose devices."""
if info is None:
info = {}
api_version = info.get("API_VERSION", "").removesuffix("/")
return DeviceInfo(
identifiers={(DOMAIN, unique_id)},
manufacturer=MANUFACTURER,
model=info.get("MODEL") or None,
model_id=info.get("MODEL_ID") or None,
name=info.get("NAME") or None,
serial_number=unique_id,
sw_version=(
f"{info.get('FW_VERSION')} (SW v{info.get('SW_VERSION')}, API {api_version})"
if info.get("FW_VERSION") and info.get("SW_VERSION") and api_version
else None
),
hw_version=info.get("FW_CODE") or None,
connections=(
{(CONNECTION_NETWORK_MAC, str(info["MAC"]))} if info.get("MAC") else set()
),
configuration_url=(
f"http://{info['IP']}/index.html" if info.get("IP") else None
),
)
class PooldoseEntity(CoordinatorEntity[PooldoseCoordinator]):
"""Base class for all PoolDose entities."""
_attr_has_entity_name = True
def __init__(
self,
coordinator: PooldoseCoordinator,
serial_number: str,
device_properties: dict[str, Any],
entity_description: EntityDescription,
platform_name: str,
) -> None:
"""Initialize PoolDose entity."""
super().__init__(coordinator)
self.entity_description = entity_description
self.platform_name = platform_name
self._attr_unique_id = f"{serial_number}_{entity_description.key}"
self._attr_device_info = device_info(device_properties, serial_number)
@property
def available(self) -> bool:
"""Return True if the entity is available."""
if not super().available or self.coordinator.data is None:
return False
# Check if the entity type exists in coordinator data
platform_data = self.coordinator.data.get(self.platform_name, {})
return self.entity_description.key in platform_data
def get_data(self) -> dict | None:
"""Get data for this entity, only if available."""
if not self.available:
return None
platform_data = self.coordinator.data.get(self.platform_name, {})
return platform_data.get(self.entity_description.key)

45
homeassistant/components/pooldose/icons.json Normal file
View File

@@ -0,0 +1,45 @@
{
"entity": {
"sensor": {
"orp": {
"default": "mdi:water-check"
},
"ph_type_dosing": {
"default": "mdi:flask"
},
"peristaltic_ph_dosing": {
"default": "mdi:pump"
},
"ofa_ph_value": {
"default": "mdi:clock"
},
"orp_type_dosing": {
"default": "mdi:flask"
},
"peristaltic_orp_dosing": {
"default": "mdi:pump"
},
"ofa_orp_value": {
"default": "mdi:clock"
},
"ph_calibration_type": {
"default": "mdi:form-select"
},
"ph_calibration_offset": {
"default": "mdi:tune"
},
"ph_calibration_slope": {
"default": "mdi:slope-downhill"
},
"orp_calibration_type": {
"default": "mdi:form-select"
},
"orp_calibration_offset": {
"default": "mdi:tune"
},
"orp_calibration_slope": {
"default": "mdi:slope-downhill"
}
}
}
}

10
homeassistant/components/pooldose/manifest.json Normal file
View File

@@ -0,0 +1,10 @@
{
"domain": "pooldose",
"name": "SEKO PoolDose",
"codeowners": ["@lmaertin"],
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/pooldose",
"iot_class": "local_polling",
"quality_scale": "gold",
"requirements": ["python-pooldose==0.5.0"]
}

94
homeassistant/components/pooldose/quality_scale.yaml Normal file
View File

@@ -0,0 +1,94 @@
rules:
# Bronze
action-setup:
status: exempt
comment: This integration does not provide any actions.
appropriate-polling: done
brands: done
common-modules: done
config-flow-test-coverage: done
config-flow: done
dependency-transparency: done
docs-actions:
status: exempt
comment: This integration does not provide any actions.
docs-high-level-description: done
docs-installation-instructions: done
docs-removal-instructions: done
entity-event-setup:
status: exempt
comment: This integration does not subscribe to any events.
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:
status: exempt
comment: This integration does not provide any actions.
config-entry-unloading: done
docs-configuration-parameters: done
docs-installation-parameters: done
entity-unavailable: done
integration-owner: done
log-when-unavailable: done
parallel-updates:
status: exempt
comment: This integration uses a central coordinator to manage updates, which is not compatible with parallel updates.
reauthentication-flow:
status: exempt
comment: This integration does not need authentication for the local API.
test-coverage: done
# Gold
devices: done
diagnostics:
status: exempt
comment: This integration does not provide any diagnostic information, but can provide detailed logs if needed.
discovery-update-info:
status: exempt
comment: This integration does not support discovery features.
discovery:
status: exempt
comment: This integration does not support discovery updates since the PoolDose device does not support standard discovery methods.
docs-data-update: done
docs-examples:
status: exempt
comment: This integration does not provide any examples, as it is a simple integration that does not require complex configurations.
docs-known-limitations:
status: exempt
comment: This integration has known and documented limitations in frequency of data polling and stability of the connection to the device.
docs-supported-devices: done
docs-supported-functions: done
docs-troubleshooting: done
docs-use-cases:
status: exempt
comment: This integration does not provide use cases, as it is a simple integration that does not require complex configurations.
dynamic-devices:
status: exempt
comment: This integration does not support dynamic devices, as it is designed for a single PoolDose device.
entity-category: done
entity-device-class: done
entity-disabled-by-default: done
entity-translations: done
exception-translations: done
icon-translations: done
reconfiguration-flow:
status: exempt
comment: This integration does not support reconfiguration flows, as it is designed for a single PoolDose device with a fixed configuration.
repair-issues:
status: exempt
comment: This integration does not provide repair issues, as it is designed for a single PoolDose device with a fixed configuration.
stale-devices:
status: exempt
comment: This integration does not support stale devices, as it is designed for a single PoolDose device with a fixed configuration.
# Platinum
async-dependency: done
inject-websession: done
strict-typing:
status: exempt
comment: Dependency python-pooldose is not strictly typed and does not include a py.typed file.

186
homeassistant/components/pooldose/sensor.py Normal file
View File

@@ -0,0 +1,186 @@
"""Sensors for the Seko PoolDose integration."""
from __future__ import annotations
import logging
from typing import TYPE_CHECKING
from homeassistant.components.sensor import (
SensorDeviceClass,
SensorEntity,
SensorEntityDescription,
)
from homeassistant.const import EntityCategory, UnitOfElectricPotential, UnitOfTime
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from . import PooldoseConfigEntry
from .entity import PooldoseEntity
_LOGGER = logging.getLogger(__name__)
PLATFORM_NAME = "sensor"
SENSOR_DESCRIPTIONS: tuple[SensorEntityDescription, ...] = (
SensorEntityDescription(
key="temperature",
device_class=SensorDeviceClass.TEMPERATURE,
# Unit dynamically determined via API
),
SensorEntityDescription(key="ph", device_class=SensorDeviceClass.PH),
SensorEntityDescription(
key="orp",
translation_key="orp",
device_class=SensorDeviceClass.VOLTAGE,
native_unit_of_measurement=UnitOfElectricPotential.MILLIVOLT,
),
SensorEntityDescription(
key="ph_type_dosing",
translation_key="ph_type_dosing",
entity_category=EntityCategory.DIAGNOSTIC,
device_class=SensorDeviceClass.ENUM,
options=["alcalyne", "acid"],
),
SensorEntityDescription(
key="peristaltic_ph_dosing",
translation_key="peristaltic_ph_dosing",
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
device_class=SensorDeviceClass.ENUM,
options=["proportional", "on_off", "timed"],
),
SensorEntityDescription(
key="ofa_ph_value",
translation_key="ofa_ph_value",
entity_category=EntityCategory.DIAGNOSTIC,
device_class=SensorDeviceClass.DURATION,
entity_registry_enabled_default=False,
native_unit_of_measurement=UnitOfTime.MINUTES,
),
SensorEntityDescription(
key="orp_type_dosing",
translation_key="orp_type_dosing",
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
device_class=SensorDeviceClass.ENUM,
options=["low", "high"],
),
SensorEntityDescription(
key="peristaltic_orp_dosing",
translation_key="peristaltic_orp_dosing",
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
device_class=SensorDeviceClass.ENUM,
options=["off", "proportional", "on_off", "timed"],
),
SensorEntityDescription(
key="ofa_orp_value",
translation_key="ofa_orp_value",
device_class=SensorDeviceClass.DURATION,
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
native_unit_of_measurement=UnitOfTime.MINUTES,
),
SensorEntityDescription(
key="ph_calibration_type",
translation_key="ph_calibration_type",
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
device_class=SensorDeviceClass.ENUM,
options=["off", "reference", "1_point", "2_points"],
),
SensorEntityDescription(
key="ph_calibration_offset",
translation_key="ph_calibration_offset",
entity_category=EntityCategory.DIAGNOSTIC,
device_class=SensorDeviceClass.VOLTAGE,
suggested_display_precision=2,
entity_registry_enabled_default=False,
native_unit_of_measurement=UnitOfElectricPotential.MILLIVOLT,
),
SensorEntityDescription(
key="ph_calibration_slope",
translation_key="ph_calibration_slope",
entity_category=EntityCategory.DIAGNOSTIC,
device_class=SensorDeviceClass.VOLTAGE,
suggested_display_precision=2,
entity_registry_enabled_default=False,
native_unit_of_measurement=UnitOfElectricPotential.MILLIVOLT,
),
SensorEntityDescription(
key="orp_calibration_type",
translation_key="orp_calibration_type",
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
device_class=SensorDeviceClass.ENUM,
options=["off", "reference", "1_point"],
),
SensorEntityDescription(
key="orp_calibration_offset",
translation_key="orp_calibration_offset",
entity_category=EntityCategory.DIAGNOSTIC,
device_class=SensorDeviceClass.VOLTAGE,
suggested_display_precision=2,
entity_registry_enabled_default=False,
native_unit_of_measurement=UnitOfElectricPotential.MILLIVOLT,
),
SensorEntityDescription(
key="orp_calibration_slope",
translation_key="orp_calibration_slope",
entity_category=EntityCategory.DIAGNOSTIC,
device_class=SensorDeviceClass.VOLTAGE,
suggested_display_precision=2,
entity_registry_enabled_default=False,
native_unit_of_measurement=UnitOfElectricPotential.MILLIVOLT,
),
)
async def async_setup_entry(
hass: HomeAssistant,
config_entry: PooldoseConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up PoolDose sensor entities from a config entry."""
if TYPE_CHECKING:
assert config_entry.unique_id is not None
coordinator = config_entry.runtime_data
data = coordinator.data
serial_number = config_entry.unique_id
sensor_data = data.get(PLATFORM_NAME, {}) if data else {}
async_add_entities(
PooldoseSensor(
coordinator,
serial_number,
coordinator.device_info,
description,
PLATFORM_NAME,
)
for description in SENSOR_DESCRIPTIONS
if description.key in sensor_data
)
class PooldoseSensor(PooldoseEntity, SensorEntity):
"""Sensor entity for the Seko PoolDose Python API."""
@property
def native_value(self) -> float | int | str | None:
"""Return the current value of the sensor."""
data = self.get_data()
if isinstance(data, dict) and "value" in data:
return data["value"]
return None
@property
def native_unit_of_measurement(self) -> str | None:
"""Return the unit of measurement."""
if self.entity_description.key == "temperature":
data = self.get_data()
if isinstance(data, dict) and "unit" in data and data["unit"] is not None:
return data["unit"] # °C or °F
return super().native_unit_of_measurement

102
homeassistant/components/pooldose/strings.json Normal file
View File

@@ -0,0 +1,102 @@
{
"config": {
"step": {
"user": {
"title": "Set up SEKO PoolDose device",
"description": "Login handling not supported by API. Device password must be deactivated, i.e., set to default value (0000).",
"data": {
"host": "[%key:common::config_flow::data::host%]"
},
"data_description": {
"host": "IP address or hostname of your device"
}
}
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"api_not_set": "API version not found in device response. Device firmware may not be compatible with this integration.",
"api_not_supported": "Unsupported API version {api_version_is} (expected: {api_version_should}). Device firmware may not be compatible with this integration.",
"params_fetch_failed": "Unable to fetch core parameters from device. Device firmware may not be compatible with this integration.",
"no_device_info": "Unable to retrieve device information. Device may not be properly initialized or may be an unsupported model.",
"no_serial_number": "No serial number found on the device. Device may not be properly configured or may be an unsupported model.",
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
}
},
"entity": {
"sensor": {
"orp": {
"name": "ORP"
},
"ph_type_dosing": {
"name": "pH dosing type",
"state": {
"alcalyne": "pH+",
"acid": "pH-"
}
},
"peristaltic_ph_dosing": {
"name": "pH peristaltic dosing",
"state": {
"off": "[%key:common::state::off%]",
"proportional": "Proportional",
"on_off": "On/Off",
"timed": "Timed"
}
},
"ofa_ph_value": {
"name": "pH overfeed alert time"
},
"orp_type_dosing": {
"name": "ORP dosing type",
"state": {
"low": "[%key:common::state::low%]",
"high": "[%key:common::state::high%]"
}
},
"peristaltic_orp_dosing": {
"name": "ORP peristaltic dosing",
"state": {
"off": "[%key:common::state::off%]",
"proportional": "[%key:component::pooldose::entity::sensor::peristaltic_ph_dosing::state::proportional%]",
"on_off": "[%key:component::pooldose::entity::sensor::peristaltic_ph_dosing::state::on_off%]",
"timed": "[%key:component::pooldose::entity::sensor::peristaltic_ph_dosing::state::timed%]"
}
},
"ofa_orp_value": {
"name": "ORP overfeed alert time"
},
"ph_calibration_type": {
"name": "pH calibration type",
"state": {
"off": "[%key:common::state::off%]",
"reference": "Reference",
"1_point": "1 point",
"2_points": "2 points"
}
},
"ph_calibration_offset": {
"name": "pH calibration offset"
},
"ph_calibration_slope": {
"name": "pH calibration slope"
},
"orp_calibration_type": {
"name": "ORP calibration type",
"state": {
"off": "[%key:common::state::off%]",
"reference": "[%key:component::pooldose::entity::sensor::ph_calibration_type::state::reference%]",
"1_point": "[%key:component::pooldose::entity::sensor::ph_calibration_type::state::1_point%]"
}
},
"orp_calibration_offset": {
"name": "ORP calibration offset"
},
"orp_calibration_slope": {
"name": "ORP calibration slope"
}
}
}
}

1
homeassistant/generated/config_flows.py generated
View File

@@ -489,6 +489,7 @@ FLOWS = {
"plugwise",
"plum_lightpad",
"point",
"pooldose",
"poolsense",
"powerfox",
"powerwall",

6
homeassistant/generated/integrations.json
View File

@@ -5037,6 +5037,12 @@
"config_flow": true,
"iot_class": "cloud_polling"
},
"pooldose": {
"name": "SEKO PoolDose",
"integration_type": "hub",
"config_flow": true,
"iot_class": "local_polling"
},
"poolsense": {
"name": "PoolSense",
"integration_type": "hub",

3
requirements_all.txt generated
View File

@@ -2494,6 +2494,9 @@ python-overseerr==0.7.1
# homeassistant.components.picnic
python-picnic-api2==1.3.1
# homeassistant.components.pooldose
python-pooldose==0.5.0
# homeassistant.components.rabbitair
python-rabbitair==0.0.8

3
requirements_test_all.txt generated
View File

@@ -2067,6 +2067,9 @@ python-overseerr==0.7.1
# homeassistant.components.picnic
python-picnic-api2==1.3.1
# homeassistant.components.pooldose
python-pooldose==0.5.0
# homeassistant.components.rabbitair
python-rabbitair==0.0.8

1
tests/components/pooldose/__init__.py Normal file
View File

@@ -0,0 +1 @@
"""Tests for the Pooldose integration."""

85
tests/components/pooldose/conftest.py Normal file
View File

@@ -0,0 +1,85 @@
"""Test fixtures for the Seko PoolDose integration."""
from collections.abc import Generator
from typing import Any
from unittest.mock import AsyncMock, MagicMock, patch
from pooldose.request_status import RequestStatus
import pytest
from homeassistant.components.pooldose.const import DOMAIN
from homeassistant.const import CONF_HOST
from homeassistant.core import HomeAssistant
from tests.common import (
MockConfigEntry,
async_load_json_object_fixture,
load_json_object_fixture,
)
@pytest.fixture
def mock_setup_entry() -> Generator[AsyncMock]:
"""Override async_setup_entry."""
with patch(
"homeassistant.components.pooldose.async_setup_entry", return_value=True
) as mock_setup_entry:
yield mock_setup_entry
@pytest.fixture
async def device_info(hass: HomeAssistant) -> dict[str, Any]:
"""Return the device info from the fixture."""
return await async_load_json_object_fixture(hass, "deviceinfo.json", DOMAIN)
@pytest.fixture(autouse=True)
def mock_pooldose_client(device_info: dict[str, Any]) -> Generator[MagicMock]:
"""Mock a PooldoseClient for end-to-end testing."""
with (
patch(
"homeassistant.components.pooldose.config_flow.PooldoseClient",
autospec=True,
) as mock_client_class,
patch(
"homeassistant.components.pooldose.PooldoseClient", new=mock_client_class
),
):
client = mock_client_class.return_value
client.device_info = device_info
# Setup client methods with realistic responses
client.connect.return_value = RequestStatus.SUCCESS
client.check_apiversion_supported.return_value = (RequestStatus.SUCCESS, {})
# Load instant values from fixture
instant_values_data = load_json_object_fixture("instantvalues.json", DOMAIN)
client.instant_values_structured.return_value = (
RequestStatus.SUCCESS,
instant_values_data,
)
client.is_connected = True
yield client
@pytest.fixture
def mock_config_entry(device_info: dict[str, Any]) -> MockConfigEntry:
"""Return the default mocked config entry."""
return MockConfigEntry(
title="Pool Device",
domain=DOMAIN,
data={CONF_HOST: "192.168.1.100"},
unique_id=device_info["SERIAL_NUMBER"],
entry_id="01JG00V55WEVTJ0CJHM0GAD7PC",
)
async def init_integration(
hass: HomeAssistant, mock_config_entry: MockConfigEntry
) -> MockConfigEntry:
"""Set up the integration for testing."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
return mock_config_entry

15
tests/components/pooldose/fixtures/deviceinfo.json Normal file
View File

@@ -0,0 +1,15 @@
{
"NAME": "Pool Device",
"SERIAL_NUMBER": "TEST123456789",
"DEVICE_ID": "TEST123456789_DEVICE",
"MODEL": "POOL DOSE",
"MODEL_ID": "PDPR1H1HAW100",
"OWNERID": "GBL00001ENDUSERS",
"GROUPNAME": "Pool Device",
"FW_VERSION": "1.30",
"SW_VERSION": "2.10",
"API_VERSION": "v1/",
"FW_CODE": "539187",
"MAC": "AA:BB:CC:DD:EE:FF",
"IP": "192.168.1.100"
}

126
tests/components/pooldose/fixtures/instantvalues.json Normal file
View File

@@ -0,0 +1,126 @@
{
"sensor": {
"temperature": {
"value": 25,
"unit": "°C"
},
"ph": {
"value": 6.8,
"unit": null
},
"orp": {
"value": 718,
"unit": "mV"
},
"ph_type_dosing": {
"value": "alcalyne",
"unit": null
},
"peristaltic_ph_dosing": {
"value": "proportional",
"unit": null
},
"ofa_ph_value": {
"value": 0,
"unit": "min"
},
"orp_type_dosing": {
"value": "low",
"unit": null
},
"peristaltic_orp_dosing": {
"value": "proportional",
"unit": null
},
"ofa_orp_value": {
"value": 0,
"unit": "min"
},
"ph_calibration_type": {
"value": "2_points",
"unit": null
},
"ph_calibration_offset": {
"value": 8,
"unit": "mV"
},
"ph_calibration_slope": {
"value": 57.34,
"unit": "mV"
},
"orp_calibration_type": {
"value": "1_point",
"unit": null
},
"orp_calibration_offset": {
"value": 0,
"unit": "mV"
},
"orp_calibration_slope": {
"value": 0.96,
"unit": "mV"
}
},
"binary_sensor": {
"pump_running": {
"value": true
},
"ph_level_ok": {
"value": false
},
"orp_level_ok": {
"value": false
},
"flow_rate_ok": {
"value": false
},
"alarm_relay": {
"value": true
},
"relay_aux1_ph": {
"value": false
},
"relay_aux2_orpcl": {
"value": false
}
},
"number": {
"ph_target": {
"value": 6.5,
"unit": null,
"min": 6,
"max": 8,
"step": 0.1
},
"orp_target": {
"value": 680,
"unit": "mV",
"min": 400,
"max": 850,
"step": 1
},
"cl_target": {
"value": 1,
"unit": "ppm",
"min": 0,
"max": 65535,
"step": 0.01
}
},
"switch": {
"stop_pool_dosing": {
"value": false
},
"pump_detection": {
"value": true
},
"frequency_input": {
"value": false
}
},
"select": {
"water_meter_unit": {
"value": "m³"
}
}
}

36
tests/components/pooldose/snapshots/test_init.ambr Normal file
View File

@@ -0,0 +1,36 @@
# serializer version: 1
# name: test_devices
DeviceRegistryEntrySnapshot({
'area_id': None,
'config_entries': <ANY>,
'config_entries_subentries': <ANY>,
'configuration_url': 'http://192.168.1.100/index.html',
'connections': set({
tuple(
'mac',
'aa:bb:cc:dd:ee:ff',
),
}),
'disabled_by': None,
'entry_type': None,
'hw_version': '539187',
'id': <ANY>,
'identifiers': set({
tuple(
'pooldose',
'TEST123456789',
),
}),
'labels': set({
}),
'manufacturer': 'SEKO',
'model': 'POOL DOSE',
'model_id': 'PDPR1H1HAW100',
'name': 'Pool Device',
'name_by_user': None,
'primary_config_entry': <ANY>,
'serial_number': 'TEST123456789',
'sw_version': '1.30 (SW v2.10, API v1)',
'via_device_id': None,
})
# ---

834
tests/components/pooldose/snapshots/test_sensor.ambr Normal file
View File

@@ -0,0 +1,834 @@
# serializer version: 1
# name: test_all_sensors[sensor.pool_device_orp-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.pool_device_orp',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
'sensor': dict({
'suggested_display_precision': 0,
}),
}),
'original_device_class': <SensorDeviceClass.VOLTAGE: 'voltage'>,
'original_icon': None,
'original_name': 'ORP',
'platform': 'pooldose',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'orp',
'unique_id': 'TEST123456789_orp',
'unit_of_measurement': <UnitOfElectricPotential.MILLIVOLT: 'mV'>,
})
# ---
# name: test_all_sensors[sensor.pool_device_orp-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'voltage',
'friendly_name': 'Pool Device ORP',
'unit_of_measurement': <UnitOfElectricPotential.MILLIVOLT: 'mV'>,
}),
'context': <ANY>,
'entity_id': 'sensor.pool_device_orp',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '718',
})
# ---
# name: test_all_sensors[sensor.pool_device_orp_calibration_offset-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.pool_device_orp_calibration_offset',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
'sensor': dict({
'suggested_display_precision': 2,
}),
}),
'original_device_class': <SensorDeviceClass.VOLTAGE: 'voltage'>,
'original_icon': None,
'original_name': 'ORP calibration offset',
'platform': 'pooldose',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'orp_calibration_offset',
'unique_id': 'TEST123456789_orp_calibration_offset',
'unit_of_measurement': <UnitOfElectricPotential.MILLIVOLT: 'mV'>,
})
# ---
# name: test_all_sensors[sensor.pool_device_orp_calibration_offset-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'voltage',
'friendly_name': 'Pool Device ORP calibration offset',
'unit_of_measurement': <UnitOfElectricPotential.MILLIVOLT: 'mV'>,
}),
'context': <ANY>,
'entity_id': 'sensor.pool_device_orp_calibration_offset',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '0',
})
# ---
# name: test_all_sensors[sensor.pool_device_orp_calibration_slope-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.pool_device_orp_calibration_slope',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
'sensor': dict({
'suggested_display_precision': 2,
}),
}),
'original_device_class': <SensorDeviceClass.VOLTAGE: 'voltage'>,
'original_icon': None,
'original_name': 'ORP calibration slope',
'platform': 'pooldose',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'orp_calibration_slope',
'unique_id': 'TEST123456789_orp_calibration_slope',
'unit_of_measurement': <UnitOfElectricPotential.MILLIVOLT: 'mV'>,
})
# ---
# name: test_all_sensors[sensor.pool_device_orp_calibration_slope-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'voltage',
'friendly_name': 'Pool Device ORP calibration slope',
'unit_of_measurement': <UnitOfElectricPotential.MILLIVOLT: 'mV'>,
}),
'context': <ANY>,
'entity_id': 'sensor.pool_device_orp_calibration_slope',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '0.96',
})
# ---
# name: test_all_sensors[sensor.pool_device_orp_calibration_type-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'options': list([
'off',
'reference',
'1_point',
]),
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.pool_device_orp_calibration_type',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SensorDeviceClass.ENUM: 'enum'>,
'original_icon': None,
'original_name': 'ORP calibration type',
'platform': 'pooldose',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'orp_calibration_type',
'unique_id': 'TEST123456789_orp_calibration_type',
'unit_of_measurement': None,
})
# ---
# name: test_all_sensors[sensor.pool_device_orp_calibration_type-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'enum',
'friendly_name': 'Pool Device ORP calibration type',
'options': list([
'off',
'reference',
'1_point',
]),
}),
'context': <ANY>,
'entity_id': 'sensor.pool_device_orp_calibration_type',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '1_point',
})
# ---
# name: test_all_sensors[sensor.pool_device_orp_dosing_type-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'options': list([
'low',
'high',
]),
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.pool_device_orp_dosing_type',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SensorDeviceClass.ENUM: 'enum'>,
'original_icon': None,
'original_name': 'ORP dosing type',
'platform': 'pooldose',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'orp_type_dosing',
'unique_id': 'TEST123456789_orp_type_dosing',
'unit_of_measurement': None,
})
# ---
# name: test_all_sensors[sensor.pool_device_orp_dosing_type-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'enum',
'friendly_name': 'Pool Device ORP dosing type',
'options': list([
'low',
'high',
]),
}),
'context': <ANY>,
'entity_id': 'sensor.pool_device_orp_dosing_type',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'low',
})
# ---
# name: test_all_sensors[sensor.pool_device_orp_overfeed_alert_time-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.pool_device_orp_overfeed_alert_time',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
'sensor': dict({
'suggested_display_precision': 2,
}),
}),
'original_device_class': <SensorDeviceClass.DURATION: 'duration'>,
'original_icon': None,
'original_name': 'ORP overfeed alert time',
'platform': 'pooldose',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'ofa_orp_value',
'unique_id': 'TEST123456789_ofa_orp_value',
'unit_of_measurement': <UnitOfTime.MINUTES: 'min'>,
})
# ---
# name: test_all_sensors[sensor.pool_device_orp_overfeed_alert_time-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'duration',
'friendly_name': 'Pool Device ORP overfeed alert time',
'unit_of_measurement': <UnitOfTime.MINUTES: 'min'>,
}),
'context': <ANY>,
'entity_id': 'sensor.pool_device_orp_overfeed_alert_time',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '0',
})
# ---
# name: test_all_sensors[sensor.pool_device_orp_peristaltic_dosing-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'options': list([
'off',
'proportional',
'on_off',
'timed',
]),
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.pool_device_orp_peristaltic_dosing',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SensorDeviceClass.ENUM: 'enum'>,
'original_icon': None,
'original_name': 'ORP peristaltic dosing',
'platform': 'pooldose',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'peristaltic_orp_dosing',
'unique_id': 'TEST123456789_peristaltic_orp_dosing',
'unit_of_measurement': None,
})
# ---
# name: test_all_sensors[sensor.pool_device_orp_peristaltic_dosing-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'enum',
'friendly_name': 'Pool Device ORP peristaltic dosing',
'options': list([
'off',
'proportional',
'on_off',
'timed',
]),
}),
'context': <ANY>,
'entity_id': 'sensor.pool_device_orp_peristaltic_dosing',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'proportional',
})
# ---
# name: test_all_sensors[sensor.pool_device_ph-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.pool_device_ph',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SensorDeviceClass.PH: 'ph'>,
'original_icon': None,
'original_name': 'pH',
'platform': 'pooldose',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': 'TEST123456789_ph',
'unit_of_measurement': None,
})
# ---
# name: test_all_sensors[sensor.pool_device_ph-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'ph',
'friendly_name': 'Pool Device pH',
}),
'context': <ANY>,
'entity_id': 'sensor.pool_device_ph',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '6.8',
})
# ---
# name: test_all_sensors[sensor.pool_device_ph_calibration_offset-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.pool_device_ph_calibration_offset',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
'sensor': dict({
'suggested_display_precision': 2,
}),
}),
'original_device_class': <SensorDeviceClass.VOLTAGE: 'voltage'>,
'original_icon': None,
'original_name': 'pH calibration offset',
'platform': 'pooldose',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'ph_calibration_offset',
'unique_id': 'TEST123456789_ph_calibration_offset',
'unit_of_measurement': <UnitOfElectricPotential.MILLIVOLT: 'mV'>,
})
# ---
# name: test_all_sensors[sensor.pool_device_ph_calibration_offset-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'voltage',
'friendly_name': 'Pool Device pH calibration offset',
'unit_of_measurement': <UnitOfElectricPotential.MILLIVOLT: 'mV'>,
}),
'context': <ANY>,
'entity_id': 'sensor.pool_device_ph_calibration_offset',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '8',
})
# ---
# name: test_all_sensors[sensor.pool_device_ph_calibration_slope-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.pool_device_ph_calibration_slope',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
'sensor': dict({
'suggested_display_precision': 2,
}),
}),
'original_device_class': <SensorDeviceClass.VOLTAGE: 'voltage'>,
'original_icon': None,
'original_name': 'pH calibration slope',
'platform': 'pooldose',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'ph_calibration_slope',
'unique_id': 'TEST123456789_ph_calibration_slope',
'unit_of_measurement': <UnitOfElectricPotential.MILLIVOLT: 'mV'>,
})
# ---
# name: test_all_sensors[sensor.pool_device_ph_calibration_slope-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'voltage',
'friendly_name': 'Pool Device pH calibration slope',
'unit_of_measurement': <UnitOfElectricPotential.MILLIVOLT: 'mV'>,
}),
'context': <ANY>,
'entity_id': 'sensor.pool_device_ph_calibration_slope',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '57.34',
})
# ---
# name: test_all_sensors[sensor.pool_device_ph_calibration_type-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'options': list([
'off',
'reference',
'1_point',
'2_points',
]),
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.pool_device_ph_calibration_type',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SensorDeviceClass.ENUM: 'enum'>,
'original_icon': None,
'original_name': 'pH calibration type',
'platform': 'pooldose',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'ph_calibration_type',
'unique_id': 'TEST123456789_ph_calibration_type',
'unit_of_measurement': None,
})
# ---
# name: test_all_sensors[sensor.pool_device_ph_calibration_type-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'enum',
'friendly_name': 'Pool Device pH calibration type',
'options': list([
'off',
'reference',
'1_point',
'2_points',
]),
}),
'context': <ANY>,
'entity_id': 'sensor.pool_device_ph_calibration_type',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '2_points',
})
# ---
# name: test_all_sensors[sensor.pool_device_ph_dosing_type-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'options': list([
'alcalyne',
'acid',
]),
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.pool_device_ph_dosing_type',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SensorDeviceClass.ENUM: 'enum'>,
'original_icon': None,
'original_name': 'pH dosing type',
'platform': 'pooldose',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'ph_type_dosing',
'unique_id': 'TEST123456789_ph_type_dosing',
'unit_of_measurement': None,
})
# ---
# name: test_all_sensors[sensor.pool_device_ph_dosing_type-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'enum',
'friendly_name': 'Pool Device pH dosing type',
'options': list([
'alcalyne',
'acid',
]),
}),
'context': <ANY>,
'entity_id': 'sensor.pool_device_ph_dosing_type',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'alcalyne',
})
# ---
# name: test_all_sensors[sensor.pool_device_ph_overfeed_alert_time-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.pool_device_ph_overfeed_alert_time',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
'sensor': dict({
'suggested_display_precision': 2,
}),
}),
'original_device_class': <SensorDeviceClass.DURATION: 'duration'>,
'original_icon': None,
'original_name': 'pH overfeed alert time',
'platform': 'pooldose',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'ofa_ph_value',
'unique_id': 'TEST123456789_ofa_ph_value',
'unit_of_measurement': <UnitOfTime.MINUTES: 'min'>,
})
# ---
# name: test_all_sensors[sensor.pool_device_ph_overfeed_alert_time-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'duration',
'friendly_name': 'Pool Device pH overfeed alert time',
'unit_of_measurement': <UnitOfTime.MINUTES: 'min'>,
}),
'context': <ANY>,
'entity_id': 'sensor.pool_device_ph_overfeed_alert_time',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '0',
})
# ---
# name: test_all_sensors[sensor.pool_device_ph_peristaltic_dosing-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'options': list([
'proportional',
'on_off',
'timed',
]),
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.pool_device_ph_peristaltic_dosing',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SensorDeviceClass.ENUM: 'enum'>,
'original_icon': None,
'original_name': 'pH peristaltic dosing',
'platform': 'pooldose',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'peristaltic_ph_dosing',
'unique_id': 'TEST123456789_peristaltic_ph_dosing',
'unit_of_measurement': None,
})
# ---
# name: test_all_sensors[sensor.pool_device_ph_peristaltic_dosing-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'enum',
'friendly_name': 'Pool Device pH peristaltic dosing',
'options': list([
'proportional',
'on_off',
'timed',
]),
}),
'context': <ANY>,
'entity_id': 'sensor.pool_device_ph_peristaltic_dosing',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'proportional',
})
# ---
# name: test_all_sensors[sensor.pool_device_temperature-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.pool_device_temperature',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
'sensor': dict({
'suggested_display_precision': 1,
}),
}),
'original_device_class': <SensorDeviceClass.TEMPERATURE: 'temperature'>,
'original_icon': None,
'original_name': 'Temperature',
'platform': 'pooldose',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': 'TEST123456789_temperature',
'unit_of_measurement': <UnitOfTemperature.CELSIUS: '°C'>,
})
# ---
# name: test_all_sensors[sensor.pool_device_temperature-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'temperature',
'friendly_name': 'Pool Device Temperature',
'unit_of_measurement': <UnitOfTemperature.CELSIUS: '°C'>,
}),
'context': <ANY>,
'entity_id': 'sensor.pool_device_temperature',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '25',
})
# ---

239
tests/components/pooldose/test_config_flow.py Normal file
View File

@@ -0,0 +1,239 @@
"""Test the PoolDose config flow."""
from typing import Any
from unittest.mock import AsyncMock
import pytest
from homeassistant.components.pooldose.const import DOMAIN
from homeassistant.config_entries import SOURCE_USER
from homeassistant.const import CONF_HOST
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from .conftest import RequestStatus
from tests.common import MockConfigEntry
async def test_full_flow(hass: HomeAssistant, mock_setup_entry: AsyncMock) -> 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["step_id"] == "user"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_HOST: "192.168.1.100"}
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "PoolDose TEST123456789"
assert result["data"] == {CONF_HOST: "192.168.1.100"}
assert result["result"].unique_id == "TEST123456789"
async def test_device_unreachable(
hass: HomeAssistant, mock_pooldose_client: AsyncMock, mock_setup_entry: AsyncMock
) -> None:
"""Test that the form shows error when device is unreachable."""
mock_pooldose_client.is_connected = False
mock_pooldose_client.connect.return_value = RequestStatus.HOST_UNREACHABLE
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_HOST: "192.168.1.100"}
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": "cannot_connect"}
mock_pooldose_client.is_connected = True
mock_pooldose_client.connect.return_value = RequestStatus.SUCCESS
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_HOST: "192.168.1.100"}
)
assert result["type"] is FlowResultType.CREATE_ENTRY
async def test_api_version_unsupported(
hass: HomeAssistant, mock_pooldose_client: AsyncMock, mock_setup_entry: AsyncMock
) -> None:
"""Test that the form shows error when API version is unsupported."""
mock_pooldose_client.check_apiversion_supported.return_value = (
RequestStatus.API_VERSION_UNSUPPORTED,
{"api_version_is": "v0.9", "api_version_should": "v1.0"},
)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_HOST: "192.168.1.100"}
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": "api_not_supported"}
mock_pooldose_client.is_connected = True
mock_pooldose_client.check_apiversion_supported.return_value = (
RequestStatus.SUCCESS,
{},
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_HOST: "192.168.1.100"}
)
assert result["type"] is FlowResultType.CREATE_ENTRY
async def test_form_no_device_info(
hass: HomeAssistant,
mock_pooldose_client: AsyncMock,
mock_setup_entry: AsyncMock,
device_info: dict[str, Any],
) -> None:
"""Test that the form shows error when device_info is None."""
mock_pooldose_client.device_info = None
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_HOST: "192.168.1.100"}
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": "no_device_info"}
mock_pooldose_client.device_info = device_info
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_HOST: "192.168.1.100"}
)
assert result["type"] is FlowResultType.CREATE_ENTRY
@pytest.mark.parametrize(
("client_status", "expected_error"),
[
(RequestStatus.HOST_UNREACHABLE, "cannot_connect"),
(RequestStatus.PARAMS_FETCH_FAILED, "params_fetch_failed"),
(RequestStatus.UNKNOWN_ERROR, "cannot_connect"),
],
)
async def test_connection_errors(
hass: HomeAssistant,
mock_pooldose_client: AsyncMock,
mock_setup_entry: AsyncMock,
client_status: str,
expected_error: str,
) -> None:
"""Test that the form shows appropriate errors for various connection issues."""
mock_pooldose_client.connect.return_value = client_status
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_HOST: "192.168.1.100"}
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": expected_error}
mock_pooldose_client.connect.return_value = RequestStatus.SUCCESS
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_HOST: "192.168.1.100"}
)
assert result["type"] is FlowResultType.CREATE_ENTRY
async def test_api_no_data(
hass: HomeAssistant, mock_pooldose_client: AsyncMock, mock_setup_entry: AsyncMock
) -> None:
"""Test that the form shows error when API returns NO_DATA."""
mock_pooldose_client.check_apiversion_supported.return_value = (
RequestStatus.NO_DATA,
{},
)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_HOST: "192.168.1.100"}
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": "api_not_set"}
mock_pooldose_client.check_apiversion_supported.return_value = (
RequestStatus.SUCCESS,
{},
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_HOST: "192.168.1.100"}
)
assert result["type"] is FlowResultType.CREATE_ENTRY
async def test_form_no_serial_number(
hass: HomeAssistant,
mock_pooldose_client: AsyncMock,
mock_setup_entry: AsyncMock,
device_info: dict[str, Any],
) -> None:
"""Test that the form shows error when device_info has no serial number."""
mock_pooldose_client.device_info = {"NAME": "Pool Device", "MODEL": "POOL DOSE"}
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_HOST: "192.168.1.100"}
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": "no_serial_number"}
mock_pooldose_client.device_info = device_info
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_HOST: "192.168.1.100"}
)
assert result["type"] is FlowResultType.CREATE_ENTRY
async def test_duplicate_entry_aborts(
hass: HomeAssistant, mock_config_entry: MockConfigEntry
) -> None:
"""Test that the flow aborts if the device is already configured."""
mock_config_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_HOST: "192.168.1.100"}
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"

119
tests/components/pooldose/test_init.py Normal file
View File

@@ -0,0 +1,119 @@
"""Test the PoolDose integration initialization."""
from unittest.mock import AsyncMock
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.pooldose.const import DOMAIN
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr
from .conftest import RequestStatus
from tests.common import MockConfigEntry
async def test_devices(
hass: HomeAssistant,
snapshot: SnapshotAssertion,
mock_config_entry: MockConfigEntry,
device_registry: dr.DeviceRegistry,
) -> None:
"""Test all entities."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
device = device_registry.async_get_device({(DOMAIN, "TEST123456789")})
assert device is not None
assert device == snapshot
async def test_setup_entry_success(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test successful setup of config entry."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.LOADED
await hass.config_entries.async_unload(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.NOT_LOADED
async def test_setup_entry_coordinator_refresh_fails(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_pooldose_client: AsyncMock,
) -> None:
"""Test setup failure when coordinator first refresh fails."""
mock_config_entry.add_to_hass(hass)
mock_pooldose_client.instant_values_structured.side_effect = Exception(
"API communication failed"
)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.SETUP_RETRY
@pytest.mark.parametrize(
"status",
[
RequestStatus.HOST_UNREACHABLE,
RequestStatus.PARAMS_FETCH_FAILED,
RequestStatus.API_VERSION_UNSUPPORTED,
RequestStatus.NO_DATA,
RequestStatus.UNKNOWN_ERROR,
],
)
async def test_setup_entry_various_client_failures(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_pooldose_client: AsyncMock,
status: RequestStatus,
) -> None:
"""Test setup fails with various client error statuses."""
mock_pooldose_client.connect.return_value = RequestStatus.HOST_UNREACHABLE
mock_pooldose_client.is_connected = False
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.SETUP_RETRY
@pytest.mark.parametrize(
"exception",
[
TimeoutError("Connection timeout"),
OSError("Network error"),
],
)
async def test_setup_entry_timeout_error(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_pooldose_client: AsyncMock,
exception: Exception,
) -> None:
"""Test setup failure when client connection times out."""
mock_pooldose_client.connect.side_effect = exception
mock_pooldose_client.is_connected = False
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.SETUP_RETRY

256
tests/components/pooldose/test_sensor.py Normal file
View File

@@ -0,0 +1,256 @@
"""Test the PoolDose sensor platform."""
from datetime import timedelta
import json
from unittest.mock import AsyncMock, patch
from freezegun.api import FrozenDateTimeFactory
from pooldose.request_status import RequestStatus
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.pooldose.const import DOMAIN
from homeassistant.const import (
ATTR_UNIT_OF_MEASUREMENT,
STATE_UNAVAILABLE,
STATE_UNKNOWN,
Platform,
UnitOfTemperature,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from tests.common import (
MockConfigEntry,
async_fire_time_changed,
async_load_fixture,
snapshot_platform,
)
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
async def test_all_sensors(
hass: HomeAssistant,
mock_pooldose_client: AsyncMock,
mock_config_entry: MockConfigEntry,
entity_registry: er.EntityRegistry,
snapshot: SnapshotAssertion,
) -> None:
"""Test the Pooldose sensors."""
with patch("homeassistant.components.pooldose.PLATFORMS", [Platform.SENSOR]):
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
@pytest.mark.parametrize("exception", [TimeoutError, ConnectionError, OSError])
async def test_exception_raising(
hass: HomeAssistant,
mock_pooldose_client: AsyncMock,
mock_config_entry: MockConfigEntry,
exception: Exception,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test the Pooldose sensors."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert hass.states.get("sensor.pool_device_ph").state == "6.8"
mock_pooldose_client.instant_values_structured.side_effect = exception
freezer.tick(timedelta(minutes=10))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert hass.states.get("sensor.pool_device_ph").state == STATE_UNAVAILABLE
async def test_no_data(
hass: HomeAssistant,
mock_pooldose_client: AsyncMock,
mock_config_entry: MockConfigEntry,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test the Pooldose sensors."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert hass.states.get("sensor.pool_device_ph").state == "6.8"
mock_pooldose_client.instant_values_structured.return_value = (
RequestStatus.SUCCESS,
None,
)
freezer.tick(timedelta(minutes=10))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert hass.states.get("sensor.pool_device_ph").state == STATE_UNAVAILABLE
@pytest.mark.usefixtures("mock_pooldose_client")
async def test_ph_sensor_dynamic_unit(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_pooldose_client,
) -> None:
"""Test pH sensor unit behavior - pH should not have unit_of_measurement."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
# Mock pH data with custom unit (should be ignored for pH sensor)
instant_values_raw = await async_load_fixture(hass, "instantvalues.json", DOMAIN)
updated_data = json.loads(instant_values_raw)
updated_data["sensor"]["ph"]["unit"] = "pH units"
mock_pooldose_client.instant_values_structured.return_value = (
RequestStatus.SUCCESS,
updated_data,
)
# Trigger refresh by reloading the integration (blackbox approach)
await hass.config_entries.async_reload(mock_config_entry.entry_id)
await hass.async_block_till_done()
await hass.async_block_till_done()
# pH sensor should not have unit_of_measurement (device class pH)
ph_state = hass.states.get("sensor.pool_device_ph")
assert "unit_of_measurement" not in ph_state.attributes
async def test_sensor_entity_unavailable_no_coordinator_data(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_pooldose_client: AsyncMock,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test sensor entity becomes unavailable when coordinator has no data."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
# Verify initial working state
temp_state = hass.states.get("sensor.pool_device_temperature")
assert temp_state.state == "25"
# Set coordinator data to None by making API return empty
mock_pooldose_client.instant_values_structured.return_value = (
RequestStatus.HOST_UNREACHABLE,
None,
)
freezer.tick(timedelta(minutes=10))
async_fire_time_changed(hass)
await hass.async_block_till_done()
# Check sensor becomes unavailable
temp_state = hass.states.get("sensor.pool_device_temperature")
assert temp_state.state == STATE_UNAVAILABLE
async def test_sensor_entity_unavailable_missing_platform_data(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_pooldose_client: AsyncMock,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test sensor entity becomes unavailable when platform data is missing."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
# Verify initial working state
temp_state = hass.states.get("sensor.pool_device_temperature")
assert temp_state.state == "25"
# Remove sensor platform data by making API return data without sensors
mock_pooldose_client.instant_values_structured.return_value = (
RequestStatus.SUCCESS,
{"other_platform": {}}, # No sensor data
)
freezer.tick(timedelta(minutes=10))
async_fire_time_changed(hass)
await hass.async_block_till_done()
# Check sensor becomes unavailable
temp_state = hass.states.get("sensor.pool_device_temperature")
assert temp_state.state == STATE_UNAVAILABLE
@pytest.mark.usefixtures("mock_pooldose_client")
async def test_temperature_sensor_dynamic_unit(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_pooldose_client: AsyncMock,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test temperature sensor uses dynamic unit from API data."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
# Verify initial Celsius unit
temp_state = hass.states.get("sensor.pool_device_temperature")
assert temp_state.attributes[ATTR_UNIT_OF_MEASUREMENT] == UnitOfTemperature.CELSIUS
# Change to Fahrenheit via mock update
instant_values_raw = await async_load_fixture(hass, "instantvalues.json", DOMAIN)
updated_data = json.loads(instant_values_raw)
updated_data["sensor"]["temperature"]["unit"] = "°F"
updated_data["sensor"]["temperature"]["value"] = 77
mock_pooldose_client.instant_values_structured.return_value = (
RequestStatus.SUCCESS,
updated_data,
)
freezer.tick(timedelta(minutes=10))
async_fire_time_changed(hass)
await hass.async_block_till_done()
# Check unit changed to Fahrenheit
temp_state = hass.states.get("sensor.pool_device_temperature")
# After reload, the original fixture data is restored, so we expect °C
assert temp_state.attributes["unit_of_measurement"] == UnitOfTemperature.CELSIUS
assert temp_state.state == "25.0" # Original fixture value
async def test_native_value_with_non_dict_data(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_pooldose_client: AsyncMock,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test native_value returns None when data is not a dict."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
# Mock get_data to return non-dict value
instant_values_raw = await async_load_fixture(hass, "instantvalues.json", DOMAIN)
malformed_data = json.loads(instant_values_raw)
malformed_data["sensor"]["temperature"] = "not_a_dict"
mock_pooldose_client.instant_values_structured.return_value = (
RequestStatus.SUCCESS,
malformed_data,
)
freezer.tick(timedelta(minutes=10))
async_fire_time_changed(hass)
await hass.async_block_till_done()
# Should handle non-dict data gracefully
temp_state = hass.states.get("sensor.pool_device_temperature")
assert temp_state.state == STATE_UNKNOWN