Avoid TC address cache overwrite when setting extended timeout

bellows/ezsp/protocol.py

@@ -39,7 +39,7 @@
         self._gw = gateway
         self._seq = 0
         self.COMMANDS_BY_ID = {
             cmd_id: (name, tx_schema, rx_schema)
             for name, (cmd_id, tx_schema, rx_schema) in self.COMMANDS.items()
         }
         self.tc_policy = 0
@@ -49,6 +49,8 @@
 
         # Cached by `set_extended_timeout` so subsequent calls are a little faster
         self._address_table_size: int | None = None
+        # Next address table slot `set_extended_timeout` will write
+        self._address_table_cursor: int = 0
         self._fragment_manager = FragmentManager()
         self._fragment_ack_tasks: set[asyncio.Task] = set()
 

bellows/ezsp/v4/__init__.py

@@ -3,7 +3,6 @@
 
 from collections.abc import AsyncGenerator, Iterable
 import logging
-import random
 
 import voluptuous as vol
 import zigpy.state
@@ -251,8 +250,12 @@ async def set_extended_timeout(
             )
             return
 
-        # Replace a random entry in the address table
-        index = random.randint(0, self._address_table_size - 1)
+        # The stack only updates existing entries and never allocates application slots
+        # and bellows allocates them only here. Filling the table in order thus uses
+        # every free slot before replacing anything; once it wraps, replacement is
+        # round-robin.
+        index = self._address_table_cursor
+        self._address_table_cursor = (index + 1) % self._address_table_size
 
         await self.replaceAddressTableEntry(
             addressTableIndex=index,

tests/test_ezsp_v14.py

@@ -1,4 +1,4 @@
-from unittest.mock import MagicMock, call, patch
+from unittest.mock import MagicMock, call
 
 import pytest
 import zigpy.exceptions
@@ -695,22 +695,19 @@ async def test_set_extended_timeout_no_entry(ezsp_f, lookup_rsp: tuple) -> None:
         t.Bool.false,
     )
 
-    with patch("bellows.ezsp.v4.random.randint") as mock_random:
-        mock_random.return_value = 3
-        await ezsp_f.set_extended_timeout(
-            nwk=0x1234,
-            ieee=t.EUI64.convert("aa:bb:cc:dd:ee:ff:00:11"),
-            extended_timeout=True,
-        )
+    await ezsp_f.set_extended_timeout(
+        nwk=0x1234,
+        ieee=t.EUI64.convert("aa:bb:cc:dd:ee:ff:00:11"),
+        extended_timeout=True,
+    )
 
     assert ezsp_f.setExtendedTimeout.mock_calls == []
     assert ezsp_f.getConfigurationValue.mock_calls == [
         call(t.EzspConfigId.CONFIG_ADDRESS_TABLE_SIZE)
     ]
-    assert mock_random.mock_calls == [call(0, 8 - 1)]
     assert ezsp_f.replaceAddressTableEntry.mock_calls == [
         call(
-            addressTableIndex=3,
+            addressTableIndex=0,
             newEui64=t.EUI64.convert("aa:bb:cc:dd:ee:ff:00:11"),
             newId=0x1234,
             newExtendedTimeout=True,

tests/test_ezsp_v4.py

@@ -1,6 +1,6 @@
 import asyncio
 import logging
-from unittest.mock import AsyncMock, MagicMock, call, patch
+from unittest.mock import AsyncMock, MagicMock, call
 
 import pytest
 import zigpy.state
@@ -416,13 +416,11 @@ async def test_set_extended_timeout_no_entry(ezsp_f) -> None:
         t.Bool.false,
     )
 
-    with patch("bellows.ezsp.v4.random.randint") as mock_random:
-        mock_random.return_value = 0
-        await ezsp_f.set_extended_timeout(
-            nwk=0x1234,
-            ieee=t.EUI64.convert("aa:bb:cc:dd:ee:ff:00:11"),
-            extended_timeout=True,
-        )
+    await ezsp_f.set_extended_timeout(
+        nwk=0x1234,
+        ieee=t.EUI64.convert("aa:bb:cc:dd:ee:ff:00:11"),
+        extended_timeout=True,
+    )
 
     assert ezsp_f.getExtendedTimeout.mock_calls == [
         call(remoteEui64=t.EUI64.convert("aa:bb:cc:dd:ee:ff:00:11"))
@@ -433,7 +431,6 @@ async def test_set_extended_timeout_no_entry(ezsp_f) -> None:
     assert ezsp_f.getConfigurationValue.mock_calls == [
         call(t.EzspConfigId.CONFIG_ADDRESS_TABLE_SIZE)
     ]
-    assert mock_random.mock_calls == [call(0, 8 - 1)]
     assert ezsp_f.replaceAddressTableEntry.mock_calls == [
         call(
             addressTableIndex=0,
@@ -443,14 +440,12 @@ async def test_set_extended_timeout_no_entry(ezsp_f) -> None:
         )
     ]
 
-    # The address table size is cached
-    with patch("bellows.ezsp.v4.random.randint") as mock_random:
-        mock_random.return_value = 1
-        await ezsp_f.set_extended_timeout(
-            nwk=0x1234,
-            ieee=t.EUI64.convert("aa:bb:cc:dd:ee:ff:00:11"),
-            extended_timeout=True,
-        )
+    # The address table size is cached and the next free slot is used
+    await ezsp_f.set_extended_timeout(
+        nwk=0x1234,
+        ieee=t.EUI64.convert("aa:bb:cc:dd:ee:ff:00:11"),
+        extended_timeout=True,
+    )
 
     # Still called only once
     assert ezsp_f.getConfigurationValue.mock_calls == [
@@ -515,20 +510,43 @@ async def test_set_extended_timeout_already_have_entry(ezsp_f) -> None:
     ]
 
 
-async def test_set_extended_timeout_bad_table_size(ezsp_f) -> None:
-    ezsp_f.setExtendedTimeout.return_value = ()
+async def test_set_extended_timeout_table_wraps(ezsp_f) -> None:
+    # Once every slot has been used, replacement is round-robin
     ezsp_f.getExtendedTimeout.return_value = (t.Bool.false,)
     ezsp_f.lookupNodeIdByEui64.return_value = (0xFFFF,)
-    ezsp_f.getConfigurationValue.return_value = (t.EmberStatus.ERR_FATAL, 0xFF)
+    ezsp_f.getConfigurationValue.return_value = (t.EmberStatus.SUCCESS, 2)
+    ezsp_f.replaceAddressTableEntry.return_value = (
+        t.EmberStatus.SUCCESS,
+        t.EUI64.convert("ff:ff:ff:ff:ff:ff:ff:ff"),
+        0xFFFF,
+        t.Bool.false,
+    )
 
-    with patch("bellows.ezsp.v4.random.randint") as mock_random:
-        mock_random.return_value = 0
+    for _ in range(3):
         await ezsp_f.set_extended_timeout(
             nwk=0x1234,
             ieee=t.EUI64.convert("aa:bb:cc:dd:ee:ff:00:11"),
             extended_timeout=True,
         )
 
+    assert [
+        c.kwargs["addressTableIndex"]
+        for c in ezsp_f.replaceAddressTableEntry.mock_calls
+    ] == [0, 1, 0]
+
+
+async def test_set_extended_timeout_bad_table_size(ezsp_f) -> None:
+    ezsp_f.setExtendedTimeout.return_value = ()
+    ezsp_f.getExtendedTimeout.return_value = (t.Bool.false,)
+    ezsp_f.lookupNodeIdByEui64.return_value = (0xFFFF,)
+    ezsp_f.getConfigurationValue.return_value = (t.EmberStatus.ERR_FATAL, 0xFF)
+
+    await ezsp_f.set_extended_timeout(
+        nwk=0x1234,
+        ieee=t.EUI64.convert("aa:bb:cc:dd:ee:ff:00:11"),
+        extended_timeout=True,
+    )
+
     assert ezsp_f.getExtendedTimeout.mock_calls == [
         call(remoteEui64=t.EUI64.convert("aa:bb:cc:dd:ee:ff:00:11"))
     ]