-
-
Notifications
You must be signed in to change notification settings - Fork 96
/
tunnelling_test.py
98 lines (87 loc) · 3.62 KB
/
tunnelling_test.py
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
"""Unit test for KNX/IP Tunnelling Request/Response."""
import asyncio
import unittest
from unittest.mock import patch
from xknx import XKNX
from xknx.dpt import DPTArray
from xknx.io import UDPClient
from xknx.io.request_response import Tunnelling
from xknx.knxip import (
ErrorCode,
KNXIPFrame,
KNXIPServiceType,
TunnellingAck,
TunnellingRequest,
)
from xknx.telegram import GroupAddress, IndividualAddress, Telegram
from xknx.telegram.apci import GroupValueWrite
class TestTunnelling(unittest.TestCase):
"""Test class for xknx/io/Tunnelling objects."""
def setUp(self):
"""Set up test class."""
self.loop = asyncio.new_event_loop()
asyncio.set_event_loop(self.loop)
def tearDown(self):
"""Tear down test class."""
self.loop.close()
def test_tunnelling(self):
"""Test tunnelling from KNX bus."""
# pylint: disable=too-many-locals
xknx = XKNX()
communication_channel_id = 23
udp_client = UDPClient(xknx, ("192.168.1.1", 0), ("192.168.1.2", 1234))
telegram = Telegram(
destination_address=GroupAddress("1/2/3"),
payload=GroupValueWrite(DPTArray((0x1, 0x2, 0x3))),
)
sequence_counter = 42
src_address = IndividualAddress("2.2.2")
tunnelling = Tunnelling(
xknx,
udp_client,
telegram,
src_address,
sequence_counter,
communication_channel_id,
)
tunnelling.timeout_in_seconds = 0
self.assertEqual(tunnelling.awaited_response_class, TunnellingAck)
self.assertEqual(tunnelling.communication_channel_id, communication_channel_id)
# Expected KNX/IP-Frame:
tunnelling_request = TunnellingRequest(
xknx,
communication_channel_id=communication_channel_id,
sequence_counter=sequence_counter,
)
tunnelling_request.cemi.telegram = telegram
tunnelling_request.cemi.src_addr = src_address
exp_knxipframe = KNXIPFrame.init_from_body(tunnelling_request)
with patch("xknx.io.UDPClient.send") as mock_udp_send, patch(
"xknx.io.UDPClient.getsockname"
) as mock_udp_getsockname:
mock_udp_getsockname.return_value = ("192.168.1.3", 4321)
self.loop.run_until_complete(tunnelling.start())
mock_udp_send.assert_called_with(exp_knxipframe)
# Response KNX/IP-Frame with wrong type
wrong_knxipframe = KNXIPFrame(xknx)
wrong_knxipframe.init(KNXIPServiceType.CONNECTIONSTATE_REQUEST)
with patch("logging.Logger.warning") as mock_warning:
tunnelling.response_rec_callback(wrong_knxipframe, None)
mock_warning.assert_called_with("Could not understand knxipframe")
# Response KNX/IP-Frame with error:
err_knxipframe = KNXIPFrame(xknx)
err_knxipframe.init(KNXIPServiceType.TUNNELLING_ACK)
err_knxipframe.body.status_code = ErrorCode.E_CONNECTION_ID
with patch("logging.Logger.debug") as mock_warning:
tunnelling.response_rec_callback(err_knxipframe, None)
mock_warning.assert_called_with(
"Error: KNX bus responded to request of type '%s' with error in '%s': %s",
type(tunnelling).__name__,
type(err_knxipframe.body).__name__,
ErrorCode.E_CONNECTION_ID,
)
# Correct Response KNX/IP-Frame:
res_knxipframe = KNXIPFrame(xknx)
res_knxipframe.init(KNXIPServiceType.TUNNELLING_ACK)
tunnelling.response_rec_callback(res_knxipframe, None)
self.assertTrue(tunnelling.success)