Use zigpy types where possible (#162)

* use zigpy types

* tests for EUI64

Author: Shulyaka

tests/test_api.py

@@ -4,8 +4,9 @@
 import pytest
 import serial
 import zigpy.exceptions
+import zigpy.types as t
 
-from zigpy_xbee import api as xbee_api, types as t, uart
+from zigpy_xbee import api as xbee_api, types as xbee_t, uart
 import zigpy_xbee.config
 from zigpy_xbee.zigbee.application import ControllerApplication
 
@@ -335,19 +336,19 @@ def _send_modem_event(api, event):
 def test_handle_modem_status(api):
     api._running.clear()
     api._reset.set()
-    _send_modem_event(api, xbee_api.ModemStatus.COORDINATOR_STARTED)
+    _send_modem_event(api, xbee_t.ModemStatus.COORDINATOR_STARTED)
     assert api.is_running is True
     assert api.reset_event.is_set() is True
 
     api._running.set()
     api._reset.set()
-    _send_modem_event(api, xbee_api.ModemStatus.DISASSOCIATED)
+    _send_modem_event(api, xbee_t.ModemStatus.DISASSOCIATED)
     assert api.is_running is False
     assert api.reset_event.is_set() is True
 
     api._running.set()
     api._reset.clear()
-    _send_modem_event(api, xbee_api.ModemStatus.HARDWARE_RESET)
+    _send_modem_event(api, xbee_t.ModemStatus.HARDWARE_RESET)
     assert api.is_running is False
     assert api.reset_event.is_set() is True
 
@@ -371,26 +372,28 @@ def test_handle_explicit_rx_indicator(api):
 
 
 def _handle_tx_status(api, status, wrong_frame_id=False):
-    status = t.TXStatus(status)
+    status = xbee_t.TXStatus(status)
     frame_id = 0x12
     send_fut = mock.MagicMock(spec=asyncio.Future)
     api._awaiting[frame_id] = (send_fut,)
     s = mock.sentinel
     if wrong_frame_id:
         frame_id += 1
-    api._handle_tx_status(frame_id, s.dst_nwk, s.retries, status, t.DiscoveryStatus())
+    api._handle_tx_status(
+        frame_id, s.dst_nwk, s.retries, status, xbee_t.DiscoveryStatus()
+    )
     return send_fut
 
 
 def test_handle_tx_status_success(api):
-    fut = _handle_tx_status(api, t.TXStatus.SUCCESS)
+    fut = _handle_tx_status(api, xbee_t.TXStatus.SUCCESS)
     assert len(api._awaiting) == 0
     assert fut.set_result.call_count == 1
     assert fut.set_exception.call_count == 0
 
 
 def test_handle_tx_status_except(api):
-    fut = _handle_tx_status(api, t.TXStatus.ADDRESS_NOT_FOUND)
+    fut = _handle_tx_status(api, xbee_t.TXStatus.ADDRESS_NOT_FOUND)
     assert len(api._awaiting) == 0
     assert fut.set_result.call_count == 0
     assert fut.set_exception.call_count == 1
@@ -404,7 +407,7 @@ def test_handle_tx_status_unexpected(api):
 
 
 def test_handle_tx_status_duplicate(api):
-    status = t.TXStatus.SUCCESS
+    status = xbee_t.TXStatus.SUCCESS
     frame_id = 0x12
     send_fut = mock.MagicMock(spec=asyncio.Future)
     send_fut.set_result.side_effect = asyncio.InvalidStateError
@@ -418,16 +421,16 @@ def test_handle_tx_status_duplicate(api):
 
 def test_handle_registration_status(api):
     frame_id = 0x12
-    status = xbee_api.RegistrationStatus.SUCCESS
+    status = xbee_t.RegistrationStatus.SUCCESS
     fut = asyncio.Future()
     api._awaiting[frame_id] = (fut,)
     api._handle_registration_status(frame_id, status)
     assert fut.done() is True
-    assert fut.result() == xbee_api.RegistrationStatus.SUCCESS
+    assert fut.result() == xbee_t.RegistrationStatus.SUCCESS
     assert fut.exception() is None
 
     frame_id = 0x13
-    status = xbee_api.RegistrationStatus.KEY_TABLE_IS_FULL
+    status = xbee_t.RegistrationStatus.KEY_TABLE_IS_FULL
     fut = asyncio.Future()
     api._awaiting[frame_id] = (fut,)
     api._handle_registration_status(frame_id, status)

tests/test_application.py

@@ -6,7 +6,7 @@
 import zigpy.types as t
 import zigpy.zdo.types as zdo_t
 
-from zigpy_xbee.api import ModemStatus, XBee
+from zigpy_xbee.api import XBee
 import zigpy_xbee.config as config
 import zigpy_xbee.types as xbee_t
 from zigpy_xbee.zigbee import application
@@ -74,10 +74,10 @@ def app(monkeypatch):
 
 
 def test_modem_status(app):
-    assert 0x00 in ModemStatus.__members__.values()
-    app.handle_modem_status(ModemStatus(0x00))
-    assert 0xEE not in ModemStatus.__members__.values()
-    app.handle_modem_status(ModemStatus(0xEE))
+    assert 0x00 in xbee_t.ModemStatus.__members__.values()
+    app.handle_modem_status(xbee_t.ModemStatus(0x00))
+    assert 0xEE not in xbee_t.ModemStatus.__members__.values()
+    app.handle_modem_status(xbee_t.ModemStatus(0xEE))
 
 
 def _test_rx(

tests/test_types.py

@@ -1,51 +1,32 @@
 import pytest
+import zigpy.types as t
 
-from zigpy_xbee import types as t
-
-
-def test_deserialize():
-    extra = b"\xBE\xEF"
-    data = b"\xff\xff\xfe01234567"
-    schema = (t.uint8_t, t.int16s, t.EUI64)
-    result, rest = t.deserialize(data + extra, schema)
-
-    assert rest == extra
-    assert result[0] == 0xFF
-    assert result[1] == -2
-    assert result[2] == t.EUI64((0x37, 0x36, 0x35, 0x34, 0x33, 0x32, 0x31, 0x30))
-
-
-def test_serialize():
-    data = [0xFF, -2, t.EUI64([t.uint8_t(i) for i in range(0x30, 0x38)])]
-    schema = (t.uint8_t, t.int16s, t.EUI64)
-    result = t.serialize(data, schema)
-
-    assert result == b"\xff\xff\xfe76543210"
+import zigpy_xbee.types as xbee_t
 
 
 def test_bytes_serialize():
     data = 0x89AB.to_bytes(4, "big")
-    result = t.Bytes(data).serialize()
+    result = xbee_t.Bytes(data).serialize()
     assert result == data
 
 
 def test_bytes_deserialize():
-    data, rest = t.Bytes.deserialize(0x89AB.to_bytes(3, "big"))
+    data, rest = xbee_t.Bytes.deserialize(0x89AB.to_bytes(3, "big"))
     assert data == b"\x00\x89\xAB"
     assert rest == b""
 
 
 def test_atcommand():
     cmd = b"AI"
     data = 0x06.to_bytes(4, "big")
-    r_cmd, r_data = t.ATCommand.deserialize(cmd + data)
+    r_cmd, r_data = xbee_t.ATCommand.deserialize(cmd + data)
 
     assert r_cmd == cmd
     assert r_data == data
 
 
 def test_undefined_enum_undefined_value():
-    class undEnum(t.uint8_t, t.UndefinedEnum):
+    class undEnum(t.uint8_t, xbee_t.UndefinedEnum):
         OK = 0
         ERROR = 2
         UNDEFINED_VALUE = 0xFF
@@ -67,7 +48,7 @@ class undEnum(t.uint8_t, t.UndefinedEnum):
 
 
 def test_undefined_enum_undefinede():
-    class undEnum(t.uint8_t, t.UndefinedEnum):
+    class undEnum(t.uint8_t, xbee_t.UndefinedEnum):
         OK = 0
         ERROR = 2
         UNDEFINED_VALUE = 0xFF
@@ -77,7 +58,22 @@ class undEnum(t.uint8_t, t.UndefinedEnum):
 
 
 def test_nwk():
-    nwk = t.NWK(0x1234)
+    nwk = xbee_t.NWK(0x1234)
 
     assert str(nwk) == "0x1234"
     assert repr(nwk) == "0x1234"
+
+
+def test_eui64():
+    extra = b"\xBE\xEF"
+    data = b"01234567"
+
+    result, rest = xbee_t.EUI64.deserialize(data + extra)
+
+    assert rest == extra
+    assert result == xbee_t.EUI64((0x37, 0x36, 0x35, 0x34, 0x33, 0x32, 0x31, 0x30))
+
+    data = xbee_t.EUI64([t.uint8_t(i) for i in range(0x30, 0x38)])
+    result = data.serialize()
+
+    assert result == b"76543210"