Implement the new zigpy radio API (#117)

* Begin implementing the new radio API

* Use the correct signature for `load_network_info`

* Use command and response IDs

* Parse ZiGate logging messages

* Implement `Status` type

* Erase PDM when writing new settings

* Handle responses and status callbacks in any order

* Implement `GET_DEVICES_LIST`

* Rename `ADDRESS_MODE` to `AddressMode`

* Initialize the ZiGate device on startup

* Only permit joins via the coordinator in `permit_ncp`

* Use `schedule_initialize` to prevent double initialization

* Add unhandled `NODE_DESCRIPTOR_RSP` response

* Add a stub for `add_endpoint`

* Set the TCLK's partner IEEE

* Set the network information `source` and `metadata`

* Fix unit tests

* Bump minimum required zigpy version to 0.47.0

Author: puddly

setup.py

@@ -53,7 +53,7 @@ def is_raspberry_pi(raise_on_errors=False):
     'pyserial-asyncio>=0.5; platform_system!="Windows"',
     'pyserial-asyncio!=0.5; platform_system=="Windows"',  # 0.5 broke writesv
     'pyusb>=1.1.0',
-    'zigpy>=0.22.2',
+    'zigpy>=0.47.0',
 ]
 
 if is_raspberry_pi():

tests/test_api.py

@@ -103,7 +103,7 @@ async def test_probe_fail(mock_connect, mock_raw_mode, exception):
 
 
 @pytest.mark.asyncio
-@patch.object(asyncio, "wait_for", side_effect=asyncio.TimeoutError)
+@patch.object(asyncio, "wait", return_value=([], []))
 async def test_api_command(mock_command, api):
     """Test command method."""
     try:

tests/test_application.py

@@ -3,6 +3,7 @@
 
 import pytest
 import zigpy.types as zigpy_types
+import zigpy.exceptions
 
 import zigpy_zigate.config as config
 import zigpy_zigate.types as t
@@ -31,7 +32,7 @@ def test_zigpy_ieee(app):
     data = b"\x01\x02\x03\x04\x05\x06\x07\x08"
 
     zigate_ieee, _ = t.EUI64.deserialize(data)
-    app._ieee = zigpy_types.EUI64(zigate_ieee)
+    app.state.node_info.ieee = zigpy_types.EUI64(zigate_ieee)
 
     dst_addr = app.get_dst_address(cluster)
     assert dst_addr.serialize() == b"\x03" + data[::-1] + b"\x01"
@@ -44,28 +45,54 @@ def test_model_detection(app):
 
 @pytest.mark.asyncio
 async def test_form_network_success(app):
+    app._api.erase_persistent_data = AsyncMock()
     app._api.set_channel = AsyncMock()
+    app._api.set_extended_panid = AsyncMock()
     app._api.reset = AsyncMock()
+
     async def mock_start_network():
         return [[0x00, 0x1234, 0x0123456789abcdef], 0]
     app._api.start_network = mock_start_network
+
+    async def mock_get_network_state():
+        return [
+            [
+                0x0000,
+                t.EUI64([0xef, 0xcd, 0xab, 0x89, 0x67, 0x45, 0x23, 0x01]),
+                0x1234,
+                0x1234abcdef012345,
+                0x11,
+            ],
+            0,
+        ]
+
+    app._api.get_network_state = mock_get_network_state
+
     await app.form_network()
-    assert app._nwk == 0x1234
-    assert app._ieee == 0x0123456789abcdef
+    await app.load_network_info()
+    assert app.state.node_info.nwk == 0x0000
+    assert app.state.node_info.ieee == zigpy.types.EUI64.convert(
+        "01:23:45:67:89:ab:cd:ef"
+    )
+    assert app.state.network_info.pan_id == 0x1234
+    assert app.state.network_info.extended_pan_id == zigpy.types.ExtendedPanId.convert(
+        "12:34:ab:cd:ef:01:23:45"
+    )
     assert app._api.reset.call_count == 0
 
 
 @pytest.mark.asyncio
 async def test_form_network_failed(app):
+    app._api.erase_persistent_data = AsyncMock()
     app._api.set_channel = AsyncMock()
+    app._api.set_extended_panid = AsyncMock()
     app._api.reset = AsyncMock()
     async def mock_start_network():
         return [[0x06], 0]
     app._api.start_network = mock_start_network
     async def mock_get_network_state():
         return [[0xffff, 0x0123456789abcdef, 0x1234, 0, 0x11], 0]
     app._api.get_network_state = mock_get_network_state
-    await app.form_network()
-    assert app._nwk == 0
-    assert app._ieee == 0
-    assert app._api.reset.call_count == 1
+
+    with pytest.raises(zigpy.exceptions.FormationFailure):
+        await app.form_network()

tests/test_types.py

@@ -23,8 +23,8 @@ def test_deserialize():
     assert result[2] == 0x0001
     assert result[3] == 0x01
     assert result[4] == 0x01
-    assert result[5] == t.Address(address_mode=t.ADDRESS_MODE.NWK, address=t.NWK(0x1234))
-    assert result[6] == t.Address(address_mode=t.ADDRESS_MODE.NWK, address=t.NWK(0xabcd))
+    assert result[5] == t.Address(address_mode=t.AddressMode.NWK, address=t.NWK(0x1234))
+    assert result[6] == t.Address(address_mode=t.AddressMode.NWK, address=t.NWK(0xabcd))
     assert result[7] == b'\x01\x00\xBE\xEF'
     assert rest == b''
 
@@ -34,7 +34,7 @@ def test_deserialize():
     assert result[0] == 0x00
     assert result[1] == 0x01
     assert result[2] == 0x01
-    assert result[3] == t.Address(address_mode=t.ADDRESS_MODE.NWK, address=t.NWK(0x1234))
+    assert result[3] == t.Address(address_mode=t.AddressMode.NWK, address=t.NWK(0x1234))
     assert result[4] == 0xff
 
     data = b'\x00\x01\x01\x03\x12\x34\x56\x78\x9a\xbc\xde\xf0\xff'
@@ -43,7 +43,7 @@ def test_deserialize():
     assert result[0] == 0x00
     assert result[1] == 0x01
     assert result[2] == 0x01
-    assert result[3] == t.Address(address_mode=t.ADDRESS_MODE.IEEE,
+    assert result[3] == t.Address(address_mode=t.AddressMode.IEEE,
                                   address=t.EUI64.deserialize(b'\x12\x34\x56\x78\x9a\xbc\xde\xf0')[0])
     assert result[4] == 0xff
 
@@ -73,7 +73,7 @@ def test_deserialize():
     assert result[0] == 0x00
     assert result[1] == 0x01
     assert result[2] == 0x01
-    assert result[3] == t.Address(address_mode=t.ADDRESS_MODE.NWK,
+    assert result[3] == t.Address(address_mode=t.AddressMode.NWK,
                                   address=t.NWK.deserialize(b'\xbc\x8c')[0])
     assert result[4] == 0x73
     assert len(result) == 5