Compare commits

...

7 Commits

Author SHA1 Message Date
Bram Kragten 0c816c22e0 Remove show_advanced_options from data entry flow API (#172249) 2026-05-27 11:13:24 +02:00
epenet 42f277716d Ensure local_strategy is defined in tuya tests (#172328) 2026-05-27 10:52:14 +02:00
Ronald van der Meer 6669b0de25 Use Duco state codes for ventilation state labels (#172314) 2026-05-27 10:43:46 +02:00
wollew 50fca42624 Bump pyvlx to 0.2.35 (#172320) 2026-05-27 10:38:55 +02:00
Erik Montnemery deecb4ee9c Improve cast option flow tests (#172323) 2026-05-27 10:37:50 +02:00
Erik Montnemery 762f07f450 Add device_tracker platform to kitchen_sink (#172250) 2026-05-27 10:21:09 +02:00
Kevin McCormack e02ea041b7 Add config flow for OPNsense (#151121)
Co-authored-by: Norbert Rittel <norbert@rittel.de>
Co-authored-by: Philippe Lafoucrière <12752+gravis@users.noreply.github.com>
Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>
Co-authored-by: Robert Resch <robert@resch.dev>
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
Co-authored-by: Ariel Ebersberger <ariel@ebersberger.io>
2026-05-27 10:15:16 +02:00
38 changed files with 1992 additions and 350 deletions
@@ -175,7 +175,6 @@ class ConfigManagerFlowIndexView(
vol.Schema(
{
vol.Required("handler"): vol.Any(str, list),
vol.Optional("show_advanced_options", default=False): cv.boolean,
vol.Optional("entry_id"): cv.string,
},
extra=vol.ALLOW_EXTRA,
@@ -302,7 +301,6 @@ class SubentryManagerFlowIndexView(
vol.Schema(
{
vol.Required("handler"): vol.All(vol.Coerce(tuple), (str, str)),
vol.Optional("show_advanced_options", default=False): cv.boolean,
},
extra=vol.ALLOW_EXTRA,
)
+17 -17
View File
@@ -64,23 +64,23 @@
"ventilation_state": {
"name": "Ventilation state",
"state": {
"aut1": "Automatic boost (15 min)",
"aut2": "Automatic boost (30 min)",
"aut3": "Automatic boost (45 min)",
"auto": "Automatic",
"cnt1": "Continuous low speed",
"cnt2": "Continuous medium speed",
"cnt3": "Continuous high speed",
"empt": "Empty house",
"man1": "Manual low speed (15 min)",
"man1x2": "Manual low speed (30 min)",
"man1x3": "Manual low speed (45 min)",
"man2": "Manual medium speed (15 min)",
"man2x2": "Manual medium speed (30 min)",
"man2x3": "Manual medium speed (45 min)",
"man3": "Manual high speed (15 min)",
"man3x2": "Manual high speed (30 min)",
"man3x3": "Manual high speed (45 min)"
"aut1": "AUT1",
"aut2": "AUT2",
"aut3": "AUT3",
"auto": "AUTO",
"cnt1": "CNT1",
"cnt2": "CNT2",
"cnt3": "CNT3",
"empt": "EMPT",
"man1": "MAN1",
"man1x2": "MAN1x2",
"man1x3": "MAN1x3",
"man2": "MAN2",
"man2x2": "MAN2x2",
"man2x3": "MAN2x3",
"man3": "MAN3",
"man3x2": "MAN3x2",
"man3x3": "MAN3x3"
}
}
}
@@ -8,8 +8,6 @@ import datetime
from functools import partial
from random import random
import voluptuous as vol
from homeassistant.components.labs import (
EventLabsUpdatedData,
async_is_preview_feature_enabled,
@@ -34,7 +32,7 @@ from homeassistant.const import (
UnitOfTemperature,
UnitOfVolume,
)
from homeassistant.core import HomeAssistant, ServiceCall, ServiceResponse, callback
from homeassistant.core import HomeAssistant
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.device_registry import DeviceEntry
from homeassistant.helpers.issue_registry import (
@@ -51,9 +49,11 @@ from homeassistant.util.unit_conversion import (
)
from .const import DATA_BACKUP_AGENT_LISTENERS, DOMAIN
from .services import async_setup_services
COMPONENTS_WITH_DEMO_PLATFORM = [
Platform.BUTTON,
Platform.DEVICE_TRACKER,
Platform.FAN,
Platform.EVENT,
Platform.IMAGE,
@@ -69,15 +69,6 @@ COMPONENTS_WITH_DEMO_PLATFORM = [
CONFIG_SCHEMA = cv.empty_config_schema(DOMAIN)
SCHEMA_SERVICE_TEST_SERVICE_1 = vol.Schema(
{
vol.Required("field_1"): vol.Coerce(int),
vol.Required("field_2"): vol.In(["off", "auto", "cool"]),
vol.Optional("field_3"): vol.Coerce(int),
vol.Optional("field_4"): vol.In(["forwards", "reverse"]),
}
)
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the demo environment."""
@@ -87,24 +78,7 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
)
)
@callback
def service_handler(call: ServiceCall | None = None) -> ServiceResponse:
"""Do nothing."""
return None
hass.services.async_register(
DOMAIN,
"test_service_1",
service_handler,
SCHEMA_SERVICE_TEST_SERVICE_1,
description_placeholders={
"meep_1": "foo",
"meep_2": "bar",
"meep_3": "beer",
"meep_4": "milk",
"meep_5": "https://example.com",
},
)
async_setup_services(hass)
return True
@@ -0,0 +1,97 @@
"""Demo platform that has a couple of fake device trackers."""
from homeassistant.components.device_tracker import (
BaseScannerEntity,
SourceType,
TrackerEntity,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the Everything but the Kitchen Sink config entry."""
async_add_entities(
[
DemoTracker(
unique_id="kitchen_sink_tracker_001",
name="Demo tracker",
latitude=hass.config.latitude,
longitude=hass.config.longitude,
accuracy=10,
),
DemoScanner(
unique_id="kitchen_sink_scanner_001",
name="Demo scanner",
is_connected=True,
),
]
)
class DemoTracker(TrackerEntity):
"""Representation of a demo tracker."""
_attr_should_poll = False
_attr_source_type = SourceType.GPS
def __init__(
self,
*,
unique_id: str,
name: str,
latitude: float | None,
longitude: float | None,
accuracy: float,
) -> None:
"""Initialize the tracker."""
self._attr_unique_id = unique_id
self._attr_name = name
self._attr_latitude = latitude
self._attr_longitude = longitude
self._attr_location_accuracy = accuracy
@callback
def async_set_tracker_location(
self, latitude: float, longitude: float, accuracy: float
) -> None:
"""Update the tracker location."""
self._attr_latitude = latitude
self._attr_longitude = longitude
self._attr_location_accuracy = accuracy
self.async_write_ha_state()
class DemoScanner(BaseScannerEntity):
"""Representation of a demo scanner."""
_attr_should_poll = False
_attr_source_type = SourceType.ROUTER
def __init__(
self,
*,
unique_id: str,
name: str,
is_connected: bool,
) -> None:
"""Initialize the scanner."""
self._attr_unique_id = unique_id
self._attr_name = name
self._is_connected = is_connected
@property
def is_connected(self) -> bool:
"""Return true if the device is connected."""
return self._is_connected
@callback
def async_set_scanner_connected(self, connected: bool) -> None:
"""Update the scanner connected state."""
self._is_connected = connected
self.async_write_ha_state()
@@ -9,6 +9,12 @@
}
},
"services": {
"set_scanner_connected": {
"service": "mdi:lan-connect"
},
"set_tracker_location": {
"service": "mdi:map-marker"
},
"test_service_1": {
"sections": {
"additional_fields": "mdi:test-tube"
@@ -0,0 +1,72 @@
"""Services for the Everything but the Kitchen Sink integration."""
import voluptuous as vol
from homeassistant.components.device_tracker import DOMAIN as DEVICE_TRACKER_DOMAIN
from homeassistant.const import ATTR_LATITUDE, ATTR_LONGITUDE
from homeassistant.core import HomeAssistant, ServiceCall, ServiceResponse, callback
from homeassistant.helpers import config_validation as cv, service
from .const import DOMAIN
SCHEMA_SERVICE_TEST_SERVICE_1 = vol.Schema(
{
vol.Required("field_1"): vol.Coerce(int),
vol.Required("field_2"): vol.In(["off", "auto", "cool"]),
vol.Optional("field_3"): vol.Coerce(int),
vol.Optional("field_4"): vol.In(["forward", "reverse"]),
}
)
SERVICE_TEST_SERVICE_1 = "test_service_1"
SERVICE_SET_TRACKER_LOCATION = "set_tracker_location"
SERVICE_SET_SCANNER_CONNECTED = "set_scanner_connected"
ATTR_ACCURACY = "accuracy"
ATTR_CONNECTED = "connected"
@callback
def async_setup_services(hass: HomeAssistant) -> None:
"""Register services for the Kitchen Sink integration."""
@callback
def service_handler(call: ServiceCall) -> ServiceResponse:
"""Do nothing."""
return None
hass.services.async_register(
DOMAIN,
SERVICE_TEST_SERVICE_1,
service_handler,
SCHEMA_SERVICE_TEST_SERVICE_1,
description_placeholders={
"meep_1": "foo",
"meep_2": "bar",
"meep_3": "beer",
"meep_4": "milk",
"meep_5": "https://example.com",
},
)
service.async_register_platform_entity_service(
hass,
DOMAIN,
SERVICE_SET_TRACKER_LOCATION,
entity_domain=DEVICE_TRACKER_DOMAIN,
schema={
vol.Required(ATTR_LATITUDE): cv.latitude,
vol.Required(ATTR_LONGITUDE): cv.longitude,
vol.Required(ATTR_ACCURACY): vol.All(vol.Coerce(float), vol.Range(min=0)),
},
func="async_set_tracker_location",
)
service.async_register_platform_entity_service(
hass,
DOMAIN,
SERVICE_SET_SCANNER_CONNECTED,
entity_domain=DEVICE_TRACKER_DOMAIN,
schema={vol.Required(ATTR_CONNECTED): cv.boolean},
func="async_set_scanner_connected",
)
@@ -30,3 +30,44 @@ test_service_1:
options:
- "forward"
- "reverse"
set_tracker_location:
target:
entity:
integration: kitchen_sink
domain: device_tracker
fields:
latitude:
required: true
example: 52.379189
selector:
number:
min: -90
max: 90
step: any
longitude:
required: true
example: 4.899431
selector:
number:
min: -180
max: 180
step: any
accuracy:
required: true
example: 10
selector:
number:
min: 0
max: 10000
unit_of_measurement: m
set_scanner_connected:
target:
entity:
integration: kitchen_sink
domain: device_tracker
fields:
connected:
required: true
example: true
selector:
boolean:
@@ -135,6 +135,34 @@
}
},
"services": {
"set_scanner_connected": {
"description": "Sets the connected state of a demo scanner entity.",
"fields": {
"connected": {
"description": "Whether the device should be reported as connected.",
"name": "Connected"
}
},
"name": "Set scanner connected"
},
"set_tracker_location": {
"description": "Sets the location and accuracy of a demo tracker entity.",
"fields": {
"accuracy": {
"description": "Location accuracy in meters.",
"name": "Accuracy"
},
"latitude": {
"description": "Latitude of the new location.",
"name": "Latitude"
},
"longitude": {
"description": "Longitude of the new location.",
"name": "Longitude"
}
},
"name": "Set tracker location"
},
"test_service_1": {
"description": "Fake action for testing {meep_2}",
"fields": {
+103 -73
View File
@@ -1,7 +1,5 @@
"""Support for OPNsense Routers."""
import logging
from aiopnsense import (
OPNsenseBelowMinFirmware,
OPNsenseClient,
@@ -15,22 +13,16 @@ from aiopnsense import (
)
import voluptuous as vol
from homeassistant.config_entries import SOURCE_IMPORT
from homeassistant.const import CONF_API_KEY, CONF_URL, CONF_VERIFY_SSL, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import config_validation as cv
from homeassistant.exceptions import ConfigEntryError, ConfigEntryNotReady
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.discovery import load_platform
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.typing import ConfigType
from .const import (
CONF_API_SECRET,
CONF_INTERFACE_CLIENT,
CONF_TRACKER_INTERFACES,
DOMAIN,
OPNSENSE_DATA,
)
_LOGGER = logging.getLogger(__name__)
from .const import CONF_API_SECRET, CONF_TRACKER_INTERFACES, DOMAIN
from .types import OPNsenseConfigEntry, OPNsenseRuntimeData
CONFIG_SCHEMA = vol.Schema(
{
@@ -49,86 +41,124 @@ CONFIG_SCHEMA = vol.Schema(
extra=vol.ALLOW_EXTRA,
)
PLATFORMS = [Platform.DEVICE_TRACKER]
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the opnsense component."""
"""Set up the OPNsense component."""
if DOMAIN not in config:
return True
conf = config[DOMAIN]
url = conf[CONF_URL]
api_key = conf[CONF_API_KEY]
api_secret = conf[CONF_API_SECRET]
verify_ssl = conf[CONF_VERIFY_SSL]
tracker_interfaces = conf[CONF_TRACKER_INTERFACES]
hass.async_create_task(_async_setup(hass, config))
session = async_get_clientsession(hass, verify_ssl=verify_ssl)
return True
async def _async_setup(hass: HomeAssistant, config: ConfigType) -> None:
"""Set up the OPNsense component from YAML."""
await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_IMPORT},
data=config[DOMAIN],
)
async def async_setup_entry(
hass: HomeAssistant, config_entry: OPNsenseConfigEntry
) -> bool:
"""Set up the OPNsense component from a config entry."""
url = config_entry.data[CONF_URL]
session = async_get_clientsession(
hass, verify_ssl=config_entry.data[CONF_VERIFY_SSL]
)
client = OPNsenseClient(
url,
api_key,
api_secret,
config_entry.data[CONF_API_KEY],
config_entry.data[CONF_API_SECRET],
session,
opts={"verify_ssl": verify_ssl},
opts={"verify_ssl": config_entry.data[CONF_VERIFY_SSL]},
)
tracker_interfaces = config_entry.data.get(CONF_TRACKER_INTERFACES, [])
try:
await client.validate()
if tracker_interfaces:
interfaces_resp = await client.get_interfaces()
except OPNsenseUnknownFirmware:
_LOGGER.error("Error checking the OPNsense firmware version at %s", url)
return False
except OPNsenseBelowMinFirmware:
_LOGGER.error(
"OPNsense Firmware is below the minimum supported version at %s", url
)
return False
except OPNsenseInvalidURL:
_LOGGER.error(
"Invalid URL while connecting to OPNsense API endpoint at %s", url
)
return False
except OPNsenseTimeoutError:
_LOGGER.error("Timeout while connecting to OPNsense API endpoint at %s", url)
return False
except OPNsenseSSLError:
_LOGGER.error(
"Unable to verify SSL while connecting to OPNsense API endpoint at %s", url
)
return False
except OPNsenseInvalidAuth:
_LOGGER.error(
"Authentication failure while connecting to OPNsense API endpoint at %s",
url,
)
return False
except OPNsensePrivilegeMissing:
_LOGGER.error(
"Invalid Permissions while connecting to OPNsense API endpoint at %s",
url,
)
return False
except OPNsenseConnectionError:
_LOGGER.error(
"Connection failure while connecting to OPNsense API endpoint at %s",
url,
)
return False
except OPNsenseUnknownFirmware as err:
raise ConfigEntryError(
translation_domain=DOMAIN,
translation_key="unknown_firmware",
translation_placeholders={"url": url},
) from err
except OPNsenseBelowMinFirmware as err:
raise ConfigEntryError(
translation_domain=DOMAIN,
translation_key="firmware_too_old",
translation_placeholders={"url": url},
) from err
except OPNsenseInvalidURL as err:
raise ConfigEntryError(
translation_domain=DOMAIN,
translation_key="invalid_url",
translation_placeholders={"url": url},
) from err
except OPNsenseTimeoutError as err:
raise ConfigEntryNotReady(
translation_domain=DOMAIN,
translation_key="timeout_connecting",
translation_placeholders={"url": url},
) from err
except OPNsenseSSLError as err:
raise ConfigEntryError(
translation_domain=DOMAIN,
translation_key="ssl_error",
translation_placeholders={"url": url},
) from err
except OPNsenseInvalidAuth as err:
raise ConfigEntryError(
translation_domain=DOMAIN,
translation_key="invalid_auth",
translation_placeholders={"url": url},
) from err
except OPNsensePrivilegeMissing as err:
raise ConfigEntryError(
translation_domain=DOMAIN,
translation_key="privilege_missing",
translation_placeholders={"url": url},
) from err
except OPNsenseConnectionError as err:
raise ConfigEntryNotReady(
translation_domain=DOMAIN,
translation_key="cannot_connect",
translation_placeholders={"url": url},
) from err
if tracker_interfaces:
# Verify that specified tracker interfaces are valid
known_interfaces = [
ifinfo.get("name", "") for ifinfo in interfaces_resp.values()
name for ifinfo in interfaces_resp.values() if (name := ifinfo.get("name"))
]
for intf_description in tracker_interfaces:
if intf_description not in known_interfaces:
_LOGGER.error(
"Specified OPNsense tracker interface %s is not found",
intf_description,
raise ConfigEntryError(
translation_domain=DOMAIN,
translation_key="tracker_interface_not_found",
translation_placeholders={
"interface": intf_description,
"known": ", ".join(known_interfaces),
},
)
return False
hass.data[OPNSENSE_DATA] = {
CONF_INTERFACE_CLIENT: client,
CONF_TRACKER_INTERFACES: tracker_interfaces,
}
config_entry.runtime_data = OPNsenseRuntimeData(
client=client,
tracker_interfaces=tracker_interfaces,
)
load_platform(hass, Platform.DEVICE_TRACKER, DOMAIN, tracker_interfaces, config)
await hass.config_entries.async_forward_entry_setups(config_entry, PLATFORMS)
return True
async def async_unload_entry(
hass: HomeAssistant, config_entry: OPNsenseConfigEntry
) -> bool:
"""Unload a config entry."""
return await hass.config_entries.async_unload_platforms(config_entry, PLATFORMS)
@@ -0,0 +1,315 @@
"""Config flow for OPNsense."""
import logging
from typing import Any
from aiopnsense import (
OPNsenseBelowMinFirmware,
OPNsenseClient,
OPNsenseConnectionError,
OPNsenseInvalidAuth,
OPNsenseInvalidURL,
OPNsensePrivilegeMissing,
OPNsenseSSLError,
OPNsenseTimeoutError,
OPNsenseUnknownFirmware,
)
import voluptuous as vol
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_API_KEY, CONF_URL, CONF_VERIFY_SSL
from homeassistant.core import DOMAIN as HOMEASSISTANT_DOMAIN
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.issue_registry import (
IssueSeverity,
async_create_issue,
async_delete_issue,
)
from homeassistant.helpers.selector import (
SelectSelector,
SelectSelectorConfig,
SelectSelectorMode,
)
from .const import CONF_API_SECRET, CONF_TRACKER_INTERFACES, DOMAIN
_LOGGER = logging.getLogger(__name__)
STEP_USER_DATA_SCHEMA = vol.Schema(
{
vol.Required(CONF_URL): str,
vol.Required(CONF_API_KEY): str,
vol.Required(CONF_API_SECRET): str,
vol.Required(CONF_VERIFY_SSL, default=True): bool,
}
)
def tracker_interfaces_schema(
interfaces: list[str], selected: list[str] | None = None
) -> vol.Schema:
"""Schema to display available interfaces for device tracking selection."""
return vol.Schema(
{
vol.Optional(
CONF_TRACKER_INTERFACES,
default=selected or [],
): SelectSelector(
SelectSelectorConfig(
options=interfaces, mode=SelectSelectorMode.DROPDOWN, multiple=True
)
),
}
)
class OPNsenseConfigFlow(ConfigFlow, domain=DOMAIN):
"""OPNsense config flow."""
def __init__(self) -> None:
"""Initialize OPNsense config flow."""
self.available_interfaces: list[str] | None = None
self._entry_data: dict[str, Any] = {}
async def _show_setup_form(
self,
user_input: dict[Any, Any] | None = None,
errors: dict[Any, Any] | None = None,
) -> ConfigFlowResult:
"""Show the setup form to the user."""
if user_input is None:
user_input = {}
description_placeholders = {
"doc_url": "https://www.home-assistant.io/integrations/opnsense/"
}
return self.async_show_form(
step_id="user",
data_schema=self.add_suggested_values_to_schema(
STEP_USER_DATA_SCHEMA, user_input
),
errors=errors or {},
description_placeholders=description_placeholders,
)
async def _show_interfaces_form(
self,
user_input: dict[Any, Any],
errors: dict[Any, Any] | None = None,
) -> ConfigFlowResult:
"""Show the tracker interfaces selection form to the user."""
return self.async_show_form(
step_id="interfaces",
data_schema=self.add_suggested_values_to_schema(
tracker_interfaces_schema(
self.available_interfaces or [],
user_input.get(CONF_TRACKER_INTERFACES),
),
user_input,
),
errors=errors or {},
)
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle user step: credentials and connection test."""
errors = {}
if user_input is None:
return await self._show_setup_form(user_input, None)
verify_ssl = user_input[CONF_VERIFY_SSL]
session = async_get_clientsession(self.hass, verify_ssl=verify_ssl)
client = OPNsenseClient(
user_input[CONF_URL],
user_input[CONF_API_KEY],
user_input[CONF_API_SECRET],
session,
opts={"verify_ssl": verify_ssl},
)
try:
await client.validate()
interfaces_resp = await client.get_interfaces()
known_interfaces = [
name
for ifinfo in interfaces_resp.values()
if (name := ifinfo.get("name"))
]
self.available_interfaces = list(known_interfaces)
except OPNsenseInvalidAuth:
errors["base"] = "invalid_auth"
except OPNsensePrivilegeMissing:
errors["base"] = "privilege_missing"
except OPNsenseInvalidURL:
errors["base"] = "invalid_url"
except OPNsenseSSLError:
errors["base"] = "ssl_error"
except OPNsenseConnectionError, OPNsenseTimeoutError:
errors["base"] = "cannot_connect"
except OPNsenseUnknownFirmware:
errors["base"] = "unknown_version"
except OPNsenseBelowMinFirmware:
errors["base"] = "invalid_version"
except Exception:
_LOGGER.exception("Unexpected exception")
errors["base"] = "unknown"
else:
unique_id = await client.get_device_unique_id()
if not unique_id:
return self.async_abort(reason="no_unique_id")
await self.async_set_unique_id(unique_id)
self._abort_if_unique_id_configured()
self._entry_data = user_input
return await self.async_step_interfaces()
return await self._show_setup_form(user_input, errors)
async def async_step_interfaces(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle tracker interface selection step."""
if user_input is None:
return await self._show_interfaces_form({}, None)
if user_input.get(CONF_TRACKER_INTERFACES):
self._entry_data[CONF_TRACKER_INTERFACES] = user_input[
CONF_TRACKER_INTERFACES
]
return self.async_create_entry(
title=self._entry_data[CONF_URL], data=self._entry_data
)
async def async_step_import(self, import_data: dict[str, Any]) -> ConfigFlowResult:
"""Import a Yaml config."""
# Test connection
session = async_get_clientsession(
self.hass, verify_ssl=import_data[CONF_VERIFY_SSL]
)
client = OPNsenseClient(
import_data[CONF_URL],
import_data[CONF_API_KEY],
import_data[CONF_API_SECRET],
session,
opts={"verify_ssl": import_data[CONF_VERIFY_SSL]},
)
try:
await client.validate()
interfaces_resp = await client.get_interfaces()
except OPNsenseInvalidURL:
return self._abort_import(reason="invalid_url")
except OPNsenseInvalidAuth:
return self._abort_import(reason="invalid_auth")
except OPNsensePrivilegeMissing:
return self._abort_import(reason="privilege_missing")
except OPNsenseSSLError:
return self._abort_import(reason="ssl_error")
except OPNsenseConnectionError, OPNsenseTimeoutError:
return self._abort_import(reason="cannot_connect")
except OPNsenseUnknownFirmware:
return self._abort_import(reason="unknown_version")
except OPNsenseBelowMinFirmware:
return self._abort_import(reason="invalid_version")
except Exception: # Allowed in config flows
_LOGGER.exception("Unexpected exception during import")
return self._abort_import(reason="unknown")
async_create_issue(
self.hass,
HOMEASSISTANT_DOMAIN,
f"deprecated_yaml_{DOMAIN}",
breaks_in_ha_version="2026.12.0",
is_fixable=False,
issue_domain=DOMAIN,
severity=IssueSeverity.WARNING,
translation_key="deprecated_yaml",
translation_placeholders={
"domain": DOMAIN,
"integration_title": "OPNsense",
},
)
unique_id = await client.get_device_unique_id()
if not unique_id:
return self._abort_import(reason="no_unique_id")
await self.async_set_unique_id(unique_id)
self._abort_if_unique_id_configured()
# Validate CONF_TRACKER_INTERFACES if present and not empty
verified_data = dict(import_data)
if CONF_TRACKER_INTERFACES in verified_data:
if not verified_data[CONF_TRACKER_INTERFACES]:
verified_data.pop(CONF_TRACKER_INTERFACES)
else:
known_interfaces = [
name
for ifinfo in interfaces_resp.values()
if (name := ifinfo.get("name"))
]
self.available_interfaces = sorted(known_interfaces)
# Abort import if any specified tracker interface is not found
missing = [
intf_description
for intf_description in verified_data[CONF_TRACKER_INTERFACES]
if intf_description not in known_interfaces
]
if missing:
# Create a repair to guide the user
async_create_issue(
self.hass,
DOMAIN,
"import_failed_missing_interfaces",
breaks_in_ha_version="2026.12.0",
is_fixable=False,
severity=IssueSeverity.ERROR,
translation_key="import_failed_missing_interfaces",
translation_placeholders={
"missing": ", ".join(missing),
"found": ", ".join(known_interfaces),
"integration_title": "OPNsense",
},
)
return self.async_abort(
reason="import_failed_missing_interfaces",
description_placeholders={
"missing": ", ".join(missing),
"found": ", ".join(known_interfaces),
"integration_title": "OPNsense",
},
)
# Clear any previous import issues if interfaces are now valid
async_delete_issue(
self.hass,
DOMAIN,
"import_failed_missing_interfaces",
)
return self.async_create_entry(
title=verified_data[CONF_URL], data=verified_data
)
def _abort_import(self, reason: str) -> ConfigFlowResult:
"""Create an issue for import errors and abort the import."""
async_create_issue(
self.hass,
DOMAIN,
f"import_failed_{reason}",
breaks_in_ha_version="2026.12.0",
is_fixable=False,
severity=IssueSeverity.ERROR,
translation_key=f"import_failed_{reason}",
translation_placeholders={
"domain": DOMAIN,
"integration_title": "OPNsense",
},
)
return self.async_abort(
reason=reason,
description_placeholders={
"integration_title": "OPNsense",
},
)
+5 -2
View File
@@ -1,8 +1,11 @@
"""Constants for OPNsense component."""
from datetime import timedelta
DOMAIN = "opnsense"
OPNSENSE_DATA = DOMAIN
CONF_API_SECRET = "api_secret"
CONF_INTERFACE_CLIENT = "interface_client"
CONF_TRACKER_INTERFACES = "tracker_interfaces"
# Update interval for device scanning
SCAN_INTERVAL = timedelta(seconds=30)
@@ -0,0 +1,80 @@
"""Coordinator for OPNsense device tracker updates."""
import logging
from aiopnsense import (
OPNsenseBelowMinFirmware,
OPNsenseClient,
OPNsenseConnectionError,
OPNsenseInvalidAuth,
OPNsenseInvalidURL,
OPNsensePrivilegeMissing,
OPNsenseSSLError,
OPNsenseTimeoutError,
OPNsenseUnknownFirmware,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryError
from homeassistant.helpers.device_registry import format_mac
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import SCAN_INTERVAL
from .types import DeviceDetails, DeviceDetailsByMAC, OPNsenseConfigEntry
_LOGGER = logging.getLogger(__name__)
class OPNsenseDeviceTrackerCoordinator(DataUpdateCoordinator[DeviceDetailsByMAC]):
"""Coordinator for OPNsense device tracker updates."""
def __init__(
self,
hass: HomeAssistant,
config_entry: OPNsenseConfigEntry,
client: OPNsenseClient,
interfaces: list[str],
) -> None:
"""Initialize the coordinator."""
super().__init__(
hass,
_LOGGER,
name="OPNsense Device Tracker",
update_interval=SCAN_INTERVAL,
config_entry=config_entry,
)
self.client = client
self.interfaces = interfaces
self.tracked_devices: set[str] = set()
def _get_mac_addrs(self, devices: list[DeviceDetails]) -> DeviceDetailsByMAC:
"""Create dict with mac address keys from list of devices."""
out_devices: DeviceDetailsByMAC = {}
for device in devices:
if not self.interfaces or device["intf_description"] in self.interfaces:
formatted_mac = format_mac(device["mac"])
out_devices[formatted_mac] = device
return out_devices
async def _async_update_data(self) -> DeviceDetailsByMAC:
"""Fetch data from OPNsense."""
try:
devices = await self.client.get_arp_table(True)
except (
OPNsenseInvalidAuth,
OPNsenseInvalidURL,
OPNsensePrivilegeMissing,
OPNsenseSSLError,
OPNsenseBelowMinFirmware,
OPNsenseUnknownFirmware,
) as err:
raise ConfigEntryError(f"Error with OPNsense configuration: {err}") from err
except (
OPNsenseConnectionError,
OPNsenseTimeoutError,
) as err:
raise UpdateFailed(
f"Error communicating with OPNsense router: {err}"
) from err
return self._get_mac_addrs(devices)
@@ -1,71 +1,117 @@
"""Device tracker support for OPNsense routers."""
from typing import Any, NewType
from typing import Any
from aiopnsense import OPNsenseClient
from homeassistant.components.device_tracker import DeviceScanner
from homeassistant.components.device_tracker import ScannerEntity
from homeassistant.core import HomeAssistant
from homeassistant.helpers.typing import ConfigType
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import CONF_INTERFACE_CLIENT, CONF_TRACKER_INTERFACES, OPNSENSE_DATA
DeviceDetails = NewType("DeviceDetails", dict[str, Any])
DeviceDetailsByMAC = NewType("DeviceDetailsByMAC", dict[str, DeviceDetails])
from .coordinator import OPNsenseDeviceTrackerCoordinator
from .types import DeviceDetails, OPNsenseConfigEntry
async def async_get_scanner(
hass: HomeAssistant, config: ConfigType
) -> DeviceScanner | None:
"""Configure the OPNsense device_tracker."""
return OPNsenseDeviceScanner(
hass.data[OPNSENSE_DATA][CONF_INTERFACE_CLIENT],
hass.data[OPNSENSE_DATA][CONF_TRACKER_INTERFACES],
)
async def async_setup_entry(
hass: HomeAssistant,
entry: OPNsenseConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up device tracker for OPNsense component."""
client = entry.runtime_data.client
interfaces = entry.runtime_data.tracker_interfaces
coordinator = OPNsenseDeviceTrackerCoordinator(hass, entry, client, interfaces)
def _async_add_new_entities() -> None:
"""Add entities for newly discovered devices."""
if not coordinator.data:
return
entities = []
for mac_address in coordinator.data:
if mac_address in coordinator.tracked_devices:
continue
entity = OPNsenseDeviceTrackerEntity(coordinator, mac_address)
coordinator.tracked_devices.add(mac_address)
entities.append(entity)
if entities:
async_add_entities(entities)
entry.async_on_unload(coordinator.async_add_listener(_async_add_new_entities))
# Initial data fetch
await coordinator.async_config_entry_first_refresh()
_async_add_new_entities()
class OPNsenseDeviceScanner(DeviceScanner):
"""This class queries a router running OPNsense."""
class OPNsenseDeviceTrackerEntity(
CoordinatorEntity[OPNsenseDeviceTrackerCoordinator], ScannerEntity
):
"""Representation of a tracked device."""
def __init__(self, client: OPNsenseClient, interfaces: list[str]) -> None:
"""Initialize the scanner."""
self.last_results: dict[str, Any] = {}
self.client = client
self.interfaces = interfaces
def __init__(
self,
coordinator: OPNsenseDeviceTrackerCoordinator,
mac_address: str,
) -> None:
"""Initialize the device tracker entity."""
super().__init__(coordinator)
self._attr_mac_address = mac_address
def _get_mac_addrs(self, devices: list[DeviceDetails]) -> DeviceDetailsByMAC | dict:
"""Create dict with mac address keys from list of devices."""
out_devices = {}
for device in devices:
if not self.interfaces or device["intf_description"] in self.interfaces:
out_devices[device["mac"]] = device
return out_devices
@property
def device_data(self) -> DeviceDetails | None:
"""Return device data for current device."""
if self.coordinator.data and self.mac_address in self.coordinator.data:
return self.coordinator.data[self.mac_address]
return None
async def async_scan_devices(self) -> list[str]:
"""Scan for new devices and return a list with found device IDs."""
await self._async_update_info()
return list(self.last_results)
@property
def is_connected(self) -> bool:
"""Return true if the device is connected to the network."""
return (
self.coordinator.data is not None
and self.mac_address in self.coordinator.data
)
def get_device_name(self, device: str) -> str | None:
"""Return the name of the given device or None if we don't know."""
if device not in self.last_results:
return None
return self.last_results[device].get("hostname") or None
@property
def name(self) -> str:
"""Return device name."""
device_data = self.device_data
if device_data and device_data.get("hostname"):
return str(device_data["hostname"])
return f"OPNsense {self.mac_address}"
async def _async_update_info(self) -> bool:
"""Ensure the information from the OPNsense router is up to date.
@property
def ip_address(self) -> str | None:
"""Return the primary IP address of the device."""
device_data = self.device_data
if device_data:
return device_data.get("ip")
return None
Return boolean if scanning successful.
"""
devices = await self.client.get_arp_table(True)
self.last_results = self._get_mac_addrs(devices)
return True
@property
def hostname(self) -> str | None:
"""Return hostname of the device."""
device_data = self.device_data
if device_data:
hostname = device_data.get("hostname")
return hostname or None
return None
def get_extra_attributes(self, device: str) -> dict[Any, Any]:
"""Return the extra attrs of the given device."""
if device not in self.last_results:
@property
def extra_state_attributes(self) -> dict[str, Any]:
"""Return the state attributes."""
device_data = self.device_data
if not device_data:
return {}
mfg = self.last_results[device].get("manufacturer")
if not mfg:
return {}
return {"manufacturer": mfg}
attrs = {}
if manufacturer := device_data.get("manufacturer"):
attrs["manufacturer"] = manufacturer
if interface := device_data.get("intf_description"):
attrs["interface"] = interface
if expires := device_data.get("expires"):
attrs["expires"] = expires
return attrs
@@ -2,6 +2,7 @@
"domain": "opnsense",
"name": "OPNsense",
"codeowners": ["@HarlemSquirrel", "@Snuffy2"],
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/opnsense",
"integration_type": "hub",
"iot_class": "local_polling",
@@ -0,0 +1,117 @@
{
"config": {
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
"already_in_progress": "[%key:common::config_flow::abort::already_in_progress%]",
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"import_failed_missing_interfaces": "The following OPNsense tracker interfaces were not found: {missing}. Found interfaces were: {found}. Please manually set up a config entry and remove the YAML config",
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
"invalid_url": "URL is invalid or unreachable",
"invalid_version": "Unsupported OPNsense firmware version",
"no_unique_id": "Could not determine a unique identifier for this OPNsense router. Please ensure the API key has sufficient privileges, and your OPNsense instance has physical ports with a MAC address.",
"privilege_missing": "The API key used does not have sufficient privileges. Please check the integration documentation for required permissions",
"ssl_error": "SSL certificate verification failed",
"unknown": "[%key:common::config_flow::error::unknown%]",
"unknown_version": "The OPNsense firmware version is unknown. Please ensure you are running a supported version of OPNsense and the user has the correct permissions."
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
"invalid_interface": "Interface(s) do not exist",
"invalid_url": "URL is invalid or unreachable",
"invalid_version": "Unsupported OPNsense firmware version",
"privilege_missing": "[%key:component::opnsense::config::abort::privilege_missing%]",
"ssl_error": "SSL certificate verification failed",
"unknown": "[%key:common::config_flow::error::unknown%]",
"unknown_version": "The OPNsense firmware version is unknown. Please ensure you are running a supported version of OPNsense and the user has the correct permissions."
},
"step": {
"interfaces": {
"data": {
"tracker_interfaces": "Interface(s) to use for tracking devices"
},
"description": "Select the OPNsense interfaces to use for tracking devices. If no interfaces are selected then all interfaces will be used for tracking."
},
"user": {
"data": {
"api_key": "[%key:common::config_flow::data::api_key%]",
"api_secret": "API secret",
"url": "[%key:common::config_flow::data::url%]",
"verify_ssl": "[%key:common::config_flow::data::verify_ssl%]"
},
"description": "Set required parameters to connect to your router. For more information, please refer to the [integration documentation]({doc_url})"
}
}
},
"exceptions": {
"cannot_connect": {
"message": "Connection failure while connecting to OPNsense API endpoint at {url}"
},
"firmware_too_old": {
"message": "OPNsense firmware at {url} is below the minimum supported version"
},
"invalid_auth": {
"message": "Authentication failure while connecting to OPNsense API endpoint at {url}"
},
"invalid_url": {
"message": "Invalid URL while connecting to OPNsense API endpoint at {url}"
},
"privilege_missing": {
"message": "The API user connecting to {url} does not have sufficient privileges"
},
"ssl_error": {
"message": "Unable to verify SSL certificate while connecting to OPNsense API endpoint at {url}"
},
"timeout_connecting": {
"message": "Timeout while connecting to OPNsense API endpoint at {url}"
},
"tracker_interface_not_found": {
"message": "Configured tracker interface {interface} is not present on the OPNsense router. Known interfaces: {known}"
},
"unknown_firmware": {
"message": "Could not determine the OPNsense firmware version at {url}"
}
},
"issues": {
"import_failed_cannot_connect": {
"description": "Configuring {integration_title} via YAML is deprecated and will be removed in a future release. While importing your configuration, a connection error occurred. Please correct your YAML configuration and restart Home Assistant, or remove the {domain} key from your configuration and configure the integration via the UI.",
"title": "The {integration_title} YAML configuration is being removed"
},
"import_failed_invalid_auth": {
"description": "Configuring {integration_title} via YAML is deprecated and will be removed in a future release. While importing your configuration, an authentication error occurred. Please correct your YAML configuration and restart Home Assistant, or remove the {domain} key from your configuration and configure the integration via the UI.",
"title": "The {integration_title} YAML configuration is being removed"
},
"import_failed_invalid_url": {
"description": "Configuring {integration_title} via YAML is deprecated and will be removed in a future release. While importing your configuration, an error occurred indicating the URL provided is invalid or unreachable. Please correct your YAML configuration and restart Home Assistant, or remove the {domain} key from your configuration and configure the integration via the UI.",
"title": "The {integration_title} YAML configuration is being removed"
},
"import_failed_invalid_version": {
"description": "Configuring {integration_title} via YAML is deprecated and will be removed in a future release. While importing your configuration, an error occurred indicating the OPNsense firmware version is unsupported. Please correct your YAML configuration and restart Home Assistant, or remove the {domain} key from your configuration and configure the integration via the UI.",
"title": "The {integration_title} YAML configuration is being removed"
},
"import_failed_missing_interfaces": {
"description": "The following OPNsense tracker interfaces were not found: {missing}. Found interfaces were: {found}. Please manually set up a config entry and remove the YAML config",
"title": "The {integration_title} YAML import failed: Missing tracker interfaces"
},
"import_failed_no_unique_id": {
"description": "Configuring {integration_title} via YAML is deprecated and will be removed in a future release. While importing your configuration, a unique identifier for the router could not be determined. Please ensure the API key has sufficient privileges, and your OPNsense instance has physical ports with a MAC address.",
"title": "The {integration_title} YAML configuration is being removed"
},
"import_failed_privilege_missing": {
"description": "Configuring {integration_title} via YAML is deprecated and will be removed in a future release. While importing your configuration, an error occurred indicating the API key used does not have sufficient privileges. Please check the integration documentation for required permissions, correct your YAML configuration, and restart Home Assistant, or remove the {domain} key from your configuration and configure the integration via the UI.",
"title": "The {integration_title} YAML configuration is being removed"
},
"import_failed_ssl_error": {
"description": "Configuring {integration_title} via YAML is deprecated and will be removed in a future release. While importing your configuration, an SSL error occurred. Please correct your YAML configuration and restart Home Assistant, or remove the {domain} key from your configuration and configure the integration via the UI.",
"title": "The {integration_title} YAML configuration is being removed"
},
"import_failed_unknown": {
"description": "Configuring {integration_title} via YAML is deprecated and will be removed in a future release. While importing your configuration, an unknown error occurred. Please correct your YAML configuration and restart Home Assistant, or remove the {domain} key from your configuration and configure the integration via the UI.",
"title": "The {integration_title} YAML configuration is being removed"
},
"import_failed_unknown_version": {
"description": "Configuring {integration_title} via YAML is deprecated and will be removed in a future release. While importing your configuration, an error occurred indicating the OPNsense firmware version is unknown. Please correct your YAML configuration and restart Home Assistant, or remove the {domain} key from your configuration and configure the integration via the UI.",
"title": "The {integration_title} YAML configuration is being removed"
}
}
}
@@ -0,0 +1,21 @@
"""Types for OPNsense routers."""
from dataclasses import dataclass
from typing import Any
from aiopnsense import OPNsenseClient
from homeassistant.config_entries import ConfigEntry
@dataclass(slots=True)
class OPNsenseRuntimeData:
"""Runtime data for OPNsense config entries."""
client: OPNsenseClient
tracker_interfaces: list[str]
type DeviceDetails = dict[str, Any]
type DeviceDetailsByMAC = dict[str, DeviceDetails]
type OPNsenseConfigEntry = ConfigEntry[OPNsenseRuntimeData]
+1 -1
View File
@@ -14,5 +14,5 @@
"iot_class": "local_polling",
"loggers": ["pyvlx"],
"quality_scale": "silver",
"requirements": ["pyvlx==0.2.34"]
"requirements": ["pyvlx==0.2.35"]
}
-1
View File
@@ -118,7 +118,6 @@ class AbortFlow(FlowError):
class FlowContext(TypedDict, total=False):
"""Typed context dict."""
show_advanced_options: bool
source: str
+1
View File
@@ -541,6 +541,7 @@ FLOWS = {
"opentherm_gw",
"openuv",
"openweathermap",
"opnsense",
"opower",
"oralb",
"orvibo",
+1 -1
View File
@@ -5119,7 +5119,7 @@
"opnsense": {
"name": "OPNsense",
"integration_type": "hub",
"config_flow": false,
"config_flow": true,
"iot_class": "local_polling"
},
"opower": {
+1 -2
View File
@@ -60,7 +60,6 @@ class FlowManagerIndexView(_BaseFlowManagerView[_FlowManagerT]):
vol.Schema(
{
vol.Required("handler"): str,
vol.Optional("show_advanced_options", default=False): cv.boolean,
},
extra=vol.ALLOW_EXTRA,
)
@@ -93,7 +92,7 @@ class FlowManagerIndexView(_BaseFlowManagerView[_FlowManagerT]):
def get_context(self, data: dict[str, Any]) -> dict[str, Any]:
"""Return context."""
return {"show_advanced_options": data["show_advanced_options"]}
return {}
class FlowManagerResourceView(_BaseFlowManagerView[_FlowManagerT]):
+1 -1
View File
@@ -2791,7 +2791,7 @@ pyvesync==3.4.1
pyvizio==0.1.61
# homeassistant.components.velux
pyvlx==0.2.34
pyvlx==0.2.35
# homeassistant.components.volumio
pyvolumio==0.1.5
-6
View File
@@ -1176,8 +1176,6 @@ class MockConfigEntry(config_entries.ConfigEntry):
async def start_reconfigure_flow(
self,
hass: HomeAssistant,
*,
show_advanced_options: bool = False,
) -> ConfigFlowResult:
"""Start a reconfiguration flow."""
if self.entry_id not in hass.config_entries._entries:
@@ -1189,7 +1187,6 @@ class MockConfigEntry(config_entries.ConfigEntry):
context={
"source": config_entries.SOURCE_RECONFIGURE,
"entry_id": self.entry_id,
"show_advanced_options": show_advanced_options,
},
)
@@ -1197,8 +1194,6 @@ class MockConfigEntry(config_entries.ConfigEntry):
self,
hass: HomeAssistant,
subentry_id: str,
*,
show_advanced_options: bool = False,
) -> ConfigFlowResult:
"""Start a subentry reconfiguration flow."""
if self.entry_id not in hass.config_entries._entries:
@@ -1212,7 +1207,6 @@ class MockConfigEntry(config_entries.ConfigEntry):
context={
"source": config_entries.SOURCE_RECONFIGURE,
"subentry_id": subentry_id,
"show_advanced_options": show_advanced_options,
},
)
+86 -68
View File
@@ -1,5 +1,6 @@
"""Tests for the Cast config flow."""
from typing import Any
from unittest.mock import ANY, patch
import pytest
@@ -13,6 +14,17 @@ from homeassistant.data_entry_flow import FlowResultType
from tests.common import MockConfigEntry, get_schema_suggested_value
def _get_schema_suggested_values(data_schema, keys: list[str]) -> dict[str, Any]:
"""Get suggested values from a data schema."""
suggested_values = {}
for key in keys:
if (
suggested_value := get_schema_suggested_value(data_schema, key)
) is not None:
suggested_values[key] = suggested_value
return suggested_values
async def test_creating_entry_sets_up_media_player(hass: HomeAssistant) -> None:
"""Test setting up Cast loads the media player."""
with (
@@ -142,50 +154,89 @@ async def test_zeroconf_setup_onboarding(hass: HomeAssistant) -> None:
@pytest.mark.parametrize(
("parameter", "initial", "suggested", "user_input", "updated"),
("initial", "expected_suggested_values", "user_input", "updated"),
[
(
"known_hosts",
["192.168.0.10", "192.168.0.11"],
["192.168.0.10", "192.168.0.11"],
["192.168.0.1", " ", " 192.168.0.2 "],
["192.168.0.1", "192.168.0.2"],
{},
{},
{"more_options": {}},
{"ignore_cec": [], "known_hosts": [], "user_id": ANY, "uuid": []},
),
(
"uuid",
["bla", "blu"],
["bla", "blu"],
["foo", " ", " bar "],
["foo", "bar"],
{"ignore_cec": [], "known_hosts": [], "uuid": []},
{"ignore_cec": [], "known_hosts": [], "uuid": []},
{"more_options": {}},
{"ignore_cec": [], "known_hosts": [], "user_id": ANY, "uuid": []},
),
(
"ignore_cec",
["cast1", "cast2"],
["cast1", "cast2"],
["other_cast", " ", " some_cast "],
["other_cast", "some_cast"],
{
"ignore_cec": ["cast1", "cast2"],
"known_hosts": ["192.168.0.10", "192.168.0.11"],
"uuid": ["bla", "blu"],
},
{
"ignore_cec": ["cast1", "cast2"],
"known_hosts": ["192.168.0.10", "192.168.0.11"],
"uuid": ["bla", "blu"],
},
{
"known_hosts": ["192.168.0.1", " ", " 192.168.0.2 "],
"more_options": {
"ignore_cec": ["other_cast", " ", " some_cast "],
"uuid": ["foo", " ", " bar "],
},
},
{
"ignore_cec": ["other_cast", "some_cast"],
"known_hosts": ["192.168.0.1", "192.168.0.2"],
"user_id": ANY,
"uuid": ["foo", "bar"],
},
),
# Implicit clearing of the lists when not passing values
(
{
"ignore_cec": ["cast1", "cast2"],
"known_hosts": ["192.168.0.10", "192.168.0.11"],
"uuid": ["bla", "blu"],
},
{
"ignore_cec": ["cast1", "cast2"],
"known_hosts": ["192.168.0.10", "192.168.0.11"],
"uuid": ["bla", "blu"],
},
{"more_options": {}},
{"ignore_cec": [], "known_hosts": [], "user_id": ANY, "uuid": []},
),
# Explicit clearing of the lists
(
{
"ignore_cec": ["cast1", "cast2"],
"known_hosts": ["192.168.0.10", "192.168.0.11"],
"uuid": ["bla", "blu"],
},
{
"ignore_cec": ["cast1", "cast2"],
"known_hosts": ["192.168.0.10", "192.168.0.11"],
"uuid": ["bla", "blu"],
},
{"known_hosts": [], "more_options": {"ignore_cec": [], "uuid": []}},
{"ignore_cec": [], "known_hosts": [], "user_id": ANY, "uuid": []},
),
],
)
async def test_option_flow(
hass: HomeAssistant,
parameter: str,
initial: list[str],
suggested: str | list[str],
user_input: str | list[str],
updated: list[str],
initial: dict[str, Any],
expected_suggested_values: dict[str, Any],
user_input: dict[str, Any],
updated: dict[str, Any],
) -> None:
"""Test config flow options."""
basic_parameters = ["known_hosts"]
extra_parameters = ["ignore_cec", "uuid"]
data = {
"ignore_cec": [],
"known_hosts": [],
"uuid": [],
}
data[parameter] = initial
config_entry = MockConfigEntry(domain="cast", data=data)
config_entry = MockConfigEntry(domain="cast", data=initial)
config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
@@ -199,54 +250,21 @@ async def test_option_flow(
more_options_schema = data_schema["more_options"].schema.schema
assert set(more_options_schema) == {"ignore_cec", "uuid"}
orig_data = dict(config_entry.data)
# Check suggested values
for other_param in basic_parameters:
if other_param == parameter:
continue
assert get_schema_suggested_value(data_schema, other_param) == []
if parameter in basic_parameters:
assert get_schema_suggested_value(data_schema, parameter) == suggested
for other_param in extra_parameters:
if other_param == parameter:
continue
assert get_schema_suggested_value(more_options_schema, other_param) == []
if parameter in extra_parameters:
assert get_schema_suggested_value(more_options_schema, parameter) == suggested
suggested_values = _get_schema_suggested_values(data_schema, basic_parameters)
suggested_values |= _get_schema_suggested_values(
more_options_schema, extra_parameters
)
assert suggested_values == expected_suggested_values
# Reconfigure
user_input_dict = {"more_options": {}}
if parameter in basic_parameters:
user_input_dict[parameter] = user_input
if parameter in extra_parameters:
user_input_dict["more_options"][parameter] = user_input
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input=user_input_dict,
user_input=user_input,
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["data"] == {}
for other_param in basic_parameters:
if other_param == parameter:
continue
assert config_entry.data[other_param] == []
for other_param in extra_parameters:
if other_param == parameter:
continue
assert config_entry.data[other_param] == []
assert config_entry.data[parameter] == updated
# Clear lists
result = await hass.config_entries.options.async_init(config_entry.entry_id)
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={"more_options": {}},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["data"] == {}
expected_data = {**orig_data, "ignore_cec": [], "known_hosts": [], "uuid": []}
assert dict(config_entry.data) == expected_data
assert config_entry.data == updated
async def test_known_hosts(hass: HomeAssistant, castbrowser_mock) -> None:
@@ -419,7 +419,7 @@ async def test_initialize_flow(hass: HomeAssistant, client: TestClient) -> None:
with mock_config_flow("test", TestFlow):
resp = await client.post(
"/api/config/config_entries/flow",
json={"handler": "test", "show_advanced_options": True},
json={"handler": "test"},
)
assert resp.status == HTTPStatus.OK
@@ -469,7 +469,7 @@ async def test_initialize_flow_unmet_dependency(
with mock_config_flow("test2", TestFlow):
resp = await client.post(
"/api/config/config_entries/flow",
json={"handler": "test2", "show_advanced_options": True},
json={"handler": "test2"},
)
assert resp.status == HTTPStatus.BAD_REQUEST
+3 -3
View File
@@ -71,12 +71,12 @@ async def test_form(hass: HomeAssistant) -> None:
assert len(mock_setup_entry.mock_calls) == 1
async def test_form_adv(hass: HomeAssistant) -> None:
"""Test we get the form with advanced options on."""
async def test_form_with_advanced_options(hass: HomeAssistant) -> None:
"""Test we can submit the form with custom resolver and port options."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_USER, "show_advanced_options": True},
context={"source": config_entries.SOURCE_USER},
)
assert result["data_schema"] == DATA_SCHEMA
+7 -13
View File
@@ -317,20 +317,16 @@ async def test_options(
@pytest.mark.parametrize(
("group_type", "extra_options", "extra_options_after", "advanced"),
("group_type", "extra_options", "extra_options_after"),
[
("light", {"all": False}, {"all": False}, False),
("light", {"all": True}, {"all": False}, False),
("light", {"all": False}, {"all": False}, True),
("light", {"all": True}, {"all": False}, True),
("switch", {"all": False}, {"all": False}, False),
("switch", {"all": True}, {"all": False}, False),
("switch", {"all": False}, {"all": False}, True),
("switch", {"all": True}, {"all": False}, True),
("light", {"all": False}, {"all": False}),
("light", {"all": True}, {"all": False}),
("switch", {"all": False}, {"all": False}),
("switch", {"all": True}, {"all": False}),
],
)
async def test_all_options(
hass: HomeAssistant, group_type, extra_options, extra_options_after, advanced
hass: HomeAssistant, group_type, extra_options, extra_options_after
) -> None:
"""Test reconfiguring."""
members1 = [f"{group_type}.one", f"{group_type}.two"]
@@ -356,9 +352,7 @@ async def test_all_options(
config_entry = hass.config_entries.async_entries(DOMAIN)[0]
result = await hass.config_entries.options.async_init(
config_entry.entry_id, context={"show_advanced_options": advanced}
)
result = await hass.config_entries.options.async_init(config_entry.entry_id)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == group_type
@@ -0,0 +1,57 @@
# serializer version: 1
# name: test_states
set({
StateSnapshot({
'attributes': ReadOnlyDict({
'editable': True,
'friendly_name': 'test home',
'icon': 'mdi:home',
'latitude': 32.87336,
'longitude': -117.22743,
'passive': False,
'persons': list([
]),
'radius': 100,
}),
'context': <ANY>,
'entity_id': 'zone.home',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '0',
}),
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Demo scanner',
'in_zones': list([
'zone.home',
]),
'source_type': <SourceType.ROUTER: 'router'>,
}),
'context': <ANY>,
'entity_id': 'device_tracker.demo_scanner',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'home',
}),
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Demo tracker',
'gps_accuracy': 10,
'in_zones': list([
'zone.home',
]),
'latitude': 32.87336,
'longitude': -117.22743,
'source_type': <SourceType.GPS: 'gps'>,
}),
'context': <ANY>,
'entity_id': 'device_tracker.demo_tracker',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'home',
}),
})
# ---
@@ -0,0 +1,146 @@
"""The tests for the kitchen_sink device_tracker platform."""
from collections.abc import Generator
from unittest.mock import patch
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.device_tracker import ATTR_SOURCE_TYPE, SourceType
from homeassistant.components.kitchen_sink import DOMAIN
from homeassistant.components.kitchen_sink.services import (
ATTR_ACCURACY,
ATTR_CONNECTED,
SERVICE_SET_SCANNER_CONNECTED,
SERVICE_SET_TRACKER_LOCATION,
)
from homeassistant.const import (
ATTR_ENTITY_ID,
ATTR_GPS_ACCURACY,
ATTR_LATITUDE,
ATTR_LONGITUDE,
STATE_HOME,
STATE_NOT_HOME,
Platform,
)
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
TRACKER_ENTITY_ID = "device_tracker.demo_tracker"
SCANNER_ENTITY_ID = "device_tracker.demo_scanner"
@pytest.fixture
def device_tracker_only() -> Generator[None]:
"""Enable only the device_tracker platform."""
with patch(
"homeassistant.components.kitchen_sink.COMPONENTS_WITH_DEMO_PLATFORM",
[Platform.DEVICE_TRACKER],
):
yield
@pytest.fixture(autouse=True)
async def setup_comp(hass: HomeAssistant, device_tracker_only: None) -> None:
"""Set up demo component."""
hass.config.latitude = 32.87336
hass.config.longitude = -117.22743
assert await async_setup_component(hass, DOMAIN, {DOMAIN: {}})
await hass.async_block_till_done()
async def test_states(hass: HomeAssistant, snapshot: SnapshotAssertion) -> None:
"""Test the expected device_tracker entities are added."""
states = hass.states.async_all()
assert set(states) == snapshot
async def test_set_tracker_location(hass: HomeAssistant) -> None:
"""Test the set_tracker_location service updates tracker attributes."""
state = hass.states.get(TRACKER_ENTITY_ID)
assert state is not None
assert state.attributes[ATTR_SOURCE_TYPE] == SourceType.GPS
await hass.services.async_call(
DOMAIN,
SERVICE_SET_TRACKER_LOCATION,
{
ATTR_ENTITY_ID: TRACKER_ENTITY_ID,
ATTR_LATITUDE: 12.34,
ATTR_LONGITUDE: 56.78,
ATTR_ACCURACY: 42,
},
blocking=True,
)
state = hass.states.get(TRACKER_ENTITY_ID)
assert state.attributes[ATTR_LATITUDE] == 12.34
assert state.attributes[ATTR_LONGITUDE] == 56.78
assert state.attributes[ATTR_GPS_ACCURACY] == 42
assert state.state == STATE_NOT_HOME
async def test_set_scanner_connected(hass: HomeAssistant) -> None:
"""Test the set_scanner_connected service updates scanner state."""
state = hass.states.get(SCANNER_ENTITY_ID)
assert state is not None
assert state.state == STATE_HOME
assert state.attributes[ATTR_SOURCE_TYPE] == SourceType.ROUTER
await hass.services.async_call(
DOMAIN,
SERVICE_SET_SCANNER_CONNECTED,
{ATTR_ENTITY_ID: SCANNER_ENTITY_ID, ATTR_CONNECTED: False},
blocking=True,
)
state = hass.states.get(SCANNER_ENTITY_ID)
assert state.state == STATE_NOT_HOME
await hass.services.async_call(
DOMAIN,
SERVICE_SET_SCANNER_CONNECTED,
{ATTR_ENTITY_ID: SCANNER_ENTITY_ID, ATTR_CONNECTED: True},
blocking=True,
)
state = hass.states.get(SCANNER_ENTITY_ID)
assert state.state == STATE_HOME
async def test_set_tracker_location_on_scanner_raises(hass: HomeAssistant) -> None:
"""Calling set_tracker_location on the scanner surfaces an AttributeError.
The service is registered for the device_tracker domain and dispatches by
method name, so targeting the scanner (which has no async_set_tracker_location)
bubbles up the missing-attribute error from the entity.
"""
with pytest.raises(
AttributeError,
match="'DemoScanner' object has no attribute 'async_set_tracker_location'",
):
await hass.services.async_call(
DOMAIN,
SERVICE_SET_TRACKER_LOCATION,
{
ATTR_ENTITY_ID: SCANNER_ENTITY_ID,
ATTR_LATITUDE: 12.34,
ATTR_LONGITUDE: 56.78,
ATTR_ACCURACY: 42,
},
blocking=True,
)
async def test_set_scanner_connected_on_tracker_raises(hass: HomeAssistant) -> None:
"""Calling set_scanner_connected on the tracker surfaces an AttributeError."""
with pytest.raises(
AttributeError,
match="'DemoTracker' object has no attribute 'async_set_scanner_connected'",
):
await hass.services.async_call(
DOMAIN,
SERVICE_SET_SCANNER_CONNECTED,
{ATTR_ENTITY_ID: TRACKER_ENTITY_ID, ATTR_CONNECTED: False},
blocking=True,
)
+1 -1
View File
@@ -369,7 +369,7 @@ async def test_service(
await hass.services.async_call(
DOMAIN,
"test_service_1",
{"field_1": 1, "field_2": "auto", "field_3": 1, "field_4": "forwards"},
{"field_1": 1, "field_2": "auto", "field_3": 1, "field_4": "forward"},
blocking=True,
)
+59
View File
@@ -0,0 +1,59 @@
"""OPNsense session fixtures."""
from collections.abc import Generator
from unittest.mock import AsyncMock, patch
import pytest
from homeassistant.components.opnsense.const import DOMAIN
from homeassistant.core import HomeAssistant
from .const import ARP, CONFIG_DATA, INTERFACES
from tests.common import MockConfigEntry
CONF_OPNSENSE_CLIENT = "opnsense_client"
@pytest.fixture
def mock_config_entry(
hass: HomeAssistant, mock_opnsense_client: AsyncMock
) -> MockConfigEntry:
"""Return the default mocked config entry."""
mock_config_entry = MockConfigEntry(
domain=DOMAIN,
data=CONFIG_DATA,
unique_id="mocked_unique_id",
)
mock_config_entry.add_to_hass(hass)
return mock_config_entry
@pytest.fixture
def mock_opnsense_client() -> Generator[AsyncMock]:
"""Override OPNsenseClient in both config_flow and component."""
with (
patch(
"homeassistant.components.opnsense.config_flow.OPNsenseClient",
autospec=True,
) as mock_client,
patch(
"homeassistant.components.opnsense.OPNsenseClient",
new=mock_client,
),
):
client = mock_client.return_value
client.get_host_firmware_version.return_value = "25.7.8"
client.get_arp_table.return_value = ARP
client.get_interfaces.return_value = INTERFACES
client.get_device_unique_id.return_value = "mocked_unique_id"
yield client
@pytest.fixture
def mock_setup_entry() -> Generator[None]:
"""Mock setting up a config entry."""
with patch(
"homeassistant.components.opnsense.async_setup_entry", return_value=True
):
yield
+42
View File
@@ -0,0 +1,42 @@
"""Constants for opnsense tests."""
from homeassistant.components.opnsense.const import (
CONF_API_SECRET,
CONF_TRACKER_INTERFACES,
)
from homeassistant.const import CONF_API_KEY, CONF_URL, CONF_VERIFY_SSL
TITLE = "OPNsense"
CONFIG_DATA = {
CONF_URL: "http://router.lan/api",
CONF_API_KEY: "key",
CONF_API_SECRET: "secret",
CONF_VERIFY_SSL: False,
}
CONFIG_DATA_IMPORT = {
CONF_URL: "http://router.lan/api",
CONF_API_KEY: "key",
CONF_API_SECRET: "secret",
CONF_VERIFY_SSL: False,
CONF_TRACKER_INTERFACES: ["LAN"],
}
ARP = [
{
"hostname": "",
"intf": "igb1",
"intf_description": "LAN",
"ip": "192.168.0.123",
"mac": "ff:ff:ff:ff:ff:ff",
"manufacturer": "",
},
{
"hostname": "Desktop",
"intf": "igb1",
"intf_description": "LAN",
"ip": "192.168.0.167",
"mac": "ff:ff:ff:ff:ff:fe",
"manufacturer": "OEM",
},
]
INTERFACES = {"igb0": {"name": "WAN"}, "igb1": {"name": "LAN"}}
@@ -0,0 +1,325 @@
"""Tests for the OPNsense config flow."""
from unittest.mock import AsyncMock
from aiopnsense import (
OPNsenseBelowMinFirmware,
OPNsenseConnectionError,
OPNsenseInvalidAuth,
OPNsenseInvalidURL,
OPNsensePrivilegeMissing,
OPNsenseTimeoutError,
)
import pytest
from homeassistant.components.opnsense import OPNsenseSSLError, OPNsenseUnknownFirmware
from homeassistant.components.opnsense.const import CONF_TRACKER_INTERFACES, DOMAIN
from homeassistant.config_entries import SOURCE_IMPORT, SOURCE_USER
from homeassistant.const import CONF_URL, CONF_VERIFY_SSL
from homeassistant.core import DOMAIN as HOMEASSISTANT_DOMAIN, HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from homeassistant.helpers import issue_registry as ir
from .const import CONFIG_DATA, CONFIG_DATA_IMPORT
# Constants for test values
TEST_URL = "http://router.lan/api"
@pytest.mark.usefixtures("mock_setup_entry")
async def test_user(hass: HomeAssistant, mock_opnsense_client: AsyncMock) -> None:
"""Test user config."""
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"
# Submit user step, should go to interfaces step
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input=CONFIG_DATA,
)
assert result.get("type") is FlowResultType.FORM
assert result.get("step_id") == "interfaces"
# Submit interfaces step (simulate user selecting all interfaces)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={CONF_TRACKER_INTERFACES: []},
)
assert result.get("type") is FlowResultType.CREATE_ENTRY
assert result.get("title") == CONFIG_DATA[CONF_URL]
assert result.get("data") == CONFIG_DATA
assert result["result"].unique_id == "mocked_unique_id"
@pytest.mark.usefixtures("mock_setup_entry")
@pytest.mark.parametrize(
("exc", "expected"),
[
(OPNsenseInvalidAuth, "invalid_auth"),
(OPNsensePrivilegeMissing, "privilege_missing"),
(OPNsenseInvalidURL, "invalid_url"),
(OPNsenseSSLError, "ssl_error"),
(OPNsenseConnectionError, "cannot_connect"),
(OPNsenseTimeoutError, "cannot_connect"),
(OPNsenseUnknownFirmware, "unknown_version"),
(OPNsenseBelowMinFirmware, "invalid_version"),
],
)
async def test_user_exceptions(
hass: HomeAssistant,
mock_opnsense_client: AsyncMock,
exc: type[Exception],
expected: str,
) -> None:
"""Test all exception branches in async_step_user."""
mock_opnsense_client.validate.side_effect = exc
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input=CONFIG_DATA
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"] == {"base": expected}
mock_opnsense_client.validate.side_effect = None
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input=CONFIG_DATA
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
@pytest.mark.usefixtures("mock_setup_entry", "mock_config_entry")
async def test_user_unique_id_already_configured(
hass: HomeAssistant, mock_opnsense_client: AsyncMock
) -> None:
"""Test user flow aborts when unique ID is already configured."""
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=CONFIG_DATA
)
assert result.get("type") is FlowResultType.ABORT
assert result.get("reason") == "already_configured"
@pytest.mark.usefixtures("mock_setup_entry")
async def test_user_no_unique_id_aborts(
hass: HomeAssistant, mock_opnsense_client: AsyncMock
) -> None:
"""Test that the user flow aborts if the router has no unique id."""
mock_opnsense_client.get_device_unique_id.return_value = None
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={**CONFIG_DATA, CONF_URL: TEST_URL},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "no_unique_id"
@pytest.mark.usefixtures("mock_setup_entry")
async def test_on_unknown_error(
hass: HomeAssistant, mock_opnsense_client: AsyncMock
) -> None:
"""Test when we have unknown errors."""
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"
mock_opnsense_client.validate.side_effect = TypeError
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input=CONFIG_DATA,
)
assert result.get("type") is FlowResultType.FORM
assert result.get("errors") == {"base": "unknown"}
mock_opnsense_client.validate.side_effect = None
# Submit user step, should go to interfaces step
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input=CONFIG_DATA,
)
assert result.get("type") is FlowResultType.FORM
assert result.get("step_id") == "interfaces"
# Submit interfaces step
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={CONF_TRACKER_INTERFACES: []},
)
assert result.get("type") is FlowResultType.CREATE_ENTRY
@pytest.mark.usefixtures("mock_setup_entry")
async def test_interfaces_step_with_tracker_interfaces(
hass: HomeAssistant, mock_opnsense_client: AsyncMock
) -> None:
"""Test interfaces step with tracker_interfaces in user_input (covering the missing branch)."""
# Patch the client to return interfaces
mock_opnsense_client.return_value.get_device_unique_id.return_value = (
"unique_id_789"
)
mock_opnsense_client.return_value.get_interfaces.return_value = {
"LAN": {"name": "LAN"},
"WAN": {"name": "WAN"},
}
# Go through user step
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={**CONFIG_DATA, CONF_VERIFY_SSL: True},
)
# Now submit interfaces step with tracker_interfaces
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={CONF_TRACKER_INTERFACES: ["LAN", "WAN"]},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["data"][CONF_TRACKER_INTERFACES] == ["LAN", "WAN"]
@pytest.mark.usefixtures("mock_setup_entry")
async def test_import(hass: HomeAssistant, mock_opnsense_client: AsyncMock) -> None:
"""Test import step."""
mock_opnsense_client.return_value.get_device_unique_id.return_value = (
"unique_id_123"
)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_IMPORT},
data=CONFIG_DATA_IMPORT,
)
assert result.get("type") is FlowResultType.CREATE_ENTRY
assert result.get("title") == CONFIG_DATA_IMPORT[CONF_URL]
@pytest.mark.usefixtures(
"mock_opnsense_client", "mock_setup_entry", "mock_config_entry"
)
async def test_import_unique_id_already_configured(
hass: HomeAssistant, issue_registry: ir.IssueRegistry
) -> None:
"""Test import step when unique ID is already configured."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_IMPORT},
data=CONFIG_DATA_IMPORT,
)
assert result.get("type") is FlowResultType.ABORT
assert result.get("reason") == "already_configured"
# The deprecation issue must still be created so the YAML block gets removed
issue = issue_registry.async_get_issue(
HOMEASSISTANT_DOMAIN, f"deprecated_yaml_{DOMAIN}"
)
assert issue is not None
assert issue.translation_key == "deprecated_yaml"
@pytest.mark.usefixtures("mock_setup_entry")
async def test_import_no_unique_id_aborts(
hass: HomeAssistant,
mock_opnsense_client: AsyncMock,
issue_registry: ir.IssueRegistry,
) -> None:
"""Test that the import flow aborts and raises a repair if no unique id."""
mock_opnsense_client.get_device_unique_id.return_value = None
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_IMPORT},
data=CONFIG_DATA_IMPORT,
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "no_unique_id"
assert issue_registry.async_get_issue(DOMAIN, "import_failed_no_unique_id")
@pytest.mark.usefixtures("mock_setup_entry")
@pytest.mark.parametrize(
("exc", "reason"),
[
(OPNsenseInvalidURL, "invalid_url"),
(OPNsenseInvalidAuth, "invalid_auth"),
(OPNsensePrivilegeMissing, "privilege_missing"),
(OPNsenseSSLError, "ssl_error"),
(OPNsenseConnectionError, "cannot_connect"),
(OPNsenseTimeoutError, "cannot_connect"),
(OPNsenseUnknownFirmware, "unknown_version"),
(OPNsenseBelowMinFirmware, "invalid_version"),
(Exception, "unknown"),
],
)
async def test_import_exceptions(
hass: HomeAssistant,
mock_opnsense_client: AsyncMock,
issue_registry: ir.IssueRegistry,
exc: type[Exception],
reason: str,
) -> None:
"""Test all exception branches in async_step_import."""
mock_opnsense_client.validate.side_effect = exc
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_IMPORT},
data=CONFIG_DATA_IMPORT,
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == reason
assert issue_registry.async_get_issue(DOMAIN, f"import_failed_{reason}")
@pytest.mark.usefixtures("mock_opnsense_client", "mock_setup_entry")
async def test_import_empty_tracker_interfaces(hass: HomeAssistant) -> None:
"""Test import with empty CONF_TRACKER_INTERFACES (should pop the key)."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_IMPORT},
data={**CONFIG_DATA_IMPORT, CONF_TRACKER_INTERFACES: []},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert CONF_TRACKER_INTERFACES not in result["data"]
@pytest.mark.usefixtures("mock_setup_entry")
async def test_import_missing_interfaces(
hass: HomeAssistant,
mock_opnsense_client: AsyncMock,
issue_registry: ir.IssueRegistry,
) -> None:
"""Test import with missing tracker interfaces (should create issue and abort)."""
mock_opnsense_client.get_interfaces.return_value = {"LAN": {"name": "LAN"}}
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_IMPORT},
data={**CONFIG_DATA_IMPORT, CONF_TRACKER_INTERFACES: ["MISSING"]},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "import_failed_missing_interfaces"
assert issue_registry.async_get_issue(DOMAIN, "import_failed_missing_interfaces")
+152 -53
View File
@@ -1,70 +1,169 @@
"""The tests for the opnsense device tracker platform."""
from datetime import timedelta
from unittest import mock
from aiopnsense import OPNsenseConnectionError
from freezegun.api import FrozenDateTimeFactory
import pytest
from homeassistant.components import opnsense
from homeassistant.components.device_tracker import legacy
from homeassistant.components.opnsense import CONF_API_SECRET, DOMAIN
from homeassistant.const import CONF_API_KEY, CONF_URL, CONF_VERIFY_SSL
from homeassistant.components import device_tracker
from homeassistant.components.opnsense import OPNsenseRuntimeData
from homeassistant.components.opnsense.const import DOMAIN
from homeassistant.const import STATE_UNAVAILABLE
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
from homeassistant.helpers import entity_registry as er
from tests.common import MockConfigEntry, async_fire_time_changed
@pytest.fixture(name="mocked_opnsense")
def mocked_opnsense():
"""Mock for aiopnsense.OPNsenseClient."""
with mock.patch.object(opnsense, "OPNsenseClient") as mocked_opn:
yield mocked_opn
async def test_get_scanner(
hass: HomeAssistant, mocked_opnsense, mock_device_tracker_conf: list[legacy.Device]
@pytest.mark.usefixtures("mock_opnsense_client")
async def test_device_tracker_setup(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
entity_registry: er.EntityRegistry,
) -> None:
"""Test creating an opnsense scanner."""
opnsense_client = mock.AsyncMock()
mocked_opnsense.return_value = opnsense_client
opnsense_client.get_arp_table.return_value = [
{
"hostname": "",
"intf": "igb1",
"intf_description": "LAN",
"ip": "192.168.0.123",
"mac": "ff:ff:ff:ff:ff:ff",
"manufacturer": "",
},
{
"hostname": "Desktop",
"intf": "igb1",
"intf_description": "LAN",
"ip": "192.168.0.167",
"mac": "ff:ff:ff:ff:ff:fe",
"manufacturer": "OEM",
},
"""Test device tracker platform setup."""
# Setup the integration
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
# Check that device tracker entities are created
device_tracker_entities = er.async_entries_for_config_entry(
entity_registry, mock_config_entry.entry_id
)
device_tracker_entities = [
entity
for entity in device_tracker_entities
if entity.domain == device_tracker.DOMAIN
]
opnsense_client.get_interfaces.return_value = {
"wan": {"name": "WAN"},
"lan": {"name": "LAN"},
# Should have 2 devices from ARP table
assert len(device_tracker_entities) == 2
# Check the unique IDs are correct
entity_unique_ids = {entity.unique_id for entity in device_tracker_entities}
assert "ff:ff:ff:ff:ff:ff" in entity_unique_ids
assert "ff:ff:ff:ff:ff:fe" in entity_unique_ids
@pytest.mark.usefixtures("mock_opnsense_client")
async def test_device_tracker_states(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
entity_registry: er.EntityRegistry,
) -> None:
"""Test device tracker entity states and attributes."""
# Setup the integration
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
device_tracker_entities = er.async_entries_for_config_entry(
entity_registry, mock_config_entry.entry_id
)
device_tracker_entities = [
entity
for entity in device_tracker_entities
if entity.domain == device_tracker.DOMAIN
]
entity_ids_by_unique_id = {
entity.unique_id: entity.entity_id for entity in device_tracker_entities
}
result = await async_setup_component(
hass,
DOMAIN,
{
DOMAIN: {
CONF_URL: "https://fake_host_fun/api",
CONF_API_KEY: "fake_key",
CONF_API_SECRET: "fake_secret",
CONF_VERIFY_SSL: False,
}
# Enable entities (device trackers are disabled by default)
entity_registry.async_update_entity(
entity_ids_by_unique_id["ff:ff:ff:ff:ff:ff"], disabled_by=None
)
entity_registry.async_update_entity(
entity_ids_by_unique_id["ff:ff:ff:ff:ff:fe"], disabled_by=None
)
# Reload the config entry to activate the enabled entities
await hass.config_entries.async_reload(mock_config_entry.entry_id)
await hass.async_block_till_done()
# Test first device (no hostname)
entity_id_1 = entity_ids_by_unique_id["ff:ff:ff:ff:ff:ff"]
state_1 = hass.states.get(entity_id_1)
assert state_1 is not None
assert state_1.state == "home" # Should be connected since it's in ARP table
assert state_1.attributes.get("ip") == "192.168.0.123"
assert state_1.attributes.get("mac") == "ff:ff:ff:ff:ff:ff"
assert state_1.attributes.get("interface") == "LAN"
# Test second device (with hostname and manufacturer)
entity_id_2 = entity_ids_by_unique_id["ff:ff:ff:ff:ff:fe"]
state_2 = hass.states.get(entity_id_2)
assert state_2 is not None
assert state_2.state == "home" # Should be connected since it's in ARP table
assert state_2.attributes.get("ip") == "192.168.0.167"
assert state_2.attributes.get("mac") == "ff:ff:ff:ff:ff:fe"
assert state_2.attributes.get("interface") == "LAN"
assert state_2.attributes.get("manufacturer") == "OEM"
async def test_device_tracker_with_interfaces_filter(
hass: HomeAssistant,
mock_opnsense_client: mock.AsyncMock,
entity_registry: er.EntityRegistry,
) -> None:
"""Test device tracker with interface filtering."""
# Create config entry with interface filtering
mock_config_entry = MockConfigEntry(
domain=DOMAIN,
data={
"url": "http://router.lan/api",
"api_key": "key",
"api_secret": "secret",
"verify_ssl": False,
"tracker_interfaces": ["WAN"], # Filter to only WAN interface
},
)
mock_config_entry.runtime_data = OPNsenseRuntimeData(
client=mock_opnsense_client.return_value,
tracker_interfaces=["WAN"],
)
mock_config_entry.add_to_hass(hass)
# Setup the integration
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert result
device_1 = hass.states.get("device_tracker.desktop")
assert device_1 is not None
assert device_1.state == "home"
device_2 = hass.states.get("device_tracker.ff_ff_ff_ff_ff_ff")
assert device_2.state == "home"
# Check that no device tracker entities are created (since all devices are on LAN)
device_tracker_entities = er.async_entries_for_config_entry(
entity_registry, mock_config_entry.entry_id
)
device_tracker_entities = [
entity
for entity in device_tracker_entities
if entity.domain == device_tracker.DOMAIN
]
assert len(device_tracker_entities) == 0
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
async def test_device_tracker_coordinator_update_failure(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_opnsense_client: mock.AsyncMock,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test coordinator wraps client errors as UpdateFailed."""
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert hass.states.get("device_tracker.desktop").state != STATE_UNAVAILABLE
mock_opnsense_client.get_arp_table.side_effect = OPNsenseConnectionError(
"connection failed"
)
freezer.tick(timedelta(seconds=30))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert hass.states.get("device_tracker.desktop").state == STATE_UNAVAILABLE
assert mock_opnsense_client.get_arp_table.call_count == 2
+88
View File
@@ -0,0 +1,88 @@
"""Tests for the opnsense integration setup."""
from unittest import mock
from aiopnsense import (
OPNsenseBelowMinFirmware,
OPNsenseConnectionError,
OPNsenseInvalidAuth,
OPNsenseInvalidURL,
OPNsensePrivilegeMissing,
OPNsenseSSLError,
OPNsenseTimeoutError,
OPNsenseUnknownFirmware,
)
import pytest
from homeassistant.components.opnsense.const import (
CONF_API_SECRET,
CONF_TRACKER_INTERFACES,
DOMAIN,
)
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import CONF_API_KEY, CONF_URL, CONF_VERIFY_SSL
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
@pytest.mark.parametrize(
("exc", "expected_state", "expected_translation_key"),
[
(OPNsenseUnknownFirmware, ConfigEntryState.SETUP_ERROR, "unknown_firmware"),
(OPNsenseBelowMinFirmware, ConfigEntryState.SETUP_ERROR, "firmware_too_old"),
(OPNsenseInvalidURL, ConfigEntryState.SETUP_ERROR, "invalid_url"),
(OPNsenseTimeoutError, ConfigEntryState.SETUP_RETRY, "timeout_connecting"),
(OPNsenseSSLError, ConfigEntryState.SETUP_ERROR, "ssl_error"),
(OPNsenseInvalidAuth, ConfigEntryState.SETUP_ERROR, "invalid_auth"),
(OPNsensePrivilegeMissing, ConfigEntryState.SETUP_ERROR, "privilege_missing"),
(OPNsenseConnectionError, ConfigEntryState.SETUP_RETRY, "cannot_connect"),
],
)
async def test_setup_entry_exceptions(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_opnsense_client: mock.AsyncMock,
exc: type[Exception],
expected_state: ConfigEntryState,
expected_translation_key: str,
) -> None:
"""Test async_setup_entry surfaces translation-keyed errors."""
mock_opnsense_client.validate.side_effect = exc
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is expected_state
assert mock_config_entry.error_reason_translation_key == expected_translation_key
assert mock_config_entry.error_reason_translation_placeholders == {
"url": mock_config_entry.data[CONF_URL]
}
async def test_setup_entry_tracker_interface_not_found(
hass: HomeAssistant,
mock_opnsense_client: mock.AsyncMock,
) -> None:
"""Test async_setup_entry rejects unknown tracker interfaces."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_URL: "http://router.lan/api",
CONF_API_KEY: "key",
CONF_API_SECRET: "secret",
CONF_VERIFY_SSL: False,
CONF_TRACKER_INTERFACES: ["NOPE"],
},
)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert entry.state is ConfigEntryState.SETUP_ERROR
assert entry.error_reason_translation_key == "tracker_interface_not_found"
assert entry.error_reason_translation_placeholders == {
"interface": "NOPE",
"known": "WAN, LAN",
}
+1 -1
View File
@@ -119,7 +119,7 @@ async def create_device(hass: HomeAssistant, mock_device_code: str) -> CustomerD
if device.update_time:
device.update_time = int(dt_util.as_timestamp(device.update_time))
device.support_local = details.get("support_local")
device.local_strategy = details.get("local_strategy")
device.local_strategy = details.get("local_strategy") or {}
device.mqtt_connected = details.get("mqtt_connected")
device.function = {
@@ -198,7 +198,8 @@
'name_by_user': None,
}),
'id': '2pxfek1jjrtctiyglam',
'local_strategy': None,
'local_strategy': dict({
}),
'mqtt_connected': True,
'name': 'Multifunction alarm',
'online': True,
@@ -388,7 +389,8 @@
'name_by_user': None,
}),
'id': 'cwwk68dyfsh2eqi4jbqr',
'local_strategy': None,
'local_strategy': dict({
}),
'mqtt_connected': True,
'name': 'Gas sensor',
'online': True,
@@ -539,7 +541,8 @@
'name_by_user': None,
}),
'id': 'vrhdtr5fawoiyth9qdt',
'local_strategy': None,
'local_strategy': dict({
}),
'mqtt_connected': True,
'name': 'Framboisiers',
'online': True,
@@ -682,7 +685,8 @@
'name_by_user': None,
}),
'id': 'cwwk68dyfsh2eqi4jbqr',
'local_strategy': None,
'local_strategy': dict({
}),
'name': 'Gas sensor',
'online': True,
'product_id': '4iqe2hsfyd86kwwc',
+3 -15
View File
@@ -3,7 +3,6 @@
import asyncio
import dataclasses
import logging
from typing import Any
from unittest.mock import Mock, patch
import pytest
@@ -1275,32 +1274,21 @@ def test_nested_section_in_serializer() -> None:
)
@pytest.mark.parametrize(
("context", "expected_show_advanced"),
[
# The property is deprecated and now unconditionally returns True
({}, True),
({"show_advanced_options": False}, True),
({"show_advanced_options": True}, True),
],
)
async def test_show_advanced_options(
manager: MockFlowManager,
context: dict[str, Any],
expected_show_advanced: bool,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test FlowHandler show_advanced_options property."""
"""Test FlowHandler show_advanced_options property is deprecated and always True."""
@manager.mock_reg_handler("test")
class TestFlow(data_entry_flow.FlowHandler):
VERSION = 5
async def async_step_init(self, info):
assert self.show_advanced_options == expected_show_advanced
assert self.show_advanced_options is True
return self.async_create_entry(title="hello", data={})
await manager.async_init("test", context=context, data={})
await manager.async_init("test", context={}, data={})
assert len(manager.async_progress()) == 0
assert len(manager.mock_created_entries) == 1