diff --git a/.strict-typing b/.strict-typing index d5e01d6ea7ff..b4f3b14ff01f 100644 --- a/.strict-typing +++ b/.strict-typing @@ -118,6 +118,7 @@ homeassistant.components.bayesian.* homeassistant.components.besen.* homeassistant.components.binary_sensor.* homeassistant.components.bitcoin.* +homeassistant.components.bitvis.* homeassistant.components.blockchain.* homeassistant.components.blue_current.* homeassistant.components.blueprint.* diff --git a/CODEOWNERS b/CODEOWNERS index c7a1b38d075a..b627d73220ba 100644 --- a/CODEOWNERS +++ b/CODEOWNERS @@ -233,6 +233,8 @@ CLAUDE.md @home-assistant/core /tests/components/besen/ @moryoav /homeassistant/components/binary_sensor/ @home-assistant/core /tests/components/binary_sensor/ @home-assistant/core +/homeassistant/components/bitvis/ @MandusBorjesson @real-tintin @simontegelid +/tests/components/bitvis/ @MandusBorjesson @real-tintin @simontegelid /homeassistant/components/bizkaibus/ @UgaitzEtxebarria /homeassistant/components/blebox/ @bbx-a @swistakm @bkobus-bbx /tests/components/blebox/ @bbx-a @swistakm @bkobus-bbx diff --git a/homeassistant/components/bitvis/__init__.py b/homeassistant/components/bitvis/__init__.py new file mode 100644 index 000000000000..cb1f689a0d73 --- /dev/null +++ b/homeassistant/components/bitvis/__init__.py @@ -0,0 +1,56 @@ +"""The Bitvis Power Hub integration.""" + +from typing import TYPE_CHECKING + +from homeassistant.const import CONF_PORT, Platform +from homeassistant.core import HomeAssistant +from homeassistant.helpers import config_validation as cv +from homeassistant.helpers.typing import ConfigType + +from .const import DATA_LISTENER_REGISTRY, DOMAIN +from .coordinator import ( + BitvisConfigEntry, + BitvisDataUpdateCoordinator, + async_get_listener_registry, +) + +_PLATFORMS: list[Platform] = [Platform.SENSOR] + +CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN) + + +async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool: + """Set up the Bitvis Power Hub integration.""" + async_get_listener_registry(hass) + return True + + +async def async_setup_entry(hass: HomeAssistant, entry: BitvisConfigEntry) -> bool: + """Set up Bitvis Power Hub from a config entry.""" + async_get_listener_registry(hass) + if TYPE_CHECKING: + assert entry.unique_id is not None + coordinator = BitvisDataUpdateCoordinator( + hass, + entry, + entry.data[CONF_PORT], + entry.unique_id, + ) + + await coordinator.async_config_entry_first_refresh() + + entry.runtime_data = coordinator + + await hass.config_entries.async_forward_entry_setups(entry, _PLATFORMS) + + return True + + +async def async_unload_entry(hass: HomeAssistant, entry: BitvisConfigEntry) -> bool: + """Unload a config entry.""" + unload_ok = await hass.config_entries.async_unload_platforms(entry, _PLATFORMS) + if unload_ok: + await entry.runtime_data.async_stop() + if not hass.config_entries.async_loaded_entries(DOMAIN): + hass.data.pop(DATA_LISTENER_REGISTRY, None) + return unload_ok diff --git a/homeassistant/components/bitvis/config_flow.py b/homeassistant/components/bitvis/config_flow.py new file mode 100644 index 000000000000..21f8aff7e324 --- /dev/null +++ b/homeassistant/components/bitvis/config_flow.py @@ -0,0 +1,235 @@ +"""Config flow for the Bitvis Power Hub integration.""" + +import asyncio +import logging +from typing import Any, Self, override + +from bitvis_protobuf.listener import FilterIp +from bitvis_protobuf.parse import PayloadDiagnostic, PayloadSample +from bitvis_protobuf.utils import ( + InvalidMacAddressError, + async_resolve_host, + async_verify_udp_port_bindable, + normalize_host, +) +import voluptuous as vol + +from homeassistant.config_entries import ConfigFlow, ConfigFlowResult +from homeassistant.const import CONF_HOST, CONF_PORT +from homeassistant.core import HomeAssistant, callback +from homeassistant.data_entry_flow import AbortFlow +from homeassistant.helpers import config_validation as cv +from homeassistant.helpers.device_registry import format_mac +from homeassistant.helpers.service_info.zeroconf import ZeroconfServiceInfo + +from .const import DEFAULT_NAME, DEFAULT_PORT, DISCOVERY_TIMEOUT, DOMAIN +from .coordinator import async_get_listener_registry + +_LOGGER = logging.getLogger(__name__) + +STEP_USER_DATA_SCHEMA = vol.Schema( + { + vol.Required(CONF_HOST): cv.string, + } +) + + +def _get_friendly_name(name: str | None) -> str: + """Return a user-friendly name derived from the zeroconf name.""" + if not name: + return DEFAULT_NAME + instance = name.split(".", 1)[0] + return instance or DEFAULT_NAME + + +async def _async_test_port(hass: HomeAssistant, port: int) -> None: + """Verify the UDP port can be bound.""" + + if async_get_listener_registry(hass).has_listener(port): + return + + await async_verify_udp_port_bindable(port) + + +async def _async_discover_mac_address(hass: HomeAssistant, host: str, port: int) -> str: + """Wait for a UDP message from the device and return its MAC address.""" + resolved_ips = await async_resolve_host(host) + listener_registry = async_get_listener_registry(hass) + listener = await listener_registry.async_get_or_create(port) + + loop = asyncio.get_running_loop() + future: asyncio.Future[str] = loop.create_future() + + @callback + def _on_payload( + payload: PayloadSample | PayloadDiagnostic, _addr: tuple[str, int] + ) -> None: + if not future.done(): + future.set_result(payload.mac_address) + + @callback + def _on_error(err: Exception, addr: tuple[str, int]) -> None: + if ( + addr[0] in resolved_ips + and not future.done() + and isinstance(err, InvalidMacAddressError) + ): + future.set_exception(err) + + filters: list[FilterIp] = [] + listener.register_error_callback(_on_error) + try: + for ip in resolved_ips: + filt = FilterIp(ip) + try: + listener.register(filt, _on_payload) + except RuntimeError as err: + raise AbortFlow("already_in_progress") from err + filters.append(filt) + + return await asyncio.wait_for(future, timeout=DISCOVERY_TIMEOUT) + finally: + listener.unregister_error_callback(_on_error) + for filt in filters: + listener.unregister(filt) + await listener_registry.async_remove_if_unused(port) + + +class BitvisConfigFlow(ConfigFlow, domain=DOMAIN): + """Handle a config flow for Bitvis Power Hub.""" + + def __init__(self) -> None: + """Initialize the config flow.""" + self._discovery_info: ZeroconfServiceInfo | None = None + self._validated_host: str | None = None + self._host: str | None = None + + @override + def is_matching(self, other_flow: Self) -> bool: + """Return True if other_flow is matching this flow.""" + return self._host is not None and self._host == other_flow._host + + async def _async_validate_host(self, host: str) -> str: + """Verify port availability and discover the device MAC address.""" + await _async_test_port(self.hass, DEFAULT_PORT) + return await _async_discover_mac_address(self.hass, host, DEFAULT_PORT) + + async def _async_create_entry_from_host( + self, host: str, title: str + ) -> ConfigFlowResult: + """Validate connectivity, discover MAC address, and create the entry.""" + self._host = host + if self.hass.config_entries.flow.async_has_matching_flow(self): + return self.async_abort(reason="already_in_progress") + + try: + mac_address = await self._async_validate_host(host) + except TimeoutError: + return self.async_show_form( + step_id="user", + data_schema=self.add_suggested_values_to_schema( + STEP_USER_DATA_SCHEMA, {CONF_HOST: host} + ), + errors={"base": "timeout_connect"}, + ) + except InvalidMacAddressError: + return self.async_show_form( + step_id="user", + data_schema=self.add_suggested_values_to_schema( + STEP_USER_DATA_SCHEMA, {CONF_HOST: host} + ), + errors={"base": "invalid_mac"}, + ) + except OSError: + return self.async_show_form( + step_id="user", + data_schema=self.add_suggested_values_to_schema( + STEP_USER_DATA_SCHEMA, {CONF_HOST: host} + ), + errors={"base": "cannot_connect"}, + ) + + await self.async_set_unique_id(format_mac(mac_address)) + self._abort_if_unique_id_configured() + + return self.async_create_entry( + title=title, + data={ + CONF_HOST: host, + CONF_PORT: DEFAULT_PORT, + }, + ) + + @override + async def async_step_user( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: + """Handle the initial step.""" + if user_input is not None: + host = normalize_host(user_input[CONF_HOST]) + return await self._async_create_entry_from_host(host, DEFAULT_NAME) + + return self.async_show_form( + step_id="user", + data_schema=STEP_USER_DATA_SCHEMA, + ) + + @override + async def async_step_zeroconf( + self, discovery_info: ZeroconfServiceInfo + ) -> ConfigFlowResult: + """Handle zeroconf discovery.""" + _LOGGER.debug("Discovered Bitvis Power Hub via Zeroconf: %s", discovery_info) + + host = discovery_info.host + self._host = host + + if self.hass.config_entries.flow.async_has_matching_flow(self): + return self.async_abort(reason="already_in_progress") + + try: + mac_address = await self._async_validate_host(host) + except TimeoutError: + return self.async_abort(reason="timeout_connect") + except InvalidMacAddressError: + return self.async_abort(reason="invalid_mac") + except OSError: + return self.async_abort(reason="cannot_connect") + + await self.async_set_unique_id(format_mac(mac_address)) + self._abort_if_unique_id_configured(updates={CONF_HOST: host}) + + self._discovery_info = discovery_info + self._validated_host = host + + self.context["title_placeholders"] = { + "name": _get_friendly_name(discovery_info.name), + "host": host, + } + + return await self.async_step_zeroconf_confirm() + + async def async_step_zeroconf_confirm( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: + """Confirm discovery.""" + assert self._discovery_info is not None + + if user_input is not None: + assert self._validated_host is not None + + return self.async_create_entry( + title=_get_friendly_name(self._discovery_info.name), + data={ + CONF_HOST: self._validated_host, + CONF_PORT: DEFAULT_PORT, + }, + ) + + return self.async_show_form( + step_id="zeroconf_confirm", + description_placeholders={ + "name": _get_friendly_name(self._discovery_info.name), + "host": self._discovery_info.host, + }, + ) diff --git a/homeassistant/components/bitvis/const.py b/homeassistant/components/bitvis/const.py new file mode 100644 index 000000000000..c1bdd6d85e42 --- /dev/null +++ b/homeassistant/components/bitvis/const.py @@ -0,0 +1,18 @@ +"""Constants for the Bitvis Power Hub integration.""" + +from typing import TYPE_CHECKING + +from homeassistant.util.hass_dict import HassKey + +if TYPE_CHECKING: + from .coordinator import BitvisListenerRegistry + +DOMAIN = "bitvis" +MANUFACTURER = "Bitvis" +MODEL_NAME = "Power Hub" + +DEFAULT_NAME = "Bitvis Power Hub" +DEFAULT_PORT = 58220 +DISCOVERY_TIMEOUT = 30 + +DATA_LISTENER_REGISTRY: HassKey[BitvisListenerRegistry] = HassKey(DOMAIN) diff --git a/homeassistant/components/bitvis/coordinator.py b/homeassistant/components/bitvis/coordinator.py new file mode 100644 index 000000000000..06ef13592fd6 --- /dev/null +++ b/homeassistant/components/bitvis/coordinator.py @@ -0,0 +1,197 @@ +"""Data coordinator for Bitvis Power Hub.""" + +import asyncio +from dataclasses import dataclass +from datetime import datetime, timedelta +import logging +from typing import override + +from bitvis_protobuf.listener import FilterMac, SharedListener +from bitvis_protobuf.parse import PayloadDiagnostic, PayloadSample + +from homeassistant.config_entries import ConfigEntry +from homeassistant.core import HomeAssistant, callback +from homeassistant.exceptions import ConfigEntryError +from homeassistant.helpers import device_registry as dr +from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed +from homeassistant.util import dt as dt_util +from homeassistant.util.variance import ignore_variance + +from .const import DATA_LISTENER_REGISTRY, DOMAIN, MODEL_NAME + +_LOGGER = logging.getLogger(__name__) + +type BitvisConfigEntry = ConfigEntry[BitvisDataUpdateCoordinator] + + +def _uptime_to_boot_time(uptime_s: int) -> datetime: + """Convert uptime in seconds to an absolute boot datetime.""" + return dt_util.utcnow().replace(microsecond=0) - timedelta(seconds=uptime_s) + + +@dataclass(kw_only=True) +class BitvisData: + """Data structure for Bitvis measurements.""" + + sample: PayloadSample | None = None + diagnostic: PayloadDiagnostic | None = None + boot_time: datetime | None = None + + +class BitvisListenerRegistry: + """Registry that manages one shared UDP listener per port. + + Stored at hass.data[DATA_LISTENER_REGISTRY] so all coordinators can + look it up without duplicating state-management logic. + """ + + def __init__(self) -> None: + """Initialize registry storage.""" + self._listeners: dict[int, SharedListener] = {} + self._locks: dict[int, asyncio.Lock] = {} + + async def async_get_or_create(self, port: int) -> SharedListener: + """Return the listener for *port*, creating and starting it if needed.""" + port_lock = self._locks.setdefault(port, asyncio.Lock()) + async with port_lock: + if port not in self._listeners: + listener = SharedListener() + await listener.start(port) + self._listeners[port] = listener + return self._listeners[port] + + async def async_remove_if_unused(self, port: int) -> None: + """Stop and remove the listener for *port* when no coordinators remain.""" + port_lock = self._locks.setdefault(port, asyncio.Lock()) + async with port_lock: + listener = self._listeners.get(port) + if listener is None or not listener.is_empty: + return + await listener.stop() + del self._listeners[port] + + def get(self, port: int) -> SharedListener | None: + """Return an existing listener for *port*, or None.""" + return self._listeners.get(port) + + def has_listener(self, port: int) -> bool: + """Return True if a listener is already active on *port*.""" + return port in self._listeners + + +def async_get_listener_registry(hass: HomeAssistant) -> BitvisListenerRegistry: + """Return (creating if needed) the Bitvis listener registry for this HA instance.""" + if DATA_LISTENER_REGISTRY not in hass.data: + hass.data[DATA_LISTENER_REGISTRY] = BitvisListenerRegistry() + return hass.data[DATA_LISTENER_REGISTRY] + + +class BitvisDataUpdateCoordinator(DataUpdateCoordinator[BitvisData]): + """Coordinator to manage data updates from UDP packets.""" + + config_entry: BitvisConfigEntry + + def __init__( + self, + hass: HomeAssistant, + config_entry: BitvisConfigEntry, + port: int, + mac_address: str, + ) -> None: + """Initialize the coordinator.""" + super().__init__( + hass, + _LOGGER, + name=DOMAIN, + config_entry=config_entry, + ) + self.port = port + self.mac_address = mac_address + self._filter = FilterMac(mac_address) + self._registered = False + self._stable_boot_time = ignore_variance( + _uptime_to_boot_time, timedelta(minutes=5) + ) + self.data = BitvisData() + + @override + async def _async_setup(self) -> None: + """Set up the coordinator by registering with the shared UDP listener.""" + try: + listener = await self.hass.data[DATA_LISTENER_REGISTRY].async_get_or_create( + self.port + ) + listener.register(self._filter, self._handle_payload) + self._registered = True + except OSError as err: + raise UpdateFailed( + f"Failed to start UDP listener on port {self.port}" + ) from err + except RuntimeError as err: + raise ConfigEntryError( + f"Failed to register MAC filter for {self.mac_address} " + f"on port {self.port}" + ) from err + + async def async_stop(self) -> None: + """Unregister from the shared listener, stopping it when no longer needed.""" + if not self._registered: + return + + listener_registry = self.hass.data[DATA_LISTENER_REGISTRY] + if listener := listener_registry.get(self.port): + listener.unregister(self._filter) + await listener_registry.async_remove_if_unused(self.port) + + self._registered = False + _LOGGER.debug( + "Unregistered coordinator from shared UDP listener for port %s", self.port + ) + + @callback + def _handle_payload( + self, + payload: PayloadSample | PayloadDiagnostic, + addr: tuple[str, int], + ) -> None: + """Handle a parsed payload dispatched by the shared listener.""" + _LOGGER.debug("Received payload from %s", addr) + if isinstance(payload, PayloadSample): + self._handle_sample(payload) + else: + self._handle_diagnostic(payload) + + @callback + def _handle_sample(self, payload: PayloadSample) -> None: + """Update sample data and notify listeners.""" + self.data.sample = payload + self.async_set_updated_data(self.data) + + @callback + def _handle_diagnostic(self, payload: PayloadDiagnostic) -> None: + """Update diagnostic data and notify listeners.""" + self.data.diagnostic = payload + diagnostic = payload.diagnostic + self.data.boot_time = self._stable_boot_time(diagnostic.uptime_s) + + if diagnostic.HasField("device_info"): + device_reg = dr.async_get(self.hass) + if device := device_reg.async_get_device_by_identifier( + (DOMAIN, self.mac_address), self.config_entry.entry_id + ): + device_info = diagnostic.device_info + model = device_info.model_name or MODEL_NAME + sw_version = device_info.sw_version or None + if device.model != model or device.sw_version != sw_version: + device_reg.async_update_device( + device.id, + model=model, + sw_version=sw_version, + ) + + self.async_set_updated_data(self.data) + + @override + async def _async_update_data(self) -> BitvisData: + """Return current data (updates are push-based via UDP datagrams).""" + return self.data diff --git a/homeassistant/components/bitvis/manifest.json b/homeassistant/components/bitvis/manifest.json new file mode 100644 index 000000000000..59012b8b318b --- /dev/null +++ b/homeassistant/components/bitvis/manifest.json @@ -0,0 +1,12 @@ +{ + "domain": "bitvis", + "name": "Bitvis Power Hub", + "codeowners": ["@MandusBorjesson", "@real-tintin", "@simontegelid"], + "config_flow": true, + "documentation": "https://www.home-assistant.io/integrations/bitvis", + "integration_type": "device", + "iot_class": "local_push", + "quality_scale": "bronze", + "requirements": ["bitvis-protobuf==2.0.4"], + "zeroconf": ["_powerhub._udp.local."] +} diff --git a/homeassistant/components/bitvis/quality_scale.yaml b/homeassistant/components/bitvis/quality_scale.yaml new file mode 100644 index 000000000000..6e28d51252ec --- /dev/null +++ b/homeassistant/components/bitvis/quality_scale.yaml @@ -0,0 +1,82 @@ +rules: + # Bronze + action-setup: + status: exempt + comment: "The integration does not have any actions." + appropriate-polling: + status: exempt + comment: "The integration does not poll." + brands: done + common-modules: done + config-flow-test-coverage: done + config-flow: done + dependency-transparency: done + docs-actions: + status: exempt + comment: "The integration does not have actions." + docs-conditions: + status: exempt + comment: "The integration does not have conditions." + docs-high-level-description: done + docs-installation-instructions: done + docs-removal-instructions: done + docs-triggers: + status: exempt + comment: "The integration does not have triggers." + entity-event-setup: done + entity-unique-id: done + has-entity-name: done + runtime-data: done + test-before-configure: done + test-before-setup: done + unique-config-entry: done + + # Silver + action-exceptions: + status: exempt + comment: "The integration does not have any service actions." + config-entry-unloading: done + docs-configuration-parameters: + status: exempt + comment: "The integration has no options flow." + docs-installation-parameters: done + entity-unavailable: todo + integration-owner: done + log-when-unavailable: todo + parallel-updates: done + reauthentication-flow: + status: exempt + comment: "The integration does not use any credentials. It connects to a local device using only host and port via unauthenticated UDP push." + test-coverage: done + + # Gold + devices: done + diagnostics: todo + discovery-update-info: done + discovery: done + docs-data-update: todo + docs-examples: todo + docs-known-limitations: todo + docs-supported-devices: todo + docs-supported-functions: todo + docs-troubleshooting: todo + docs-use-cases: todo + dynamic-devices: + status: exempt + comment: "The integration connects to a single device." + entity-category: done + entity-device-class: done + entity-disabled-by-default: done + entity-translations: done + exception-translations: todo + icon-translations: todo + reconfiguration-flow: todo + repair-issues: todo + stale-devices: todo + + # Platinum + async-dependency: done + inject-websession: + status: exempt + comment: "The integration does not use HTTP." + strict-typing: done diff --git a/homeassistant/components/bitvis/sensor.py b/homeassistant/components/bitvis/sensor.py new file mode 100644 index 000000000000..299544885ce7 --- /dev/null +++ b/homeassistant/components/bitvis/sensor.py @@ -0,0 +1,501 @@ +"""Sensor platform for Bitvis Power Hub.""" + +from collections.abc import Callable +from dataclasses import dataclass +from datetime import datetime +from typing import TYPE_CHECKING, cast, override + +from bitvis_protobuf.han_port_pb2 import HanPortSample +from bitvis_protobuf.powerhub_pb2 import Diagnostic + +from homeassistant.components.sensor import ( + SensorDeviceClass, + SensorEntity, + SensorEntityDescription, + SensorStateClass, +) +from homeassistant.const import ( + SIGNAL_STRENGTH_DECIBELS_MILLIWATT, + EntityCategory, + UnitOfElectricCurrent, + UnitOfElectricPotential, + UnitOfEnergy, + UnitOfPower, + UnitOfReactiveEnergy, + UnitOfReactivePower, +) +from homeassistant.core import HomeAssistant, callback +from homeassistant.helpers.device_registry import CONNECTION_NETWORK_MAC, DeviceInfo +from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback +from homeassistant.helpers.update_coordinator import CoordinatorEntity + +from . import BitvisConfigEntry +from .const import DOMAIN, MANUFACTURER +from .coordinator import BitvisDataUpdateCoordinator + +PARALLEL_UPDATES = 0 + + +def _optional(field: str) -> Callable[[HanPortSample], float | None]: + """Return a getter that yields None when a protobuf field is unset.""" + + def _get(data: HanPortSample) -> float | None: + if data.HasField(field): + return cast(float, getattr(data, field)) + return None + + return _get + + +@dataclass(frozen=True, kw_only=True) +class BitvisSensorEntityDescription(SensorEntityDescription): + """Describes Bitvis sensor entity.""" + + value_fn: Callable[[HanPortSample], float | None] + + +@dataclass(frozen=True, kw_only=True) +class BitvisDiagnosticSensorEntityDescription(SensorEntityDescription): + """Describes Bitvis diagnostic sensor entity.""" + + value_fn: Callable[[Diagnostic], float | int | str | datetime | None] + + +SENSOR_DESCRIPTIONS: tuple[BitvisSensorEntityDescription, ...] = ( + # Phase voltages + BitvisSensorEntityDescription( + key="phase_voltage_l1", + translation_key="phase_voltage", + translation_placeholders={"phase": "L1"}, + device_class=SensorDeviceClass.VOLTAGE, + native_unit_of_measurement=UnitOfElectricPotential.VOLT, + state_class=SensorStateClass.MEASUREMENT, + suggested_display_precision=1, + entity_registry_enabled_default=False, + value_fn=_optional("phase_voltage_l1_v"), + ), + BitvisSensorEntityDescription( + key="phase_voltage_l2", + translation_key="phase_voltage", + translation_placeholders={"phase": "L2"}, + device_class=SensorDeviceClass.VOLTAGE, + native_unit_of_measurement=UnitOfElectricPotential.VOLT, + state_class=SensorStateClass.MEASUREMENT, + suggested_display_precision=1, + entity_registry_enabled_default=False, + value_fn=_optional("phase_voltage_l2_v"), + ), + BitvisSensorEntityDescription( + key="phase_voltage_l3", + translation_key="phase_voltage", + translation_placeholders={"phase": "L3"}, + device_class=SensorDeviceClass.VOLTAGE, + native_unit_of_measurement=UnitOfElectricPotential.VOLT, + state_class=SensorStateClass.MEASUREMENT, + suggested_display_precision=1, + entity_registry_enabled_default=False, + value_fn=_optional("phase_voltage_l3_v"), + ), + # Phase currents + BitvisSensorEntityDescription( + key="phase_current_l1", + translation_key="phase_current", + translation_placeholders={"phase": "L1"}, + device_class=SensorDeviceClass.CURRENT, + native_unit_of_measurement=UnitOfElectricCurrent.AMPERE, + state_class=SensorStateClass.MEASUREMENT, + suggested_display_precision=2, + value_fn=_optional("phase_current_l1_a"), + ), + BitvisSensorEntityDescription( + key="phase_current_l2", + translation_key="phase_current", + translation_placeholders={"phase": "L2"}, + device_class=SensorDeviceClass.CURRENT, + native_unit_of_measurement=UnitOfElectricCurrent.AMPERE, + state_class=SensorStateClass.MEASUREMENT, + suggested_display_precision=2, + value_fn=_optional("phase_current_l2_a"), + ), + BitvisSensorEntityDescription( + key="phase_current_l3", + translation_key="phase_current", + translation_placeholders={"phase": "L3"}, + device_class=SensorDeviceClass.CURRENT, + native_unit_of_measurement=UnitOfElectricCurrent.AMPERE, + state_class=SensorStateClass.MEASUREMENT, + suggested_display_precision=2, + value_fn=_optional("phase_current_l3_a"), + ), + # Total active power + BitvisSensorEntityDescription( + key="power_active_delivered_to_client", + translation_key="power_active_import", + device_class=SensorDeviceClass.POWER, + native_unit_of_measurement=UnitOfPower.KILO_WATT, + state_class=SensorStateClass.MEASUREMENT, + suggested_display_precision=3, + value_fn=_optional("power_active_delivered_to_client_kw"), + ), + BitvisSensorEntityDescription( + key="power_active_delivered_by_client", + translation_key="power_active_export", + device_class=SensorDeviceClass.POWER, + native_unit_of_measurement=UnitOfPower.KILO_WATT, + state_class=SensorStateClass.MEASUREMENT, + suggested_display_precision=3, + value_fn=_optional("power_active_delivered_by_client_kw"), + ), + # Total reactive power + BitvisSensorEntityDescription( + key="power_reactive_delivered_to_client", + translation_key="power_reactive_import", + device_class=SensorDeviceClass.REACTIVE_POWER, + native_unit_of_measurement=UnitOfReactivePower.KILO_VOLT_AMPERE_REACTIVE, + state_class=SensorStateClass.MEASUREMENT, + suggested_display_precision=3, + entity_registry_enabled_default=False, + value_fn=_optional("power_reactive_delivered_to_client_kvar"), + ), + BitvisSensorEntityDescription( + key="power_reactive_delivered_by_client", + translation_key="power_reactive_export", + device_class=SensorDeviceClass.REACTIVE_POWER, + native_unit_of_measurement=UnitOfReactivePower.KILO_VOLT_AMPERE_REACTIVE, + state_class=SensorStateClass.MEASUREMENT, + suggested_display_precision=3, + entity_registry_enabled_default=False, + value_fn=_optional("power_reactive_delivered_by_client_kvar"), + ), + # Per-phase active power (to client) + BitvisSensorEntityDescription( + key="power_active_l1_delivered_to_client", + translation_key="power_active_phase_import", + translation_placeholders={"phase": "L1"}, + device_class=SensorDeviceClass.POWER, + native_unit_of_measurement=UnitOfPower.KILO_WATT, + state_class=SensorStateClass.MEASUREMENT, + suggested_display_precision=3, + entity_registry_enabled_default=False, + value_fn=_optional("power_active_l1_delivered_to_client_kw"), + ), + BitvisSensorEntityDescription( + key="power_active_l2_delivered_to_client", + translation_key="power_active_phase_import", + translation_placeholders={"phase": "L2"}, + device_class=SensorDeviceClass.POWER, + native_unit_of_measurement=UnitOfPower.KILO_WATT, + state_class=SensorStateClass.MEASUREMENT, + suggested_display_precision=3, + entity_registry_enabled_default=False, + value_fn=_optional("power_active_l2_delivered_to_client_kw"), + ), + BitvisSensorEntityDescription( + key="power_active_l3_delivered_to_client", + translation_key="power_active_phase_import", + translation_placeholders={"phase": "L3"}, + device_class=SensorDeviceClass.POWER, + native_unit_of_measurement=UnitOfPower.KILO_WATT, + state_class=SensorStateClass.MEASUREMENT, + suggested_display_precision=3, + entity_registry_enabled_default=False, + value_fn=_optional("power_active_l3_delivered_to_client_kw"), + ), + # Per-phase active power (by client) + BitvisSensorEntityDescription( + key="power_active_l1_delivered_by_client", + translation_key="power_active_phase_export", + translation_placeholders={"phase": "L1"}, + device_class=SensorDeviceClass.POWER, + native_unit_of_measurement=UnitOfPower.KILO_WATT, + state_class=SensorStateClass.MEASUREMENT, + suggested_display_precision=3, + entity_registry_enabled_default=False, + value_fn=_optional("power_active_l1_delivered_by_client_kw"), + ), + BitvisSensorEntityDescription( + key="power_active_l2_delivered_by_client", + translation_key="power_active_phase_export", + translation_placeholders={"phase": "L2"}, + device_class=SensorDeviceClass.POWER, + native_unit_of_measurement=UnitOfPower.KILO_WATT, + state_class=SensorStateClass.MEASUREMENT, + suggested_display_precision=3, + entity_registry_enabled_default=False, + value_fn=_optional("power_active_l2_delivered_by_client_kw"), + ), + BitvisSensorEntityDescription( + key="power_active_l3_delivered_by_client", + translation_key="power_active_phase_export", + translation_placeholders={"phase": "L3"}, + device_class=SensorDeviceClass.POWER, + native_unit_of_measurement=UnitOfPower.KILO_WATT, + state_class=SensorStateClass.MEASUREMENT, + suggested_display_precision=3, + entity_registry_enabled_default=False, + value_fn=_optional("power_active_l3_delivered_by_client_kw"), + ), + # Per-phase reactive power (to client) + BitvisSensorEntityDescription( + key="power_reactive_l1_delivered_to_client", + translation_key="power_reactive_phase_import", + translation_placeholders={"phase": "L1"}, + device_class=SensorDeviceClass.REACTIVE_POWER, + native_unit_of_measurement=UnitOfReactivePower.KILO_VOLT_AMPERE_REACTIVE, + state_class=SensorStateClass.MEASUREMENT, + suggested_display_precision=3, + entity_registry_enabled_default=False, + value_fn=_optional("power_reactive_l1_delivered_to_client_kvar"), + ), + BitvisSensorEntityDescription( + key="power_reactive_l2_delivered_to_client", + translation_key="power_reactive_phase_import", + translation_placeholders={"phase": "L2"}, + device_class=SensorDeviceClass.REACTIVE_POWER, + native_unit_of_measurement=UnitOfReactivePower.KILO_VOLT_AMPERE_REACTIVE, + state_class=SensorStateClass.MEASUREMENT, + suggested_display_precision=3, + entity_registry_enabled_default=False, + value_fn=_optional("power_reactive_l2_delivered_to_client_kvar"), + ), + BitvisSensorEntityDescription( + key="power_reactive_l3_delivered_to_client", + translation_key="power_reactive_phase_import", + translation_placeholders={"phase": "L3"}, + device_class=SensorDeviceClass.REACTIVE_POWER, + native_unit_of_measurement=UnitOfReactivePower.KILO_VOLT_AMPERE_REACTIVE, + state_class=SensorStateClass.MEASUREMENT, + suggested_display_precision=3, + entity_registry_enabled_default=False, + value_fn=_optional("power_reactive_l3_delivered_to_client_kvar"), + ), + # Per-phase reactive power (by client) + BitvisSensorEntityDescription( + key="power_reactive_l1_delivered_by_client", + translation_key="power_reactive_phase_export", + translation_placeholders={"phase": "L1"}, + device_class=SensorDeviceClass.REACTIVE_POWER, + native_unit_of_measurement=UnitOfReactivePower.KILO_VOLT_AMPERE_REACTIVE, + state_class=SensorStateClass.MEASUREMENT, + suggested_display_precision=3, + entity_registry_enabled_default=False, + value_fn=_optional("power_reactive_l1_delivered_by_client_kvar"), + ), + BitvisSensorEntityDescription( + key="power_reactive_l2_delivered_by_client", + translation_key="power_reactive_phase_export", + translation_placeholders={"phase": "L2"}, + device_class=SensorDeviceClass.REACTIVE_POWER, + native_unit_of_measurement=UnitOfReactivePower.KILO_VOLT_AMPERE_REACTIVE, + state_class=SensorStateClass.MEASUREMENT, + suggested_display_precision=3, + entity_registry_enabled_default=False, + value_fn=_optional("power_reactive_l2_delivered_by_client_kvar"), + ), + BitvisSensorEntityDescription( + key="power_reactive_l3_delivered_by_client", + translation_key="power_reactive_phase_export", + translation_placeholders={"phase": "L3"}, + device_class=SensorDeviceClass.REACTIVE_POWER, + native_unit_of_measurement=UnitOfReactivePower.KILO_VOLT_AMPERE_REACTIVE, + state_class=SensorStateClass.MEASUREMENT, + suggested_display_precision=3, + entity_registry_enabled_default=False, + value_fn=_optional("power_reactive_l3_delivered_by_client_kvar"), + ), + # Energy - active + BitvisSensorEntityDescription( + key="energy_active_delivered_to_client", + translation_key="energy_active_import", + device_class=SensorDeviceClass.ENERGY, + native_unit_of_measurement=UnitOfEnergy.KILO_WATT_HOUR, + state_class=SensorStateClass.TOTAL_INCREASING, + suggested_display_precision=2, + value_fn=_optional("energy_active_delivered_to_client_kwh"), + ), + BitvisSensorEntityDescription( + key="energy_active_delivered_by_client", + translation_key="energy_active_export", + device_class=SensorDeviceClass.ENERGY, + native_unit_of_measurement=UnitOfEnergy.KILO_WATT_HOUR, + state_class=SensorStateClass.TOTAL_INCREASING, + suggested_display_precision=2, + value_fn=_optional("energy_active_delivered_by_client_kwh"), + ), + # Energy - reactive + BitvisSensorEntityDescription( + key="energy_reactive_delivered_to_client", + translation_key="energy_reactive_import", + device_class=SensorDeviceClass.REACTIVE_ENERGY, + native_unit_of_measurement=UnitOfReactiveEnergy.KILO_VOLT_AMPERE_REACTIVE_HOUR, + state_class=SensorStateClass.TOTAL_INCREASING, + suggested_display_precision=2, + entity_registry_enabled_default=False, + value_fn=_optional("energy_reactive_delivered_to_client_kvarh"), + ), + BitvisSensorEntityDescription( + key="energy_reactive_delivered_by_client", + translation_key="energy_reactive_export", + device_class=SensorDeviceClass.REACTIVE_ENERGY, + native_unit_of_measurement=UnitOfReactiveEnergy.KILO_VOLT_AMPERE_REACTIVE_HOUR, + state_class=SensorStateClass.TOTAL_INCREASING, + suggested_display_precision=2, + entity_registry_enabled_default=False, + value_fn=_optional("energy_reactive_delivered_by_client_kvarh"), + ), +) + +UPTIME_DESCRIPTION = SensorEntityDescription( + key="uptime", + device_class=SensorDeviceClass.UPTIME, + entity_category=EntityCategory.DIAGNOSTIC, + entity_registry_enabled_default=False, +) + +DIAGNOSTIC_SENSOR_DESCRIPTIONS: tuple[BitvisDiagnosticSensorEntityDescription, ...] = ( + BitvisDiagnosticSensorEntityDescription( + key="wifi_rssi", + translation_key="wifi_rssi", + device_class=SensorDeviceClass.SIGNAL_STRENGTH, + native_unit_of_measurement=SIGNAL_STRENGTH_DECIBELS_MILLIWATT, + state_class=SensorStateClass.MEASUREMENT, + entity_category=EntityCategory.DIAGNOSTIC, + entity_registry_enabled_default=False, + value_fn=lambda data: data.wifi_rssi_dbm, + ), + BitvisDiagnosticSensorEntityDescription( + key="han_msg_successfully_parsed", + translation_key="han_msg_successfully_parsed", + state_class=SensorStateClass.TOTAL_INCREASING, + entity_category=EntityCategory.DIAGNOSTIC, + entity_registry_enabled_default=False, + value_fn=lambda data: data.han_msg_successfully_parsed, + ), + BitvisDiagnosticSensorEntityDescription( + key="han_msg_buffer_overflow", + translation_key="han_msg_buffer_overflow", + state_class=SensorStateClass.TOTAL_INCREASING, + entity_category=EntityCategory.DIAGNOSTIC, + entity_registry_enabled_default=False, + value_fn=lambda data: data.han_msg_buffer_overflow, + ), +) + + +async def async_setup_entry( + hass: HomeAssistant, + entry: BitvisConfigEntry, + async_add_entities: AddConfigEntryEntitiesCallback, +) -> None: + """Set up Bitvis sensor platform.""" + coordinator = entry.runtime_data + known_keys: set[str] = set() + + async_add_entities( + [ + BitvisUptimeSensorEntity(coordinator, UPTIME_DESCRIPTION), + *( + BitvisDiagnosticSensorEntity(coordinator, description) + for description in DIAGNOSTIC_SENSOR_DESCRIPTIONS + ), + ] + ) + + @callback + def _check_entities() -> None: + if (payload := coordinator.data.sample) is None: + return + entities = [ + BitvisSensorEntity(coordinator, description) + for description in SENSOR_DESCRIPTIONS + if description.key not in known_keys + and description.value_fn(payload.sample) is not None + ] + if entities: + known_keys.update(entity.entity_description.key for entity in entities) + async_add_entities(entities) + + _check_entities() + entry.async_on_unload(coordinator.async_add_listener(_check_entities)) + + +class BitvisBaseSensorEntity( + CoordinatorEntity[BitvisDataUpdateCoordinator], SensorEntity +): + """Base class for Bitvis sensor entities.""" + + _attr_has_entity_name = True + + def __init__( + self, + coordinator: BitvisDataUpdateCoordinator, + description: SensorEntityDescription, + ) -> None: + """Initialize the sensor.""" + super().__init__(coordinator) + self.entity_description = description + mac_address = coordinator.mac_address + self._attr_unique_id = f"{mac_address}_{description.key}" + self._attr_device_info = DeviceInfo( + identifiers={(DOMAIN, mac_address)}, + connections={(CONNECTION_NETWORK_MAC, mac_address)}, + manufacturer=MANUFACTURER, + ) + + +class BitvisSensorEntity(BitvisBaseSensorEntity): + """Representation of a Bitvis sensor.""" + + entity_description: BitvisSensorEntityDescription + + @property + @override + def native_value(self) -> float | None: + """Return the state of the sensor.""" + payload = self.coordinator.data.sample + if TYPE_CHECKING: + assert payload is not None + return self.entity_description.value_fn(payload.sample) + + @property + @override + def available(self) -> bool: + """Return if entity is available.""" + return super().available and self.coordinator.data.sample is not None + + +class BitvisDiagnosticSensorEntity(BitvisBaseSensorEntity): + """Representation of a Bitvis diagnostic sensor.""" + + entity_description: BitvisDiagnosticSensorEntityDescription + + @property + @override + def native_value(self) -> float | int | str | datetime | None: + """Return the state of the sensor.""" + payload = self.coordinator.data.diagnostic + if TYPE_CHECKING: + assert payload is not None + return self.entity_description.value_fn(payload.diagnostic) + + @property + @override + def available(self) -> bool: + """Return if entity is available.""" + return super().available and self.coordinator.data.diagnostic is not None + + +class BitvisUptimeSensorEntity(BitvisBaseSensorEntity): + """Sensor entity for device uptime (boot time).""" + + @property + @override + def native_value(self) -> datetime | None: + """Return the stable boot time computed by the coordinator.""" + return self.coordinator.data.boot_time + + @property + @override + def available(self) -> bool: + """Return if entity is available.""" + return super().available and self.coordinator.data.boot_time is not None diff --git a/homeassistant/components/bitvis/strings.json b/homeassistant/components/bitvis/strings.json new file mode 100644 index 000000000000..f8f68fcb6e9a --- /dev/null +++ b/homeassistant/components/bitvis/strings.json @@ -0,0 +1,87 @@ +{ + "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%]", + "invalid_mac": "The device did not provide a valid MAC address. Update the Power Hub to the latest firmware and try again.", + "timeout_connect": "[%key:common::config_flow::error::timeout_connect%]" + }, + "error": { + "cannot_connect": "[%key:common::config_flow::error::cannot_connect%]", + "invalid_mac": "The device did not provide a valid MAC address. Update the Power Hub to the latest firmware and try again.", + "timeout_connect": "[%key:common::config_flow::error::timeout_connect%]" + }, + "step": { + "user": { + "data": { + "host": "[%key:common::config_flow::data::host%]" + }, + "data_description": { + "host": "Hostname or IP address of your Bitvis Power Hub device." + }, + "description": "Enter the hostname or IP address of your Bitvis Power Hub. Setup waits for a UDP packet from the device.", + "title": "Set up Bitvis Power Hub" + }, + "zeroconf_confirm": { + "description": "Do you want to add the discovered Bitvis Power Hub ({name}) at {host} to Home Assistant?", + "title": "Discovered Bitvis Power Hub" + } + } + }, + "entity": { + "sensor": { + "energy_active_export": { + "name": "Active energy export" + }, + "energy_active_import": { + "name": "Active energy import" + }, + "energy_reactive_export": { + "name": "Reactive energy export" + }, + "energy_reactive_import": { + "name": "Reactive energy import" + }, + "han_msg_buffer_overflow": { + "name": "HAN buffer overflows" + }, + "han_msg_successfully_parsed": { + "name": "HAN messages successfully parsed" + }, + "phase_current": { + "name": "Current {phase}" + }, + "phase_voltage": { + "name": "Voltage {phase}" + }, + "power_active_export": { + "name": "Active power export" + }, + "power_active_import": { + "name": "Active power import" + }, + "power_active_phase_export": { + "name": "Active power export {phase}" + }, + "power_active_phase_import": { + "name": "Active power import {phase}" + }, + "power_reactive_export": { + "name": "Reactive power export" + }, + "power_reactive_import": { + "name": "Reactive power import" + }, + "power_reactive_phase_export": { + "name": "Reactive power export {phase}" + }, + "power_reactive_phase_import": { + "name": "Reactive power import {phase}" + }, + "wifi_rssi": { + "name": "Wi-Fi signal strength" + } + } + } +} diff --git a/homeassistant/generated/config_flows.py b/homeassistant/generated/config_flows.py index dec26af9bfff..26f525c1a545 100644 --- a/homeassistant/generated/config_flows.py +++ b/homeassistant/generated/config_flows.py @@ -102,6 +102,7 @@ FLOWS = { "bang_olufsen", "bayesian", "besen", + "bitvis", "blebox", "blink", "blue_current", diff --git a/homeassistant/generated/integrations.json b/homeassistant/generated/integrations.json index 190556a56d3f..cf9ab63316b6 100644 --- a/homeassistant/generated/integrations.json +++ b/homeassistant/generated/integrations.json @@ -756,6 +756,12 @@ "config_flow": false, "iot_class": "cloud_polling" }, + "bitvis": { + "name": "Bitvis Power Hub", + "integration_type": "device", + "config_flow": true, + "iot_class": "local_push" + }, "bizkaibus": { "name": "Bizkaibus", "integration_type": "hub", diff --git a/homeassistant/generated/zeroconf.py b/homeassistant/generated/zeroconf.py index 7b08acf0ad40..2d013ca31d52 100644 --- a/homeassistant/generated/zeroconf.py +++ b/homeassistant/generated/zeroconf.py @@ -888,6 +888,11 @@ ZEROCONF = { "domain": "plugwise", }, ], + "_powerhub._udp.local.": [ + { + "domain": "bitvis", + }, + ], "_powerview._tcp.local.": [ { "domain": "hunterdouglas_powerview", diff --git a/mypy.ini b/mypy.ini index f4e044aa4925..998c3a6c5767 100644 --- a/mypy.ini +++ b/mypy.ini @@ -937,6 +937,16 @@ disallow_untyped_defs = true warn_return_any = true warn_unreachable = true +[mypy-homeassistant.components.bitvis.*] +check_untyped_defs = true +disallow_incomplete_defs = true +disallow_subclassing_any = true +disallow_untyped_calls = true +disallow_untyped_decorators = true +disallow_untyped_defs = true +warn_return_any = true +warn_unreachable = true + [mypy-homeassistant.components.blockchain.*] check_untyped_defs = true disallow_incomplete_defs = true diff --git a/requirements_all.txt b/requirements_all.txt index 1b1321161254..71cb729f8650 100644 --- a/requirements_all.txt +++ b/requirements_all.txt @@ -669,6 +669,9 @@ beautifulsoup4==4.13.3 # homeassistant.components.besen besen==0.4.0 +# homeassistant.components.bitvis +bitvis-protobuf==2.0.4 + # homeassistant.components.bizkaibus bizkaibus==0.1.1 diff --git a/tests/components/bitvis/__init__.py b/tests/components/bitvis/__init__.py new file mode 100644 index 000000000000..0f84a7803690 --- /dev/null +++ b/tests/components/bitvis/__init__.py @@ -0,0 +1,35 @@ +"""Tests for the Bitvis Power Hub integration.""" + +from collections.abc import Callable + +from bitvis_protobuf.listener import FilterMac +from bitvis_protobuf.parse import PayloadDiagnostic, PayloadSample +import pytest + +from homeassistant.core import HomeAssistant + +from tests.common import MockConfigEntry + + +async def setup_integration(hass: HomeAssistant, config_entry: MockConfigEntry) -> None: + """Set up the integration.""" + config_entry.add_to_hass(hass) + + await hass.config_entries.async_setup(config_entry.entry_id) + await hass.async_block_till_done() + + +def find_listener_callback( + listener: object, + mac_address: str, +) -> Callable[[PayloadSample | PayloadDiagnostic, tuple[str, int]], None]: + """Find the listener callback registered for a MAC address.""" + register = listener.register + for call in register.call_args_list: + filt = call[0][0] + if ( + isinstance(filt, FilterMac) + and filt.mac_address.lower() == mac_address.lower() + ): + return call[0][1] + pytest.fail(f"Callback for MAC {mac_address} not found") diff --git a/tests/components/bitvis/conftest.py b/tests/components/bitvis/conftest.py new file mode 100644 index 000000000000..f62bcab053f9 --- /dev/null +++ b/tests/components/bitvis/conftest.py @@ -0,0 +1,233 @@ +"""Common fixtures for Bitvis Power Hub tests.""" + +from collections.abc import Callable, Generator, Iterator +from contextlib import ExitStack, contextmanager +from unittest.mock import AsyncMock, MagicMock, patch + +from bitvis_protobuf.listener import Filter, FilterIp +from bitvis_protobuf.parse import PayloadDiagnostic, PayloadSample, parse_payload +from bitvis_protobuf.powerhub_pb2 import Payload +from bitvis_protobuf.utils import InvalidMacAddressError +import pytest + +from homeassistant.components.bitvis.const import DEFAULT_NAME, DEFAULT_PORT, DOMAIN +from homeassistant.const import CONF_HOST, CONF_PORT +from homeassistant.core import HomeAssistant + +from . import setup_integration + +from tests.common import MockConfigEntry + +TEST_DEVICE_MAC = "aa:bb:cc:dd:ee:ff" +SECOND_DEVICE_MAC = "11:22:33:44:55:66" + +type ListenerCallback = Callable[ + [PayloadSample | PayloadDiagnostic, tuple[str, int]], None +] +type ErrorCallback = Callable[[Exception, tuple[str, int]], None] + + +class FakeListener: + """In-memory SharedListener stand-in for config-flow and coordinator tests.""" + + def __init__(self) -> None: + """Initialize callback storage and async start/stop mocks.""" + self._callbacks: dict[Filter, ListenerCallback] = {} + self._error_callbacks: list[ErrorCallback] = [] + self.start = AsyncMock() + self.stop = AsyncMock() + self.register = MagicMock(side_effect=self._register) + self.unregister = MagicMock(side_effect=self._unregister) + self.register_error_callback = MagicMock( + side_effect=self._error_callbacks.append + ) + self.unregister_error_callback = MagicMock(side_effect=self._unregister_error) + self.dispatch = MagicMock(side_effect=self._dispatch) + + @property + def is_empty(self) -> bool: + """Return True when no payload callbacks are registered.""" + return not self._callbacks + + def _register(self, filt: Filter, callback: ListenerCallback) -> None: + if filt in self._callbacks: + raise RuntimeError(f"Filter already registered: {filt}") + self._callbacks[filt] = callback + + def _unregister(self, filt: Filter) -> None: + self._callbacks.pop(filt, None) + + def _unregister_error(self, callback: ErrorCallback) -> None: + if callback in self._error_callbacks: + self._error_callbacks.remove(callback) + + def _dispatch(self, data: bytes, addr: tuple[str, int]) -> None: + try: + payload = parse_payload(data) + except InvalidMacAddressError as err: + for callback in self._error_callbacks: + callback(err, addr) + return + if payload is None: + return + self.deliver(payload, addr) + + def deliver( + self, + payload: PayloadSample | PayloadDiagnostic, + addr: tuple[str, int], + ) -> None: + """Deliver an already-parsed payload to matching callbacks.""" + host = addr[0] + for filt, callback in self._callbacks.items(): + if filt.match(payload, host): + callback(payload, addr) + + +@contextmanager +def patch_config_flow_connectivity( + resolved_host: str, + *, + mac_address: str = TEST_DEVICE_MAC, + deliver_mac: bool = True, + invalid_mac: bool = False, + port_bind_side_effect: BaseException | None = None, + discovery_timeout: bool = False, + register_side_effect: BaseException | None = None, + shared_listener: FakeListener | None = None, +) -> Iterator[AsyncMock]: + """Patch library connectivity helpers used by the config flow.""" + listener = shared_listener or FakeListener() + + if register_side_effect is not None: + listener.register.side_effect = register_side_effect + else: + original_register = listener.register.side_effect + + def _on_register(filt: Filter, callback: ListenerCallback) -> None: + original_register(filt, callback) + if not isinstance(filt, FilterIp): + return + if invalid_mac: + payload = Payload() + payload.sample.SetInParent() + listener.dispatch(payload.SerializeToString(), (resolved_host, 1234)) + elif deliver_mac and not discovery_timeout: + callback( + PayloadSample(mac_address=mac_address, sample=MagicMock()), + (resolved_host, 1234), + ) + + listener.register.side_effect = _on_register + + with ExitStack() as stack: + mock_verify = stack.enter_context( + patch( + "homeassistant.components.bitvis.config_flow.async_verify_udp_port_bindable", + new_callable=AsyncMock, + side_effect=port_bind_side_effect, + ) + ) + stack.enter_context( + patch( + "homeassistant.components.bitvis.config_flow.async_resolve_host", + new_callable=AsyncMock, + return_value={resolved_host}, + ) + ) + stack.enter_context( + patch( + "homeassistant.components.bitvis.coordinator.SharedListener", + return_value=listener, + ) + ) + if discovery_timeout: + stack.enter_context( + patch( + "homeassistant.components.bitvis.config_flow.DISCOVERY_TIMEOUT", + 0, + ) + ) + yield mock_verify + + +@pytest.fixture +def mock_config_entry() -> MockConfigEntry: + """Return the default mocked config entry.""" + return MockConfigEntry( + domain=DOMAIN, + data={CONF_HOST: "192.168.1.100", CONF_PORT: DEFAULT_PORT}, + unique_id=TEST_DEVICE_MAC, + title=DEFAULT_NAME, + ) + + +@pytest.fixture +def mock_zeroconf_config_entry() -> MockConfigEntry: + """Return a mocked config entry for zeroconf discovery host.""" + return MockConfigEntry( + domain=DOMAIN, + data={CONF_HOST: "192.168.1.200", CONF_PORT: DEFAULT_PORT}, + unique_id=TEST_DEVICE_MAC, + title=DEFAULT_NAME, + ) + + +@pytest.fixture +def mock_ipv6_config_entry() -> MockConfigEntry: + """Return a mocked config entry with an IPv6 host.""" + return MockConfigEntry( + domain=DOMAIN, + data={CONF_HOST: "2001:db8::10", CONF_PORT: DEFAULT_PORT}, + unique_id=SECOND_DEVICE_MAC, + title=DEFAULT_NAME, + ) + + +@pytest.fixture +def mock_second_config_entry() -> MockConfigEntry: + """Return a second mocked config entry on the same UDP port.""" + return MockConfigEntry( + domain=DOMAIN, + data={CONF_HOST: "192.168.1.101", CONF_PORT: DEFAULT_PORT}, + unique_id=SECOND_DEVICE_MAC, + title=DEFAULT_NAME, + ) + + +@pytest.fixture +def mock_shared_listener() -> FakeListener: + """Return a fake bitvis_protobuf SharedListener.""" + return FakeListener() + + +@pytest.fixture +def patch_shared_listener( + mock_shared_listener: FakeListener, +) -> Generator[FakeListener]: + """Patch SharedListener to return a mocked instance.""" + with patch( + "homeassistant.components.bitvis.coordinator.SharedListener", + return_value=mock_shared_listener, + ): + yield mock_shared_listener + + +@pytest.fixture +def mock_setup_entry() -> Generator[AsyncMock]: + """Override async_setup_entry.""" + with patch( + "homeassistant.components.bitvis.async_setup_entry", return_value=True + ) as mock_setup_entry: + yield mock_setup_entry + + +@pytest.fixture +async def init_integration( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + patch_shared_listener: FakeListener, +) -> MockConfigEntry: + """Set up the integration with a mocked UDP listener.""" + await setup_integration(hass, mock_config_entry) + return mock_config_entry diff --git a/tests/components/bitvis/snapshots/test_sensor.ambr b/tests/components/bitvis/snapshots/test_sensor.ambr new file mode 100644 index 000000000000..27e9306840f0 --- /dev/null +++ b/tests/components/bitvis/snapshots/test_sensor.ambr @@ -0,0 +1,1721 @@ +# serializer version: 1 +# name: test_all_entities[sensor.bitvis_power_hub_active_energy_export-entry] + EntityRegistryEntrySnapshot({ + 'aliases': list([ + None, + ]), + 'area_id': None, + 'capabilities': dict({ + : , + }), + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'sensor', + 'entity_category': None, + 'entity_id': 'sensor.bitvis_power_hub_active_energy_export', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'object_id_base': 'Active energy export', + 'options': dict({ + 'sensor': dict({ + 'suggested_display_precision': 2, + }), + }), + 'original_device_class': , + 'original_icon': None, + 'original_name': 'Active energy export', + 'platform': 'bitvis', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': 0, + 'translation_key': 'energy_active_export', + 'unique_id': 'aa:bb:cc:dd:ee:ff_energy_active_delivered_by_client', + 'unit_of_measurement': , + }) +# --- +# name: test_all_entities[sensor.bitvis_power_hub_active_energy_export-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + : 'energy', + : 'Bitvis Power Hub Active energy export', + : , + : , + }), + 'context': , + 'entity_id': 'sensor.bitvis_power_hub_active_energy_export', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': '789.119995117188', + }) +# --- +# name: test_all_entities[sensor.bitvis_power_hub_active_energy_import-entry] + EntityRegistryEntrySnapshot({ + 'aliases': list([ + None, + ]), + 'area_id': None, + 'capabilities': dict({ + : , + }), + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'sensor', + 'entity_category': None, + 'entity_id': 'sensor.bitvis_power_hub_active_energy_import', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'object_id_base': 'Active energy import', + 'options': dict({ + 'sensor': dict({ + 'suggested_display_precision': 2, + }), + }), + 'original_device_class': , + 'original_icon': None, + 'original_name': 'Active energy import', + 'platform': 'bitvis', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': 0, + 'translation_key': 'energy_active_import', + 'unique_id': 'aa:bb:cc:dd:ee:ff_energy_active_delivered_to_client', + 'unit_of_measurement': , + }) +# --- +# name: test_all_entities[sensor.bitvis_power_hub_active_energy_import-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + : 'energy', + : 'Bitvis Power Hub Active energy import', + : , + : , + }), + 'context': , + 'entity_id': 'sensor.bitvis_power_hub_active_energy_import', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': '1234.56005859375', + }) +# --- +# name: test_all_entities[sensor.bitvis_power_hub_active_power_export-entry] + EntityRegistryEntrySnapshot({ + 'aliases': list([ + None, + ]), + 'area_id': None, + 'capabilities': dict({ + : , + }), + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'sensor', + 'entity_category': None, + 'entity_id': 'sensor.bitvis_power_hub_active_power_export', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'object_id_base': 'Active power export', + 'options': dict({ + 'sensor': dict({ + 'suggested_display_precision': 3, + }), + }), + 'original_device_class': , + 'original_icon': None, + 'original_name': 'Active power export', + 'platform': 'bitvis', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': 0, + 'translation_key': 'power_active_export', + 'unique_id': 'aa:bb:cc:dd:ee:ff_power_active_delivered_by_client', + 'unit_of_measurement': , + }) +# --- +# name: test_all_entities[sensor.bitvis_power_hub_active_power_export-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + : 'power', + : 'Bitvis Power Hub Active power export', + : , + : , + }), + 'context': , + 'entity_id': 'sensor.bitvis_power_hub_active_power_export', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': '0.0', + }) +# --- +# name: test_all_entities[sensor.bitvis_power_hub_active_power_export_l1-entry] + EntityRegistryEntrySnapshot({ + 'aliases': list([ + None, + ]), + 'area_id': None, + 'capabilities': dict({ + : , + }), + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'sensor', + 'entity_category': None, + 'entity_id': 'sensor.bitvis_power_hub_active_power_export_l1', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'object_id_base': 'Active power export L1', + 'options': dict({ + 'sensor': dict({ + 'suggested_display_precision': 3, + }), + }), + 'original_device_class': , + 'original_icon': None, + 'original_name': 'Active power export L1', + 'platform': 'bitvis', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': 0, + 'translation_key': 'power_active_phase_export', + 'unique_id': 'aa:bb:cc:dd:ee:ff_power_active_l1_delivered_by_client', + 'unit_of_measurement': , + }) +# --- +# name: test_all_entities[sensor.bitvis_power_hub_active_power_export_l1-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + : 'power', + : 'Bitvis Power Hub Active power export L1', + : , + : , + }), + 'context': , + 'entity_id': 'sensor.bitvis_power_hub_active_power_export_l1', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': '0.0', + }) +# --- +# name: test_all_entities[sensor.bitvis_power_hub_active_power_export_l2-entry] + EntityRegistryEntrySnapshot({ + 'aliases': list([ + None, + ]), + 'area_id': None, + 'capabilities': dict({ + : , + }), + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'sensor', + 'entity_category': None, + 'entity_id': 'sensor.bitvis_power_hub_active_power_export_l2', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'object_id_base': 'Active power export L2', + 'options': dict({ + 'sensor': dict({ + 'suggested_display_precision': 3, + }), + }), + 'original_device_class': , + 'original_icon': None, + 'original_name': 'Active power export L2', + 'platform': 'bitvis', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': 0, + 'translation_key': 'power_active_phase_export', + 'unique_id': 'aa:bb:cc:dd:ee:ff_power_active_l2_delivered_by_client', + 'unit_of_measurement': , + }) +# --- +# name: test_all_entities[sensor.bitvis_power_hub_active_power_export_l2-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + : 'power', + : 'Bitvis Power Hub Active power export L2', + : , + : , + }), + 'context': , + 'entity_id': 'sensor.bitvis_power_hub_active_power_export_l2', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': '0.0', + }) +# --- +# name: test_all_entities[sensor.bitvis_power_hub_active_power_export_l3-entry] + EntityRegistryEntrySnapshot({ + 'aliases': list([ + None, + ]), + 'area_id': None, + 'capabilities': dict({ + : , + }), + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'sensor', + 'entity_category': None, + 'entity_id': 'sensor.bitvis_power_hub_active_power_export_l3', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'object_id_base': 'Active power export L3', + 'options': dict({ + 'sensor': dict({ + 'suggested_display_precision': 3, + }), + }), + 'original_device_class': , + 'original_icon': None, + 'original_name': 'Active power export L3', + 'platform': 'bitvis', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': 0, + 'translation_key': 'power_active_phase_export', + 'unique_id': 'aa:bb:cc:dd:ee:ff_power_active_l3_delivered_by_client', + 'unit_of_measurement': , + }) +# --- +# name: test_all_entities[sensor.bitvis_power_hub_active_power_export_l3-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + : 'power', + : 'Bitvis Power Hub Active power export L3', + : , + : , + }), + 'context': , + 'entity_id': 'sensor.bitvis_power_hub_active_power_export_l3', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': '0.0', + }) +# --- +# name: test_all_entities[sensor.bitvis_power_hub_active_power_import-entry] + EntityRegistryEntrySnapshot({ + 'aliases': list([ + None, + ]), + 'area_id': None, + 'capabilities': dict({ + : , + }), + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'sensor', + 'entity_category': None, + 'entity_id': 'sensor.bitvis_power_hub_active_power_import', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'object_id_base': 'Active power import', + 'options': dict({ + 'sensor': dict({ + 'suggested_display_precision': 3, + }), + }), + 'original_device_class': , + 'original_icon': None, + 'original_name': 'Active power import', + 'platform': 'bitvis', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': 0, + 'translation_key': 'power_active_import', + 'unique_id': 'aa:bb:cc:dd:ee:ff_power_active_delivered_to_client', + 'unit_of_measurement': , + }) +# --- +# name: test_all_entities[sensor.bitvis_power_hub_active_power_import-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + : 'power', + : 'Bitvis Power Hub Active power import', + : , + : , + }), + 'context': , + 'entity_id': 'sensor.bitvis_power_hub_active_power_import', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': '2.41499996185303', + }) +# --- +# name: test_all_entities[sensor.bitvis_power_hub_active_power_import_l1-entry] + EntityRegistryEntrySnapshot({ + 'aliases': list([ + None, + ]), + 'area_id': None, + 'capabilities': dict({ + : , + }), + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'sensor', + 'entity_category': None, + 'entity_id': 'sensor.bitvis_power_hub_active_power_import_l1', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'object_id_base': 'Active power import L1', + 'options': dict({ + 'sensor': dict({ + 'suggested_display_precision': 3, + }), + }), + 'original_device_class': , + 'original_icon': None, + 'original_name': 'Active power import L1', + 'platform': 'bitvis', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': 0, + 'translation_key': 'power_active_phase_import', + 'unique_id': 'aa:bb:cc:dd:ee:ff_power_active_l1_delivered_to_client', + 'unit_of_measurement': , + }) +# --- +# name: test_all_entities[sensor.bitvis_power_hub_active_power_import_l1-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + : 'power', + : 'Bitvis Power Hub Active power import L1', + : , + : , + }), + 'context': , + 'entity_id': 'sensor.bitvis_power_hub_active_power_import_l1', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': '0.800000011920929', + }) +# --- +# name: test_all_entities[sensor.bitvis_power_hub_active_power_import_l2-entry] + EntityRegistryEntrySnapshot({ + 'aliases': list([ + None, + ]), + 'area_id': None, + 'capabilities': dict({ + : , + }), + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'sensor', + 'entity_category': None, + 'entity_id': 'sensor.bitvis_power_hub_active_power_import_l2', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'object_id_base': 'Active power import L2', + 'options': dict({ + 'sensor': dict({ + 'suggested_display_precision': 3, + }), + }), + 'original_device_class': , + 'original_icon': None, + 'original_name': 'Active power import L2', + 'platform': 'bitvis', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': 0, + 'translation_key': 'power_active_phase_import', + 'unique_id': 'aa:bb:cc:dd:ee:ff_power_active_l2_delivered_to_client', + 'unit_of_measurement': , + }) +# --- +# name: test_all_entities[sensor.bitvis_power_hub_active_power_import_l2-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + : 'power', + : 'Bitvis Power Hub Active power import L2', + : , + : , + }), + 'context': , + 'entity_id': 'sensor.bitvis_power_hub_active_power_import_l2', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': '0.699999988079071', + }) +# --- +# name: test_all_entities[sensor.bitvis_power_hub_active_power_import_l3-entry] + EntityRegistryEntrySnapshot({ + 'aliases': list([ + None, + ]), + 'area_id': None, + 'capabilities': dict({ + : , + }), + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'sensor', + 'entity_category': None, + 'entity_id': 'sensor.bitvis_power_hub_active_power_import_l3', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'object_id_base': 'Active power import L3', + 'options': dict({ + 'sensor': dict({ + 'suggested_display_precision': 3, + }), + }), + 'original_device_class': , + 'original_icon': None, + 'original_name': 'Active power import L3', + 'platform': 'bitvis', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': 0, + 'translation_key': 'power_active_phase_import', + 'unique_id': 'aa:bb:cc:dd:ee:ff_power_active_l3_delivered_to_client', + 'unit_of_measurement': , + }) +# --- +# name: test_all_entities[sensor.bitvis_power_hub_active_power_import_l3-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + : 'power', + : 'Bitvis Power Hub Active power import L3', + : , + : , + }), + 'context': , + 'entity_id': 'sensor.bitvis_power_hub_active_power_import_l3', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': '0.915000021457672', + }) +# --- +# name: test_all_entities[sensor.bitvis_power_hub_current_l1-entry] + EntityRegistryEntrySnapshot({ + 'aliases': list([ + None, + ]), + 'area_id': None, + 'capabilities': dict({ + : , + }), + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'sensor', + 'entity_category': None, + 'entity_id': 'sensor.bitvis_power_hub_current_l1', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'object_id_base': 'Current L1', + 'options': dict({ + 'sensor': dict({ + 'suggested_display_precision': 2, + }), + }), + 'original_device_class': , + 'original_icon': None, + 'original_name': 'Current L1', + 'platform': 'bitvis', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': 0, + 'translation_key': 'phase_current', + 'unique_id': 'aa:bb:cc:dd:ee:ff_phase_current_l1', + 'unit_of_measurement': , + }) +# --- +# name: test_all_entities[sensor.bitvis_power_hub_current_l1-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + : 'current', + : 'Bitvis Power Hub Current L1', + : , + : , + }), + 'context': , + 'entity_id': 'sensor.bitvis_power_hub_current_l1', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': '10.5', + }) +# --- +# name: test_all_entities[sensor.bitvis_power_hub_current_l2-entry] + EntityRegistryEntrySnapshot({ + 'aliases': list([ + None, + ]), + 'area_id': None, + 'capabilities': dict({ + : , + }), + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'sensor', + 'entity_category': None, + 'entity_id': 'sensor.bitvis_power_hub_current_l2', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'object_id_base': 'Current L2', + 'options': dict({ + 'sensor': dict({ + 'suggested_display_precision': 2, + }), + }), + 'original_device_class': , + 'original_icon': None, + 'original_name': 'Current L2', + 'platform': 'bitvis', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': 0, + 'translation_key': 'phase_current', + 'unique_id': 'aa:bb:cc:dd:ee:ff_phase_current_l2', + 'unit_of_measurement': , + }) +# --- +# name: test_all_entities[sensor.bitvis_power_hub_current_l2-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + : 'current', + : 'Bitvis Power Hub Current L2', + : , + : , + }), + 'context': , + 'entity_id': 'sensor.bitvis_power_hub_current_l2', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': '8.30000019073486', + }) +# --- +# name: test_all_entities[sensor.bitvis_power_hub_current_l3-entry] + EntityRegistryEntrySnapshot({ + 'aliases': list([ + None, + ]), + 'area_id': None, + 'capabilities': dict({ + : , + }), + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'sensor', + 'entity_category': None, + 'entity_id': 'sensor.bitvis_power_hub_current_l3', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'object_id_base': 'Current L3', + 'options': dict({ + 'sensor': dict({ + 'suggested_display_precision': 2, + }), + }), + 'original_device_class': , + 'original_icon': None, + 'original_name': 'Current L3', + 'platform': 'bitvis', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': 0, + 'translation_key': 'phase_current', + 'unique_id': 'aa:bb:cc:dd:ee:ff_phase_current_l3', + 'unit_of_measurement': , + }) +# --- +# name: test_all_entities[sensor.bitvis_power_hub_current_l3-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + : 'current', + : 'Bitvis Power Hub Current L3', + : , + : , + }), + 'context': , + 'entity_id': 'sensor.bitvis_power_hub_current_l3', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': '12.1000003814697', + }) +# --- +# name: test_all_entities[sensor.bitvis_power_hub_han_buffer_overflows-entry] + EntityRegistryEntrySnapshot({ + 'aliases': list([ + None, + ]), + 'area_id': None, + 'capabilities': dict({ + : , + }), + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'sensor', + 'entity_category': , + 'entity_id': 'sensor.bitvis_power_hub_han_buffer_overflows', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'object_id_base': 'HAN buffer overflows', + 'options': dict({ + }), + 'original_device_class': None, + 'original_icon': None, + 'original_name': 'HAN buffer overflows', + 'platform': 'bitvis', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': 0, + 'translation_key': 'han_msg_buffer_overflow', + 'unique_id': 'aa:bb:cc:dd:ee:ff_han_msg_buffer_overflow', + 'unit_of_measurement': None, + }) +# --- +# name: test_all_entities[sensor.bitvis_power_hub_han_buffer_overflows-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + : 'Bitvis Power Hub HAN buffer overflows', + : , + }), + 'context': , + 'entity_id': 'sensor.bitvis_power_hub_han_buffer_overflows', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': '5', + }) +# --- +# name: test_all_entities[sensor.bitvis_power_hub_han_messages_successfully_parsed-entry] + EntityRegistryEntrySnapshot({ + 'aliases': list([ + None, + ]), + 'area_id': None, + 'capabilities': dict({ + : , + }), + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'sensor', + 'entity_category': , + 'entity_id': 'sensor.bitvis_power_hub_han_messages_successfully_parsed', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'object_id_base': 'HAN messages successfully parsed', + 'options': dict({ + }), + 'original_device_class': None, + 'original_icon': None, + 'original_name': 'HAN messages successfully parsed', + 'platform': 'bitvis', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': 0, + 'translation_key': 'han_msg_successfully_parsed', + 'unique_id': 'aa:bb:cc:dd:ee:ff_han_msg_successfully_parsed', + 'unit_of_measurement': None, + }) +# --- +# name: test_all_entities[sensor.bitvis_power_hub_han_messages_successfully_parsed-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + : 'Bitvis Power Hub HAN messages successfully parsed', + : , + }), + 'context': , + 'entity_id': 'sensor.bitvis_power_hub_han_messages_successfully_parsed', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': '1000', + }) +# --- +# name: test_all_entities[sensor.bitvis_power_hub_reactive_energy_export-entry] + EntityRegistryEntrySnapshot({ + 'aliases': list([ + None, + ]), + 'area_id': None, + 'capabilities': dict({ + : , + }), + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'sensor', + 'entity_category': None, + 'entity_id': 'sensor.bitvis_power_hub_reactive_energy_export', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'object_id_base': 'Reactive energy export', + 'options': dict({ + 'sensor': dict({ + 'suggested_display_precision': 2, + }), + }), + 'original_device_class': , + 'original_icon': None, + 'original_name': 'Reactive energy export', + 'platform': 'bitvis', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': 0, + 'translation_key': 'energy_reactive_export', + 'unique_id': 'aa:bb:cc:dd:ee:ff_energy_reactive_delivered_by_client', + 'unit_of_measurement': , + }) +# --- +# name: test_all_entities[sensor.bitvis_power_hub_reactive_energy_export-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + : 'reactive_energy', + : 'Bitvis Power Hub Reactive energy export', + : , + : , + }), + 'context': , + 'entity_id': 'sensor.bitvis_power_hub_reactive_energy_export', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': '23.4500007629395', + }) +# --- +# name: test_all_entities[sensor.bitvis_power_hub_reactive_energy_import-entry] + EntityRegistryEntrySnapshot({ + 'aliases': list([ + None, + ]), + 'area_id': None, + 'capabilities': dict({ + : , + }), + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'sensor', + 'entity_category': None, + 'entity_id': 'sensor.bitvis_power_hub_reactive_energy_import', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'object_id_base': 'Reactive energy import', + 'options': dict({ + 'sensor': dict({ + 'suggested_display_precision': 2, + }), + }), + 'original_device_class': , + 'original_icon': None, + 'original_name': 'Reactive energy import', + 'platform': 'bitvis', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': 0, + 'translation_key': 'energy_reactive_import', + 'unique_id': 'aa:bb:cc:dd:ee:ff_energy_reactive_delivered_to_client', + 'unit_of_measurement': , + }) +# --- +# name: test_all_entities[sensor.bitvis_power_hub_reactive_energy_import-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + : 'reactive_energy', + : 'Bitvis Power Hub Reactive energy import', + : , + : , + }), + 'context': , + 'entity_id': 'sensor.bitvis_power_hub_reactive_energy_import', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': '45.6699981689453', + }) +# --- +# name: test_all_entities[sensor.bitvis_power_hub_reactive_power_export-entry] + EntityRegistryEntrySnapshot({ + 'aliases': list([ + None, + ]), + 'area_id': None, + 'capabilities': dict({ + : , + }), + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'sensor', + 'entity_category': None, + 'entity_id': 'sensor.bitvis_power_hub_reactive_power_export', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'object_id_base': 'Reactive power export', + 'options': dict({ + 'sensor': dict({ + 'suggested_display_precision': 3, + }), + }), + 'original_device_class': , + 'original_icon': None, + 'original_name': 'Reactive power export', + 'platform': 'bitvis', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': 0, + 'translation_key': 'power_reactive_export', + 'unique_id': 'aa:bb:cc:dd:ee:ff_power_reactive_delivered_by_client', + 'unit_of_measurement': , + }) +# --- +# name: test_all_entities[sensor.bitvis_power_hub_reactive_power_export-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + : 'reactive_power', + : 'Bitvis Power Hub Reactive power export', + : , + : , + }), + 'context': , + 'entity_id': 'sensor.bitvis_power_hub_reactive_power_export', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': '0.0', + }) +# --- +# name: test_all_entities[sensor.bitvis_power_hub_reactive_power_export_l1-entry] + EntityRegistryEntrySnapshot({ + 'aliases': list([ + None, + ]), + 'area_id': None, + 'capabilities': dict({ + : , + }), + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'sensor', + 'entity_category': None, + 'entity_id': 'sensor.bitvis_power_hub_reactive_power_export_l1', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'object_id_base': 'Reactive power export L1', + 'options': dict({ + 'sensor': dict({ + 'suggested_display_precision': 3, + }), + }), + 'original_device_class': , + 'original_icon': None, + 'original_name': 'Reactive power export L1', + 'platform': 'bitvis', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': 0, + 'translation_key': 'power_reactive_phase_export', + 'unique_id': 'aa:bb:cc:dd:ee:ff_power_reactive_l1_delivered_by_client', + 'unit_of_measurement': , + }) +# --- +# name: test_all_entities[sensor.bitvis_power_hub_reactive_power_export_l1-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + : 'reactive_power', + : 'Bitvis Power Hub Reactive power export L1', + : , + : , + }), + 'context': , + 'entity_id': 'sensor.bitvis_power_hub_reactive_power_export_l1', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': '0.0', + }) +# --- +# name: test_all_entities[sensor.bitvis_power_hub_reactive_power_export_l2-entry] + EntityRegistryEntrySnapshot({ + 'aliases': list([ + None, + ]), + 'area_id': None, + 'capabilities': dict({ + : , + }), + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'sensor', + 'entity_category': None, + 'entity_id': 'sensor.bitvis_power_hub_reactive_power_export_l2', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'object_id_base': 'Reactive power export L2', + 'options': dict({ + 'sensor': dict({ + 'suggested_display_precision': 3, + }), + }), + 'original_device_class': , + 'original_icon': None, + 'original_name': 'Reactive power export L2', + 'platform': 'bitvis', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': 0, + 'translation_key': 'power_reactive_phase_export', + 'unique_id': 'aa:bb:cc:dd:ee:ff_power_reactive_l2_delivered_by_client', + 'unit_of_measurement': , + }) +# --- +# name: test_all_entities[sensor.bitvis_power_hub_reactive_power_export_l2-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + : 'reactive_power', + : 'Bitvis Power Hub Reactive power export L2', + : , + : , + }), + 'context': , + 'entity_id': 'sensor.bitvis_power_hub_reactive_power_export_l2', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': '0.0', + }) +# --- +# name: test_all_entities[sensor.bitvis_power_hub_reactive_power_export_l3-entry] + EntityRegistryEntrySnapshot({ + 'aliases': list([ + None, + ]), + 'area_id': None, + 'capabilities': dict({ + : , + }), + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'sensor', + 'entity_category': None, + 'entity_id': 'sensor.bitvis_power_hub_reactive_power_export_l3', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'object_id_base': 'Reactive power export L3', + 'options': dict({ + 'sensor': dict({ + 'suggested_display_precision': 3, + }), + }), + 'original_device_class': , + 'original_icon': None, + 'original_name': 'Reactive power export L3', + 'platform': 'bitvis', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': 0, + 'translation_key': 'power_reactive_phase_export', + 'unique_id': 'aa:bb:cc:dd:ee:ff_power_reactive_l3_delivered_by_client', + 'unit_of_measurement': , + }) +# --- +# name: test_all_entities[sensor.bitvis_power_hub_reactive_power_export_l3-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + : 'reactive_power', + : 'Bitvis Power Hub Reactive power export L3', + : , + : , + }), + 'context': , + 'entity_id': 'sensor.bitvis_power_hub_reactive_power_export_l3', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': '0.0', + }) +# --- +# name: test_all_entities[sensor.bitvis_power_hub_reactive_power_import-entry] + EntityRegistryEntrySnapshot({ + 'aliases': list([ + None, + ]), + 'area_id': None, + 'capabilities': dict({ + : , + }), + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'sensor', + 'entity_category': None, + 'entity_id': 'sensor.bitvis_power_hub_reactive_power_import', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'object_id_base': 'Reactive power import', + 'options': dict({ + 'sensor': dict({ + 'suggested_display_precision': 3, + }), + }), + 'original_device_class': , + 'original_icon': None, + 'original_name': 'Reactive power import', + 'platform': 'bitvis', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': 0, + 'translation_key': 'power_reactive_import', + 'unique_id': 'aa:bb:cc:dd:ee:ff_power_reactive_delivered_to_client', + 'unit_of_measurement': , + }) +# --- +# name: test_all_entities[sensor.bitvis_power_hub_reactive_power_import-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + : 'reactive_power', + : 'Bitvis Power Hub Reactive power import', + : , + : , + }), + 'context': , + 'entity_id': 'sensor.bitvis_power_hub_reactive_power_import', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': '0.5', + }) +# --- +# name: test_all_entities[sensor.bitvis_power_hub_reactive_power_import_l1-entry] + EntityRegistryEntrySnapshot({ + 'aliases': list([ + None, + ]), + 'area_id': None, + 'capabilities': dict({ + : , + }), + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'sensor', + 'entity_category': None, + 'entity_id': 'sensor.bitvis_power_hub_reactive_power_import_l1', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'object_id_base': 'Reactive power import L1', + 'options': dict({ + 'sensor': dict({ + 'suggested_display_precision': 3, + }), + }), + 'original_device_class': , + 'original_icon': None, + 'original_name': 'Reactive power import L1', + 'platform': 'bitvis', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': 0, + 'translation_key': 'power_reactive_phase_import', + 'unique_id': 'aa:bb:cc:dd:ee:ff_power_reactive_l1_delivered_to_client', + 'unit_of_measurement': , + }) +# --- +# name: test_all_entities[sensor.bitvis_power_hub_reactive_power_import_l1-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + : 'reactive_power', + : 'Bitvis Power Hub Reactive power import L1', + : , + : , + }), + 'context': , + 'entity_id': 'sensor.bitvis_power_hub_reactive_power_import_l1', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': '0.200000002980232', + }) +# --- +# name: test_all_entities[sensor.bitvis_power_hub_reactive_power_import_l2-entry] + EntityRegistryEntrySnapshot({ + 'aliases': list([ + None, + ]), + 'area_id': None, + 'capabilities': dict({ + : , + }), + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'sensor', + 'entity_category': None, + 'entity_id': 'sensor.bitvis_power_hub_reactive_power_import_l2', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'object_id_base': 'Reactive power import L2', + 'options': dict({ + 'sensor': dict({ + 'suggested_display_precision': 3, + }), + }), + 'original_device_class': , + 'original_icon': None, + 'original_name': 'Reactive power import L2', + 'platform': 'bitvis', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': 0, + 'translation_key': 'power_reactive_phase_import', + 'unique_id': 'aa:bb:cc:dd:ee:ff_power_reactive_l2_delivered_to_client', + 'unit_of_measurement': , + }) +# --- +# name: test_all_entities[sensor.bitvis_power_hub_reactive_power_import_l2-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + : 'reactive_power', + : 'Bitvis Power Hub Reactive power import L2', + : , + : , + }), + 'context': , + 'entity_id': 'sensor.bitvis_power_hub_reactive_power_import_l2', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': '0.150000005960464', + }) +# --- +# name: test_all_entities[sensor.bitvis_power_hub_reactive_power_import_l3-entry] + EntityRegistryEntrySnapshot({ + 'aliases': list([ + None, + ]), + 'area_id': None, + 'capabilities': dict({ + : , + }), + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'sensor', + 'entity_category': None, + 'entity_id': 'sensor.bitvis_power_hub_reactive_power_import_l3', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'object_id_base': 'Reactive power import L3', + 'options': dict({ + 'sensor': dict({ + 'suggested_display_precision': 3, + }), + }), + 'original_device_class': , + 'original_icon': None, + 'original_name': 'Reactive power import L3', + 'platform': 'bitvis', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': 0, + 'translation_key': 'power_reactive_phase_import', + 'unique_id': 'aa:bb:cc:dd:ee:ff_power_reactive_l3_delivered_to_client', + 'unit_of_measurement': , + }) +# --- +# name: test_all_entities[sensor.bitvis_power_hub_reactive_power_import_l3-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + : 'reactive_power', + : 'Bitvis Power Hub Reactive power import L3', + : , + : , + }), + 'context': , + 'entity_id': 'sensor.bitvis_power_hub_reactive_power_import_l3', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': '0.150000005960464', + }) +# --- +# name: test_all_entities[sensor.bitvis_power_hub_uptime-entry] + EntityRegistryEntrySnapshot({ + 'aliases': list([ + None, + ]), + 'area_id': None, + 'capabilities': None, + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'sensor', + 'entity_category': , + 'entity_id': 'sensor.bitvis_power_hub_uptime', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'object_id_base': 'Uptime', + 'options': dict({ + }), + 'original_device_class': , + 'original_icon': None, + 'original_name': 'Uptime', + 'platform': 'bitvis', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': 0, + 'translation_key': None, + 'unique_id': 'aa:bb:cc:dd:ee:ff_uptime', + 'unit_of_measurement': None, + }) +# --- +# name: test_all_entities[sensor.bitvis_power_hub_uptime-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + : 'uptime', + : 'Bitvis Power Hub Uptime', + }), + 'context': , + 'entity_id': 'sensor.bitvis_power_hub_uptime', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': '2025-12-31T12:00:00+00:00', + }) +# --- +# name: test_all_entities[sensor.bitvis_power_hub_voltage_l1-entry] + EntityRegistryEntrySnapshot({ + 'aliases': list([ + None, + ]), + 'area_id': None, + 'capabilities': dict({ + : , + }), + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'sensor', + 'entity_category': None, + 'entity_id': 'sensor.bitvis_power_hub_voltage_l1', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'object_id_base': 'Voltage L1', + 'options': dict({ + 'sensor': dict({ + 'suggested_display_precision': 1, + }), + }), + 'original_device_class': , + 'original_icon': None, + 'original_name': 'Voltage L1', + 'platform': 'bitvis', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': 0, + 'translation_key': 'phase_voltage', + 'unique_id': 'aa:bb:cc:dd:ee:ff_phase_voltage_l1', + 'unit_of_measurement': , + }) +# --- +# name: test_all_entities[sensor.bitvis_power_hub_voltage_l1-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + : 'voltage', + : 'Bitvis Power Hub Voltage L1', + : , + : , + }), + 'context': , + 'entity_id': 'sensor.bitvis_power_hub_voltage_l1', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': '230.0', + }) +# --- +# name: test_all_entities[sensor.bitvis_power_hub_voltage_l2-entry] + EntityRegistryEntrySnapshot({ + 'aliases': list([ + None, + ]), + 'area_id': None, + 'capabilities': dict({ + : , + }), + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'sensor', + 'entity_category': None, + 'entity_id': 'sensor.bitvis_power_hub_voltage_l2', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'object_id_base': 'Voltage L2', + 'options': dict({ + 'sensor': dict({ + 'suggested_display_precision': 1, + }), + }), + 'original_device_class': , + 'original_icon': None, + 'original_name': 'Voltage L2', + 'platform': 'bitvis', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': 0, + 'translation_key': 'phase_voltage', + 'unique_id': 'aa:bb:cc:dd:ee:ff_phase_voltage_l2', + 'unit_of_measurement': , + }) +# --- +# name: test_all_entities[sensor.bitvis_power_hub_voltage_l2-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + : 'voltage', + : 'Bitvis Power Hub Voltage L2', + : , + : , + }), + 'context': , + 'entity_id': 'sensor.bitvis_power_hub_voltage_l2', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': '229.5', + }) +# --- +# name: test_all_entities[sensor.bitvis_power_hub_voltage_l3-entry] + EntityRegistryEntrySnapshot({ + 'aliases': list([ + None, + ]), + 'area_id': None, + 'capabilities': dict({ + : , + }), + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'sensor', + 'entity_category': None, + 'entity_id': 'sensor.bitvis_power_hub_voltage_l3', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'object_id_base': 'Voltage L3', + 'options': dict({ + 'sensor': dict({ + 'suggested_display_precision': 1, + }), + }), + 'original_device_class': , + 'original_icon': None, + 'original_name': 'Voltage L3', + 'platform': 'bitvis', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': 0, + 'translation_key': 'phase_voltage', + 'unique_id': 'aa:bb:cc:dd:ee:ff_phase_voltage_l3', + 'unit_of_measurement': , + }) +# --- +# name: test_all_entities[sensor.bitvis_power_hub_voltage_l3-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + : 'voltage', + : 'Bitvis Power Hub Voltage L3', + : , + : , + }), + 'context': , + 'entity_id': 'sensor.bitvis_power_hub_voltage_l3', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': '231.199996948242', + }) +# --- +# name: test_all_entities[sensor.bitvis_power_hub_wi_fi_signal_strength-entry] + EntityRegistryEntrySnapshot({ + 'aliases': list([ + None, + ]), + 'area_id': None, + 'capabilities': dict({ + : , + }), + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'sensor', + 'entity_category': , + 'entity_id': 'sensor.bitvis_power_hub_wi_fi_signal_strength', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'object_id_base': 'Wi-Fi signal strength', + 'options': dict({ + }), + 'original_device_class': , + 'original_icon': None, + 'original_name': 'Wi-Fi signal strength', + 'platform': 'bitvis', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': 0, + 'translation_key': 'wifi_rssi', + 'unique_id': 'aa:bb:cc:dd:ee:ff_wifi_rssi', + 'unit_of_measurement': 'dBm', + }) +# --- +# name: test_all_entities[sensor.bitvis_power_hub_wi_fi_signal_strength-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + : 'signal_strength', + : 'Bitvis Power Hub Wi-Fi signal strength', + : , + : 'dBm', + }), + 'context': , + 'entity_id': 'sensor.bitvis_power_hub_wi_fi_signal_strength', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': '-65', + }) +# --- diff --git a/tests/components/bitvis/test_config_flow.py b/tests/components/bitvis/test_config_flow.py new file mode 100644 index 000000000000..8fc292c01215 --- /dev/null +++ b/tests/components/bitvis/test_config_flow.py @@ -0,0 +1,613 @@ +"""Tests for the Bitvis Power Hub config flow.""" + +import asyncio +from ipaddress import ip_address +from unittest.mock import AsyncMock, MagicMock, patch + +from bitvis_protobuf.parse import PayloadSample +from bitvis_protobuf.powerhub_pb2 import Payload +import pytest + +from homeassistant.components.bitvis.const import DEFAULT_NAME, DEFAULT_PORT, DOMAIN +from homeassistant.components.bitvis.coordinator import async_get_listener_registry +from homeassistant.config_entries import SOURCE_USER, SOURCE_ZEROCONF +from homeassistant.const import CONF_HOST, CONF_PORT +from homeassistant.core import HomeAssistant +from homeassistant.data_entry_flow import FlowResultType +from homeassistant.helpers.service_info.zeroconf import ZeroconfServiceInfo + +from .conftest import ( + SECOND_DEVICE_MAC, + TEST_DEVICE_MAC, + FakeListener, + patch_config_flow_connectivity, +) + +from tests.common import MockConfigEntry + +pytestmark = pytest.mark.usefixtures("mock_setup_entry") + +ZEROCONF_HOST = "192.168.1.200" +USER_HOST = "192.168.1.100" + + +def _zeroconf_discovery( + host: str = ZEROCONF_HOST, + name: str = "Bitvis Power Hub._powerhub._udp.local.", + port: int | None = DEFAULT_PORT, +) -> ZeroconfServiceInfo: + return ZeroconfServiceInfo( + ip_address=ip_address(host), + ip_addresses=[ip_address(host)], + hostname="powerhub.local.", + name=name, + port=port, + properties={}, + type="_powerhub._udp.local.", + ) + + +ZEROCONF_DISCOVERY = _zeroconf_discovery() +UNRELATED_HOST = "10.9.9.9" + + +def _invalid_mac_datagram() -> bytes: + payload = Payload() + payload.sample.SetInParent() + return payload.SerializeToString() + + +@pytest.mark.parametrize( + ("input_host", "resolved_ip", "expected_host"), + [ + pytest.param(USER_HOST, USER_HOST, USER_HOST, id="ipv4"), + pytest.param("2001:db8::10", "2001:db8::10", "2001:db8::10", id="ipv6"), + pytest.param( + "my-powerhub.local", "10.0.0.5", "my-powerhub.local", id="hostname" + ), + pytest.param( + "[2001:db8::10]", "2001:db8::10", "2001:db8::10", id="bracketed-ipv6" + ), + ], +) +async def test_user_form_create_entry( + hass: HomeAssistant, + input_host: str, + resolved_ip: str, + expected_host: str, +) -> None: + """Test creating an entry via user flow.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": SOURCE_USER} + ) + + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "user" + + with patch_config_flow_connectivity(resolved_ip): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + { + CONF_HOST: input_host, + }, + ) + + assert result["type"] is FlowResultType.CREATE_ENTRY + assert result["title"] == DEFAULT_NAME + assert result["data"] == { + CONF_HOST: expected_host, + CONF_PORT: DEFAULT_PORT, + } + assert result["result"].unique_id == TEST_DEVICE_MAC + + +@pytest.mark.parametrize( + ("connectivity_kwargs", "error_key"), + [ + pytest.param( + {"port_bind_side_effect": OSError("UDP port is unavailable")}, + "cannot_connect", + id="cannot-connect", + ), + pytest.param({"invalid_mac": True}, "invalid_mac", id="invalid-mac"), + pytest.param( + {"deliver_mac": False, "discovery_timeout": True}, + "timeout_connect", + id="timeout", + ), + ], +) +async def test_user_form_error_and_recovery( + hass: HomeAssistant, + connectivity_kwargs: dict[str, object], + error_key: str, +) -> None: + """Test user form error then successful recovery.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": SOURCE_USER} + ) + + with patch_config_flow_connectivity(USER_HOST, **connectivity_kwargs): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + { + CONF_HOST: USER_HOST, + }, + ) + + assert result["type"] is FlowResultType.FORM + assert result["errors"] == {"base": error_key} + + with patch_config_flow_connectivity(USER_HOST): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + { + CONF_HOST: USER_HOST, + }, + ) + + assert result["type"] is FlowResultType.CREATE_ENTRY + assert result["title"] == DEFAULT_NAME + assert result["data"] == { + CONF_HOST: USER_HOST, + CONF_PORT: DEFAULT_PORT, + } + assert result["result"].unique_id == TEST_DEVICE_MAC + + +@pytest.mark.parametrize( + "input_host", + [ + pytest.param(USER_HOST, id="same-host"), + pytest.param("192.168.1.101", id="different-host"), + ], +) +async def test_user_form_duplicate_mac( + hass: HomeAssistant, mock_config_entry: MockConfigEntry, input_host: str +) -> None: + """Test duplicate detection is based on MAC address, not host.""" + mock_config_entry.add_to_hass(hass) + + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": SOURCE_USER} + ) + + with patch_config_flow_connectivity(input_host): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + { + CONF_HOST: input_host, + }, + ) + + assert result["type"] is FlowResultType.ABORT + assert result["reason"] == "already_configured" + + +async def test_user_form_reused_ip_new_device( + hass: HomeAssistant, mock_config_entry: MockConfigEntry +) -> None: + """Test a new device can be added at an IP already stored on another entry.""" + mock_config_entry.add_to_hass(hass) + + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": SOURCE_USER} + ) + + with patch_config_flow_connectivity(USER_HOST, mac_address=SECOND_DEVICE_MAC): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + { + CONF_HOST: USER_HOST, + }, + ) + + assert result["type"] is FlowResultType.CREATE_ENTRY + assert result["data"] == { + CONF_HOST: USER_HOST, + CONF_PORT: DEFAULT_PORT, + } + assert result["result"].unique_id == SECOND_DEVICE_MAC + + +@pytest.mark.parametrize( + ("connectivity_kwargs", "reason"), + [ + pytest.param( + {"port_bind_side_effect": OSError("UDP port is unavailable")}, + "cannot_connect", + id="cannot-connect", + ), + pytest.param({"invalid_mac": True}, "invalid_mac", id="invalid-mac"), + pytest.param( + {"deliver_mac": False, "discovery_timeout": True}, + "timeout_connect", + id="timeout", + ), + ], +) +async def test_zeroconf_abort( + hass: HomeAssistant, + connectivity_kwargs: dict[str, object], + reason: str, +) -> None: + """Test zeroconf discovery abort reasons.""" + with patch_config_flow_connectivity(ZEROCONF_HOST, **connectivity_kwargs): + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": SOURCE_ZEROCONF}, + data=ZEROCONF_DISCOVERY, + ) + + assert result["type"] is FlowResultType.ABORT + assert result["reason"] == reason + + +async def test_zeroconf_duplicate( + hass: HomeAssistant, mock_zeroconf_config_entry: MockConfigEntry +) -> None: + """Test that a duplicate zeroconf discovery is aborted by MAC address.""" + mock_zeroconf_config_entry.add_to_hass(hass) + + with patch_config_flow_connectivity(ZEROCONF_HOST): + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": SOURCE_ZEROCONF}, + data=ZEROCONF_DISCOVERY, + ) + + assert result["type"] is FlowResultType.ABORT + assert result["reason"] == "already_configured" + + +async def test_zeroconf_reused_ip_new_device( + hass: HomeAssistant, mock_zeroconf_config_entry: MockConfigEntry +) -> None: + """Test zeroconf can add a new device at an IP already stored on another entry.""" + mock_zeroconf_config_entry.add_to_hass(hass) + + with patch_config_flow_connectivity(ZEROCONF_HOST, mac_address=SECOND_DEVICE_MAC): + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": SOURCE_ZEROCONF}, + data=ZEROCONF_DISCOVERY, + ) + + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "zeroconf_confirm" + + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input={} + ) + + assert result["type"] is FlowResultType.CREATE_ENTRY + assert result["data"] == { + CONF_HOST: ZEROCONF_HOST, + CONF_PORT: DEFAULT_PORT, + } + assert result["result"].unique_id == SECOND_DEVICE_MAC + + +@pytest.mark.parametrize( + ("name", "port", "expected_title"), + [ + pytest.param( + "Bitvis Power Hub._powerhub._udp.local.", + DEFAULT_PORT, + "Bitvis Power Hub", + id="happy-path", + ), + pytest.param( + "Bitvis Power Hub._powerhub._udp.local.", + None, + "Bitvis Power Hub", + id="none-port", + ), + pytest.param( + "My Custom Hub._powerhub._udp.local.", + DEFAULT_PORT, + "My Custom Hub", + id="friendly-name", + ), + pytest.param("", DEFAULT_PORT, DEFAULT_NAME, id="empty-name"), + pytest.param( + "._powerhub._udp.local.", DEFAULT_PORT, DEFAULT_NAME, id="dot-prefixed" + ), + ], +) +async def test_zeroconf_create_entry( + hass: HomeAssistant, + name: str, + port: int | None, + expected_title: str, +) -> None: + """Test zeroconf confirm creates an entry with title, data, and unique_id.""" + discovery = _zeroconf_discovery(name=name, port=port) + + with patch_config_flow_connectivity(ZEROCONF_HOST): + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": SOURCE_ZEROCONF}, + data=discovery, + ) + + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "zeroconf_confirm" + + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input={} + ) + + assert result["type"] is FlowResultType.CREATE_ENTRY + assert result["title"] == expected_title + assert result["data"] == { + CONF_HOST: ZEROCONF_HOST, + CONF_PORT: DEFAULT_PORT, + } + assert result["result"].unique_id == TEST_DEVICE_MAC + + +async def test_zeroconf_updates_host_on_new_ip( + hass: HomeAssistant, mock_config_entry: MockConfigEntry +) -> None: + """Test rediscovery on a new IP updates the stored host and aborts.""" + mock_config_entry.add_to_hass(hass) + assert mock_config_entry.data[CONF_HOST] != ZEROCONF_HOST + + with patch_config_flow_connectivity(ZEROCONF_HOST): + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": SOURCE_ZEROCONF}, + data=ZEROCONF_DISCOVERY, + ) + + assert result["type"] is FlowResultType.ABORT + assert result["reason"] == "already_configured" + assert mock_config_entry.data[CONF_HOST] == ZEROCONF_HOST + + +async def test_aborted_flow_removes_listener( + hass: HomeAssistant, + mock_shared_listener: FakeListener, +) -> None: + """Test listener is stopped after an aborted config flow.""" + with patch_config_flow_connectivity( + USER_HOST, + deliver_mac=False, + discovery_timeout=True, + shared_listener=mock_shared_listener, + ): + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": SOURCE_USER} + ) + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + { + CONF_HOST: USER_HOST, + }, + ) + + assert result["type"] is FlowResultType.FORM + assert result["errors"] == {"base": "timeout_connect"} + mock_shared_listener.stop.assert_awaited_once() + assert not async_get_listener_registry(hass).has_listener(DEFAULT_PORT) + + +async def test_invalid_mac_from_other_host_is_ignored( + hass: HomeAssistant, mock_shared_listener: FakeListener +) -> None: + """Test an invalid-MAC datagram from another host does not fail the flow.""" + with patch_config_flow_connectivity( + USER_HOST, deliver_mac=False, shared_listener=mock_shared_listener + ): + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": SOURCE_USER} + ) + configure_task = asyncio.create_task( + hass.config_entries.flow.async_configure( + result["flow_id"], + { + CONF_HOST: USER_HOST, + }, + ) + ) + await hass.async_block_till_done() + + mock_shared_listener.dispatch(_invalid_mac_datagram(), (UNRELATED_HOST, 1234)) + mock_shared_listener.deliver( + PayloadSample(mac_address=TEST_DEVICE_MAC, sample=MagicMock()), + (USER_HOST, 1234), + ) + result = await configure_task + + assert result["type"] is FlowResultType.CREATE_ENTRY + assert result["result"].unique_id == TEST_DEVICE_MAC + + +async def test_invalid_mac_does_not_fail_other_flow( + hass: HomeAssistant, mock_shared_listener: FakeListener +) -> None: + """Test an invalid-MAC datagram only fails the flow waiting for that host.""" + + async def resolve_host(host: str) -> set[str]: + return {host} + + with ( + patch( + "homeassistant.components.bitvis.config_flow.async_verify_udp_port_bindable", + new_callable=AsyncMock, + ), + patch( + "homeassistant.components.bitvis.config_flow.async_resolve_host", + side_effect=resolve_host, + ), + patch( + "homeassistant.components.bitvis.coordinator.SharedListener", + return_value=mock_shared_listener, + ), + ): + first_result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": SOURCE_USER} + ) + first_task = asyncio.create_task( + hass.config_entries.flow.async_configure( + first_result["flow_id"], + { + CONF_HOST: USER_HOST, + }, + ) + ) + await hass.async_block_till_done() + + second_result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": SOURCE_USER} + ) + second_task = asyncio.create_task( + hass.config_entries.flow.async_configure( + second_result["flow_id"], + { + CONF_HOST: ZEROCONF_HOST, + }, + ) + ) + await hass.async_block_till_done() + + mock_shared_listener.dispatch(_invalid_mac_datagram(), (ZEROCONF_HOST, 1234)) + mock_shared_listener.deliver( + PayloadSample(mac_address=TEST_DEVICE_MAC, sample=MagicMock()), + (USER_HOST, 1234), + ) + first_result = await first_task + second_result = await second_task + + assert first_result["type"] is FlowResultType.CREATE_ENTRY + assert first_result["result"].unique_id == TEST_DEVICE_MAC + assert second_result["type"] is FlowResultType.FORM + assert second_result["errors"] == {"base": "invalid_mac"} + + +async def test_concurrent_flow_same_host_aborts(hass: HomeAssistant) -> None: + """Test concurrent flows for the same host abort with already_in_progress.""" + with patch_config_flow_connectivity(USER_HOST, deliver_mac=False): + first_result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": SOURCE_USER} + ) + first_task = asyncio.create_task( + hass.config_entries.flow.async_configure( + first_result["flow_id"], + { + CONF_HOST: USER_HOST, + }, + ) + ) + await hass.async_block_till_done() + + second_result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": SOURCE_USER} + ) + second_result = await hass.config_entries.flow.async_configure( + second_result["flow_id"], + { + CONF_HOST: USER_HOST, + }, + ) + + first_task.cancel() + with pytest.raises(asyncio.CancelledError): + await first_task + + assert second_result["type"] is FlowResultType.ABORT + assert second_result["reason"] == "already_in_progress" + + +async def test_discovery_register_runtime_error_aborts(hass: HomeAssistant) -> None: + """Test discovery aborts when filter registration raises RuntimeError.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": SOURCE_USER} + ) + + with patch_config_flow_connectivity( + USER_HOST, + deliver_mac=False, + register_side_effect=RuntimeError("Filter already registered"), + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + { + CONF_HOST: USER_HOST, + }, + ) + + assert result["type"] is FlowResultType.ABORT + assert result["reason"] == "already_in_progress" + + +async def test_zeroconf_concurrent_flow_same_host_aborts(hass: HomeAssistant) -> None: + """Test concurrent zeroconf flows for the same host abort.""" + with patch_config_flow_connectivity(ZEROCONF_HOST, deliver_mac=False): + first_task = asyncio.create_task( + hass.config_entries.flow.async_init( + DOMAIN, + context={"source": SOURCE_ZEROCONF}, + data=ZEROCONF_DISCOVERY, + ) + ) + await hass.async_block_till_done() + + second_result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": SOURCE_ZEROCONF}, + data=ZEROCONF_DISCOVERY, + ) + + first_task.cancel() + with pytest.raises(asyncio.CancelledError): + await first_task + + assert second_result["type"] is FlowResultType.ABORT + assert second_result["reason"] == "already_in_progress" + + +@pytest.mark.parametrize( + ("listener_already_running", "expected_awaits"), + [ + pytest.param(True, 0, id="listener-exists"), + pytest.param(False, 1, id="no-listener"), + ], +) +async def test_user_form_port_bind_check( + hass: HomeAssistant, + mock_shared_listener: FakeListener, + listener_already_running: bool, + expected_awaits: int, +) -> None: + """Test user flow skips the port bind check only when a listener exists.""" + if listener_already_running: + with patch( + "homeassistant.components.bitvis.coordinator.SharedListener", + return_value=mock_shared_listener, + ): + await async_get_listener_registry(hass).async_get_or_create(DEFAULT_PORT) + + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": SOURCE_USER} + ) + + kwargs: dict[str, object] = { + "mac_address": SECOND_DEVICE_MAC + if listener_already_running + else TEST_DEVICE_MAC + } + if listener_already_running: + kwargs["shared_listener"] = mock_shared_listener + + with patch_config_flow_connectivity("192.168.1.101", **kwargs) as mock_verify: + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + { + CONF_HOST: "192.168.1.101", + }, + ) + + assert mock_verify.await_count == expected_awaits + assert result["type"] is FlowResultType.CREATE_ENTRY diff --git a/tests/components/bitvis/test_coordinator.py b/tests/components/bitvis/test_coordinator.py new file mode 100644 index 000000000000..0a3b79d2d201 --- /dev/null +++ b/tests/components/bitvis/test_coordinator.py @@ -0,0 +1,45 @@ +"""Tests for the Bitvis Power Hub coordinator.""" + +from unittest.mock import AsyncMock + +import pytest + +from homeassistant.config_entries import ConfigEntryState +from homeassistant.core import HomeAssistant + +from .conftest import FakeListener + +from tests.common import MockConfigEntry + +pytestmark = pytest.mark.usefixtures("patch_shared_listener") + + +async def test_setup_oserror_results_in_setup_retry( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + mock_shared_listener: FakeListener, +) -> None: + """Test that OSError from SharedListener.start results in SETUP_RETRY.""" + mock_shared_listener.start = AsyncMock(side_effect=OSError("port in use")) + 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 mock_config_entry.state is ConfigEntryState.SETUP_RETRY + + +async def test_setup_runtime_error_results_in_setup_error( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + mock_shared_listener: FakeListener, +) -> None: + """Test that RuntimeError from SharedListener.register results in SETUP_ERROR.""" + mock_shared_listener.register.side_effect = RuntimeError("duplicate filter") + 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 mock_config_entry.state is ConfigEntryState.SETUP_ERROR + mock_shared_listener.unregister.assert_not_called() diff --git a/tests/components/bitvis/test_init.py b/tests/components/bitvis/test_init.py new file mode 100644 index 000000000000..41f3b39999a7 --- /dev/null +++ b/tests/components/bitvis/test_init.py @@ -0,0 +1,99 @@ +"""Tests for the Bitvis Power Hub integration.""" + +from bitvis_protobuf import powerhub_pb2 +from bitvis_protobuf.parse import PayloadSample +import pytest + +from homeassistant.components.bitvis.const import DATA_LISTENER_REGISTRY, DOMAIN +from homeassistant.config_entries import ConfigEntryState +from homeassistant.const import STATE_UNAVAILABLE +from homeassistant.core import HomeAssistant +from homeassistant.helpers import entity_registry as er + +from . import find_listener_callback, setup_integration +from .conftest import TEST_DEVICE_MAC, FakeListener + +from tests.common import MockConfigEntry + + +async def test_setup_entry( + init_integration: MockConfigEntry, +) -> None: + """Test successful integration setup.""" + assert init_integration.state is ConfigEntryState.LOADED + + +async def test_unload_entry( + hass: HomeAssistant, init_integration: MockConfigEntry +) -> None: + """Test that unloading stops the coordinator and unloads platforms.""" + assert DATA_LISTENER_REGISTRY in hass.data + assert await hass.config_entries.async_unload(init_integration.entry_id) + assert init_integration.state is ConfigEntryState.NOT_LOADED + assert DATA_LISTENER_REGISTRY not in hass.data + + +async def test_two_entries_share_listener( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + mock_second_config_entry: MockConfigEntry, + patch_shared_listener: FakeListener, +) -> None: + """Test that two entries on the same port share one library listener.""" + mock_config_entry.add_to_hass(hass) + await hass.config_entries.async_setup(mock_config_entry.entry_id) + await hass.async_block_till_done() + + mock_second_config_entry.add_to_hass(hass) + await hass.config_entries.async_setup(mock_second_config_entry.entry_id) + await hass.async_block_till_done() + + assert mock_config_entry.state is ConfigEntryState.LOADED + assert mock_second_config_entry.state is ConfigEntryState.LOADED + patch_shared_listener.start.assert_awaited_once() + + assert await hass.config_entries.async_unload(mock_config_entry.entry_id) + patch_shared_listener.stop.assert_not_called() + assert DATA_LISTENER_REGISTRY in hass.data + + assert await hass.config_entries.async_unload(mock_second_config_entry.entry_id) + patch_shared_listener.stop.assert_awaited_once() + assert DATA_LISTENER_REGISTRY not in hass.data + + +async def test_unload_after_dynamic_entities( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + patch_shared_listener: FakeListener, +) -> None: + """Test unload succeeds after HAN sensors have been created dynamically.""" + await setup_integration(hass, mock_config_entry) + + payload = powerhub_pb2.Payload() + payload.sample.power_active_delivered_to_client_kw = 2.0 + find_listener_callback(patch_shared_listener, TEST_DEVICE_MAC)( + PayloadSample(mac_address=TEST_DEVICE_MAC, sample=payload.sample), + ("192.168.1.100", 1234), + ) + await hass.async_block_till_done() + + assert await hass.config_entries.async_unload(mock_config_entry.entry_id) + assert mock_config_entry.state is ConfigEntryState.NOT_LOADED + + +@pytest.mark.usefixtures("entity_registry_enabled_by_default") +async def test_entities_unavailable_before_data( + hass: HomeAssistant, + init_integration: MockConfigEntry, + entity_registry: er.EntityRegistry, +) -> None: + """Test diagnostic entities are unavailable after setup before any packet.""" + wifi_entity_id = entity_registry.async_get_entity_id( + "sensor", + DOMAIN, + f"{TEST_DEVICE_MAC}_wifi_rssi", + ) + assert wifi_entity_id is not None + state = hass.states.get(wifi_entity_id) + assert state is not None + assert state.state == STATE_UNAVAILABLE diff --git a/tests/components/bitvis/test_sensor.py b/tests/components/bitvis/test_sensor.py new file mode 100644 index 000000000000..2aab629157f3 --- /dev/null +++ b/tests/components/bitvis/test_sensor.py @@ -0,0 +1,293 @@ +"""Tests for the Bitvis Power Hub sensor platform.""" + +from unittest.mock import patch + +from bitvis_protobuf import powerhub_pb2 +from bitvis_protobuf.parse import PayloadDiagnostic, PayloadSample +import pytest +from syrupy.assertion import SnapshotAssertion + +from homeassistant.components.bitvis.const import DOMAIN +from homeassistant.const import Platform +from homeassistant.core import HomeAssistant +from homeassistant.helpers import device_registry as dr, entity_registry as er +from homeassistant.helpers.device_registry import CONNECTION_NETWORK_MAC + +from . import find_listener_callback, setup_integration +from .conftest import TEST_DEVICE_MAC, FakeListener + +from tests.common import MockConfigEntry, snapshot_platform + +DIAGNOSTIC_UNIQUE_IDS = { + f"{TEST_DEVICE_MAC}_uptime", + f"{TEST_DEVICE_MAC}_wifi_rssi", + f"{TEST_DEVICE_MAC}_han_msg_successfully_parsed", + f"{TEST_DEVICE_MAC}_han_msg_buffer_overflow", +} + + +@pytest.fixture(autouse=True) +def enable_all_entities(entity_registry_enabled_by_default: None) -> None: + """Make sure all entities are enabled.""" + + +@pytest.fixture +def sample_payload() -> PayloadSample: + """Return a sample payload with test data.""" + payload = powerhub_pb2.Payload() + payload.sample.phase_voltage_l1_v = 230.0 + payload.sample.phase_voltage_l2_v = 229.5 + payload.sample.phase_voltage_l3_v = 231.2 + payload.sample.phase_current_l1_a = 10.5 + payload.sample.phase_current_l2_a = 8.3 + payload.sample.phase_current_l3_a = 12.1 + payload.sample.power_active_delivered_to_client_kw = 2.415 + payload.sample.power_active_delivered_by_client_kw = 0.0 + payload.sample.power_reactive_delivered_to_client_kvar = 0.5 + payload.sample.power_reactive_delivered_by_client_kvar = 0.0 + payload.sample.power_active_l1_delivered_to_client_kw = 0.8 + payload.sample.power_active_l2_delivered_to_client_kw = 0.7 + payload.sample.power_active_l3_delivered_to_client_kw = 0.915 + payload.sample.power_active_l1_delivered_by_client_kw = 0.0 + payload.sample.power_active_l2_delivered_by_client_kw = 0.0 + payload.sample.power_active_l3_delivered_by_client_kw = 0.0 + payload.sample.power_reactive_l1_delivered_to_client_kvar = 0.2 + payload.sample.power_reactive_l2_delivered_to_client_kvar = 0.15 + payload.sample.power_reactive_l3_delivered_to_client_kvar = 0.15 + payload.sample.power_reactive_l1_delivered_by_client_kvar = 0.0 + payload.sample.power_reactive_l2_delivered_by_client_kvar = 0.0 + payload.sample.power_reactive_l3_delivered_by_client_kvar = 0.0 + payload.sample.energy_active_delivered_to_client_kwh = 1234.56 + payload.sample.energy_active_delivered_by_client_kwh = 789.12 + payload.sample.energy_reactive_delivered_to_client_kvarh = 45.67 + payload.sample.energy_reactive_delivered_by_client_kvarh = 23.45 + return PayloadSample(mac_address=TEST_DEVICE_MAC, sample=payload.sample) + + +@pytest.fixture +def diagnostic_payload() -> PayloadDiagnostic: + """Return a diagnostic payload with test data.""" + payload = powerhub_pb2.Payload() + payload.diagnostic.uptime_s = 86400 + payload.diagnostic.wifi_rssi_dbm = -65 + payload.diagnostic.device_info.model_name = "PowerHub Gen2" + payload.diagnostic.device_info.sw_version = "2.0.0" + payload.diagnostic.device_info.mac_address = b"\xaa\xbb\xcc\xdd\xee\xff" + payload.diagnostic.han_msg_successfully_parsed = 1000 + payload.diagnostic.han_msg_buffer_overflow = 5 + return PayloadDiagnostic(mac_address=TEST_DEVICE_MAC, diagnostic=payload.diagnostic) + + +@pytest.mark.freeze_time("2026-01-01 12:00:00") +async def test_all_entities( + hass: HomeAssistant, + snapshot: SnapshotAssertion, + mock_config_entry: MockConfigEntry, + entity_registry: er.EntityRegistry, + sample_payload: PayloadSample, + diagnostic_payload: PayloadDiagnostic, + patch_shared_listener: FakeListener, +) -> None: + """Test all entities with snapshot.""" + with patch("homeassistant.components.bitvis._PLATFORMS", [Platform.SENSOR]): + await setup_integration(hass, mock_config_entry) + + callback = find_listener_callback(patch_shared_listener, TEST_DEVICE_MAC) + callback(sample_payload, ("192.168.1.100", 1234)) + callback(diagnostic_payload, ("192.168.1.100", 1234)) + await hass.async_block_till_done() + + await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id) + + +@pytest.mark.usefixtures("init_integration") +async def test_entities_added_when_fields_become_available( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + entity_registry: er.EntityRegistry, + patch_shared_listener: FakeListener, +) -> None: + """Test that HAN sensors are created when their fields first appear.""" + base_unique_id = mock_config_entry.unique_id + + payload = powerhub_pb2.Payload() + payload.sample.power_active_delivered_to_client_kw = 2.0 + find_listener_callback(patch_shared_listener, TEST_DEVICE_MAC)( + PayloadSample(mac_address=TEST_DEVICE_MAC, sample=payload.sample), + ("192.168.1.100", 1234), + ) + await hass.async_block_till_done() + + unique_ids = { + entry.unique_id + for entry in er.async_entries_for_config_entry( + entity_registry, mock_config_entry.entry_id + ) + } + assert unique_ids == DIAGNOSTIC_UNIQUE_IDS | { + f"{base_unique_id}_power_active_delivered_to_client" + } + + payload.sample.phase_voltage_l1_v = 230.0 + find_listener_callback(patch_shared_listener, TEST_DEVICE_MAC)( + PayloadSample(mac_address=TEST_DEVICE_MAC, sample=payload.sample), + ("192.168.1.100", 1234), + ) + await hass.async_block_till_done() + + unique_ids = { + entry.unique_id + for entry in er.async_entries_for_config_entry( + entity_registry, mock_config_entry.entry_id + ) + } + assert unique_ids == DIAGNOSTIC_UNIQUE_IDS | { + f"{base_unique_id}_power_active_delivered_to_client", + f"{base_unique_id}_phase_voltage_l1", + } + + +@pytest.mark.usefixtures("init_integration") +async def test_sensors_become_available_with_data( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + entity_registry: er.EntityRegistry, + patch_shared_listener: FakeListener, +) -> None: + """Test that sensors become available when data arrives.""" + payload = powerhub_pb2.Payload() + payload.sample.power_active_delivered_to_client_kw = 2.0 + find_listener_callback(patch_shared_listener, TEST_DEVICE_MAC)( + PayloadSample(mac_address=TEST_DEVICE_MAC, sample=payload.sample), + ("192.168.1.100", 1234), + ) + await hass.async_block_till_done() + + entity_id = entity_registry.async_get_entity_id( + "sensor", + DOMAIN, + f"{mock_config_entry.unique_id}_power_active_delivered_to_client", + ) + assert entity_id is not None + state = hass.states.get(entity_id) + assert state is not None + assert state.state != "unavailable" + assert float(state.state) == pytest.approx(2.0) + + +@pytest.mark.usefixtures("init_integration") +async def test_diagnostic_sensors_update_with_data( + hass: HomeAssistant, + entity_registry: er.EntityRegistry, + patch_shared_listener: FakeListener, +) -> None: + """Test that diagnostic sensors update when a diagnostic payload arrives.""" + payload = powerhub_pb2.Payload() + payload.diagnostic.uptime_s = 999 + payload.diagnostic.wifi_rssi_dbm = -70 + find_listener_callback(patch_shared_listener, TEST_DEVICE_MAC)( + PayloadDiagnostic(mac_address=TEST_DEVICE_MAC, diagnostic=payload.diagnostic), + ("192.168.1.100", 1234), + ) + await hass.async_block_till_done() + + wifi_entity_id = entity_registry.async_get_entity_id( + "sensor", + DOMAIN, + f"{TEST_DEVICE_MAC}_wifi_rssi", + ) + assert wifi_entity_id is not None + wifi_state = hass.states.get(wifi_entity_id) + assert wifi_state is not None + assert wifi_state.state != "unavailable" + assert float(wifi_state.state) == pytest.approx(-70) + + uptime_entity_id = entity_registry.async_get_entity_id( + "sensor", + DOMAIN, + f"{TEST_DEVICE_MAC}_uptime", + ) + assert uptime_entity_id is not None + uptime_state = hass.states.get(uptime_entity_id) + assert uptime_state is not None + assert uptime_state.state != "unavailable" + + +@pytest.mark.usefixtures("init_integration") +async def test_device_info_updated_from_diagnostic( + hass: HomeAssistant, + init_integration: MockConfigEntry, + device_registry: dr.DeviceRegistry, + patch_shared_listener: FakeListener, +) -> None: + """Test that device info is updated from a diagnostic payload.""" + device = device_registry.async_get_device_by_identifier( + (DOMAIN, TEST_DEVICE_MAC), init_integration.entry_id + ) + assert device is not None + assert device.model is None + assert device.sw_version is None + assert (CONNECTION_NETWORK_MAC, TEST_DEVICE_MAC) in device.connections + + payload = powerhub_pb2.Payload() + payload.diagnostic.uptime_s = 10 + payload.diagnostic.device_info.model_name = "PowerHub Gen2" + payload.diagnostic.device_info.sw_version = "1.2.3" + payload.diagnostic.device_info.mac_address = b"\xaa\xbb\xcc\xdd\xee\xff" + find_listener_callback(patch_shared_listener, TEST_DEVICE_MAC)( + PayloadDiagnostic(mac_address=TEST_DEVICE_MAC, diagnostic=payload.diagnostic), + ("192.168.1.100", 1234), + ) + await hass.async_block_till_done() + + device = device_registry.async_get_device_by_identifier( + (DOMAIN, TEST_DEVICE_MAC), init_integration.entry_id + ) + assert device is not None + assert device.model == "PowerHub Gen2" + assert device.sw_version == "1.2.3" + + +@pytest.mark.usefixtures("init_integration") +async def test_device_info_kept_when_absent_in_later_payload( + hass: HomeAssistant, + init_integration: MockConfigEntry, + device_registry: dr.DeviceRegistry, + sample_payload: PayloadSample, + patch_shared_listener: FakeListener, +) -> None: + """Test that known model/sw_version are kept when later payloads omit them.""" + payload = powerhub_pb2.Payload() + payload.diagnostic.uptime_s = 10 + payload.diagnostic.device_info.model_name = "PowerHub" + payload.diagnostic.device_info.sw_version = "1.0" + payload.diagnostic.device_info.mac_address = b"\xaa\xbb\xcc\xdd\xee\xff" + callback = find_listener_callback(patch_shared_listener, TEST_DEVICE_MAC) + callback( + PayloadDiagnostic(mac_address=TEST_DEVICE_MAC, diagnostic=payload.diagnostic), + ("192.168.1.100", 1234), + ) + await hass.async_block_till_done() + + device = device_registry.async_get_device_by_identifier( + (DOMAIN, TEST_DEVICE_MAC), init_integration.entry_id + ) + assert device is not None + assert device.model == "PowerHub" + assert device.sw_version == "1.0" + + payload2 = powerhub_pb2.Payload() + payload2.diagnostic.uptime_s = 20 + callback( + PayloadDiagnostic(mac_address=TEST_DEVICE_MAC, diagnostic=payload2.diagnostic), + ("192.168.1.100", 1234), + ) + callback(sample_payload, ("192.168.1.100", 1234)) + await hass.async_block_till_done() + + device = device_registry.async_get_device_by_identifier( + (DOMAIN, TEST_DEVICE_MAC), init_integration.entry_id + ) + assert device is not None + assert device.model == "PowerHub" + assert device.sw_version == "1.0"