Add webmin integration (#106976)
* add webmin integration 1 * refactor, add memory sensors * Fix docstring * addressed reviews * address reviews * address reviews * use translation strings for sensors * add async_abort_entries_match * apply review comments * address reviews * add async_set_unique_id * add identifiers to device_info * disable all sensors by default * move icons to icons.json * show Faults when given from server in config flow * add test for Fault * Apply review suggestions * Create helper functions for webmin instance and sorted mac addresses * fix testspull/111492/head
parent
4ad7f420e7
commit
174ebe70d7
|
@ -1588,6 +1588,7 @@ omit =
|
|||
homeassistant/components/weatherflow/__init__.py
|
||||
homeassistant/components/weatherflow/const.py
|
||||
homeassistant/components/weatherflow/sensor.py
|
||||
homeassistant/components/webmin/sensor.py
|
||||
homeassistant/components/wiffi/__init__.py
|
||||
homeassistant/components/wiffi/binary_sensor.py
|
||||
homeassistant/components/wiffi/sensor.py
|
||||
|
|
|
@ -1515,6 +1515,8 @@ build.json @home-assistant/supervisor
|
|||
/tests/components/weatherkit/ @tjhorner
|
||||
/homeassistant/components/webhook/ @home-assistant/core
|
||||
/tests/components/webhook/ @home-assistant/core
|
||||
/homeassistant/components/webmin/ @autinerd
|
||||
/tests/components/webmin/ @autinerd
|
||||
/homeassistant/components/webostv/ @thecode
|
||||
/tests/components/webostv/ @thecode
|
||||
/homeassistant/components/websocket_api/ @home-assistant/core
|
||||
|
|
|
@ -0,0 +1,30 @@
|
|||
"""The Webmin integration."""
|
||||
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import Platform
|
||||
from homeassistant.core import HomeAssistant
|
||||
|
||||
from .const import DOMAIN
|
||||
from .coordinator import WebminUpdateCoordinator
|
||||
|
||||
PLATFORMS = [Platform.SENSOR]
|
||||
|
||||
|
||||
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||
"""Set up Webmin from a config entry."""
|
||||
|
||||
coordinator = WebminUpdateCoordinator(hass, entry)
|
||||
await coordinator.async_config_entry_first_refresh()
|
||||
await coordinator.async_setup()
|
||||
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = coordinator
|
||||
|
||||
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
|
||||
return True
|
||||
|
||||
|
||||
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||
"""Unload a config entry."""
|
||||
if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
|
||||
hass.data[DOMAIN].pop(entry.entry_id)
|
||||
|
||||
return unload_ok
|
|
@ -0,0 +1,95 @@
|
|||
"""Config flow for Webmin."""
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Mapping
|
||||
from http import HTTPStatus
|
||||
from typing import Any, cast
|
||||
from xmlrpc.client import Fault
|
||||
|
||||
from aiohttp.client_exceptions import ClientConnectionError, ClientResponseError
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.const import (
|
||||
CONF_HOST,
|
||||
CONF_PASSWORD,
|
||||
CONF_PORT,
|
||||
CONF_SSL,
|
||||
CONF_USERNAME,
|
||||
CONF_VERIFY_SSL,
|
||||
)
|
||||
from homeassistant.helpers import selector
|
||||
from homeassistant.helpers.schema_config_entry_flow import (
|
||||
SchemaCommonFlowHandler,
|
||||
SchemaConfigFlowHandler,
|
||||
SchemaFlowError,
|
||||
SchemaFlowFormStep,
|
||||
)
|
||||
|
||||
from .const import DEFAULT_PORT, DEFAULT_SSL, DEFAULT_VERIFY_SSL, DOMAIN
|
||||
from .helpers import get_instance_from_options, get_sorted_mac_addresses
|
||||
|
||||
|
||||
async def validate_user_input(
|
||||
handler: SchemaCommonFlowHandler, user_input: dict[str, Any]
|
||||
) -> dict[str, Any]:
|
||||
"""Validate user input."""
|
||||
# pylint: disable-next=protected-access
|
||||
handler.parent_handler._async_abort_entries_match(
|
||||
{CONF_HOST: user_input[CONF_HOST]}
|
||||
)
|
||||
instance, _ = get_instance_from_options(handler.parent_handler.hass, user_input)
|
||||
try:
|
||||
data = await instance.update()
|
||||
except ClientResponseError as err:
|
||||
if err.status == HTTPStatus.UNAUTHORIZED:
|
||||
raise SchemaFlowError("invalid_auth") from err
|
||||
raise SchemaFlowError("cannot_connect") from err
|
||||
except Fault as fault:
|
||||
raise SchemaFlowError(
|
||||
f"Fault {fault.faultCode}: {fault.faultString}"
|
||||
) from fault
|
||||
except ClientConnectionError as err:
|
||||
raise SchemaFlowError("cannot_connect") from err
|
||||
except Exception as err:
|
||||
raise SchemaFlowError("unknown") from err
|
||||
|
||||
await cast(SchemaConfigFlowHandler, handler.parent_handler).async_set_unique_id(
|
||||
get_sorted_mac_addresses(data)[0]
|
||||
)
|
||||
return user_input
|
||||
|
||||
|
||||
CONFIG_SCHEMA = vol.Schema(
|
||||
{
|
||||
vol.Required(CONF_HOST): selector.TextSelector(),
|
||||
vol.Required(CONF_PORT, default=DEFAULT_PORT): selector.NumberSelector(
|
||||
selector.NumberSelectorConfig(
|
||||
min=1, max=65535, mode=selector.NumberSelectorMode.BOX
|
||||
)
|
||||
),
|
||||
vol.Required(CONF_USERNAME): selector.TextSelector(),
|
||||
vol.Required(CONF_PASSWORD): selector.TextSelector(
|
||||
selector.TextSelectorConfig(type=selector.TextSelectorType.PASSWORD)
|
||||
),
|
||||
vol.Required(CONF_SSL, default=DEFAULT_SSL): selector.BooleanSelector(),
|
||||
vol.Required(
|
||||
CONF_VERIFY_SSL, default=DEFAULT_VERIFY_SSL
|
||||
): selector.BooleanSelector(),
|
||||
}
|
||||
)
|
||||
|
||||
CONFIG_FLOW = {
|
||||
"user": SchemaFlowFormStep(
|
||||
schema=CONFIG_SCHEMA, validate_user_input=validate_user_input
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
class WebminConfigFlowHandler(SchemaConfigFlowHandler, domain=DOMAIN):
|
||||
"""Handle a config flow for Webmin."""
|
||||
|
||||
config_flow = CONFIG_FLOW
|
||||
|
||||
def async_config_entry_title(self, options: Mapping[str, Any]) -> str:
|
||||
"""Return config entry title."""
|
||||
return str(options[CONF_HOST])
|
|
@ -0,0 +1,10 @@
|
|||
"""Constants for the Webmin integration."""
|
||||
|
||||
from logging import Logger, getLogger
|
||||
|
||||
LOGGER: Logger = getLogger(__package__)
|
||||
DOMAIN = "webmin"
|
||||
|
||||
DEFAULT_PORT = 10000
|
||||
DEFAULT_SSL = True
|
||||
DEFAULT_VERIFY_SSL = False
|
|
@ -0,0 +1,53 @@
|
|||
"""Data update coordinator for the Webmin integration."""
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import ATTR_CONNECTIONS, ATTR_IDENTIFIERS, CONF_HOST
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers.device_registry import (
|
||||
CONNECTION_NETWORK_MAC,
|
||||
DeviceInfo,
|
||||
format_mac,
|
||||
)
|
||||
from homeassistant.helpers.entity_component import DEFAULT_SCAN_INTERVAL
|
||||
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
|
||||
|
||||
from .const import DOMAIN, LOGGER
|
||||
from .helpers import get_instance_from_options, get_sorted_mac_addresses
|
||||
|
||||
|
||||
class WebminUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]):
|
||||
"""The Webmin data update coordinator."""
|
||||
|
||||
mac_address: str
|
||||
|
||||
def __init__(self, hass: HomeAssistant, config_entry: ConfigEntry) -> None:
|
||||
"""Initialize the Webmin data update coordinator."""
|
||||
|
||||
super().__init__(
|
||||
hass, logger=LOGGER, name=DOMAIN, update_interval=DEFAULT_SCAN_INTERVAL
|
||||
)
|
||||
|
||||
self.instance, base_url = get_instance_from_options(hass, config_entry.options)
|
||||
|
||||
self.device_info = DeviceInfo(
|
||||
configuration_url=base_url,
|
||||
name=config_entry.options[CONF_HOST],
|
||||
)
|
||||
|
||||
async def async_setup(self) -> None:
|
||||
"""Provide needed data to the device info."""
|
||||
mac_addresses = get_sorted_mac_addresses(self.data)
|
||||
self.mac_address = mac_addresses[0]
|
||||
self.device_info[ATTR_CONNECTIONS] = {
|
||||
(CONNECTION_NETWORK_MAC, format_mac(mac_address))
|
||||
for mac_address in mac_addresses
|
||||
}
|
||||
self.device_info[ATTR_IDENTIFIERS] = {
|
||||
(DOMAIN, format_mac(mac_address)) for mac_address in mac_addresses
|
||||
}
|
||||
|
||||
async def _async_update_data(self) -> dict[str, Any]:
|
||||
return await self.instance.update()
|
|
@ -0,0 +1,47 @@
|
|||
"""Helper functions for the Webmin integration."""
|
||||
|
||||
from collections.abc import Mapping
|
||||
from typing import Any
|
||||
|
||||
from webmin_xmlrpc.client import WebminInstance
|
||||
from yarl import URL
|
||||
|
||||
from homeassistant.const import (
|
||||
CONF_HOST,
|
||||
CONF_PASSWORD,
|
||||
CONF_PORT,
|
||||
CONF_SSL,
|
||||
CONF_USERNAME,
|
||||
CONF_VERIFY_SSL,
|
||||
)
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers.aiohttp_client import async_create_clientsession
|
||||
|
||||
|
||||
def get_instance_from_options(
|
||||
hass: HomeAssistant, options: Mapping[str, Any]
|
||||
) -> tuple[WebminInstance, URL]:
|
||||
"""Retrieve a Webmin instance and the base URL from config options."""
|
||||
|
||||
base_url = URL.build(
|
||||
scheme="https" if options[CONF_SSL] else "http",
|
||||
user=options[CONF_USERNAME],
|
||||
password=options[CONF_PASSWORD],
|
||||
host=options[CONF_HOST],
|
||||
port=int(options[CONF_PORT]),
|
||||
)
|
||||
|
||||
return WebminInstance(
|
||||
session=async_create_clientsession(
|
||||
hass,
|
||||
verify_ssl=options[CONF_VERIFY_SSL],
|
||||
base_url=base_url,
|
||||
)
|
||||
), base_url
|
||||
|
||||
|
||||
def get_sorted_mac_addresses(data: dict[str, Any]) -> list[str]:
|
||||
"""Return a sorted list of mac addresses."""
|
||||
return sorted(
|
||||
[iface["ether"] for iface in data["active_interfaces"] if "ether" in iface]
|
||||
)
|
|
@ -0,0 +1,27 @@
|
|||
{
|
||||
"entity": {
|
||||
"sensor": {
|
||||
"load_1m": {
|
||||
"default": "mdi:chip"
|
||||
},
|
||||
"load_5m": {
|
||||
"default": "mdi:chip"
|
||||
},
|
||||
"load_15m": {
|
||||
"default": "mdi:chip"
|
||||
},
|
||||
"mem_total": {
|
||||
"default": "mdi:memory"
|
||||
},
|
||||
"mem_free": {
|
||||
"default": "mdi:memory"
|
||||
},
|
||||
"swap_total": {
|
||||
"default": "mdi:memory"
|
||||
},
|
||||
"swap_free": {
|
||||
"default": "mdi:memory"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
|
@ -0,0 +1,11 @@
|
|||
{
|
||||
"domain": "webmin",
|
||||
"name": "Webmin",
|
||||
"codeowners": ["@autinerd"],
|
||||
"config_flow": true,
|
||||
"documentation": "https://www.home-assistant.io/integrations/webmin",
|
||||
"integration_type": "device",
|
||||
"iot_class": "local_polling",
|
||||
"loggers": ["webmin"],
|
||||
"requirements": ["webmin-xmlrpc==0.0.1"]
|
||||
}
|
|
@ -0,0 +1,112 @@
|
|||
"""Support for Webmin sensors."""
|
||||
from __future__ import annotations
|
||||
|
||||
from homeassistant.components.sensor import (
|
||||
SensorDeviceClass,
|
||||
SensorEntity,
|
||||
SensorEntityDescription,
|
||||
SensorStateClass,
|
||||
)
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import UnitOfInformation
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||
from homeassistant.helpers.update_coordinator import CoordinatorEntity
|
||||
|
||||
from .const import DOMAIN
|
||||
from .coordinator import WebminUpdateCoordinator
|
||||
|
||||
SENSOR_TYPES: list[SensorEntityDescription] = [
|
||||
SensorEntityDescription(
|
||||
key="load_1m",
|
||||
translation_key="load_1m",
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
entity_registry_enabled_default=False,
|
||||
),
|
||||
SensorEntityDescription(
|
||||
key="load_5m",
|
||||
translation_key="load_5m",
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
entity_registry_enabled_default=False,
|
||||
),
|
||||
SensorEntityDescription(
|
||||
key="load_15m",
|
||||
translation_key="load_15m",
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
entity_registry_enabled_default=False,
|
||||
),
|
||||
SensorEntityDescription(
|
||||
key="mem_total",
|
||||
translation_key="mem_total",
|
||||
native_unit_of_measurement=UnitOfInformation.KIBIBYTES,
|
||||
suggested_unit_of_measurement=UnitOfInformation.GIBIBYTES,
|
||||
device_class=SensorDeviceClass.DATA_SIZE,
|
||||
suggested_display_precision=2,
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
entity_registry_enabled_default=False,
|
||||
),
|
||||
SensorEntityDescription(
|
||||
key="mem_free",
|
||||
translation_key="mem_free",
|
||||
native_unit_of_measurement=UnitOfInformation.KIBIBYTES,
|
||||
suggested_unit_of_measurement=UnitOfInformation.GIBIBYTES,
|
||||
device_class=SensorDeviceClass.DATA_SIZE,
|
||||
suggested_display_precision=2,
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
entity_registry_enabled_default=False,
|
||||
),
|
||||
SensorEntityDescription(
|
||||
key="swap_total",
|
||||
translation_key="swap_total",
|
||||
native_unit_of_measurement=UnitOfInformation.KIBIBYTES,
|
||||
suggested_unit_of_measurement=UnitOfInformation.GIBIBYTES,
|
||||
device_class=SensorDeviceClass.DATA_SIZE,
|
||||
suggested_display_precision=2,
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
entity_registry_enabled_default=False,
|
||||
),
|
||||
SensorEntityDescription(
|
||||
key="swap_free",
|
||||
translation_key="swap_free",
|
||||
native_unit_of_measurement=UnitOfInformation.KIBIBYTES,
|
||||
suggested_unit_of_measurement=UnitOfInformation.GIBIBYTES,
|
||||
device_class=SensorDeviceClass.DATA_SIZE,
|
||||
suggested_display_precision=2,
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
entity_registry_enabled_default=False,
|
||||
),
|
||||
]
|
||||
|
||||
|
||||
async def async_setup_entry(
|
||||
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
|
||||
) -> None:
|
||||
"""Set up Webmin sensors based on a config entry."""
|
||||
coordinator: WebminUpdateCoordinator = hass.data[DOMAIN][entry.entry_id]
|
||||
async_add_entities(
|
||||
WebminSensor(coordinator, description)
|
||||
for description in SENSOR_TYPES
|
||||
if description.key in coordinator.data
|
||||
)
|
||||
|
||||
|
||||
class WebminSensor(CoordinatorEntity[WebminUpdateCoordinator], SensorEntity):
|
||||
"""Represents a Webmin sensor."""
|
||||
|
||||
entity_description: SensorEntityDescription
|
||||
_attr_has_entity_name = True
|
||||
|
||||
def __init__(
|
||||
self, coordinator: WebminUpdateCoordinator, description: SensorEntityDescription
|
||||
) -> None:
|
||||
"""Initialize a Webmin sensor."""
|
||||
|
||||
super().__init__(coordinator)
|
||||
self.entity_description = description
|
||||
self._attr_device_info = coordinator.device_info
|
||||
self._attr_unique_id = f"{coordinator.mac_address}_{description.key}"
|
||||
|
||||
@property
|
||||
def native_value(self) -> int | float:
|
||||
"""Return the state of the sensor."""
|
||||
return self.coordinator.data[self.entity_description.key]
|
|
@ -0,0 +1,54 @@
|
|||
{
|
||||
"config": {
|
||||
"flow_title": "{name}",
|
||||
"step": {
|
||||
"user": {
|
||||
"description": "Please enter the connection details of your instance.",
|
||||
"data": {
|
||||
"host": "[%key:common::config_flow::data::host%]",
|
||||
"port": "[%key:common::config_flow::data::port%]",
|
||||
"ssl": "[%key:common::config_flow::data::ssl%]",
|
||||
"username": "[%key:common::config_flow::data::username%]",
|
||||
"password": "[%key:common::config_flow::data::password%]",
|
||||
"verify_ssl": "[%key:common::config_flow::data::verify_ssl%]"
|
||||
}
|
||||
}
|
||||
},
|
||||
"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%]"
|
||||
},
|
||||
"abort": {
|
||||
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
|
||||
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
|
||||
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
|
||||
"unknown": "[%key:common::config_flow::error::unknown%]"
|
||||
}
|
||||
},
|
||||
"entity": {
|
||||
"sensor": {
|
||||
"load_1m": {
|
||||
"name": "Load (1m)"
|
||||
},
|
||||
"load_5m": {
|
||||
"name": "Load (5m)"
|
||||
},
|
||||
"load_15m": {
|
||||
"name": "Load (15m)"
|
||||
},
|
||||
"mem_total": {
|
||||
"name": "Memory total"
|
||||
},
|
||||
"mem_free": {
|
||||
"name": "Memory free"
|
||||
},
|
||||
"swap_total": {
|
||||
"name": "Swap total"
|
||||
},
|
||||
"swap_free": {
|
||||
"name": "Swap free"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
|
@ -587,6 +587,7 @@ FLOWS = {
|
|||
"waze_travel_time",
|
||||
"weatherflow",
|
||||
"weatherkit",
|
||||
"webmin",
|
||||
"webostv",
|
||||
"wemo",
|
||||
"whirlpool",
|
||||
|
|
|
@ -6668,6 +6668,12 @@
|
|||
"integration_type": "hub",
|
||||
"config_flow": false
|
||||
},
|
||||
"webmin": {
|
||||
"name": "Webmin",
|
||||
"integration_type": "device",
|
||||
"config_flow": true,
|
||||
"iot_class": "local_polling"
|
||||
},
|
||||
"wemo": {
|
||||
"name": "Belkin WeMo",
|
||||
"integration_type": "hub",
|
||||
|
|
|
@ -2838,6 +2838,9 @@ watchdog==2.3.1
|
|||
# homeassistant.components.waterfurnace
|
||||
waterfurnace==1.1.0
|
||||
|
||||
# homeassistant.components.webmin
|
||||
webmin-xmlrpc==0.0.1
|
||||
|
||||
# homeassistant.components.assist_pipeline
|
||||
webrtc-noise-gain==1.2.3
|
||||
|
||||
|
|
|
@ -2173,6 +2173,9 @@ wallbox==0.6.0
|
|||
# homeassistant.components.folder_watcher
|
||||
watchdog==2.3.1
|
||||
|
||||
# homeassistant.components.webmin
|
||||
webmin-xmlrpc==0.0.1
|
||||
|
||||
# homeassistant.components.assist_pipeline
|
||||
webrtc-noise-gain==1.2.3
|
||||
|
||||
|
|
|
@ -0,0 +1 @@
|
|||
"""Tests for the Webmin integration."""
|
|
@ -0,0 +1,33 @@
|
|||
"""Fixtures for Webmin integration tests."""
|
||||
from collections.abc import Generator
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from homeassistant.components.webmin.const import DEFAULT_PORT
|
||||
from homeassistant.const import (
|
||||
CONF_HOST,
|
||||
CONF_PASSWORD,
|
||||
CONF_PORT,
|
||||
CONF_SSL,
|
||||
CONF_USERNAME,
|
||||
CONF_VERIFY_SSL,
|
||||
)
|
||||
|
||||
TEST_USER_INPUT = {
|
||||
CONF_HOST: "192.168.1.1",
|
||||
CONF_USERNAME: "user",
|
||||
CONF_PASSWORD: "pass",
|
||||
CONF_PORT: DEFAULT_PORT,
|
||||
CONF_SSL: True,
|
||||
CONF_VERIFY_SSL: False,
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_setup_entry() -> Generator[AsyncMock, None, None]:
|
||||
"""Mock setting up a config entry."""
|
||||
with patch(
|
||||
"homeassistant.components.webmin.async_setup_entry", return_value=True
|
||||
) as mock_setup:
|
||||
yield mock_setup
|
|
@ -0,0 +1,45 @@
|
|||
{
|
||||
"active_interfaces": [
|
||||
{
|
||||
"fullname": "lo",
|
||||
"up": 1,
|
||||
"index": 0,
|
||||
"scope6": ["host"],
|
||||
"netmask": "255.0.0.0",
|
||||
"netmask6": [128],
|
||||
"edit": 1,
|
||||
"broadcast": 0,
|
||||
"mtu": 65536,
|
||||
"name": "lo",
|
||||
"address": "127.0.0.1",
|
||||
"address6": ["::1"]
|
||||
},
|
||||
{
|
||||
"mtu": 1500,
|
||||
"fullname": "enp6s0",
|
||||
"up": 1,
|
||||
"index": 1,
|
||||
"ether": "12:34:56:78:9a:bc",
|
||||
"address6": [],
|
||||
"netmask6": [],
|
||||
"edit": 1,
|
||||
"scope6": [],
|
||||
"name": "enp6s0"
|
||||
},
|
||||
{
|
||||
"edit": 1,
|
||||
"netmask6": [64],
|
||||
"netmask": "255.255.255.0",
|
||||
"scope6": ["link"],
|
||||
"up": 1,
|
||||
"index": 2,
|
||||
"fullname": "eno1",
|
||||
"address6": ["fe80::2:3:4"],
|
||||
"address": "192.168.1.4",
|
||||
"name": "eno1",
|
||||
"mtu": 1500,
|
||||
"broadcast": "192.168.1.255",
|
||||
"ether": "12:34:56:78:9a:bd"
|
||||
}
|
||||
]
|
||||
}
|
|
@ -0,0 +1,97 @@
|
|||
{
|
||||
"load_1m": 0.98,
|
||||
"load_5m": 1.02,
|
||||
"load_15m": 1.0,
|
||||
"mem_total": 32767008,
|
||||
"mem_free": 26162544,
|
||||
"swap_total": 1953088,
|
||||
"swap_free": 1953088,
|
||||
"total_space": 18104905818112,
|
||||
"free_space": 8641328926720,
|
||||
"fs": [
|
||||
{
|
||||
"free": 174511820800,
|
||||
"dir": "/",
|
||||
"iused": 391146,
|
||||
"used": 61225123840,
|
||||
"type": "ext4",
|
||||
"device": "UUID=00000000-80b6-0000-8a06-000000000000",
|
||||
"iused_percent": 3,
|
||||
"used_percent": 26,
|
||||
"total": 248431161344,
|
||||
"itotal": 15482880,
|
||||
"ifree": 15091734
|
||||
},
|
||||
{
|
||||
"iused": 8877,
|
||||
"used": 4608079593472,
|
||||
"type": "ext4",
|
||||
"dir": "/media/disk1",
|
||||
"free": 1044483624960,
|
||||
"used_percent": 82,
|
||||
"ifree": 183131475,
|
||||
"itotal": 183140352,
|
||||
"total": 5952635744256,
|
||||
"device": "UUID=00000000-2bb2-0000-896c-000000000000",
|
||||
"iused_percent": 1
|
||||
},
|
||||
{
|
||||
"used": 3881508986880,
|
||||
"type": "ext4",
|
||||
"iused": 3411401,
|
||||
"dir": "/media/disk2",
|
||||
"free": 7422333480960,
|
||||
"used_percent": 35,
|
||||
"total": 11903838912512,
|
||||
"itotal": 366198784,
|
||||
"ifree": 362787383,
|
||||
"device": "/dev/md127",
|
||||
"iused_percent": 1
|
||||
}
|
||||
],
|
||||
"used_space": 8550813704192,
|
||||
"uptime": { "days": 3, "minutes": 23, "seconds": 12 },
|
||||
"active_interfaces": [
|
||||
{
|
||||
"fullname": "lo",
|
||||
"up": 1,
|
||||
"index": 0,
|
||||
"scope6": ["host"],
|
||||
"netmask": "255.0.0.0",
|
||||
"netmask6": [128],
|
||||
"edit": 1,
|
||||
"broadcast": 0,
|
||||
"mtu": 65536,
|
||||
"name": "lo",
|
||||
"address": "127.0.0.1",
|
||||
"address6": ["::1"]
|
||||
},
|
||||
{
|
||||
"mtu": 1500,
|
||||
"fullname": "enp6s0",
|
||||
"up": 1,
|
||||
"index": 1,
|
||||
"ether": "12:34:56:78:9a:bc",
|
||||
"address6": [],
|
||||
"netmask6": [],
|
||||
"edit": 1,
|
||||
"scope6": [],
|
||||
"name": "enp6s0"
|
||||
},
|
||||
{
|
||||
"edit": 1,
|
||||
"netmask6": [64],
|
||||
"netmask": "255.255.255.0",
|
||||
"scope6": ["link"],
|
||||
"up": 1,
|
||||
"index": 2,
|
||||
"fullname": "eno1",
|
||||
"address6": ["fe80::2:3:4"],
|
||||
"address": "192.168.1.4",
|
||||
"name": "eno1",
|
||||
"mtu": 1500,
|
||||
"broadcast": "192.168.1.255",
|
||||
"ether": "12:34:56:78:9a:bd"
|
||||
}
|
||||
]
|
||||
}
|
|
@ -0,0 +1,140 @@
|
|||
"""Test the Webmin config flow."""
|
||||
from __future__ import annotations
|
||||
|
||||
from http import HTTPStatus
|
||||
from unittest.mock import AsyncMock, patch
|
||||
from xmlrpc.client import Fault
|
||||
|
||||
from aiohttp.client_exceptions import ClientConnectionError, ClientResponseError
|
||||
import pytest
|
||||
|
||||
from homeassistant import config_entries
|
||||
from homeassistant.components.webmin.const import DOMAIN
|
||||
from homeassistant.const import CONF_HOST
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.data_entry_flow import FlowResultType
|
||||
|
||||
from .conftest import TEST_USER_INPUT
|
||||
|
||||
from tests.common import load_json_object_fixture
|
||||
|
||||
pytestmark = pytest.mark.usefixtures("mock_setup_entry")
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def user_flow(hass: HomeAssistant) -> str:
|
||||
"""Return a user-initiated flow after filling in host info."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
assert result["type"] == FlowResultType.FORM
|
||||
assert result["errors"] is None
|
||||
return result["flow_id"]
|
||||
|
||||
|
||||
async def test_form_user(
|
||||
hass: HomeAssistant,
|
||||
user_flow: str,
|
||||
mock_setup_entry: AsyncMock,
|
||||
) -> None:
|
||||
"""Test a successful user initiated flow."""
|
||||
with patch(
|
||||
"homeassistant.components.webmin.helpers.WebminInstance.update",
|
||||
return_value=load_json_object_fixture("webmin_update.json", DOMAIN),
|
||||
):
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
user_flow, TEST_USER_INPUT
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
assert result["type"] == FlowResultType.CREATE_ENTRY
|
||||
assert result["title"] == TEST_USER_INPUT[CONF_HOST]
|
||||
assert result["options"] == TEST_USER_INPUT
|
||||
|
||||
assert len(mock_setup_entry.mock_calls) == 1
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("exception", "error_type"),
|
||||
[
|
||||
(
|
||||
ClientResponseError(
|
||||
request_info=None, history=None, status=HTTPStatus.UNAUTHORIZED
|
||||
),
|
||||
"invalid_auth",
|
||||
),
|
||||
(
|
||||
ClientResponseError(
|
||||
request_info=None, history=None, status=HTTPStatus.BAD_REQUEST
|
||||
),
|
||||
"cannot_connect",
|
||||
),
|
||||
(ClientConnectionError, "cannot_connect"),
|
||||
(Exception, "unknown"),
|
||||
(
|
||||
Fault("5", "Webmin module net does not exist"),
|
||||
"Fault 5: Webmin module net does not exist",
|
||||
),
|
||||
],
|
||||
)
|
||||
async def test_form_user_errors(
|
||||
hass: HomeAssistant, user_flow: str, exception: Exception, error_type: str
|
||||
) -> None:
|
||||
"""Test we handle errors."""
|
||||
with patch(
|
||||
"homeassistant.components.webmin.helpers.WebminInstance.update",
|
||||
side_effect=exception,
|
||||
):
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
user_flow, TEST_USER_INPUT
|
||||
)
|
||||
|
||||
assert result["type"] == FlowResultType.FORM
|
||||
assert result["step_id"] == "user"
|
||||
assert result["errors"] == {"base": error_type}
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.webmin.helpers.WebminInstance.update",
|
||||
return_value=load_json_object_fixture("webmin_update.json", DOMAIN),
|
||||
):
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], TEST_USER_INPUT
|
||||
)
|
||||
|
||||
assert result["type"] == FlowResultType.CREATE_ENTRY
|
||||
assert result["title"] == TEST_USER_INPUT[CONF_HOST]
|
||||
assert result["options"] == TEST_USER_INPUT
|
||||
|
||||
|
||||
async def test_duplicate_entry(
|
||||
hass: HomeAssistant,
|
||||
user_flow: str,
|
||||
mock_setup_entry: AsyncMock,
|
||||
) -> None:
|
||||
"""Test a successful user initiated flow."""
|
||||
with patch(
|
||||
"homeassistant.components.webmin.helpers.WebminInstance.update",
|
||||
return_value=load_json_object_fixture("webmin_update.json", DOMAIN),
|
||||
):
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
user_flow, TEST_USER_INPUT
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert result["type"] == FlowResultType.CREATE_ENTRY
|
||||
assert result["title"] == TEST_USER_INPUT[CONF_HOST]
|
||||
assert result["options"] == TEST_USER_INPUT
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.webmin.helpers.WebminInstance.update",
|
||||
return_value=load_json_object_fixture("webmin_update.json", DOMAIN),
|
||||
):
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], TEST_USER_INPUT
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert result["type"] == FlowResultType.ABORT
|
||||
assert result["reason"] == "already_configured"
|
|
@ -0,0 +1,32 @@
|
|||
"""Tests for the Webmin integration."""
|
||||
from unittest.mock import patch
|
||||
|
||||
from homeassistant.components.webmin.const import DOMAIN
|
||||
from homeassistant.config_entries import ConfigEntryState
|
||||
from homeassistant.core import HomeAssistant
|
||||
|
||||
from .conftest import TEST_USER_INPUT
|
||||
|
||||
from tests.common import MockConfigEntry, load_json_object_fixture
|
||||
|
||||
|
||||
async def test_unload_entry(hass: HomeAssistant) -> None:
|
||||
"""Test successful unload of entry."""
|
||||
|
||||
entry = MockConfigEntry(domain=DOMAIN, options=TEST_USER_INPUT, title="name")
|
||||
entry.add_to_hass(hass)
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.webmin.helpers.WebminInstance.update",
|
||||
return_value=load_json_object_fixture("webmin_update.json", DOMAIN),
|
||||
):
|
||||
await hass.config_entries.async_setup(entry.entry_id)
|
||||
|
||||
await hass.async_block_till_done()
|
||||
assert entry.state is ConfigEntryState.LOADED
|
||||
|
||||
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)
|
Loading…
Reference in New Issue