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
|
"""Unit test for KNX/IP ConnectResponses."""
import pytest
from xknx.exceptions import CouldNotParseKNXIP
from xknx.knxip import (
HPAI,
ConnectRequestType,
ConnectResponse,
ConnectResponseData,
ErrorCode,
KNXIPFrame,
)
from xknx.telegram import IndividualAddress
class TestKNXIPConnectResponse:
"""Test class for KNX/IP ConnectResponses."""
def test_tunnel_connect_response(self) -> None:
"""Test parsing and streaming connection response KNX/IP packet."""
raw = bytes.fromhex(
"06 10 02 06 00 14 01 00 08 01 C0 A8 2A 0A 0E 57 04 04 11 FF"
)
knxipframe, _ = KNXIPFrame.from_knx(raw)
assert isinstance(knxipframe.body, ConnectResponse)
assert knxipframe.body.communication_channel == 1
assert knxipframe.body.status_code == ErrorCode.E_NO_ERROR
assert knxipframe.body.data_endpoint == HPAI(ip_addr="192.168.42.10", port=3671)
assert knxipframe.body.crd.request_type == ConnectRequestType.TUNNEL_CONNECTION
assert knxipframe.body.crd.individual_address.raw == 4607
connect_response = ConnectResponse(
communication_channel=1,
status_code=ErrorCode.E_NO_ERROR,
data_endpoint=HPAI(ip_addr="192.168.42.10", port=3671),
crd=ConnectResponseData(
request_type=ConnectRequestType.TUNNEL_CONNECTION,
individual_address=IndividualAddress(4607),
),
)
knxipframe2 = KNXIPFrame.init_from_body(connect_response)
assert knxipframe2.to_knx() == raw
def test_mgmt_connect_response(self) -> None:
"""Test parsing and streaming connection response KNX/IP packet."""
raw = bytes.fromhex("06 10 02 06 00 12 01 00 08 01 C0 A8 2A 0A 0E 57 02 03")
knxipframe, _ = KNXIPFrame.from_knx(raw)
assert isinstance(knxipframe.body, ConnectResponse)
assert knxipframe.body.communication_channel == 1
assert knxipframe.body.status_code == ErrorCode.E_NO_ERROR
assert knxipframe.body.data_endpoint == HPAI(ip_addr="192.168.42.10", port=3671)
assert (
knxipframe.body.crd.request_type
== ConnectRequestType.DEVICE_MGMT_CONNECTION
)
assert knxipframe.body.crd.individual_address is None
connect_response = ConnectResponse(
communication_channel=1,
status_code=ErrorCode.E_NO_ERROR,
data_endpoint=HPAI(ip_addr="192.168.42.10", port=3671),
crd=ConnectResponseData(
request_type=ConnectRequestType.DEVICE_MGMT_CONNECTION,
),
)
knxipframe2 = KNXIPFrame.init_from_body(connect_response)
assert knxipframe2.to_knx() == raw
def test_from_knx_wrong_crd(self) -> None:
"""Test parsing and streaming wrong ConnectRequest (wrong CRD length byte)."""
raw = bytes.fromhex(
"06 10 02 06 00 14 01 00 08 01 C0 A8 2A 0A 0E 57 03 04 11 FF"
)
with pytest.raises(CouldNotParseKNXIP):
KNXIPFrame.from_knx(raw)
def test_from_knx_wrong_crd2(self) -> None:
"""Test parsing and streaming wrong ConnectRequest (wrong CRD length)."""
raw = bytes.fromhex("06 10 02 06 00 12 01 00 08 01 C0 A8 2A 0A 0E 57 04 04")
with pytest.raises(CouldNotParseKNXIP):
KNXIPFrame.from_knx(raw)
def test_from_knx_wrong_crd3(self) -> None:
"""Test parsing and streaming wrong ConnectRequest (CRD length too small)."""
raw = bytes.fromhex("06 10 02 06 00 12 01 00 08 01 C0 A8 2A 0A 0E 57 01 04")
with pytest.raises(CouldNotParseKNXIP):
KNXIPFrame.from_knx(raw)
def test_from_knx_wrong_crd4(self) -> None:
"""Test parsing and streaming wrong ConnectRequest (wrong CRD length)."""
raw = bytes.fromhex("06 10 02 06 00 12 01 00 08 01 C0 A8 2A 0A 0E 57 04 03")
with pytest.raises(CouldNotParseKNXIP):
KNXIPFrame.from_knx(raw)
def test_from_knx_wrong_crd5(self) -> None:
"""Test parsing and streaming wrong ConnectRequest (wrong CRD length)."""
raw = bytes.fromhex("06 10 02 06 00 13 01 00 08 01 C0 A8 2A 0A 0E 57 03 03 01")
with pytest.raises(CouldNotParseKNXIP):
KNXIPFrame.from_knx(raw)
def test_connect_response_connection_error_gira(self) -> None:
"""
Test parsing and streaming connection response KNX/IP packet with error e_no_more_connections.
HPAI and CRD normal. This was received from Gira devices (2020).
"""
raw = bytes.fromhex(
"06 10 02 06 00 14 C0 24 08 01 0A 01 00 29 0E 57 04 04 00 00"
)
knxipframe, _ = KNXIPFrame.from_knx(raw)
assert isinstance(knxipframe.body, ConnectResponse)
assert knxipframe.body.status_code == ErrorCode.E_NO_MORE_CONNECTIONS
assert knxipframe.body.communication_channel == 192
connect_response = ConnectResponse(
communication_channel=192,
status_code=ErrorCode.E_NO_MORE_CONNECTIONS,
data_endpoint=HPAI(ip_addr="10.1.0.41", port=3671),
crd=ConnectResponseData(
request_type=ConnectRequestType.TUNNEL_CONNECTION,
individual_address=IndividualAddress(0),
),
)
knxipframe2 = KNXIPFrame.init_from_body(connect_response)
assert knxipframe2.to_knx() == raw
def test_connect_response_connection_error_lox(self) -> None:
"""
Test parsing and streaming connection response KNX/IP packet with error e_no_more_connections.
HPAI given, CRD all zero. This was received from Loxone device (2020).
"""
raw = bytes.fromhex(
"06 10 02 06 00 14 00 24 08 01 C0 A8 01 01 0E 57 00 00 00 00"
)
knxipframe, _ = KNXIPFrame.from_knx(raw)
assert isinstance(knxipframe.body, ConnectResponse)
assert knxipframe.body.status_code == ErrorCode.E_NO_MORE_CONNECTIONS
assert knxipframe.body.communication_channel == 0
# no further tests: the current API can't (and shouldn't) create such odd packets
def test_connect_response_connection_error_mdt(self) -> None:
"""
Test parsing and streaming connection response KNX/IP packet with error e_no_more_connections.
HPAI and CRD all zero. This was received from MDT device (2020).
"""
raw = bytes.fromhex("06 10 02 06 00 08 00 24 00 00 00 00 00 00 00 00 00 00")
knxipframe, _ = KNXIPFrame.from_knx(raw)
assert isinstance(knxipframe.body, ConnectResponse)
assert knxipframe.body.status_code == ErrorCode.E_NO_MORE_CONNECTIONS
assert knxipframe.body.communication_channel == 0
# no further tests: the current API can't (and shouldn't) create such odd packets
|
