diff --git a/.coverage b/.coverage index f372c82..5156dd2 100644 Binary files a/.coverage and b/.coverage differ diff --git a/custom_components/blanco_unit/__init__.py b/custom_components/blanco_unit/__init__.py index b87cc78..36c2e26 100644 --- a/custom_components/blanco_unit/__init__.py +++ b/custom_components/blanco_unit/__init__.py @@ -4,6 +4,8 @@ import logging +from bleak.backends.device import BLEDevice +from bleak_retry_connector import BleakClientWithServiceCache, establish_connection from packaging import version from homeassistant.components import bluetooth @@ -22,7 +24,17 @@ IntegrationError, ) -from .const import BLE_CALLBACK, CONF_MAC, DOMAIN, MIN_HA_VERSION +from .client import validate_pin +from .const import ( + BLE_CALLBACK, + CHARACTERISTIC_UUID, + CONF_DEV_ID, + CONF_MAC, + CONF_PIN, + DOMAIN, + MIN_HA_VERSION, + RANDOM_MAC_PLACEHOLDER, +) from .coordinator import BlancoUnitCoordinator from .services import async_setup_services @@ -51,16 +63,164 @@ async def async_setup(hass: HomeAssistant, entry: BlancoUnitConfigEntry) -> bool return True -async def async_setup_entry( +def _is_random_mac(config_entry: BlancoUnitConfigEntry) -> bool: + """Check if config entry has a randomized MAC address.""" + return config_entry.data.get(CONF_MAC) == RANDOM_MAC_PLACEHOLDER + + +async def _find_device_by_scanning( + hass: HomeAssistant, pin: str, expected_dev_id: str +) -> BLEDevice: + """Find a BLE device by active scanning for CHARACTERISTIC_UUID and matching PIN + dev_id. + + Uses the shared HA BLE scanner to actively scan for devices that advertise + the Blanco Unit service UUID, sorts them by RSSI (closest first), + and tries to connect and validate each one. + + Raises: + ConfigEntryNotReady: No devices with the correct UUID were found. + ConfigEntryAuthFailed: Devices were found but none matched PIN + dev_id. + """ + scanner = bluetooth.async_get_scanner(hass) + discovered = scanner.discovered_devices_and_advertisement_data + + # Filter by CHARACTERISTIC_UUID and sort by RSSI (closest first) + candidates: list[tuple[BLEDevice, int]] = [] + for device, adv_data in discovered.values(): + if CHARACTERISTIC_UUID in adv_data.service_uuids: + candidates.append((device, adv_data.rssi)) + candidates.sort(key=lambda item: item[1], reverse=True) + + _LOGGER.debug( + "Random MAC scan: found %d candidates with UUID %s", + len(candidates), + CHARACTERISTIC_UUID, + ) + + if not candidates: + raise ConfigEntryNotReady( + translation_key="error_device_not_found", + ) + + had_auth_failure = False + + for device, rssi in candidates: + _LOGGER.debug( + "Random MAC scan: trying %s (RSSI: %s)", device.address, rssi + ) + client = None + try: + client = await establish_connection( + client_class=BleakClientWithServiceCache, + device=device, + name=device.name or "Unknown Device", + ) + + result = await validate_pin(client, pin) + + if not result.is_valid: + _LOGGER.debug( + "Random MAC scan: PIN rejected by %s", device.address + ) + had_auth_failure = True + continue + + if result.dev_id == expected_dev_id: + _LOGGER.debug( + "Random MAC scan: matched device %s (dev_id: %s)", + device.address, + result.dev_id, + ) + return device + + _LOGGER.debug( + "Random MAC scan: dev_id mismatch on %s (got %s, expected %s)", + device.address, + result.dev_id, + expected_dev_id, + ) + except (OSError, TimeoutError): + _LOGGER.debug( + "Random MAC scan: connection failed for %s", + device.address, + exc_info=True, + ) + finally: + if client is not None and client.is_connected: + await client.disconnect() + + # Tried all candidates, none matched + if had_auth_failure: + raise ConfigEntryAuthFailed( + translation_key="error_invalid_authentication", + ) + + raise ConfigEntryNotReady( + translation_key="error_device_not_found", + ) + + +def _register_retry_callback( hass: HomeAssistant, config_entry: BlancoUnitConfigEntry -) -> bool: - """Set up Blanco Unit Integration from a config entry.""" - _LOGGER.debug("async_setup_entry called with config_entry: %s", config_entry) +) -> None: + """Register a BLE callback to retry setup when device appears.""" + if hass.data[DOMAIN][config_entry.entry_id].get(BLE_CALLBACK) is not None: + return - hass.data.setdefault(DOMAIN, {}) - hass.data[DOMAIN].setdefault(config_entry.entry_id, {}) + random_mac = _is_random_mac(config_entry) + + def _available_callback( + info: BluetoothServiceInfoBleak, _change: BluetoothChange + ) -> None: + if random_mac: + _LOGGER.debug( + "Random MAC: device with UUID discovered at %s", info.address + ) + else: + _LOGGER.debug("%s is discovered again", info.address) + hass.async_create_task( + hass.config_entries.async_reload(config_entry.entry_id) + ) + + _LOGGER.debug("async_setup_entry async_register_callback (random_mac=%s)", random_mac) + + if random_mac: + # For random MAC, listen for any device advertising our service UUID + unregister_ble_callback = bluetooth.async_register_callback( + hass, + _available_callback, + {"service_uuid": CHARACTERISTIC_UUID, "connectable": True}, + BluetoothScanningMode.ACTIVE, + ) + else: + # For static MAC, listen for the specific address + unregister_ble_callback = bluetooth.async_register_callback( + hass, + _available_callback, + {"address": config_entry.data[CONF_MAC], "connectable": True}, + BluetoothScanningMode.ACTIVE, + ) + + hass.data[DOMAIN][config_entry.entry_id][BLE_CALLBACK] = unregister_ble_callback + + +async def _resolve_device( + hass: HomeAssistant, config_entry: BlancoUnitConfigEntry +) -> BLEDevice: + """Resolve the BLE device, handling both static and random MAC cases. + + Raises: + ConfigEntryNotReady: Device not found. + ConfigEntryAuthFailed: Device found but PIN/dev_id mismatch (random MAC only). + """ + if _is_random_mac(config_entry): + _LOGGER.debug("async_setup_entry: random MAC, scanning for device") + return await _find_device_by_scanning( + hass, + pin=str(config_entry.data[CONF_PIN]), + expected_dev_id=config_entry.data[CONF_DEV_ID], + ) - # Initialise the coordinator that manages data updates from your api. device = bluetooth.async_ble_device_from_address( hass=hass, address=config_entry.data[CONF_MAC], @@ -69,34 +229,31 @@ async def async_setup_entry( if device is None: _LOGGER.debug("async_setup_entry device not found") - - if hass.data[DOMAIN][config_entry.entry_id].get(BLE_CALLBACK) is None: - # Register a callback to retry setup when the device appears - def _available_callback( - info: BluetoothServiceInfoBleak, change: BluetoothChange - ) -> None: - _LOGGER.debug("%s is discovered again", info) - if info.address == config_entry.data[CONF_MAC]: - _LOGGER.debug("%s is discovered again", info.address) - # Schedule a reload of the config entry immediately - hass.async_create_task( - hass.config_entries.async_reload(config_entry.entry_id) - ) - - _LOGGER.debug("async_setup_entry async_register_callback") - unregister_ble_callback = bluetooth.async_register_callback( - hass, - _available_callback, - {"address": config_entry.data[CONF_MAC], "connectable": True}, - BluetoothScanningMode.ACTIVE, - ) - hass.data[DOMAIN][config_entry.entry_id][BLE_CALLBACK] = ( - unregister_ble_callback - ) + _register_retry_callback(hass, config_entry) raise ConfigEntryNotReady( translation_key="error_device_not_found", ) + return device + + +async def async_setup_entry( + hass: HomeAssistant, config_entry: BlancoUnitConfigEntry +) -> bool: + """Set up Blanco Unit Integration from a config entry.""" + _LOGGER.debug("async_setup_entry called with config_entry: %s", config_entry) + + hass.data.setdefault(DOMAIN, {}) + hass.data[DOMAIN].setdefault(config_entry.entry_id, {}) + + try: + device = await _resolve_device(hass, config_entry) + except ConfigEntryNotReady: + _register_retry_callback(hass, config_entry) + raise + except ConfigEntryAuthFailed: + raise + # Registers update listener to update config entry when options are updated. unsub_update_listener = config_entry.add_update_listener(async_reload_entry) @@ -166,6 +323,7 @@ async def async_unload_entry( ): coordinator: BlancoUnitCoordinator = config_entry.runtime_data await coordinator.unload() - bluetooth.async_rediscover_address(hass, config_entry.data[CONF_MAC]) + if not _is_random_mac(config_entry): + bluetooth.async_rediscover_address(hass, config_entry.data[CONF_MAC]) return unload_ok diff --git a/custom_components/blanco_unit/client.py b/custom_components/blanco_unit/client.py index 774831e..50fc3de 100644 --- a/custom_components/blanco_unit/client.py +++ b/custom_components/blanco_unit/client.py @@ -64,15 +64,17 @@ class _RequestMeta: evt_type: int dev_id: str | None = None - dev_type: int = 1 + dev_type: int | None = None evt_ver: int = 1 evt_ts: int = field(default_factory=lambda: int(time.time() * 1000)) def to_dict(self) -> dict[str, Any]: - """Convert to dictionary, omitting None dev_id.""" + """Convert to dictionary, omitting None dev_id and dev_type.""" data = asdict(self) if self.dev_id is None: del data["dev_id"] + if self.dev_type is None: + del data["dev_type"] return data @@ -302,7 +304,7 @@ async def send_pairing_request( self, client: BleakClient, pin: str ) -> dict[str, Any]: """Send pairing request and return parsed response.""" - meta = _RequestMeta(evt_type=10, dev_id=None) + meta = _RequestMeta(evt_type=10, dev_id=None, dev_type=None) body = _RequestBody(meta=meta, pars={}) req_id = random.randint(1000000, 9999999) @@ -319,7 +321,7 @@ async def send_pairing_request( request_dict = envelope.to_dict() packets = self.create_packets(request_dict, self.msg_id_counter) - _LOGGER.debug("Sending pairing data: %s", envelope) + _LOGGER.debug("Sending pairing data: %s", request_dict) _LOGGER.debug("Sending pairing request (ReqID: %s)", req_id) # Send packets @@ -335,12 +337,13 @@ async def send_request( client: BleakClient, pin: str, dev_id: str, + dev_type: int, evt_type: int, ctrl: int | None = None, pars: dict[str, Any] | None = None, ) -> dict[str, Any]: """Send a general request and return parsed response.""" - meta = _RequestMeta(evt_type=evt_type, dev_id=dev_id) + meta = _RequestMeta(evt_type=evt_type, dev_id=dev_id, dev_type=dev_type) opts_dict: dict[str, int] | None = {"ctrl": ctrl} if ctrl is not None else None body = _RequestBody(meta=meta, opts=opts_dict, pars=pars) @@ -362,7 +365,7 @@ async def send_request( request_dict = envelope.to_dict() packets = self.create_packets(request_dict, self.msg_id_counter) - _LOGGER.debug("Sending data: %s", envelope) + _LOGGER.debug("Sending data: %s", request_dict) _LOGGER.debug("Sending request (ReqID: %s, %d packets)", req_id, len(packets)) # Send packets @@ -412,6 +415,11 @@ def device_id(self) -> str | None: """Return the device ID from the current session, or None if not connected.""" return self._session_data.dev_id if self._session_data else None + @property + def device_type(self) -> int | None: + """Return the device type from the current session, or None if not connected.""" + return self._session_data.dev_type if self._session_data else None + @property def is_connected(self) -> bool: """Return True if the BLE client is currently connected.""" @@ -448,12 +456,13 @@ async def _connect(self) -> _BlancoUnitSessionData: protocol = _BlancoUnitProtocol(mtu=MTU_SIZE) # Perform initial pairing - dev_id = await self._perform_pairing(client, protocol) + result = await self._perform_pairing(client, protocol) - _LOGGER.debug("Connected and paired with device ID: %s", dev_id) + _LOGGER.debug("Connected and paired with device ID: %s, device type: %d", result.dev_id, result.dev_type,) self._session_data = _BlancoUnitSessionData( client=client, - dev_id=dev_id, + dev_id=result.dev_id, + dev_type=result.dev_type, protocol=protocol, ) self._connection_callback(self._session_data.client.is_connected) @@ -467,23 +476,27 @@ def _handle_disconnect(self, _: BleakClient) -> None: async def _perform_pairing( self, client: BleakClient, protocol: _BlancoUnitProtocol - ) -> str: - """Perform initial pairing to get device ID. + ) -> PinValidationResult: + """Perform initial pairing to get device ID and device type. + + Returns: + Tuple of (dev_id, dev_type). Raises: BlancoUnitAuthenticationError: If PIN is wrong (error code 4). BlancoUnitConnectionError: If device ID cannot be extracted. """ # Validate PIN and get response - is_valid, response = await validate_pin(client, self._pin, protocol) - if not is_valid: + validation = await validate_pin(client, self._pin, protocol) + if not validation.is_valid: raise BlancoUnitAuthenticationError("Wrong PIN - Authentication failed") - # Extract device ID using shared helper - dev_id = _extract_device_id(response) - if dev_id is None: + if validation.dev_id is None: raise BlancoUnitConnectionError("No device ID in pairing response") - return dev_id + + if validation.dev_type is None: + raise BlancoUnitConnectionError("No device type in pairing response") + return validation async def _execute_transaction( self, @@ -498,6 +511,7 @@ async def _execute_transaction( client=session_data.client, pin=self._pin, dev_id=session_data.dev_id, + dev_type=session_data.dev_type, evt_type=evt_type, ctrl=ctrl, pars=pars, @@ -704,7 +718,7 @@ async def set_calibration_soda(self, amount: int) -> bool: async def test_protocol_parameters( self, evt_type: int, ctrl: int | None = None, pars: dict[str, Any] | None = None - ) -> dict[str, Any] | None: + ) -> dict[str, Any]: """Test protocol parameters and return response if it contains meaningful data. Args: @@ -716,15 +730,9 @@ async def test_protocol_parameters( Response dictionary if it contains meaningful data, None otherwise. """ try: - response = await self._execute_transaction( + return await self._execute_transaction( evt_type=evt_type, ctrl=ctrl, pars=pars ) - - # Check if response contains meaningful data - if self._is_response_empty(response): - return None - - return response # noqa: TRY300 except Exception as e: # noqa: BLE001 _LOGGER.debug( "Test failed for evt_type=%s, ctrl=%s, pars=%s: %s", @@ -735,27 +743,20 @@ async def test_protocol_parameters( ) return None - def _is_response_empty(self, response: dict[str, Any]) -> bool: - """Check if a response contains meaningful data. - - Args: - response: Response dictionary to check. - - Returns: - True if response is empty or contains only empty structures. - """ - if not response: - return True - - # Check if body exists - body = response.get("body", {}) - return not body - # ------------------------------- # region Standalone Functions # ------------------------------- +@dataclass +class PinValidationResult: + """Result of PIN validation.""" + + is_valid: bool + dev_id: str | None + dev_type: int | None + + def _extract_device_id(response: dict[str, Any]) -> str | None: """Extract device ID from a pairing response. @@ -775,9 +776,28 @@ def _extract_device_id(response: dict[str, Any]) -> str | None: return None +def _extract_device_type(response: dict[str, Any]) -> int | None: + """Extract device type from a pairing response. + + Args: + response: The response dictionary from a pairing request. + + Returns: + The device type if found, None otherwise. + """ + try: + body = response.get("body", {}) + meta = body.get("meta", {}) + if "dev_type" in meta: + return meta["dev_type"] + except (KeyError, TypeError): + pass + return None + + async def validate_pin( client: BleakClient, pin: str, protocol: _BlancoUnitProtocol | None = None -) -> tuple[bool, dict[str, Any]]: +) -> PinValidationResult: """Test if a PIN is valid by attempting to pair with the device. This is a standalone function that works with an existing BleakClient. @@ -788,9 +808,10 @@ async def validate_pin( protocol: Optional protocol instance. If None, creates a new one. Returns: - Tuple of (is_valid, response_dict): + PinValidationResult containing: - is_valid: True if PIN is valid, False if wrong PIN (error code 4) - - response_dict: The full response from the pairing attempt + - response: The full response from the pairing attempt + - dev_id: The device ID if pairing was successful, None otherwise Raises: ValueError: If PIN format is invalid. @@ -809,21 +830,22 @@ async def validate_pin( # Send pairing request and get response response = await protocol.send_pairing_request(client, pin) + dev_id = _extract_device_id(response) + dev_type = _extract_device_type(response) + # Check for authentication error (error code 4) errors = protocol.extract_errors(response) for error in errors: if error.get("err_code") == 4: _LOGGER.debug("PIN validation failed: wrong PIN (error code 4)") - return (False, response) + return PinValidationResult(is_valid=False, dev_type=dev_type, dev_id=dev_id) - # Check if we got a device ID in results (successful pairing) - dev_id = _extract_device_id(response) if dev_id is not None: - _LOGGER.debug("PIN validation successful") - return (True, response) + _LOGGER.debug("PIN validation successful, dev_id: %s", dev_id) + return PinValidationResult(is_valid=True, dev_type=dev_type, dev_id=dev_id) - _LOGGER.debug("PIN validation failed: no device ID in response") - return (False, response) + _LOGGER.debug("PIN validation failed: no device ID or device type in response") + return PinValidationResult(is_valid=False, dev_type=dev_type, dev_id=dev_id) # ------------------------------- @@ -837,4 +859,5 @@ class _BlancoUnitSessionData: client: BleakClient dev_id: str + dev_type: int protocol: _BlancoUnitProtocol diff --git a/custom_components/blanco_unit/config_flow.py b/custom_components/blanco_unit/config_flow.py index adadb73..ad3ea5a 100644 --- a/custom_components/blanco_unit/config_flow.py +++ b/custom_components/blanco_unit/config_flow.py @@ -21,7 +21,15 @@ ) from .client import validate_pin -from .const import CONF_ERROR, CONF_MAC, CONF_NAME, CONF_PIN, DOMAIN +from .const import ( + CONF_DEV_ID, + CONF_ERROR, + CONF_MAC, + CONF_NAME, + CONF_PIN, + DOMAIN, + RANDOM_MAC_PLACEHOLDER, +) _LOGGER = logging.getLogger(__name__) @@ -31,6 +39,8 @@ class ValidationResult: """Result of the validation, errors is empty if successful.""" errors: dict[str, str] + dev_id: str | None = None + mac_address: str | None = None description_placeholders: dict[str, Any] | None = None @@ -117,16 +127,39 @@ async def validate_input(self, user_input: dict[str, Any]) -> ValidationResult: client = None try: - _LOGGER.debug("await async_ble_device_from_address") - device = bluetooth.async_ble_device_from_address( - hass=self.hass, - address=user_input[CONF_MAC], - connectable=True, - ) + # Use device from discovery_info if available + if self._discovery_info is not None: + _LOGGER.debug("Using device from discovery_info") + device = self._discovery_info.device + else: + _LOGGER.debug("await async_ble_device_from_address") + device = bluetooth.async_ble_device_from_address( + hass=self.hass, + address=user_input[CONF_MAC], + connectable=True, + ) if device is None: return ValidationResult({CONF_ERROR: "error_device_not_found"}) + # Check if device has randomized MAC address + # BLEDevice.address can be random if device uses BLE privacy + has_random_mac = getattr(device.details, "address_type", "").lower() in ( + "random", + "random_static", + "random_resolvable", + ) + mac_to_store = ( + RANDOM_MAC_PLACEHOLDER if has_random_mac else device.address + ) + + _LOGGER.debug( + "Device MAC: %s, Random: %s, Storing as: %s", + device.address, + has_random_mac, + mac_to_store, + ) + _LOGGER.debug("await establish_connection") client = await establish_connection( client_class=BleakClientWithServiceCache, @@ -135,15 +168,28 @@ async def validate_input(self, user_input: dict[str, Any]) -> ValidationResult: ) _LOGGER.debug("await validate_pin") - is_valid, _ = await validate_pin(client, pin_str) - _LOGGER.debug("validate_pin returned %s", is_valid) + validation = await validate_pin(client, pin_str) + _LOGGER.debug( + "validate_pin returned %s, dev_id: %s, dev_type: %s", + validation.is_valid, + validation.dev_id, + validation.dev_type, + ) - if not is_valid: + if not validation.is_valid: return ValidationResult({CONF_ERROR: "error_invalid_authentication"}) + if validation.dev_id is None or validation.dev_type is None: + return ValidationResult({CONF_ERROR: "error_device_not_found"}) + _LOGGER.debug( - "Successfully tested connection to %s", - user_input[CONF_MAC], + "Successfully tested connection to %s (dev_id: %s, dev_type: %s)", + mac_to_store, + validation.dev_id, + validation.dev_type, + ) + return ValidationResult( + {}, dev_id=validation.dev_id, mac_address=mac_to_store ) except ValueError as err: _LOGGER.error("Validation error: %s", err) @@ -159,14 +205,13 @@ async def validate_input(self, user_input: dict[str, Any]) -> ValidationResult: if client is not None and client.is_connected: await client.disconnect() - return ValidationResult({}) - async def async_step_bluetooth( self, discovery_info: BluetoothServiceInfoBleak ) -> ConfigFlowResult: """Handle a bluetooth device being discovered.""" + _LOGGER.debug("async_step_bluetooth called with %s and advertisement %s", discovery_info, discovery_info.advertisement) # Check if the device already exists. - await self.async_set_unique_id(discovery_info.address) + await self._async_handle_discovery_without_unique_id() self._abort_if_unique_id_configured() _LOGGER.debug("async_step_bluetooth %s", discovery_info) @@ -187,12 +232,24 @@ async def async_step_user( result = await self.validate_input(user_input) if not result.errors: # Validation was successful, create a unique id and create the config entry. - await self.async_set_unique_id(user_input[CONF_MAC]) + # Use MAC as unique_id for static MAC, dev_id for random MAC + if result.mac_address == RANDOM_MAC_PLACEHOLDER: + await self.async_set_unique_id(result.dev_id) + else: + await self.async_set_unique_id(result.mac_address) self._abort_if_unique_id_configured() - _LOGGER.debug("Create entry with %s", user_input) + + # Store MAC address and dev_id in config + config_data = user_input.copy() + config_data[CONF_MAC] = result.mac_address + config_data[CONF_DEV_ID] = result.dev_id + + _LOGGER.debug("Create entry with %s", config_data) + # Clean up discovery_info after successful validation + self._discovery_info = None return self.async_create_entry( title=user_input[CONF_NAME], - data=user_input, + data=config_data, ) return self.async_show_form( @@ -212,11 +269,23 @@ async def async_step_reauth( if user_input is not None: result = await self.validate_input(user_input) if not result.errors: - await self.async_set_unique_id(user_input[CONF_MAC]) + # Verify this is the same device + if result.mac_address == RANDOM_MAC_PLACEHOLDER: + await self.async_set_unique_id(result.dev_id) + else: + await self.async_set_unique_id(result.mac_address) self._abort_if_unique_id_mismatch(reason="wrong_device") + + # Update config with new PIN and potentially updated MAC + data_updates = user_input.copy() + data_updates[CONF_MAC] = result.mac_address + data_updates[CONF_DEV_ID] = result.dev_id + + # Clean up discovery_info after successful validation + self._discovery_info = None return self.async_update_reload_and_abort( entry=self._get_reauth_entry(), - data_updates=user_input, + data_updates=data_updates, ) return self.async_show_form( step_id="reauth", @@ -239,11 +308,23 @@ async def async_step_reconfigure( if user_input is not None: result = await self.validate_input(user_input) if not result.errors: - await self.async_set_unique_id(user_input[CONF_MAC]) + # Verify this is the same device + if result.mac_address == RANDOM_MAC_PLACEHOLDER: + await self.async_set_unique_id(result.dev_id) + else: + await self.async_set_unique_id(result.mac_address) self._abort_if_unique_id_mismatch(reason="wrong_device") + + # Update config with potentially updated MAC and dev_id + data_updates = user_input.copy() + data_updates[CONF_MAC] = result.mac_address + data_updates[CONF_DEV_ID] = result.dev_id + + # Clean up discovery_info after successful validation + self._discovery_info = None return self.async_update_reload_and_abort( entry=self._get_reconfigure_entry(), - data_updates=user_input, + data_updates=data_updates, ) return self.async_show_form( step_id="reconfigure", diff --git a/custom_components/blanco_unit/const.py b/custom_components/blanco_unit/const.py index d1f0d74..3e35d0a 100644 --- a/custom_components/blanco_unit/const.py +++ b/custom_components/blanco_unit/const.py @@ -8,7 +8,9 @@ CONF_MAC = "conf_mac" CONF_NAME = "conf_name" CONF_PIN = "conf_pin" +CONF_DEV_ID = "conf_dev_id" CONF_ERROR = "base" +RANDOM_MAC_PLACEHOLDER = "randomized:mac" BLE_CALLBACK = "unregister_ble_callback" # BLE Protocol Constants diff --git a/custom_components/blanco_unit/coordinator.py b/custom_components/blanco_unit/coordinator.py index 2a1d2ab..0acb6d6 100644 --- a/custom_components/blanco_unit/coordinator.py +++ b/custom_components/blanco_unit/coordinator.py @@ -21,7 +21,13 @@ from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed from .client import BlancoUnitAuthenticationError, BlancoUnitBluetoothClient -from .const import CONF_MAC, CONF_PIN, DOMAIN +from .const import ( + CHARACTERISTIC_UUID, + CONF_MAC, + CONF_PIN, + DOMAIN, + RANDOM_MAC_PLACEHOLDER, +) from .data import BlancoUnitData _LOGGER = logging.getLogger(__name__) @@ -49,6 +55,7 @@ def __init__( # Store setup data self.address = device.address self.mac_address = config_entry.data[CONF_MAC] + self._random_mac = self.mac_address == RANDOM_MAC_PLACEHOLDER # Create client self._client = BlancoUnitBluetoothClient( @@ -72,14 +79,23 @@ def __init__( self._unsub_unavailable_update_listener = bluetooth.async_track_unavailable( hass, self._unavailable_callback, self.address, connectable=True ) - self._unsub_available_update_listener = bluetooth.async_register_callback( - hass, - self._available_callback, - {"address": self.address, "connectable": True}, - BluetoothScanningMode.ACTIVE, - ) + if self._random_mac: + # For random MAC, listen for any device advertising our service UUID + self._unsub_available_update_listener = bluetooth.async_register_callback( + hass, + self._available_callback, + {"service_uuid": CHARACTERISTIC_UUID, "connectable": True}, + BluetoothScanningMode.ACTIVE, + ) + else: + self._unsub_available_update_listener = bluetooth.async_register_callback( + hass, + self._available_callback, + {"address": self.address, "connectable": True}, + BluetoothScanningMode.ACTIVE, + ) - _LOGGER.debug("Coordinator startup finished") + _LOGGER.debug("Coordinator startup finished (random_mac=%s)", self._random_mac) def _available_callback( self, info: BluetoothServiceInfoBleak, change: BluetoothChange @@ -186,7 +202,7 @@ def _connection_changed(self, connected: bool) -> None: async def test_protocol_parameters( self, evt_type: int, ctrl: int | None = None, pars: dict[str, Any] | None = None - ) -> dict[str, Any] | None: + ) -> dict[str, Any]: """Test protocol parameters by sending a custom event.""" return await self._call( self._client.test_protocol_parameters, evt_type, ctrl, pars @@ -208,6 +224,7 @@ async def _async_update_data(self) -> BlancoUnitData: connected=self._client.is_connected, available=True, device_id=self._client.device_id or "", + device_type=self._client.device_type, ) except BlancoUnitAuthenticationError as err: self._set_unavailable() @@ -263,9 +280,10 @@ async def _call(self, func: Callable[..., Any], *args: Any, **kwargs: Any) -> An ) from err def _set_unavailable(self) -> None: - _LOGGER.debug("_set_unavailable width data %s", str(self.data)) - # trigger rediscovery for the device - bluetooth.async_rediscover_address(self.hass, self.mac_address) + _LOGGER.debug("_set_unavailable with data %s", str(self.data)) + # trigger rediscovery for the device (only for static MAC) + if not self._random_mac: + bluetooth.async_rediscover_address(self.hass, self.mac_address) if self.data is None: # may be called before data is available return # tell HA to refresh all entities diff --git a/custom_components/blanco_unit/data.py b/custom_components/blanco_unit/data.py index fc6777b..abd27af 100644 --- a/custom_components/blanco_unit/data.py +++ b/custom_components/blanco_unit/data.py @@ -72,6 +72,7 @@ class BlancoUnitData: connected: bool available: bool device_id: str + device_type: int | None = None system_info: BlancoUnitSystemInfo | None = None settings: BlancoUnitSettings | None = None status: BlancoUnitStatus | None = None diff --git a/custom_components/blanco_unit/sensor.py b/custom_components/blanco_unit/sensor.py index 2ce2215..f0209da 100644 --- a/custom_components/blanco_unit/sensor.py +++ b/custom_components/blanco_unit/sensor.py @@ -44,6 +44,8 @@ async def async_setup_entry( FirmwareElecSensor(coordinator), DeviceNameSensor(coordinator), ResetCountSensor(coordinator), + DeviceTypeSensor(coordinator), + DeviceIdSensor(coordinator), # Identity sensors SerialNumberSensor(coordinator), ServiceCodeSensor(coordinator), @@ -334,6 +336,34 @@ def native_value(self) -> int | None: return self.coordinator.data.system_info.reset_cnt +class DeviceTypeSensor(BlancoUnitBaseEntity, SensorEntity): + """Sensor for device type.""" + + _attr_unique_id = "device_type" + _attr_translation_key = _attr_unique_id + _attr_icon = "mdi:information" + _attr_entity_category = EntityCategory.DIAGNOSTIC + + @property + def native_value(self) -> int | None: + """Return the device type.""" + return self.coordinator.data.device_type + + +class DeviceIdSensor(BlancoUnitBaseEntity, SensorEntity): + """Sensor for device id.""" + + _attr_unique_id = "device_id" + _attr_translation_key = _attr_unique_id + _attr_icon = "mdi:information" + _attr_entity_category = EntityCategory.DIAGNOSTIC + + @property + def native_value(self) -> int | None: + """Return the device id.""" + return self.coordinator.data.device_id + + # ------------------------------- # Identity Sensors # ------------------------------- diff --git a/custom_components/blanco_unit/services.py b/custom_components/blanco_unit/services.py index 05250c4..cae15b9 100644 --- a/custom_components/blanco_unit/services.py +++ b/custom_components/blanco_unit/services.py @@ -146,22 +146,7 @@ async def handle_scan_protocol(call: ServiceCall) -> dict: "response": response if response else None, } - if response: - _LOGGER.info( - "Found data: evt_type=%d, ctrl=%s, pars=%s", - evt_type, - ctrl, - pars, - ) - _LOGGER.info("Response: %s", json.dumps(response, indent=2)) - else: - _LOGGER.warning( - "No data: evt_type=%d, ctrl=%s, pars=%s", - evt_type, - ctrl, - pars, - ) - + _LOGGER.info("Response: %s", json.dumps(response, indent=2)) return result hass.services.async_register( diff --git a/custom_components/blanco_unit/translations/en.json b/custom_components/blanco_unit/translations/en.json index d219ea9..b9ecac6 100644 --- a/custom_components/blanco_unit/translations/en.json +++ b/custom_components/blanco_unit/translations/en.json @@ -175,6 +175,9 @@ "reset_count": { "name": "Reset Count" }, + "device_type": { + "name": "Device Type" + }, "serial_number": { "name": "Serial Number" }, diff --git a/test_cli.py b/test_cli.py index 6742a20..60b43a9 100755 --- a/test_cli.py +++ b/test_cli.py @@ -1,4 +1,6 @@ #!/usr/bin/env python3 +# ruff: noqa: PGH004 +# ruff: noqa """CLI tool for testing Blanco Unit Bluetooth client functionality.""" from __future__ import annotations diff --git a/tests/snapshots/test_sensor.ambr b/tests/snapshots/test_sensor.ambr index 78aa39b..6d08e4c 100644 --- a/tests/snapshots/test_sensor.ambr +++ b/tests/snapshots/test_sensor.ambr @@ -248,6 +248,55 @@ 'state': 'Test Device', }) # --- +# name: test_all_entities[sensor.test_blanco_unit_device_type-entry] + EntityRegistryEntrySnapshot({ + 'aliases': set({ + }), + '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.test_blanco_unit_device_type', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'options': dict({ + }), + 'original_device_class': None, + 'original_icon': 'mdi:information', + 'original_name': 'Device Type', + 'platform': 'blanco_unit', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': 0, + 'translation_key': 'device_type', + 'unique_id': 'device_type', + 'unit_of_measurement': None, + }) +# --- +# name: test_all_entities[sensor.test_blanco_unit_device_type-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + 'friendly_name': 'Test Blanco Unit Device Type', + 'icon': 'mdi:information', + }), + 'context': , + 'entity_id': 'sensor.test_blanco_unit_device_type', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': 'unknown', + }) +# --- # name: test_all_entities[sensor.test_blanco_unit_electronic_controller_firmware-entry] EntityRegistryEntrySnapshot({ 'aliases': set({ @@ -649,6 +698,55 @@ 'state': '192.168.1.1', }) # --- +# name: test_all_entities[sensor.test_blanco_unit_none-entry] + EntityRegistryEntrySnapshot({ + 'aliases': set({ + }), + '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.test_blanco_unit_none', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'options': dict({ + }), + 'original_device_class': None, + 'original_icon': 'mdi:information', + 'original_name': None, + 'platform': 'blanco_unit', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': 0, + 'translation_key': 'device_id', + 'unique_id': 'device_id', + 'unit_of_measurement': None, + }) +# --- +# name: test_all_entities[sensor.test_blanco_unit_none-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + 'friendly_name': 'Test Blanco Unit None', + 'icon': 'mdi:information', + }), + 'context': , + 'entity_id': 'sensor.test_blanco_unit_none', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': 'test_device_id', + }) +# --- # name: test_all_entities[sensor.test_blanco_unit_post_flush_quantity-entry] EntityRegistryEntrySnapshot({ 'aliases': set({ diff --git a/tests/test_binary_sensor.py b/tests/test_binary_sensor.py index b42ce2c..6aa6a91 100644 --- a/tests/test_binary_sensor.py +++ b/tests/test_binary_sensor.py @@ -1,242 +1,242 @@ -"""Tests for the Blanco Unit binary sensor entities.""" - -from unittest.mock import MagicMock, patch - -import pytest -from pytest_homeassistant_custom_component.common import ( - MockConfigEntry, - snapshot_platform, -) -from syrupy.assertion import SnapshotAssertion - -from custom_components.blanco_unit.binary_sensor import ( - CloudConnectBinarySensor, - ConnectionBinarySensor, - FirmwareUpdateBinarySensor, - WaterDispensingBinarySensor, - async_setup_entry, -) -from custom_components.blanco_unit.const import CONF_MAC, CONF_NAME, CONF_PIN, DOMAIN -from custom_components.blanco_unit.data import ( - BlancoUnitData, - BlancoUnitStatus, - BlancoUnitWifiInfo, -) -from homeassistant.const import Platform -from homeassistant.core import HomeAssistant -from homeassistant.helpers import entity_registry as er - -from .conftest import setup_integration # noqa: TID251 - - -@pytest.fixture -def mock_coordinator(): - """Create a mock coordinator.""" - coordinator = MagicMock() - coordinator.data = BlancoUnitData( - connected=True, - available=True, - device_id="test_device_id", - status=BlancoUnitStatus( - tap_state=1, - filter_rest=85, - co2_rest=90, - wtr_disp_active=True, - firm_upd_avlb=True, - set_point_cooling=7, - clean_mode_state=0, - err_bits=0, - ), - wifi_info=BlancoUnitWifiInfo( - cloud_connect=True, - ssid="TestSSID", - signal=-50, - ip="192.168.1.100", - ble_mac="AA:BB:CC:DD:EE:FF", - wifi_mac="11:22:33:44:55:66", - gateway="192.168.1.1", - gateway_mac="AA:BB:CC:DD:EE:00", - subnet="255.255.255.0", - ), - ) - return coordinator - - -@pytest.fixture -def mock_config_entry(): - """Create a mock config entry.""" - entry = MagicMock() - entry.entry_id = "test_entry_id" - return entry - - -async def test_all_entities( - hass: HomeAssistant, - snapshot: SnapshotAssertion, - entity_registry: er.EntityRegistry, - mock_blanco_unit_data, - mock_bluetooth_device, -) -> None: - """Test all entities.""" - mock_config_entry = MockConfigEntry( - domain=DOMAIN, - data={ - CONF_MAC: "AA:BB:CC:DD:EE:FF", - CONF_NAME: "Test Blanco Unit", - CONF_PIN: 12345, - }, - unique_id="AA:BB:CC:DD:EE:FF", - ) - - with ( - patch("custom_components.blanco_unit.PLATFORMS", [Platform.BINARY_SENSOR]), - patch( - "custom_components.blanco_unit.bluetooth.async_ble_device_from_address", - return_value=mock_bluetooth_device, - ), - ): - await setup_integration(hass, mock_config_entry) - - await snapshot_platform( - hass, entity_registry, snapshot, mock_config_entry.entry_id - ) - - -async def test_async_setup_entry( - hass: HomeAssistant, mock_config_entry, mock_coordinator -) -> None: - """Test async_setup_entry creates all binary sensors.""" - mock_config_entry.runtime_data = mock_coordinator - entities_added = [] - - def mock_add_entities(entities): - entities_added.extend(entities) - - await async_setup_entry(hass, mock_config_entry, mock_add_entities) - - # Verify all 4 binary sensors were added - assert len(entities_added) == 4 - - # Verify sensor types - sensor_types = [type(entity).__name__ for entity in entities_added] - assert "ConnectionBinarySensor" in sensor_types - assert "WaterDispensingBinarySensor" in sensor_types - assert "FirmwareUpdateBinarySensor" in sensor_types - assert "CloudConnectBinarySensor" in sensor_types - - -async def test_connection_binary_sensor(mock_coordinator) -> None: - """Test ConnectionBinarySensor.""" - sensor = ConnectionBinarySensor(mock_coordinator) - - assert sensor.is_on is True - assert sensor.unique_id == "connection" - assert sensor.icon == "mdi:bluetooth-connect" - - -async def test_connection_binary_sensor_disconnected(mock_coordinator) -> None: - """Test ConnectionBinarySensor when disconnected.""" - mock_coordinator.data.connected = False - sensor = ConnectionBinarySensor(mock_coordinator) - - assert sensor.is_on is False - assert sensor.icon == "mdi:bluetooth-off" - - -async def test_water_dispensing_binary_sensor(mock_coordinator) -> None: - """Test WaterDispensingBinarySensor.""" - sensor = WaterDispensingBinarySensor(mock_coordinator) - - assert sensor.available is True - assert sensor.is_on is True - assert sensor.unique_id == "water_dispensing" - - -async def test_water_dispensing_binary_sensor_not_active(mock_coordinator) -> None: - """Test WaterDispensingBinarySensor when not active.""" - mock_coordinator.data.status.wtr_disp_active = False - sensor = WaterDispensingBinarySensor(mock_coordinator) - - assert sensor.is_on is False - - -async def test_water_dispensing_binary_sensor_unavailable(mock_coordinator) -> None: - """Test WaterDispensingBinarySensor when status is None.""" - mock_coordinator.data.status = None - sensor = WaterDispensingBinarySensor(mock_coordinator) - - assert sensor.available is False - assert sensor.is_on is None - - -async def test_firmware_update_binary_sensor(mock_coordinator) -> None: - """Test FirmwareUpdateBinarySensor.""" - sensor = FirmwareUpdateBinarySensor(mock_coordinator) - - assert sensor.available is True - assert sensor.is_on is True - assert sensor.unique_id == "firmware_update" - - -async def test_firmware_update_binary_sensor_not_available(mock_coordinator) -> None: - """Test FirmwareUpdateBinarySensor when update not available.""" - mock_coordinator.data.status.firm_upd_avlb = False - sensor = FirmwareUpdateBinarySensor(mock_coordinator) - - assert sensor.is_on is False - - -async def test_firmware_update_binary_sensor_unavailable(mock_coordinator) -> None: - """Test FirmwareUpdateBinarySensor when status is None.""" - mock_coordinator.data.status = None - sensor = FirmwareUpdateBinarySensor(mock_coordinator) - - assert sensor.available is False - assert sensor.is_on is None - - -async def test_cloud_connect_binary_sensor(mock_coordinator) -> None: - """Test CloudConnectBinarySensor.""" - sensor = CloudConnectBinarySensor(mock_coordinator) - - assert sensor.available is True - assert sensor.is_on is True - assert sensor.unique_id == "cloud_connection" - - -async def test_cloud_connect_binary_sensor_disconnected(mock_coordinator) -> None: - """Test CloudConnectBinarySensor when not connected to cloud.""" - mock_coordinator.data.wifi_info.cloud_connect = False - sensor = CloudConnectBinarySensor(mock_coordinator) - - assert sensor.is_on is False - - -async def test_cloud_connect_binary_sensor_unavailable(mock_coordinator) -> None: - """Test CloudConnectBinarySensor when wifi_info is None.""" - mock_coordinator.data.wifi_info = None - sensor = CloudConnectBinarySensor(mock_coordinator) - - assert sensor.available is False - assert sensor.is_on is None - - -async def test_binary_sensor_when_data_unavailable(mock_coordinator) -> None: - """Test binary sensor when data is unavailable.""" - mock_coordinator.data.available = False - sensor = ConnectionBinarySensor(mock_coordinator) - - assert sensor.available is False - - -async def test_connection_binary_sensor_icon_property(mock_coordinator) -> None: - """Test ConnectionBinarySensor icon property changes.""" - sensor = ConnectionBinarySensor(mock_coordinator) - - # Connected - mock_coordinator.data.connected = True - assert sensor.icon == "mdi:bluetooth-connect" - - # Disconnected - mock_coordinator.data.connected = False - assert sensor.icon == "mdi:bluetooth-off" +"""Tests for the Blanco Unit binary sensor entities.""" + +from unittest.mock import MagicMock, patch + +import pytest +from pytest_homeassistant_custom_component.common import ( + MockConfigEntry, + snapshot_platform, +) +from syrupy.assertion import SnapshotAssertion + +from custom_components.blanco_unit.binary_sensor import ( + CloudConnectBinarySensor, + ConnectionBinarySensor, + FirmwareUpdateBinarySensor, + WaterDispensingBinarySensor, + async_setup_entry, +) +from custom_components.blanco_unit.const import CONF_MAC, CONF_NAME, CONF_PIN, DOMAIN +from custom_components.blanco_unit.data import ( + BlancoUnitData, + BlancoUnitStatus, + BlancoUnitWifiInfo, +) +from homeassistant.const import Platform +from homeassistant.core import HomeAssistant +from homeassistant.helpers import entity_registry as er + +from .conftest import setup_integration # noqa: TID251 + + +@pytest.fixture +def mock_coordinator(): + """Create a mock coordinator.""" + coordinator = MagicMock() + coordinator.data = BlancoUnitData( + connected=True, + available=True, + device_id="test_device_id", + status=BlancoUnitStatus( + tap_state=1, + filter_rest=85, + co2_rest=90, + wtr_disp_active=True, + firm_upd_avlb=True, + set_point_cooling=7, + clean_mode_state=0, + err_bits=0, + ), + wifi_info=BlancoUnitWifiInfo( + cloud_connect=True, + ssid="TestSSID", + signal=-50, + ip="192.168.1.100", + ble_mac="AA:BB:CC:DD:EE:FF", + wifi_mac="11:22:33:44:55:66", + gateway="192.168.1.1", + gateway_mac="AA:BB:CC:DD:EE:00", + subnet="255.255.255.0", + ), + ) + return coordinator + + +@pytest.fixture +def mock_config_entry(): + """Create a mock config entry.""" + entry = MagicMock() + entry.entry_id = "test_entry_id" + return entry + + +async def test_all_entities( + hass: HomeAssistant, + snapshot: SnapshotAssertion, + entity_registry: er.EntityRegistry, + mock_blanco_unit_data, + mock_bluetooth_device, +) -> None: + """Test all entities.""" + mock_config_entry = MockConfigEntry( + domain=DOMAIN, + data={ + CONF_MAC: "AA:BB:CC:DD:EE:FF", + CONF_NAME: "Test Blanco Unit", + CONF_PIN: 12345, + }, + unique_id="AA:BB:CC:DD:EE:FF", + ) + + with ( + patch("custom_components.blanco_unit.PLATFORMS", [Platform.BINARY_SENSOR]), + patch( + "custom_components.blanco_unit.bluetooth.async_ble_device_from_address", + return_value=mock_bluetooth_device, + ), + ): + await setup_integration(hass, mock_config_entry) + + await snapshot_platform( + hass, entity_registry, snapshot, mock_config_entry.entry_id + ) + + +async def test_async_setup_entry( + hass: HomeAssistant, mock_config_entry, mock_coordinator +) -> None: + """Test async_setup_entry creates all binary sensors.""" + mock_config_entry.runtime_data = mock_coordinator + entities_added = [] + + def mock_add_entities(entities): + entities_added.extend(entities) + + await async_setup_entry(hass, mock_config_entry, mock_add_entities) + + # Verify all 4 binary sensors were added + assert len(entities_added) == 4 + + # Verify sensor types + sensor_types = [type(entity).__name__ for entity in entities_added] + assert "ConnectionBinarySensor" in sensor_types + assert "WaterDispensingBinarySensor" in sensor_types + assert "FirmwareUpdateBinarySensor" in sensor_types + assert "CloudConnectBinarySensor" in sensor_types + + +async def test_connection_binary_sensor(mock_coordinator) -> None: + """Test ConnectionBinarySensor.""" + sensor = ConnectionBinarySensor(mock_coordinator) + + assert sensor.is_on is True + assert sensor.unique_id == "connection" + assert sensor.icon == "mdi:bluetooth-connect" + + +async def test_connection_binary_sensor_disconnected(mock_coordinator) -> None: + """Test ConnectionBinarySensor when disconnected.""" + mock_coordinator.data.connected = False + sensor = ConnectionBinarySensor(mock_coordinator) + + assert sensor.is_on is False + assert sensor.icon == "mdi:bluetooth-off" + + +async def test_water_dispensing_binary_sensor(mock_coordinator) -> None: + """Test WaterDispensingBinarySensor.""" + sensor = WaterDispensingBinarySensor(mock_coordinator) + + assert sensor.available is True + assert sensor.is_on is True + assert sensor.unique_id == "water_dispensing" + + +async def test_water_dispensing_binary_sensor_not_active(mock_coordinator) -> None: + """Test WaterDispensingBinarySensor when not active.""" + mock_coordinator.data.status.wtr_disp_active = False + sensor = WaterDispensingBinarySensor(mock_coordinator) + + assert sensor.is_on is False + + +async def test_water_dispensing_binary_sensor_unavailable(mock_coordinator) -> None: + """Test WaterDispensingBinarySensor when status is None.""" + mock_coordinator.data.status = None + sensor = WaterDispensingBinarySensor(mock_coordinator) + + assert sensor.available is False + assert sensor.is_on is None + + +async def test_firmware_update_binary_sensor(mock_coordinator) -> None: + """Test FirmwareUpdateBinarySensor.""" + sensor = FirmwareUpdateBinarySensor(mock_coordinator) + + assert sensor.available is True + assert sensor.is_on is True + assert sensor.unique_id == "firmware_update" + + +async def test_firmware_update_binary_sensor_not_available(mock_coordinator) -> None: + """Test FirmwareUpdateBinarySensor when update not available.""" + mock_coordinator.data.status.firm_upd_avlb = False + sensor = FirmwareUpdateBinarySensor(mock_coordinator) + + assert sensor.is_on is False + + +async def test_firmware_update_binary_sensor_unavailable(mock_coordinator) -> None: + """Test FirmwareUpdateBinarySensor when status is None.""" + mock_coordinator.data.status = None + sensor = FirmwareUpdateBinarySensor(mock_coordinator) + + assert sensor.available is False + assert sensor.is_on is None + + +async def test_cloud_connect_binary_sensor(mock_coordinator) -> None: + """Test CloudConnectBinarySensor.""" + sensor = CloudConnectBinarySensor(mock_coordinator) + + assert sensor.available is True + assert sensor.is_on is True + assert sensor.unique_id == "cloud_connection" + + +async def test_cloud_connect_binary_sensor_disconnected(mock_coordinator) -> None: + """Test CloudConnectBinarySensor when not connected to cloud.""" + mock_coordinator.data.wifi_info.cloud_connect = False + sensor = CloudConnectBinarySensor(mock_coordinator) + + assert sensor.is_on is False + + +async def test_cloud_connect_binary_sensor_unavailable(mock_coordinator) -> None: + """Test CloudConnectBinarySensor when wifi_info is None.""" + mock_coordinator.data.wifi_info = None + sensor = CloudConnectBinarySensor(mock_coordinator) + + assert sensor.available is False + assert sensor.is_on is None + + +async def test_binary_sensor_when_data_unavailable(mock_coordinator) -> None: + """Test binary sensor when data is unavailable.""" + mock_coordinator.data.available = False + sensor = ConnectionBinarySensor(mock_coordinator) + + assert sensor.available is False + + +async def test_connection_binary_sensor_icon_property(mock_coordinator) -> None: + """Test ConnectionBinarySensor icon property changes.""" + sensor = ConnectionBinarySensor(mock_coordinator) + + # Connected + mock_coordinator.data.connected = True + assert sensor.icon == "mdi:bluetooth-connect" + + # Disconnected + mock_coordinator.data.connected = False + assert sensor.icon == "mdi:bluetooth-off" diff --git a/tests/test_client.py b/tests/test_client.py index 7dbfd3b..ba860ad 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -13,6 +13,7 @@ BlancoUnitBluetoothClient, BlancoUnitClientError, BlancoUnitConnectionError, + PinValidationResult, _BlancoUnitProtocol, _ChangePinPars, _DispensePars, @@ -446,7 +447,9 @@ async def test_protocol_send_pairing_request(): mock_client = AsyncMock() # Mock response - response_data = {"body": {"results": [{"pars": {"dev_id": "device123"}}]}} + response_data = { + "body": {"results": [{"pars": {"dev_id": "device123", "dev_type": 1}}]} + } json_str = json.dumps(response_data) response_packet = ( bytes([0xFF, 0x00, 1, 10, 0x00]) + json_str.encode("utf-8") + b"\x00\xff" @@ -478,7 +481,7 @@ async def test_protocol_send_request_with_ctrl(): mock_client.read_gatt_char = AsyncMock(return_value=response_packet) result = await protocol.send_request( - mock_client, "12345", "device123", evt_type=1, ctrl=2 + mock_client, "12345", "device123", dev_type=1, evt_type=1, ctrl=2 ) assert result["body"]["results"][0]["pars"]["status"] == "ok" @@ -502,7 +505,13 @@ async def test_protocol_send_request_without_ctrl(): mock_client.read_gatt_char = AsyncMock(return_value=response_packet) result = await protocol.send_request( - mock_client, "12345", "device123", evt_type=1, ctrl=None, pars={"test": "data"} + mock_client, + "12345", + "device123", + 1, + evt_type=1, + ctrl=None, + pars={"test": "data"}, ) assert result["body"]["results"][0]["pars"]["status"] == "ok" @@ -539,7 +548,10 @@ async def test_validate_pin_success_with_dev_id(): # Mock successful pairing response response_data = { - "body": {"results": [{"pars": {}}], "meta": {"dev_id": "device123"}} + "body": { + "results": [{"pars": {}}], + "meta": {"dev_id": "device123", "dev_type": 1}, + } } json_str = json.dumps(response_data) response_packet = ( @@ -549,10 +561,11 @@ async def test_validate_pin_success_with_dev_id(): mock_client.write_gatt_char = AsyncMock() mock_client.read_gatt_char = AsyncMock(return_value=response_packet) - is_valid, response = await validate_pin(mock_client, "12345") + validation = await validate_pin(mock_client, "12345") - assert is_valid is True - assert response["body"]["meta"]["dev_id"] == "device123" + assert validation.is_valid is True + assert validation.dev_id == "device123" + assert validation.dev_type == 1 @pytest.mark.asyncio @@ -570,9 +583,9 @@ async def test_validate_pin_wrong_pin_error_code(): mock_client.write_gatt_char = AsyncMock() mock_client.read_gatt_char = AsyncMock(return_value=response_packet) - is_valid, response = await validate_pin(mock_client, "99999") + validation = await validate_pin(mock_client, "99999") - assert is_valid is False + assert validation.is_valid is False @pytest.mark.asyncio @@ -590,9 +603,9 @@ async def test_validate_pin_no_device_id(): mock_client.write_gatt_char = AsyncMock() mock_client.read_gatt_char = AsyncMock(return_value=response_packet) - is_valid, response = await validate_pin(mock_client, "12345") + validation = await validate_pin(mock_client, "12345") - assert is_valid is False + assert validation.is_valid is False @pytest.mark.asyncio @@ -621,7 +634,10 @@ async def test_validate_pin_with_provided_protocol(): # Mock successful pairing response response_data = { - "body": {"results": [{"pars": {}}], "meta": {"dev_id": "device789"}} + "body": { + "results": [{"pars": {}}], + "meta": {"dev_id": "device789", "dev_type": 2}, + } } json_str = json.dumps(response_data) response_packet = ( @@ -631,10 +647,11 @@ async def test_validate_pin_with_provided_protocol(): mock_client.write_gatt_char = AsyncMock() mock_client.read_gatt_char = AsyncMock(return_value=response_packet) - is_valid, response = await validate_pin(mock_client, "12345", protocol=protocol) + validation = await validate_pin(mock_client, "12345", protocol=protocol) - assert is_valid is True - assert response["body"]["meta"]["dev_id"] == "device789" + assert validation.is_valid is True + assert validation.dev_id == "device789" + assert validation.dev_type == 2 # ------------------------------- @@ -706,7 +723,7 @@ def test_bluetooth_client_device_id_when_connected(): mock_client = AsyncMock() mock_protocol = MagicMock() client._session_data = _BlancoUnitSessionData( - client=mock_client, dev_id="device123", protocol=mock_protocol + client=mock_client, dev_id="device123", dev_type=1, protocol=mock_protocol ) assert client.device_id == "device123" @@ -740,7 +757,7 @@ def test_bluetooth_client_is_connected_when_connected(): mock_client.is_connected = True mock_protocol = MagicMock() client._session_data = _BlancoUnitSessionData( - client=mock_client, dev_id="device123", protocol=mock_protocol + client=mock_client, dev_id="device123", dev_type=1, protocol=mock_protocol ) assert client.is_connected is True @@ -762,7 +779,7 @@ async def test_bluetooth_client_disconnect_when_connected(): mock_client = AsyncMock() mock_protocol = MagicMock() client._session_data = _BlancoUnitSessionData( - client=mock_client, dev_id="device123", protocol=mock_protocol + client=mock_client, dev_id="device123", dev_type=1, protocol=mock_protocol ) await client.disconnect() @@ -801,7 +818,7 @@ async def test_bluetooth_client_connect_first_time(mock_establish): mock_establish.return_value = mock_ble_client # Mock pairing response - response_data = {"body": {"meta": {"dev_id": "device123"}}} + response_data = {"body": {"meta": {"dev_id": "device123", "dev_type": 1}}} json_str = json.dumps(response_data) response_packet = ( bytes([0xFF, 0x00, 1, 10, 0x00]) + json_str.encode("utf-8") + b"\x00\xff" @@ -835,7 +852,7 @@ async def test_bluetooth_client_connect_already_connected(mock_establish): mock_ble_client = AsyncMock() mock_protocol = MagicMock() existing_session = _BlancoUnitSessionData( - client=mock_ble_client, dev_id="device123", protocol=mock_protocol + client=mock_ble_client, dev_id="device123", dev_type=1, protocol=mock_protocol ) client._session_data = existing_session @@ -861,7 +878,7 @@ def test_bluetooth_client_handle_disconnect(): mock_ble_client = AsyncMock() mock_protocol = MagicMock() client._session_data = _BlancoUnitSessionData( - client=mock_ble_client, dev_id="device123", protocol=mock_protocol + client=mock_ble_client, dev_id="device123", dev_type=1, protocol=mock_protocol ) # Trigger disconnect @@ -886,7 +903,10 @@ async def test_bluetooth_client_perform_pairing_success(): # Mock successful pairing response response_data = { - "body": {"results": [{"pars": {}}], "meta": {"dev_id": "device456"}} + "body": { + "results": [{"pars": {}}], + "meta": {"dev_id": "device456", "dev_type": 1}, + } } json_str = json.dumps(response_data) response_packet = ( @@ -896,9 +916,11 @@ async def test_bluetooth_client_perform_pairing_success(): mock_ble_client.write_gatt_char = AsyncMock() mock_ble_client.read_gatt_char = AsyncMock(return_value=response_packet) - dev_id = await client._perform_pairing(mock_ble_client, mock_protocol) + result = await client._perform_pairing(mock_ble_client, mock_protocol) - assert dev_id == "device456" + assert result.is_valid is True + assert result.dev_id == "device456" + assert result.dev_type == 1 @pytest.mark.asyncio @@ -943,10 +965,11 @@ async def test_bluetooth_client_perform_pairing_no_device_id(mock_validate_pin): mock_protocol = _BlancoUnitProtocol() # Mock validate_pin to return True but with a response that has no device ID - response_data = {"body": {"results": [{"pars": {}}], "meta": {}}} - mock_validate_pin.return_value = (True, response_data) + mock_validate_pin.return_value = PinValidationResult(True, None, 1) - with pytest.raises(BlancoUnitConnectionError, match="No device ID"): + with pytest.raises( + BlancoUnitConnectionError, match="No device ID in pairing response" + ): await client._perform_pairing(mock_ble_client, mock_protocol) @@ -967,7 +990,7 @@ async def test_bluetooth_client_execute_transaction_success(mock_establish): mock_establish.return_value = mock_ble_client # Mock pairing response - pairing_data = {"body": {"meta": {"dev_id": "device123"}}} + pairing_data = {"body": {"meta": {"dev_id": "device123", "dev_type": 1}}} pairing_json = json.dumps(pairing_data) pairing_packet = ( bytes([0xFF, 0x00, 1, 10, 0x00]) + pairing_json.encode("utf-8") + b"\x00\xff" @@ -1019,7 +1042,7 @@ async def test_bluetooth_client_execute_transaction_auth_error(mock_establish): mock_establish.return_value = mock_ble_client # Mock pairing response - pairing_data = {"body": {"meta": {"dev_id": "device123"}}} + pairing_data = {"body": {"meta": {"dev_id": "device123", "dev_type": 1}}} pairing_json = json.dumps(pairing_data) pairing_packet = ( bytes([0xFF, 0x00, 1, 10, 0x00]) + pairing_json.encode("utf-8") + b"\x00\xff" @@ -1070,7 +1093,7 @@ async def test_bluetooth_client_get_system_info(mock_establish): mock_establish.return_value = mock_ble_client # Mock responses - pairing_data = {"body": {"meta": {"dev_id": "device123"}}} + pairing_data = {"body": {"meta": {"dev_id": "device123", "dev_type": 1}}} pairing_json = json.dumps(pairing_data) pairing_packet = ( bytes([0xFF, 0x00, 1, 10, 0x00]) + pairing_json.encode("utf-8") + b"\x00\xff" @@ -1134,7 +1157,7 @@ async def test_bluetooth_client_get_settings(mock_establish): mock_establish.return_value = mock_ble_client # Mock responses - pairing_data = {"body": {"meta": {"dev_id": "device123"}}} + pairing_data = {"body": {"meta": {"dev_id": "device123", "dev_type": 1}}} pairing_json = json.dumps(pairing_data) pairing_packet = ( bytes([0xFF, 0x00, 1, 10, 0x00]) + pairing_json.encode("utf-8") + b"\x00\xff" @@ -1200,7 +1223,7 @@ async def test_bluetooth_client_get_status(mock_establish): mock_establish.return_value = mock_ble_client # Mock responses - pairing_data = {"body": {"meta": {"dev_id": "device123"}}} + pairing_data = {"body": {"meta": {"dev_id": "device123", "dev_type": 1}}} pairing_json = json.dumps(pairing_data) pairing_packet = ( bytes([0xFF, 0x00, 1, 10, 0x00]) + pairing_json.encode("utf-8") + b"\x00\xff" @@ -1267,7 +1290,7 @@ async def test_bluetooth_client_get_device_identity(mock_establish): mock_establish.return_value = mock_ble_client # Mock responses - pairing_data = {"body": {"meta": {"dev_id": "device123"}}} + pairing_data = {"body": {"meta": {"dev_id": "device123", "dev_type": 1}}} pairing_json = json.dumps(pairing_data) pairing_packet = ( bytes([0xFF, 0x00, 1, 10, 0x00]) + pairing_json.encode("utf-8") + b"\x00\xff" @@ -1316,7 +1339,7 @@ async def test_bluetooth_client_get_wifi_info(mock_establish): mock_establish.return_value = mock_ble_client # Mock responses - pairing_data = {"body": {"meta": {"dev_id": "device123"}}} + pairing_data = {"body": {"meta": {"dev_id": "device123", "dev_type": 1}}} pairing_json = json.dumps(pairing_data) pairing_packet = ( bytes([0xFF, 0x00, 1, 10, 0x00]) + pairing_json.encode("utf-8") + b"\x00\xff" @@ -1383,7 +1406,7 @@ async def test_bluetooth_client_set_temperature_success(mock_establish): mock_establish.return_value = mock_ble_client # Mock responses - pairing_data = {"body": {"meta": {"dev_id": "device123"}}} + pairing_data = {"body": {"meta": {"dev_id": "device123", "dev_type": 1}}} pairing_json = json.dumps(pairing_data) pairing_packet = ( bytes([0xFF, 0x00, 1, 10, 0x00]) + pairing_json.encode("utf-8") + b"\x00\xff" @@ -1457,7 +1480,7 @@ async def test_bluetooth_client_set_water_hardness_success(mock_establish): mock_establish.return_value = mock_ble_client # Mock responses - pairing_data = {"body": {"meta": {"dev_id": "device123"}}} + pairing_data = {"body": {"meta": {"dev_id": "device123", "dev_type": 1}}} pairing_json = json.dumps(pairing_data) pairing_packet = ( bytes([0xFF, 0x00, 1, 10, 0x00]) + pairing_json.encode("utf-8") + b"\x00\xff" @@ -1503,7 +1526,7 @@ async def test_bluetooth_client_change_pin_success(mock_establish): mock_establish.return_value = mock_ble_client # Mock responses - pairing_data = {"body": {"meta": {"dev_id": "device123"}}} + pairing_data = {"body": {"meta": {"dev_id": "device123", "dev_type": 1}}} pairing_json = json.dumps(pairing_data) pairing_packet = ( bytes([0xFF, 0x00, 1, 10, 0x00]) + pairing_json.encode("utf-8") + b"\x00\xff" @@ -1550,7 +1573,7 @@ async def test_bluetooth_client_change_pin_failure(mock_establish): mock_establish.return_value = mock_ble_client # Mock responses - pairing_data = {"body": {"meta": {"dev_id": "device123"}}} + pairing_data = {"body": {"meta": {"dev_id": "device123", "dev_type": 1}}} pairing_json = json.dumps(pairing_data) pairing_packet = ( bytes([0xFF, 0x00, 1, 10, 0x00]) + pairing_json.encode("utf-8") + b"\x00\xff" @@ -1597,7 +1620,7 @@ async def test_bluetooth_client_dispense_water_success(mock_establish): mock_establish.return_value = mock_ble_client # Mock responses - pairing_data = {"body": {"meta": {"dev_id": "device123"}}} + pairing_data = {"body": {"meta": {"dev_id": "device123", "dev_type": 1}}} pairing_json = json.dumps(pairing_data) pairing_packet = ( bytes([0xFF, 0x00, 1, 10, 0x00]) + pairing_json.encode("utf-8") + b"\x00\xff" @@ -1699,7 +1722,7 @@ async def test_bluetooth_client_set_calibration_still(mock_establish): mock_establish.return_value = mock_ble_client # Mock responses - pairing_data = {"body": {"meta": {"dev_id": "device123"}}} + pairing_data = {"body": {"meta": {"dev_id": "device123", "dev_type": 1}}} pairing_json = json.dumps(pairing_data) pairing_packet = ( bytes([0xFF, 0x00, 1, 10, 0x00]) + pairing_json.encode("utf-8") + b"\x00\xff" @@ -1745,7 +1768,7 @@ async def test_bluetooth_client_set_calibration_soda(mock_establish): mock_establish.return_value = mock_ble_client # Mock responses - pairing_data = {"body": {"meta": {"dev_id": "device123"}}} + pairing_data = {"body": {"meta": {"dev_id": "device123", "dev_type": 1}}} pairing_json = json.dumps(pairing_data) pairing_packet = ( bytes([0xFF, 0x00, 1, 10, 0x00]) + pairing_json.encode("utf-8") + b"\x00\xff" diff --git a/tests/test_config_flow.py b/tests/test_config_flow.py index 403ec40..a3a0a8f 100644 --- a/tests/test_config_flow.py +++ b/tests/test_config_flow.py @@ -1,11 +1,12 @@ """Tests for the Blanco Unit config flow.""" from typing import Any -from unittest.mock import AsyncMock, patch +from unittest.mock import AsyncMock, MagicMock, patch import pytest from pytest_homeassistant_custom_component.common import MockConfigEntry +from custom_components.blanco_unit.client import PinValidationResult from custom_components.blanco_unit.const import ( CONF_ERROR, CONF_MAC, @@ -25,6 +26,7 @@ MOCKED_CONF_MAC = "AA:BB:CC:DD:EE:FF" MOCKED_CONF_NAME = "Test Blanco Unit" MOCKED_CONF_PIN = "01234" +MOCKED_CONF_DEV_ID = "test_device_id" MOCKED_CONFIG: dict[str, Any] = { CONF_MAC: MOCKED_CONF_MAC, @@ -81,7 +83,7 @@ async def test_user_flow_success( """Test successful user configuration flow.""" mock_device_from_address.return_value = mock_bluetooth_device mock_establish_connection.return_value = mock_bleak_client - mock_validate_pin.return_value = (True, None) + mock_validate_pin.return_value = PinValidationResult(True, "test_device_id", 2) # Initialize flow flow_result = await hass.config_entries.flow.async_init( @@ -116,7 +118,7 @@ async def test_user_flow_already_configured(hass: HomeAssistant) -> None: with ( patch( "custom_components.blanco_unit.config_flow.validate_pin", - return_value=(True, None), + return_value=PinValidationResult(True, MOCKED_CONF_DEV_ID, 2), ), patch( "custom_components.blanco_unit.config_flow.establish_connection", @@ -209,7 +211,7 @@ async def test_user_flow_invalid_authentication( """Test user flow with invalid authentication.""" mock_device_from_address.return_value = mock_bluetooth_device mock_establish_connection.return_value = mock_bleak_client - mock_validate_pin.return_value = (False, None) + mock_validate_pin.return_value = PinValidationResult(False, "devId", 2) flow_result = await hass.config_entries.flow.async_init( DOMAIN, context={"source": SOURCE_USER} @@ -310,7 +312,7 @@ async def test_reauth_flow_success( """Test successful reauth flow.""" mock_device_from_address.return_value = mock_bluetooth_device mock_establish_connection.return_value = mock_bleak_client - mock_validate_pin.return_value = (True, None) + mock_validate_pin.return_value = PinValidationResult(True, MOCKED_CONF_DEV_ID, 2) entry = MockConfigEntry( domain=DOMAIN, @@ -345,17 +347,24 @@ async def test_reauth_flow_wrong_device( mock_establish_connection: AsyncMock, mock_validate_pin: AsyncMock, hass: HomeAssistant, - mock_bluetooth_device, mock_bleak_client, ) -> None: - """Test reauth flow with wrong device MAC.""" - mock_device_from_address.return_value = mock_bluetooth_device + """Test reauth flow with wrong device (random MAC, dev_id mismatch).""" + # Create a mock device with random MAC so dev_id is used as unique_id + mock_device = AsyncMock() + mock_device.address = MOCKED_CONF_MAC + mock_device.name = MOCKED_CONF_NAME + mock_device.details = MagicMock() + mock_device.details.address_type = "random" + + mock_device_from_address.return_value = mock_device mock_establish_connection.return_value = mock_bleak_client - mock_validate_pin.return_value = (True, None) + # Return a different dev_id than the one stored in the entry + mock_validate_pin.return_value = PinValidationResult(True, "different_dev_id", 2) entry = MockConfigEntry( domain=DOMAIN, - unique_id=MOCKED_CONF_MAC, + unique_id=MOCKED_CONF_DEV_ID, data=MOCKED_CONFIG, ) entry.add_to_hass(hass) @@ -367,7 +376,7 @@ async def test_reauth_flow_wrong_device( configure_result = await hass.config_entries.flow.async_configure( flow_result["flow_id"], - {**MOCKED_CONFIG, CONF_MAC: "11:22:33:44:55:66"}, + MOCKED_CONFIG, ) assert configure_result["type"] is FlowResultType.ABORT @@ -395,7 +404,7 @@ async def test_reconfigure_flow_success( """Test successful reconfigure flow.""" mock_device_from_address.return_value = mock_bluetooth_device mock_establish_connection.return_value = mock_bleak_client - mock_validate_pin.return_value = (True, None) + mock_validate_pin.return_value = PinValidationResult(True, MOCKED_CONF_DEV_ID, 2) entry = MockConfigEntry( domain=DOMAIN, @@ -443,7 +452,7 @@ async def test_validate_input_success( mock_device_from_address.return_value = mock_bluetooth_device mock_establish_connection.return_value = mock_bleak_client - mock_validate_pin.return_value = (True, None) + mock_validate_pin.return_value = PinValidationResult(True, "test_device_id", 2) flow = BlancoUnitConfigFlow() flow.hass = hass diff --git a/tests/test_init.py b/tests/test_init.py index 67c653e..01c662b 100644 --- a/tests/test_init.py +++ b/tests/test_init.py @@ -5,12 +5,24 @@ import pytest from custom_components.blanco_unit import ( + _find_device_by_scanning, + _is_random_mac, + _register_retry_callback, async_reload_entry, async_setup, async_setup_entry, async_unload_entry, ) -from custom_components.blanco_unit.const import BLE_CALLBACK, CONF_MAC, DOMAIN +from custom_components.blanco_unit.client import PinValidationResult +from custom_components.blanco_unit.const import ( + BLE_CALLBACK, + CHARACTERISTIC_UUID, + CONF_DEV_ID, + CONF_MAC, + CONF_PIN, + DOMAIN, + RANDOM_MAC_PLACEHOLDER, +) from homeassistant.components.bluetooth import ( BluetoothChange, BluetoothServiceInfoBleak, @@ -434,3 +446,473 @@ async def test_async_unload_entry_no_entry_data(hass: HomeAssistant) -> None: assert result is True mock_coordinator.unload.assert_called_once() mock_rediscover.assert_called_once() + + +# ------------------------------- +# Random MAC Tests +# ------------------------------- + + +def _make_discovered_device( + address: str, rssi: int = -50 +) -> tuple[MagicMock, MagicMock]: + """Create a mock (BLEDevice, AdvertisementData) tuple for scanner results.""" + device = MagicMock() + device.address = address + device.name = f"Device_{address}" + + adv_data = MagicMock() + adv_data.service_uuids = [CHARACTERISTIC_UUID] + adv_data.rssi = rssi + + return device, adv_data + + +def _make_scanner( + discovered: dict[str, tuple[MagicMock, MagicMock]], +) -> MagicMock: + """Create a mock BLE scanner with discovered devices.""" + scanner = MagicMock() + scanner.discovered_devices_and_advertisement_data = discovered + return scanner + + +def _make_random_mac_entry() -> MagicMock: + """Create a mock config entry with random MAC.""" + entry = MagicMock(spec=ConfigEntry) + entry.entry_id = "test_entry_id" + entry.data = { + CONF_MAC: RANDOM_MAC_PLACEHOLDER, + CONF_PIN: "12345", + CONF_DEV_ID: "expected_dev_id", + } + return entry + + +def test_is_random_mac_true() -> None: + """Test _is_random_mac returns True for randomized MAC.""" + entry = MagicMock(spec=ConfigEntry) + entry.data = {CONF_MAC: RANDOM_MAC_PLACEHOLDER} + assert _is_random_mac(entry) is True + + +def test_is_random_mac_false() -> None: + """Test _is_random_mac returns False for static MAC.""" + entry = MagicMock(spec=ConfigEntry) + entry.data = {CONF_MAC: "AA:BB:CC:DD:EE:FF"} + assert _is_random_mac(entry) is False + + +async def test_find_device_by_scanning_match_found(hass: HomeAssistant) -> None: + """Test _find_device_by_scanning finds matching device.""" + dev_close, adv_close = _make_discovered_device("11:22:33:44:55:66", rssi=-30) + dev_far, adv_far = _make_discovered_device("AA:BB:CC:DD:EE:FF", rssi=-80) + + scanner = _make_scanner( + { + "AA:BB:CC:DD:EE:FF": (dev_far, adv_far), + "11:22:33:44:55:66": (dev_close, adv_close), + } + ) + + mock_client = AsyncMock() + mock_client.is_connected = True + mock_client.disconnect = AsyncMock() + + with ( + patch( + "custom_components.blanco_unit.bluetooth.async_get_scanner", + return_value=scanner, + ), + patch( + "custom_components.blanco_unit.establish_connection", + return_value=mock_client, + ), + patch( + "custom_components.blanco_unit.validate_pin", + return_value=PinValidationResult( + is_valid=True, dev_id="expected_dev_id", dev_type=1 + ), + ), + ): + device = await _find_device_by_scanning(hass, "12345", "expected_dev_id") + + # Should return the closest device (higher RSSI first) + assert device == dev_close + + +async def test_find_device_by_scanning_no_candidates(hass: HomeAssistant) -> None: + """Test _find_device_by_scanning raises ConfigEntryNotReady when no devices found.""" + scanner = _make_scanner({}) + + with patch( + "custom_components.blanco_unit.bluetooth.async_get_scanner", + return_value=scanner, + ): + with pytest.raises(ConfigEntryNotReady) as exc_info: + await _find_device_by_scanning(hass, "12345", "expected_dev_id") + + assert exc_info.value.translation_key == "error_device_not_found" + + +async def test_find_device_by_scanning_filters_by_uuid(hass: HomeAssistant) -> None: + """Test _find_device_by_scanning ignores devices without matching UUID.""" + dev_match, adv_match = _make_discovered_device("11:22:33:44:55:66") + + dev_other = MagicMock() + dev_other.address = "AA:BB:CC:DD:EE:FF" + adv_other = MagicMock() + adv_other.service_uuids = ["00001800-0000-1000-8000-00805f9b34fb"] + adv_other.rssi = -20 + + scanner = _make_scanner( + { + "11:22:33:44:55:66": (dev_match, adv_match), + "AA:BB:CC:DD:EE:FF": (dev_other, adv_other), + } + ) + + mock_client = AsyncMock() + mock_client.is_connected = True + mock_client.disconnect = AsyncMock() + + with ( + patch( + "custom_components.blanco_unit.bluetooth.async_get_scanner", + return_value=scanner, + ), + patch( + "custom_components.blanco_unit.establish_connection", + return_value=mock_client, + ), + patch( + "custom_components.blanco_unit.validate_pin", + return_value=PinValidationResult( + is_valid=True, dev_id="expected_dev_id", dev_type=1 + ), + ), + ): + device = await _find_device_by_scanning(hass, "12345", "expected_dev_id") + + assert device == dev_match + + +async def test_find_device_by_scanning_auth_failure(hass: HomeAssistant) -> None: + """Test _find_device_by_scanning raises ConfigEntryAuthFailed when PIN rejected.""" + dev, adv = _make_discovered_device("11:22:33:44:55:66") + scanner = _make_scanner({"11:22:33:44:55:66": (dev, adv)}) + + mock_client = AsyncMock() + mock_client.is_connected = True + mock_client.disconnect = AsyncMock() + + with ( + patch( + "custom_components.blanco_unit.bluetooth.async_get_scanner", + return_value=scanner, + ), + patch( + "custom_components.blanco_unit.establish_connection", + return_value=mock_client, + ), + patch( + "custom_components.blanco_unit.validate_pin", + return_value=PinValidationResult( + is_valid=False, dev_id=None, dev_type=None + ), + ), + ): + with pytest.raises(ConfigEntryAuthFailed) as exc_info: + await _find_device_by_scanning(hass, "12345", "expected_dev_id") + + assert exc_info.value.translation_key == "error_invalid_authentication" + + +async def test_find_device_by_scanning_dev_id_mismatch(hass: HomeAssistant) -> None: + """Test _find_device_by_scanning raises ConfigEntryNotReady when dev_id doesn't match.""" + dev, adv = _make_discovered_device("11:22:33:44:55:66") + scanner = _make_scanner({"11:22:33:44:55:66": (dev, adv)}) + + mock_client = AsyncMock() + mock_client.is_connected = True + mock_client.disconnect = AsyncMock() + + with ( + patch( + "custom_components.blanco_unit.bluetooth.async_get_scanner", + return_value=scanner, + ), + patch( + "custom_components.blanco_unit.establish_connection", + return_value=mock_client, + ), + patch( + "custom_components.blanco_unit.validate_pin", + return_value=PinValidationResult( + is_valid=True, dev_id="wrong_dev_id", dev_type=1 + ), + ), + ): + with pytest.raises(ConfigEntryNotReady) as exc_info: + await _find_device_by_scanning(hass, "12345", "expected_dev_id") + + assert exc_info.value.translation_key == "error_device_not_found" + + +async def test_find_device_by_scanning_connection_failure_skipped( + hass: HomeAssistant, +) -> None: + """Test _find_device_by_scanning skips devices with connection failures.""" + dev_fail, adv_fail = _make_discovered_device("11:22:33:44:55:66", rssi=-20) + dev_ok, adv_ok = _make_discovered_device("AA:BB:CC:DD:EE:FF", rssi=-50) + + scanner = _make_scanner( + { + "11:22:33:44:55:66": (dev_fail, adv_fail), + "AA:BB:CC:DD:EE:FF": (dev_ok, adv_ok), + } + ) + + mock_client_ok = AsyncMock() + mock_client_ok.is_connected = True + mock_client_ok.disconnect = AsyncMock() + + call_count = 0 + + async def mock_establish(client_class, device, name): + nonlocal call_count + call_count += 1 + if call_count == 1: + raise OSError("Connection failed") + return mock_client_ok + + with ( + patch( + "custom_components.blanco_unit.bluetooth.async_get_scanner", + return_value=scanner, + ), + patch( + "custom_components.blanco_unit.establish_connection", + side_effect=mock_establish, + ), + patch( + "custom_components.blanco_unit.validate_pin", + return_value=PinValidationResult( + is_valid=True, dev_id="expected_dev_id", dev_type=1 + ), + ), + ): + device = await _find_device_by_scanning(hass, "12345", "expected_dev_id") + + # Should skip the failed device and return the second one + assert device == dev_ok + + +async def test_find_device_by_scanning_sorts_by_rssi(hass: HomeAssistant) -> None: + """Test _find_device_by_scanning tries closest device first.""" + dev_far, adv_far = _make_discovered_device("AA:BB:CC:DD:EE:FF", rssi=-80) + dev_close, adv_close = _make_discovered_device("11:22:33:44:55:66", rssi=-20) + dev_medium, adv_medium = _make_discovered_device("22:33:44:55:66:77", rssi=-50) + + scanner = _make_scanner( + { + "AA:BB:CC:DD:EE:FF": (dev_far, adv_far), + "11:22:33:44:55:66": (dev_close, adv_close), + "22:33:44:55:66:77": (dev_medium, adv_medium), + } + ) + + mock_client = AsyncMock() + mock_client.is_connected = True + mock_client.disconnect = AsyncMock() + + tried_addresses = [] + + async def mock_establish(client_class, device, name): + tried_addresses.append(device.address) + return mock_client + + with ( + patch( + "custom_components.blanco_unit.bluetooth.async_get_scanner", + return_value=scanner, + ), + patch( + "custom_components.blanco_unit.establish_connection", + side_effect=mock_establish, + ), + patch( + "custom_components.blanco_unit.validate_pin", + return_value=PinValidationResult( + is_valid=True, dev_id="expected_dev_id", dev_type=1 + ), + ), + ): + await _find_device_by_scanning(hass, "12345", "expected_dev_id") + + # Closest device (highest RSSI) should be tried first + assert tried_addresses[0] == "11:22:33:44:55:66" + + +async def test_register_retry_callback_random_mac(hass: HomeAssistant) -> None: + """Test _register_retry_callback uses service_uuid filter for random MAC.""" + entry = _make_random_mac_entry() + + hass.data[DOMAIN] = {} + hass.data[DOMAIN][entry.entry_id] = {} + + with patch( + "custom_components.blanco_unit.bluetooth.async_register_callback" + ) as mock_register: + mock_register.return_value = MagicMock() + _register_retry_callback(hass, entry) + + mock_register.assert_called_once() + call_args = mock_register.call_args + # Second argument is the filter dict + filter_dict = call_args[0][2] + assert "service_uuid" in filter_dict + assert filter_dict["service_uuid"] == CHARACTERISTIC_UUID + + +async def test_register_retry_callback_static_mac(hass: HomeAssistant) -> None: + """Test _register_retry_callback uses address filter for static MAC.""" + entry = MagicMock(spec=ConfigEntry) + entry.entry_id = "test_entry_id" + entry.data = {CONF_MAC: "AA:BB:CC:DD:EE:FF"} + + hass.data[DOMAIN] = {} + hass.data[DOMAIN][entry.entry_id] = {} + + with patch( + "custom_components.blanco_unit.bluetooth.async_register_callback" + ) as mock_register: + mock_register.return_value = MagicMock() + _register_retry_callback(hass, entry) + + mock_register.assert_called_once() + call_args = mock_register.call_args + filter_dict = call_args[0][2] + assert "address" in filter_dict + assert filter_dict["address"] == "AA:BB:CC:DD:EE:FF" + + +async def test_register_retry_callback_skips_if_already_registered( + hass: HomeAssistant, +) -> None: + """Test _register_retry_callback does not register twice.""" + entry = _make_random_mac_entry() + + hass.data[DOMAIN] = {} + hass.data[DOMAIN][entry.entry_id] = {BLE_CALLBACK: MagicMock()} + + with patch( + "custom_components.blanco_unit.bluetooth.async_register_callback" + ) as mock_register: + _register_retry_callback(hass, entry) + mock_register.assert_not_called() + + +async def test_async_setup_entry_random_mac_success(hass: HomeAssistant) -> None: + """Test config entry setup with random MAC finds device via scanning.""" + mock_device = MagicMock() + mock_device.address = "11:22:33:44:55:66" + + mock_coordinator = MagicMock() + mock_coordinator.async_config_entry_first_refresh = AsyncMock() + + mock_entry = _make_random_mac_entry() + mock_entry.title = "Test Blanco" + mock_entry.add_update_listener = MagicMock(return_value=MagicMock()) + + with ( + patch( + "custom_components.blanco_unit._find_device_by_scanning", + return_value=mock_device, + ), + patch( + "custom_components.blanco_unit.BlancoUnitCoordinator", + return_value=mock_coordinator, + ), + patch.object(hass.config_entries, "async_forward_entry_setups") as mock_forward, + ): + result = await async_setup_entry(hass, mock_entry) + + assert result is True + assert mock_entry.runtime_data == mock_coordinator + mock_forward.assert_called_once() + + +async def test_async_setup_entry_random_mac_not_found(hass: HomeAssistant) -> None: + """Test config entry setup with random MAC when no device found.""" + mock_entry = _make_random_mac_entry() + + hass.data[DOMAIN] = {} + hass.data[DOMAIN][mock_entry.entry_id] = {} + + with ( + patch( + "custom_components.blanco_unit._find_device_by_scanning", + side_effect=ConfigEntryNotReady(translation_key="error_device_not_found"), + ), + patch( + "custom_components.blanco_unit.bluetooth.async_register_callback", + return_value=MagicMock(), + ) as mock_register, + ): + with pytest.raises(ConfigEntryNotReady) as exc_info: + await async_setup_entry(hass, mock_entry) + + assert exc_info.value.translation_key == "error_device_not_found" + # Should register retry callback with service_uuid + mock_register.assert_called_once() + + +async def test_async_setup_entry_random_mac_auth_failed(hass: HomeAssistant) -> None: + """Test config entry setup with random MAC when PIN fails.""" + mock_entry = _make_random_mac_entry() + + hass.data[DOMAIN] = {} + hass.data[DOMAIN][mock_entry.entry_id] = {} + + with patch( + "custom_components.blanco_unit._find_device_by_scanning", + side_effect=ConfigEntryAuthFailed( + translation_key="error_invalid_authentication" + ), + ): + with pytest.raises(ConfigEntryAuthFailed) as exc_info: + await async_setup_entry(hass, mock_entry) + + assert exc_info.value.translation_key == "error_invalid_authentication" + + +async def test_async_unload_entry_random_mac_skips_rediscover( + hass: HomeAssistant, +) -> None: + """Test config entry unload with random MAC skips async_rediscover_address.""" + mock_coordinator = MagicMock() + mock_coordinator.unload = AsyncMock() + + mock_entry = _make_random_mac_entry() + mock_entry.runtime_data = mock_coordinator + + hass.data[DOMAIN] = {} + hass.data[DOMAIN][mock_entry.entry_id] = {} + + with ( + patch.object( + hass.config_entries, + "async_unload_platforms", + return_value=True, + ), + patch( + "custom_components.blanco_unit.bluetooth.async_rediscover_address" + ) as mock_rediscover, + ): + result = await async_unload_entry(hass, mock_entry) + + assert result is True + mock_coordinator.unload.assert_called_once() + # Should NOT call rediscover for random MAC + mock_rediscover.assert_not_called() diff --git a/tests/test_sensor.py b/tests/test_sensor.py index 5923dca..a864339 100644 --- a/tests/test_sensor.py +++ b/tests/test_sensor.py @@ -23,6 +23,7 @@ CleanModeStateSensor, CO2RemainingSensor, DeviceNameSensor, + DeviceTypeSensor, ErrorBitsSensor, FilterLifetimeSensor, FilterRemainingSensor, @@ -58,6 +59,7 @@ def mock_coordinator(): connected=True, available=True, device_id="test_device_id", + device_type=1, status=BlancoUnitStatus( tap_state=1, filter_rest=85, @@ -154,8 +156,8 @@ def mock_add_entities(entities): await async_setup_entry(hass, mock_config_entry, mock_add_entities) - # Verify all 22 sensors were added - assert len(entities_added) == 22 + # Verify all 24 sensors were added + assert len(entities_added) == 24 # Verify sensor types sensor_types = [type(entity).__name__ for entity in entities_added] @@ -171,6 +173,7 @@ def mock_add_entities(entities): assert "FirmwareElecSensor" in sensor_types assert "DeviceNameSensor" in sensor_types assert "ResetCountSensor" in sensor_types + assert "DeviceTypeSensor" in sensor_types assert "SerialNumberSensor" in sensor_types assert "ServiceCodeSensor" in sensor_types assert "WiFiSSIDSensor" in sensor_types @@ -399,6 +402,22 @@ async def test_reset_count_sensor_unavailable(mock_coordinator) -> None: assert sensor.native_value is None +async def test_device_type_sensor(mock_coordinator) -> None: + """Test DeviceTypeSensor.""" + sensor = DeviceTypeSensor(mock_coordinator) + + assert sensor.native_value == 1 + assert sensor.unique_id == "device_type" + + +async def test_device_type_sensor_none(mock_coordinator) -> None: + """Test DeviceTypeSensor when device_type is None.""" + mock_coordinator.data.device_type = None + sensor = DeviceTypeSensor(mock_coordinator) + + assert sensor.native_value is None + + async def test_serial_number_sensor(mock_coordinator) -> None: """Test SerialNumberSensor.""" sensor = SerialNumberSensor(mock_coordinator)