1
0
mirror of https://github.com/home-assistant/core.git synced 2026-02-15 07:36:16 +00:00

Prana integration (#156599)

Co-authored-by: Josef Zweck <josef@zweck.dev>
This commit is contained in:
prana-dev-official
2026-01-28 18:22:19 +02:00
committed by GitHub
parent 8057de408e
commit bb3c977448
26 changed files with 1210 additions and 0 deletions

2
CODEOWNERS generated
View File

@@ -1265,6 +1265,8 @@ build.json @home-assistant/supervisor
/tests/components/powerfox/ @klaasnicolaas
/homeassistant/components/powerwall/ @bdraco @jrester @daniel-simpson
/tests/components/powerwall/ @bdraco @jrester @daniel-simpson
/homeassistant/components/prana/ @prana-dev-official
/tests/components/prana/ @prana-dev-official
/homeassistant/components/private_ble_device/ @Jc2k
/tests/components/private_ble_device/ @Jc2k
/homeassistant/components/probe_plus/ @pantherale0

View File

@@ -0,0 +1,35 @@
"""Home Assistant Prana integration entry point.
Sets up the update coordinator and forwards platform setups.
"""
from __future__ import annotations
import logging
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from .coordinator import PranaConfigEntry, PranaCoordinator
_LOGGER = logging.getLogger(__name__)
# Keep platforms sorted alphabetically to satisfy lint rule
PLATFORMS = [Platform.SWITCH]
async def async_setup_entry(hass: HomeAssistant, entry: PranaConfigEntry) -> bool:
"""Set up Prana from a config entry."""
coordinator = PranaCoordinator(hass, entry)
await coordinator.async_config_entry_first_refresh()
entry.runtime_data = coordinator
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True
async def async_unload_entry(hass: HomeAssistant, entry: PranaConfigEntry) -> bool:
"""Unload Prana integration platforms and coordinator."""
_LOGGER.info("Unloading Prana integration")
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)

View File

@@ -0,0 +1,102 @@
"""Configuration flow for Prana integration."""
import logging
from typing import Any
from prana_local_api_client.exceptions import PranaApiCommunicationError
from prana_local_api_client.models.prana_device_info import PranaDeviceInfo
from prana_local_api_client.prana_api_client import PranaLocalApiClient
import voluptuous as vol
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_HOST
from homeassistant.helpers.service_info.zeroconf import ZeroconfServiceInfo
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
USER_SCHEMA = vol.Schema(
{
vol.Required(CONF_HOST): str,
}
)
class PranaConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a Prana config flow."""
_host: str
_device_info: PranaDeviceInfo
async def async_step_zeroconf(
self, discovery_info: ZeroconfServiceInfo
) -> ConfigFlowResult:
"""Handle Zeroconf discovery of a Prana device."""
_LOGGER.debug("Discovered device via Zeroconf: %s", discovery_info)
friendly_name = discovery_info.properties.get("label", "")
self.context["title_placeholders"] = {"name": friendly_name}
self._host = discovery_info.host
try:
self._device_info = await self._validate_device()
except ValueError:
return self.async_abort(reason="invalid_device")
except PranaApiCommunicationError:
return self.async_abort(reason="invalid_device_or_unreachable")
self._set_confirm_only()
return await self.async_step_confirm()
async def async_step_confirm(self, user_input=None) -> ConfigFlowResult:
"""Handle the user confirming a discovered Prana device."""
if user_input is not None:
return self.async_create_entry(
title=self._device_info.label,
data={CONF_HOST: self._host},
)
return self.async_show_form(
step_id="confirm",
description_placeholders={
"name": self._device_info.label,
"host": self._host,
},
)
async def async_step_user(
self,
user_input: dict[str, Any] | None = None,
) -> ConfigFlowResult:
"""Manual entry by IP address."""
errors: dict[str, str] = {}
if user_input is not None:
self._host = user_input[CONF_HOST]
try:
self._device_info = await self._validate_device()
except ValueError:
return self.async_abort(reason="invalid_device")
except PranaApiCommunicationError:
errors = {"base": "invalid_device_or_unreachable"}
if not errors:
return self.async_create_entry(
title=self._device_info.label,
data={CONF_HOST: self._host},
)
return self.async_show_form(
step_id="user", data_schema=USER_SCHEMA, errors=errors
)
async def _validate_device(self) -> PranaDeviceInfo:
"""Validate that a Prana device is reachable and valid."""
client = PranaLocalApiClient(host=self._host, port=80)
device_info = await client.get_device_info()
if not device_info.isValid:
raise ValueError("invalid_device")
await self.async_set_unique_id(device_info.manufactureId)
self._abort_if_unique_id_configured()
return device_info

View File

@@ -0,0 +1,3 @@
"""Constants for Prana integration."""
DOMAIN = "prana"

View File

@@ -0,0 +1,67 @@
"""Coordinator for Prana integration.
Responsible for polling the device REST endpoints and normalizing data for entities.
"""
from datetime import timedelta
import logging
from prana_local_api_client.exceptions import (
PranaApiCommunicationError,
PranaApiUpdateFailed,
)
from prana_local_api_client.models.prana_device_info import PranaDeviceInfo
from prana_local_api_client.models.prana_state import PranaState
from prana_local_api_client.prana_api_client import PranaLocalApiClient
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_HOST
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
UPDATE_INTERVAL = timedelta(seconds=10)
COORDINATOR_NAME = f"{DOMAIN} coordinator"
type PranaConfigEntry = ConfigEntry[PranaCoordinator]
class PranaCoordinator(DataUpdateCoordinator[PranaState]):
"""Universal coordinator for Prana (fan, switch, sensor, light data)."""
config_entry: PranaConfigEntry
device_info: PranaDeviceInfo
def __init__(self, hass: HomeAssistant, entry: PranaConfigEntry) -> None:
"""Initialize the Prana data update coordinator."""
super().__init__(
hass,
logger=_LOGGER,
name=COORDINATOR_NAME,
update_interval=UPDATE_INTERVAL,
config_entry=entry,
)
self.api_client = PranaLocalApiClient(host=entry.data[CONF_HOST], port=80)
async def _async_setup(self) -> None:
try:
self.device_info = await self.api_client.get_device_info()
except PranaApiCommunicationError as err:
raise ConfigEntryNotReady("Could not fetch device info") from err
async def _async_update_data(self) -> PranaState:
"""Fetch and normalize device state for all platforms."""
try:
state = await self.api_client.get_state()
except PranaApiUpdateFailed as err:
raise UpdateFailed(f"HTTP error communicating with device: {err}") from err
except PranaApiCommunicationError as err:
raise UpdateFailed(
f"Network error communicating with device: {err}"
) from err
return state

View File

@@ -0,0 +1,50 @@
"""Defines base Prana entity."""
from dataclasses import dataclass
import logging
from typing import TYPE_CHECKING
from homeassistant.components.switch import StrEnum
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity import EntityDescription
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN
from .coordinator import PranaCoordinator
_LOGGER = logging.getLogger(__name__)
@dataclass(frozen=True, kw_only=True)
class PranaEntityDescription(EntityDescription):
"""Description for all Prana entities."""
key: StrEnum
class PranaBaseEntity(CoordinatorEntity[PranaCoordinator]):
"""Defines a base Prana entity."""
_attr_has_entity_name = True
_attr_entity_description: PranaEntityDescription
def __init__(
self,
coordinator: PranaCoordinator,
description: PranaEntityDescription,
) -> None:
"""Initialize the Prana entity."""
super().__init__(coordinator)
if TYPE_CHECKING:
assert coordinator.config_entry.unique_id
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, coordinator.config_entry.unique_id)},
manufacturer="Prana",
name=coordinator.device_info.label,
model=coordinator.device_info.pranaModel,
serial_number=coordinator.device_info.manufactureId,
sw_version=str(coordinator.device_info.fwVersion),
)
self._attr_unique_id = f"{coordinator.config_entry.unique_id}_{description.key}"
self.entity_description = description

View File

@@ -0,0 +1,21 @@
{
"entity": {
"switch": {
"auto": {
"default": "mdi:fan-auto"
},
"auto_plus": {
"default": "mdi:fan-auto"
},
"bound": {
"default": "mdi:link"
},
"heater": {
"default": "mdi:radiator"
},
"winter": {
"default": "mdi:snowflake"
}
}
}
}

View File

@@ -0,0 +1,16 @@
{
"domain": "prana",
"name": "Prana",
"codeowners": ["@prana-dev-official"],
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/prana",
"integration_type": "device",
"iot_class": "local_polling",
"quality_scale": "bronze",
"requirements": ["prana-api-client==0.10.0"],
"zeroconf": [
{
"type": "_prana._tcp.local."
}
]
}

View File

@@ -0,0 +1,76 @@
rules:
# Bronze (must be satisfied or exempt)
action-setup:
status: exempt
comment: Integration registers no custom services/actions
appropriate-polling: done # coordinator with 10s local interval (>5s min)
brands: done
common-modules: done
config-flow-test-coverage: done
config-flow: done
dependency-transparency: done
docs-actions:
status: exempt
comment: Integration has no custom services/actions
docs-high-level-description: done
docs-installation-instructions: done
docs-removal-instructions: done
entity-event-setup:
status: exempt
comment: Integration has no entity-specific event handling
entity-unique-id: done
has-entity-name: done
runtime-data: done # coordinator stored in ConfigEntry.runtime_data
test-before-configure: done # config flow validates discovery metadata
test-before-setup: done # coordinator refreshed prior to platform setup
unique-config-entry: done # unique_id based on device manufacturer id prevents duplicates
# Silver (future work)
action-exceptions: todo
config-entry-unloading: done # async_unload_entry implemented
docs-configuration-parameters:
status: exempt
comment: No configuration parameters
docs-installation-parameters: todo
entity-unavailable: done
integration-owner: done # codeowners present
log-when-unavailable: done # logged in coordinator
parallel-updates: done # PARALLEL_UPDATES defined in platforms
reauthentication-flow:
status: exempt
comment: Integration does not use authentication
test-coverage: todo
# Gold (future work)
devices: done
diagnostics: todo
discovery-update-info: todo
discovery: done # zeroconf implemented
docs-data-update: todo
docs-examples: todo
docs-known-limitations: todo
docs-supported-devices: todo
docs-supported-functions: todo
docs-troubleshooting: todo
docs-use-cases: todo
dynamic-devices:
status: exempt
comment: Device type integration
entity-category: todo
entity-device-class: done
entity-disabled-by-default: todo
entity-translations: done # strings + translation keys
exception-translations: todo
icon-translations: done
reconfiguration-flow: todo
repair-issues:
status: exempt
comment: There is no to repair
stale-devices:
status: exempt
comment: Device type integration
# Platinum (future work)
async-dependency: todo
inject-websession: todo
strict-typing: todo

View File

@@ -0,0 +1,46 @@
{
"config": {
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
"invalid_device": "The device is invalid",
"invalid_device_or_unreachable": "The device is invalid or unreachable"
},
"error": {
"invalid_device_or_unreachable": "The device is invalid or unreachable"
},
"step": {
"confirm": {
"description": "Set up {name} at {host}?",
"title": "Confirm Prana device"
},
"user": {
"data": {
"host": "[%key:common::config_flow::data::host%]"
},
"data_description": {
"host": "The IP address or hostname of your Prana device."
},
"title": "Add Prana device by IP"
}
}
},
"entity": {
"switch": {
"auto": {
"name": "Auto"
},
"auto_plus": {
"name": "Auto plus"
},
"bound": {
"name": "Bound"
},
"heater": {
"name": "Heater"
},
"winter": {
"name": "Winter"
}
}
}
}

View File

@@ -0,0 +1,100 @@
"""Switch platform for Prana integration."""
from collections.abc import Callable
from typing import Any
from aioesphomeapi import dataclass
from homeassistant.components.switch import (
StrEnum,
SwitchEntity,
SwitchEntityDescription,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from . import PranaConfigEntry, PranaCoordinator
from .entity import PranaBaseEntity, PranaEntityDescription
PARALLEL_UPDATES = 1
class PranaSwitchType(StrEnum):
"""Enumerates Prana switch types exposed by the device API."""
BOUND = "bound"
HEATER = "heater"
NIGHT = "night"
BOOST = "boost"
AUTO = "auto"
AUTO_PLUS = "auto_plus"
WINTER = "winter"
@dataclass(frozen=True, kw_only=True)
class PranaSwitchEntityDescription(SwitchEntityDescription, PranaEntityDescription):
"""Description of a Prana switch entity."""
value_fn: Callable[[PranaCoordinator], bool]
ENTITIES: tuple[PranaEntityDescription, ...] = (
PranaSwitchEntityDescription(
key=PranaSwitchType.BOUND,
translation_key="bound",
value_fn=lambda coord: coord.data.bound,
),
PranaSwitchEntityDescription(
key=PranaSwitchType.HEATER,
translation_key="heater",
value_fn=lambda coord: coord.data.heater,
),
PranaSwitchEntityDescription(
key=PranaSwitchType.AUTO,
translation_key="auto",
value_fn=lambda coord: coord.data.auto,
),
PranaSwitchEntityDescription(
key=PranaSwitchType.AUTO_PLUS,
translation_key="auto_plus",
value_fn=lambda coord: coord.data.auto_plus,
),
PranaSwitchEntityDescription(
key=PranaSwitchType.WINTER,
translation_key="winter",
value_fn=lambda coord: coord.data.winter,
),
)
async def async_setup_entry(
hass: HomeAssistant,
entry: PranaConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up Prana switch entities from a config entry."""
async_add_entities(
PranaSwitch(entry.runtime_data, entity_description)
for entity_description in ENTITIES
)
class PranaSwitch(PranaBaseEntity, SwitchEntity):
"""Representation of a Prana switch (bound/heater/auto/etc)."""
entity_description: PranaSwitchEntityDescription
@property
def is_on(self) -> bool:
"""Return switch on/off state."""
return self.entity_description.value_fn(self.coordinator)
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn the switch on."""
await self.coordinator.api_client.set_switch(self.entity_description.key, True)
await self.coordinator.async_refresh()
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn the switch off."""
await self.coordinator.api_client.set_switch(self.entity_description.key, False)
await self.coordinator.async_refresh()

View File

@@ -533,6 +533,7 @@ FLOWS = {
"portainer",
"powerfox",
"powerwall",
"prana",
"private_ble_device",
"probe_plus",
"profiler",

View File

@@ -5210,6 +5210,12 @@
"config_flow": true,
"iot_class": "cloud_polling"
},
"prana": {
"name": "Prana",
"integration_type": "device",
"config_flow": true,
"iot_class": "local_polling"
},
"private_ble_device": {
"name": "Private BLE Device",
"integration_type": "device",

View File

@@ -832,6 +832,11 @@ ZEROCONF = {
"domain": "hunterdouglas_powerview",
},
],
"_prana._tcp.local.": [
{
"domain": "prana",
},
],
"_printer._tcp.local.": [
{
"domain": "brother",

3
requirements_all.txt generated
View File

@@ -1765,6 +1765,9 @@ poolsense==0.0.8
# homeassistant.components.powerfox
powerfox==2.0.0
# homeassistant.components.prana
prana-api-client==0.10.0
# homeassistant.components.reddit
praw==7.5.0

View File

@@ -1517,6 +1517,9 @@ poolsense==0.0.8
# homeassistant.components.powerfox
powerfox==2.0.0
# homeassistant.components.prana
prana-api-client==0.10.0
# homeassistant.components.reddit
praw==7.5.0

View File

@@ -0,0 +1,14 @@
"""Mock inputs for Prana tests."""
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
async def async_init_integration(
hass: HomeAssistant, mock_config_entry: MockConfigEntry
) -> None:
"""Set up the Prana 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()

View File

@@ -0,0 +1,55 @@
"""Common fixtures for the Prana tests."""
from collections.abc import Generator
from types import SimpleNamespace
from unittest.mock import AsyncMock, patch
import pytest
from homeassistant.components.prana.const import DOMAIN
from homeassistant.const import CONF_HOST
from tests.common import MockConfigEntry, load_json_object_fixture
@pytest.fixture
def mock_config_entry() -> MockConfigEntry:
"""Create Prana mock config entry."""
device_info_data = load_json_object_fixture("device_info.json", DOMAIN)
device_info_obj = SimpleNamespace(**device_info_data)
return MockConfigEntry(
domain=DOMAIN,
unique_id=device_info_obj.manufactureId,
entry_id="0123456789abcdef0123456789abcdef",
title=device_info_obj.label,
data={
CONF_HOST: "127.0.0.1",
},
)
@pytest.fixture
def mock_prana_api() -> Generator[AsyncMock]:
"""Mock the Prana API client used by the integration."""
with (
patch(
"homeassistant.components.prana.config_flow.PranaLocalApiClient",
autospec=True,
) as mock_api_class,
patch(
"homeassistant.components.prana.coordinator.PranaLocalApiClient",
mock_api_class,
),
):
device_info_data = load_json_object_fixture("device_info.json", DOMAIN)
state_data = load_json_object_fixture("state.json", DOMAIN)
device_info_obj = SimpleNamespace(**device_info_data)
state_obj = SimpleNamespace(**state_data)
mock_api_class.return_value.get_device_info = AsyncMock(
return_value=device_info_obj
)
mock_api_class.return_value.get_state = AsyncMock(return_value=state_obj)
yield mock_api_class.return_value

View File

@@ -0,0 +1,7 @@
{
"manufactureId": "ECC9FFE0E574",
"isValid": true,
"fwVersion": 46,
"pranaModel": "PRANA RECUPERATOR 150",
"label": "PRANA RECUPERATOR"
}

View File

@@ -0,0 +1,7 @@
{
"manufactureId": "ECC9FFE0E574",
"isValid": false,
"fwVersion": 46,
"pranaModel": "PRANA RECUPERATOR 150",
"label": "PRANA RECUPERATOR"
}

View File

@@ -0,0 +1,25 @@
{
"extract": {
"speed": 1,
"is_on": true,
"max_speed": 100
},
"supply": {
"speed": 1,
"is_on": true,
"max_speed": 100
},
"bounded": {
"speed": 1,
"is_on": true,
"max_speed": 100
},
"bound": true,
"heater": false,
"auto": false,
"auto_plus": false,
"winter": false,
"inside_temperature": 217,
"humidity": 56,
"brightness": 6
}

View File

@@ -0,0 +1,32 @@
# serializer version: 1
# name: test_device_info_registered
DeviceRegistryEntrySnapshot({
'area_id': None,
'config_entries': <ANY>,
'config_entries_subentries': <ANY>,
'configuration_url': None,
'connections': set({
}),
'disabled_by': None,
'entry_type': None,
'hw_version': None,
'id': <ANY>,
'identifiers': set({
tuple(
'prana',
'ECC9FFE0E574',
),
}),
'labels': set({
}),
'manufacturer': 'Prana',
'model': 'PRANA RECUPERATOR 150',
'model_id': None,
'name': 'PRANA RECUPERATOR',
'name_by_user': None,
'primary_config_entry': <ANY>,
'serial_number': 'ECC9FFE0E574',
'sw_version': '46',
'via_device_id': None,
})
# ---

View File

@@ -0,0 +1,246 @@
# serializer version: 1
# name: test_switches[switch.prana_recuperator_auto-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': 'switch',
'entity_category': None,
'entity_id': 'switch.prana_recuperator_auto',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Auto',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Auto',
'platform': 'prana',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'auto',
'unique_id': 'ECC9FFE0E574_auto',
'unit_of_measurement': None,
})
# ---
# name: test_switches[switch.prana_recuperator_auto-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'PRANA RECUPERATOR Auto',
}),
'context': <ANY>,
'entity_id': 'switch.prana_recuperator_auto',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'off',
})
# ---
# name: test_switches[switch.prana_recuperator_auto_plus-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': 'switch',
'entity_category': None,
'entity_id': 'switch.prana_recuperator_auto_plus',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Auto plus',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Auto plus',
'platform': 'prana',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'auto_plus',
'unique_id': 'ECC9FFE0E574_auto_plus',
'unit_of_measurement': None,
})
# ---
# name: test_switches[switch.prana_recuperator_auto_plus-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'PRANA RECUPERATOR Auto plus',
}),
'context': <ANY>,
'entity_id': 'switch.prana_recuperator_auto_plus',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'off',
})
# ---
# name: test_switches[switch.prana_recuperator_bound-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': 'switch',
'entity_category': None,
'entity_id': 'switch.prana_recuperator_bound',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Bound',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Bound',
'platform': 'prana',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'bound',
'unique_id': 'ECC9FFE0E574_bound',
'unit_of_measurement': None,
})
# ---
# name: test_switches[switch.prana_recuperator_bound-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'PRANA RECUPERATOR Bound',
}),
'context': <ANY>,
'entity_id': 'switch.prana_recuperator_bound',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'on',
})
# ---
# name: test_switches[switch.prana_recuperator_heater-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': 'switch',
'entity_category': None,
'entity_id': 'switch.prana_recuperator_heater',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Heater',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Heater',
'platform': 'prana',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'heater',
'unique_id': 'ECC9FFE0E574_heater',
'unit_of_measurement': None,
})
# ---
# name: test_switches[switch.prana_recuperator_heater-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'PRANA RECUPERATOR Heater',
}),
'context': <ANY>,
'entity_id': 'switch.prana_recuperator_heater',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'off',
})
# ---
# name: test_switches[switch.prana_recuperator_winter-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': 'switch',
'entity_category': None,
'entity_id': 'switch.prana_recuperator_winter',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Winter',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Winter',
'platform': 'prana',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'winter',
'unique_id': 'ECC9FFE0E574_winter',
'unit_of_measurement': None,
})
# ---
# name: test_switches[switch.prana_recuperator_winter-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'PRANA RECUPERATOR Winter',
}),
'context': <ANY>,
'entity_id': 'switch.prana_recuperator_winter',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'off',
})
# ---

View File

@@ -0,0 +1,166 @@
"""Tests for the Prana config flow."""
from types import SimpleNamespace
from prana_local_api_client.exceptions import (
PranaApiCommunicationError as PranaCommunicationError,
)
from homeassistant.components.prana.const import DOMAIN
from homeassistant.config_entries import SOURCE_USER, SOURCE_ZEROCONF
from homeassistant.const import CONF_HOST
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from homeassistant.helpers.service_info.zeroconf import ZeroconfServiceInfo
from tests.common import load_json_object_fixture
ZEROCONF_INFO = ZeroconfServiceInfo(
ip_address="192.168.1.30",
ip_addresses=["192.168.1.30"],
hostname="prana.local",
name="TestNew._prana._tcp.local.",
type="_prana._tcp.local.",
port=1234,
properties={},
)
async def async_load_fixture(hass: HomeAssistant, filename: str) -> dict:
"""Load a fixture file."""
return await hass.async_add_executor_job(load_json_object_fixture, filename, DOMAIN)
async def test_zeroconf_new_device_and_confirm(
hass: HomeAssistant, mock_prana_api
) -> None:
"""Zeroconf discovery shows confirm form and creates a config entry."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_ZEROCONF}, data=ZEROCONF_INFO
)
device_info = await async_load_fixture(hass, "device_info.json")
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "confirm"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={}
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == device_info["label"]
assert result["result"].unique_id == device_info["manufactureId"]
assert result["result"].data == {CONF_HOST: "192.168.1.30"}
async def test_user_flow_with_manual_entry(hass: HomeAssistant, mock_prana_api) -> None:
"""User flow accepts manual host and creates entry after confirmation."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
device_info = await async_load_fixture(hass, "device_info.json")
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={CONF_HOST: "192.168.1.40"}
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == device_info["label"]
assert result["result"].unique_id == device_info["manufactureId"]
assert result["result"].data == {CONF_HOST: "192.168.1.40"}
async def test_communication_error_on_device_info(
hass: HomeAssistant, mock_prana_api
) -> None:
"""Communication errors when fetching device info surface as form errors."""
# Setting an invalid device info, for abort the flow
device_info_invalid = await async_load_fixture(hass, "device_info_invalid.json")
mock_prana_api.get_device_info.return_value = SimpleNamespace(**device_info_invalid)
mock_prana_api.get_device_info.side_effect = None
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={CONF_HOST: "192.168.1.50"}
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "invalid_device"
# Simulating a communication error
device_info = await async_load_fixture(hass, "device_info.json")
mock_prana_api.get_device_info.return_value = SimpleNamespace(**device_info)
mock_prana_api.get_device_info.side_effect = PranaCommunicationError(
"Network error"
)
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"], user_input={CONF_HOST: "192.168.1.50"}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
assert "invalid_device_or_unreachable" in result["errors"].values()
# Now simulating a successful fetch, without aborting
mock_prana_api.get_device_info.side_effect = None
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={CONF_HOST: "192.168.1.50"}
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == device_info["label"]
assert result["result"].unique_id == device_info["manufactureId"]
assert result["result"].data == {CONF_HOST: "192.168.1.50"}
async def test_user_flow_already_configured(
hass: HomeAssistant, mock_prana_api, mock_config_entry
) -> None:
"""Second configuration for the same device should be aborted."""
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"], user_input={CONF_HOST: "192.168.1.40"}
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
async def test_zeroconf_already_configured(
hass: HomeAssistant, mock_prana_api, mock_config_entry
) -> None:
"""Zeroconf discovery of an already configured device should be aborted."""
mock_config_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_ZEROCONF}, data=ZEROCONF_INFO
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
async def test_zeroconf_invalid_device(hass: HomeAssistant, mock_prana_api) -> None:
"""Zeroconf discovery of an invalid device should be aborted."""
mock_prana_api.get_device_info.side_effect = PranaCommunicationError(
"Network error"
)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_ZEROCONF}, data=ZEROCONF_INFO
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "invalid_device_or_unreachable"

View File

@@ -0,0 +1,43 @@
"""Tests for Prana integration entry points (async_setup_entry / async_unload_entry)."""
from homeassistant.components.prana.const import DOMAIN
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr
from . import async_init_integration
from tests.common import SnapshotAssertion
async def test_async_setup_entry_and_unload_entry(
hass: HomeAssistant, mock_config_entry, mock_prana_api
) -> None:
"""async_setup_entry should create coordinator, refresh it, store runtime_data and forward setups."""
await async_init_integration(hass, mock_config_entry)
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_device_info_registered(
hass: HomeAssistant,
mock_config_entry,
mock_prana_api,
device_registry: dr.DeviceRegistry,
snapshot: SnapshotAssertion,
) -> None:
"""Device info from the API should be registered on the device registry."""
await async_init_integration(hass, mock_config_entry)
device = device_registry.async_get_device(
identifiers={(DOMAIN, mock_config_entry.unique_id)}
)
assert device is not None
assert snapshot == device

View File

@@ -0,0 +1,79 @@
"""Integration-style tests for Prana switches."""
from unittest.mock import MagicMock, patch
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.switch import (
DOMAIN as SWITCH_DOMAIN,
SERVICE_TURN_OFF,
SERVICE_TURN_ON,
)
from homeassistant.const import ATTR_ENTITY_ID, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from . import async_init_integration
from tests.common import MockConfigEntry, snapshot_platform
async def test_switches(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
entity_registry: er.EntityRegistry,
mock_prana_api: MagicMock,
snapshot: SnapshotAssertion,
) -> None:
"""Test the Prana switches snapshot."""
with patch("homeassistant.components.prana.PLATFORMS", [Platform.SWITCH]):
await async_init_integration(hass, mock_config_entry)
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
@pytest.mark.parametrize(
("type_key", "entity_suffix"),
[
("winter", "_winter"),
("heater", "_heater"),
("auto", "_auto"),
("bound", "_bound"),
("auto_plus", "_auto_plus"),
],
)
async def test_switches_actions(
hass: HomeAssistant,
mock_prana_api: MagicMock,
mock_config_entry: MockConfigEntry,
entity_registry: er.EntityRegistry,
type_key: str,
entity_suffix: str,
) -> None:
"""Test turning switches on and off calls the API through the coordinator."""
await async_init_integration(hass, mock_config_entry)
entries = er.async_entries_for_config_entry(
entity_registry, mock_config_entry.entry_id
)
assert entries
target = f"switch.prana_recuperator{entity_suffix}"
await hass.services.async_call(
SWITCH_DOMAIN,
SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: target},
blocking=True,
)
mock_prana_api.set_switch.assert_called()
await hass.services.async_call(
SWITCH_DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: target},
blocking=True,
)
assert mock_prana_api.set_switch.call_count >= 2