Merge pull request #97 from puddly/puddly/new-radio-settings-api

Implement new zigpy radio API

Author: puddly

.pre-commit-config.yaml

@@ -15,7 +15,7 @@ repos:
       - id: flake8
         entry: pflake8
         additional_dependencies: 
-          - pyproject-flake8==0.0.1a2
+          - pyproject-flake8==0.0.1a3
           - flake8-bugbear==22.1.11
           - flake8-comprehensions==3.8.0
           - flake8_2020==1.6.1
@@ -29,11 +29,11 @@ repos:
       - id: isort
 
   - repo: https://github.com/pre-commit/mirrors-mypy
-    rev: v0.931
+    rev: v0.942
     hooks:
       - id: mypy
         additional_dependencies:
-          - zigpy==0.43.0
+          - zigpy
 
   - repo: https://github.com/asottile/pyupgrade
     rev: v2.31.0
@@ -43,4 +43,4 @@ repos:
   - repo: https://github.com/fsouza/autoflake8
     rev: v0.3.1
     hooks:
-      - id: autoflake8
+      - id: autoflake8

setup.cfg

@@ -15,7 +15,7 @@ python_requires = >=3.7
 install_requires =
     pyserial-asyncio; platform_system!="Windows"
     pyserial-asyncio!=0.5; platform_system=="Windows"  # 0.5 broke writes
-    zigpy>=0.40.0
+    zigpy>=0.47.0
     async_timeout
     voluptuous
     coloredlogs

tests/api/test_network_state.py

@@ -1,8 +1,8 @@
 import logging
-import dataclasses
 
 import pytest
 
+import zigpy_znp.types as t
 from zigpy_znp.types.nvids import ExNvIds, OsalNvIds
 
 from ..conftest import (
@@ -31,16 +31,22 @@ async def test_state_transfer(from_device, to_device, make_connected_znp):
 
     # Z-Stack 1 devices can't have some security info read out
     if issubclass(from_device, BaseZStack1CC2531):
-        assert formed_znp.network_info == dataclasses.replace(
-            empty_znp.network_info, stack_specific={}
+        assert formed_znp.network_info == empty_znp.network_info.replace(
+            stack_specific={},
+            metadata=formed_znp.network_info.metadata,
         )
     elif issubclass(to_device, BaseZStack1CC2531):
         assert (
-            dataclasses.replace(formed_znp.network_info, stack_specific={})
+            formed_znp.network_info.replace(
+                stack_specific={},
+                metadata=empty_znp.network_info.metadata,
+            )
             == empty_znp.network_info
         )
     else:
-        assert formed_znp.network_info == empty_znp.network_info
+        assert formed_znp.network_info == empty_znp.network_info.replace(
+            metadata=formed_znp.network_info.metadata
+        )
 
     assert formed_znp.node_info == empty_znp.node_info
 
@@ -59,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_nvram_migration.py renamed to tests/api/test_nvram_migration.py

@@ -44,7 +44,7 @@ async def test_addrmgr_empty_entries(make_connected_znp, device):
 
 
 @pytest.mark.parametrize("device", [FormedZStack3CC2531])
-async def test_addrmgr_rewrite_fix(device, make_application, mocker):
+async def test_addrmgr_rewrite_fix(device, make_connected_znp):
     # Keep track of reads
     addrmgr_reads = []
 
@@ -60,7 +60,7 @@ async def test_addrmgr_rewrite_fix(device, make_application, mocker):
         extAddr=t.EUI64.convert("FF:FF:FF:FF:FF:FF:FF:FF"),
     )
 
-    app, znp_server = make_application(server_cls=device)
+    znp, znp_server = await make_connected_znp(server_cls=device)
     znp_server.callback_for_response(
         c.SYS.OSALNVReadExt.Req(Id=OsalNvIds.ADDRMGR, Offset=0), addrmgr_reads.append
     )
@@ -81,8 +81,7 @@ async def test_addrmgr_rewrite_fix(device, make_application, mocker):
     assert old_addrmgr != nvram[OsalNvIds.ADDRMGR]
 
     assert len(addrmgr_reads) == 0
-    await app.startup()
-    await app.shutdown()
+    await znp.migrate_nvram()
     assert len(addrmgr_reads) == 2
 
     # Bad entries have been fixed
@@ -94,8 +93,7 @@ async def test_addrmgr_rewrite_fix(device, make_application, mocker):
 
     # Will not be read again
     assert len(addrmgr_reads) == 2
-    await app.startup()
-    await app.shutdown()
+    await znp.migrate_nvram()
     assert len(addrmgr_reads) == 2
 
     # Will be migrated again if the migration NVID is deleted
@@ -104,8 +102,7 @@ async def test_addrmgr_rewrite_fix(device, make_application, mocker):
     old_addrmgr2 = nvram[OsalNvIds.ADDRMGR]
 
     assert len(addrmgr_reads) == 2
-    await app.startup()
-    await app.shutdown()
+    await znp.migrate_nvram()
     assert len(addrmgr_reads) == 3
 
     # But nothing will change

tests/application/test_connect.py

@@ -1,12 +1,13 @@
 import asyncio
+from unittest.mock import patch
 
 import pytest
 
 import zigpy_znp.config as conf
 from zigpy_znp.uart import connect as uart_connect
 from zigpy_znp.zigbee.application import ControllerApplication
 
-from ..conftest import FORMED_DEVICES, FormedLaunchpadCC26X2R1, swap_attribute
+from ..conftest import FORMED_DEVICES, FormedLaunchpadCC26X2R1
 
 
 async def test_no_double_connect(make_znp_server, mocker):
@@ -57,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
@@ -74,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)
@@ -100,8 +119,8 @@ async def test_probe_multiple(device, make_znp_server):
 
 
 @pytest.mark.parametrize("device", FORMED_DEVICES)
-async def test_reconnect(device, event_loop, make_application):
-    app, znp_server = make_application(
+async def test_reconnect(device, make_application):
+    app, znp_server = await make_application(
         server_cls=device,
         client_config={
             # Make auto-reconnection happen really fast
@@ -118,7 +137,7 @@ async def test_reconnect(device, event_loop, make_application):
     assert app._znp is not None
 
     # Don't reply to anything for a bit
-    with swap_attribute(znp_server, "frame_received", lambda _: None):
+    with patch.object(znp_server, "frame_received", lambda _: None):
         # Now that we're connected, have the server close the connection
         znp_server._uart._transport.close()
 
@@ -143,7 +162,7 @@ async def test_reconnect(device, event_loop, make_application):
 
 @pytest.mark.parametrize("device", FORMED_DEVICES)
 async def test_shutdown_from_app(device, mocker, make_application):
-    app, znp_server = make_application(server_cls=device)
+    app, znp_server = await make_application(server_cls=device)
 
     await app.startup(auto_form=False)
 
@@ -159,7 +178,7 @@ async def test_shutdown_from_app(device, mocker, make_application):
 
 
 async def test_clean_shutdown(make_application):
-    app, znp_server = make_application(server_cls=FormedLaunchpadCC26X2R1)
+    app, znp_server = await make_application(server_cls=FormedLaunchpadCC26X2R1)
     await app.startup(auto_form=False)
 
     # This should not throw
@@ -170,7 +189,7 @@ async def test_clean_shutdown(make_application):
 
 
 async def test_multiple_shutdown(make_application):
-    app, znp_server = make_application(server_cls=FormedLaunchpadCC26X2R1)
+    app, znp_server = await make_application(server_cls=FormedLaunchpadCC26X2R1)
     await app.startup(auto_form=False)
 
     await app.shutdown()
@@ -179,10 +198,10 @@ 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 = make_application(
+    app, znp_server = await make_application(
         server_cls=device,
         client_config={
             # Make auto-reconnection happen really fast
@@ -197,7 +216,7 @@ async def test_reconnect_lockup(device, event_loop, make_application, mocker):
     await app.startup(auto_form=False)
 
     # Stop responding
-    with swap_attribute(znp_server, "frame_received", lambda _: None):
+    with patch.object(znp_server, "frame_received", lambda _: None):
         assert app._znp is not None
         assert app._reconnect_task.done()
 
@@ -219,15 +238,16 @@ async def test_reconnect_lockup(device, event_loop, make_application, mocker):
     await app.shutdown()
 
 
-@pytest.mark.parametrize("device", FORMED_DEVICES)
-async def test_reconnect_lockup_pyserial(device, event_loop, make_application, mocker):
+@pytest.mark.parametrize("device", [FormedLaunchpadCC26X2R1])
+async def test_reconnect_lockup_pyserial(device, make_application, mocker):
     mocker.patch("zigpy_znp.zigbee.application.WATCHDOG_PERIOD", 0.1)
 
-    app, znp_server = make_application(
+    app, znp_server = await make_application(
         server_cls=device,
         client_config={
             conf.CONF_ZNP_CONFIG: {
-                conf.CONF_AUTO_RECONNECT_RETRY_DELAY: 0.1,
+                conf.CONF_AUTO_RECONNECT_RETRY_DELAY: 0.01,
+                conf.CONF_SREQ_TIMEOUT: 0.1,
             }
         },
     )
@@ -242,20 +262,20 @@ async def test_reconnect_lockup_pyserial(device, event_loop, make_application, m
     # We are connected
     assert app._znp is not None
 
-    did_load_info = asyncio.get_running_loop().create_future()
+    did_start_network = asyncio.get_running_loop().create_future()
 
-    async def patched_load_network_info(*, old_load=app.load_network_info):
+    async def patched_start_network(old_start_network=app.start_network, **kwargs):
         try:
-            return await old_load()
+            return await old_start_network(**kwargs)
         finally:
-            did_load_info.set_result(True)
+            did_start_network.set_result(True)
 
-    with swap_attribute(app, "load_network_info", patched_load_network_info):
+    with patch.object(app, "start_network", patched_start_network):
         # "Drop" the connection like PySerial
         app._znp._uart.connection_lost(exc=None)
 
         # Wait until we are reconnecting
-        await did_load_info
+        await did_start_network
 
     # "Drop" the connection like PySerial again, but during connect
     app._znp._uart.connection_lost(exc=None)
@@ -269,3 +289,49 @@ async def patched_load_network_info(*, old_load=app.load_network_info):
     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