Add some tests

Author: puddly

tests/test_ash.py

@@ -605,6 +605,43 @@ async def test_ash_end_to_end(transport_cls: type[FakeTransport]) -> None:
             await host.send_data(b"ncp NAKing until failure")
 
 
+async def test_rstack_cancels_pending_frames() -> None:
+    """Test that RSTACK frame cancels pending data frames."""
+    host_ezsp = MagicMock()
+    ncp_ezsp = MagicMock()
+
+    host = ash.AshProtocol(host_ezsp)
+    ncp = AshNcpProtocol(ncp_ezsp)
+
+    host_transport = FakeTransport(ncp)
+    ncp_transport = FakeTransport(host)
+
+    host.connection_made(host_transport)
+    ncp.connection_made(ncp_transport)
+
+    # Pause the NCP transport so ACKs can't be sent back, creating a pending frame
+    ncp_transport.paused = True
+
+    # Start sending data without awaiting - this will create a pending frame
+    send_task = asyncio.create_task(host.send_data(b"test data"))
+
+    # Give task time to start and create the pending frame
+    await asyncio.sleep(0.1)
+
+    # Verify we have a pending frame
+    assert len(host._pending_data_frames) == 1
+
+    # Trigger RSTACK frame to cancel the pending frame
+    rstack = ash.RStackFrame(version=2, reset_code=t.NcpResetCode.RESET_POWER_ON)
+    host.rstack_frame_received(rstack)
+
+    # Verify task was cancelled with NcpFailure containing the reset code
+    with pytest.raises(ash.NcpFailure) as exc_info:
+        await send_task
+
+    assert exc_info.value.code == t.NcpResetCode.RESET_POWER_ON
+
+
 def test_ncp_failure_comparison() -> None:
     exc1 = ash.NcpFailure(code=t.NcpResetCode.ERROR_EXCEEDED_MAXIMUM_ACK_TIMEOUT_COUNT)
     exc2 = ash.NcpFailure(code=t.NcpResetCode.RESET_POWER_ON)

tests/test_ezsp.py

@@ -304,6 +304,28 @@ async def test_ezsp_connect_failure(disconnect_mock, reset_mock, version_mock):
     assert disconnect_mock.call_count == 1
 
 
+@pytest.mark.parametrize("failures_before_success", [1, 2, 3, 4])
+@patch.object(EZSP, "disconnect", new_callable=AsyncMock)
+async def test_ezsp_connect_retry_success(disconnect_mock, failures_before_success):
+    """Test connection succeeding after N failures."""
+    call_count = 0
+
+    async def startup_reset_mock():
+        nonlocal call_count
+        call_count += 1
+        if call_count <= failures_before_success:
+            raise RuntimeError(f"Startup failed (attempt {call_count})")
+
+    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 == failures_before_success + 1
+    assert disconnect_mock.call_count == 0
+
+
 async def test_ezsp_newer_version(ezsp_f):
     """Test newer version of ezsp."""
     with patch.object(

tests/test_uart.py

@@ -211,9 +211,18 @@ def on_transport_close():
     assert len(threads) == 0
 
 
-async def test_wait_for_startup_reset(gw):
+@pytest.mark.parametrize(
+    "reset_code",
+    [
+        t.NcpResetCode.RESET_SOFTWARE,
+        t.NcpResetCode.RESET_POWER_ON,
+        t.NcpResetCode.RESET_WATCHDOG,
+        t.NcpResetCode.RESET_EXTERNAL,
+    ],
+)
+async def test_wait_for_startup_reset(gw, reset_code):
     loop = asyncio.get_running_loop()
-    loop.call_later(0.01, gw.reset_received, t.NcpResetCode.RESET_SOFTWARE)
+    loop.call_later(0.01, gw.reset_received, reset_code)
 
     assert gw._startup_reset_future is None
     await gw.wait_for_startup_reset()
@@ -239,8 +248,25 @@ async def test_callbacks(gw):
     ]
 
 
-def test_reset_propagation(gw):
-    gw.reset_received(t.NcpResetCode.ERROR_EXCEEDED_MAXIMUM_ACK_TIMEOUT_COUNT)
+async def test_error_received_during_reset_ignored(gw):
+    # Set up a reset future to simulate being in the middle of a reset
+    loop = asyncio.get_running_loop()
+    gw._reset_future = loop.create_future()
+
+    # Error should be ignored (not trigger failed state)
+    gw.error_received(t.NcpResetCode.ERROR_EXCEEDED_MAXIMUM_ACK_TIMEOUT_COUNT)
+    assert gw._api.enter_failed_state.call_count == 0
+
+    # Clean up
+    gw._reset_future.cancel()
+
+
+def test_unexpected_reset_triggers_failed_state(gw):
+    # When no reset is expected, any reset should trigger failed state
+    assert gw._reset_future is None
+    assert gw._startup_reset_future is None
+
+    gw.reset_received(t.NcpResetCode.RESET_SOFTWARE)
     assert gw._api.enter_failed_state.mock_calls == [
-        call(t.NcpResetCode.ERROR_EXCEEDED_MAXIMUM_ACK_TIMEOUT_COUNT)
+        call(t.NcpResetCode.RESET_SOFTWARE)
     ]