Home-Assistant/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-12-24 12:59:34 +00:00
Code Activity

La Marzocco add Bluetooth offline mode (#157011)

Browse Source
This commit is contained in:
Josef Zweck
2025-12-03 05:53:27 +01:00
committed by GitHub
parent b6786c5a42
commit 639a96f8cb
16 changed files with 1045 additions and 180 deletions
Download Patch File Download Diff File Expand all files Collapse all files

81
homeassistant/components/lamarzocco/__init__.py
View File

@@ -35,6 +35,7 @@ from homeassistant.helpers.aiohttp_client import async_create_clientsession
from .const import CONF_INSTALLATION_KEY, CONF_USE_BLUETOOTH, DOMAIN
from .coordinator import (
LaMarzoccoBluetoothUpdateCoordinator,
LaMarzoccoConfigEntry,
LaMarzoccoConfigUpdateCoordinator,
LaMarzoccoRuntimeData,
@@ -72,38 +73,10 @@ async def async_setup_entry(hass: HomeAssistant, entry: LaMarzoccoConfigEntry) -
client=create_client_session(hass),
)
try:
settings = await cloud_client.get_thing_settings(serial)
except AuthFail as ex:
raise ConfigEntryAuthFailed(
translation_domain=DOMAIN, translation_key="authentication_failed"
) from ex
except (RequestNotSuccessful, TimeoutError) as ex:
_LOGGER.debug(ex, exc_info=True)
raise ConfigEntryNotReady(
translation_domain=DOMAIN, translation_key="api_error"
) from ex
gateway_version = version.parse(
settings.firmwares[FirmwareType.GATEWAY].build_version
)
if gateway_version < version.parse("v5.0.9"):
# incompatible gateway firmware, create an issue
ir.async_create_issue(
hass,
DOMAIN,
"unsupported_gateway_firmware",
is_fixable=False,
severity=ir.IssueSeverity.ERROR,
translation_key="unsupported_gateway_firmware",
translation_placeholders={"gateway_version": str(gateway_version)},
)
# initialize Bluetooth
bluetooth_client: LaMarzoccoBluetoothClient | None = None
if entry.options.get(CONF_USE_BLUETOOTH, True) and (
token := (entry.data.get(CONF_TOKEN) or settings.ble_auth_token)
token := entry.data.get(CONF_TOKEN)
):
if CONF_MAC not in entry.data:
for discovery_info in async_discovered_service_info(hass):
@@ -145,6 +118,44 @@ async def async_setup_entry(hass: HomeAssistant, entry: LaMarzoccoConfigEntry) -
_LOGGER.info(
"Bluetooth device not found during lamarzocco setup, continuing with cloud only"
)
try:
settings = await cloud_client.get_thing_settings(serial)
except AuthFail as ex:
raise ConfigEntryAuthFailed(
translation_domain=DOMAIN, translation_key="authentication_failed"
) from ex
except (RequestNotSuccessful, TimeoutError) as ex:
_LOGGER.debug(ex, exc_info=True)
if not bluetooth_client:
raise ConfigEntryNotReady(
translation_domain=DOMAIN, translation_key="api_error"
) from ex
_LOGGER.debug("Cloud failed, continuing with Bluetooth only", exc_info=True)
else:
gateway_version = version.parse(
settings.firmwares[FirmwareType.GATEWAY].build_version
)
if gateway_version < version.parse("v5.0.9"):
# incompatible gateway firmware, create an issue
ir.async_create_issue(
hass,
DOMAIN,
"unsupported_gateway_firmware",
is_fixable=False,
severity=ir.IssueSeverity.ERROR,
translation_key="unsupported_gateway_firmware",
translation_placeholders={"gateway_version": str(gateway_version)},
)
# Update BLE Token if exists
if settings.ble_auth_token:
hass.config_entries.async_update_entry(
entry,
data={
**entry.data,
CONF_TOKEN: settings.ble_auth_token,
},
)
device = LaMarzoccoMachine(
serial_number=entry.unique_id,
@@ -153,7 +164,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: LaMarzoccoConfigEntry) -
)
coordinators = LaMarzoccoRuntimeData(
LaMarzoccoConfigUpdateCoordinator(hass, entry, device, cloud_client),
LaMarzoccoConfigUpdateCoordinator(hass, entry, device),
LaMarzoccoSettingsUpdateCoordinator(hass, entry, device),
LaMarzoccoScheduleUpdateCoordinator(hass, entry, device),
LaMarzoccoStatisticsUpdateCoordinator(hass, entry, device),
@@ -166,6 +177,16 @@ async def async_setup_entry(hass: HomeAssistant, entry: LaMarzoccoConfigEntry) -
coordinators.statistics_coordinator.async_config_entry_first_refresh(),
)
# bt coordinator only if bluetooth client is available
# and after the initial refresh of the config coordinator
# to fetch only if the others failed
if bluetooth_client:
bluetooth_coordinator = LaMarzoccoBluetoothUpdateCoordinator(
hass, entry, device
)
await bluetooth_coordinator.async_config_entry_first_refresh()
coordinators.bluetooth_coordinator = bluetooth_coordinator
entry.runtime_data = coordinators
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)

15
homeassistant/components/lamarzocco/binary_sensor.py
View File

@@ -6,7 +6,7 @@ from typing import cast
from pylamarzocco import LaMarzoccoMachine
from pylamarzocco.const import BackFlushStatus, MachineState, ModelName, WidgetType
from pylamarzocco.models import BackFlush, MachineStatus
from pylamarzocco.models import BackFlush, MachineStatus, NoWater
from homeassistant.components.binary_sensor import (
BinarySensorDeviceClass,
@@ -39,8 +39,15 @@ ENTITIES: tuple[LaMarzoccoBinarySensorEntityDescription, ...] = (
key="water_tank",
translation_key="water_tank",
device_class=BinarySensorDeviceClass.PROBLEM,
is_on_fn=lambda machine: WidgetType.CM_NO_WATER in machine.dashboard.config,
is_on_fn=(
lambda machine: cast(
NoWater, machine.dashboard.config[WidgetType.CM_NO_WATER]
).allarm
if WidgetType.CM_NO_WATER in machine.dashboard.config
else False
),
entity_category=EntityCategory.DIAGNOSTIC,
bt_offline_mode=True,
),
LaMarzoccoBinarySensorEntityDescription(
key="brew_active",
@@ -93,7 +100,9 @@ async def async_setup_entry(
coordinator = entry.runtime_data.config_coordinator
async_add_entities(
LaMarzoccoBinarySensorEntity(coordinator, description)
LaMarzoccoBinarySensorEntity(
coordinator, description, entry.runtime_data.bluetooth_coordinator
)
for description in ENTITIES
if description.supported_fn(coordinator)
)

54
homeassistant/components/lamarzocco/coordinator.py
View File

@@ -10,8 +10,12 @@ from datetime import timedelta
import logging
from typing import Any
from pylamarzocco import LaMarzoccoCloudClient, LaMarzoccoMachine
from pylamarzocco.exceptions import AuthFail, RequestNotSuccessful
from pylamarzocco import LaMarzoccoMachine
from pylamarzocco.exceptions import (
AuthFail,
BluetoothConnectionFailed,
RequestNotSuccessful,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import EVENT_HOMEASSISTANT_STOP
@@ -36,6 +40,7 @@ class LaMarzoccoRuntimeData:
settings_coordinator: LaMarzoccoSettingsUpdateCoordinator
schedule_coordinator: LaMarzoccoScheduleUpdateCoordinator
statistics_coordinator: LaMarzoccoStatisticsUpdateCoordinator
bluetooth_coordinator: LaMarzoccoBluetoothUpdateCoordinator | None = None
type LaMarzoccoConfigEntry = ConfigEntry[LaMarzoccoRuntimeData]
@@ -46,14 +51,13 @@ class LaMarzoccoUpdateCoordinator(DataUpdateCoordinator[None]):
_default_update_interval = SCAN_INTERVAL
config_entry: LaMarzoccoConfigEntry
_websocket_task: Task | None = None
update_success = False
def __init__(
self,
hass: HomeAssistant,
entry: LaMarzoccoConfigEntry,
device: LaMarzoccoMachine,
cloud_client: LaMarzoccoCloudClient | None = None,
) -> None:
"""Initialize coordinator."""
super().__init__(
@@ -64,7 +68,7 @@ class LaMarzoccoUpdateCoordinator(DataUpdateCoordinator[None]):
update_interval=self._default_update_interval,
)
self.device = device
self.cloud_client = cloud_client
self._websocket_task: Task | None = None
@property
def websocket_terminated(self) -> bool:
@@ -81,14 +85,28 @@ class LaMarzoccoUpdateCoordinator(DataUpdateCoordinator[None]):
await func()
except AuthFail as ex:
_LOGGER.debug("Authentication failed", exc_info=True)
self.update_success = False
raise ConfigEntryAuthFailed(
translation_domain=DOMAIN, translation_key="authentication_failed"
) from ex
except RequestNotSuccessful as ex:
_LOGGER.debug(ex, exc_info=True)
self.update_success = False
# if no bluetooth coordinator, this is a fatal error
# otherwise, bluetooth may still work
if not self.device.bluetooth_client_available:
raise UpdateFailed(
translation_domain=DOMAIN, translation_key="api_error"
) from ex
except BluetoothConnectionFailed as err:
self.update_success = False
raise UpdateFailed(
translation_domain=DOMAIN, translation_key="api_error"
) from ex
translation_domain=DOMAIN,
translation_key="bluetooth_connection_failed",
) from err
else:
self.update_success = True
_LOGGER.debug("Current status: %s", self.device.dashboard.to_dict())
async def _async_setup(self) -> None:
"""Set up coordinator."""
@@ -109,11 +127,9 @@ class LaMarzoccoUpdateCoordinator(DataUpdateCoordinator[None]):
class LaMarzoccoConfigUpdateCoordinator(LaMarzoccoUpdateCoordinator):
"""Class to handle fetching data from the La Marzocco API centrally."""
cloud_client: LaMarzoccoCloudClient
async def _internal_async_setup(self) -> None:
"""Set up the coordinator."""
await self.cloud_client.async_get_access_token()
await self.device.ensure_token_valid()
await self.device.get_dashboard()
_LOGGER.debug("Current status: %s", self.device.dashboard.to_dict())
@@ -121,7 +137,7 @@ class LaMarzoccoConfigUpdateCoordinator(LaMarzoccoUpdateCoordinator):
"""Fetch data from API endpoint."""
# ensure token stays valid; does nothing if token is still valid
await self.cloud_client.async_get_access_token()
await self.device.ensure_token_valid()
# Only skip websocket reconnection if it's currently connected and the task is still running
if self.device.websocket.connected and not self.websocket_terminated:
@@ -193,3 +209,19 @@ class LaMarzoccoStatisticsUpdateCoordinator(LaMarzoccoUpdateCoordinator):
"""Fetch data from API endpoint."""
await self.device.get_coffee_and_flush_counter()
_LOGGER.debug("Current statistics: %s", self.device.statistics.to_dict())
class LaMarzoccoBluetoothUpdateCoordinator(LaMarzoccoUpdateCoordinator):
"""Class to handle fetching data from the La Marzocco Bluetooth API centrally."""
async def _internal_async_setup(self) -> None:
"""Initial setup for Bluetooth coordinator."""
await self.device.get_model_info_from_bluetooth()
async def _internal_async_update_data(self) -> None:
"""Fetch data from Bluetooth endpoint."""
# if the websocket is connected and the machine is connected to the cloud
# skip bluetooth update, because we get push updates
if self.device.websocket.connected and self.device.dashboard.connected:
return
await self.device.get_dashboard_from_bluetooth()

38
homeassistant/components/lamarzocco/entity.py
View File

@@ -17,7 +17,10 @@ from homeassistant.helpers.entity import EntityDescription
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN
from .coordinator import LaMarzoccoUpdateCoordinator
from .coordinator import (
LaMarzoccoBluetoothUpdateCoordinator,
LaMarzoccoUpdateCoordinator,
)
@dataclass(frozen=True, kw_only=True)
@@ -26,6 +29,7 @@ class LaMarzoccoEntityDescription(EntityDescription):
available_fn: Callable[[LaMarzoccoUpdateCoordinator], bool] = lambda _: True
supported_fn: Callable[[LaMarzoccoUpdateCoordinator], bool] = lambda _: True
bt_offline_mode: bool = False
class LaMarzoccoBaseEntity(
@@ -45,14 +49,19 @@ class LaMarzoccoBaseEntity(
super().__init__(coordinator)
device = coordinator.device
self._attr_unique_id = f"{device.serial_number}_{key}"
sw_version = (
device.settings.firmwares[FirmwareType.MACHINE].build_version
if FirmwareType.MACHINE in device.settings.firmwares
else None
)
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, device.serial_number)},
name=device.dashboard.name,
name=device.dashboard.name or self.coordinator.config_entry.title,
manufacturer="La Marzocco",
model=device.dashboard.model_name.value,
model_id=device.dashboard.model_code.value,
serial_number=device.serial_number,
sw_version=device.settings.firmwares[FirmwareType.MACHINE].build_version,
sw_version=sw_version,
)
connections: set[tuple[str, str]] = set()
if coordinator.config_entry.data.get(CONF_ADDRESS):
@@ -77,8 +86,12 @@ class LaMarzoccoBaseEntity(
if WidgetType.CM_MACHINE_STATUS in self.coordinator.device.dashboard.config
else MachineState.OFF
)
return super().available and not (
self._unavailable_when_machine_off and machine_state is MachineState.OFF
return (
super().available
and not (
self._unavailable_when_machine_off and machine_state is MachineState.OFF
)
and self.coordinator.update_success
)
@@ -90,6 +103,11 @@ class LaMarzoccoEntity(LaMarzoccoBaseEntity):
@property
def available(self) -> bool:
"""Return True if entity is available."""
if (
self.entity_description.bt_offline_mode
and self.bluetooth_coordinator is not None
):
return self.bluetooth_coordinator.last_update_success
if super().available:
return self.entity_description.available_fn(self.coordinator)
return False
@@ -98,7 +116,17 @@ class LaMarzoccoEntity(LaMarzoccoBaseEntity):
self,
coordinator: LaMarzoccoUpdateCoordinator,
entity_description: LaMarzoccoEntityDescription,
bluetooth_coordinator: LaMarzoccoBluetoothUpdateCoordinator | None = None,
) -> None:
"""Initialize the entity."""
super().__init__(coordinator, entity_description.key)
self.entity_description = entity_description
self.bluetooth_coordinator = bluetooth_coordinator
async def async_added_to_hass(self) -> None:
"""Handle when entity is added to hass."""
await super().async_added_to_hass()
if self.bluetooth_coordinator is not None:
self.async_on_remove(
self.bluetooth_coordinator.async_add_listener(self.async_write_ha_state)
)

14
homeassistant/components/lamarzocco/number.py
View File

@@ -58,6 +58,7 @@ ENTITIES: tuple[LaMarzoccoNumberEntityDescription, ...] = (
CoffeeBoiler, machine.dashboard.config[WidgetType.CM_COFFEE_BOILER]
).target_temperature
),
bt_offline_mode=True,
),
LaMarzoccoNumberEntityDescription(
key="steam_temp",
@@ -78,6 +79,7 @@ ENTITIES: tuple[LaMarzoccoNumberEntityDescription, ...] = (
lambda coordinator: coordinator.device.dashboard.model_name
in (ModelName.GS3_AV, ModelName.GS3_MP)
),
bt_offline_mode=True,
),
LaMarzoccoNumberEntityDescription(
key="smart_standby_time",
@@ -96,6 +98,7 @@ ENTITIES: tuple[LaMarzoccoNumberEntityDescription, ...] = (
)
),
native_value_fn=lambda machine: machine.schedule.smart_wake_up_sleep.smart_stand_by_minutes,
bt_offline_mode=True,
),
LaMarzoccoNumberEntityDescription(
key="preinfusion_off",
@@ -226,13 +229,14 @@ async def async_setup_entry(
) -> None:
"""Set up number entities."""
coordinator = entry.runtime_data.config_coordinator
entities: list[NumberEntity] = [
LaMarzoccoNumberEntity(coordinator, description)
async_add_entities(
LaMarzoccoNumberEntity(
coordinator, description, entry.runtime_data.bluetooth_coordinator
)
for description in ENTITIES
if description.supported_fn(coordinator)
]
async_add_entities(entities)
)
class LaMarzoccoNumberEntity(LaMarzoccoEntity, NumberEntity):

5
homeassistant/components/lamarzocco/select.py
View File

@@ -80,6 +80,7 @@ ENTITIES: tuple[LaMarzoccoSelectEntityDescription, ...] = (
lambda coordinator: coordinator.device.dashboard.model_name
in (ModelName.LINEA_MINI_R, ModelName.LINEA_MICRA)
),
bt_offline_mode=True,
),
LaMarzoccoSelectEntityDescription(
key="prebrew_infusion_select",
@@ -128,7 +129,9 @@ async def async_setup_entry(
coordinator = entry.runtime_data.config_coordinator
async_add_entities(
LaMarzoccoSelectEntity(coordinator, description)
LaMarzoccoSelectEntity(
coordinator, description, entry.runtime_data.bluetooth_coordinator
)
for description in ENTITIES
if description.supported_fn(coordinator)
)

3
homeassistant/components/lamarzocco/strings.json
View File

@@ -183,6 +183,9 @@
"auto_on_off_error": {
"message": "Error while setting auto on/off to {state} for {id}"
},
"bluetooth_connection_failed": {
"message": "Error while connecting to machine via Bluetooth"
},
"button_error": {
"message": "Error while executing button {key}"
},

8
homeassistant/components/lamarzocco/switch.py
View File

@@ -50,6 +50,7 @@ ENTITIES: tuple[LaMarzoccoSwitchEntityDescription, ...] = (
).mode
is MachineMode.BREWING_MODE
),
bt_offline_mode=True,
),
LaMarzoccoSwitchEntityDescription(
key="steam_boiler_enable",
@@ -65,6 +66,7 @@ ENTITIES: tuple[LaMarzoccoSwitchEntityDescription, ...] = (
lambda coordinator: coordinator.device.dashboard.model_name
in (ModelName.LINEA_MINI_R, ModelName.LINEA_MICRA)
),
bt_offline_mode=True,
),
LaMarzoccoSwitchEntityDescription(
key="steam_boiler_enable",
@@ -80,6 +82,7 @@ ENTITIES: tuple[LaMarzoccoSwitchEntityDescription, ...] = (
lambda coordinator: coordinator.device.dashboard.model_name
not in (ModelName.LINEA_MINI_R, ModelName.LINEA_MICRA)
),
bt_offline_mode=True,
),
LaMarzoccoSwitchEntityDescription(
key="smart_standby_enabled",
@@ -91,6 +94,7 @@ ENTITIES: tuple[LaMarzoccoSwitchEntityDescription, ...] = (
minutes=machine.schedule.smart_wake_up_sleep.smart_stand_by_minutes,
),
is_on_fn=lambda machine: machine.schedule.smart_wake_up_sleep.smart_stand_by_enabled,
bt_offline_mode=True,
),
)
@@ -106,7 +110,9 @@ async def async_setup_entry(
entities: list[SwitchEntity] = []
entities.extend(
LaMarzoccoSwitchEntity(coordinator, description)
LaMarzoccoSwitchEntity(
coordinator, description, entry.runtime_data.bluetooth_coordinator
)
for description in ENTITIES
if description.supported_fn(coordinator)
)

74
tests/components/lamarzocco/conftest.py
View File

@@ -16,7 +16,7 @@ from pylamarzocco.util import InstallationKey
import pytest
from homeassistant.components.lamarzocco.const import CONF_INSTALLATION_KEY, DOMAIN
from homeassistant.const import CONF_ADDRESS, CONF_TOKEN
from homeassistant.const import CONF_ADDRESS, CONF_MAC, CONF_TOKEN
from homeassistant.core import HomeAssistant
from . import MOCK_INSTALLATION_KEY, SERIAL_DICT, USER_INPUT, async_init_integration
@@ -150,6 +150,78 @@ def mock_ble_device() -> BLEDevice:
return BLEDevice("00:00:00:00:00:00", "GS_GS012345", details={"path": "path"})
@pytest.fixture
def mock_bluetooth_client() -> Generator[MagicMock]:
"""Return a mocked Bluetooth client."""
with patch(
"homeassistant.components.lamarzocco.LaMarzoccoBluetoothClient",
autospec=True,
) as mock_bt_client_cls:
mock_bt_client = mock_bt_client_cls.return_value
mock_bt_client.disconnect = AsyncMock()
yield mock_bt_client
@pytest.fixture
def mock_ble_device_from_address(
mock_ble_device: BLEDevice,
) -> Generator[MagicMock]:
"""Return a mocked async_ble_device_from_address."""
with patch(
"homeassistant.components.lamarzocco.async_ble_device_from_address",
return_value=mock_ble_device,
) as mock_ble_device_from_address:
yield mock_ble_device_from_address
@pytest.fixture
def mock_lamarzocco_bluetooth(mock_lamarzocco: MagicMock) -> MagicMock:
"""Return a mocked LM client with Bluetooth config."""
if mock_lamarzocco.dashboard.model_name == ModelName.LINEA_MICRA:
config = load_json_object_fixture("config_micra_bluetooth.json", DOMAIN)
else:
config = load_json_object_fixture("config_gs3_bluetooth.json", DOMAIN)
mock_lamarzocco.dashboard = ThingDashboardConfig.from_dict(config)
mock_lamarzocco.dashboard.model_name = mock_lamarzocco.dashboard.model_name
mock_lamarzocco.schedule = ThingSchedulingSettings(
serial_number=mock_lamarzocco.serial_number
)
mock_lamarzocco.settings = ThingSettings(
serial_number=mock_lamarzocco.serial_number
)
mock_lamarzocco.statistics = ThingStatistics(
serial_number=mock_lamarzocco.serial_number
)
mock_lamarzocco.to_dict.return_value = {
"serial_number": mock_lamarzocco.serial_number,
"dashboard": mock_lamarzocco.dashboard.to_dict(),
}
return mock_lamarzocco
@pytest.fixture
def mock_config_entry_bluetooth(
mock_lamarzocco: MagicMock,
mock_ble_device: BLEDevice,
) -> MockConfigEntry:
"""Return a mocked config entry with Bluetooth enabled."""
return MockConfigEntry(
title=mock_lamarzocco.serial_number,
domain=DOMAIN,
version=4,
data=USER_INPUT
| {
CONF_MAC: mock_ble_device.address,
CONF_TOKEN: "token",
CONF_INSTALLATION_KEY: MOCK_INSTALLATION_KEY,
},
unique_id=mock_lamarzocco.serial_number,
)
@pytest.fixture
def mock_websocket_terminated() -> Generator[PropertyMock]:
"""Mock websocket terminated."""

74
tests/components/lamarzocco/fixtures/config_gs3_bluetooth.json Normal file
View File

@@ -0,0 +1,74 @@
{
"serialNumber": "GS012345",
"type": "CoffeeMachine",
"name": "",
"location": null,
"modelCode": "GS3AV",
"modelName": "GS3AV",
"connected": true,
"connectionDate": 1742489087479,
"offlineMode": false,
"requireFirmwareUpdate": false,
"availableFirmwareUpdate": false,
"coffeeStation": null,
"imageUrl": "",
"bleAuthToken": null,
"widgets": [
{
"code": "CMMachineStatus",
"index": 1,
"output": {
"status": "PoweredOn",
"availableModes": ["BrewingMode", "StandBy"],
"mode": "BrewingMode",
"nextStatus": {
"status": "StandBy",
"startTime": 1742857195332
},
"brewingStartTime": 1746641060000
},
"tutorialUrl": null
},
{
"code": "CMCoffeeBoiler",
"index": 1,
"output": {
"status": "Ready",
"enabled": true,
"enabledSupported": false,
"targetTemperature": 98.0,
"targetTemperatureMin": 80,
"targetTemperatureMax": 110,
"targetTemperatureStep": 0.1,
"readyStartTime": null
},
"tutorialUrl": null
},
{
"code": "CMSteamBoilerTemperature",
"index": 1,
"output": {
"status": "Off",
"enabled": true,
"enabledSupported": true,
"targetTemperature": 121.0,
"targetTemperatureSupported": true,
"targetTemperatureMin": 95,
"targetTemperatureMax": 140,
"targetTemperatureStep": 0.1,
"readyStartTime": null
},
"tutorialUrl": null
},
{
"code": "CMNoWater",
"index": 1,
"output": {
"allarm": false
},
"tutorialUrl": null
}
],
"invalidWidgets": [],
"runningCommands": []
}

68
tests/components/lamarzocco/fixtures/config_micra_bluetooth.json Normal file
View File

@@ -0,0 +1,68 @@
{
"serialNumber": "MR012345",
"type": "CoffeeMachine",
"name": "",
"location": null,
"modelCode": "LINEAMICRA",
"modelName": "LINEA MICRA",
"connected": false,
"connectionDate": 1742526019892,
"offlineMode": false,
"requireFirmwareUpdate": false,
"availableFirmwareUpdate": false,
"coffeeStation": null,
"imageUrl": "",
"bleAuthToken": null,
"widgets": [
{
"code": "CMMachineStatus",
"index": 1,
"output": {
"status": "StandBy",
"availableModes": ["BrewingMode", "StandBy"],
"mode": "StandBy",
"nextStatus": null,
"brewingStartTime": null
},
"tutorialUrl": null
},
{
"code": "CMCoffeeBoiler",
"index": 1,
"output": {
"status": "StandBy",
"enabled": true,
"enabledSupported": false,
"targetTemperature": 94.0,
"targetTemperatureMin": 80,
"targetTemperatureMax": 100,
"targetTemperatureStep": 0.1,
"readyStartTime": null
},
"tutorialUrl": null
},
{
"code": "CMSteamBoilerLevel",
"index": 1,
"output": {
"status": "StandBy",
"enabled": true,
"enabledSupported": true,
"targetLevel": "Level2",
"targetLevelSupported": true,
"readyStartTime": null
},
"tutorialUrl": null
},
{
"code": "CMNoWater",
"index": 1,
"output": {
"allarm": true
},
"tutorialUrl": null
}
],
"invalidWidgets": [],
"runningCommands": []
}

290
tests/components/lamarzocco/snapshots/test_bluetooth.ambr Normal file
View File

@@ -0,0 +1,290 @@
# serializer version: 1
# name: test_setup_through_bluetooth_only[GS3 AV-entities1][binary_sensor.GS012345_water_tank_empty]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'problem',
'friendly_name': 'GS012345 Water tank empty',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.gs012345_water_tank_empty',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'off',
})
# ---
# name: test_setup_through_bluetooth_only[GS3 AV-entities1][device_bluetooth_GS012345]
DeviceRegistryEntrySnapshot({
'area_id': None,
'config_entries': <ANY>,
'config_entries_subentries': <ANY>,
'configuration_url': None,
'connections': set({
tuple(
'bluetooth',
'00:00:00:00:00:00',
),
}),
'disabled_by': None,
'entry_type': None,
'hw_version': None,
'id': <ANY>,
'identifiers': set({
tuple(
'lamarzocco',
'GS012345',
),
}),
'labels': set({
}),
'manufacturer': 'La Marzocco',
'model': 'GS3 AV',
'model_id': 'GS3AV',
'name': 'GS012345',
'name_by_user': None,
'primary_config_entry': <ANY>,
'serial_number': 'GS012345',
'sw_version': None,
'via_device_id': None,
})
# ---
# name: test_setup_through_bluetooth_only[GS3 AV-entities1][number.GS012345_coffee_target_temperature]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'temperature',
'friendly_name': 'GS012345 Coffee target temperature',
'max': 104,
'min': 85,
'mode': <NumberMode.AUTO: 'auto'>,
'step': 0.1,
'unit_of_measurement': <UnitOfTemperature.CELSIUS: '°C'>,
}),
'context': <ANY>,
'entity_id': 'number.gs012345_coffee_target_temperature',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '98.0',
})
# ---
# name: test_setup_through_bluetooth_only[GS3 AV-entities1][number.GS012345_smart_standby_time]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'duration',
'friendly_name': 'GS012345 Smart standby time',
'max': 240,
'min': 0,
'mode': <NumberMode.AUTO: 'auto'>,
'step': 1,
'unit_of_measurement': <UnitOfTime.MINUTES: 'min'>,
}),
'context': <ANY>,
'entity_id': 'number.gs012345_smart_standby_time',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '0',
})
# ---
# name: test_setup_through_bluetooth_only[GS3 AV-entities1][number.GS012345_steam_target_temperature]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'temperature',
'friendly_name': 'GS012345 Steam target temperature',
'max': 134,
'min': 20,
'mode': <NumberMode.AUTO: 'auto'>,
'step': 0.1,
'unit_of_measurement': <UnitOfTemperature.CELSIUS: '°C'>,
}),
'context': <ANY>,
'entity_id': 'number.gs012345_steam_target_temperature',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '121.0',
})
# ---
# name: test_setup_through_bluetooth_only[GS3 AV-entities1][switch.GS012345]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'GS012345',
}),
'context': <ANY>,
'entity_id': 'switch.gs012345',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'on',
})
# ---
# name: test_setup_through_bluetooth_only[GS3 AV-entities1][switch.GS012345_smart_standby_enabled]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'GS012345 Smart standby enabled',
}),
'context': <ANY>,
'entity_id': 'switch.gs012345_smart_standby_enabled',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'off',
})
# ---
# name: test_setup_through_bluetooth_only[GS3 AV-entities1][switch.GS012345_steam_boiler]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'GS012345 Steam boiler',
}),
'context': <ANY>,
'entity_id': 'switch.gs012345_steam_boiler',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'on',
})
# ---
# name: test_setup_through_bluetooth_only[Linea Micra-entities0][binary_sensor.MR012345_water_tank_empty]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'problem',
'friendly_name': 'MR012345 Water tank empty',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.mr012345_water_tank_empty',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'on',
})
# ---
# name: test_setup_through_bluetooth_only[Linea Micra-entities0][device_bluetooth_MR012345]
DeviceRegistryEntrySnapshot({
'area_id': None,
'config_entries': <ANY>,
'config_entries_subentries': <ANY>,
'configuration_url': None,
'connections': set({
tuple(
'bluetooth',
'00:00:00:00:00:00',
),
}),
'disabled_by': None,
'entry_type': None,
'hw_version': None,
'id': <ANY>,
'identifiers': set({
tuple(
'lamarzocco',
'MR012345',
),
}),
'labels': set({
}),
'manufacturer': 'La Marzocco',
'model': 'Linea Micra',
'model_id': 'LINEAMICRA',
'name': 'MR012345',
'name_by_user': None,
'primary_config_entry': <ANY>,
'serial_number': 'MR012345',
'sw_version': None,
'via_device_id': None,
})
# ---
# name: test_setup_through_bluetooth_only[Linea Micra-entities0][number.MR012345_coffee_target_temperature]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'temperature',
'friendly_name': 'MR012345 Coffee target temperature',
'max': 104,
'min': 85,
'mode': <NumberMode.AUTO: 'auto'>,
'step': 0.1,
'unit_of_measurement': <UnitOfTemperature.CELSIUS: '°C'>,
}),
'context': <ANY>,
'entity_id': 'number.mr012345_coffee_target_temperature',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '94.0',
})
# ---
# name: test_setup_through_bluetooth_only[Linea Micra-entities0][number.MR012345_smart_standby_time]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'duration',
'friendly_name': 'MR012345 Smart standby time',
'max': 240,
'min': 0,
'mode': <NumberMode.AUTO: 'auto'>,
'step': 1,
'unit_of_measurement': <UnitOfTime.MINUTES: 'min'>,
}),
'context': <ANY>,
'entity_id': 'number.mr012345_smart_standby_time',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '0',
})
# ---
# name: test_setup_through_bluetooth_only[Linea Micra-entities0][select.MR012345_steam_level]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'MR012345 Steam level',
'options': list([
'1',
'2',
'3',
]),
}),
'context': <ANY>,
'entity_id': 'select.mr012345_steam_level',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '2',
})
# ---
# name: test_setup_through_bluetooth_only[Linea Micra-entities0][switch.MR012345]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'MR012345',
}),
'context': <ANY>,
'entity_id': 'switch.mr012345',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'off',
})
# ---
# name: test_setup_through_bluetooth_only[Linea Micra-entities0][switch.MR012345_smart_standby_enabled]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'MR012345 Smart standby enabled',
}),
'context': <ANY>,
'entity_id': 'switch.mr012345_smart_standby_enabled',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'off',
})
# ---
# name: test_setup_through_bluetooth_only[Linea Micra-entities0][switch.MR012345_steam_boiler]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'MR012345 Steam boiler',
}),
'context': <ANY>,
'entity_id': 'switch.mr012345_steam_boiler',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'on',
})
# ---

3
tests/components/lamarzocco/test_binary_sensor.py
View File

@@ -56,7 +56,6 @@ async def test_sensor_going_unavailable(
hass: HomeAssistant,
mock_lamarzocco: MagicMock,
mock_config_entry: MockConfigEntry,
mock_cloud_client: MagicMock,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test sensor is going unavailable after an unsuccessful update."""
@@ -70,7 +69,7 @@ async def test_sensor_going_unavailable(
assert state.state != STATE_UNAVAILABLE
mock_lamarzocco.websocket.connected = False
mock_cloud_client.async_get_access_token.side_effect = RequestNotSuccessful("")
mock_lamarzocco.ensure_token_valid.side_effect = RequestNotSuccessful("")
freezer.tick(timedelta(minutes=10))
async_fire_time_changed(hass)
await hass.async_block_till_done()

365
tests/components/lamarzocco/test_bluetooth.py Normal file
View File

@@ -0,0 +1,365 @@
"""Tests for La Marzocco Bluetooth connection."""
from datetime import timedelta
from unittest.mock import MagicMock, patch
from bleak.backends.device import BLEDevice
from freezegun.api import FrozenDateTimeFactory
from pylamarzocco.const import MachineMode, ModelName, WidgetType
from pylamarzocco.exceptions import BluetoothConnectionFailed, RequestNotSuccessful
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.lamarzocco.const import DOMAIN
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import (
EVENT_HOMEASSISTANT_STOP,
STATE_OFF,
STATE_ON,
STATE_UNAVAILABLE,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr
from . import async_init_integration, get_bluetooth_service_info
from tests.common import MockConfigEntry, async_fire_time_changed
# Entities with bt_offline_mode=True
BLUETOOTH_ONLY_BASE_ENTITIES = [
("binary_sensor", "water_tank_empty"),
("switch", ""),
("switch", "steam_boiler"),
("number", "coffee_target_temperature"),
("switch", "smart_standby_enabled"),
("number", "smart_standby_time"),
]
MICRA_BT_OFFLINE_ENTITIES = [
*BLUETOOTH_ONLY_BASE_ENTITIES,
("select", "steam_level"),
]
GS3_BT_OFFLINE_ENTITIES = [
*BLUETOOTH_ONLY_BASE_ENTITIES,
("number", "steam_target_temperature"),
]
def build_entity_id(
platform: str,
serial_number: str,
entity_suffix: str,
) -> str:
"""Build full entity ID."""
if entity_suffix:
return f"{platform}.{serial_number}_{entity_suffix}"
return f"{platform}.{serial_number}"
async def test_bluetooth_coordinator_updates_based_on_websocket_state(
hass: HomeAssistant,
mock_lamarzocco: MagicMock,
mock_config_entry_bluetooth: MockConfigEntry,
mock_ble_device_from_address: MagicMock,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test Bluetooth coordinator updates based on websocket connection state."""
mock_lamarzocco.websocket.connected = False
await async_init_integration(hass, mock_config_entry_bluetooth)
await hass.async_block_till_done()
# Reset call count after initial setup
mock_lamarzocco.get_dashboard_from_bluetooth.reset_mock()
# Test 1: When websocket is connected, Bluetooth should skip updates
mock_lamarzocco.websocket.connected = True
mock_lamarzocco.dashboard.connected = True
freezer.tick(timedelta(seconds=61))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert not mock_lamarzocco.get_dashboard_from_bluetooth.called
# Test 2: When websocket is disconnected, Bluetooth should update
mock_lamarzocco.dashboard.connected = False
freezer.tick(timedelta(seconds=61))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert mock_lamarzocco.get_dashboard_from_bluetooth.called
@pytest.mark.parametrize(
("device_fixture", "entities"),
[
(ModelName.LINEA_MICRA, MICRA_BT_OFFLINE_ENTITIES),
(ModelName.GS3_AV, GS3_BT_OFFLINE_ENTITIES),
],
)
async def test_bt_offline_mode_entity_available_when_cloud_fails(
hass: HomeAssistant,
mock_lamarzocco: MagicMock,
mock_config_entry_bluetooth: MockConfigEntry,
freezer: FrozenDateTimeFactory,
device_fixture: ModelName,
entities: list[tuple[str, str]],
) -> None:
"""Test entities with bt_offline_mode=True remain available when cloud coordinators fail."""
await async_init_integration(hass, mock_config_entry_bluetooth)
# Check all entities are initially available
for entity_id in entities:
state = hass.states.get(
build_entity_id(entity_id[0], mock_lamarzocco.serial_number, entity_id[1])
)
assert state
assert state.state != STATE_UNAVAILABLE
# Simulate cloud coordinator failures
mock_lamarzocco.websocket.connected = False
mock_lamarzocco.get_dashboard.side_effect = RequestNotSuccessful("")
# Trigger update
freezer.tick(timedelta(seconds=61))
async_fire_time_changed(hass)
await hass.async_block_till_done()
# All bt_offline_mode entities should still be available
for entity_id in entities:
state = hass.states.get(
build_entity_id(entity_id[0], mock_lamarzocco.serial_number, entity_id[1])
)
assert state
assert state.state != STATE_UNAVAILABLE
async def test_entity_without_bt_becomes_unavailable_when_cloud_fails_no_bt(
hass: HomeAssistant,
mock_lamarzocco: MagicMock,
mock_config_entry: MockConfigEntry,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test entities become unavailable when cloud fails and no bluetooth coordinator exists."""
await async_init_integration(hass, mock_config_entry)
# Water tank sensor (even with bt_offline_mode=True, needs BT coordinator to work)
water_tank_sensor = (
f"binary_sensor.{mock_lamarzocco.serial_number}_water_tank_empty"
)
state = hass.states.get(water_tank_sensor)
assert state
# Initially should be available
initial_state = state.state
assert initial_state != STATE_UNAVAILABLE
# Simulate cloud coordinator failures without bluetooth fallback
mock_lamarzocco.websocket.connected = False
mock_lamarzocco.ensure_token_valid.side_effect = RequestNotSuccessful("")
# Trigger update
freezer.tick(timedelta(seconds=61))
async_fire_time_changed(hass)
await hass.async_block_till_done()
# Water tank sensor should become unavailable because cloud failed and no BT
state = hass.states.get(water_tank_sensor)
assert state
assert state.state == STATE_UNAVAILABLE
async def test_bluetooth_coordinator_handles_connection_failure(
hass: HomeAssistant,
mock_lamarzocco: MagicMock,
mock_config_entry_bluetooth: MockConfigEntry,
mock_ble_device_from_address: MagicMock,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test Bluetooth coordinator handles connection failures gracefully."""
# Start with websocket terminated to ensure Bluetooth coordinator is active
mock_lamarzocco.websocket.connected = False
await async_init_integration(hass, mock_config_entry_bluetooth)
# Water tank sensor has bt_offline_mode=True
water_tank_sensor = (
f"binary_sensor.{mock_lamarzocco.serial_number}_water_tank_empty"
)
state = hass.states.get(water_tank_sensor)
assert state
assert state.state != STATE_UNAVAILABLE
# Simulate Bluetooth connection failure
mock_lamarzocco.websocket.connected = False
mock_lamarzocco.dashboard.connected = False
mock_lamarzocco.get_dashboard_from_bluetooth.side_effect = (
BluetoothConnectionFailed("")
)
# Trigger Bluetooth coordinator update
freezer.tick(timedelta(seconds=61))
async_fire_time_changed(hass)
await hass.async_block_till_done()
# now it should be unavailable due to BT failure
state = hass.states.get(water_tank_sensor)
assert state
assert state.state == STATE_UNAVAILABLE
async def test_bluetooth_coordinator_triggers_entity_updates(
hass: HomeAssistant,
mock_lamarzocco: MagicMock,
mock_config_entry_bluetooth: MockConfigEntry,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test Bluetooth coordinator updates trigger entity state updates."""
mock_lamarzocco.dashboard.config[
WidgetType.CM_MACHINE_STATUS
].mode = MachineMode.STANDBY
await async_init_integration(hass, mock_config_entry_bluetooth)
main_switch = f"switch.{mock_lamarzocco.serial_number}"
state = hass.states.get(main_switch)
assert state
assert state.state == STATE_OFF
# Simulate Bluetooth update changing machine mode to brewing
mock_lamarzocco.dashboard.config[
WidgetType.CM_MACHINE_STATUS
].mode = MachineMode.BREWING_MODE
mock_lamarzocco.websocket.connected = False
mock_lamarzocco.dashboard.connected = False
# Trigger Bluetooth coordinator update
freezer.tick(timedelta(seconds=61))
async_fire_time_changed(hass)
await hass.async_block_till_done()
# Verify entity state was updated
state = hass.states.get(main_switch)
assert state
assert state.state == STATE_ON
@pytest.mark.parametrize(
("device_fixture", "entities"),
[
(ModelName.LINEA_MICRA, MICRA_BT_OFFLINE_ENTITIES),
(ModelName.GS3_AV, GS3_BT_OFFLINE_ENTITIES),
],
)
async def test_setup_through_bluetooth_only(
hass: HomeAssistant,
mock_config_entry_bluetooth: MockConfigEntry,
mock_lamarzocco_bluetooth: MagicMock,
mock_ble_device_from_address: MagicMock,
mock_cloud_client: MagicMock,
device_registry: dr.DeviceRegistry,
device_fixture: ModelName,
entities: list[tuple[str, str]],
snapshot: SnapshotAssertion,
) -> None:
"""Test we can setup without a cloud connection."""
# Simulate cloud connection failures
mock_cloud_client.get_thing_settings.side_effect = RequestNotSuccessful("")
mock_cloud_client.async_get_access_token.side_effect = RequestNotSuccessful("")
mock_lamarzocco_bluetooth.get_dashboard.side_effect = RequestNotSuccessful("")
mock_lamarzocco_bluetooth.get_coffee_and_flush_counter.side_effect = (
RequestNotSuccessful("")
)
mock_lamarzocco_bluetooth.get_schedule.side_effect = RequestNotSuccessful("")
mock_lamarzocco_bluetooth.get_settings.side_effect = RequestNotSuccessful("")
await async_init_integration(hass, mock_config_entry_bluetooth)
assert mock_config_entry_bluetooth.state is ConfigEntryState.LOADED
# Check all Bluetooth entities are available
for entity_id in entities:
entity = build_entity_id(
entity_id[0], mock_lamarzocco_bluetooth.serial_number, entity_id[1]
)
state = hass.states.get(entity)
assert state
assert state.state != STATE_UNAVAILABLE
assert state == snapshot(name=entity)
# snapshot device
device = device_registry.async_get_device(
{(DOMAIN, mock_lamarzocco_bluetooth.serial_number)}
)
assert device
assert device == snapshot(
name=f"device_bluetooth_{mock_lamarzocco_bluetooth.serial_number}"
)
@pytest.mark.parametrize(
("mock_ble_device", "has_client"),
[
(None, False),
(
BLEDevice(
address="aa:bb:cc:dd:ee:ff",
name="name",
details={},
),
True,
),
],
)
async def test_bluetooth_is_set_from_discovery(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_lamarzocco: MagicMock,
mock_cloud_client: MagicMock,
mock_ble_device: BLEDevice | None,
has_client: bool,
mock_ble_device_from_address: MagicMock,
) -> None:
"""Check we can fill a device from discovery info."""
service_info = get_bluetooth_service_info(
ModelName.GS3_MP, mock_lamarzocco.serial_number
)
mock_cloud_client.get_thing_settings.return_value.ble_auth_token = "token"
with (
patch(
"homeassistant.components.lamarzocco.async_discovered_service_info",
return_value=[service_info],
) as discovery,
patch(
"homeassistant.components.lamarzocco.LaMarzoccoMachine"
) as mock_machine_class,
):
mock_machine_class.return_value = mock_lamarzocco
await async_init_integration(hass, mock_config_entry)
discovery.assert_called_once()
assert mock_machine_class.call_count == 1
_, kwargs = mock_machine_class.call_args
assert (kwargs["bluetooth_client"] is not None) == has_client
assert mock_config_entry.data["mac"] == service_info.address
assert mock_config_entry.data["token"] == "token"
async def test_disconnect_on_stop(
hass: HomeAssistant,
mock_config_entry_bluetooth: MockConfigEntry,
mock_ble_device_from_address: MagicMock,
mock_bluetooth_client: MagicMock,
) -> None:
"""Test we close the connection with the La Marzocco when Home Assistant stops."""
await async_init_integration(hass, mock_config_entry_bluetooth)
await hass.async_block_till_done()
assert mock_config_entry_bluetooth.state is ConfigEntryState.LOADED
hass.bus.async_fire(EVENT_HOMEASSISTANT_STOP)
await hass.async_block_till_done()
mock_bluetooth_client.disconnect.assert_awaited_once()

131
tests/components/lamarzocco/test_init.py
View File

@@ -1,11 +1,10 @@
"""Test initialization of lamarzocco."""
from datetime import timedelta
from unittest.mock import AsyncMock, MagicMock, patch
from unittest.mock import AsyncMock, MagicMock, PropertyMock
from bleak.backends.device import BLEDevice
from freezegun.api import FrozenDateTimeFactory
from pylamarzocco.const import FirmwareType, ModelName
from pylamarzocco.const import FirmwareType
from pylamarzocco.exceptions import AuthFail, RequestNotSuccessful
from pylamarzocco.models import WebSocketDetails
import pytest
@@ -26,12 +25,7 @@ from homeassistant.helpers import (
issue_registry as ir,
)
from . import (
MOCK_INSTALLATION_KEY,
USER_INPUT,
async_init_integration,
get_bluetooth_service_info,
)
from . import MOCK_INSTALLATION_KEY, USER_INPUT, async_init_integration
from tests.common import MockConfigEntry, async_fire_time_changed
@@ -57,6 +51,7 @@ async def test_config_entry_not_ready(
mock_lamarzocco: MagicMock,
) -> None:
"""Test the La Marzocco configuration entry not ready."""
mock_lamarzocco.bluetooth_client_available = False
mock_lamarzocco.websocket.connected = False
mock_lamarzocco.get_dashboard.side_effect = RequestNotSuccessful("")
@@ -197,58 +192,6 @@ async def test_config_flow_entry_migration_downgrade(
assert not await hass.config_entries.async_setup(entry.entry_id)
@pytest.mark.parametrize(
("ble_device", "has_client"),
[
(None, False),
(
BLEDevice(
address="aa:bb:cc:dd:ee:ff",
name="name",
details={},
),
True,
),
],
)
async def test_bluetooth_is_set_from_discovery(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_lamarzocco: MagicMock,
mock_cloud_client: MagicMock,
ble_device: BLEDevice | None,
has_client: bool,
) -> None:
"""Check we can fill a device from discovery info."""
service_info = get_bluetooth_service_info(
ModelName.GS3_MP, mock_lamarzocco.serial_number
)
mock_cloud_client.get_thing_settings.return_value.ble_auth_token = "token"
with (
patch(
"homeassistant.components.lamarzocco.async_discovered_service_info",
return_value=[service_info],
) as discovery,
patch(
"homeassistant.components.lamarzocco.LaMarzoccoMachine"
) as mock_machine_class,
patch(
"homeassistant.components.lamarzocco.async_ble_device_from_address",
return_value=ble_device,
),
):
mock_machine_class.return_value = mock_lamarzocco
await async_init_integration(hass, mock_config_entry)
discovery.assert_called_once()
assert mock_machine_class.call_count == 1
_, kwargs = mock_machine_class.call_args
assert (kwargs["bluetooth_client"] is not None) == has_client
assert mock_config_entry.data[CONF_MAC] == service_info.address
assert mock_config_entry.data[CONF_TOKEN] == "token"
async def test_websocket_closed_on_unload(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
@@ -335,61 +278,16 @@ async def test_device(
assert device == snapshot
async def test_disconnect_on_stop(
hass: HomeAssistant,
mock_lamarzocco: MagicMock,
mock_ble_device: BLEDevice,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test we close the connection with the La Marzocco when Home Assistants stops."""
mock_config_entry = MockConfigEntry(
title="My LaMarzocco",
domain=DOMAIN,
version=4,
data=USER_INPUT
| {
CONF_MAC: mock_ble_device.address,
CONF_TOKEN: "token",
CONF_INSTALLATION_KEY: MOCK_INSTALLATION_KEY,
},
unique_id=mock_lamarzocco.serial_number,
)
with (
patch(
"homeassistant.components.lamarzocco.async_ble_device_from_address",
return_value=mock_ble_device,
),
patch(
"homeassistant.components.lamarzocco.LaMarzoccoBluetoothClient",
autospec=True,
) as mock_bt_client_cls,
):
mock_bt_client = mock_bt_client_cls.return_value
mock_bt_client.disconnect = AsyncMock()
await async_init_integration(hass, mock_config_entry)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.LOADED
hass.bus.async_fire(EVENT_HOMEASSISTANT_STOP)
await hass.async_block_till_done()
mock_bt_client.disconnect.assert_awaited_once()
async def test_websocket_reconnects_after_termination(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_lamarzocco: MagicMock,
freezer: FrozenDateTimeFactory,
mock_websocket_terminated: PropertyMock,
) -> None:
"""Test the websocket reconnects after background task terminates."""
# Setup: websocket connected initially
mock_websocket = MagicMock()
mock_websocket.closed = False
mock_lamarzocco.websocket = WebSocketDetails(mock_websocket, None)
# Setup: websocket disconnected initially
mock_websocket_terminated.return_value = True
await async_init_integration(hass, mock_config_entry)
@@ -397,17 +295,12 @@ async def test_websocket_reconnects_after_termination(
assert mock_lamarzocco.connect_dashboard_websocket.call_count == 1
# Simulate websocket disconnection (e.g., after internet outage)
mock_websocket.closed = True
mock_websocket_terminated.return_value = True
# Simulate the background task terminating by patching websocket_terminated
with patch(
"homeassistant.components.lamarzocco.coordinator.LaMarzoccoConfigUpdateCoordinator.websocket_terminated",
new=True,
):
# Trigger the coordinator's update (which runs every 60 seconds)
freezer.tick(timedelta(seconds=61))
async_fire_time_changed(hass)
await hass.async_block_till_done()
# Trigger the coordinator's update (which runs every 60 seconds)
freezer.tick(timedelta(seconds=61))
async_fire_time_changed(hass)
await hass.async_block_till_done()
# Verify websocket reconnection was attempted
assert mock_lamarzocco.connect_dashboard_websocket.call_count == 2

2
tests/components/lamarzocco/test_sensor.py
View File

@@ -14,8 +14,6 @@ from . import async_init_integration
from tests.common import MockConfigEntry, snapshot_platform
pytestmark = pytest.mark.usefixtures("mock_websocket_terminated")
async def test_sensors(
hass: HomeAssistant,