mirror of
https://github.com/home-assistant/core.git
synced 2026-08-24 10:13:52 -05:00
Allow SAJ Solar Inverter to be configured through the UI (#160052)
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
This commit is contained in:
Eduard Reñé Claramunt
co-authored by
Copilot
Copilot
parent
7e684dbcca
commit
7e805f2f2a
Generated
+1
@@ -1559,6 +1559,7 @@ CLAUDE.md @home-assistant/core
|
||||
/homeassistant/components/sabnzbd/ @shaiu @jpbede
|
||||
/tests/components/sabnzbd/ @shaiu @jpbede
|
||||
/homeassistant/components/saj/ @fredericvl
|
||||
/tests/components/saj/ @fredericvl
|
||||
/homeassistant/components/samsung_infrared/ @lmaertin
|
||||
/tests/components/samsung_infrared/ @lmaertin
|
||||
/homeassistant/components/samsungtv/ @chemelli74
|
||||
|
||||
@@ -1 +1,240 @@
|
||||
"""The saj component."""
|
||||
|
||||
from collections.abc import Callable, Coroutine
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
import pysaj
|
||||
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import (
|
||||
CONF_HOST,
|
||||
CONF_PASSWORD,
|
||||
CONF_TYPE,
|
||||
CONF_USERNAME,
|
||||
EVENT_HOMEASSISTANT_STOP,
|
||||
Platform,
|
||||
)
|
||||
from homeassistant.core import CALLBACK_TYPE, Event, HomeAssistant, callback
|
||||
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
|
||||
from homeassistant.helpers.event import async_call_later
|
||||
from homeassistant.helpers.start import async_at_start
|
||||
|
||||
from .const import CONNECTION_TYPES
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
PLATFORMS: list[Platform] = [Platform.SENSOR]
|
||||
|
||||
MIN_INTERVAL_SEC = 5
|
||||
MAX_INTERVAL_SEC = 300
|
||||
|
||||
|
||||
@callback
|
||||
def async_track_time_interval_backoff(
|
||||
hass: HomeAssistant, action: Callable[[], Coroutine[Any, Any, bool]]
|
||||
) -> CALLBACK_TYPE:
|
||||
"""Fire `action` on an interval; double the interval (capped) when it returns False."""
|
||||
remove: CALLBACK_TYPE | None = None
|
||||
interval = MIN_INTERVAL_SEC
|
||||
stopped = False
|
||||
|
||||
async def interval_listener(_now: datetime | None = None) -> None:
|
||||
nonlocal interval, remove, stopped
|
||||
try:
|
||||
if await action():
|
||||
interval = MIN_INTERVAL_SEC
|
||||
else:
|
||||
interval = min(interval * 2, MAX_INTERVAL_SEC)
|
||||
finally:
|
||||
if not stopped:
|
||||
remove = async_call_later(hass, interval, interval_listener)
|
||||
|
||||
hass.async_create_task(interval_listener())
|
||||
|
||||
def remove_listener() -> None:
|
||||
nonlocal remove, stopped
|
||||
stopped = True
|
||||
if remove:
|
||||
remove()
|
||||
remove = None
|
||||
|
||||
return remove_listener
|
||||
|
||||
|
||||
class SAJPolling:
|
||||
"""Interval polling with backoff; entities register for per-poll callbacks."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
hass: HomeAssistant,
|
||||
entry: ConfigEntry,
|
||||
saj: pysaj.SAJ,
|
||||
sensor_def: pysaj.Sensors,
|
||||
) -> None:
|
||||
"""Initialize polling for one config entry."""
|
||||
self._hass = hass
|
||||
self._entry = entry
|
||||
self._saj = saj
|
||||
self._sensor_def = sensor_def
|
||||
self._listeners: list[Callable[[bool], None]] = []
|
||||
self._remove_backoff: CALLBACK_TYPE | None = None
|
||||
self._cancel_at_start: CALLBACK_TYPE | None = None
|
||||
self._unsub_stop: CALLBACK_TYPE | None = None
|
||||
|
||||
@callback
|
||||
def async_add_poll_listener(
|
||||
self, target: Callable[[bool], None]
|
||||
) -> Callable[[], None]:
|
||||
"""Register to be called after each poll with the read success flag."""
|
||||
|
||||
@callback
|
||||
def remove_listener() -> None:
|
||||
self._listeners.remove(target)
|
||||
if not self._listeners:
|
||||
self._async_stop_backoff()
|
||||
if self._cancel_at_start:
|
||||
self._cancel_at_start()
|
||||
self._cancel_at_start = None
|
||||
|
||||
self._listeners.append(target)
|
||||
if len(self._listeners) == 1:
|
||||
self._schedule_polling_start()
|
||||
return remove_listener
|
||||
|
||||
def _schedule_polling_start(self) -> None:
|
||||
@callback
|
||||
def start(_hass: HomeAssistant) -> None:
|
||||
self._cancel_at_start = None
|
||||
if not self._listeners:
|
||||
return
|
||||
self._async_start_backoff()
|
||||
|
||||
self._cancel_at_start = async_at_start(self._hass, start)
|
||||
|
||||
@callback
|
||||
def _async_start_backoff(self) -> None:
|
||||
self._remove_backoff = async_track_time_interval_backoff(
|
||||
self._hass, self._async_poll_with_notify
|
||||
)
|
||||
|
||||
@callback
|
||||
def stop_on_hass_stop(_event: Event) -> None:
|
||||
self._async_stop_backoff()
|
||||
|
||||
self._unsub_stop = self._hass.bus.async_listen(
|
||||
EVENT_HOMEASSISTANT_STOP, stop_on_hass_stop
|
||||
)
|
||||
|
||||
async def _async_poll_with_notify(self) -> bool:
|
||||
success = False
|
||||
try:
|
||||
success = await self._saj.read(self._sensor_def)
|
||||
except pysaj.UnauthorizedException:
|
||||
_LOGGER.error(
|
||||
"Username and/or password rejected during polling for %s",
|
||||
self._entry.title,
|
||||
)
|
||||
except pysaj.UnexpectedResponseException as err:
|
||||
_LOGGER.error(
|
||||
"Error in SAJ, please check host/ip address. Original error: %s", err
|
||||
)
|
||||
except (TimeoutError, OSError) as err:
|
||||
_LOGGER.error("Error communicating with SAJ: %s", err)
|
||||
except Exception as err: # noqa: BLE001
|
||||
_LOGGER.error(
|
||||
"Unexpected error polling SAJ inverter %s: %s",
|
||||
self._entry.title,
|
||||
err,
|
||||
)
|
||||
|
||||
for listener in list(self._listeners):
|
||||
listener(success)
|
||||
return success
|
||||
|
||||
@callback
|
||||
def _async_stop_backoff(self) -> None:
|
||||
if self._remove_backoff:
|
||||
self._remove_backoff()
|
||||
self._remove_backoff = None
|
||||
if self._unsub_stop:
|
||||
self._unsub_stop()
|
||||
self._unsub_stop = None
|
||||
|
||||
@callback
|
||||
def async_shutdown(self) -> None:
|
||||
"""Cancel polling and any deferred start."""
|
||||
self._listeners.clear()
|
||||
self._async_stop_backoff()
|
||||
if self._cancel_at_start:
|
||||
self._cancel_at_start()
|
||||
self._cancel_at_start = None
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class SAJRuntimeData:
|
||||
"""Runtime data attached to a SAJ config entry."""
|
||||
|
||||
saj: pysaj.SAJ
|
||||
sensor_def: pysaj.Sensors
|
||||
polling: SAJPolling
|
||||
|
||||
|
||||
type SAJConfigEntry = ConfigEntry[SAJRuntimeData]
|
||||
|
||||
|
||||
async def async_setup_entry(hass: HomeAssistant, entry: SAJConfigEntry) -> bool:
|
||||
"""Set up SAJ from a config entry."""
|
||||
host = entry.data[CONF_HOST]
|
||||
connection_type = entry.data[CONF_TYPE]
|
||||
username = entry.data.get(CONF_USERNAME, None)
|
||||
password = entry.data.get(CONF_PASSWORD, None)
|
||||
|
||||
# Create SAJ connection
|
||||
kwargs: dict[str, Any] = {}
|
||||
wifi = connection_type == CONNECTION_TYPES[1]
|
||||
if wifi:
|
||||
kwargs["wifi"] = True
|
||||
if username:
|
||||
kwargs["username"] = username
|
||||
if password:
|
||||
kwargs["password"] = password
|
||||
|
||||
async def _async_connect() -> tuple[pysaj.SAJ, pysaj.Sensors]:
|
||||
"""Connect to SAJ and verify connection."""
|
||||
saj = pysaj.SAJ(host, **kwargs)
|
||||
sensor_def = pysaj.Sensors(wifi)
|
||||
done = await saj.read(sensor_def)
|
||||
if not done:
|
||||
raise ConfigEntryNotReady("Failed to read initial sensor data")
|
||||
return saj, sensor_def
|
||||
|
||||
try:
|
||||
saj, sensor_def = await _async_connect()
|
||||
except pysaj.UnauthorizedException as err:
|
||||
if wifi:
|
||||
raise ConfigEntryAuthFailed("Authentication failed") from err
|
||||
raise ConfigEntryNotReady(
|
||||
"Wrong connection type or device rejected connection"
|
||||
) from err
|
||||
except pysaj.UnexpectedResponseException as err:
|
||||
raise ConfigEntryNotReady(f"Connection error: {err}") from err
|
||||
except TimeoutError as err:
|
||||
raise ConfigEntryNotReady(f"Connection timeout: {err}") from err
|
||||
except OSError as err:
|
||||
raise ConfigEntryNotReady(f"Network error: {err}") from err
|
||||
|
||||
polling = SAJPolling(hass, entry, saj, sensor_def)
|
||||
entry.runtime_data = SAJRuntimeData(saj=saj, sensor_def=sensor_def, polling=polling)
|
||||
entry.async_on_unload(polling.async_shutdown)
|
||||
|
||||
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
|
||||
|
||||
return True
|
||||
|
||||
|
||||
async def async_unload_entry(hass: HomeAssistant, entry: SAJConfigEntry) -> bool:
|
||||
"""Unload a config entry."""
|
||||
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
|
||||
|
||||
@@ -0,0 +1,244 @@
|
||||
"""Config flow for SAJ."""
|
||||
|
||||
import logging
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
import pysaj
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
|
||||
from homeassistant.const import (
|
||||
CONF_HOST,
|
||||
CONF_NAME,
|
||||
CONF_PASSWORD,
|
||||
CONF_TYPE,
|
||||
CONF_USERNAME,
|
||||
)
|
||||
from homeassistant.exceptions import HomeAssistantError
|
||||
from homeassistant.helpers.typing import ConfigType
|
||||
|
||||
from .const import CONNECTION_TYPES, DOMAIN, INTEGRATION_TITLE
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class CannotConnect(HomeAssistantError):
|
||||
"""Error to indicate we cannot connect."""
|
||||
|
||||
|
||||
class InvalidAuth(HomeAssistantError):
|
||||
"""Error to indicate invalid credentials (config flow)."""
|
||||
|
||||
|
||||
class SAJConfigFlow(ConfigFlow, domain=DOMAIN):
|
||||
"""Handle a config flow for the SAJ Solar Inverter."""
|
||||
|
||||
VERSION = 1
|
||||
|
||||
def __init__(self) -> None:
|
||||
"""Initialize flow."""
|
||||
self._host: str | None = None
|
||||
self._connection_type: str | None = None
|
||||
self._pending_username: str = ""
|
||||
self._pending_password: str = ""
|
||||
|
||||
async def async_step_import(self, import_config: ConfigType) -> ConfigFlowResult:
|
||||
"""Import a config entry from configuration.yaml (sensor platform)."""
|
||||
_LOGGER.warning("Importing SAJ from YAML is deprecated and will be removed")
|
||||
entry_input: dict[str, Any] = {
|
||||
CONF_HOST: import_config[CONF_HOST],
|
||||
CONF_TYPE: import_config.get(CONF_TYPE, CONNECTION_TYPES[0]),
|
||||
CONF_USERNAME: import_config.get(CONF_USERNAME),
|
||||
CONF_PASSWORD: import_config.get(CONF_PASSWORD),
|
||||
}
|
||||
try:
|
||||
serial_number = await self._async_validate_input(entry_input)
|
||||
except InvalidAuth:
|
||||
return self.async_abort(reason="invalid_auth")
|
||||
except CannotConnect:
|
||||
return self.async_abort(reason="cannot_connect")
|
||||
except Exception:
|
||||
_LOGGER.exception("Unexpected error importing SAJ from YAML")
|
||||
return self.async_abort(reason="unknown")
|
||||
|
||||
data = {
|
||||
CONF_HOST: entry_input[CONF_HOST],
|
||||
CONF_TYPE: entry_input[CONF_TYPE],
|
||||
CONF_USERNAME: entry_input.get(CONF_USERNAME),
|
||||
CONF_PASSWORD: entry_input.get(CONF_PASSWORD),
|
||||
}
|
||||
title = (import_config.get(CONF_NAME) or "").strip() or INTEGRATION_TITLE
|
||||
|
||||
await self.async_set_unique_id(serial_number)
|
||||
self._abort_if_unique_id_configured(updates=data)
|
||||
|
||||
return self.async_create_entry(title=title, data=data)
|
||||
|
||||
async def _async_validate_input(self, user_input: dict[str, Any]) -> str:
|
||||
"""Validate the user input allows us to connect.
|
||||
|
||||
Returns the device serial number (required after a successful read).
|
||||
"""
|
||||
host = user_input[CONF_HOST]
|
||||
connection_type = user_input[CONF_TYPE]
|
||||
username = user_input.get(CONF_USERNAME)
|
||||
password = user_input.get(CONF_PASSWORD)
|
||||
|
||||
wifi = connection_type == CONNECTION_TYPES[1]
|
||||
kwargs: dict[str, Any] = {}
|
||||
if wifi:
|
||||
kwargs["wifi"] = True
|
||||
if username:
|
||||
kwargs["username"] = username
|
||||
if password:
|
||||
kwargs["password"] = password
|
||||
|
||||
async def _async_validate_connection() -> str:
|
||||
"""Validate connection and get serial number."""
|
||||
saj = pysaj.SAJ(host, **kwargs)
|
||||
sensor_def = pysaj.Sensors(wifi)
|
||||
done = await saj.read(sensor_def)
|
||||
if not done:
|
||||
raise CannotConnect("Failed to read sensor data")
|
||||
|
||||
serial_number = saj.serialnumber
|
||||
if not serial_number:
|
||||
raise CannotConnect("Device did not return a serial number")
|
||||
return serial_number
|
||||
|
||||
try:
|
||||
serial_number = await _async_validate_connection()
|
||||
|
||||
except pysaj.UnauthorizedException as err:
|
||||
# Only raise auth error for WiFi connections with wrong credentials
|
||||
if wifi:
|
||||
raise InvalidAuth("Invalid authentication") from err
|
||||
# For ethernet, this likely means wrong connection type - treat as connection error
|
||||
raise CannotConnect("Wrong connection type or cannot connect") from err
|
||||
except pysaj.UnexpectedResponseException as err:
|
||||
raise CannotConnect(f"Connection error: {err}") from err
|
||||
except CannotConnect, InvalidAuth:
|
||||
raise
|
||||
except Exception as err:
|
||||
raise CannotConnect(f"Connection failed: {err}") from err
|
||||
|
||||
return serial_number
|
||||
|
||||
async def async_step_user(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> ConfigFlowResult:
|
||||
"""Handle a flow started by the user."""
|
||||
errors = {}
|
||||
|
||||
if user_input is not None:
|
||||
host = user_input[CONF_HOST]
|
||||
connection_type = user_input.get(CONF_TYPE, CONNECTION_TYPES[0])
|
||||
|
||||
# Store host/type for a possible WiFi credentials step
|
||||
self._host = host
|
||||
self._connection_type = connection_type
|
||||
self._pending_username = user_input.get(CONF_USERNAME) or ""
|
||||
self._pending_password = user_input.get(CONF_PASSWORD) or ""
|
||||
|
||||
entry_input = {
|
||||
CONF_HOST: host,
|
||||
CONF_TYPE: connection_type,
|
||||
CONF_USERNAME: user_input.get(CONF_USERNAME),
|
||||
CONF_PASSWORD: user_input.get(CONF_PASSWORD),
|
||||
}
|
||||
|
||||
# Ethernet ignores username/password; WiFi tries open access or creds from this step
|
||||
try:
|
||||
serial_number = await self._async_validate_input(entry_input)
|
||||
except InvalidAuth:
|
||||
# WiFi-only: device is SAJ but requires credentials
|
||||
return await self.async_step_device_credentials()
|
||||
except CannotConnect:
|
||||
errors["base"] = "cannot_connect"
|
||||
except Exception:
|
||||
_LOGGER.exception("Unexpected error during user flow")
|
||||
errors["base"] = "unknown"
|
||||
else:
|
||||
data = {
|
||||
CONF_HOST: entry_input[CONF_HOST],
|
||||
CONF_TYPE: entry_input[CONF_TYPE],
|
||||
CONF_USERNAME: entry_input.get(CONF_USERNAME),
|
||||
CONF_PASSWORD: entry_input.get(CONF_PASSWORD),
|
||||
}
|
||||
|
||||
await self.async_set_unique_id(serial_number)
|
||||
self._abort_if_unique_id_configured(updates=data)
|
||||
|
||||
return self.async_create_entry(title=INTEGRATION_TITLE, data=data)
|
||||
|
||||
return self.async_show_form(
|
||||
step_id="user",
|
||||
data_schema=self._schema_user(),
|
||||
errors=errors or None,
|
||||
)
|
||||
|
||||
def _schema_user(self) -> vol.Schema:
|
||||
"""Define the schema for the user step."""
|
||||
return vol.Schema(
|
||||
{
|
||||
vol.Required(CONF_HOST): str,
|
||||
vol.Optional(CONF_TYPE, default=CONNECTION_TYPES[0]): vol.In(
|
||||
CONNECTION_TYPES
|
||||
),
|
||||
vol.Optional(CONF_USERNAME, default=""): str,
|
||||
vol.Optional(CONF_PASSWORD, default=""): str,
|
||||
}
|
||||
)
|
||||
|
||||
async def async_step_device_credentials(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> ConfigFlowResult:
|
||||
"""Handle device credentials step (only shown for WiFi connections)."""
|
||||
errors = {}
|
||||
|
||||
if user_input is not None:
|
||||
if TYPE_CHECKING:
|
||||
assert self._host is not None
|
||||
assert self._connection_type is not None
|
||||
|
||||
combined_input = {
|
||||
CONF_HOST: self._host,
|
||||
CONF_TYPE: self._connection_type,
|
||||
CONF_USERNAME: user_input.get(CONF_USERNAME),
|
||||
CONF_PASSWORD: user_input.get(CONF_PASSWORD),
|
||||
}
|
||||
|
||||
try:
|
||||
serial_number = await self._async_validate_input(combined_input)
|
||||
data = {
|
||||
CONF_HOST: combined_input[CONF_HOST],
|
||||
CONF_TYPE: combined_input[CONF_TYPE],
|
||||
CONF_USERNAME: combined_input.get(CONF_USERNAME),
|
||||
CONF_PASSWORD: combined_input.get(CONF_PASSWORD),
|
||||
}
|
||||
except InvalidAuth:
|
||||
errors["base"] = "invalid_auth"
|
||||
except CannotConnect:
|
||||
errors["base"] = "cannot_connect"
|
||||
except Exception:
|
||||
_LOGGER.exception("Unexpected error during device credentials flow")
|
||||
errors["base"] = "unknown"
|
||||
else:
|
||||
await self.async_set_unique_id(serial_number)
|
||||
self._abort_if_unique_id_configured(updates=data)
|
||||
|
||||
return self.async_create_entry(title=INTEGRATION_TITLE, data=data)
|
||||
return self.async_show_form(
|
||||
step_id="device_credentials",
|
||||
data_schema=self._schema_device_credentials(),
|
||||
errors=errors or None,
|
||||
)
|
||||
|
||||
def _schema_device_credentials(self) -> vol.Schema:
|
||||
"""Define the schema for the device credentials step."""
|
||||
return vol.Schema(
|
||||
{
|
||||
vol.Optional(CONF_USERNAME, default=self._pending_username): str,
|
||||
vol.Optional(CONF_PASSWORD, default=self._pending_password): str,
|
||||
}
|
||||
)
|
||||
@@ -0,0 +1,9 @@
|
||||
"""Constants for the SAJ Solar Inverter integration."""
|
||||
|
||||
from typing import Final
|
||||
|
||||
DOMAIN: Final = "saj"
|
||||
|
||||
INTEGRATION_TITLE: Final = "SAJ Solar Inverter"
|
||||
|
||||
CONNECTION_TYPES = ["ethernet", "wifi"]
|
||||
@@ -2,7 +2,9 @@
|
||||
"domain": "saj",
|
||||
"name": "SAJ Solar Inverter",
|
||||
"codeowners": ["@fredericvl"],
|
||||
"config_flow": true,
|
||||
"documentation": "https://www.home-assistant.io/integrations/saj",
|
||||
"integration_type": "hub",
|
||||
"iot_class": "local_polling",
|
||||
"loggers": ["pysaj"],
|
||||
"quality_scale": "legacy",
|
||||
|
||||
@@ -1,9 +1,6 @@
|
||||
"""SAJ solar inverter interface."""
|
||||
|
||||
from collections.abc import Callable, Coroutine
|
||||
from datetime import date, datetime
|
||||
import logging
|
||||
from typing import Any
|
||||
from datetime import date
|
||||
|
||||
import pysaj
|
||||
import voluptuous as vol
|
||||
@@ -14,34 +11,31 @@ from homeassistant.components.sensor import (
|
||||
SensorEntity,
|
||||
SensorStateClass,
|
||||
)
|
||||
from homeassistant.config_entries import SOURCE_IMPORT
|
||||
from homeassistant.const import (
|
||||
CONF_HOST,
|
||||
CONF_NAME,
|
||||
CONF_PASSWORD,
|
||||
CONF_TYPE,
|
||||
CONF_USERNAME,
|
||||
EVENT_HOMEASSISTANT_STOP,
|
||||
UnitOfEnergy,
|
||||
UnitOfMass,
|
||||
UnitOfPower,
|
||||
UnitOfTemperature,
|
||||
UnitOfTime,
|
||||
)
|
||||
from homeassistant.core import CALLBACK_TYPE, HomeAssistant, callback
|
||||
from homeassistant.exceptions import PlatformNotReady
|
||||
from homeassistant.helpers import config_validation as cv
|
||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||
from homeassistant.helpers.event import async_call_later
|
||||
from homeassistant.helpers.start import async_at_start
|
||||
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
|
||||
from homeassistant.core import DOMAIN as HOMEASSISTANT_DOMAIN, HomeAssistant, callback
|
||||
from homeassistant.data_entry_flow import FlowResultType
|
||||
from homeassistant.helpers import config_validation as cv, issue_registry as ir
|
||||
from homeassistant.helpers.entity_platform import (
|
||||
AddConfigEntryEntitiesCallback,
|
||||
AddEntitiesCallback,
|
||||
)
|
||||
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType, StateType
|
||||
from homeassistant.util import dt as dt_util
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
MIN_INTERVAL = 5
|
||||
MAX_INTERVAL = 300
|
||||
|
||||
INVERTER_TYPES = ["ethernet", "wifi"]
|
||||
from . import SAJConfigEntry, SAJRuntimeData
|
||||
from .const import CONNECTION_TYPES, DOMAIN, INTEGRATION_TITLE
|
||||
|
||||
SAJ_UNIT_MAPPINGS = {
|
||||
"": None,
|
||||
@@ -56,138 +50,93 @@ PLATFORM_SCHEMA = SENSOR_PLATFORM_SCHEMA.extend(
|
||||
{
|
||||
vol.Required(CONF_HOST): cv.string,
|
||||
vol.Optional(CONF_NAME): cv.string,
|
||||
vol.Optional(CONF_TYPE, default=INVERTER_TYPES[0]): vol.In(INVERTER_TYPES),
|
||||
vol.Optional(CONF_TYPE, default=CONNECTION_TYPES[0]): vol.In(CONNECTION_TYPES),
|
||||
vol.Inclusive(CONF_USERNAME, "credentials"): cv.string,
|
||||
vol.Inclusive(CONF_PASSWORD, "credentials"): cv.string,
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
async def async_setup_entry(
|
||||
hass: HomeAssistant,
|
||||
entry: SAJConfigEntry,
|
||||
async_add_entities: AddConfigEntryEntitiesCallback,
|
||||
) -> None:
|
||||
"""Set up the SAJ sensors from a config entry."""
|
||||
runtime = entry.runtime_data
|
||||
sensor_def = runtime.sensor_def
|
||||
|
||||
hass_sensors = [
|
||||
SAJsensor(runtime, entry.unique_id, sensor, inverter_name=None)
|
||||
for sensor in sensor_def
|
||||
if sensor.enabled
|
||||
]
|
||||
|
||||
async_add_entities(hass_sensors)
|
||||
|
||||
|
||||
async def async_setup_platform(
|
||||
hass: HomeAssistant,
|
||||
config: ConfigType,
|
||||
async_add_entities: AddEntitiesCallback,
|
||||
discovery_info: DiscoveryInfoType | None = None,
|
||||
) -> None:
|
||||
"""Set up the SAJ sensors."""
|
||||
|
||||
remove_interval_update = None
|
||||
wifi = config[CONF_TYPE] == INVERTER_TYPES[1]
|
||||
|
||||
# Init all sensors
|
||||
sensor_def = pysaj.Sensors(wifi)
|
||||
|
||||
# Use all sensors by default
|
||||
hass_sensors: list[SAJsensor] = []
|
||||
|
||||
kwargs = {}
|
||||
if wifi:
|
||||
kwargs["wifi"] = True
|
||||
if config.get(CONF_USERNAME) and config.get(CONF_PASSWORD):
|
||||
kwargs["username"] = config[CONF_USERNAME]
|
||||
kwargs["password"] = config[CONF_PASSWORD]
|
||||
|
||||
try:
|
||||
saj = pysaj.SAJ(config[CONF_HOST], **kwargs)
|
||||
done = await saj.read(sensor_def)
|
||||
except pysaj.UnauthorizedException:
|
||||
_LOGGER.error("Username and/or password is wrong")
|
||||
return
|
||||
except pysaj.UnexpectedResponseException as err:
|
||||
_LOGGER.error(
|
||||
"Error in SAJ, please check host/ip address. Original error: %s", err
|
||||
"""Migrate YAML sensor platform configuration to a config entry."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_IMPORT},
|
||||
data=dict(config),
|
||||
)
|
||||
if (
|
||||
result.get("type") is FlowResultType.ABORT
|
||||
and result.get("reason") != "already_configured"
|
||||
):
|
||||
reason = result.get("reason", "unknown")
|
||||
ir.async_create_issue(
|
||||
hass,
|
||||
DOMAIN,
|
||||
f"deprecated_yaml_import_issue_{reason}",
|
||||
is_fixable=False,
|
||||
issue_domain=DOMAIN,
|
||||
severity=ir.IssueSeverity.WARNING,
|
||||
translation_key=f"deprecated_yaml_import_issue_{reason}",
|
||||
translation_placeholders={
|
||||
"domain": DOMAIN,
|
||||
"integration_title": INTEGRATION_TITLE,
|
||||
},
|
||||
)
|
||||
return
|
||||
|
||||
if not done:
|
||||
raise PlatformNotReady
|
||||
|
||||
hass_sensors.extend(
|
||||
SAJsensor(saj.serialnumber, sensor, inverter_name=config.get(CONF_NAME))
|
||||
for sensor in sensor_def
|
||||
if sensor.enabled
|
||||
ir.async_create_issue(
|
||||
hass,
|
||||
HOMEASSISTANT_DOMAIN,
|
||||
f"deprecated_yaml_{DOMAIN}",
|
||||
is_fixable=False,
|
||||
issue_domain=DOMAIN,
|
||||
severity=ir.IssueSeverity.WARNING,
|
||||
translation_key="deprecated_yaml",
|
||||
translation_placeholders={
|
||||
"domain": DOMAIN,
|
||||
"integration_title": INTEGRATION_TITLE,
|
||||
},
|
||||
)
|
||||
|
||||
async_add_entities(hass_sensors)
|
||||
|
||||
async def async_saj() -> bool:
|
||||
"""Update all the SAJ sensors."""
|
||||
success = await saj.read(sensor_def)
|
||||
|
||||
for sensor in hass_sensors:
|
||||
state_unknown = False
|
||||
# SAJ inverters are powered by DC via solar panels and thus are
|
||||
# offline after the sun has set. If a sensor resets on a daily
|
||||
# basis like "today_yield", this reset won't happen automatically.
|
||||
# Code below checks if today > day when sensor was last updated
|
||||
# and if so: set state to None.
|
||||
# Sensors with live values like "temperature" or "current_power"
|
||||
# will also be reset to None.
|
||||
if not success and (
|
||||
(sensor.per_day_basis and dt_util.now().date() > sensor.date_updated)
|
||||
or (not sensor.per_day_basis and not sensor.per_total_basis)
|
||||
):
|
||||
state_unknown = True
|
||||
sensor.async_update_values(unknown_state=state_unknown)
|
||||
|
||||
return success
|
||||
|
||||
@callback
|
||||
def start_update_interval(hass: HomeAssistant) -> None:
|
||||
"""Start the update interval scheduling."""
|
||||
nonlocal remove_interval_update
|
||||
remove_interval_update = async_track_time_interval_backoff(hass, async_saj)
|
||||
|
||||
@callback
|
||||
def stop_update_interval(event):
|
||||
"""Properly cancel the scheduled update."""
|
||||
remove_interval_update()
|
||||
|
||||
hass.bus.async_listen(EVENT_HOMEASSISTANT_STOP, stop_update_interval)
|
||||
async_at_start(hass, start_update_interval)
|
||||
|
||||
|
||||
@callback
|
||||
def async_track_time_interval_backoff(
|
||||
hass: HomeAssistant, action: Callable[[], Coroutine[Any, Any, bool]]
|
||||
) -> CALLBACK_TYPE:
|
||||
"""Add a listener that fires repetitively and increases the interval when failed."""
|
||||
remove = None
|
||||
interval = MIN_INTERVAL
|
||||
|
||||
async def interval_listener(now: datetime | None = None) -> None:
|
||||
"""Handle elapsed interval with backoff."""
|
||||
nonlocal interval, remove
|
||||
try:
|
||||
if await action():
|
||||
interval = MIN_INTERVAL
|
||||
else:
|
||||
interval = min(interval * 2, MAX_INTERVAL)
|
||||
finally:
|
||||
remove = async_call_later(hass, interval, interval_listener)
|
||||
|
||||
hass.async_create_task(interval_listener())
|
||||
|
||||
def remove_listener() -> None:
|
||||
"""Remove interval listener."""
|
||||
if remove:
|
||||
remove()
|
||||
|
||||
return remove_listener
|
||||
|
||||
|
||||
class SAJsensor(SensorEntity):
|
||||
"""Representation of a SAJ sensor."""
|
||||
|
||||
_attr_should_poll = False
|
||||
_state: StateType
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
runtime: SAJRuntimeData,
|
||||
serialnumber: str | None,
|
||||
pysaj_sensor: pysaj.Sensor,
|
||||
inverter_name: str | None = None,
|
||||
) -> None:
|
||||
"""Initialize the SAJ sensor."""
|
||||
self._runtime = runtime
|
||||
self._sensor = pysaj_sensor
|
||||
self._inverter_name = inverter_name
|
||||
self._serialnumber = serialnumber
|
||||
@@ -215,8 +164,15 @@ class SAJsensor(SensorEntity):
|
||||
):
|
||||
self._attr_device_class = SensorDeviceClass.TEMPERATURE
|
||||
|
||||
async def async_added_to_hass(self) -> None:
|
||||
"""Register for inverter poll updates."""
|
||||
await super().async_added_to_hass()
|
||||
self.async_on_remove(
|
||||
self._runtime.polling.async_add_poll_listener(self._on_poll_success)
|
||||
)
|
||||
|
||||
@property
|
||||
def native_value(self):
|
||||
def native_value(self) -> StateType:
|
||||
"""Return the state of the sensor."""
|
||||
return self._state
|
||||
|
||||
@@ -236,15 +192,21 @@ class SAJsensor(SensorEntity):
|
||||
return self._sensor.date
|
||||
|
||||
@callback
|
||||
def async_update_values(self, unknown_state=False):
|
||||
"""Update this sensor."""
|
||||
update = False
|
||||
def _on_poll_success(self, success: bool) -> None:
|
||||
"""Update state from the inverter after a poll."""
|
||||
state_unknown = False
|
||||
if not success and (
|
||||
(self.per_day_basis and dt_util.now().date() > self.date_updated)
|
||||
or (not self.per_day_basis and not self.per_total_basis)
|
||||
):
|
||||
state_unknown = True
|
||||
|
||||
update = False
|
||||
if self._sensor.value != self._state:
|
||||
update = True
|
||||
self._state = self._sensor.value
|
||||
|
||||
if unknown_state and self._state is not None:
|
||||
if state_unknown and self._state is not None:
|
||||
update = True
|
||||
self._state = None
|
||||
|
||||
|
||||
@@ -0,0 +1,48 @@
|
||||
{
|
||||
"config": {
|
||||
"abort": {
|
||||
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
|
||||
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
|
||||
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
|
||||
"unknown": "[%key:common::config_flow::error::unknown%]"
|
||||
},
|
||||
"error": {
|
||||
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
|
||||
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
|
||||
"unknown": "[%key:common::config_flow::error::unknown%]"
|
||||
},
|
||||
"step": {
|
||||
"device_credentials": {
|
||||
"data": {
|
||||
"password": "[%key:common::config_flow::data::password%]",
|
||||
"username": "[%key:common::config_flow::data::username%]"
|
||||
},
|
||||
"description": "The inverter required WiFi login. Enter or correct username and password.",
|
||||
"title": "Device credentials"
|
||||
},
|
||||
"user": {
|
||||
"data": {
|
||||
"host": "[%key:common::config_flow::data::host%]",
|
||||
"password": "[%key:common::config_flow::data::password%]",
|
||||
"type": "Connection type",
|
||||
"username": "[%key:common::config_flow::data::username%]"
|
||||
},
|
||||
"description": "Configure your SAJ solar inverter. Enter the IP address or hostname. If you use WiFi, username and password are optional here (open networks); if login is required you can enter them now or on the next step after we detect the inverter."
|
||||
}
|
||||
}
|
||||
},
|
||||
"issues": {
|
||||
"deprecated_yaml_import_issue_cannot_connect": {
|
||||
"description": "Configuring {integration_title} using the YAML sensor platform is deprecated.\n\nHome Assistant could not reach the inverter while importing your configuration. Check the host, network, and that the device is an SAJ inverter, then restart to retry the import, or remove the `sensor` / `saj` YAML and add the integration from the UI.",
|
||||
"title": "{domain} YAML configuration import failed"
|
||||
},
|
||||
"deprecated_yaml_import_issue_invalid_auth": {
|
||||
"description": "Configuring {integration_title} using the YAML sensor platform is deprecated.\n\nThe username or password in your YAML was rejected. Update the credentials in YAML and restart, or remove the `sensor` / `saj` YAML and set up the integration from the UI.",
|
||||
"title": "{domain} YAML configuration import failed"
|
||||
},
|
||||
"deprecated_yaml_import_issue_unknown": {
|
||||
"description": "Configuring {integration_title} using the YAML sensor platform is deprecated.\n\nAn unexpected error occurred while importing your configuration. Restart Home Assistant to try again, or remove the `sensor` / `saj` YAML and add the integration from the UI.",
|
||||
"title": "[%key:component::saj::issues::deprecated_yaml_import_issue_cannot_connect::title%]"
|
||||
}
|
||||
}
|
||||
}
|
||||
Generated
+1
@@ -656,6 +656,7 @@ FLOWS = {
|
||||
"ruuvitag_ble",
|
||||
"rympro",
|
||||
"sabnzbd",
|
||||
"saj",
|
||||
"samsung_infrared",
|
||||
"samsungtv",
|
||||
"sanix",
|
||||
|
||||
@@ -6165,7 +6165,7 @@
|
||||
"saj": {
|
||||
"name": "SAJ Solar Inverter",
|
||||
"integration_type": "hub",
|
||||
"config_flow": false,
|
||||
"config_flow": true,
|
||||
"iot_class": "local_polling"
|
||||
},
|
||||
"samsam": {
|
||||
|
||||
@@ -0,0 +1,33 @@
|
||||
"""Tests for the saj integration."""
|
||||
|
||||
from homeassistant.components.saj.const import CONNECTION_TYPES
|
||||
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_TYPE, CONF_USERNAME
|
||||
from homeassistant.core import HomeAssistant
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
MOCK_USER_INPUT_ETHERNET = {
|
||||
CONF_HOST: "192.168.1.100",
|
||||
CONF_TYPE: CONNECTION_TYPES[0],
|
||||
}
|
||||
|
||||
MOCK_USER_INPUT_WIFI = {
|
||||
CONF_HOST: "192.168.1.100",
|
||||
CONF_TYPE: CONNECTION_TYPES[1],
|
||||
CONF_USERNAME: "admin",
|
||||
CONF_PASSWORD: "password",
|
||||
}
|
||||
|
||||
MOCK_SERIAL_NUMBER = "TEST123456789"
|
||||
|
||||
|
||||
async def setup_integration(
|
||||
hass: HomeAssistant, config_entry: MockConfigEntry
|
||||
) -> MockConfigEntry:
|
||||
"""Set up the integration for testing."""
|
||||
config_entry.add_to_hass(hass)
|
||||
|
||||
await hass.config_entries.async_setup(config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
return config_entry
|
||||
@@ -0,0 +1,114 @@
|
||||
"""Fixtures for saj tests."""
|
||||
|
||||
from collections.abc import Generator
|
||||
from datetime import date
|
||||
from typing import Any
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from homeassistant.components.saj.const import DOMAIN
|
||||
|
||||
from . import MOCK_SERIAL_NUMBER, MOCK_USER_INPUT_ETHERNET, MOCK_USER_INPUT_WIFI
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def connection_method(request: pytest.FixtureRequest) -> str:
|
||||
"""Connection method for the config entry fixture."""
|
||||
return getattr(request, "param", "ethernet")
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def config_entry_data(connection_method: str) -> dict[str, Any]:
|
||||
"""Return config entry data for the connection method."""
|
||||
if connection_method == "ethernet":
|
||||
return MOCK_USER_INPUT_ETHERNET
|
||||
return MOCK_USER_INPUT_WIFI
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_config_entry(
|
||||
config_entry_data: dict[str, Any],
|
||||
connection_method: str,
|
||||
) -> MockConfigEntry:
|
||||
"""Return a mocked config entry for the selected connection method."""
|
||||
return MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
title="SAJ Solar Inverter",
|
||||
unique_id=MOCK_SERIAL_NUMBER,
|
||||
data=config_entry_data,
|
||||
entry_id=f"saj_entry_{connection_method}",
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_setup_entry() -> Generator[AsyncMock]:
|
||||
"""Mock the setup entry."""
|
||||
with patch(
|
||||
"homeassistant.components.saj.async_setup_entry", return_value=True
|
||||
) as mock_setup_entry:
|
||||
yield mock_setup_entry
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_pysaj_sensors() -> Generator[list[MagicMock]]:
|
||||
"""Mock pysaj.Sensors across SAJ integration modules."""
|
||||
sensors: list[MagicMock] = []
|
||||
for key, value, unit, per_day_basis, per_total_basis in (
|
||||
("current_power", 5000.0, "W", False, False),
|
||||
("today_yield", 25.5, "kWh", True, False),
|
||||
):
|
||||
sensor = MagicMock()
|
||||
sensor.name = key
|
||||
sensor.key = key
|
||||
sensor.value = value
|
||||
sensor.unit = unit
|
||||
sensor.enabled = True
|
||||
sensor.per_day_basis = per_day_basis
|
||||
sensor.per_total_basis = per_total_basis
|
||||
sensor.date = date.today()
|
||||
sensors.append(sensor)
|
||||
|
||||
with (
|
||||
patch(
|
||||
"homeassistant.components.saj.pysaj.Sensors",
|
||||
autospec=True,
|
||||
return_value=sensors,
|
||||
) as sensors_cls,
|
||||
patch(
|
||||
"homeassistant.components.saj.config_flow.pysaj.Sensors",
|
||||
new=sensors_cls,
|
||||
),
|
||||
patch(
|
||||
"homeassistant.components.saj.sensor.pysaj.Sensors",
|
||||
new=sensors_cls,
|
||||
),
|
||||
):
|
||||
yield sensors
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_pysaj_saj(mock_pysaj_sensors: list[MagicMock]) -> Generator[MagicMock]:
|
||||
"""Mock pysaj.SAJ across SAJ integration modules."""
|
||||
saj_instance = MagicMock()
|
||||
saj_instance.serialnumber = MOCK_SERIAL_NUMBER
|
||||
saj_instance.read = AsyncMock(return_value=True)
|
||||
|
||||
with (
|
||||
patch(
|
||||
"homeassistant.components.saj.pysaj.SAJ",
|
||||
autospec=True,
|
||||
return_value=saj_instance,
|
||||
) as saj_cls,
|
||||
patch(
|
||||
"homeassistant.components.saj.config_flow.pysaj.SAJ",
|
||||
new=saj_cls,
|
||||
),
|
||||
patch(
|
||||
"homeassistant.components.saj.sensor.pysaj.SAJ",
|
||||
new=saj_cls,
|
||||
),
|
||||
):
|
||||
yield saj_instance
|
||||
@@ -0,0 +1,114 @@
|
||||
# serializer version: 1
|
||||
# name: test_sensors[sensor.saj_current_power-entry]
|
||||
EntityRegistryEntrySnapshot({
|
||||
'aliases': list([
|
||||
None,
|
||||
]),
|
||||
'area_id': None,
|
||||
'capabilities': dict({
|
||||
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
|
||||
}),
|
||||
'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.saj_current_power',
|
||||
'has_entity_name': False,
|
||||
'hidden_by': None,
|
||||
'icon': None,
|
||||
'id': <ANY>,
|
||||
'labels': set({
|
||||
}),
|
||||
'name': None,
|
||||
'object_id_base': 'saj_current_power',
|
||||
'options': dict({
|
||||
'sensor': dict({
|
||||
'suggested_display_precision': 0,
|
||||
}),
|
||||
}),
|
||||
'original_device_class': <SensorDeviceClass.POWER: 'power'>,
|
||||
'original_icon': None,
|
||||
'original_name': 'saj_current_power',
|
||||
'platform': 'saj',
|
||||
'previous_unique_id': None,
|
||||
'suggested_object_id': None,
|
||||
'supported_features': 0,
|
||||
'translation_key': None,
|
||||
'unique_id': 'TEST123456789_current_power',
|
||||
'unit_of_measurement': <UnitOfPower.WATT: 'W'>,
|
||||
})
|
||||
# ---
|
||||
# name: test_sensors[sensor.saj_current_power-state]
|
||||
StateSnapshot({
|
||||
'attributes': ReadOnlyDict({
|
||||
'device_class': 'power',
|
||||
'friendly_name': 'saj_current_power',
|
||||
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
|
||||
'unit_of_measurement': <UnitOfPower.WATT: 'W'>,
|
||||
}),
|
||||
'context': <ANY>,
|
||||
'entity_id': 'sensor.saj_current_power',
|
||||
'last_changed': <ANY>,
|
||||
'last_reported': <ANY>,
|
||||
'last_updated': <ANY>,
|
||||
'state': '5000.0',
|
||||
})
|
||||
# ---
|
||||
# name: test_sensors[sensor.saj_today_yield-entry]
|
||||
EntityRegistryEntrySnapshot({
|
||||
'aliases': list([
|
||||
None,
|
||||
]),
|
||||
'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.saj_today_yield',
|
||||
'has_entity_name': False,
|
||||
'hidden_by': None,
|
||||
'icon': None,
|
||||
'id': <ANY>,
|
||||
'labels': set({
|
||||
}),
|
||||
'name': None,
|
||||
'object_id_base': 'saj_today_yield',
|
||||
'options': dict({
|
||||
'sensor': dict({
|
||||
'suggested_display_precision': 2,
|
||||
}),
|
||||
}),
|
||||
'original_device_class': <SensorDeviceClass.ENERGY: 'energy'>,
|
||||
'original_icon': None,
|
||||
'original_name': 'saj_today_yield',
|
||||
'platform': 'saj',
|
||||
'previous_unique_id': None,
|
||||
'suggested_object_id': None,
|
||||
'supported_features': 0,
|
||||
'translation_key': None,
|
||||
'unique_id': 'TEST123456789_today_yield',
|
||||
'unit_of_measurement': <UnitOfEnergy.KILO_WATT_HOUR: 'kWh'>,
|
||||
})
|
||||
# ---
|
||||
# name: test_sensors[sensor.saj_today_yield-state]
|
||||
StateSnapshot({
|
||||
'attributes': ReadOnlyDict({
|
||||
'device_class': 'energy',
|
||||
'friendly_name': 'saj_today_yield',
|
||||
'unit_of_measurement': <UnitOfEnergy.KILO_WATT_HOUR: 'kWh'>,
|
||||
}),
|
||||
'context': <ANY>,
|
||||
'entity_id': 'sensor.saj_today_yield',
|
||||
'last_changed': <ANY>,
|
||||
'last_reported': <ANY>,
|
||||
'last_updated': <ANY>,
|
||||
'state': '25.5',
|
||||
})
|
||||
# ---
|
||||
@@ -0,0 +1,372 @@
|
||||
"""Test the saj config flow."""
|
||||
|
||||
from unittest.mock import AsyncMock, MagicMock
|
||||
|
||||
import pysaj
|
||||
import pytest
|
||||
|
||||
from homeassistant.components.saj.const import CONNECTION_TYPES, DOMAIN
|
||||
from homeassistant.config_entries import SOURCE_IMPORT, SOURCE_USER
|
||||
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_TYPE, CONF_USERNAME
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.data_entry_flow import FlowResultType
|
||||
|
||||
from . import MOCK_SERIAL_NUMBER, MOCK_USER_INPUT_ETHERNET, MOCK_USER_INPUT_WIFI
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
IMPORT_DATA_ETHERNET = {
|
||||
CONF_HOST: "192.168.1.88",
|
||||
CONF_TYPE: CONNECTION_TYPES[0],
|
||||
}
|
||||
|
||||
IMPORT_DATA_WIFI = {
|
||||
CONF_HOST: "192.168.1.88",
|
||||
CONF_TYPE: CONNECTION_TYPES[1],
|
||||
CONF_USERNAME: "u",
|
||||
CONF_PASSWORD: "p",
|
||||
}
|
||||
|
||||
|
||||
async def test_form_ethernet(
|
||||
hass: HomeAssistant, mock_setup_entry: AsyncMock, mock_pysaj_saj: MagicMock
|
||||
) -> None:
|
||||
"""Test we get the form for ethernet connection."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_USER}
|
||||
)
|
||||
assert result.get("type") is FlowResultType.FORM
|
||||
assert result.get("errors") is None
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
MOCK_USER_INPUT_ETHERNET,
|
||||
)
|
||||
|
||||
assert result.get("type") is FlowResultType.CREATE_ENTRY
|
||||
assert result.get("title") == "SAJ Solar Inverter"
|
||||
result_data = result.get("data")
|
||||
assert result_data is not None
|
||||
assert result_data[CONF_HOST] == MOCK_USER_INPUT_ETHERNET[CONF_HOST]
|
||||
assert result_data[CONF_TYPE] == MOCK_USER_INPUT_ETHERNET[CONF_TYPE]
|
||||
assert not result_data.get(CONF_USERNAME)
|
||||
assert not result_data.get(CONF_PASSWORD)
|
||||
result_entry = result.get("result")
|
||||
assert result_entry is not None
|
||||
assert result_entry.unique_id == MOCK_SERIAL_NUMBER
|
||||
assert len(mock_setup_entry.mock_calls) == 1
|
||||
|
||||
|
||||
async def test_form_wifi_open_network(
|
||||
hass: HomeAssistant, mock_setup_entry: AsyncMock, mock_pysaj_saj: MagicMock
|
||||
) -> None:
|
||||
"""Test WiFi without credentials when the device allows open access."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_USER}
|
||||
)
|
||||
assert result.get("type") is FlowResultType.FORM
|
||||
assert result.get("step_id") == "user"
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{
|
||||
CONF_HOST: MOCK_USER_INPUT_WIFI[CONF_HOST],
|
||||
CONF_TYPE: MOCK_USER_INPUT_WIFI[CONF_TYPE],
|
||||
},
|
||||
)
|
||||
|
||||
assert result.get("type") is FlowResultType.CREATE_ENTRY
|
||||
assert result.get("title") == "SAJ Solar Inverter"
|
||||
result_data = result.get("data")
|
||||
assert result_data is not None
|
||||
assert result_data[CONF_HOST] == MOCK_USER_INPUT_WIFI[CONF_HOST]
|
||||
assert result_data[CONF_TYPE] == MOCK_USER_INPUT_WIFI[CONF_TYPE]
|
||||
assert not result_data.get(CONF_USERNAME)
|
||||
assert not result_data.get(CONF_PASSWORD)
|
||||
result_entry = result.get("result")
|
||||
assert result_entry is not None
|
||||
assert result_entry.unique_id == MOCK_SERIAL_NUMBER
|
||||
assert len(mock_setup_entry.mock_calls) == 1
|
||||
|
||||
|
||||
async def test_form_wifi_requires_credentials(
|
||||
hass: HomeAssistant, mock_setup_entry: AsyncMock, mock_pysaj_saj: MagicMock
|
||||
) -> None:
|
||||
"""Test WiFi flow when the first probe requires authentication."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_USER}
|
||||
)
|
||||
|
||||
mock_pysaj_saj.read.side_effect = [
|
||||
pysaj.UnauthorizedException("Auth required"),
|
||||
True,
|
||||
]
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{
|
||||
CONF_HOST: MOCK_USER_INPUT_WIFI[CONF_HOST],
|
||||
CONF_TYPE: MOCK_USER_INPUT_WIFI[CONF_TYPE],
|
||||
},
|
||||
)
|
||||
|
||||
assert result.get("type") is FlowResultType.FORM
|
||||
assert result.get("step_id") == "device_credentials"
|
||||
assert result.get("errors") is None
|
||||
|
||||
mock_pysaj_saj.read.side_effect = None
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{
|
||||
CONF_USERNAME: MOCK_USER_INPUT_WIFI[CONF_USERNAME],
|
||||
CONF_PASSWORD: MOCK_USER_INPUT_WIFI[CONF_PASSWORD],
|
||||
},
|
||||
)
|
||||
|
||||
assert result.get("type") is FlowResultType.CREATE_ENTRY
|
||||
assert result.get("title") == "SAJ Solar Inverter"
|
||||
result_data = result.get("data")
|
||||
assert result_data is not None
|
||||
assert result_data == MOCK_USER_INPUT_WIFI
|
||||
result_entry = result.get("result")
|
||||
assert result_entry is not None
|
||||
assert result_entry.unique_id == MOCK_SERIAL_NUMBER
|
||||
assert len(mock_setup_entry.mock_calls) == 1
|
||||
|
||||
|
||||
async def test_form_missing_serial_number(
|
||||
hass: HomeAssistant, mock_pysaj_saj: MagicMock
|
||||
) -> None:
|
||||
"""Test we reject devices that respond but do not expose a serial number."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_USER}
|
||||
)
|
||||
|
||||
mock_pysaj_saj.serialnumber = None
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
MOCK_USER_INPUT_ETHERNET,
|
||||
)
|
||||
|
||||
assert result.get("type") is FlowResultType.FORM
|
||||
assert result.get("step_id") == "user"
|
||||
assert result.get("errors") == {"base": "cannot_connect"}
|
||||
|
||||
# Recover on the same flow once the device responds correctly.
|
||||
mock_pysaj_saj.serialnumber = MOCK_SERIAL_NUMBER
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
MOCK_USER_INPUT_ETHERNET,
|
||||
)
|
||||
assert result.get("type") is FlowResultType.CREATE_ENTRY
|
||||
|
||||
|
||||
async def test_form_wifi_probe_fails_shows_user_error(
|
||||
hass: HomeAssistant, mock_pysaj_saj: MagicMock
|
||||
) -> None:
|
||||
"""Test WiFi: failed probe (not SAJ / wrong host) keeps the user on host step."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_USER}
|
||||
)
|
||||
|
||||
mock_pysaj_saj.read.side_effect = pysaj.UnexpectedResponseException("not a saj")
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{
|
||||
CONF_HOST: MOCK_USER_INPUT_WIFI[CONF_HOST],
|
||||
CONF_TYPE: MOCK_USER_INPUT_WIFI[CONF_TYPE],
|
||||
},
|
||||
)
|
||||
|
||||
assert result.get("type") is FlowResultType.FORM
|
||||
assert result.get("step_id") == "user"
|
||||
assert result.get("errors") == {"base": "cannot_connect"}
|
||||
|
||||
# Recover without restarting the flow once the probe works.
|
||||
mock_pysaj_saj.read.side_effect = None
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{
|
||||
CONF_HOST: MOCK_USER_INPUT_WIFI[CONF_HOST],
|
||||
CONF_TYPE: MOCK_USER_INPUT_WIFI[CONF_TYPE],
|
||||
},
|
||||
)
|
||||
assert result.get("type") is FlowResultType.CREATE_ENTRY
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("exception", "error"),
|
||||
[
|
||||
(pysaj.UnexpectedResponseException("Connection failed"), "cannot_connect"),
|
||||
(pysaj.UnauthorizedException("Auth failed"), "cannot_connect"),
|
||||
(Exception("Unknown error"), "cannot_connect"),
|
||||
],
|
||||
)
|
||||
async def test_form_exceptions(
|
||||
hass: HomeAssistant,
|
||||
exception: Exception,
|
||||
error: str,
|
||||
mock_pysaj_saj: MagicMock,
|
||||
) -> None:
|
||||
"""Test we handle exceptions during form submission."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_USER}
|
||||
)
|
||||
|
||||
mock_pysaj_saj.read.side_effect = exception
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
MOCK_USER_INPUT_ETHERNET,
|
||||
)
|
||||
|
||||
assert result.get("type") is FlowResultType.FORM
|
||||
assert result.get("errors") == {"base": error}
|
||||
|
||||
# Recover once the inverter responds.
|
||||
mock_pysaj_saj.read.side_effect = None
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
MOCK_USER_INPUT_ETHERNET,
|
||||
)
|
||||
assert result.get("type") is FlowResultType.CREATE_ENTRY
|
||||
|
||||
|
||||
async def test_form_already_configured(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
mock_setup_entry: AsyncMock,
|
||||
mock_pysaj_saj: MagicMock,
|
||||
) -> None:
|
||||
"""Test starting a flow by user when already configured."""
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_USER}
|
||||
)
|
||||
assert result.get("type") is FlowResultType.FORM
|
||||
assert result.get("step_id") == "user"
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input=MOCK_USER_INPUT_ETHERNET,
|
||||
)
|
||||
|
||||
assert result.get("type") is FlowResultType.ABORT
|
||||
assert result.get("reason") == "already_configured"
|
||||
|
||||
|
||||
async def test_import_success(
|
||||
hass: HomeAssistant, mock_setup_entry: AsyncMock, mock_pysaj_saj: MagicMock
|
||||
) -> None:
|
||||
"""Test YAML import creates a config entry."""
|
||||
data = {
|
||||
CONF_HOST: "192.168.1.88",
|
||||
CONF_TYPE: CONNECTION_TYPES[0],
|
||||
}
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_IMPORT},
|
||||
data=data,
|
||||
)
|
||||
assert result.get("type") is FlowResultType.CREATE_ENTRY
|
||||
assert result.get("title") == "SAJ Solar Inverter"
|
||||
assert result.get("data") == {
|
||||
**data,
|
||||
CONF_USERNAME: None,
|
||||
CONF_PASSWORD: None,
|
||||
}
|
||||
result_entry = result.get("result")
|
||||
assert result_entry is not None
|
||||
assert result_entry.unique_id == MOCK_SERIAL_NUMBER
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("import_data", "exception", "abort_reason"),
|
||||
[
|
||||
pytest.param(
|
||||
IMPORT_DATA_ETHERNET,
|
||||
pysaj.UnexpectedResponseException("bad response"),
|
||||
"cannot_connect",
|
||||
id="unexpected_response",
|
||||
),
|
||||
pytest.param(
|
||||
IMPORT_DATA_ETHERNET,
|
||||
pysaj.UnauthorizedException("auth failed"),
|
||||
"cannot_connect",
|
||||
id="unauthorized_ethernet",
|
||||
),
|
||||
pytest.param(
|
||||
IMPORT_DATA_ETHERNET,
|
||||
Exception("Unknown error"),
|
||||
"cannot_connect",
|
||||
id="unexpected_error",
|
||||
),
|
||||
pytest.param(
|
||||
IMPORT_DATA_WIFI,
|
||||
pysaj.UnauthorizedException("auth failed"),
|
||||
"invalid_auth",
|
||||
id="unauthorized_wifi",
|
||||
),
|
||||
],
|
||||
)
|
||||
async def test_import_aborts(
|
||||
hass: HomeAssistant,
|
||||
mock_pysaj_saj: MagicMock,
|
||||
import_data: dict[str, str],
|
||||
exception: Exception,
|
||||
abort_reason: str,
|
||||
) -> None:
|
||||
"""Test YAML import aborts for validation failures."""
|
||||
mock_pysaj_saj.read.side_effect = exception
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_IMPORT},
|
||||
data=import_data,
|
||||
)
|
||||
assert result.get("type") is FlowResultType.ABORT
|
||||
assert result.get("reason") == abort_reason
|
||||
|
||||
|
||||
async def test_import_aborts_read_failed(
|
||||
hass: HomeAssistant, mock_pysaj_saj: MagicMock
|
||||
) -> None:
|
||||
"""Test YAML import aborts when the inverter read returns false."""
|
||||
mock_pysaj_saj.read.return_value = False
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_IMPORT},
|
||||
data=IMPORT_DATA_ETHERNET,
|
||||
)
|
||||
assert result.get("type") is FlowResultType.ABORT
|
||||
assert result.get("reason") == "cannot_connect"
|
||||
|
||||
|
||||
async def test_import_aborts_missing_serial(
|
||||
hass: HomeAssistant, mock_pysaj_saj: MagicMock
|
||||
) -> None:
|
||||
"""Test YAML import aborts when the device does not return a serial number."""
|
||||
mock_pysaj_saj.serialnumber = None
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_IMPORT},
|
||||
data=IMPORT_DATA_ETHERNET,
|
||||
)
|
||||
assert result.get("type") is FlowResultType.ABORT
|
||||
assert result.get("reason") == "cannot_connect"
|
||||
|
||||
|
||||
async def test_import_already_configured_aborts(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
mock_pysaj_saj: MagicMock,
|
||||
) -> None:
|
||||
"""Test YAML import aborts when the device is already configured."""
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_IMPORT},
|
||||
data=IMPORT_DATA_ETHERNET,
|
||||
)
|
||||
assert result.get("type") is FlowResultType.ABORT
|
||||
assert result.get("reason") == "already_configured"
|
||||
@@ -0,0 +1,97 @@
|
||||
"""Tests for the saj integration initialization."""
|
||||
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import pysaj
|
||||
import pytest
|
||||
|
||||
from homeassistant.config_entries import ConfigEntryState
|
||||
from homeassistant.core import HomeAssistant
|
||||
|
||||
from . import setup_integration
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
|
||||
@pytest.mark.parametrize("connection_method", ["ethernet", "wifi"], indirect=True)
|
||||
@pytest.mark.usefixtures("mock_pysaj_saj")
|
||||
async def test_setup_entry(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
) -> None:
|
||||
"""Test async_setup_entry for ethernet and wifi connections."""
|
||||
entry = await setup_integration(hass, mock_config_entry)
|
||||
|
||||
assert entry.state is ConfigEntryState.LOADED
|
||||
|
||||
|
||||
async def test_setup_entry_connection_failed(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
mock_pysaj_saj: MagicMock,
|
||||
) -> None:
|
||||
"""Test async_setup_entry handles connection failures."""
|
||||
mock_pysaj_saj.read.return_value = False
|
||||
entry = await setup_integration(hass, mock_config_entry)
|
||||
# Entry should be in SETUP_RETRY state when setup fails
|
||||
assert entry.state is ConfigEntryState.SETUP_RETRY
|
||||
|
||||
|
||||
@pytest.mark.parametrize("connection_method", ["wifi"], indirect=True)
|
||||
async def test_setup_entry_auth_failed(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
mock_pysaj_saj: MagicMock,
|
||||
) -> None:
|
||||
"""Test async_setup_entry fails WiFi auth at setup without endless retries."""
|
||||
mock_pysaj_saj.read.side_effect = pysaj.UnauthorizedException("Auth failed")
|
||||
entry = await setup_integration(hass, mock_config_entry)
|
||||
assert entry.state is ConfigEntryState.SETUP_ERROR
|
||||
|
||||
|
||||
async def test_setup_entry_ethernet_unauthorized_retries(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
mock_pysaj_saj: MagicMock,
|
||||
) -> None:
|
||||
"""Ethernet UnauthorizedException is treated as not ready (e.g. wrong type)."""
|
||||
mock_pysaj_saj.read.side_effect = pysaj.UnauthorizedException("unexpected")
|
||||
entry = await setup_integration(hass, mock_config_entry)
|
||||
assert entry.state is ConfigEntryState.SETUP_RETRY
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"exception",
|
||||
[
|
||||
Exception("Unexpected error"),
|
||||
RuntimeError("Unexpected runtime error"),
|
||||
],
|
||||
)
|
||||
async def test_setup_entry_unexpected_error(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
mock_pysaj_saj: MagicMock,
|
||||
exception: Exception,
|
||||
) -> None:
|
||||
"""Test async_setup_entry handles unexpected errors."""
|
||||
mock_pysaj_saj.read.side_effect = exception
|
||||
entry = await setup_integration(hass, mock_config_entry)
|
||||
# Truly unexpected exceptions should result in SETUP_ERROR
|
||||
# so the actual error is visible rather than being hidden
|
||||
assert entry.state is ConfigEntryState.SETUP_ERROR
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("mock_pysaj_saj")
|
||||
async def test_unload_entry(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
) -> None:
|
||||
"""Test async_unload_entry."""
|
||||
entry = await setup_integration(hass, mock_config_entry)
|
||||
|
||||
assert entry.state is ConfigEntryState.LOADED
|
||||
|
||||
await hass.config_entries.async_unload(entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert entry.state is ConfigEntryState.NOT_LOADED
|
||||
@@ -0,0 +1,107 @@
|
||||
"""Test the saj sensor platform."""
|
||||
|
||||
from datetime import timedelta
|
||||
from unittest.mock import AsyncMock, MagicMock
|
||||
|
||||
from freezegun.api import FrozenDateTimeFactory
|
||||
import pysaj
|
||||
import pytest
|
||||
from syrupy.assertion import SnapshotAssertion
|
||||
|
||||
from homeassistant.components.saj import MIN_INTERVAL_SEC
|
||||
from homeassistant.components.saj.const import DOMAIN
|
||||
from homeassistant.config_entries import ConfigEntryState
|
||||
from homeassistant.const import CONF_HOST, STATE_UNKNOWN, Platform
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers import entity_registry as er, issue_registry as ir
|
||||
from homeassistant.setup import async_setup_component
|
||||
|
||||
from . import setup_integration
|
||||
|
||||
from tests.common import MockConfigEntry, async_fire_time_changed, snapshot_platform
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def platforms() -> list[Platform]:
|
||||
"""Fixture to specify platforms to test."""
|
||||
return [Platform.SENSOR]
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("entity_registry_enabled_by_default", "mock_pysaj_saj")
|
||||
async def test_sensors(
|
||||
hass: HomeAssistant,
|
||||
snapshot: SnapshotAssertion,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
entity_registry: er.EntityRegistry,
|
||||
) -> None:
|
||||
"""Test the sensor entities."""
|
||||
await setup_integration(hass, mock_config_entry)
|
||||
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
|
||||
|
||||
|
||||
async def test_sensor_update_failure(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
mock_pysaj_saj: MagicMock,
|
||||
freezer: FrozenDateTimeFactory,
|
||||
) -> None:
|
||||
"""Test sensor update handles failures."""
|
||||
# Setup read + initial scheduled poll succeed; next poll fails (unknown state).
|
||||
mock_pysaj_saj.read = AsyncMock(side_effect=[True, True, False])
|
||||
|
||||
entry = await setup_integration(hass, mock_config_entry)
|
||||
assert entry.state is ConfigEntryState.LOADED
|
||||
|
||||
await hass.async_block_till_done()
|
||||
|
||||
state = hass.states.get("sensor.saj_current_power")
|
||||
assert state is not None
|
||||
assert state.state == "5000.0"
|
||||
assert mock_pysaj_saj.read.await_count == 2
|
||||
|
||||
freezer.tick(timedelta(seconds=MIN_INTERVAL_SEC + 1))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert mock_pysaj_saj.read.await_count == 3
|
||||
state = hass.states.get("sensor.saj_current_power")
|
||||
assert state is not None
|
||||
assert state.state == STATE_UNKNOWN
|
||||
|
||||
|
||||
async def test_yaml_import_creates_deprecated_issue(
|
||||
hass: HomeAssistant,
|
||||
issue_registry: ir.IssueRegistry,
|
||||
mock_pysaj_saj: MagicMock,
|
||||
) -> None:
|
||||
"""YAML platform triggers import; successful import creates remove-YAML issue."""
|
||||
assert await async_setup_component(
|
||||
hass,
|
||||
"sensor",
|
||||
{"sensor": {"platform": DOMAIN, CONF_HOST: "192.168.1.10"}},
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
issue = issue_registry.async_get_issue("homeassistant", f"deprecated_yaml_{DOMAIN}")
|
||||
assert issue is not None
|
||||
assert issue.issue_domain == DOMAIN
|
||||
|
||||
|
||||
async def test_yaml_import_failure_creates_domain_issue(
|
||||
hass: HomeAssistant,
|
||||
issue_registry: ir.IssueRegistry,
|
||||
mock_pysaj_saj: MagicMock,
|
||||
) -> None:
|
||||
"""YAML import failure creates an integration issue explaining the error."""
|
||||
mock_pysaj_saj.read.side_effect = pysaj.UnexpectedResponseException("bad")
|
||||
assert await async_setup_component(
|
||||
hass,
|
||||
"sensor",
|
||||
{"sensor": {"platform": DOMAIN, CONF_HOST: "192.168.1.10"}},
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
issue = issue_registry.async_get_issue(
|
||||
DOMAIN, "deprecated_yaml_import_issue_cannot_connect"
|
||||
)
|
||||
assert issue is not None
|
||||
Reference in New Issue
Block a user