Compare commits

..

1 Commits

Author SHA1 Message Date
Paulus Schoutsen
c080c98cfe do not require admin for instance URLs 2025-11-21 13:41:06 -05:00
42 changed files with 361 additions and 11712 deletions

View File

@@ -1,10 +1,10 @@
"""The Actron Air integration."""
from actron_neo_api import (
ActronAirACSystem,
ActronAirAPI,
ActronAirAPIError,
ActronAirAuthError,
ActronAirNeoACSystem,
ActronNeoAPI,
ActronNeoAPIError,
ActronNeoAuthError,
)
from homeassistant.const import CONF_API_TOKEN, Platform
@@ -23,16 +23,16 @@ PLATFORM = [Platform.CLIMATE]
async def async_setup_entry(hass: HomeAssistant, entry: ActronAirConfigEntry) -> bool:
"""Set up Actron Air integration from a config entry."""
api = ActronAirAPI(refresh_token=entry.data[CONF_API_TOKEN])
systems: list[ActronAirACSystem] = []
api = ActronNeoAPI(refresh_token=entry.data[CONF_API_TOKEN])
systems: list[ActronAirNeoACSystem] = []
try:
systems = await api.get_ac_systems()
await api.update_status()
except ActronAirAuthError:
except ActronNeoAuthError:
_LOGGER.error("Authentication error while setting up Actron Air integration")
raise
except ActronAirAPIError as err:
except ActronNeoAPIError as err:
_LOGGER.error("API error while setting up Actron Air integration: %s", err)
raise

View File

@@ -2,7 +2,7 @@
from typing import Any
from actron_neo_api import ActronAirStatus, ActronAirZone
from actron_neo_api import ActronAirNeoStatus, ActronAirNeoZone
from homeassistant.components.climate import (
FAN_AUTO,
@@ -132,7 +132,7 @@ class ActronSystemClimate(BaseClimateEntity):
return self._status.max_temp
@property
def _status(self) -> ActronAirStatus:
def _status(self) -> ActronAirNeoStatus:
"""Get the current status from the coordinator."""
return self.coordinator.data
@@ -194,7 +194,7 @@ class ActronZoneClimate(BaseClimateEntity):
def __init__(
self,
coordinator: ActronAirSystemCoordinator,
zone: ActronAirZone,
zone: ActronAirNeoZone,
) -> None:
"""Initialize an Actron Air unit."""
super().__init__(coordinator, zone.title)
@@ -221,7 +221,7 @@ class ActronZoneClimate(BaseClimateEntity):
return self._zone.max_temp
@property
def _zone(self) -> ActronAirZone:
def _zone(self) -> ActronAirNeoZone:
"""Get the current zone data from the coordinator."""
status = self.coordinator.data
return status.zones[self._zone_id]

View File

@@ -3,7 +3,7 @@
import asyncio
from typing import Any
from actron_neo_api import ActronAirAPI, ActronAirAuthError
from actron_neo_api import ActronNeoAPI, ActronNeoAuthError
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_API_TOKEN
@@ -17,7 +17,7 @@ class ActronAirConfigFlow(ConfigFlow, domain=DOMAIN):
def __init__(self) -> None:
"""Initialize the config flow."""
self._api: ActronAirAPI | None = None
self._api: ActronNeoAPI | None = None
self._device_code: str | None = None
self._user_code: str = ""
self._verification_uri: str = ""
@@ -30,10 +30,10 @@ class ActronAirConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle the initial step."""
if self._api is None:
_LOGGER.debug("Initiating device authorization")
self._api = ActronAirAPI()
self._api = ActronNeoAPI()
try:
device_code_response = await self._api.request_device_code()
except ActronAirAuthError as err:
except ActronNeoAuthError as err:
_LOGGER.error("OAuth2 flow failed: %s", err)
return self.async_abort(reason="oauth2_error")
@@ -50,7 +50,7 @@ class ActronAirConfigFlow(ConfigFlow, domain=DOMAIN):
try:
await self._api.poll_for_token(self._device_code)
_LOGGER.debug("Authorization successful")
except ActronAirAuthError as ex:
except ActronNeoAuthError as ex:
_LOGGER.exception("Error while waiting for device authorization")
raise CannotConnect from ex
@@ -89,7 +89,7 @@ class ActronAirConfigFlow(ConfigFlow, domain=DOMAIN):
try:
user_data = await self._api.get_user_info()
except ActronAirAuthError as err:
except ActronNeoAuthError as err:
_LOGGER.error("Error getting user info: %s", err)
return self.async_abort(reason="oauth2_error")

View File

@@ -5,7 +5,7 @@ from __future__ import annotations
from dataclasses import dataclass
from datetime import timedelta
from actron_neo_api import ActronAirACSystem, ActronAirAPI, ActronAirStatus
from actron_neo_api import ActronAirNeoACSystem, ActronAirNeoStatus, ActronNeoAPI
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
@@ -23,7 +23,7 @@ ERROR_UNKNOWN = "unknown_error"
class ActronAirRuntimeData:
"""Runtime data for the Actron Air integration."""
api: ActronAirAPI
api: ActronNeoAPI
system_coordinators: dict[str, ActronAirSystemCoordinator]
@@ -33,15 +33,15 @@ AUTH_ERROR_THRESHOLD = 3
SCAN_INTERVAL = timedelta(seconds=30)
class ActronAirSystemCoordinator(DataUpdateCoordinator[ActronAirACSystem]):
class ActronAirSystemCoordinator(DataUpdateCoordinator[ActronAirNeoACSystem]):
"""System coordinator for Actron Air integration."""
def __init__(
self,
hass: HomeAssistant,
entry: ActronAirConfigEntry,
api: ActronAirAPI,
system: ActronAirACSystem,
api: ActronNeoAPI,
system: ActronAirNeoACSystem,
) -> None:
"""Initialize the coordinator."""
super().__init__(
@@ -57,7 +57,7 @@ class ActronAirSystemCoordinator(DataUpdateCoordinator[ActronAirACSystem]):
self.status = self.api.state_manager.get_status(self.serial_number)
self.last_seen = dt_util.utcnow()
async def _async_update_data(self) -> ActronAirStatus:
async def _async_update_data(self) -> ActronAirNeoStatus:
"""Fetch updates and merge incremental changes into the full state."""
await self.api.update_status()
self.status = self.api.state_manager.get_status(self.serial_number)

View File

@@ -12,5 +12,5 @@
"documentation": "https://www.home-assistant.io/integrations/actron_air",
"iot_class": "cloud_polling",
"quality_scale": "bronze",
"requirements": ["actron-neo-api==0.1.87"]
"requirements": ["actron-neo-api==0.1.84"]
}

View File

@@ -0,0 +1,259 @@
"""Support for Dominos Pizza ordering."""
from datetime import timedelta
import logging
from pizzapi import Address, Customer, Order
import voluptuous as vol
from homeassistant.components import http
from homeassistant.core import HomeAssistant, ServiceCall, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.entity_component import EntityComponent
from homeassistant.helpers.typing import ConfigType
from homeassistant.util import Throttle
_LOGGER = logging.getLogger(__name__)
# The domain of your component. Should be equal to the name of your component.
DOMAIN = "dominos"
ENTITY_ID_FORMAT = DOMAIN + ".{}"
ATTR_COUNTRY = "country_code"
ATTR_FIRST_NAME = "first_name"
ATTR_LAST_NAME = "last_name"
ATTR_EMAIL = "email"
ATTR_PHONE = "phone"
ATTR_ADDRESS = "address"
ATTR_ORDERS = "orders"
ATTR_SHOW_MENU = "show_menu"
ATTR_ORDER_ENTITY = "order_entity_id"
ATTR_ORDER_NAME = "name"
ATTR_ORDER_CODES = "codes"
MIN_TIME_BETWEEN_UPDATES = timedelta(minutes=10)
MIN_TIME_BETWEEN_STORE_UPDATES = timedelta(minutes=3330)
_ORDERS_SCHEMA = vol.Schema(
{
vol.Required(ATTR_ORDER_NAME): cv.string,
vol.Required(ATTR_ORDER_CODES): vol.All(cv.ensure_list, [cv.string]),
}
)
CONFIG_SCHEMA = vol.Schema(
{
DOMAIN: vol.Schema(
{
vol.Required(ATTR_COUNTRY): cv.string,
vol.Required(ATTR_FIRST_NAME): cv.string,
vol.Required(ATTR_LAST_NAME): cv.string,
vol.Required(ATTR_EMAIL): cv.string,
vol.Required(ATTR_PHONE): cv.string,
vol.Required(ATTR_ADDRESS): cv.string,
vol.Optional(ATTR_SHOW_MENU): cv.boolean,
vol.Optional(ATTR_ORDERS, default=[]): vol.All(
cv.ensure_list, [_ORDERS_SCHEMA]
),
}
)
},
extra=vol.ALLOW_EXTRA,
)
def setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up is called when Home Assistant is loading our component."""
dominos = Dominos(hass, config)
component = EntityComponent[DominosOrder](_LOGGER, DOMAIN, hass)
hass.data[DOMAIN] = {}
entities: list[DominosOrder] = []
conf = config[DOMAIN]
hass.services.register(
DOMAIN,
"order",
dominos.handle_order,
vol.Schema(
{
vol.Required(ATTR_ORDER_ENTITY): cv.entity_ids,
}
),
)
if conf.get(ATTR_SHOW_MENU):
hass.http.register_view(DominosProductListView(dominos))
for order_info in conf.get(ATTR_ORDERS):
order = DominosOrder(order_info, dominos)
entities.append(order)
component.add_entities(entities)
# Return boolean to indicate that initialization was successfully.
return True
class Dominos:
"""Main Dominos service."""
def __init__(self, hass, config):
"""Set up main service."""
conf = config[DOMAIN]
self.hass = hass
self.customer = Customer(
conf.get(ATTR_FIRST_NAME),
conf.get(ATTR_LAST_NAME),
conf.get(ATTR_EMAIL),
conf.get(ATTR_PHONE),
conf.get(ATTR_ADDRESS),
)
self.address = Address(
*self.customer.address.split(","), country=conf.get(ATTR_COUNTRY)
)
self.country = conf.get(ATTR_COUNTRY)
try:
self.closest_store = self.address.closest_store()
except Exception: # noqa: BLE001
self.closest_store = None
def handle_order(self, call: ServiceCall) -> None:
"""Handle ordering pizza."""
entity_ids = call.data[ATTR_ORDER_ENTITY]
target_orders = [
order
for order in self.hass.data[DOMAIN]["entities"]
if order.entity_id in entity_ids
]
for order in target_orders:
order.place()
@Throttle(MIN_TIME_BETWEEN_STORE_UPDATES)
def update_closest_store(self):
"""Update the shared closest store (if open)."""
try:
self.closest_store = self.address.closest_store()
except Exception: # noqa: BLE001
self.closest_store = None
return False
return True
def get_menu(self):
"""Return the products from the closest stores menu."""
self.update_closest_store()
if self.closest_store is None:
_LOGGER.warning("Cannot get menu. Store may be closed")
return []
menu = self.closest_store.get_menu()
product_entries = []
for product in menu.products:
item = {}
if isinstance(product.menu_data["Variants"], list):
variants = ", ".join(product.menu_data["Variants"])
else:
variants = product.menu_data["Variants"]
item["name"] = product.name
item["variants"] = variants
product_entries.append(item)
return product_entries
class DominosProductListView(http.HomeAssistantView):
"""View to retrieve product list content."""
url = "/api/dominos"
name = "api:dominos"
def __init__(self, dominos):
"""Initialize suite view."""
self.dominos = dominos
@callback
def get(self, request):
"""Retrieve if API is running."""
return self.json(self.dominos.get_menu())
class DominosOrder(Entity):
"""Represents a Dominos order entity."""
def __init__(self, order_info, dominos):
"""Set up the entity."""
self._name = order_info["name"]
self._product_codes = order_info["codes"]
self._orderable = False
self.dominos = dominos
@property
def name(self):
"""Return the orders name."""
return self._name
@property
def product_codes(self):
"""Return the orders product codes."""
return self._product_codes
@property
def orderable(self):
"""Return the true if orderable."""
return self._orderable
@property
def state(self):
"""Return the state either closed, orderable or unorderable."""
if self.dominos.closest_store is None:
return "closed"
return "orderable" if self._orderable else "unorderable"
@Throttle(MIN_TIME_BETWEEN_UPDATES)
def update(self):
"""Update the order state and refreshes the store."""
try:
self.dominos.update_closest_store()
except Exception: # noqa: BLE001
self._orderable = False
return
try:
order = self.order()
order.pay_with()
self._orderable = True
except Exception: # noqa: BLE001
self._orderable = False
def order(self):
"""Create the order object."""
if self.dominos.closest_store is None:
raise HomeAssistantError("No store available")
order = Order(
self.dominos.closest_store,
self.dominos.customer,
self.dominos.address,
self.dominos.country,
)
for code in self._product_codes:
order.add_item(code)
return order
def place(self):
"""Place the order."""
try:
order = self.order()
order.place()
except Exception: # noqa: BLE001
self._orderable = False
_LOGGER.warning(
"Attempted to order Dominos - Order invalid or store closed"
)

View File

@@ -0,0 +1,7 @@
{
"services": {
"order": {
"service": "mdi:pizza"
}
}
}

View File

@@ -0,0 +1,11 @@
{
"domain": "dominos",
"name": "Dominos Pizza",
"codeowners": [],
"dependencies": ["http"],
"documentation": "https://www.home-assistant.io/integrations/dominos",
"iot_class": "cloud_polling",
"loggers": ["pizzapi"],
"quality_scale": "legacy",
"requirements": ["pizzapi==0.0.6"]
}

View File

@@ -0,0 +1,6 @@
order:
fields:
order_entity_id:
example: dominos.medium_pan
selector:
text:

View File

@@ -0,0 +1,14 @@
{
"services": {
"order": {
"description": "Places a set of orders with Domino's Pizza.",
"fields": {
"order_entity_id": {
"description": "The ID (as specified in the configuration) of an order to place. If provided as an array, all the identified orders will be placed.",
"name": "Order entity"
}
},
"name": "Order"
}
}
}

View File

@@ -62,5 +62,5 @@
"dependencies": ["bluetooth_adapters"],
"documentation": "https://www.home-assistant.io/integrations/inkbird",
"iot_class": "local_push",
"requirements": ["inkbird-ble==1.1.1"]
"requirements": ["inkbird-ble==1.1.0"]
}

View File

@@ -68,7 +68,6 @@ async def websocket_network_adapters_configure(
@callback
@websocket_api.require_admin
@websocket_api.websocket_command(
{
vol.Required("type"): "network/url",

View File

@@ -17,7 +17,7 @@ from .coordinator import PooldoseConfigEntry, PooldoseCoordinator
_LOGGER = logging.getLogger(__name__)
PLATFORMS: list[Platform] = [Platform.BINARY_SENSOR, Platform.SENSOR]
PLATFORMS: list[Platform] = [Platform.SENSOR]
async def async_migrate_entry(hass: HomeAssistant, entry: PooldoseConfigEntry) -> bool:

View File

@@ -1,129 +0,0 @@
"""Binary sensors for the Seko PoolDose integration."""
from __future__ import annotations
import logging
from typing import TYPE_CHECKING, cast
from homeassistant.components.binary_sensor import (
BinarySensorDeviceClass,
BinarySensorEntity,
BinarySensorEntityDescription,
)
from homeassistant.const import EntityCategory
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from . import PooldoseConfigEntry
from .entity import PooldoseEntity
_LOGGER = logging.getLogger(__name__)
BINARY_SENSOR_DESCRIPTIONS: tuple[BinarySensorEntityDescription, ...] = (
BinarySensorEntityDescription(
key="pump_alarm",
translation_key="pump_alarm",
device_class=BinarySensorDeviceClass.PROBLEM,
entity_category=EntityCategory.DIAGNOSTIC,
),
BinarySensorEntityDescription(
key="ph_level_alarm",
translation_key="ph_level_alarm",
device_class=BinarySensorDeviceClass.PROBLEM,
entity_category=EntityCategory.DIAGNOSTIC,
),
BinarySensorEntityDescription(
key="orp_level_alarm",
translation_key="orp_level_alarm",
device_class=BinarySensorDeviceClass.PROBLEM,
entity_category=EntityCategory.DIAGNOSTIC,
),
BinarySensorEntityDescription(
key="flow_rate_alarm",
translation_key="flow_rate_alarm",
device_class=BinarySensorDeviceClass.PROBLEM,
entity_category=EntityCategory.DIAGNOSTIC,
),
BinarySensorEntityDescription(
key="alarm_ofa_ph",
translation_key="alarm_ofa_ph",
device_class=BinarySensorDeviceClass.PROBLEM,
entity_category=EntityCategory.DIAGNOSTIC,
),
BinarySensorEntityDescription(
key="alarm_ofa_orp",
translation_key="alarm_ofa_orp",
device_class=BinarySensorDeviceClass.PROBLEM,
entity_category=EntityCategory.DIAGNOSTIC,
),
BinarySensorEntityDescription(
key="alarm_ofa_cl",
translation_key="alarm_ofa_cl",
device_class=BinarySensorDeviceClass.PROBLEM,
entity_category=EntityCategory.DIAGNOSTIC,
),
BinarySensorEntityDescription(
key="relay_alarm",
translation_key="relay_alarm",
device_class=BinarySensorDeviceClass.POWER,
entity_category=EntityCategory.DIAGNOSTIC,
),
BinarySensorEntityDescription(
key="relay_aux1",
translation_key="relay_aux1",
device_class=BinarySensorDeviceClass.POWER,
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
),
BinarySensorEntityDescription(
key="relay_aux2",
translation_key="relay_aux2",
device_class=BinarySensorDeviceClass.POWER,
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
),
BinarySensorEntityDescription(
key="relay_aux3",
translation_key="relay_aux3",
device_class=BinarySensorDeviceClass.POWER,
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
),
)
async def async_setup_entry(
hass: HomeAssistant,
config_entry: PooldoseConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up PoolDose binary sensor entities from a config entry."""
if TYPE_CHECKING:
assert config_entry.unique_id is not None
coordinator = config_entry.runtime_data
binary_sensor_data = coordinator.data["binary_sensor"]
serial_number = config_entry.unique_id
async_add_entities(
PooldoseBinarySensor(
coordinator,
serial_number,
coordinator.device_info,
description,
"binary_sensor",
)
for description in BINARY_SENSOR_DESCRIPTIONS
if description.key in binary_sensor_data
)
class PooldoseBinarySensor(PooldoseEntity, BinarySensorEntity):
"""Binary sensor entity for the Seko PoolDose Python API."""
@property
def is_on(self) -> bool:
"""Return true if the binary sensor is on."""
data = cast(dict, self.get_data())
return cast(bool, data["value"])

View File

@@ -68,11 +68,6 @@ class PooldoseEntity(CoordinatorEntity[PooldoseCoordinator]):
coordinator.config_entry.data.get(CONF_MAC),
)
@property
def available(self) -> bool:
"""Return if entity is available."""
return super().available and self.get_data() is not None
def get_data(self) -> ValueDict | None:
"""Get data for this entity, only if available."""
platform_data = self.coordinator.data[self.platform_name]

View File

@@ -1,79 +1,11 @@
{
"entity": {
"binary_sensor": {
"alarm_ofa_cl": {
"default": "mdi:clock-alert-outline",
"state": {
"on": "mdi:clock-alert"
}
},
"alarm_ofa_orp": {
"default": "mdi:clock-alert-outline",
"state": {
"on": "mdi:clock-alert"
}
},
"alarm_ofa_ph": {
"default": "mdi:clock-alert-outline",
"state": {
"on": "mdi:clock-alert"
}
},
"flow_rate_alarm": {
"default": "mdi:autorenew",
"state": {
"on": "mdi:autorenew-off"
}
},
"orp_level_alarm": {
"default": "mdi:flask",
"state": {
"on": "mdi:flask-empty"
}
},
"ph_level_alarm": {
"default": "mdi:flask",
"state": {
"on": "mdi:flask-empty"
}
},
"pump_alarm": {
"default": "mdi:pump",
"state": {
"on": "mdi:pump-off"
}
},
"relay_alarm": {
"default": "mdi:electric-switch-closed",
"state": {
"on": "mdi:electric-switch"
}
},
"relay_aux1": {
"default": "mdi:electric-switch-closed",
"state": {
"on": "mdi:electric-switch"
}
},
"relay_aux2": {
"default": "mdi:electric-switch-closed",
"state": {
"on": "mdi:electric-switch"
}
},
"relay_aux3": {
"default": "mdi:electric-switch-closed",
"state": {
"on": "mdi:electric-switch"
}
}
},
"sensor": {
"cl": {
"default": "mdi:pool"
},
"cl_type_dosing": {
"default": "mdi:beaker"
"default": "mdi:flask"
},
"flow_rate": {
"default": "mdi:pipe-valve"
@@ -97,7 +29,7 @@
"default": "mdi:form-select"
},
"orp_type_dosing": {
"default": "mdi:beaker"
"default": "mdi:flask"
},
"peristaltic_cl_dosing": {
"default": "mdi:pump"
@@ -118,7 +50,7 @@
"default": "mdi:form-select"
},
"ph_type_dosing": {
"default": "mdi:beaker"
"default": "mdi:flask"
}
}
}

View File

@@ -33,41 +33,6 @@
}
},
"entity": {
"binary_sensor": {
"alarm_ofa_cl": {
"name": "Chlorine tank level"
},
"alarm_ofa_orp": {
"name": "ORP overfeed"
},
"alarm_ofa_ph": {
"name": "pH overfeed"
},
"flow_rate_alarm": {
"name": "Flow rate"
},
"orp_level_alarm": {
"name": "ORP tank level"
},
"ph_level_alarm": {
"name": "pH tank level"
},
"pump_alarm": {
"name": "Recirculation"
},
"relay_alarm": {
"name": "Alarm relay status"
},
"relay_aux1": {
"name": "Auxiliary relay 1 status"
},
"relay_aux2": {
"name": "Auxiliary relay 2 status"
},
"relay_aux3": {
"name": "Auxiliary relay 3 status"
}
},
"sensor": {
"cl": {
"name": "Chlorine"

View File

@@ -83,7 +83,7 @@ class RpcBinarySensor(ShellyRpcAttributeEntity, BinarySensorEntity):
"""Initialize sensor."""
super().__init__(coordinator, key, attribute, description)
if description.role != ROLE_GENERIC:
if hasattr(self, "_attr_name") and description.role != ROLE_GENERIC:
if not description.role and description.key == "input":
_, component, component_id = get_rpc_key(key)
if not get_rpc_custom_name(coordinator.device, key) and (
@@ -94,8 +94,7 @@ class RpcBinarySensor(ShellyRpcAttributeEntity, BinarySensorEntity):
else:
return
if hasattr(self, "_attr_name"):
delattr(self, "_attr_name")
delattr(self, "_attr_name")
if not description.role and description.key != "input":
translation_placeholders, translation_key = (

View File

@@ -183,8 +183,6 @@ class ShellyConfigFlow(ConfigFlow, domain=DOMAIN):
selected_ssid: str = ""
_provision_task: asyncio.Task | None = None
_provision_result: ConfigFlowResult | None = None
disable_ap_after_provision: bool = True
disable_ble_rpc_after_provision: bool = True
async def async_step_user(
self, user_input: dict[str, Any] | None = None
@@ -428,20 +426,10 @@ class ShellyConfigFlow(ConfigFlow, domain=DOMAIN):
) -> ConfigFlowResult:
"""Confirm bluetooth provisioning."""
if user_input is not None:
self.disable_ap_after_provision = user_input.get("disable_ap", True)
self.disable_ble_rpc_after_provision = user_input.get(
"disable_ble_rpc", True
)
return await self.async_step_wifi_scan()
return self.async_show_form(
step_id="bluetooth_confirm",
data_schema=vol.Schema(
{
vol.Optional("disable_ap", default=True): bool,
vol.Optional("disable_ble_rpc", default=True): bool,
}
),
description_placeholders={
"name": self.context["title_placeholders"]["name"]
},
@@ -533,62 +521,6 @@ class ShellyConfigFlow(ConfigFlow, domain=DOMAIN):
description_placeholders={"ssid": self.selected_ssid},
)
async def _async_secure_device_after_provision(self, host: str, port: int) -> None:
"""Disable AP and/or BLE RPC after successful WiFi provisioning.
Must be called via IP after device is on WiFi, not via BLE.
"""
if (
not self.disable_ap_after_provision
and not self.disable_ble_rpc_after_provision
):
return
# Connect to device via IP
options = ConnectionOptions(
host,
None,
None,
device_mac=self.unique_id,
port=port,
)
device: RpcDevice | None = None
try:
device = await RpcDevice.create(
async_get_clientsession(self.hass), None, options
)
await device.initialize()
restart_required = False
# Disable WiFi AP if requested
if self.disable_ap_after_provision:
result = await device.wifi_setconfig(ap_enable=False)
LOGGER.debug("Disabled WiFi AP on %s", host)
restart_required = restart_required or result.get(
"restart_required", False
)
# Disable BLE RPC if requested (keep BLE enabled for sensors/buttons)
if self.disable_ble_rpc_after_provision:
result = await device.ble_setconfig(enable=True, enable_rpc=False)
LOGGER.debug("Disabled BLE RPC on %s", host)
restart_required = restart_required or result.get(
"restart_required", False
)
# Restart device once if either operation requires it
if restart_required:
await device.trigger_reboot(delay_ms=1000)
except (TimeoutError, DeviceConnectionError, RpcCallError) as err:
LOGGER.warning(
"Failed to secure device after provisioning at %s: %s", host, err
)
# Don't fail the flow - device is already on WiFi and functional
finally:
if device:
await device.shutdown()
async def _async_provision_wifi_and_wait_for_zeroconf(
self, mac: str, password: str, state: ProvisioningState
) -> ConfigFlowResult | None:
@@ -682,9 +614,6 @@ class ShellyConfigFlow(ConfigFlow, domain=DOMAIN):
if not device_info[CONF_MODEL]:
return self.async_abort(reason="firmware_not_fully_provisioned")
# Secure device after provisioning if requested (disable AP/BLE)
await self._async_secure_device_after_provision(self.host, self.port)
# User just provisioned this device - create entry directly without confirmation
return self.async_create_entry(
title=device_info["title"],

View File

@@ -31,16 +31,7 @@
},
"step": {
"bluetooth_confirm": {
"data": {
"disable_ap": "Disable WiFi access point after provisioning",
"disable_ble_rpc": "Disable Bluetooth RPC after provisioning"
},
"data_description": {
"disable_ap": "For improved security, disable the WiFi access point after successfully connecting to your network.",
"disable_ble_rpc": "For improved security, disable Bluetooth RPC access after WiFi is configured. Bluetooth will remain enabled for BLE sensors and buttons."
},
"description": "The Shelly device {name} has been discovered via Bluetooth but is not connected to WiFi.\n\nDo you want to provision WiFi credentials to this device?",
"title": "Provision WiFi via Bluetooth"
"description": "The Shelly device {name} has been discovered via Bluetooth but is not connected to WiFi.\n\nDo you want to provision WiFi credentials to this device?"
},
"confirm_discovery": {
"description": "Do you want to set up the {model} at {host}?\n\nBattery-powered devices that are password-protected must be woken up before continuing with setting up.\nBattery-powered devices that are not password-protected will be added when the device wakes up, you can now manually wake the device up using a button on it or wait for the next data update from the device."

View File

@@ -40,7 +40,7 @@
"integration_type": "hub",
"iot_class": "local_push",
"loggers": ["uiprotect", "unifi_discovery"],
"requirements": ["uiprotect==7.26.0", "unifi-discovery==1.2.0"],
"requirements": ["uiprotect==7.23.0", "unifi-discovery==1.2.0"],
"ssdp": [
{
"manufacturer": "Ubiquiti Networks",

View File

@@ -64,11 +64,6 @@ _DESCRIPTIONS: tuple[VolvoButtonDescription, ...] = (
api_command="honk-flash",
required_command_key="HONK_AND_FLASH",
),
VolvoButtonDescription(
key="lock_reduced_guard",
api_command="lock-reduced-guard",
required_command_key="LOCK_REDUCED_GUARD",
),
)

View File

@@ -281,9 +281,6 @@
},
"honk_flash": {
"default": "mdi:alarm-light"
},
"lock_reduced_guard": {
"default": "mdi:lock-minus"
}
},
"device_tracker": {

View File

@@ -208,9 +208,6 @@
},
"honk_flash": {
"name": "Honk & flash"
},
"lock_reduced_guard": {
"name": "Lock reduced guard"
}
},
"device_tracker": {

View File

@@ -1,56 +0,0 @@
"""Diagnostics platform for the Xbox integration."""
from __future__ import annotations
from typing import Any
from homeassistant.components.diagnostics import async_redact_data
from homeassistant.core import HomeAssistant
from .coordinator import XboxConfigEntry
TO_REDACT = {
"bio",
"display_name",
"display_pic_raw",
"gamertag",
"linked_accounts",
"location",
"modern_gamertag_suffix",
"modern_gamertag",
"real_name",
"unique_modern_gamertag",
"xuid",
}
async def async_get_config_entry_diagnostics(
hass: HomeAssistant, config_entry: XboxConfigEntry
) -> dict[str, Any]:
"""Return diagnostics for a config entry."""
coordinator = config_entry.runtime_data.status
consoles_coordinator = config_entry.runtime_data.consoles
presence = [
async_redact_data(person.model_dump(), TO_REDACT)
for person in coordinator.data.presence.values()
]
consoles_status = [
{
"status": console.status.model_dump(),
"app_details": (
console.app_details.model_dump() if console.app_details else None
),
}
for console in coordinator.data.consoles.values()
]
consoles_list = consoles_coordinator.data.model_dump()
title_info = [title.model_dump() for title in coordinator.data.title_info.values()]
return {
"consoles_status": consoles_status,
"consoles_list": consoles_list,
"presence": presence,
"title_info": title_info,
}

View File

@@ -11,7 +11,7 @@
],
"documentation": "https://www.home-assistant.io/integrations/xbox",
"iot_class": "cloud_polling",
"requirements": ["python-xbox==0.1.2"],
"requirements": ["python-xbox==0.1.1"],
"ssdp": [
{
"manufacturer": "Microsoft Corporation",

View File

@@ -1387,6 +1387,12 @@
"config_flow": true,
"iot_class": "cloud_polling"
},
"dominos": {
"name": "Dominos Pizza",
"integration_type": "hub",
"config_flow": false,
"iot_class": "cloud_polling"
},
"doods": {
"name": "DOODS - Dedicated Open Object Detection Service",
"integration_type": "hub",

View File

@@ -85,6 +85,7 @@ _ENTITY_COMPONENTS: set[str] = {platform.value for platform in Platform}.union(
"alert",
"automation",
"counter",
"dominos",
"input_boolean",
"input_button",
"input_datetime",

11
requirements_all.txt generated
View File

@@ -133,7 +133,7 @@ WSDiscovery==2.1.2
accuweather==4.2.2
# homeassistant.components.actron_air
actron-neo-api==0.1.87
actron-neo-api==0.1.84
# homeassistant.components.adax
adax==0.4.0
@@ -1263,7 +1263,7 @@ influxdb-client==1.48.0
influxdb==5.3.1
# homeassistant.components.inkbird
inkbird-ble==1.1.1
inkbird-ble==1.1.0
# homeassistant.components.insteon
insteon-frontend-home-assistant==0.5.0
@@ -1713,6 +1713,9 @@ pigpio==1.78
# homeassistant.components.pilight
pilight==0.1.1
# homeassistant.components.dominos
pizzapi==0.0.6
# homeassistant.components.plex
plexauth==0.0.6
@@ -2565,7 +2568,7 @@ python-telegram-bot[socks]==22.1
python-vlc==3.0.18122
# homeassistant.components.xbox
python-xbox==0.1.2
python-xbox==0.1.1
# homeassistant.components.egardia
pythonegardia==1.0.52
@@ -3039,7 +3042,7 @@ typedmonarchmoney==0.4.4
uasiren==0.0.1
# homeassistant.components.unifiprotect
uiprotect==7.26.0
uiprotect==7.23.0
# homeassistant.components.landisgyr_heat_meter
ultraheat-api==0.5.7

View File

@@ -121,7 +121,7 @@ WSDiscovery==2.1.2
accuweather==4.2.2
# homeassistant.components.actron_air
actron-neo-api==0.1.87
actron-neo-api==0.1.84
# homeassistant.components.adax
adax==0.4.0
@@ -1103,7 +1103,7 @@ influxdb-client==1.48.0
influxdb==5.3.1
# homeassistant.components.inkbird
inkbird-ble==1.1.1
inkbird-ble==1.1.0
# homeassistant.components.insteon
insteon-frontend-home-assistant==0.5.0
@@ -2134,7 +2134,7 @@ python-technove==2.0.0
python-telegram-bot[socks]==22.1
# homeassistant.components.xbox
python-xbox==0.1.2
python-xbox==0.1.1
# homeassistant.components.uptime_kuma
pythonkuma==0.3.2
@@ -2518,7 +2518,7 @@ typedmonarchmoney==0.4.4
uasiren==0.0.1
# homeassistant.components.unifiprotect
uiprotect==7.26.0
uiprotect==7.23.0
# homeassistant.components.landisgyr_heat_meter
ultraheat-api==0.5.7

View File

@@ -272,6 +272,8 @@ FORBIDDEN_PACKAGE_FILES_EXCEPTIONS = {
"abode": {"jaraco-abode": {"jaraco-net"}},
# https://github.com/coinbase/coinbase-advanced-py
"coinbase": {"homeassistant": {"coinbase-advanced-py"}},
# https://github.com/ggrammar/pizzapi
"dominos": {"homeassistant": {"pizzapi"}},
# https://github.com/u9n/dlms-cosem
"dsmr": {"dsmr-parser": {"dlms-cosem"}},
# https://github.com/ChrisMandich/PyFlume # Fixed with >=0.7.1

View File

@@ -12,11 +12,11 @@ def mock_actron_api() -> Generator[AsyncMock]:
"""Mock the Actron Air API class."""
with (
patch(
"homeassistant.components.actron_air.ActronAirAPI",
"homeassistant.components.actron_air.ActronNeoAPI",
autospec=True,
) as mock_api,
patch(
"homeassistant.components.actron_air.config_flow.ActronAirAPI",
"homeassistant.components.actron_air.config_flow.ActronNeoAPI",
new=mock_api,
),
):

View File

@@ -3,7 +3,7 @@
import asyncio
from unittest.mock import AsyncMock
from actron_neo_api import ActronAirAuthError
from actron_neo_api import ActronNeoAuthError
from homeassistant import config_entries
from homeassistant.components.actron_air.const import DOMAIN
@@ -76,7 +76,7 @@ async def test_user_flow_oauth2_error(hass: HomeAssistant, mock_actron_api) -> N
"""Test OAuth2 flow with authentication error during device code request."""
# Override the default mock to raise an error
mock_actron_api.request_device_code = AsyncMock(
side_effect=ActronAirAuthError("OAuth2 error")
side_effect=ActronNeoAuthError("OAuth2 error")
)
# Start the flow
@@ -95,7 +95,7 @@ async def test_user_flow_token_polling_error(
"""Test OAuth2 flow with error during token polling."""
# Override the default mock to raise an error during token polling
mock_actron_api.poll_for_token = AsyncMock(
side_effect=ActronAirAuthError("Token polling error")
side_effect=ActronNeoAuthError("Token polling error")
)
# Start the config flow

View File

@@ -90,7 +90,7 @@
"flow_rate_alarm": {
"value": false
},
"relay_alarm": {
"alarm_relay": {
"value": true
},
"relay_aux1": {
@@ -98,18 +98,6 @@
},
"relay_aux2": {
"value": false
},
"relay_aux3": {
"value": false
},
"alarm_ofa_ph": {
"value": false
},
"alarm_ofa_orp": {
"value": false
},
"alarm_ofa_cl": {
"value": false
}
},
"number": {

View File

@@ -1,540 +0,0 @@
# serializer version: 1
# name: test_all_binary_sensors[binary_sensor.pool_device_alarm_relay_status-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'binary_sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'binary_sensor.pool_device_alarm_relay_status',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <BinarySensorDeviceClass.POWER: 'power'>,
'original_icon': None,
'original_name': 'Alarm relay status',
'platform': 'pooldose',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'relay_alarm',
'unique_id': 'TEST123456789_relay_alarm',
'unit_of_measurement': None,
})
# ---
# name: test_all_binary_sensors[binary_sensor.pool_device_alarm_relay_status-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'power',
'friendly_name': 'Pool Device Alarm relay status',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.pool_device_alarm_relay_status',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'on',
})
# ---
# name: test_all_binary_sensors[binary_sensor.pool_device_auxiliary_relay_1_status-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'binary_sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'binary_sensor.pool_device_auxiliary_relay_1_status',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <BinarySensorDeviceClass.POWER: 'power'>,
'original_icon': None,
'original_name': 'Auxiliary relay 1 status',
'platform': 'pooldose',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'relay_aux1',
'unique_id': 'TEST123456789_relay_aux1',
'unit_of_measurement': None,
})
# ---
# name: test_all_binary_sensors[binary_sensor.pool_device_auxiliary_relay_1_status-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'power',
'friendly_name': 'Pool Device Auxiliary relay 1 status',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.pool_device_auxiliary_relay_1_status',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'off',
})
# ---
# name: test_all_binary_sensors[binary_sensor.pool_device_auxiliary_relay_2_status-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'binary_sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'binary_sensor.pool_device_auxiliary_relay_2_status',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <BinarySensorDeviceClass.POWER: 'power'>,
'original_icon': None,
'original_name': 'Auxiliary relay 2 status',
'platform': 'pooldose',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'relay_aux2',
'unique_id': 'TEST123456789_relay_aux2',
'unit_of_measurement': None,
})
# ---
# name: test_all_binary_sensors[binary_sensor.pool_device_auxiliary_relay_2_status-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'power',
'friendly_name': 'Pool Device Auxiliary relay 2 status',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.pool_device_auxiliary_relay_2_status',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'off',
})
# ---
# name: test_all_binary_sensors[binary_sensor.pool_device_auxiliary_relay_3_status-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'binary_sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'binary_sensor.pool_device_auxiliary_relay_3_status',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <BinarySensorDeviceClass.POWER: 'power'>,
'original_icon': None,
'original_name': 'Auxiliary relay 3 status',
'platform': 'pooldose',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'relay_aux3',
'unique_id': 'TEST123456789_relay_aux3',
'unit_of_measurement': None,
})
# ---
# name: test_all_binary_sensors[binary_sensor.pool_device_auxiliary_relay_3_status-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'power',
'friendly_name': 'Pool Device Auxiliary relay 3 status',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.pool_device_auxiliary_relay_3_status',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'off',
})
# ---
# name: test_all_binary_sensors[binary_sensor.pool_device_chlorine_tank_level-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'binary_sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'binary_sensor.pool_device_chlorine_tank_level',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <BinarySensorDeviceClass.PROBLEM: 'problem'>,
'original_icon': None,
'original_name': 'Chlorine tank level',
'platform': 'pooldose',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'alarm_ofa_cl',
'unique_id': 'TEST123456789_alarm_ofa_cl',
'unit_of_measurement': None,
})
# ---
# name: test_all_binary_sensors[binary_sensor.pool_device_chlorine_tank_level-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'problem',
'friendly_name': 'Pool Device Chlorine tank level',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.pool_device_chlorine_tank_level',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'off',
})
# ---
# name: test_all_binary_sensors[binary_sensor.pool_device_flow_rate-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'binary_sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'binary_sensor.pool_device_flow_rate',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <BinarySensorDeviceClass.PROBLEM: 'problem'>,
'original_icon': None,
'original_name': 'Flow rate',
'platform': 'pooldose',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'flow_rate_alarm',
'unique_id': 'TEST123456789_flow_rate_alarm',
'unit_of_measurement': None,
})
# ---
# name: test_all_binary_sensors[binary_sensor.pool_device_flow_rate-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'problem',
'friendly_name': 'Pool Device Flow rate',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.pool_device_flow_rate',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'off',
})
# ---
# name: test_all_binary_sensors[binary_sensor.pool_device_orp_overfeed-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'binary_sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'binary_sensor.pool_device_orp_overfeed',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <BinarySensorDeviceClass.PROBLEM: 'problem'>,
'original_icon': None,
'original_name': 'ORP overfeed',
'platform': 'pooldose',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'alarm_ofa_orp',
'unique_id': 'TEST123456789_alarm_ofa_orp',
'unit_of_measurement': None,
})
# ---
# name: test_all_binary_sensors[binary_sensor.pool_device_orp_overfeed-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'problem',
'friendly_name': 'Pool Device ORP overfeed',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.pool_device_orp_overfeed',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'off',
})
# ---
# name: test_all_binary_sensors[binary_sensor.pool_device_orp_tank_level-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'binary_sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'binary_sensor.pool_device_orp_tank_level',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <BinarySensorDeviceClass.PROBLEM: 'problem'>,
'original_icon': None,
'original_name': 'ORP tank level',
'platform': 'pooldose',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'orp_level_alarm',
'unique_id': 'TEST123456789_orp_level_alarm',
'unit_of_measurement': None,
})
# ---
# name: test_all_binary_sensors[binary_sensor.pool_device_orp_tank_level-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'problem',
'friendly_name': 'Pool Device ORP tank level',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.pool_device_orp_tank_level',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'off',
})
# ---
# name: test_all_binary_sensors[binary_sensor.pool_device_ph_overfeed-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'binary_sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'binary_sensor.pool_device_ph_overfeed',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <BinarySensorDeviceClass.PROBLEM: 'problem'>,
'original_icon': None,
'original_name': 'pH overfeed',
'platform': 'pooldose',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'alarm_ofa_ph',
'unique_id': 'TEST123456789_alarm_ofa_ph',
'unit_of_measurement': None,
})
# ---
# name: test_all_binary_sensors[binary_sensor.pool_device_ph_overfeed-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'problem',
'friendly_name': 'Pool Device pH overfeed',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.pool_device_ph_overfeed',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'off',
})
# ---
# name: test_all_binary_sensors[binary_sensor.pool_device_ph_tank_level-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'binary_sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'binary_sensor.pool_device_ph_tank_level',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <BinarySensorDeviceClass.PROBLEM: 'problem'>,
'original_icon': None,
'original_name': 'pH tank level',
'platform': 'pooldose',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'ph_level_alarm',
'unique_id': 'TEST123456789_ph_level_alarm',
'unit_of_measurement': None,
})
# ---
# name: test_all_binary_sensors[binary_sensor.pool_device_ph_tank_level-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'problem',
'friendly_name': 'Pool Device pH tank level',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.pool_device_ph_tank_level',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'off',
})
# ---
# name: test_all_binary_sensors[binary_sensor.pool_device_recirculation-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'binary_sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'binary_sensor.pool_device_recirculation',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <BinarySensorDeviceClass.PROBLEM: 'problem'>,
'original_icon': None,
'original_name': 'Recirculation',
'platform': 'pooldose',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'pump_alarm',
'unique_id': 'TEST123456789_pump_alarm',
'unit_of_measurement': None,
})
# ---
# name: test_all_binary_sensors[binary_sensor.pool_device_recirculation-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'problem',
'friendly_name': 'Pool Device Recirculation',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.pool_device_recirculation',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'on',
})
# ---

View File

@@ -1,193 +0,0 @@
"""Test the PoolDose binary sensor platform."""
from datetime import timedelta
from unittest.mock import AsyncMock, patch
from freezegun.api import FrozenDateTimeFactory
from pooldose.request_status import RequestStatus
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.const import STATE_OFF, STATE_ON, STATE_UNAVAILABLE, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from tests.common import MockConfigEntry, async_fire_time_changed, snapshot_platform
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
async def test_all_binary_sensors(
hass: HomeAssistant,
mock_pooldose_client: AsyncMock,
mock_config_entry: MockConfigEntry,
entity_registry: er.EntityRegistry,
snapshot: SnapshotAssertion,
) -> None:
"""Test the Pooldose binary sensors."""
with patch("homeassistant.components.pooldose.PLATFORMS", [Platform.BINARY_SENSOR]):
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
@pytest.mark.parametrize("exception", [TimeoutError, ConnectionError, OSError])
async def test_exception_raising(
hass: HomeAssistant,
mock_pooldose_client: AsyncMock,
mock_config_entry: MockConfigEntry,
exception: Exception,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test the Pooldose binary sensors."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert hass.states.get("binary_sensor.pool_device_recirculation").state == STATE_ON
mock_pooldose_client.instant_values_structured.side_effect = exception
freezer.tick(timedelta(minutes=10))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert (
hass.states.get("binary_sensor.pool_device_recirculation").state
== STATE_UNAVAILABLE
)
async def test_no_data(
hass: HomeAssistant,
mock_pooldose_client: AsyncMock,
mock_config_entry: MockConfigEntry,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test the Pooldose binary sensors."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert hass.states.get("binary_sensor.pool_device_recirculation").state == STATE_ON
mock_pooldose_client.instant_values_structured.return_value = (
RequestStatus.SUCCESS,
None,
)
freezer.tick(timedelta(minutes=10))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert (
hass.states.get("binary_sensor.pool_device_recirculation").state
== STATE_UNAVAILABLE
)
async def test_binary_sensor_entity_unavailable_no_coordinator_data(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_pooldose_client: AsyncMock,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test binary sensor entity becomes unavailable when coordinator has no data."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
# Verify initial working state
pump_state = hass.states.get("binary_sensor.pool_device_recirculation")
assert pump_state.state == STATE_ON
# Set coordinator data to None by making API return empty
mock_pooldose_client.instant_values_structured.return_value = (
RequestStatus.HOST_UNREACHABLE,
None,
)
freezer.tick(timedelta(minutes=10))
async_fire_time_changed(hass)
await hass.async_block_till_done()
# Check binary sensor becomes unavailable
pump_state = hass.states.get("binary_sensor.pool_device_recirculation")
assert pump_state.state == STATE_UNAVAILABLE
async def test_binary_sensor_state_changes(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_pooldose_client: AsyncMock,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test binary sensor state changes."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
# Verify initial states
pump_state = hass.states.get("binary_sensor.pool_device_recirculation")
assert pump_state.state == STATE_ON
ph_level_state = hass.states.get("binary_sensor.pool_device_ph_tank_level")
assert ph_level_state.state == STATE_OFF
# Update data with changed values
current_data = mock_pooldose_client.instant_values_structured.return_value[1]
updated_data = current_data.copy()
updated_data["binary_sensor"]["pump_alarm"]["value"] = False
updated_data["binary_sensor"]["ph_level_alarm"]["value"] = True
mock_pooldose_client.instant_values_structured.return_value = (
RequestStatus.SUCCESS,
updated_data,
)
freezer.tick(timedelta(minutes=10))
async_fire_time_changed(hass)
await hass.async_block_till_done()
# Check states have changed
pump_state = hass.states.get("binary_sensor.pool_device_recirculation")
assert pump_state.state == STATE_OFF
ph_level_state = hass.states.get("binary_sensor.pool_device_ph_tank_level")
assert ph_level_state.state == STATE_ON
async def test_binary_sensor_missing_from_data(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_pooldose_client: AsyncMock,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test binary sensor becomes unavailable when missing from coordinator data."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
# Verify initial working state
flow_alarm_state = hass.states.get("binary_sensor.pool_device_flow_rate")
assert flow_alarm_state.state == STATE_OFF
# Update data with missing sensor
current_data = mock_pooldose_client.instant_values_structured.return_value[1]
updated_data = current_data.copy()
del updated_data["binary_sensor"]["flow_rate_alarm"]
mock_pooldose_client.instant_values_structured.return_value = (
RequestStatus.SUCCESS,
updated_data,
)
freezer.tick(timedelta(minutes=10))
async_fire_time_changed(hass)
await hass.async_block_till_done()
# Check sensor becomes unavailable when not in coordinator data
flow_alarm_state = hass.states.get("binary_sensor.pool_device_flow_rate")
assert flow_alarm_state.state == STATE_UNAVAILABLE

View File

@@ -576,9 +576,6 @@ def _mock_rpc_device(version: str | None = None):
zigbee_enabled=False,
zigbee_firmware=False,
ip_address="10.10.10.10",
wifi_setconfig=AsyncMock(return_value={}),
ble_setconfig=AsyncMock(return_value={"restart_required": False}),
shutdown=AsyncMock(),
)
type(device).name = PropertyMock(return_value="Test name")
return device

View File

@@ -13,7 +13,6 @@ from aioshelly.exceptions import (
DeviceConnectionError,
InvalidAuthError,
InvalidHostError,
RpcCallError,
)
import pytest
@@ -2327,16 +2326,6 @@ async def test_bluetooth_wifi_scan_failure(
)
# Complete provisioning
mock_device = AsyncMock()
mock_device.initialize = AsyncMock()
mock_device.name = "Test name"
mock_device.status = {"sys": {}}
mock_device.xmod_info = {}
mock_device.shelly = {"model": MODEL_PLUS_2PM}
mock_device.wifi_setconfig = AsyncMock(return_value={})
mock_device.ble_setconfig = AsyncMock(return_value={"restart_required": False})
mock_device.shutdown = AsyncMock()
with (
patch(
"homeassistant.components.shelly.config_flow.async_provision_wifi",
@@ -2349,10 +2338,6 @@ async def test_bluetooth_wifi_scan_failure(
"homeassistant.components.shelly.config_flow.get_info",
return_value=MOCK_DEVICE_INFO,
),
patch(
"homeassistant.components.shelly.config_flow.RpcDevice.create",
return_value=mock_device,
),
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
@@ -2446,16 +2431,6 @@ async def test_bluetooth_wifi_credentials_and_provision_success(
assert result["step_id"] == "wifi_credentials"
# Enter password and provision
mock_device = AsyncMock()
mock_device.initialize = AsyncMock()
mock_device.name = "Test name"
mock_device.status = {"sys": {}}
mock_device.xmod_info = {}
mock_device.shelly = {"model": MODEL_PLUS_2PM}
mock_device.wifi_setconfig = AsyncMock(return_value={})
mock_device.ble_setconfig = AsyncMock(return_value={"restart_required": False})
mock_device.shutdown = AsyncMock()
with (
patch(
"homeassistant.components.shelly.config_flow.async_provision_wifi",
@@ -2468,10 +2443,6 @@ async def test_bluetooth_wifi_credentials_and_provision_success(
"homeassistant.components.shelly.config_flow.get_info",
return_value=MOCK_DEVICE_INFO,
),
patch(
"homeassistant.components.shelly.config_flow.RpcDevice.create",
return_value=mock_device,
),
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
@@ -3028,17 +2999,6 @@ async def test_bluetooth_provision_with_zeroconf_discovery_fast_path(
# Ensure the zeroconf discovery completes before returning
await hass.async_block_till_done()
# Mock device for secure device feature
mock_device = AsyncMock()
mock_device.initialize = AsyncMock()
mock_device.name = "Test name"
mock_device.status = {"sys": {}}
mock_device.xmod_info = {}
mock_device.shelly = {"model": MODEL_PLUS_2PM}
mock_device.wifi_setconfig = AsyncMock(return_value={})
mock_device.ble_setconfig = AsyncMock(return_value={"restart_required": False})
mock_device.shutdown = AsyncMock()
with (
patch(
"homeassistant.components.shelly.config_flow.PROVISIONING_TIMEOUT",
@@ -3056,10 +3016,6 @@ async def test_bluetooth_provision_with_zeroconf_discovery_fast_path(
"homeassistant.components.shelly.config_flow.get_info",
return_value=MOCK_DEVICE_INFO,
),
patch(
"homeassistant.components.shelly.config_flow.RpcDevice.create",
return_value=mock_device,
),
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
@@ -3146,390 +3102,6 @@ async def test_bluetooth_provision_timeout_active_lookup_fails(
assert result["reason"] == "unknown"
async def test_bluetooth_provision_secure_device_both_enabled(
hass: HomeAssistant,
mock_setup_entry: AsyncMock,
mock_setup: AsyncMock,
) -> None:
"""Test provisioning with both AP and BLE disable enabled (default)."""
inject_bluetooth_service_info_bleak(hass, BLE_DISCOVERY_INFO)
result = await hass.config_entries.flow.async_init(
DOMAIN,
data=BLE_DISCOVERY_INFO,
context={"source": config_entries.SOURCE_BLUETOOTH},
)
# Confirm with both switches enabled (default)
with patch(
"homeassistant.components.shelly.config_flow.async_scan_wifi_networks",
return_value=[{"ssid": "MyNetwork", "rssi": -50, "auth": 2}],
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{"disable_ap": True, "disable_ble_rpc": True},
)
# Select network
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_SSID: "MyNetwork"},
)
# Provision and verify security calls
mock_device = AsyncMock()
mock_device.initialize = AsyncMock()
mock_device.wifi_setconfig = AsyncMock(return_value={})
mock_device.ble_setconfig = AsyncMock(return_value={"restart_required": False})
mock_device.shutdown = AsyncMock()
with (
patch("homeassistant.components.shelly.config_flow.async_provision_wifi"),
patch(
"homeassistant.components.shelly.config_flow.async_lookup_device_by_name",
return_value=("1.1.1.1", 80),
),
patch(
"homeassistant.components.shelly.config_flow.get_info",
return_value=MOCK_DEVICE_INFO,
),
patch(
"homeassistant.components.shelly.config_flow.RpcDevice.create",
return_value=mock_device,
),
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_PASSWORD: "my_password"},
)
await hass.async_block_till_done()
result = await hass.config_entries.flow.async_configure(result["flow_id"])
# Verify entry created
assert result["type"] is FlowResultType.CREATE_ENTRY
# Verify security calls were made
mock_device.wifi_setconfig.assert_called_once_with(ap_enable=False)
mock_device.ble_setconfig.assert_called_once_with(enable=True, enable_rpc=False)
assert mock_device.shutdown.called
async def test_bluetooth_provision_secure_device_both_disabled(
hass: HomeAssistant,
mock_setup_entry: AsyncMock,
mock_setup: AsyncMock,
) -> None:
"""Test provisioning with both AP and BLE disable disabled."""
inject_bluetooth_service_info_bleak(hass, BLE_DISCOVERY_INFO)
result = await hass.config_entries.flow.async_init(
DOMAIN,
data=BLE_DISCOVERY_INFO,
context={"source": config_entries.SOURCE_BLUETOOTH},
)
# Confirm with both switches disabled
with patch(
"homeassistant.components.shelly.config_flow.async_scan_wifi_networks",
return_value=[{"ssid": "MyNetwork", "rssi": -50, "auth": 2}],
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{"disable_ap": False, "disable_ble_rpc": False},
)
# Select network
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_SSID: "MyNetwork"},
)
# Provision - with both disabled, secure device method should not create device
with (
patch("homeassistant.components.shelly.config_flow.async_provision_wifi"),
patch(
"homeassistant.components.shelly.config_flow.async_lookup_device_by_name",
return_value=("1.1.1.1", 80),
),
patch(
"homeassistant.components.shelly.config_flow.get_info",
return_value=MOCK_DEVICE_INFO,
),
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_PASSWORD: "my_password"},
)
await hass.async_block_till_done()
result = await hass.config_entries.flow.async_configure(result["flow_id"])
# Verify entry created (secure device call is skipped when both disabled)
assert result["type"] is FlowResultType.CREATE_ENTRY
async def test_bluetooth_provision_secure_device_only_ap_disabled(
hass: HomeAssistant,
mock_setup_entry: AsyncMock,
mock_setup: AsyncMock,
) -> None:
"""Test provisioning with only AP disable enabled."""
inject_bluetooth_service_info_bleak(hass, BLE_DISCOVERY_INFO)
result = await hass.config_entries.flow.async_init(
DOMAIN,
data=BLE_DISCOVERY_INFO,
context={"source": config_entries.SOURCE_BLUETOOTH},
)
# Confirm with only AP disable
with patch(
"homeassistant.components.shelly.config_flow.async_scan_wifi_networks",
return_value=[{"ssid": "MyNetwork", "rssi": -50, "auth": 2}],
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{"disable_ap": True, "disable_ble_rpc": False},
)
# Select network
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_SSID: "MyNetwork"},
)
# Provision and verify only AP disabled
mock_device = AsyncMock()
mock_device.initialize = AsyncMock()
mock_device.wifi_setconfig = AsyncMock(return_value={})
mock_device.shutdown = AsyncMock()
with (
patch("homeassistant.components.shelly.config_flow.async_provision_wifi"),
patch(
"homeassistant.components.shelly.config_flow.async_lookup_device_by_name",
return_value=("1.1.1.1", 80),
),
patch(
"homeassistant.components.shelly.config_flow.get_info",
return_value=MOCK_DEVICE_INFO,
),
patch(
"homeassistant.components.shelly.config_flow.RpcDevice.create",
return_value=mock_device,
),
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_PASSWORD: "my_password"},
)
await hass.async_block_till_done()
result = await hass.config_entries.flow.async_configure(result["flow_id"])
# Verify entry created
assert result["type"] is FlowResultType.CREATE_ENTRY
# Verify only wifi_setconfig was called
mock_device.wifi_setconfig.assert_called_once_with(ap_enable=False)
assert mock_device.shutdown.called
async def test_bluetooth_provision_secure_device_only_ble_disabled(
hass: HomeAssistant,
mock_setup_entry: AsyncMock,
mock_setup: AsyncMock,
) -> None:
"""Test provisioning with only BLE disable enabled."""
inject_bluetooth_service_info_bleak(hass, BLE_DISCOVERY_INFO)
result = await hass.config_entries.flow.async_init(
DOMAIN,
data=BLE_DISCOVERY_INFO,
context={"source": config_entries.SOURCE_BLUETOOTH},
)
# Confirm with only BLE disable
with patch(
"homeassistant.components.shelly.config_flow.async_scan_wifi_networks",
return_value=[{"ssid": "MyNetwork", "rssi": -50, "auth": 2}],
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{"disable_ap": False, "disable_ble_rpc": True},
)
# Select network
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_SSID: "MyNetwork"},
)
# Provision and verify only BLE disabled
mock_device = AsyncMock()
mock_device.initialize = AsyncMock()
mock_device.ble_setconfig = AsyncMock(return_value={"restart_required": False})
mock_device.shutdown = AsyncMock()
with (
patch("homeassistant.components.shelly.config_flow.async_provision_wifi"),
patch(
"homeassistant.components.shelly.config_flow.async_lookup_device_by_name",
return_value=("1.1.1.1", 80),
),
patch(
"homeassistant.components.shelly.config_flow.get_info",
return_value=MOCK_DEVICE_INFO,
),
patch(
"homeassistant.components.shelly.config_flow.RpcDevice.create",
return_value=mock_device,
),
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_PASSWORD: "my_password"},
)
await hass.async_block_till_done()
result = await hass.config_entries.flow.async_configure(result["flow_id"])
# Verify entry created
assert result["type"] is FlowResultType.CREATE_ENTRY
# Verify only ble_setconfig was called
mock_device.ble_setconfig.assert_called_once_with(enable=True, enable_rpc=False)
assert mock_device.shutdown.called
async def test_bluetooth_provision_secure_device_with_restart_required(
hass: HomeAssistant,
mock_setup_entry: AsyncMock,
mock_setup: AsyncMock,
) -> None:
"""Test provisioning when BLE disable requires restart."""
inject_bluetooth_service_info_bleak(hass, BLE_DISCOVERY_INFO)
result = await hass.config_entries.flow.async_init(
DOMAIN,
data=BLE_DISCOVERY_INFO,
context={"source": config_entries.SOURCE_BLUETOOTH},
)
# Confirm with both enabled
with patch(
"homeassistant.components.shelly.config_flow.async_scan_wifi_networks",
return_value=[{"ssid": "MyNetwork", "rssi": -50, "auth": 2}],
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{"disable_ap": True, "disable_ble_rpc": True},
)
# Select network
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_SSID: "MyNetwork"},
)
# Provision and verify restart is triggered
mock_device = AsyncMock()
mock_device.initialize = AsyncMock()
mock_device.wifi_setconfig = AsyncMock(return_value={})
mock_device.ble_setconfig = AsyncMock(return_value={"restart_required": True})
mock_device.trigger_reboot = AsyncMock()
mock_device.shutdown = AsyncMock()
with (
patch("homeassistant.components.shelly.config_flow.async_provision_wifi"),
patch(
"homeassistant.components.shelly.config_flow.async_lookup_device_by_name",
return_value=("1.1.1.1", 80),
),
patch(
"homeassistant.components.shelly.config_flow.get_info",
return_value=MOCK_DEVICE_INFO,
),
patch(
"homeassistant.components.shelly.config_flow.RpcDevice.create",
return_value=mock_device,
),
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_PASSWORD: "my_password"},
)
await hass.async_block_till_done()
result = await hass.config_entries.flow.async_configure(result["flow_id"])
# Verify entry created
assert result["type"] is FlowResultType.CREATE_ENTRY
# Verify restart was triggered and shutdown called
mock_device.trigger_reboot.assert_called_once_with(delay_ms=1000)
assert mock_device.shutdown.called
async def test_bluetooth_provision_secure_device_fails_gracefully(
hass: HomeAssistant,
mock_setup_entry: AsyncMock,
mock_setup: AsyncMock,
) -> None:
"""Test provisioning succeeds even when secure device calls fail."""
inject_bluetooth_service_info_bleak(hass, BLE_DISCOVERY_INFO)
result = await hass.config_entries.flow.async_init(
DOMAIN,
data=BLE_DISCOVERY_INFO,
context={"source": config_entries.SOURCE_BLUETOOTH},
)
# Confirm with both enabled
with patch(
"homeassistant.components.shelly.config_flow.async_scan_wifi_networks",
return_value=[{"ssid": "MyNetwork", "rssi": -50, "auth": 2}],
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{"disable_ap": True, "disable_ble_rpc": True},
)
# Select network
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_SSID: "MyNetwork"},
)
# Provision with security calls failing - wifi_setconfig will fail
mock_device = AsyncMock()
mock_device.initialize = AsyncMock()
mock_device.wifi_setconfig = AsyncMock(side_effect=RpcCallError("RPC call failed"))
mock_device.shutdown = AsyncMock()
with (
patch("homeassistant.components.shelly.config_flow.async_provision_wifi"),
patch(
"homeassistant.components.shelly.config_flow.async_lookup_device_by_name",
return_value=("1.1.1.1", 80),
),
patch(
"homeassistant.components.shelly.config_flow.get_info",
return_value=MOCK_DEVICE_INFO,
),
patch(
"homeassistant.components.shelly.config_flow.RpcDevice.create",
return_value=mock_device,
),
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_PASSWORD: "my_password"},
)
await hass.async_block_till_done()
result = await hass.config_entries.flow.async_configure(result["flow_id"])
# Verify entry still created despite secure device failure
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["result"].unique_id == "C049EF8873E8"
async def test_zeroconf_aborts_idle_ble_flow(
hass: HomeAssistant,
mock_setup_entry: AsyncMock,

View File

@@ -143,54 +143,6 @@
'state': 'unknown',
})
# ---
# name: test_button[ex30_2024][button.volvo_ex30_lock_reduced_guard-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'button',
'entity_category': None,
'entity_id': 'button.volvo_ex30_lock_reduced_guard',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Lock reduced guard',
'platform': 'volvo',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'lock_reduced_guard',
'unique_id': 'yv1abcdefg1234567_lock_reduced_guard',
'unit_of_measurement': None,
})
# ---
# name: test_button[ex30_2024][button.volvo_ex30_lock_reduced_guard-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Volvo EX30 Lock reduced guard',
}),
'context': <ANY>,
'entity_id': 'button.volvo_ex30_lock_reduced_guard',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
# name: test_button[ex30_2024][button.volvo_ex30_start_climatization-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
@@ -431,54 +383,6 @@
'state': 'unknown',
})
# ---
# name: test_button[s90_diesel_2018][button.volvo_s90_lock_reduced_guard-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'button',
'entity_category': None,
'entity_id': 'button.volvo_s90_lock_reduced_guard',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Lock reduced guard',
'platform': 'volvo',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'lock_reduced_guard',
'unique_id': 'yv1abcdefg1234567_lock_reduced_guard',
'unit_of_measurement': None,
})
# ---
# name: test_button[s90_diesel_2018][button.volvo_s90_lock_reduced_guard-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Volvo S90 Lock reduced guard',
}),
'context': <ANY>,
'entity_id': 'button.volvo_s90_lock_reduced_guard',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
# name: test_button[s90_diesel_2018][button.volvo_s90_start_climatization-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
@@ -719,54 +623,6 @@
'state': 'unknown',
})
# ---
# name: test_button[xc40_electric_2024][button.volvo_xc40_lock_reduced_guard-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'button',
'entity_category': None,
'entity_id': 'button.volvo_xc40_lock_reduced_guard',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Lock reduced guard',
'platform': 'volvo',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'lock_reduced_guard',
'unique_id': 'yv1abcdefg1234567_lock_reduced_guard',
'unit_of_measurement': None,
})
# ---
# name: test_button[xc40_electric_2024][button.volvo_xc40_lock_reduced_guard-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Volvo XC40 Lock reduced guard',
}),
'context': <ANY>,
'entity_id': 'button.volvo_xc40_lock_reduced_guard',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
# name: test_button[xc40_electric_2024][button.volvo_xc40_start_climatization-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
@@ -1007,54 +863,6 @@
'state': 'unknown',
})
# ---
# name: test_button[xc90_petrol_2019][button.volvo_xc90_lock_reduced_guard-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'button',
'entity_category': None,
'entity_id': 'button.volvo_xc90_lock_reduced_guard',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Lock reduced guard',
'platform': 'volvo',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'lock_reduced_guard',
'unique_id': 'yv1abcdefg1234567_lock_reduced_guard',
'unit_of_measurement': None,
})
# ---
# name: test_button[xc90_petrol_2019][button.volvo_xc90_lock_reduced_guard-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Volvo XC90 Lock reduced guard',
}),
'context': <ANY>,
'entity_id': 'button.volvo_xc90_lock_reduced_guard',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
# name: test_button[xc90_petrol_2019][button.volvo_xc90_start_climatization-entry]
EntityRegistryEntrySnapshot({
'aliases': set({

View File

@@ -43,14 +43,7 @@ async def test_button(
@pytest.mark.usefixtures("full_model")
@pytest.mark.parametrize(
"command",
[
"start_climatization",
"stop_climatization",
"flash",
"honk",
"honk_flash",
"lock_reduced_guard",
],
["start_climatization", "stop_climatization", "flash", "honk", "honk_flash"],
)
async def test_button_press(
hass: HomeAssistant,
@@ -79,14 +72,7 @@ async def test_button_press(
@pytest.mark.usefixtures("full_model")
@pytest.mark.parametrize(
"command",
[
"start_climatization",
"stop_climatization",
"flash",
"honk",
"honk_flash",
"lock_reduced_guard",
],
["start_climatization", "stop_climatization", "flash", "honk", "honk_flash"],
)
async def test_button_press_error(
hass: HomeAssistant,
@@ -113,14 +99,7 @@ async def test_button_press_error(
@pytest.mark.usefixtures("full_model")
@pytest.mark.parametrize(
"command",
[
"start_climatization",
"stop_climatization",
"flash",
"honk",
"honk_flash",
"lock_reduced_guard",
],
["start_climatization", "stop_climatization", "flash", "honk", "honk_flash"],
)
async def test_button_press_failure(
hass: HomeAssistant,

File diff suppressed because it is too large Load Diff

View File

@@ -1,32 +0,0 @@
"""Test for diagnostics platform of the Xbox integration."""
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
from tests.components.diagnostics import get_diagnostics_for_config_entry
from tests.typing import ClientSessionGenerator
@pytest.mark.usefixtures("xbox_live_client")
async def test_diagnostics(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
config_entry: MockConfigEntry,
snapshot: SnapshotAssertion,
) -> None:
"""Test diagnostics."""
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is ConfigEntryState.LOADED
assert (
await get_diagnostics_for_config_entry(hass, hass_client, config_entry)
== snapshot
)