Use attr_device_info and add init tests for nut (#57725)
parent
bcd431e848
commit
fe0291012c
|
@ -7,6 +7,9 @@ from pynut2.nut2 import PyNUTClient, PyNUTError
|
|||
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import (
|
||||
ATTR_MANUFACTURER,
|
||||
ATTR_MODEL,
|
||||
ATTR_SW_VERSION,
|
||||
CONF_ALIAS,
|
||||
CONF_HOST,
|
||||
CONF_PASSWORD,
|
||||
|
@ -24,9 +27,6 @@ from .const import (
|
|||
DOMAIN,
|
||||
PLATFORMS,
|
||||
PYNUT_DATA,
|
||||
PYNUT_FIRMWARE,
|
||||
PYNUT_MANUFACTURER,
|
||||
PYNUT_MODEL,
|
||||
PYNUT_UNIQUE_ID,
|
||||
UNDO_UPDATE_LISTENER,
|
||||
)
|
||||
|
@ -78,7 +78,6 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
|||
undo_listener = entry.add_update_listener(_async_update_listener)
|
||||
|
||||
unique_id = _unique_id_from_status(status)
|
||||
|
||||
if unique_id is None:
|
||||
unique_id = entry.entry_id
|
||||
|
||||
|
@ -87,9 +86,6 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
|||
COORDINATOR: coordinator,
|
||||
PYNUT_DATA: data,
|
||||
PYNUT_UNIQUE_ID: unique_id,
|
||||
PYNUT_MANUFACTURER: _manufacturer_from_status(status),
|
||||
PYNUT_MODEL: _model_from_status(status),
|
||||
PYNUT_FIRMWARE: _firmware_from_status(status),
|
||||
UNDO_UPDATE_LISTENER: undo_listener,
|
||||
}
|
||||
|
||||
|
@ -185,6 +181,7 @@ class PyNUTData:
|
|||
self._client = PyNUTClient(self._host, port, username, password, 5, False)
|
||||
self.ups_list = None
|
||||
self._status = None
|
||||
self._device_info = None
|
||||
|
||||
@property
|
||||
def status(self):
|
||||
|
@ -196,6 +193,11 @@ class PyNUTData:
|
|||
"""Return the name of the ups."""
|
||||
return self._alias
|
||||
|
||||
@property
|
||||
def device_info(self):
|
||||
"""Return the device info for the ups."""
|
||||
return self._device_info or {}
|
||||
|
||||
def _get_alias(self):
|
||||
"""Get the ups alias from NUT."""
|
||||
try:
|
||||
|
@ -211,6 +213,23 @@ class PyNUTData:
|
|||
self.ups_list = ups_list
|
||||
return list(ups_list)[0]
|
||||
|
||||
def _get_device_info(self):
|
||||
"""Get the ups device info from NUT."""
|
||||
if not self._status:
|
||||
return None
|
||||
|
||||
manufacturer = _manufacturer_from_status(self._status)
|
||||
model = _model_from_status(self._status)
|
||||
firmware = _firmware_from_status(self._status)
|
||||
device_info = {}
|
||||
if model:
|
||||
device_info[ATTR_MODEL] = model
|
||||
if manufacturer:
|
||||
device_info[ATTR_MANUFACTURER] = manufacturer
|
||||
if firmware:
|
||||
device_info[ATTR_SW_VERSION] = firmware
|
||||
return device_info
|
||||
|
||||
def _get_status(self):
|
||||
"""Get the ups status from NUT."""
|
||||
if self._alias is None:
|
||||
|
@ -222,6 +241,8 @@ class PyNUTData:
|
|||
_LOGGER.debug("Error getting NUT vars for host %s: %s", self._host, err)
|
||||
return None
|
||||
|
||||
def update(self, **kwargs):
|
||||
def update(self):
|
||||
"""Fetch the latest status from NUT."""
|
||||
self._status = self._get_status()
|
||||
if self._device_info is None:
|
||||
self._device_info = self._get_device_info()
|
||||
|
|
|
@ -44,9 +44,6 @@ DEFAULT_SCAN_INTERVAL = 60
|
|||
|
||||
PYNUT_DATA = "data"
|
||||
PYNUT_UNIQUE_ID = "unique_id"
|
||||
PYNUT_MANUFACTURER = "manufacturer"
|
||||
PYNUT_MODEL = "model"
|
||||
PYNUT_FIRMWARE = "firmware"
|
||||
|
||||
SENSOR_TYPES: Final[dict[str, SensorEntityDescription]] = {
|
||||
"ups.status.display": SensorEntityDescription(
|
||||
|
|
|
@ -5,7 +5,12 @@ import logging
|
|||
|
||||
from homeassistant.components.nut import PyNUTData
|
||||
from homeassistant.components.sensor import SensorEntity, SensorEntityDescription
|
||||
from homeassistant.const import CONF_RESOURCES, STATE_UNKNOWN
|
||||
from homeassistant.const import (
|
||||
ATTR_IDENTIFIERS,
|
||||
ATTR_NAME,
|
||||
CONF_RESOURCES,
|
||||
STATE_UNKNOWN,
|
||||
)
|
||||
from homeassistant.helpers.update_coordinator import (
|
||||
CoordinatorEntity,
|
||||
DataUpdateCoordinator,
|
||||
|
@ -17,9 +22,6 @@ from .const import (
|
|||
KEY_STATUS,
|
||||
KEY_STATUS_DISPLAY,
|
||||
PYNUT_DATA,
|
||||
PYNUT_FIRMWARE,
|
||||
PYNUT_MANUFACTURER,
|
||||
PYNUT_MODEL,
|
||||
PYNUT_UNIQUE_ID,
|
||||
SENSOR_TYPES,
|
||||
STATE_TYPES,
|
||||
|
@ -32,12 +34,9 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
|
|||
"""Set up the NUT sensors."""
|
||||
|
||||
pynut_data = hass.data[DOMAIN][config_entry.entry_id]
|
||||
unique_id = pynut_data[PYNUT_UNIQUE_ID]
|
||||
manufacturer = pynut_data[PYNUT_MANUFACTURER]
|
||||
model = pynut_data[PYNUT_MODEL]
|
||||
firmware = pynut_data[PYNUT_FIRMWARE]
|
||||
coordinator = pynut_data[COORDINATOR]
|
||||
data = pynut_data[PYNUT_DATA]
|
||||
unique_id = pynut_data[PYNUT_UNIQUE_ID]
|
||||
status = coordinator.data
|
||||
|
||||
enabled_resources = [
|
||||
|
@ -52,12 +51,9 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
|
|||
entities = [
|
||||
NUTSensor(
|
||||
coordinator,
|
||||
data,
|
||||
SENSOR_TYPES[sensor_type],
|
||||
data,
|
||||
unique_id,
|
||||
manufacturer,
|
||||
model,
|
||||
firmware,
|
||||
sensor_type in enabled_resources,
|
||||
)
|
||||
for sensor_type in resources
|
||||
|
@ -72,41 +68,24 @@ class NUTSensor(CoordinatorEntity, SensorEntity):
|
|||
def __init__(
|
||||
self,
|
||||
coordinator: DataUpdateCoordinator,
|
||||
data: PyNUTData,
|
||||
sensor_description: SensorEntityDescription,
|
||||
data: PyNUTData,
|
||||
unique_id: str,
|
||||
manufacturer: str | None,
|
||||
model: str | None,
|
||||
firmware: str | None,
|
||||
enabled_default: bool,
|
||||
) -> None:
|
||||
"""Initialize the sensor."""
|
||||
super().__init__(coordinator)
|
||||
self.entity_description = sensor_description
|
||||
self._manufacturer = manufacturer
|
||||
self._firmware = firmware
|
||||
self._model = model
|
||||
self._device_name = data.name.title()
|
||||
self._unique_id = unique_id
|
||||
|
||||
device_name = data.name.title()
|
||||
self._attr_entity_registry_enabled_default = enabled_default
|
||||
self._attr_name = f"{self._device_name} {sensor_description.name}"
|
||||
self._attr_name = f"{device_name} {sensor_description.name}"
|
||||
self._attr_unique_id = f"{unique_id}_{sensor_description.key}"
|
||||
|
||||
@property
|
||||
def device_info(self):
|
||||
"""Device info for the ups."""
|
||||
device_info = {
|
||||
"identifiers": {(DOMAIN, self._unique_id)},
|
||||
"name": self._device_name,
|
||||
self._attr_device_info = {
|
||||
ATTR_IDENTIFIERS: {(DOMAIN, unique_id)},
|
||||
ATTR_NAME: device_name,
|
||||
}
|
||||
if self._model:
|
||||
device_info["model"] = self._model
|
||||
if self._manufacturer:
|
||||
device_info["manufacturer"] = self._manufacturer
|
||||
if self._firmware:
|
||||
device_info["sw_version"] = self._firmware
|
||||
return device_info
|
||||
self._attr_device_info.update(data.device_info)
|
||||
|
||||
@property
|
||||
def native_value(self):
|
||||
|
|
|
@ -0,0 +1,72 @@
|
|||
"""Test init of Nut integration."""
|
||||
from unittest.mock import patch
|
||||
|
||||
from homeassistant.components.nut.const import DOMAIN
|
||||
from homeassistant.config_entries import ConfigEntryState
|
||||
from homeassistant.const import CONF_HOST, CONF_PORT, CONF_RESOURCES, STATE_UNAVAILABLE
|
||||
|
||||
from .util import _get_mock_pynutclient
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
|
||||
async def test_async_setup_entry(hass):
|
||||
"""Test a successful setup entry."""
|
||||
entry = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
data={
|
||||
CONF_HOST: "mock",
|
||||
CONF_PORT: "mock",
|
||||
CONF_RESOURCES: ["ups.status"],
|
||||
},
|
||||
)
|
||||
entry.add_to_hass(hass)
|
||||
|
||||
mock_pynut = _get_mock_pynutclient(
|
||||
list_ups={"ups1": "UPS 1"}, list_vars={"ups.status": "OL"}
|
||||
)
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.nut.PyNUTClient",
|
||||
return_value=mock_pynut,
|
||||
):
|
||||
await hass.config_entries.async_setup(entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert len(hass.config_entries.async_entries(DOMAIN)) == 1
|
||||
assert entry.state is ConfigEntryState.LOADED
|
||||
|
||||
state = hass.states.get("sensor.ups1_status_data")
|
||||
assert state is not None
|
||||
assert state.state != STATE_UNAVAILABLE
|
||||
assert state.state == "OL"
|
||||
|
||||
assert await hass.config_entries.async_unload(entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert entry.state is ConfigEntryState.NOT_LOADED
|
||||
assert not hass.data.get(DOMAIN)
|
||||
|
||||
|
||||
async def test_config_not_ready(hass):
|
||||
"""Test for setup failure if connection to broker is missing."""
|
||||
entry = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
data={
|
||||
CONF_HOST: "mock",
|
||||
CONF_PORT: "mock",
|
||||
CONF_RESOURCES: ["ups.status"],
|
||||
},
|
||||
)
|
||||
entry.add_to_hass(hass)
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.nut.PyNUTClient.list_ups",
|
||||
return_value=["ups1"],
|
||||
), patch(
|
||||
"homeassistant.components.nut.PyNUTClient.list_vars",
|
||||
side_effect=ConnectionResetError,
|
||||
):
|
||||
await hass.config_entries.async_setup(entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
assert entry.state is ConfigEntryState.SETUP_RETRY
|
Loading…
Reference in New Issue