100% coverage

bellows/ezsp/__init__.py

@@ -259,18 +259,12 @@ async def leaveNetwork(self, timeout: float | int = NETWORK_OPS_TIMEOUT) -> None
 
     def connection_lost(self, exc):
         """Lost serial connection."""
-        LOGGER.debug(
-            "%s connection lost unexpectedly: %s",
-            self._config[conf.CONF_DEVICE_PATH],
-            exc,
-        )
         if self._application is not None:
             self._application.connection_lost(exc)
 
     def enter_failed_state(self, code: t.NcpResetCode) -> None:
         """UART received reset code."""
-        if self._application is not None:
-            self._application.connection_lost(NcpFailure(code=code))
+        self.connection_lost(NcpFailure(code=code))
 
     def __getattr__(self, name: str) -> Callable:
         if name not in self._protocol.COMMANDS:

tests/test_ash.py

@@ -600,3 +600,12 @@ async def test_ash_end_to_end(transport_cls: type[FakeTransport]) -> None:
     with patch.object(ncp, "nak_state", True):
         with pytest.raises(ash.NotAcked):
             await host.send_data(b"ncp NAKing until failure")
+
+
+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)
+
+    assert exc1 == exc1
+    assert exc1 != exc2
+    assert exc2 != t.NcpResetCode.RESET_POWER_ON

tests/test_uart.py

@@ -237,3 +237,10 @@ async def test_callbacks(gw):
     assert gw._api.enter_failed_state.mock_calls == [
         call(t.NcpResetCode.RESET_SOFTWARE)
     ]
+
+
+def test_reset_propagation(gw):
+    gw.reset_received(t.NcpResetCode.ERROR_EXCEEDED_MAXIMUM_ACK_TIMEOUT_COUNT)
+    assert gw._api.enter_failed_state.mock_calls == [
+        call(t.NcpResetCode.ERROR_EXCEEDED_MAXIMUM_ACK_TIMEOUT_COUNT)
+    ]