Increase diff test coverage

Author: puddly

tests/api/test_network_state.py

@@ -2,6 +2,7 @@
 
 import pytest
 
+import zigpy_znp.types as t
 from zigpy_znp.types.nvids import ExNvIds, OsalNvIds
 
 from ..conftest import (
@@ -64,3 +65,30 @@ async def test_broken_cc2531_load_state(device, make_connected_znp, caplog):
     assert "inconsistent" in caplog.text
 
     znp.close()
+
+
+@pytest.mark.parametrize("device", [FormedZStack3CC2531])
+async def test_state_write_tclk_zstack3(device, make_connected_znp, caplog):
+    formed_znp, _ = await make_connected_znp(server_cls=device)
+
+    await formed_znp.load_network_info()
+    formed_znp.close()
+
+    empty_znp, _ = await make_connected_znp(server_cls=device)
+
+    caplog.set_level(logging.WARNING)
+    await empty_znp.write_network_info(
+        network_info=formed_znp.network_info.replace(
+            tc_link_key=formed_znp.network_info.tc_link_key.replace(
+                # Non-standard TCLK
+                key=t.KeyData.convert("AA:BB:CC:DD:AA:BB:CC:DD:AA:BB:CC:DD:AA:BB:CC:DD")
+            )
+        ),
+        node_info=formed_znp.node_info,
+    )
+    assert "TC link key is configured at build time in Z-Stack 3" in caplog.text
+
+    await empty_znp.load_network_info()
+
+    # TCLK was not changed
+    assert formed_znp.network_info == empty_znp.network_info

tests/application/test_connect.py

@@ -58,7 +58,7 @@ async def test_probe_unsuccessful():
 
 
 @pytest.mark.parametrize("device", FORMED_DEVICES)
-async def test_probe_unsuccessful_slow(device, make_znp_server, mocker):
+async def test_probe_unsuccessful_slow1(device, make_znp_server, mocker):
     znp_server = make_znp_server(server_cls=device, shorten_delays=False)
 
     # Don't respond to anything
@@ -75,6 +75,24 @@ async def test_probe_unsuccessful_slow(device, make_znp_server, mocker):
     assert not any([t._is_connected for t in znp_server._transports])
 
 
+@pytest.mark.parametrize("device", FORMED_DEVICES)
+async def test_probe_unsuccessful_slow2(device, make_znp_server, mocker):
+    znp_server = make_znp_server(server_cls=device, shorten_delays=False)
+
+    # Don't respond to anything
+    znp_server._listeners.clear()
+
+    mocker.patch("zigpy_znp.zigbee.application.PROBE_TIMEOUT", new=0.1)
+
+    assert not (
+        await ControllerApplication.probe(
+            conf.SCHEMA_DEVICE({conf.CONF_DEVICE_PATH: znp_server.serial_port})
+        )
+    )
+
+    assert not any([t._is_connected for t in znp_server._transports])
+
+
 @pytest.mark.parametrize("device", FORMED_DEVICES)
 async def test_probe_successful(device, make_znp_server):
     znp_server = make_znp_server(server_cls=device, shorten_delays=False)
@@ -101,7 +119,7 @@ async def test_probe_multiple(device, make_znp_server):
 
 
 @pytest.mark.parametrize("device", FORMED_DEVICES)
-async def test_reconnect(device, event_loop, make_application):
+async def test_reconnect(device, make_application):
     app, znp_server = await make_application(
         server_cls=device,
         client_config={
@@ -180,7 +198,7 @@ async def test_multiple_shutdown(make_application):
 
 
 @pytest.mark.parametrize("device", FORMED_DEVICES)
-async def test_reconnect_lockup(device, event_loop, make_application, mocker):
+async def test_reconnect_lockup(device, make_application, mocker):
     mocker.patch("zigpy_znp.zigbee.application.WATCHDOG_PERIOD", 0.1)
 
     app, znp_server = await make_application(
@@ -221,7 +239,7 @@ async def test_reconnect_lockup(device, event_loop, make_application, mocker):
 
 
 @pytest.mark.parametrize("device", [FormedLaunchpadCC26X2R1])
-async def test_reconnect_lockup_pyserial(device, event_loop, make_application, mocker):
+async def test_reconnect_lockup_pyserial(device, make_application, mocker):
     mocker.patch("zigpy_znp.zigbee.application.WATCHDOG_PERIOD", 0.1)
 
     app, znp_server = await make_application(
@@ -271,3 +289,49 @@ async def patched_start_network(old_start_network=app.start_network, **kwargs):
     assert app._znp and app._znp._uart
 
     await app.shutdown()
+
+
+@pytest.mark.parametrize("device", [FormedLaunchpadCC26X2R1])
+async def test_disconnect(device, make_application):
+    app, znp_server = await make_application(
+        server_cls=device,
+        client_config={
+            conf.CONF_ZNP_CONFIG: {
+                conf.CONF_SREQ_TIMEOUT: 0.1,
+            }
+        },
+    )
+
+    assert app._znp is None
+    await app.connect()
+
+    assert app._znp is not None
+
+    await app.disconnect()
+    assert app._znp is None
+
+    await app.disconnect()
+    await app.disconnect()
+
+
+@pytest.mark.parametrize("device", [FormedLaunchpadCC26X2R1])
+async def test_disconnect_failure(device, make_application):
+    app, znp_server = await make_application(
+        server_cls=device,
+        client_config={
+            conf.CONF_ZNP_CONFIG: {
+                conf.CONF_SREQ_TIMEOUT: 0.1,
+            }
+        },
+    )
+
+    assert app._znp is None
+    await app.connect()
+
+    assert app._znp is not None
+
+    with patch.object(app._znp, "reset", side_effect=RuntimeError("An error")):
+        # Runs without error
+        await app.disconnect()
+
+    assert app._znp is None

zigpy_znp/types/commands.py

@@ -471,15 +471,7 @@ def as_dict(self) -> dict[str, typing.Any]:
         Converts the command into a dictionary.
         """
 
-        result = {}
-
-        for param, value in self._bound_params.values():
-            if issubclass(param.type, t.CStruct):
-                result[param.name] = value.as_dict()
-            else:
-                result[param.name] = value
-
-        return result
+        return {p.name: v for p, v in self._bound_params.values()}
 
     def __eq__(self, other):
         return type(self) is type(other) and self._bound_params == other._bound_params

zigpy_znp/zigbee/application.py

@@ -123,8 +123,6 @@ async def disconnect(self):
         if self._znp is not None:
             try:
                 await self._znp.reset(wait_for_reset=False)
-            except asyncio.CancelledError:
-                raise
             except Exception as e:
                 LOGGER.warning("Failed to reset before disconnect: %s", e)
             finally: