Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 4 additions & 1 deletion bellows/ezsp/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -144,7 +144,10 @@ async def _startup_reset(self) -> None:

async def startup_reset(self) -> None:
for attempt in range(RESET_ATTEMPTS):
self._protocol = v4.EZSPv4(self.handle_callback, self._gw)
# Start every attempt from a consistent v4 state. A failed attempt may
# already have switched to the stick's protocol version; keeping that
# version with a v4 protocol object makes `version()` skip the switch.
self._switch_protocol_version(v4.EZSPv4.VERSION)

try:
await self._startup_reset()
Expand Down
27 changes: 27 additions & 0 deletions tests/test_ezsp.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
InvalidTxPower,
)
from bellows.ezsp import EZSP, EZSP_LATEST, xncp
from bellows.ezsp.v4 import EZSPv4
from bellows.ezsp.v9.commands import GetTokenDataRsp
import bellows.types as t

Expand Down Expand Up @@ -341,6 +342,32 @@ async def startup_reset_mock():
assert disconnect_mock.call_count == 0


@patch.object(EZSP, "disconnect", new_callable=AsyncMock)
async def test_ezsp_connect_retry_resets_protocol_version(disconnect_mock):
"""A retry after a failure that already switched protocols starts from v4."""
call_count = 0
seen = []

async def startup_reset_mock():
nonlocal call_count
call_count += 1
seen.append((ezsp.ezsp_version, type(ezsp._protocol)))
if call_count == 1:
# The first attempt negotiates v12, then fails (e.g. a slow `zigbeed` reset)
ezsp._switch_protocol_version(12)
raise RuntimeError("Startup failed after the version switch")

with patch("bellows.uart.connect"):
ezsp = make_ezsp(version=4)

with patch.object(ezsp, "_startup_reset", side_effect=startup_reset_mock):
await ezsp.connect()

assert call_count == 2
assert seen[1] == (4, EZSPv4)
assert disconnect_mock.call_count == 0


async def test_ezsp_newer_version(ezsp_f):
"""Test newer version of ezsp."""
with patch.object(
Expand Down
Loading