Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Use zigpy types where possible #162

Merged
merged 2 commits into from
Oct 9, 2023
Merged

Use zigpy types where possible #162

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
988d7ce
use zigpy types
Shulyaka Oct 4, 2023
89a343e
tests for EUI64
Shulyaka Oct 4, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 15 additions & 12 deletions tests/test_api.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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

Expand All @@ -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
Expand All @@ -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
Expand All @@ -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)
Expand Down
10 changes: 5 additions & 5 deletions tests/test_application.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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(
Expand Down
50 changes: 23 additions & 27 deletions tests/test_types.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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
Expand All @@ -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
Expand All @@ -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"
Loading
Loading