Use zigpy types where possible (#162)

* use zigpy types

* tests for EUI64

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"