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
|
"""Unit test for KNX/IP Connect Request/Response."""
from unittest.mock import patch
from xknx.io.request_response import Connect
from xknx.io.transport import UDPTransport
from xknx.knxip import (
HPAI,
ConnectionStateRequest,
ConnectRequest,
ConnectResponse,
ConnectResponseData,
ErrorCode,
KNXIPFrame,
)
from xknx.telegram import IndividualAddress
class TestConnect:
"""Test class for xknx/io/Connect objects."""
async def test_connect(self) -> None:
"""Test connecting from KNX bus."""
udp_transport = UDPTransport(("192.168.1.1", 0), ("192.168.1.2", 1234))
local_hpai = HPAI(ip_addr="192.168.1.3", port=4321)
connect = Connect(udp_transport, local_hpai=local_hpai)
connect.timeout_in_seconds = 0
assert connect.awaited_response_class == ConnectResponse
# Expected KNX/IP-Frame:
exp_knxipframe = KNXIPFrame.init_from_body(
ConnectRequest(
control_endpoint=local_hpai,
data_endpoint=local_hpai,
)
)
with (
patch("xknx.io.transport.UDPTransport.send") as mock_udp_send,
patch("xknx.io.transport.UDPTransport.getsockname") as mock_udp_getsockname,
):
mock_udp_getsockname.return_value = ("192.168.1.3", 4321)
await connect.start()
mock_udp_send.assert_called_with(exp_knxipframe)
# Response KNX/IP-Frame with wrong type
wrong_knxipframe = KNXIPFrame.init_from_body(ConnectionStateRequest())
with patch("logging.Logger.warning") as mock_warning:
connect.response_rec_callback(wrong_knxipframe, HPAI(), None)
mock_warning.assert_called_with("Could not understand knxipframe")
# Response KNX/IP-Frame with error:
err_knxipframe = KNXIPFrame.init_from_body(
ConnectResponse(status_code=ErrorCode.E_CONNECTION_ID)
)
with patch("logging.Logger.debug") as mock_warning:
connect.response_rec_callback(err_knxipframe, HPAI(), None)
mock_warning.assert_called_with(
"Error: KNX bus responded to request of type '%s' with error in '%s': %s",
type(connect).__name__,
type(err_knxipframe.body).__name__,
ErrorCode.E_CONNECTION_ID,
)
# Correct Response KNX/IP-Frame:
res_knxipframe = KNXIPFrame.init_from_body(
ConnectResponse(
communication_channel=23,
crd=ConnectResponseData(individual_address=IndividualAddress(7)),
)
)
connect.response_rec_callback(res_knxipframe, HPAI(), None)
assert connect.success
assert connect.communication_channel == 23
assert connect.crd.individual_address.raw == 7
async def test_connect_route_back_true(self) -> None:
"""Test connecting from KNX bus."""
udp_transport = UDPTransport(("192.168.1.1", 0), ("192.168.1.2", 1234))
local_hpai = HPAI() # route_back: No IP address, no port, UDP
connect = Connect(udp_transport, local_hpai=local_hpai)
connect.timeout_in_seconds = 0
assert connect.awaited_response_class == ConnectResponse
# Expected KNX/IP-Frame:
exp_knxipframe = KNXIPFrame.init_from_body(ConnectRequest())
with (
patch("xknx.io.transport.UDPTransport.send") as mock_udp_send,
patch("xknx.io.transport.UDPTransport.getsockname") as mock_udp_getsockname,
):
mock_udp_getsockname.return_value = ("192.168.1.3", 4321)
await connect.start()
mock_udp_send.assert_called_with(exp_knxipframe)
# Response KNX/IP-Frame with wrong type
wrong_knxipframe = KNXIPFrame.init_from_body(ConnectionStateRequest())
with patch("logging.Logger.warning") as mock_warning:
connect.response_rec_callback(wrong_knxipframe, HPAI(), None)
mock_warning.assert_called_with("Could not understand knxipframe")
# Response KNX/IP-Frame with error:
err_knxipframe = KNXIPFrame.init_from_body(
ConnectResponse(status_code=ErrorCode.E_CONNECTION_ID)
)
with patch("logging.Logger.debug") as mock_warning:
connect.response_rec_callback(err_knxipframe, HPAI(), None)
mock_warning.assert_called_with(
"Error: KNX bus responded to request of type '%s' with error in '%s': %s",
type(connect).__name__,
type(err_knxipframe.body).__name__,
ErrorCode.E_CONNECTION_ID,
)
# Correct Response KNX/IP-Frame:
res_knxipframe = KNXIPFrame.init_from_body(
ConnectResponse(
communication_channel=23,
crd=ConnectResponseData(individual_address=IndividualAddress(7)),
)
)
connect.response_rec_callback(res_knxipframe, HPAI(), None)
assert connect.success
assert connect.communication_channel == 23
assert connect.crd.individual_address.raw == 7
|
