Split update WebSocket command into two commands

Make check_node_update a separate WebSocket command which only
checks for updates. The update_node command then will download
and actually apply the update.

Author: agners

matter_server/common/models.py

@@ -47,6 +47,7 @@ class APICommand(str, Enum):
     PING_NODE = "ping_node"
     GET_NODE_IP_ADDRESSES = "get_node_ip_addresses"
     IMPORT_TEST_NODE = "import_test_node"
+    CHECK_NODE_UPDATE = "check_node_update"
     UPDATE_NODE = "update_node"
 
 

matter_server/server/device_controller.py

@@ -896,8 +896,8 @@ async def import_test_node(self, dump: str) -> None:
             self._nodes[node.node_id] = node
             self.server.signal_event(EventType.NODE_ADDED, node)
 
-    @api_command(APICommand.UPDATE_NODE)
-    async def update_node(self, node_id: int) -> dict | None:
+    @api_command(APICommand.CHECK_NODE_UPDATE)
+    async def check_node_update(self, node_id: int) -> dict | None:
         """
         Check if there is an update for a particular node.
 
@@ -906,8 +906,27 @@ async def update_node(self, node_id: int) -> dict | None:
         information of the latest update available.
         """
 
-        node_logger = LOGGER.getChild(f"node_{node_id}")
-        node = self._nodes[node_id]
+        return await self._check_node_update(node_id)
+
+    @api_command(APICommand.UPDATE_NODE)
+    async def update_node(self, node_id: int, software_version: int) -> dict | None:
+        """
+        Update a node to a new software version.
+
+        This command checks if the requested software version is indeed still available
+        and if so, it will start the update process. The update process will be handled
+        by the built-in OTA provider. The OTA provider will download the update and
+        notify the node about the new update.
+        """
+
+        update = await self._check_node_update(node_id, software_version)
+        if update is None:
+            logging.error(
+                "Software version %d is not available for node %d",
+                software_version,
+                node_id,
+            )
+            return None
 
         if self.chip_controller is None:
             raise RuntimeError("Device Controller not initialized.")
@@ -916,34 +935,7 @@ async def update_node(self, node_id: int) -> dict | None:
             LOGGER.warning("No OTA provider found, updates not possible.")
             return None
 
-        node_logger.debug("Check for updates.")
-        vid = cast(int, node.attributes.get(BASIC_INFORMATION_VENDOR_ID_ATTRIBUTE_PATH))
-        pid = cast(
-            int, node.attributes.get(BASIC_INFORMATION_PRODUCT_ID_ATTRIBUTE_PATH)
-        )
-        software_version = cast(
-            int, node.attributes.get(BASIC_INFORMATION_SOFTWARE_VERSION_ATTRIBUTE_PATH)
-        )
-        software_version_string = node.attributes.get(
-            BASIC_INFORMATION_SOFTWARE_VERSION_STRING_ATTRIBUTE_PATH
-        )
-
-        update = await check_for_update(vid, pid, software_version)
-        if not update:
-            node_logger.info("No new update found.")
-            return None
-
-        if "otaUrl" not in update:
-            node_logger.warning("Update found, but no OTA URL provided.")
-            return None
-
-        node_logger.info(
-            "New software update found: %s (current %s). Preparing updates...",
-            update["softwareVersionString"],
-            software_version_string,
-        )
-
-        # Add to OTA provider
+        # Add update to the OTA provider
         await self._ota_provider.download_update(update)
 
         ota_provider_node_id = self._ota_provider.get_node_id()
@@ -1042,6 +1034,44 @@ async def update_node(self, node_id: int) -> dict | None:
 
         return update
 
+    async def _check_node_update(
+        self,
+        node_id: int,
+        requested_software_version: int | None = None,
+    ) -> dict | None:
+        node_logger = LOGGER.getChild(f"node_{node_id}")
+        node = self._nodes[node_id]
+
+        node_logger.debug("Check for updates.")
+        vid = cast(int, node.attributes.get(BASIC_INFORMATION_VENDOR_ID_ATTRIBUTE_PATH))
+        pid = cast(
+            int, node.attributes.get(BASIC_INFORMATION_PRODUCT_ID_ATTRIBUTE_PATH)
+        )
+        software_version = cast(
+            int, node.attributes.get(BASIC_INFORMATION_SOFTWARE_VERSION_ATTRIBUTE_PATH)
+        )
+        software_version_string = node.attributes.get(
+            BASIC_INFORMATION_SOFTWARE_VERSION_STRING_ATTRIBUTE_PATH
+        )
+
+        update = await check_for_update(
+            vid, pid, software_version, requested_software_version
+        )
+        if not update:
+            node_logger.info("No new update found.")
+            return None
+
+        if "otaUrl" not in update:
+            node_logger.warning("Update found, but no OTA URL provided.")
+            return None
+
+        node_logger.info(
+            "New software update found: %s (current %s).",
+            update["softwareVersionString"],
+            software_version_string,
+        )
+        return update
+
     async def _subscribe_node(self, node_id: int) -> None:
         """
         Subscribe to all node state changes/events for an individual node.

matter_server/server/ota/__init__.py

@@ -19,10 +19,18 @@
 
 
 async def check_for_update(
-    vid: int, pid: int, current_software_version: int
+    vid: int,
+    pid: int,
+    current_software_version: int,
+    requested_software_version: int | None = None,
 ) -> None | dict:
     """Check for software updates."""
     if (vid, pid) in HARDCODED_UPDATES:
-        return HARDCODED_UPDATES[(vid, pid)]
+        update = HARDCODED_UPDATES[(vid, pid)]
+        if (
+            requested_software_version is None
+            or update["softwareVersion"] == requested_software_version
+        ):
+            return update
 
     return await dcl.check_for_update(vid, pid, current_software_version)

matter_server/server/ota/dcl.py

@@ -10,7 +10,7 @@
 LOGGER = logging.getLogger(__name__)
 
 
-async def get_software_versions(vid: int, pid: int) -> Any:
+async def _get_software_versions(vid: int, pid: int) -> Any:
     """Check DCL if there are updates available for a particular node."""
     async with ClientSession(raise_for_status=True) as http_session:
         # fetch the paa certificates list
@@ -20,7 +20,7 @@ async def get_software_versions(vid: int, pid: int) -> Any:
             return await response.json()
 
 
-async def get_software_version(vid: int, pid: int, software_version: int) -> Any:
+async def _get_software_version(vid: int, pid: int, software_version: int) -> Any:
     """Check DCL if there are updates available for a particular node."""
     async with ClientSession(raise_for_status=True) as http_session:
         # fetch the paa certificates list
@@ -30,12 +30,45 @@ async def get_software_version(vid: int, pid: int, software_version: int) -> Any
             return await response.json()
 
 
+async def _check_update_version(
+    vid: int, pid: int, version: int, current_software_version: int
+) -> None | dict:
+    version_res: dict = await _get_software_version(vid, pid, version)
+    if not isinstance(version_res, dict):
+        raise TypeError("Unexpected DCL response.")
+
+    if "modelVersion" not in version_res:
+        raise ValueError("Unexpected DCL response.")
+
+    version_candidate: dict = cast(dict, version_res["modelVersion"])
+
+    # Check minApplicableSoftwareVersion/maxApplicableSoftwareVersion
+    min_sw_version = version_candidate["minApplicableSoftwareVersion"]
+    max_sw_version = version_candidate["maxApplicableSoftwareVersion"]
+    if (
+        current_software_version < min_sw_version
+        or current_software_version > max_sw_version
+    ):
+        return None
+
+    return version_candidate
+
+
 async def check_for_update(
-    vid: int, pid: int, current_software_version: int
+    vid: int,
+    pid: int,
+    current_software_version: int,
+    requested_software_version: int | None = None,
 ) -> None | dict:
-    """Check if there is a newer software version available on the DCL."""
+    """Check if there is a software update available on the DCL."""
     try:
-        versions = await get_software_versions(vid, pid)
+        if requested_software_version is not None:
+            return await _check_update_version(
+                vid, pid, requested_software_version, current_software_version
+            )
+
+        # Get all versions and check each one of them.
+        versions = await _get_software_versions(vid, pid)
 
         all_software_versions: list[int] = versions["modelVersions"]["softwareVersions"]
         newer_software_versions = [
@@ -51,26 +84,11 @@ async def check_for_update(
 
         # Check if latest firmware is applicable, and backtrack from there
         for version in sorted(newer_software_versions, reverse=True):
-            version_res: dict = await get_software_version(vid, pid, version)
-            if not isinstance(version_res, dict):
-                raise TypeError("Unexpected DCL response.")
-
-            if "modelVersion" not in version_res:
-                raise ValueError("Unexpected DCL response.")
-
-            version_candidate: dict = cast(dict, version_res["modelVersion"])
-
-            # Check minApplicableSoftwareVersion/maxApplicableSoftwareVersion
-            min_sw_version = version_candidate["minApplicableSoftwareVersion"]
-            max_sw_version = version_candidate["maxApplicableSoftwareVersion"]
-            if (
-                current_software_version < min_sw_version
-                or current_software_version > max_sw_version
+            if version_candidate := await _check_update_version(
+                vid, pid, version, current_software_version
             ):
-                LOGGER.debug("Software version %d not applicable.", version)
-                continue
-
-            return version_candidate
+                return version_candidate
+            LOGGER.debug("Software version %d not applicable.", version)
         return None
 
     except (ClientError, TimeoutError) as err:

tests/server/ota/test_dcl.py

@@ -2,6 +2,8 @@
 
 from unittest.mock import AsyncMock, patch
 
+import pytest
+
 from matter_server.server.ota.dcl import check_for_update
 
 # Mock the DCL responses (sample from https://on.dcl.csa-iot.org/dcl/model/versions/4447/8194)
@@ -35,41 +37,49 @@
 }
 
 
-async def test_check_updates():
+@pytest.fixture(name="get_software_versions")
+def mock_get_software_versions():
+    """Mock the _get_software_versions function."""
+    with patch(
+        "matter_server.server.ota.dcl._get_software_versions",
+        new_callable=AsyncMock,
+        return_value=DCL_RESPONSE_SOFTWARE_VERSIONS,
+    ) as mock:
+        yield mock
+
+
+@pytest.fixture(name="get_software_version")
+def mock_get_software_version():
+    """Mock the _get_software_version function."""
+    with patch(
+        "matter_server.server.ota.dcl._get_software_version",
+        new_callable=AsyncMock,
+        return_value=DCL_RESPONSE_SOFTWARE_VERSION_1011,
+    ) as mock:
+        yield mock
+
+
+async def test_check_updates(get_software_versions, get_software_version):
     """Test the case where the latest software version is applicable."""
-    with (
-        patch(
-            "matter_server.server.ota.dcl.get_software_versions",
-            new_callable=AsyncMock,
-            return_value=DCL_RESPONSE_SOFTWARE_VERSIONS,
-        ),
-        patch(
-            "matter_server.server.ota.dcl.get_software_version",
-            new_callable=AsyncMock,
-            return_value=DCL_RESPONSE_SOFTWARE_VERSION_1011,
-        ),
-    ):
-        # Call the function with a current software version of 1000
-        result = await check_for_update(4447, 8194, 1000)
-
-        assert result == DCL_RESPONSE_SOFTWARE_VERSION_1011["modelVersion"]
-
-
-async def test_check_updates_not_applicable():
+    # Call the function with a current software version of 1000
+    result = await check_for_update(4447, 8194, 1000)
+
+    assert result == DCL_RESPONSE_SOFTWARE_VERSION_1011["modelVersion"]
+
+
+async def test_check_updates_not_applicable(
+    get_software_versions, get_software_version
+):
     """Test the case where the latest software version is not applicable."""
-    with (
-        patch(
-            "matter_server.server.ota.dcl.get_software_versions",
-            new_callable=AsyncMock,
-            return_value=DCL_RESPONSE_SOFTWARE_VERSIONS,
-        ),
-        patch(
-            "matter_server.server.ota.dcl.get_software_version",
-            new_callable=AsyncMock,
-            return_value=DCL_RESPONSE_SOFTWARE_VERSION_1011,
-        ),
-    ):
-        # Call the function with a current software version of 1
-        result = await check_for_update(4447, 8194, 1)
-
-        assert result is None
+    # Call the function with a current software version of 1
+    result = await check_for_update(4447, 8194, 1)
+
+    assert result is None
+
+
+async def test_check_updates_specific_version(get_software_version):
+    """Test the case to get a specific version."""
+    # Call the function with a current software version of 1000 and request 1011 as update
+    result = await check_for_update(4447, 8194, 1000, 1011)
+
+    assert result == DCL_RESPONSE_SOFTWARE_VERSION_1011["modelVersion"]