Add driver firmware v15 commands (#1260)

* Add driver firmware v15 commands

* Add parameters to async_firmware_update_otw

* Bump min and max schema version to 44

* Put async_is_firmware_update_in_progress back

* Fix updateInfo parameter

* Require schema 44 for async_firmware_update_otw

Author: MartinHjelmare

test/conftest.py

@@ -246,7 +246,7 @@ def version_data_fixture() -> dict[str, Any]:
         "serverVersion": "test_server_version",
         "homeId": "test_home_id",
         "minSchemaVersion": 0,
-        "maxSchemaVersion": 43,
+        "maxSchemaVersion": 44,
     }
 
 

test/model/common.py

@@ -0,0 +1,21 @@
+"""Provide common tools for testing Z-Wave JS server models."""
+
+from zwave_js_server.model.firmware import FirmwareUpdateInfoDataType
+
+FIRMWARE_UPDATE_INFO = FirmwareUpdateInfoDataType(
+    **{
+        "version": "1.0.0",
+        "changelog": "changelog",
+        "channel": "stable",
+        "files": [{"target": 0, "url": "http://example.com", "integrity": "test"}],
+        "downgrade": True,
+        "normalizedVersion": "1.0.0",
+        "device": {
+            "manufacturerId": 1,
+            "productType": 2,
+            "productId": 3,
+            "firmwareVersion": "0.4.4",
+            "rfRegion": 1,
+        },
+    }
+)

test/model/test_controller.py

@@ -37,22 +37,7 @@
 from zwave_js_server.model.node.firmware import NodeFirmwareUpdateInfo
 
 from .. import load_fixture
-
-FIRMWARE_UPDATE_INFO = {
-    "version": "1.0.0",
-    "changelog": "changelog",
-    "channel": "stable",
-    "files": [{"target": 0, "url": "http://example.com", "integrity": "test"}],
-    "downgrade": True,
-    "normalizedVersion": "1.0.0",
-    "device": {
-        "manufacturerId": 1,
-        "productType": 2,
-        "productId": 3,
-        "firmwareVersion": "0.4.4",
-        "rfRegion": 1,
-    },
-}
+from .common import FIRMWARE_UPDATE_INFO
 
 
 def test_from_state():

test/model/test_driver.py

@@ -1,29 +1,31 @@
 """Test the driver model."""
 
 import json
+from typing import Any
 
 import pytest
 
 from zwave_js_server.const import LogLevel
 from zwave_js_server.event import Event
 from zwave_js_server.model import (
-    driver as driver_pkg,
     log_config as log_config_pkg,
     log_message as log_message_pkg,
 )
+from zwave_js_server.model.driver import Driver
 from zwave_js_server.model.driver.firmware import DriverFirmwareUpdateStatus
+from zwave_js_server.model.firmware import FirmwareUpdateData, FirmwareUpdateInfo
 
 from .. import load_fixture
+from ..common import MockCommandProtocol
+from .common import FIRMWARE_UPDATE_INFO
 
 
 def test_from_state(client, log_config):
     """Test from_state method."""
     ws_msgs = load_fixture("basic_dump.txt").strip().split("\n")
 
-    driver = driver_pkg.Driver(
-        client, json.loads(ws_msgs[0])["result"]["state"], log_config
-    )
-    assert driver == driver_pkg.Driver(
+    driver = Driver(client, json.loads(ws_msgs[0])["result"]["state"], log_config)
+    assert driver == Driver(
         client, json.loads(ws_msgs[0])["result"]["state"], log_config
     )
     assert driver != driver.controller.home_id
@@ -452,3 +454,93 @@ async def test_firmware_events(driver):
     assert result.status == DriverFirmwareUpdateStatus.OK
     assert result.success
     assert driver.firmware_update_progress is None
+
+
+@pytest.mark.parametrize("progress", [True, False])
+async def test_is_otw_firmware_update_in_progress(
+    driver: Driver,
+    uuid4: str,
+    mock_command: MockCommandProtocol,
+    progress: bool,
+) -> None:
+    """Test is_otw_firmware_update_in_progress command."""
+    ack_commands = mock_command(
+        {"command": "driver.is_otw_firmware_update_in_progress"},
+        {"progress": progress},
+    )
+    assert await driver.async_is_otw_firmware_update_in_progress() is progress
+
+    assert len(ack_commands) == 1
+    assert ack_commands[0] == {
+        "command": "driver.is_otw_firmware_update_in_progress",
+        "messageId": uuid4,
+    }
+
+
+@pytest.mark.parametrize(
+    ("firmware_update_param", "expected_params"),
+    [
+        (
+            {
+                "update_data": FirmwareUpdateData(
+                    filename="test-file", file=b"test", file_format=None
+                )
+            },
+            FirmwareUpdateData(
+                filename="test-file", file=b"test", file_format=None
+            ).to_dict(),
+        ),
+        (
+            {"update_info": FirmwareUpdateInfo.from_dict(FIRMWARE_UPDATE_INFO)},
+            {"updateInfo": FIRMWARE_UPDATE_INFO},
+        ),
+    ],
+)
+@pytest.mark.parametrize(("status", "success"), [(1, True), (0, False)])
+async def test_firmware_update_otw(
+    driver: Driver,
+    uuid4: str,
+    mock_command: MockCommandProtocol,
+    status: int,
+    success: bool,
+    firmware_update_param: dict[str, Any],
+    expected_params: dict[str, Any],
+) -> None:
+    """Test firmware_update_otw command."""
+    ack_commands = mock_command(
+        {"command": "driver.firmware_update_otw", **expected_params},
+        {"result": {"status": status, "success": success}},
+    )
+    result = await driver.async_firmware_update_otw(**firmware_update_param)
+
+    assert result.status == status
+    assert result.success is success
+
+    assert len(ack_commands) == 1
+    assert ack_commands[0] == {
+        "command": "driver.firmware_update_otw",
+        "messageId": uuid4,
+        **expected_params,
+    }
+
+
+@pytest.mark.parametrize(
+    "firmware_update_param",
+    [
+        {},  # No parameters
+        {
+            "update_data": FirmwareUpdateData(
+                filename="test-file", file=b"test", file_format=None
+            ),
+            "update_info": FirmwareUpdateInfo.from_dict(FIRMWARE_UPDATE_INFO),
+        },  # Invalid parameters
+    ],
+)
+async def test_firmware_update_otw_invalid_params(
+    driver: Driver,
+    firmware_update_param: dict[str, FirmwareUpdateData | FirmwareUpdateInfo],
+) -> None:
+    """Test firmware_update_otw command invalid parameters."""
+    with pytest.raises(ValueError):
+        # Should raise ValueError if no parameters are provided or both are provided
+        await driver.async_firmware_update_otw(**firmware_update_param)

test/test_client.py

@@ -463,7 +463,7 @@ async def test_additional_user_agent_components(client_session, url):
             {
                 "command": "initialize",
                 "messageId": "initialize",
-                "schemaVersion": 43,
+                "schemaVersion": 44,
                 "additionalUserAgentComponents": {
                     "zwave-js-server-python": __version__,
                     "foo": "bar",

test/test_dump.py

@@ -106,7 +106,7 @@ async def test_dump_additional_user_agent_components(
         {
             "command": "initialize",
             "messageId": "initialize",
-            "schemaVersion": 43,
+            "schemaVersion": 44,
             "additionalUserAgentComponents": {
                 "zwave-js-server-python": __version__,
                 "foo": "bar",

test/test_main.py

@@ -58,7 +58,7 @@ def test_dump_state(
     assert captured.out == (
         "{'type': 'version', 'driverVersion': 'test_driver_version', "
         "'serverVersion': 'test_server_version', 'homeId': 'test_home_id', "
-        "'minSchemaVersion': 0, 'maxSchemaVersion': 43}\n"
+        "'minSchemaVersion': 0, 'maxSchemaVersion': 44}\n"
         "{'type': 'result', 'success': True, 'result': {}, 'messageId': 'initialize'}\n"
         "test_result\n"
     )

zwave_js_server/const/__init__.py

@@ -11,9 +11,9 @@
 __version__ = "0.66.0"
 
 # minimal server schema version we can handle
-MIN_SERVER_SCHEMA_VERSION = 41
+MIN_SERVER_SCHEMA_VERSION = 44
 # max server schema version we can handle (and our code is compatible with)
-MAX_SERVER_SCHEMA_VERSION = 43
+MAX_SERVER_SCHEMA_VERSION = 44
 
 VALUE_UNKNOWN = "unknown"
 

zwave_js_server/model/driver/__init__.py

@@ -4,6 +4,13 @@
 
 from typing import TYPE_CHECKING, Any, Literal, cast
 
+from zwave_js_server.model.firmware import (
+    FirmwareUpdateData,
+    FirmwareUpdateDataDataType,
+    FirmwareUpdateInfo,
+    FirmwareUpdateInfoDataType,
+)
+
 from ...event import BaseEventModel, Event, EventBase
 from ..config_manager import ConfigManager
 from ..controller import Controller
@@ -242,6 +249,38 @@ async def async_install_config_update(self) -> bool:
         )
         return cast(bool, result["success"])
 
+    async def async_firmware_update_otw(
+        self,
+        *,
+        update_data: FirmwareUpdateData | None = None,
+        update_info: FirmwareUpdateInfo | None = None,
+    ) -> DriverFirmwareUpdateResult:
+        """Send firmwareUpdateOTW command to Driver."""
+        if update_data is None and update_info is None:
+            raise ValueError(
+                "Either update_data or update_info must be provided for firmware update."
+            )
+        if update_data is not None and update_info is not None:
+            raise ValueError(
+                "Only one of update_data or update_info can be provided for firmware update."
+            )
+        params: FirmwareUpdateDataDataType | dict[str, FirmwareUpdateInfoDataType]
+        if update_data is not None:
+            params = update_data.to_dict()
+        elif update_info is not None:
+            params = {"updateInfo": update_info.to_dict()}
+        data = await self._async_send_command(
+            "firmware_update_otw", require_schema=44, **params
+        )
+        return DriverFirmwareUpdateResult(data["result"])
+
+    async def async_is_otw_firmware_update_in_progress(self) -> bool:
+        """Send isOTWFirmwareUpdateInProgress command to Driver."""
+        result = await self._async_send_command(
+            "is_otw_firmware_update_in_progress", require_schema=41
+        )
+        return cast(bool, result["progress"])
+
     async def async_set_preferred_scales(
         self, scales: dict[str | int, str | int]
     ) -> None: