1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
|
import asyncio
import logging
from unittest.mock import AsyncMock, MagicMock, call, patch
import pytest
from bellows.ezsp import EZSP
import bellows.ezsp.v4
import bellows.ezsp.v9
from bellows.ezsp.v9.commands import GetTokenDataRsp
import bellows.types as t
from bellows.types import NV3KeyId
from bellows.uart import Gateway
@pytest.fixture
def prot_hndl():
"""Protocol handler mock."""
app = MagicMock()
gateway = Gateway(app)
gateway._transport = AsyncMock()
callback_handler = MagicMock()
return bellows.ezsp.v4.EZSPv4(callback_handler, gateway)
@pytest.fixture
def prot_hndl_v9():
"""Protocol handler mock."""
app = MagicMock()
gateway = Gateway(app)
gateway._transport = AsyncMock()
callback_handler = MagicMock()
return bellows.ezsp.v9.EZSPv9(callback_handler, gateway)
async def test_command(prot_hndl):
with patch.object(prot_hndl._gw, "send_data") as mock_send_data:
coro = prot_hndl.command("nop")
asyncio.get_running_loop().call_soon(
lambda: prot_hndl._awaiting[prot_hndl._seq - 1][2].set_result(True)
)
await coro
assert mock_send_data.mock_calls == [call(b"\x00\x00\x05")]
def test_receive_reply(prot_hndl):
callback_mock = MagicMock(spec_set=asyncio.Future)
prot_hndl._awaiting[0] = (0, prot_hndl.COMMANDS["version"][2], callback_mock)
prot_hndl(b"\x00\xff\x00\x04\x05\x06\x00")
assert 0 not in prot_hndl._awaiting
assert callback_mock.set_exception.call_count == 0
assert callback_mock.set_result.call_count == 1
callback_mock.set_result.assert_called_once_with([4, 5, 6])
assert prot_hndl._handle_callback.call_count == 0
def test_receive_reply_after_timeout(prot_hndl):
callback_mock = MagicMock(spec_set=asyncio.Future)
callback_mock.set_result.side_effect = asyncio.InvalidStateError()
prot_hndl._awaiting[0] = (0, prot_hndl.COMMANDS["version"][2], callback_mock)
prot_hndl(b"\x00\xff\x00\x04\x05\x06\x00")
assert 0 not in prot_hndl._awaiting
assert callback_mock.set_exception.call_count == 0
assert callback_mock.set_result.call_count == 1
callback_mock.set_result.assert_called_once_with([4, 5, 6])
assert prot_hndl._handle_callback.call_count == 0
def test_receive_reply_invalid_command(prot_hndl):
callback_mock = MagicMock(spec_set=asyncio.Future)
prot_hndl._awaiting[0] = (0, prot_hndl.COMMANDS["invalidCommand"][2], callback_mock)
prot_hndl(b"\x00\xff\x58\x31")
assert 0 not in prot_hndl._awaiting
assert callback_mock.set_exception.call_count == 1
assert callback_mock.set_result.call_count == 0
assert prot_hndl._handle_callback.call_count == 0
async def test_update_policies(prot_hndl):
"""Test update_policies."""
with patch.object(prot_hndl, "setPolicy", new=AsyncMock()) as pol_mock:
pol_mock.return_value = (t.EzspStatus.SUCCESS,)
await prot_hndl.update_policies({})
with patch.object(prot_hndl, "setPolicy", new=AsyncMock()) as pol_mock:
pol_mock.return_value = (t.EzspStatus.ERROR_OUT_OF_MEMORY,)
with pytest.raises(AssertionError):
await prot_hndl.update_policies({})
async def test_unknown_command(prot_hndl, caplog):
"""Test receiving an unknown command."""
unregistered_command = 0x04
with caplog.at_level(logging.WARNING):
prot_hndl(bytes([0x00, 0x00, unregistered_command, 0xAB, 0xCD]))
assert "0x0004 received: b'abcd' (b'000004abcd')" in caplog.text
async def test_logging_frame_parsing_failure(prot_hndl, caplog) -> None:
"""Test logging when frame parsing fails."""
with caplog.at_level(logging.WARNING):
with pytest.raises(ValueError):
prot_hndl(b"\xAA\xAA\x71\x22")
assert "Failed to parse frame getKeyTableEntry: b'22'" in caplog.text
async def test_parsing_schema_response(prot_hndl_v9):
"""Test parsing data with a struct schema."""
coro = prot_hndl_v9.command(
"getTokenData", NV3KeyId.CREATOR_STACK_RESTORED_EUI64, 0
)
asyncio.get_running_loop().call_soon(
lambda: prot_hndl_v9(
bytes([prot_hndl_v9._seq - 1, 0x00, 0x00])
+ t.uint16_t(prot_hndl_v9.COMMANDS["getTokenData"][0]).serialize()
+ bytes([0xB5])
)
)
rsp = await coro
assert rsp == GetTokenDataRsp(status=t.EmberStatus.LIBRARY_NOT_PRESENT)
|
