mirror of
https://github.com/home-assistant/core.git
synced 2025-04-26 10:17:51 +00:00
Add config flow for nut (#33457)
* Convert nut to config flow * Add a test for importing * lint * Address review items (part 1) * Address review items (part 1) * Cleanup unique id handling * Update tests for new naming scheme * No unique id, no device_info * Remove sensor types * Update tests to use resources that still exist
This commit is contained in:
J. Nick Koston
GitHub
parent
2cfa0af532
commit
cc443ff37a
@ -261,6 +261,7 @@ homeassistant/components/nsw_fuel_station/* @nickw444
|
||||
homeassistant/components/nsw_rural_fire_service_feed/* @exxamalte
|
||||
homeassistant/components/nuheat/* @bdraco
|
||||
homeassistant/components/nuki/* @pvizeli
|
||||
homeassistant/components/nut/* @bdraco
|
||||
homeassistant/components/nws/* @MatthewFlamm
|
||||
homeassistant/components/nzbget/* @chriscla
|
||||
homeassistant/components/obihai/* @dshokouhi
|
||||
|
||||
38
homeassistant/components/nut/.translations/en.json
Normal file
38
homeassistant/components/nut/.translations/en.json
Normal file
@ -0,0 +1,38 @@
|
||||
{
|
||||
"config": {
|
||||
"title": "Network UPS Tools (NUT)",
|
||||
"step": {
|
||||
"user": {
|
||||
"title": "Connect to the NUT server",
|
||||
"description": "If there are multiple UPSs attached to the NUT server, enter the name UPS to query in the 'Alias' field.",
|
||||
"data": {
|
||||
"name": "Name",
|
||||
"host": "Host",
|
||||
"port": "Port",
|
||||
"alias": "Alias",
|
||||
"username": "Username",
|
||||
"password": "Password",
|
||||
"resources": "Resources"
|
||||
}
|
||||
}
|
||||
},
|
||||
"error": {
|
||||
"cannot_connect": "Failed to connect, please try again",
|
||||
"unknown": "Unexpected error"
|
||||
},
|
||||
"abort": {
|
||||
"already_configured": "Device is already configured"
|
||||
}
|
||||
},
|
||||
"options": {
|
||||
"step": {
|
||||
"init": {
|
||||
"description": "Choose Sensor Resources",
|
||||
"data": {
|
||||
"resources": "Resources"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@ -1 +1,210 @@
|
||||
"""The nut component."""
|
||||
import asyncio
|
||||
import logging
|
||||
|
||||
from pynut2.nut2 import PyNUTClient, PyNUTError
|
||||
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import (
|
||||
CONF_ALIAS,
|
||||
CONF_HOST,
|
||||
CONF_PASSWORD,
|
||||
CONF_PORT,
|
||||
CONF_RESOURCES,
|
||||
CONF_USERNAME,
|
||||
)
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.exceptions import ConfigEntryNotReady
|
||||
|
||||
from .const import (
|
||||
DOMAIN,
|
||||
PLATFORMS,
|
||||
PYNUT_DATA,
|
||||
PYNUT_FIRMWARE,
|
||||
PYNUT_MANUFACTURER,
|
||||
PYNUT_MODEL,
|
||||
PYNUT_STATUS,
|
||||
PYNUT_UNIQUE_ID,
|
||||
)
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
async def async_setup(hass: HomeAssistant, config: dict):
|
||||
"""Set up the Network UPS Tools (NUT) component."""
|
||||
hass.data.setdefault(DOMAIN, {})
|
||||
|
||||
return True
|
||||
|
||||
|
||||
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry):
|
||||
"""Set up Network UPS Tools (NUT) from a config entry."""
|
||||
|
||||
config = entry.data
|
||||
host = config[CONF_HOST]
|
||||
port = config[CONF_PORT]
|
||||
|
||||
alias = config.get(CONF_ALIAS)
|
||||
username = config.get(CONF_USERNAME)
|
||||
password = config.get(CONF_PASSWORD)
|
||||
|
||||
data = PyNUTData(host, port, alias, username, password)
|
||||
|
||||
status = await hass.async_add_executor_job(pynutdata_status, data)
|
||||
|
||||
if not status:
|
||||
_LOGGER.error("NUT Sensor has no data, unable to set up")
|
||||
raise ConfigEntryNotReady
|
||||
|
||||
_LOGGER.debug("NUT Sensors Available: %s", status)
|
||||
|
||||
hass.data[DOMAIN][entry.entry_id] = {
|
||||
PYNUT_DATA: data,
|
||||
PYNUT_STATUS: status,
|
||||
PYNUT_UNIQUE_ID: _unique_id_from_status(status),
|
||||
PYNUT_MANUFACTURER: _manufacturer_from_status(status),
|
||||
PYNUT_MODEL: _model_from_status(status),
|
||||
PYNUT_FIRMWARE: _firmware_from_status(status),
|
||||
}
|
||||
|
||||
entry.add_update_listener(_async_update_listener)
|
||||
|
||||
for component in PLATFORMS:
|
||||
hass.async_create_task(
|
||||
hass.config_entries.async_forward_entry_setup(entry, component)
|
||||
)
|
||||
|
||||
return True
|
||||
|
||||
|
||||
async def _async_update_listener(hass: HomeAssistant, entry: ConfigEntry):
|
||||
"""Handle options update."""
|
||||
await hass.config_entries.async_reload(entry.entry_id)
|
||||
|
||||
|
||||
def _manufacturer_from_status(status):
|
||||
"""Find the best manufacturer value from the status."""
|
||||
return (
|
||||
status.get("device.mfr")
|
||||
or status.get("ups.mfr")
|
||||
or status.get("ups.vendorid")
|
||||
or status.get("driver.version.data")
|
||||
)
|
||||
|
||||
|
||||
def _model_from_status(status):
|
||||
"""Find the best model value from the status."""
|
||||
return (
|
||||
status.get("device.model")
|
||||
or status.get("ups.model")
|
||||
or status.get("ups.productid")
|
||||
)
|
||||
|
||||
|
||||
def _firmware_from_status(status):
|
||||
"""Find the best firmware value from the status."""
|
||||
return status.get("ups.firmware") or status.get("ups.firmware.aux")
|
||||
|
||||
|
||||
def _serial_from_status(status):
|
||||
"""Find the best serialvalue from the status."""
|
||||
serial = status.get("device.serial") or status.get("ups.serial")
|
||||
if serial and serial == "unknown":
|
||||
return None
|
||||
return serial
|
||||
|
||||
|
||||
def _unique_id_from_status(status):
|
||||
"""Find the best unique id value from the status."""
|
||||
serial = _serial_from_status(status)
|
||||
# We must have a serial for this to be unique
|
||||
if not serial:
|
||||
return None
|
||||
|
||||
manufacturer = _manufacturer_from_status(status)
|
||||
model = _model_from_status(status)
|
||||
|
||||
unique_id_group = []
|
||||
if manufacturer:
|
||||
unique_id_group.append(manufacturer)
|
||||
if model:
|
||||
unique_id_group.append(model)
|
||||
if serial:
|
||||
unique_id_group.append(serial)
|
||||
return "_".join(unique_id_group)
|
||||
|
||||
|
||||
def find_resources_in_config_entry(config_entry):
|
||||
"""Find the configured resources in the config entry."""
|
||||
if CONF_RESOURCES in config_entry.options:
|
||||
return config_entry.options[CONF_RESOURCES]
|
||||
return config_entry.data[CONF_RESOURCES]
|
||||
|
||||
|
||||
def pynutdata_status(data):
|
||||
"""Wrap for data update as a callable."""
|
||||
return data.status
|
||||
|
||||
|
||||
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry):
|
||||
"""Unload a config entry."""
|
||||
unload_ok = all(
|
||||
await asyncio.gather(
|
||||
*[
|
||||
hass.config_entries.async_forward_entry_unload(entry, component)
|
||||
for component in PLATFORMS
|
||||
]
|
||||
)
|
||||
)
|
||||
if unload_ok:
|
||||
hass.data[DOMAIN].pop(entry.entry_id)
|
||||
|
||||
return unload_ok
|
||||
|
||||
|
||||
class PyNUTData:
|
||||
"""Stores the data retrieved from NUT.
|
||||
|
||||
For each entity to use, acts as the single point responsible for fetching
|
||||
updates from the server.
|
||||
"""
|
||||
|
||||
def __init__(self, host, port, alias, username, password):
|
||||
"""Initialize the data object."""
|
||||
|
||||
self._host = host
|
||||
self._alias = alias
|
||||
|
||||
# Establish client with persistent=False to open/close connection on
|
||||
# each update call. This is more reliable with async.
|
||||
self._client = PyNUTClient(self._host, port, username, password, 5, False)
|
||||
self._status = None
|
||||
|
||||
@property
|
||||
def status(self):
|
||||
"""Get latest update if throttle allows. Return status."""
|
||||
self.update()
|
||||
return self._status
|
||||
|
||||
def _get_alias(self):
|
||||
"""Get the ups alias from NUT."""
|
||||
try:
|
||||
return next(iter(self._client.list_ups()))
|
||||
except PyNUTError as err:
|
||||
_LOGGER.error("Failure getting NUT ups alias, %s", err)
|
||||
return None
|
||||
|
||||
def _get_status(self):
|
||||
"""Get the ups status from NUT."""
|
||||
if self._alias is None:
|
||||
self._alias = self._get_alias()
|
||||
|
||||
try:
|
||||
return self._client.list_vars(self._alias)
|
||||
except (PyNUTError, ConnectionResetError) as err:
|
||||
_LOGGER.debug("Error getting NUT vars for host %s: %s", self._host, err)
|
||||
return None
|
||||
|
||||
def update(self, **kwargs):
|
||||
"""Fetch the latest status from NUT."""
|
||||
self._status = self._get_status()
|
||||
|
||||
143
homeassistant/components/nut/config_flow.py
Normal file
143
homeassistant/components/nut/config_flow.py
Normal file
@ -0,0 +1,143 @@
|
||||
"""Config flow for Network UPS Tools (NUT) integration."""
|
||||
import logging
|
||||
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant import config_entries, core, exceptions
|
||||
from homeassistant.const import (
|
||||
CONF_ALIAS,
|
||||
CONF_HOST,
|
||||
CONF_NAME,
|
||||
CONF_PASSWORD,
|
||||
CONF_PORT,
|
||||
CONF_RESOURCES,
|
||||
CONF_USERNAME,
|
||||
)
|
||||
from homeassistant.core import callback
|
||||
import homeassistant.helpers.config_validation as cv
|
||||
|
||||
from . import PyNUTData, find_resources_in_config_entry, pynutdata_status
|
||||
from .const import DEFAULT_HOST, DEFAULT_NAME, DEFAULT_PORT, SENSOR_TYPES
|
||||
from .const import DOMAIN # pylint:disable=unused-import
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
SENSOR_DICT = {sensor_id: SENSOR_TYPES[sensor_id][0] for sensor_id in SENSOR_TYPES}
|
||||
|
||||
DATA_SCHEMA = vol.Schema(
|
||||
{
|
||||
vol.Optional(CONF_NAME, default=DEFAULT_NAME): str,
|
||||
vol.Required(CONF_RESOURCES): cv.multi_select(SENSOR_DICT),
|
||||
vol.Optional(CONF_HOST, default=DEFAULT_HOST): str,
|
||||
vol.Optional(CONF_PORT, default=DEFAULT_PORT): int,
|
||||
vol.Optional(CONF_ALIAS): str,
|
||||
vol.Optional(CONF_USERNAME): str,
|
||||
vol.Optional(CONF_PASSWORD): str,
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
async def validate_input(hass: core.HomeAssistant, data):
|
||||
"""Validate the user input allows us to connect.
|
||||
|
||||
Data has the keys from DATA_SCHEMA with values provided by the user.
|
||||
"""
|
||||
|
||||
host = data[CONF_HOST]
|
||||
port = data[CONF_PORT]
|
||||
alias = data.get(CONF_ALIAS)
|
||||
username = data.get(CONF_USERNAME)
|
||||
password = data.get(CONF_PASSWORD)
|
||||
|
||||
data = PyNUTData(host, port, alias, username, password)
|
||||
|
||||
status = await hass.async_add_executor_job(pynutdata_status, data)
|
||||
|
||||
if not status:
|
||||
raise CannotConnect
|
||||
|
||||
return {"title": _format_host_port_alias(host, port, alias)}
|
||||
|
||||
|
||||
def _format_host_port_alias(host, port, alias):
|
||||
"""Format a host, port, and alias so it can be used for comparison or display."""
|
||||
if alias:
|
||||
return f"{alias}@{host}:{port}"
|
||||
return f"{host}:{port}"
|
||||
|
||||
|
||||
class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
||||
"""Handle a config flow for Network UPS Tools (NUT)."""
|
||||
|
||||
VERSION = 1
|
||||
CONNECTION_CLASS = config_entries.CONN_CLASS_LOCAL_POLL
|
||||
|
||||
async def async_step_user(self, user_input=None):
|
||||
"""Handle the initial step."""
|
||||
errors = {}
|
||||
if user_input is not None:
|
||||
if self._host_port_alias_already_configured(
|
||||
user_input[CONF_HOST], user_input[CONF_PORT], user_input.get(CONF_ALIAS)
|
||||
):
|
||||
return self.async_abort(reason="already_configured")
|
||||
try:
|
||||
info = await validate_input(self.hass, user_input)
|
||||
except CannotConnect:
|
||||
errors["base"] = "cannot_connect"
|
||||
except Exception: # pylint: disable=broad-except
|
||||
_LOGGER.exception("Unexpected exception")
|
||||
errors["base"] = "unknown"
|
||||
|
||||
if "base" not in errors:
|
||||
return self.async_create_entry(title=info["title"], data=user_input)
|
||||
|
||||
return self.async_show_form(
|
||||
step_id="user", data_schema=DATA_SCHEMA, errors=errors
|
||||
)
|
||||
|
||||
def _host_port_alias_already_configured(self, host, port, alias):
|
||||
"""See if we already have a nut entry matching user input configured."""
|
||||
existing_host_port_aliases = {
|
||||
_format_host_port_alias(host, port, alias)
|
||||
for entry in self._async_current_entries()
|
||||
}
|
||||
return _format_host_port_alias(host, port, alias) in existing_host_port_aliases
|
||||
|
||||
async def async_step_import(self, user_input):
|
||||
"""Handle import."""
|
||||
return await self.async_step_user(user_input)
|
||||
|
||||
@staticmethod
|
||||
@callback
|
||||
def async_get_options_flow(config_entry):
|
||||
"""Get the options flow for this handler."""
|
||||
return OptionsFlowHandler(config_entry)
|
||||
|
||||
|
||||
class OptionsFlowHandler(config_entries.OptionsFlow):
|
||||
"""Handle a option flow for nut."""
|
||||
|
||||
def __init__(self, config_entry: config_entries.ConfigEntry):
|
||||
"""Initialize options flow."""
|
||||
self.config_entry = config_entry
|
||||
|
||||
async def async_step_init(self, user_input=None):
|
||||
"""Handle options flow."""
|
||||
if user_input is not None:
|
||||
return self.async_create_entry(title="", data=user_input)
|
||||
|
||||
resources = find_resources_in_config_entry(self.config_entry)
|
||||
|
||||
data_schema = vol.Schema(
|
||||
{
|
||||
vol.Required(CONF_RESOURCES, default=resources): cv.multi_select(
|
||||
SENSOR_DICT
|
||||
),
|
||||
}
|
||||
)
|
||||
return self.async_show_form(step_id="init", data_schema=data_schema)
|
||||
|
||||
|
||||
class CannotConnect(exceptions.HomeAssistantError):
|
||||
"""Error to indicate we cannot connect."""
|
||||
125
homeassistant/components/nut/const.py
Normal file
125
homeassistant/components/nut/const.py
Normal file
@ -0,0 +1,125 @@
|
||||
"""The nut component."""
|
||||
from homeassistant.const import POWER_WATT, TEMP_CELSIUS, TIME_SECONDS, UNIT_PERCENTAGE
|
||||
|
||||
DOMAIN = "nut"
|
||||
|
||||
PLATFORMS = ["sensor"]
|
||||
|
||||
|
||||
DEFAULT_NAME = "NUT UPS"
|
||||
DEFAULT_HOST = "localhost"
|
||||
DEFAULT_PORT = 3493
|
||||
|
||||
KEY_STATUS = "ups.status"
|
||||
KEY_STATUS_DISPLAY = "ups.status.display"
|
||||
|
||||
PYNUT_DATA = "data"
|
||||
PYNUT_STATUS = "status"
|
||||
PYNUT_UNIQUE_ID = "unique_id"
|
||||
PYNUT_MANUFACTURER = "manufacturer"
|
||||
PYNUT_MODEL = "model"
|
||||
PYNUT_FIRMWARE = "firmware"
|
||||
|
||||
SENSOR_TYPES = {
|
||||
"ups.status.display": ["Status", "", "mdi:information-outline"],
|
||||
"ups.status": ["Status Data", "", "mdi:information-outline"],
|
||||
"ups.alarm": ["Alarms", "", "mdi:alarm"],
|
||||
"ups.temperature": ["UPS Temperature", TEMP_CELSIUS, "mdi:thermometer"],
|
||||
"ups.load": ["Load", UNIT_PERCENTAGE, "mdi:gauge"],
|
||||
"ups.load.high": ["Overload Setting", UNIT_PERCENTAGE, "mdi:gauge"],
|
||||
"ups.id": ["System identifier", "", "mdi:information-outline"],
|
||||
"ups.delay.start": ["Load Restart Delay", TIME_SECONDS, "mdi:timer"],
|
||||
"ups.delay.reboot": ["UPS Reboot Delay", TIME_SECONDS, "mdi:timer"],
|
||||
"ups.delay.shutdown": ["UPS Shutdown Delay", TIME_SECONDS, "mdi:timer"],
|
||||
"ups.timer.start": ["Load Start Timer", TIME_SECONDS, "mdi:timer"],
|
||||
"ups.timer.reboot": ["Load Reboot Timer", TIME_SECONDS, "mdi:timer"],
|
||||
"ups.timer.shutdown": ["Load Shutdown Timer", TIME_SECONDS, "mdi:timer"],
|
||||
"ups.test.interval": ["Self-Test Interval", TIME_SECONDS, "mdi:timer"],
|
||||
"ups.test.result": ["Self-Test Result", "", "mdi:information-outline"],
|
||||
"ups.test.date": ["Self-Test Date", "", "mdi:calendar"],
|
||||
"ups.display.language": ["Language", "", "mdi:information-outline"],
|
||||
"ups.contacts": ["External Contacts", "", "mdi:information-outline"],
|
||||
"ups.efficiency": ["Efficiency", UNIT_PERCENTAGE, "mdi:gauge"],
|
||||
"ups.power": ["Current Apparent Power", "VA", "mdi:flash"],
|
||||
"ups.power.nominal": ["Nominal Power", "VA", "mdi:flash"],
|
||||
"ups.realpower": ["Current Real Power", POWER_WATT, "mdi:flash"],
|
||||
"ups.realpower.nominal": ["Nominal Real Power", POWER_WATT, "mdi:flash"],
|
||||
"ups.beeper.status": ["Beeper Status", "", "mdi:information-outline"],
|
||||
"ups.type": ["UPS Type", "", "mdi:information-outline"],
|
||||
"ups.watchdog.status": ["Watchdog Status", "", "mdi:information-outline"],
|
||||
"ups.start.auto": ["Start on AC", "", "mdi:information-outline"],
|
||||
"ups.start.battery": ["Start on Battery", "", "mdi:information-outline"],
|
||||
"ups.start.reboot": ["Reboot on Battery", "", "mdi:information-outline"],
|
||||
"ups.shutdown": ["Shutdown Ability", "", "mdi:information-outline"],
|
||||
"battery.charge": ["Battery Charge", UNIT_PERCENTAGE, "mdi:gauge"],
|
||||
"battery.charge.low": ["Low Battery Setpoint", UNIT_PERCENTAGE, "mdi:gauge"],
|
||||
"battery.charge.restart": [
|
||||
"Minimum Battery to Start",
|
||||
UNIT_PERCENTAGE,
|
||||
"mdi:gauge",
|
||||
],
|
||||
"battery.charge.warning": [
|
||||
"Warning Battery Setpoint",
|
||||
UNIT_PERCENTAGE,
|
||||
"mdi:gauge",
|
||||
],
|
||||
"battery.charger.status": ["Charging Status", "", "mdi:information-outline"],
|
||||
"battery.voltage": ["Battery Voltage", "V", "mdi:flash"],
|
||||
"battery.voltage.nominal": ["Nominal Battery Voltage", "V", "mdi:flash"],
|
||||
"battery.voltage.low": ["Low Battery Voltage", "V", "mdi:flash"],
|
||||
"battery.voltage.high": ["High Battery Voltage", "V", "mdi:flash"],
|
||||
"battery.capacity": ["Battery Capacity", "Ah", "mdi:flash"],
|
||||
"battery.current": ["Battery Current", "A", "mdi:flash"],
|
||||
"battery.current.total": ["Total Battery Current", "A", "mdi:flash"],
|
||||
"battery.temperature": ["Battery Temperature", TEMP_CELSIUS, "mdi:thermometer"],
|
||||
"battery.runtime": ["Battery Runtime", TIME_SECONDS, "mdi:timer"],
|
||||
"battery.runtime.low": ["Low Battery Runtime", TIME_SECONDS, "mdi:timer"],
|
||||
"battery.runtime.restart": [
|
||||
"Minimum Battery Runtime to Start",
|
||||
TIME_SECONDS,
|
||||
"mdi:timer",
|
||||
],
|
||||
"battery.alarm.threshold": [
|
||||
"Battery Alarm Threshold",
|
||||
"",
|
||||
"mdi:information-outline",
|
||||
],
|
||||
"battery.date": ["Battery Date", "", "mdi:calendar"],
|
||||
"battery.mfr.date": ["Battery Manuf. Date", "", "mdi:calendar"],
|
||||
"battery.packs": ["Number of Batteries", "", "mdi:information-outline"],
|
||||
"battery.packs.bad": ["Number of Bad Batteries", "", "mdi:information-outline"],
|
||||
"battery.type": ["Battery Chemistry", "", "mdi:information-outline"],
|
||||
"input.sensitivity": ["Input Power Sensitivity", "", "mdi:information-outline"],
|
||||
"input.transfer.low": ["Low Voltage Transfer", "V", "mdi:flash"],
|
||||
"input.transfer.high": ["High Voltage Transfer", "V", "mdi:flash"],
|
||||
"input.transfer.reason": ["Voltage Transfer Reason", "", "mdi:information-outline"],
|
||||
"input.voltage": ["Input Voltage", "V", "mdi:flash"],
|
||||
"input.voltage.nominal": ["Nominal Input Voltage", "V", "mdi:flash"],
|
||||
"input.frequency": ["Input Line Frequency", "hz", "mdi:flash"],
|
||||
"input.frequency.nominal": ["Nominal Input Line Frequency", "hz", "mdi:flash"],
|
||||
"input.frequency.status": ["Input Frequency Status", "", "mdi:information-outline"],
|
||||
"output.current": ["Output Current", "A", "mdi:flash"],
|
||||
"output.current.nominal": ["Nominal Output Current", "A", "mdi:flash"],
|
||||
"output.voltage": ["Output Voltage", "V", "mdi:flash"],
|
||||
"output.voltage.nominal": ["Nominal Output Voltage", "V", "mdi:flash"],
|
||||
"output.frequency": ["Output Frequency", "hz", "mdi:flash"],
|
||||
"output.frequency.nominal": ["Nominal Output Frequency", "hz", "mdi:flash"],
|
||||
}
|
||||
|
||||
STATE_TYPES = {
|
||||
"OL": "Online",
|
||||
"OB": "On Battery",
|
||||
"LB": "Low Battery",
|
||||
"HB": "High Battery",
|
||||
"RB": "Battery Needs Replaced",
|
||||
"CHRG": "Battery Charging",
|
||||
"DISCHRG": "Battery Discharging",
|
||||
"BYPASS": "Bypass Active",
|
||||
"CAL": "Runtime Calibration",
|
||||
"OFF": "Offline",
|
||||
"OVER": "Overloaded",
|
||||
"TRIM": "Trimming Voltage",
|
||||
"BOOST": "Boosting Voltage",
|
||||
"FSD": "Forced Shutdown",
|
||||
"ALARM": "Alarm",
|
||||
}
|
||||
@ -2,7 +2,10 @@
|
||||
"domain": "nut",
|
||||
"name": "Network UPS Tools (NUT)",
|
||||
"documentation": "https://www.home-assistant.io/integrations/nut",
|
||||
"requirements": ["pynut2==2.1.2"],
|
||||
"requirements": [
|
||||
"pynut2==2.1.2"
|
||||
],
|
||||
"dependencies": [],
|
||||
"codeowners": []
|
||||
"codeowners": ["@bdraco"],
|
||||
"config_flow": true
|
||||
}
|
||||
|
||||
@ -2,10 +2,10 @@
|
||||
from datetime import timedelta
|
||||
import logging
|
||||
|
||||
from pynut2.nut2 import PyNUTClient, PyNUTError
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.components.sensor import PLATFORM_SCHEMA
|
||||
from homeassistant.config_entries import SOURCE_IMPORT
|
||||
from homeassistant.const import (
|
||||
ATTR_STATE,
|
||||
CONF_ALIAS,
|
||||
@ -15,140 +15,33 @@ from homeassistant.const import (
|
||||
CONF_PORT,
|
||||
CONF_RESOURCES,
|
||||
CONF_USERNAME,
|
||||
POWER_WATT,
|
||||
STATE_UNKNOWN,
|
||||
TEMP_CELSIUS,
|
||||
TIME_SECONDS,
|
||||
UNIT_PERCENTAGE,
|
||||
)
|
||||
from homeassistant.exceptions import PlatformNotReady
|
||||
import homeassistant.helpers.config_validation as cv
|
||||
from homeassistant.helpers.entity import Entity
|
||||
|
||||
from .const import (
|
||||
DEFAULT_HOST,
|
||||
DEFAULT_NAME,
|
||||
DEFAULT_PORT,
|
||||
DOMAIN,
|
||||
KEY_STATUS,
|
||||
KEY_STATUS_DISPLAY,
|
||||
PYNUT_DATA,
|
||||
PYNUT_FIRMWARE,
|
||||
PYNUT_MANUFACTURER,
|
||||
PYNUT_MODEL,
|
||||
PYNUT_STATUS,
|
||||
PYNUT_UNIQUE_ID,
|
||||
SENSOR_TYPES,
|
||||
STATE_TYPES,
|
||||
)
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
DEFAULT_NAME = "NUT UPS"
|
||||
DEFAULT_HOST = "localhost"
|
||||
DEFAULT_PORT = 3493
|
||||
|
||||
KEY_STATUS = "ups.status"
|
||||
KEY_STATUS_DISPLAY = "ups.status.display"
|
||||
|
||||
SCAN_INTERVAL = timedelta(seconds=60)
|
||||
|
||||
SENSOR_TYPES = {
|
||||
"ups.status.display": ["Status", "", "mdi:information-outline"],
|
||||
"ups.status": ["Status Data", "", "mdi:information-outline"],
|
||||
"ups.alarm": ["Alarms", "", "mdi:alarm"],
|
||||
"ups.time": ["Internal Time", "", "mdi:calendar-clock"],
|
||||
"ups.date": ["Internal Date", "", "mdi:calendar"],
|
||||
"ups.model": ["Model", "", "mdi:information-outline"],
|
||||
"ups.mfr": ["Manufacturer", "", "mdi:information-outline"],
|
||||
"ups.mfr.date": ["Manufacture Date", "", "mdi:calendar"],
|
||||
"ups.serial": ["Serial Number", "", "mdi:information-outline"],
|
||||
"ups.vendorid": ["Vendor ID", "", "mdi:information-outline"],
|
||||
"ups.productid": ["Product ID", "", "mdi:information-outline"],
|
||||
"ups.firmware": ["Firmware Version", "", "mdi:information-outline"],
|
||||
"ups.firmware.aux": ["Firmware Version 2", "", "mdi:information-outline"],
|
||||
"ups.temperature": ["UPS Temperature", TEMP_CELSIUS, "mdi:thermometer"],
|
||||
"ups.load": ["Load", UNIT_PERCENTAGE, "mdi:gauge"],
|
||||
"ups.load.high": ["Overload Setting", UNIT_PERCENTAGE, "mdi:gauge"],
|
||||
"ups.id": ["System identifier", "", "mdi:information-outline"],
|
||||
"ups.delay.start": ["Load Restart Delay", TIME_SECONDS, "mdi:timer"],
|
||||
"ups.delay.reboot": ["UPS Reboot Delay", TIME_SECONDS, "mdi:timer"],
|
||||
"ups.delay.shutdown": ["UPS Shutdown Delay", TIME_SECONDS, "mdi:timer"],
|
||||
"ups.timer.start": ["Load Start Timer", TIME_SECONDS, "mdi:timer"],
|
||||
"ups.timer.reboot": ["Load Reboot Timer", TIME_SECONDS, "mdi:timer"],
|
||||
"ups.timer.shutdown": ["Load Shutdown Timer", TIME_SECONDS, "mdi:timer"],
|
||||
"ups.test.interval": ["Self-Test Interval", TIME_SECONDS, "mdi:timer"],
|
||||
"ups.test.result": ["Self-Test Result", "", "mdi:information-outline"],
|
||||
"ups.test.date": ["Self-Test Date", "", "mdi:calendar"],
|
||||
"ups.display.language": ["Language", "", "mdi:information-outline"],
|
||||
"ups.contacts": ["External Contacts", "", "mdi:information-outline"],
|
||||
"ups.efficiency": ["Efficiency", UNIT_PERCENTAGE, "mdi:gauge"],
|
||||
"ups.power": ["Current Apparent Power", "VA", "mdi:flash"],
|
||||
"ups.power.nominal": ["Nominal Power", "VA", "mdi:flash"],
|
||||
"ups.realpower": ["Current Real Power", POWER_WATT, "mdi:flash"],
|
||||
"ups.realpower.nominal": ["Nominal Real Power", POWER_WATT, "mdi:flash"],
|
||||
"ups.beeper.status": ["Beeper Status", "", "mdi:information-outline"],
|
||||
"ups.type": ["UPS Type", "", "mdi:information-outline"],
|
||||
"ups.watchdog.status": ["Watchdog Status", "", "mdi:information-outline"],
|
||||
"ups.start.auto": ["Start on AC", "", "mdi:information-outline"],
|
||||
"ups.start.battery": ["Start on Battery", "", "mdi:information-outline"],
|
||||
"ups.start.reboot": ["Reboot on Battery", "", "mdi:information-outline"],
|
||||
"ups.shutdown": ["Shutdown Ability", "", "mdi:information-outline"],
|
||||
"battery.charge": ["Battery Charge", UNIT_PERCENTAGE, "mdi:gauge"],
|
||||
"battery.charge.low": ["Low Battery Setpoint", UNIT_PERCENTAGE, "mdi:gauge"],
|
||||
"battery.charge.restart": [
|
||||
"Minimum Battery to Start",
|
||||
UNIT_PERCENTAGE,
|
||||
"mdi:gauge",
|
||||
],
|
||||
"battery.charge.warning": [
|
||||
"Warning Battery Setpoint",
|
||||
UNIT_PERCENTAGE,
|
||||
"mdi:gauge",
|
||||
],
|
||||
"battery.charger.status": ["Charging Status", "", "mdi:information-outline"],
|
||||
"battery.voltage": ["Battery Voltage", "V", "mdi:flash"],
|
||||
"battery.voltage.nominal": ["Nominal Battery Voltage", "V", "mdi:flash"],
|
||||
"battery.voltage.low": ["Low Battery Voltage", "V", "mdi:flash"],
|
||||
"battery.voltage.high": ["High Battery Voltage", "V", "mdi:flash"],
|
||||
"battery.capacity": ["Battery Capacity", "Ah", "mdi:flash"],
|
||||
"battery.current": ["Battery Current", "A", "mdi:flash"],
|
||||
"battery.current.total": ["Total Battery Current", "A", "mdi:flash"],
|
||||
"battery.temperature": ["Battery Temperature", TEMP_CELSIUS, "mdi:thermometer"],
|
||||
"battery.runtime": ["Battery Runtime", TIME_SECONDS, "mdi:timer"],
|
||||
"battery.runtime.low": ["Low Battery Runtime", TIME_SECONDS, "mdi:timer"],
|
||||
"battery.runtime.restart": [
|
||||
"Minimum Battery Runtime to Start",
|
||||
TIME_SECONDS,
|
||||
"mdi:timer",
|
||||
],
|
||||
"battery.alarm.threshold": [
|
||||
"Battery Alarm Threshold",
|
||||
"",
|
||||
"mdi:information-outline",
|
||||
],
|
||||
"battery.date": ["Battery Date", "", "mdi:calendar"],
|
||||
"battery.mfr.date": ["Battery Manuf. Date", "", "mdi:calendar"],
|
||||
"battery.packs": ["Number of Batteries", "", "mdi:information-outline"],
|
||||
"battery.packs.bad": ["Number of Bad Batteries", "", "mdi:information-outline"],
|
||||
"battery.type": ["Battery Chemistry", "", "mdi:information-outline"],
|
||||
"input.sensitivity": ["Input Power Sensitivity", "", "mdi:information-outline"],
|
||||
"input.transfer.low": ["Low Voltage Transfer", "V", "mdi:flash"],
|
||||
"input.transfer.high": ["High Voltage Transfer", "V", "mdi:flash"],
|
||||
"input.transfer.reason": ["Voltage Transfer Reason", "", "mdi:information-outline"],
|
||||
"input.voltage": ["Input Voltage", "V", "mdi:flash"],
|
||||
"input.voltage.nominal": ["Nominal Input Voltage", "V", "mdi:flash"],
|
||||
"input.frequency": ["Input Line Frequency", "hz", "mdi:flash"],
|
||||
"input.frequency.nominal": ["Nominal Input Line Frequency", "hz", "mdi:flash"],
|
||||
"input.frequency.status": ["Input Frequency Status", "", "mdi:information-outline"],
|
||||
"output.current": ["Output Current", "A", "mdi:flash"],
|
||||
"output.current.nominal": ["Nominal Output Current", "A", "mdi:flash"],
|
||||
"output.voltage": ["Output Voltage", "V", "mdi:flash"],
|
||||
"output.voltage.nominal": ["Nominal Output Voltage", "V", "mdi:flash"],
|
||||
"output.frequency": ["Output Frequency", "hz", "mdi:flash"],
|
||||
"output.frequency.nominal": ["Nominal Output Frequency", "hz", "mdi:flash"],
|
||||
}
|
||||
|
||||
STATE_TYPES = {
|
||||
"OL": "Online",
|
||||
"OB": "On Battery",
|
||||
"LB": "Low Battery",
|
||||
"HB": "High Battery",
|
||||
"RB": "Battery Needs Replaced",
|
||||
"CHRG": "Battery Charging",
|
||||
"DISCHRG": "Battery Discharging",
|
||||
"BYPASS": "Bypass Active",
|
||||
"CAL": "Runtime Calibration",
|
||||
"OFF": "Offline",
|
||||
"OVER": "Overloaded",
|
||||
"TRIM": "Trimming Voltage",
|
||||
"BOOST": "Boosting Voltage",
|
||||
"FSD": "Forced Shutdown",
|
||||
"ALARM": "Alarm",
|
||||
}
|
||||
|
||||
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
|
||||
{
|
||||
@ -164,34 +57,48 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
|
||||
|
||||
|
||||
def setup_platform(hass, config, add_entities, discovery_info=None):
|
||||
"""Import the platform into a config entry."""
|
||||
|
||||
hass.async_create_task(
|
||||
hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_IMPORT}, data=config
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
async def async_setup_entry(hass, config_entry, async_add_entities):
|
||||
"""Set up the NUT sensors."""
|
||||
name = config[CONF_NAME]
|
||||
host = config[CONF_HOST]
|
||||
port = config[CONF_PORT]
|
||||
|
||||
alias = config.get(CONF_ALIAS)
|
||||
username = config.get(CONF_USERNAME)
|
||||
password = config.get(CONF_PASSWORD)
|
||||
|
||||
data = PyNUTData(host, port, alias, username, password)
|
||||
|
||||
if data.status is None:
|
||||
_LOGGER.error("NUT Sensor has no data, unable to set up")
|
||||
raise PlatformNotReady
|
||||
|
||||
_LOGGER.debug("NUT Sensors Available: %s", data.status)
|
||||
config = config_entry.data
|
||||
pynut_data = hass.data[DOMAIN][config_entry.entry_id]
|
||||
data = pynut_data[PYNUT_DATA]
|
||||
status = pynut_data[PYNUT_STATUS]
|
||||
unique_id = pynut_data[PYNUT_UNIQUE_ID]
|
||||
manufacturer = pynut_data[PYNUT_MANUFACTURER]
|
||||
model = pynut_data[PYNUT_MODEL]
|
||||
firmware = pynut_data[PYNUT_FIRMWARE]
|
||||
|
||||
entities = []
|
||||
|
||||
for resource in config[CONF_RESOURCES]:
|
||||
name = config[CONF_NAME]
|
||||
if CONF_RESOURCES in config_entry.options:
|
||||
resources = config_entry.options[CONF_RESOURCES]
|
||||
else:
|
||||
resources = config_entry.data[CONF_RESOURCES]
|
||||
|
||||
for resource in resources:
|
||||
sensor_type = resource.lower()
|
||||
|
||||
# Display status is a special case that falls back to the status value
|
||||
# of the UPS instead.
|
||||
if sensor_type in data.status or (
|
||||
sensor_type == KEY_STATUS_DISPLAY and KEY_STATUS in data.status
|
||||
if sensor_type in status or (
|
||||
sensor_type == KEY_STATUS_DISPLAY and KEY_STATUS in status
|
||||
):
|
||||
entities.append(NUTSensor(name, data, sensor_type))
|
||||
entities.append(
|
||||
NUTSensor(
|
||||
name, data, sensor_type, unique_id, manufacturer, model, firmware
|
||||
)
|
||||
)
|
||||
else:
|
||||
_LOGGER.warning(
|
||||
"Sensor type: %s does not appear in the NUT status "
|
||||
@ -199,30 +106,53 @@ def setup_platform(hass, config, add_entities, discovery_info=None):
|
||||
sensor_type,
|
||||
)
|
||||
|
||||
try:
|
||||
data.update(no_throttle=True)
|
||||
except data.pynuterror as err:
|
||||
_LOGGER.error(
|
||||
"Failure while testing NUT status retrieval. Cannot continue setup: %s", err
|
||||
)
|
||||
raise PlatformNotReady
|
||||
|
||||
add_entities(entities, True)
|
||||
async_add_entities(entities, True)
|
||||
|
||||
|
||||
class NUTSensor(Entity):
|
||||
"""Representation of a sensor entity for NUT status values."""
|
||||
|
||||
def __init__(self, name, data, sensor_type):
|
||||
def __init__(
|
||||
self, name, data, sensor_type, unique_id, manufacturer, model, firmware
|
||||
):
|
||||
"""Initialize the sensor."""
|
||||
self._data = data
|
||||
self.type = sensor_type
|
||||
self._type = sensor_type
|
||||
self._manufacturer = manufacturer
|
||||
self._firmware = firmware
|
||||
self._model = model
|
||||
self._device_name = name
|
||||
self._name = "{} {}".format(name, SENSOR_TYPES[sensor_type][0])
|
||||
self._unit = SENSOR_TYPES[sensor_type][1]
|
||||
self._state = None
|
||||
self._unique_id = unique_id
|
||||
self._display_state = None
|
||||
self._available = False
|
||||
|
||||
@property
|
||||
def device_info(self):
|
||||
"""Device info for the ups."""
|
||||
if not self._unique_id:
|
||||
return None
|
||||
device_info = {
|
||||
"identifiers": {(DOMAIN, self._unique_id)},
|
||||
"name": self._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
|
||||
|
||||
@property
|
||||
def unique_id(self):
|
||||
"""Sensor Unique id."""
|
||||
if not self._unique_id:
|
||||
return None
|
||||
return f"{self._unique_id}_{self._type}"
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
"""Return the name of the UPS sensor."""
|
||||
@ -231,7 +161,7 @@ class NUTSensor(Entity):
|
||||
@property
|
||||
def icon(self):
|
||||
"""Icon to use in the frontend, if any."""
|
||||
return SENSOR_TYPES[self.type][2]
|
||||
return SENSOR_TYPES[self._type][2]
|
||||
|
||||
@property
|
||||
def state(self):
|
||||
@ -265,12 +195,12 @@ class NUTSensor(Entity):
|
||||
self._display_state = _format_display_state(status)
|
||||
# In case of the display status sensor, keep a human-readable form
|
||||
# as the sensor state.
|
||||
if self.type == KEY_STATUS_DISPLAY:
|
||||
if self._type == KEY_STATUS_DISPLAY:
|
||||
self._state = self._display_state
|
||||
elif self.type not in status:
|
||||
elif self._type not in status:
|
||||
self._state = None
|
||||
else:
|
||||
self._state = status[self.type]
|
||||
self._state = status[self._type]
|
||||
|
||||
|
||||
def _format_display_state(status):
|
||||
@ -281,58 +211,3 @@ def _format_display_state(status):
|
||||
return " ".join(STATE_TYPES[state] for state in status[KEY_STATUS].split())
|
||||
except KeyError:
|
||||
return STATE_UNKNOWN
|
||||
|
||||
|
||||
class PyNUTData:
|
||||
"""Stores the data retrieved from NUT.
|
||||
|
||||
For each entity to use, acts as the single point responsible for fetching
|
||||
updates from the server.
|
||||
"""
|
||||
|
||||
def __init__(self, host, port, alias, username, password):
|
||||
"""Initialize the data object."""
|
||||
|
||||
self._host = host
|
||||
self._port = port
|
||||
self._alias = alias
|
||||
self._username = username
|
||||
self._password = password
|
||||
|
||||
self.pynuterror = PyNUTError
|
||||
# Establish client with persistent=False to open/close connection on
|
||||
# each update call. This is more reliable with async.
|
||||
self._client = PyNUTClient(
|
||||
self._host, self._port, self._username, self._password, 5, False
|
||||
)
|
||||
|
||||
self._status = None
|
||||
|
||||
@property
|
||||
def status(self):
|
||||
"""Get latest update if throttle allows. Return status."""
|
||||
self.update()
|
||||
return self._status
|
||||
|
||||
def _get_alias(self):
|
||||
"""Get the ups alias from NUT."""
|
||||
try:
|
||||
return next(iter(self._client.list_ups()))
|
||||
except self.pynuterror as err:
|
||||
_LOGGER.error("Failure getting NUT ups alias, %s", err)
|
||||
return None
|
||||
|
||||
def _get_status(self):
|
||||
"""Get the ups status from NUT."""
|
||||
if self._alias is None:
|
||||
self._alias = self._get_alias()
|
||||
|
||||
try:
|
||||
return self._client.list_vars(self._alias)
|
||||
except (self.pynuterror, ConnectionResetError) as err:
|
||||
_LOGGER.debug("Error getting NUT vars for host %s: %s", self._host, err)
|
||||
return None
|
||||
|
||||
def update(self, **kwargs):
|
||||
"""Fetch the latest status from NUT."""
|
||||
self._status = self._get_status()
|
||||
|
||||
38
homeassistant/components/nut/strings.json
Normal file
38
homeassistant/components/nut/strings.json
Normal file
@ -0,0 +1,38 @@
|
||||
{
|
||||
"config": {
|
||||
"title": "Network UPS Tools (NUT)",
|
||||
"step": {
|
||||
"user": {
|
||||
"title": "Connect to the NUT server",
|
||||
"description": "If there are multiple UPSs attached to the NUT server, enter the name UPS to query in the 'Alias' field.",
|
||||
"data": {
|
||||
"name": "Name",
|
||||
"host": "Host",
|
||||
"port": "Port",
|
||||
"alias": "Alias",
|
||||
"username": "Username",
|
||||
"password": "Password",
|
||||
"resources": "Resources"
|
||||
}
|
||||
}
|
||||
},
|
||||
"error": {
|
||||
"cannot_connect": "Failed to connect, please try again",
|
||||
"unknown": "Unexpected error"
|
||||
},
|
||||
"abort": {
|
||||
"already_configured": "Device is already configured"
|
||||
}
|
||||
},
|
||||
"options": {
|
||||
"step": {
|
||||
"init": {
|
||||
"description": "Choose Sensor Resources",
|
||||
"data": {
|
||||
"resources": "Resources"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@ -81,6 +81,7 @@ FLOWS = [
|
||||
"nexia",
|
||||
"notion",
|
||||
"nuheat",
|
||||
"nut",
|
||||
"opentherm_gw",
|
||||
"openuv",
|
||||
"owntracks",
|
||||
|
||||
@ -554,6 +554,9 @@ pymonoprice==0.3
|
||||
# homeassistant.components.myq
|
||||
pymyq==2.0.1
|
||||
|
||||
# homeassistant.components.nut
|
||||
pynut2==2.1.2
|
||||
|
||||
# homeassistant.components.nws
|
||||
pynws==0.10.4
|
||||
|
||||
|
||||
1
tests/components/nut/__init__.py
Normal file
1
tests/components/nut/__init__.py
Normal file
@ -0,0 +1 @@
|
||||
"""Tests for the Network UPS Tools (NUT) integration."""
|
||||
122
tests/components/nut/test_config_flow.py
Normal file
122
tests/components/nut/test_config_flow.py
Normal file
@ -0,0 +1,122 @@
|
||||
"""Test the Network UPS Tools (NUT) config flow."""
|
||||
from asynctest import MagicMock, patch
|
||||
|
||||
from homeassistant import config_entries, setup
|
||||
from homeassistant.components.nut.const import DOMAIN
|
||||
|
||||
|
||||
def _get_mock_pynutclient(list_vars=None):
|
||||
pynutclient = MagicMock()
|
||||
type(pynutclient).list_ups = MagicMock(return_value=["ups1"])
|
||||
type(pynutclient).list_vars = MagicMock(return_value=list_vars)
|
||||
return pynutclient
|
||||
|
||||
|
||||
async def test_form(hass):
|
||||
"""Test we get the form."""
|
||||
await setup.async_setup_component(hass, "persistent_notification", {})
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
assert result["type"] == "form"
|
||||
assert result["errors"] == {}
|
||||
|
||||
mock_pynut = _get_mock_pynutclient(list_vars={"battery.voltage": "voltage"})
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.nut.PyNUTClient", return_value=mock_pynut,
|
||||
), patch(
|
||||
"homeassistant.components.nut.async_setup", return_value=True
|
||||
) as mock_setup, patch(
|
||||
"homeassistant.components.nut.async_setup_entry", return_value=True,
|
||||
) as mock_setup_entry:
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{
|
||||
"host": "1.1.1.1",
|
||||
"username": "test-username",
|
||||
"password": "test-password",
|
||||
"port": 2222,
|
||||
"alias": "ups1",
|
||||
"resources": ["battery.charge"],
|
||||
},
|
||||
)
|
||||
|
||||
assert result2["type"] == "create_entry"
|
||||
assert result2["title"] == "ups1@1.1.1.1:2222"
|
||||
assert result2["data"] == {
|
||||
"alias": "ups1",
|
||||
"host": "1.1.1.1",
|
||||
"name": "NUT UPS",
|
||||
"password": "test-password",
|
||||
"port": 2222,
|
||||
"resources": ["battery.charge"],
|
||||
"username": "test-username",
|
||||
}
|
||||
await hass.async_block_till_done()
|
||||
assert len(mock_setup.mock_calls) == 1
|
||||
assert len(mock_setup_entry.mock_calls) == 1
|
||||
|
||||
|
||||
async def test_form_import(hass):
|
||||
"""Test we get the form with import source."""
|
||||
await setup.async_setup_component(hass, "persistent_notification", {})
|
||||
|
||||
mock_pynut = _get_mock_pynutclient(list_vars={"battery.voltage": "serial"})
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.nut.PyNUTClient", return_value=mock_pynut,
|
||||
), patch(
|
||||
"homeassistant.components.nut.async_setup", return_value=True
|
||||
) as mock_setup, patch(
|
||||
"homeassistant.components.nut.async_setup_entry", return_value=True,
|
||||
) as mock_setup_entry:
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": config_entries.SOURCE_IMPORT},
|
||||
data={
|
||||
"host": "localhost",
|
||||
"port": 123,
|
||||
"name": "name",
|
||||
"resources": ["battery.charge"],
|
||||
},
|
||||
)
|
||||
|
||||
assert result["type"] == "create_entry"
|
||||
assert result["title"] == "localhost:123"
|
||||
assert result["data"] == {
|
||||
"host": "localhost",
|
||||
"port": 123,
|
||||
"name": "name",
|
||||
"resources": ["battery.charge"],
|
||||
}
|
||||
await hass.async_block_till_done()
|
||||
assert len(mock_setup.mock_calls) == 1
|
||||
assert len(mock_setup_entry.mock_calls) == 1
|
||||
|
||||
|
||||
async def test_form_cannot_connect(hass):
|
||||
"""Test we handle cannot connect error."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
|
||||
mock_pynut = _get_mock_pynutclient()
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.nut.PyNUTClient", return_value=mock_pynut,
|
||||
):
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{
|
||||
"host": "1.1.1.1",
|
||||
"username": "test-username",
|
||||
"password": "test-password",
|
||||
"port": 2222,
|
||||
"alias": "ups1",
|
||||
"resources": ["battery.charge"],
|
||||
},
|
||||
)
|
||||
|
||||
assert result2["type"] == "form"
|
||||
assert result2["errors"] == {"base": "cannot_connect"}
|
||||
Loading…
x
Reference in New Issue
Block a user