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
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
|
"""Test for CEMIHandler."""
import asyncio
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from xknx import XKNX
from xknx.cemi import CEMIFrame, CEMILData, CEMIMessageCode
from xknx.dpt import DPTArray
from xknx.exceptions import ConfirmationError
from xknx.telegram import GroupAddress, IndividualAddress, Telegram, apci, tpci
from ..conftest import EventLoopClockAdvancer
async def test_wait_for_l2_confirmation(time_travel: EventLoopClockAdvancer) -> None:
"""Test waiting for L_DATA.con before sending another L_DATA.req."""
xknx = XKNX()
xknx.knxip_interface = AsyncMock()
test_telegram = Telegram(
destination_address=GroupAddress(1),
payload=apci.GroupValueWrite(DPTArray((1,))),
)
test_cemi = CEMIFrame(
code=CEMIMessageCode.L_DATA_REQ,
data=CEMILData.init_from_telegram(test_telegram),
)
test_cemi_confirmation = CEMIFrame(
code=CEMIMessageCode.L_DATA_CON,
data=CEMILData.init_from_telegram(
test_telegram,
),
)
task = asyncio.create_task(xknx.cemi_handler.send_telegram(test_telegram))
await time_travel(0)
xknx.knxip_interface.send_cemi.assert_called_once_with(test_cemi)
assert xknx.connection_manager.cemi_count_outgoing == 0
assert not task.done()
xknx.cemi_handler.handle_cemi_frame(test_cemi_confirmation)
await time_travel(0)
assert task.done()
await task
assert xknx.connection_manager.cemi_count_outgoing == 1
assert xknx.connection_manager.cemi_count_outgoing_error == 0
# no L_DATA.con received -> raise ConfirmationError
xknx.knxip_interface.send_cemi.reset_mock()
task = asyncio.create_task(xknx.cemi_handler.send_telegram(test_telegram))
await time_travel(0)
xknx.knxip_interface.send_cemi.assert_called_once_with(test_cemi)
with pytest.raises(ConfirmationError):
await time_travel(3)
assert task.done()
await task
assert xknx.connection_manager.cemi_count_outgoing == 1
assert xknx.connection_manager.cemi_count_outgoing_error == 1
@patch("xknx.management.management.Management.process")
def test_incoming_cemi(mock_management_process: MagicMock) -> None:
"""Test incoming CEMI."""
xknx = XKNX()
xknx.current_address = IndividualAddress("1.1.1")
# TDataGroup Telegram
test_telegram = Telegram(
destination_address=GroupAddress(1),
payload=apci.GroupValueWrite(DPTArray((1,))),
)
test_group_cemi = CEMIFrame(
code=CEMIMessageCode.L_DATA_IND,
data=CEMILData.init_from_telegram(test_telegram),
)
xknx.cemi_handler.handle_cemi_frame(test_group_cemi)
assert xknx.telegrams.qsize() == 1
mock_management_process.assert_not_called()
xknx.telegrams.get_nowait() # remove telegram from queue
# L_DATA_CON and L_DATA_REQ should not be forwarded to the telegram queue or management
test_incoming_l_data_con = CEMIFrame(
code=CEMIMessageCode.L_DATA_CON,
data=CEMILData.init_from_telegram(test_telegram),
)
xknx.cemi_handler.handle_cemi_frame(test_incoming_l_data_con)
assert not xknx.telegrams.qsize()
mock_management_process.assert_not_called()
test_incoming_l_data_req = CEMIFrame(
code=CEMIMessageCode.L_DATA_REQ,
data=CEMILData.init_from_telegram(test_telegram),
)
xknx.cemi_handler.handle_cemi_frame(test_incoming_l_data_req)
assert not xknx.telegrams.qsize()
mock_management_process.assert_not_called()
assert xknx.connection_manager.cemi_count_incoming == 1
@pytest.mark.parametrize(
"telegram",
[
Telegram(
destination_address=GroupAddress(0),
tpci=tpci.TDataBroadcast(),
),
Telegram(
destination_address=IndividualAddress("1.1.1"),
tpci=tpci.TConnect(),
),
Telegram(
destination_address=IndividualAddress("1.1.1"),
tpci=tpci.TDataIndividual(),
),
],
)
@patch("xknx.management.management.Management.process")
def test_incoming_management_telegram(
mock_management_process: MagicMock, telegram: Telegram
) -> None:
"""Test incoming management CEMI."""
xknx = XKNX()
xknx.current_address = IndividualAddress("1.1.1")
test_cemi = CEMIFrame(
code=CEMIMessageCode.L_DATA_IND,
data=CEMILData.init_from_telegram(telegram),
)
xknx.cemi_handler.handle_cemi_frame(test_cemi)
mock_management_process.assert_called_once()
assert xknx.telegrams.qsize() == 0
assert xknx.connection_manager.cemi_count_incoming == 1
@pytest.mark.parametrize(
"raw",
[
# <CouldNotParseCEMI description="CEMI too small. Length: 9; CEMI: 2900b06010fa10ff00" />
# communication_channel_id: 0x02 sequence_counter: 0x81
bytes.fromhex("2900b06010fa10ff00"),
],
)
@patch("xknx.cemi.cemi_handler.CEMIHandler.handle_cemi_frame")
@patch("logging.Logger.warning")
def test_invalid_cemi(
mock_warning: MagicMock, mock_handle_cemi_frame: MagicMock, raw: bytes
) -> None:
"""Test incoming invalid CEMI Frames."""
xknx = XKNX()
xknx.cemi_handler.handle_raw_cemi(raw)
mock_warning.assert_called_once()
mock_handle_cemi_frame.assert_not_called()
assert xknx.connection_manager.cemi_count_incoming_error == 1
@pytest.mark.parametrize(
"raw",
[
# LDataInd Unsupported Extended APCI from 0.0.1 to 0/0/0 broadcast
# <UnsupportedCEMIMessage description="APCI not supported: 0b1111111000 in CEMI: 2900b0d0000100000103f8" />
bytes.fromhex("2900b0d0000100000103f8"),
],
)
@patch("xknx.cemi.cemi_handler.CEMIHandler.handle_cemi_frame")
@patch("logging.Logger.info")
def test_unsupported_cemi(
mock_info: MagicMock, mock_handle_cemi_frame: MagicMock, raw: bytes
) -> None:
"""Test incoming unsupported CEMI Frames."""
xknx = XKNX()
xknx.cemi_handler.handle_raw_cemi(raw)
mock_info.assert_called_once()
mock_handle_cemi_frame.assert_not_called()
assert xknx.connection_manager.cemi_count_incoming_error == 1
@patch("xknx.cemi.cemi_handler.CEMIHandler.telegram_received")
@patch("logging.Logger.debug")
def test_incoming_from_own_ia(
mock_debug: MagicMock, mock_telegram_received: MagicMock
) -> None:
"""Test incoming CEMI from own IA."""
xknx = XKNX()
xknx.current_address = IndividualAddress("1.1.22")
# L_Data.ind GroupValueWrite from 1.1.22 to to 5/1/22 with DPT9 payload 0C 3F
raw = bytes.fromhex("2900bcd011162916030080 0c 3f")
xknx.cemi_handler.handle_raw_cemi(raw)
mock_debug.assert_called_once()
mock_telegram_received.assert_called_once()
assert xknx.connection_manager.cemi_count_incoming == 1
assert xknx.connection_manager.cemi_count_incoming_error == 0
@pytest.mark.parametrize(
"raw_cemi_data_secure",
[
# src = 4.0.9; dst = 0/4/0; GroupValueResponse; value=(116, 41, 41)
# A+C; seq_num=155806854986
bytes.fromhex("29003ce0400904001103f110002446cfef4ac085e7092ab062b44d"),
# Property Value Write PID_GRP_KEY_TABLE connectionless
# Object Idx = 5, PropId = 35h, Element Count = 1, Index = 1
# Data = 20 21 22 23 24 25 26 27 28 29 2A 2B 2C 2D 2E 2F
# A+C
# from AN158 v07 KNX Data Security AS - Annex A example
bytes.fromhex(
"29 00 b0 60 ff 67 ff 00 22 03 f1 90 00 00 00 00"
"00 04 67 67 24 2a 23 08 ca 76 a1 17 74 21 4e e4"
"cf 5d 94 90 9f 74 3d 05 0d 8f c1 68"
),
],
)
@patch("xknx.cemi.cemi_handler.CEMIHandler.telegram_received")
def test_incoming_cemi_no_data_secure_keys(
mock_telegram_received: MagicMock,
raw_cemi_data_secure: bytes,
) -> None:
"""Test incoming DataSecure CEMI when no DataSecure keys are initialized."""
xknx = XKNX()
xknx.current_address = IndividualAddress("5.0.1")
with (
patch("logging.Logger.debug") as mock_debug,
):
xknx.cemi_handler.handle_raw_cemi(raw_cemi_data_secure)
assert mock_debug.call_count == 2 # one for reception, one for DataSecure debug
# frame is dropped because no keys are available
mock_telegram_received.assert_not_called()
# not having a key isn't an error per se, so no incoming_error count
assert xknx.connection_manager.cemi_count_incoming == 1
assert xknx.connection_manager.cemi_count_incoming_error == 0
assert xknx.connection_manager.undecoded_data_secure == 1
|
