bugfixes and more tests

Author: zweckj

lmcloud/client_bluetooth.py

@@ -115,7 +115,7 @@ async def set_temp(self, boiler: BoilerType, temperature: float) -> None:
         data = {
             "name": "SettingBoilerTarget",
             "parameter": {
-                "identifier": boiler,
+                "identifier": boiler.value,
                 "value": temperature,
             },
         }

lmcloud/client_cloud.py

@@ -21,6 +21,7 @@
     TOKEN_URL,
     BoilerType,
     FirmwareType,
+    PhysicalKey,
     PrebrewMode,
 )
 from .exceptions import AuthFail, RequestNotSuccessful
@@ -142,7 +143,7 @@ async def set_steam(
         """Turn Steamboiler on or off"""
 
         data = {
-            "identifier": BoilerType.STEAM,
+            "identifier": BoilerType.STEAM.value,
             "state": enabled,
         }
         url = f"{GW_MACHINE_BASE_URL}/{serial_number}/enable-boiler"
@@ -159,7 +160,7 @@ async def set_temp(
     ) -> bool:
         """Set boiler temperature (in Celsius)."""
 
-        data = {"identifier": boiler, "value": temperature}
+        data = {"identifier": boiler.value, "value": temperature}
         url = f"{GW_MACHINE_BASE_URL}/{serial_number}/target-boiler"
         response = await self._rest_api_call(url=url, method=HTTPMethod.POST, data=data)
         if await self._check_cloud_command_status(serial_number, response):
@@ -175,7 +176,7 @@ async def set_prebrew_mode(
         """Enable/Disable Pre-Brew or Pre-Infusion (mutually exclusive)."""
 
         url = f"{GW_MACHINE_BASE_URL}/{serial_number}/enable-preinfusion"
-        data = {"mode": mode}
+        data = {"mode": mode.value}
         response = await self._rest_api_call(url=url, method=HTTPMethod.POST, data=data)
         if await self._check_cloud_command_status(serial_number, response):
             return True
@@ -186,20 +187,20 @@ async def configure_pre_brew_infusion_time(
         serial_number: str,
         on_time: float,
         off_time: float,
-        key: int = 1,
+        key: PhysicalKey,
     ) -> bool:
         """Set Pre-Brew details. Also used for preinfusion (prebrewOnTime=0, prebrewOnTime=ms)."""
 
-        on_time = round(on_time, 1) * 100
-        off_time = round(off_time, 1) * 100
-        button = f"Dose{chr(key + 64)}"
+        on_time = round(on_time, 1) * 1000
+        off_time = round(off_time, 1) * 1000
+        button = f"Dose{key.name}"
 
         url = f"{GW_MACHINE_BASE_URL}/{serial_number}/setting-preinfusion"
         data = {
             "button": button,
             "group": "Group1",
-            "holdTimeMs": off_time,
-            "wetTimeMs": on_time,
+            "holdTimeMs": int(off_time),
+            "wetTimeMs": int(on_time),
         }
         response = await self._rest_api_call(url=url, method=HTTPMethod.POST, data=data)
         if await self._check_cloud_command_status(serial_number, response):
@@ -223,12 +224,12 @@ async def enable_plumbin(
     async def set_dose(
         self,
         serial_number: str,
-        key: int,
+        key: PhysicalKey,
         value: int,
     ) -> bool:
         """Set the value for a dose"""
 
-        dose_index = f"Dose{chr(key + 64)}"
+        dose_index = f"Dose{key.name}"
 
         url = f"{GW_MACHINE_BASE_URL}/{serial_number}/dose"
         data = {
@@ -343,7 +344,7 @@ async def update_firmware(
         """Update Firmware."""
 
         _LOGGER.debug("Updating firmware for component %s", component)
-        url = f"{GW_MACHINE_BASE_URL}/{serial_number}/firmware/{component}/update"
+        url = f"{GW_MACHINE_BASE_URL}/{serial_number}/firmware/{component.value}/update"
         await self._rest_api_call(url=url, method=HTTPMethod.POST, data={})
         retry_counter = 0
         while retry_counter <= 20:

lmcloud/lm_machine.py

@@ -239,7 +239,7 @@ async def set_prebrew_time(
         self,
         prebrew_on_time: float | None = None,
         prebrew_off_time: float | None = None,
-        key: PhysicalKey = PhysicalKey(1),
+        key: PhysicalKey = PhysicalKey.A,
     ) -> bool:
         """Set prebrew time"""
 
@@ -260,7 +260,7 @@ async def set_prebrew_time(
     async def set_preinfusion_time(
         self,
         preinfusion_time: float,
-        key: PhysicalKey = PhysicalKey(1),
+        key: PhysicalKey = PhysicalKey.A,
     ) -> bool:
         """Set preinfusion time"""
 

setup.py

@@ -7,7 +7,7 @@
 
 setuptools.setup(
     name="lmcloud",
-    version="1.0.0b14",
+    version="1.0.0b15",
     description="A Python implementation of the new La Marzocco API",
     long_description=readme,
     long_description_content_type="text/markdown",

tests/__init__.py

@@ -1,14 +1,18 @@
 """Test utilities"""
 
 from lmcloud import LaMarzoccoCloudClient, LaMarzoccoMachine, LaMarzoccoGrinder
+from lmcloud.client_bluetooth import LaMarzoccoBluetoothClient
+from lmcloud.client_local import LaMarzoccoLocalClient
 from lmcloud.const import MachineModel, GrinderModel
 
 MACHINE_SERIAL = "GS01234"
 GRINDER_SERIAL = "G00000000000"
 
 
 async def init_machine(
-    cloud_client: LaMarzoccoCloudClient,
+    cloud_client: LaMarzoccoCloudClient | None = None,
+    bluetooth_client: LaMarzoccoBluetoothClient | None = None,
+    local_client: LaMarzoccoLocalClient | None = None,
 ) -> LaMarzoccoMachine:
     """Get an initialized machine"""
 
@@ -17,6 +21,8 @@ async def init_machine(
         serial_number=MACHINE_SERIAL,
         name="MyMachine",
         cloud_client=cloud_client,
+        local_client=local_client,
+        bluetooth_client=bluetooth_client,
     )
     return machine
 

tests/conftest.py

@@ -1,14 +1,18 @@
 """Fixtures for the tests."""
 
+# pylint: disable=W0212
+
 import json
 from collections.abc import Generator
 from http import HTTPMethod
 from pathlib import Path
 from unittest.mock import AsyncMock
 
 import pytest
+from bleak import BLEDevice, BleakError
 from httpx import Response
 
+from lmcloud.client_bluetooth import LaMarzoccoBluetoothClient
 from lmcloud.client_cloud import LaMarzoccoCloudClient
 from lmcloud.client_local import LaMarzoccoLocalClient
 
@@ -69,7 +73,7 @@ def cloud_client() -> Generator[LaMarzoccoCloudClient, None, None]:
     oauth_client = AsyncMock()
     oauth_client.request.side_effect = get_mock_response
 
-    client._oauth_client = oauth_client  # pylint: disable=protected-access
+    client._oauth_client = oauth_client
     yield client
 
 
@@ -80,3 +84,20 @@ def local_machine_client() -> Generator[LaMarzoccoLocalClient, None, None]:
     httpx_client.get.side_effect = get_local_machine_mock_response
     client = LaMarzoccoLocalClient("192.168.1.42", "secret", client=httpx_client)
     yield client
+
+
+@pytest.fixture
+def bluetooth_client() -> Generator[LaMarzoccoBluetoothClient, None, None]:
+    """Fixture for a bluetooth client."""
+    ble_device = BLEDevice(
+        address="00:11:22:33:44:55",
+        name="MyMachine",
+        details={"path": "path/to/device"},
+        rssi=50,
+    )
+
+    bt_client = LaMarzoccoBluetoothClient("username", "serial", "token", ble_device)
+    bt_client._client = AsyncMock()
+    bt_client._client.is_connected = True
+    bt_client._client.write_gatt_char.side_effect = BleakError("Failed to write")
+    yield bt_client

tests/test_machine.py

@@ -1,10 +1,14 @@
 """Test the LaMarzoccoMachine class."""
 
+# pylint: disable=W0212
+
 from unittest.mock import AsyncMock, patch
 
 import pytest
 from syrupy import SnapshotAssertion
+from http import HTTPMethod
 
+from lmcloud.client_bluetooth import LaMarzoccoBluetoothClient
 from lmcloud.client_cloud import LaMarzoccoCloudClient
 from lmcloud.client_local import LaMarzoccoLocalClient
 from lmcloud.const import BoilerType, MachineModel, PhysicalKey, WeekDay
@@ -31,12 +35,7 @@ async def test_local_client(
 ) -> None:
     """Ensure that the local client delivers same result"""
 
-    machine = await LaMarzoccoMachine.create(
-        model=MachineModel.GS3_AV,
-        serial_number=MACHINE_SERIAL,
-        name="MyMachine",
-        local_client=local_machine_client,
-    )
+    machine = await init_machine(local_client=local_machine_client)
 
     machine2 = await init_machine(cloud_client)
 
@@ -103,13 +102,7 @@ async def test_websocket_message(
     snapshot: SnapshotAssertion,
 ):
     """Test parsing of websocket messages."""
-    machine = await LaMarzoccoMachine.create(
-        model=MachineModel.GS3_AV,
-        serial_number=MACHINE_SERIAL,
-        name="MyMachine",
-        cloud_client=cloud_client,
-        local_client=local_machine_client,
-    )
+    machine = await init_machine(cloud_client, local_client=local_machine_client)
 
     message = r'[{"Boilers":"[{\"id\":\"SteamBoiler\",\"isEnabled\":true,\"target\":131,\"current\":113},{\"id\":\"CoffeeBoiler1\",\"isEnabled\":true,\"target\":94,\"current\":81}]"}]'
     machine.on_websocket_message_received(message)
@@ -118,3 +111,87 @@ async def test_websocket_message(
     message = r'[{"BoilersTargetTemperature":"{\"SteamBoiler\":131,\"CoffeeBoiler1\":94}"},{"Boilers":"[{\"id\":\"SteamBoiler\",\"isEnabled\":true,\"target\":131,\"current\":50},{\"id\":\"CoffeeBoiler1\",\"isEnabled\":true,\"target\":94,\"current\":36}]"}]'
     machine.on_websocket_message_received(message)
     assert machine.config == snapshot
+
+
+async def test_set_power(
+    cloud_client: LaMarzoccoCloudClient, bluetooth_client: LaMarzoccoBluetoothClient
+):
+    """Test setting the power."""
+    machine = await init_machine(cloud_client, bluetooth_client=bluetooth_client)
+
+    assert await machine.set_power(True)
+
+    bluetooth_client._client.write_gatt_char.assert_called_once_with(  # type: ignore[attr-defined]
+        "050b7847-e12b-09a8-b04b-8e0922a9abab",
+        b'{"name":"MachineChangeMode","parameter":{"mode":"BrewingMode"}}\x00',
+    )
+    cloud_client._oauth_client.request.assert_any_call(  # type: ignore[union-attr]
+        HTTPMethod.POST,
+        "https://gw-lmz.lamarzocco.io/v1/home/machines/GS01234/status",
+        json={"status": "BrewingMode"},
+    )
+
+
+async def test_set_steam(
+    cloud_client: LaMarzoccoCloudClient, bluetooth_client: LaMarzoccoBluetoothClient
+):
+    """Test setting the steam."""
+    machine = await init_machine(cloud_client, bluetooth_client=bluetooth_client)
+
+    assert await machine.set_steam(True)
+
+    bluetooth_client._client.write_gatt_char.assert_called_once_with(  # type: ignore[attr-defined]
+        "050b7847-e12b-09a8-b04b-8e0922a9abab",
+        b'{"name":"SettingBoilerEnable","parameter":{"identifier":"SteamBoiler","state":true}}\x00',
+    )
+    cloud_client._oauth_client.request.assert_any_call(  # type: ignore[union-attr]
+        HTTPMethod.POST,
+        "https://gw-lmz.lamarzocco.io/v1/home/machines/GS01234/enable-boiler",
+        json={"identifier": "SteamBoiler", "state": True},
+    )
+
+
+async def test_set_temperature(
+    cloud_client: LaMarzoccoCloudClient, bluetooth_client: LaMarzoccoBluetoothClient
+):
+    """Test setting temperature."""
+    machine = await init_machine(cloud_client, bluetooth_client=bluetooth_client)
+
+    assert await machine.set_temp(BoilerType.STEAM, 131)
+
+    bluetooth_client._client.write_gatt_char.assert_called_once_with(  # type: ignore[attr-defined]
+        "050b7847-e12b-09a8-b04b-8e0922a9abab",
+        b'{"name":"SettingBoilerTarget","parameter":{"identifier":"SteamBoiler","value":131}}\x00',
+    )
+    cloud_client._oauth_client.request.assert_any_call(  # type: ignore[union-attr]
+        HTTPMethod.POST,
+        "https://gw-lmz.lamarzocco.io/v1/home/machines/GS01234/target-boiler",
+        json={"identifier": "SteamBoiler", "value": 131},
+    )
+
+
+async def test_set_prebrew_time(cloud_client: LaMarzoccoCloudClient):
+    """Test setting prebrew time."""
+    machine = await init_machine(
+        cloud_client,
+    )
+
+    assert await machine.set_prebrew_time(1.0, 3.5)
+
+    cloud_client._oauth_client.request.assert_any_call(  # type: ignore[union-attr]
+        HTTPMethod.POST,
+        "https://gw-lmz.lamarzocco.io/v1/home/machines/GS01234/setting-preinfusion",
+        json={
+            "button": "DoseA",
+            "group": "Group1",
+            "holdTimeMs": 3500,
+            "wetTimeMs": 1000,
+        },
+    )
+
+    assert await machine.set_preinfusion_time(4.5)
+    cloud_client._oauth_client.request.assert_any_call(  # type: ignore[union-attr]
+        HTTPMethod.POST,
+        "https://gw-lmz.lamarzocco.io/v1/home/machines/GS01234/setting-preinfusion",
+        json={"button": "DoseA", "group": "Group1", "holdTimeMs": 4500, "wetTimeMs": 0},
+    )