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
|
"""Unit test for RemoteValueRaw objects."""
import pytest
from xknx import XKNX
from xknx.dpt import DPTArray, DPTBinary
from xknx.exceptions import ConversionError
from xknx.remote_value import RemoteValueRaw
from xknx.telegram import GroupAddress, Telegram
from xknx.telegram.apci import GroupValueWrite
class TestRemoteValueRaw:
"""Test class for RemoteValueRaw objects."""
def test_to_knx(self) -> None:
"""Test to_knx function with normal operation."""
xknx = XKNX()
rv_0 = RemoteValueRaw(xknx, payload_length=0)
rv_1 = RemoteValueRaw(xknx, payload_length=1)
rv_2 = RemoteValueRaw(xknx, payload_length=2)
assert rv_0.to_knx(1) == DPTBinary(True)
assert rv_0.to_knx(4) == DPTBinary(4)
assert rv_1.to_knx(100) == DPTArray((0x64,))
assert rv_2.to_knx(100) == DPTArray((0x00, 0x64))
def test_from_knx(self) -> None:
"""Test from_knx function with normal operation."""
xknx = XKNX()
rv_0 = RemoteValueRaw(xknx, payload_length=0)
rv_1 = RemoteValueRaw(xknx, payload_length=1)
rv_2 = RemoteValueRaw(xknx, payload_length=2)
assert rv_0.from_knx(DPTBinary(True)) == 1
assert rv_0.from_knx(DPTBinary(0x4)) == 4
assert rv_1.from_knx(DPTArray((0x64,))) == 100
assert rv_2.from_knx(DPTArray((0x00, 0x64))) == 100
with pytest.raises(ConversionError):
assert rv_1.from_knx(DPTArray((256,)))
async def test_set(self) -> None:
"""Test setting value."""
xknx = XKNX()
rv_0 = RemoteValueRaw(xknx, payload_length=0, group_address="1/2/3")
rv_1 = RemoteValueRaw(xknx, payload_length=1, group_address="1/2/3")
rv_2 = RemoteValueRaw(xknx, payload_length=2, group_address="1/2/3")
rv_0.set(0)
assert xknx.telegrams.qsize() == 1
telegram = xknx.telegrams.get_nowait()
assert telegram == Telegram(
destination_address=GroupAddress("1/2/3"),
payload=GroupValueWrite(DPTBinary(False)),
)
rv_0.set(63)
assert xknx.telegrams.qsize() == 1
telegram = xknx.telegrams.get_nowait()
assert telegram == Telegram(
destination_address=GroupAddress("1/2/3"),
payload=GroupValueWrite(DPTBinary(0x3F)),
)
rv_1.set(0)
assert xknx.telegrams.qsize() == 1
telegram = xknx.telegrams.get_nowait()
assert telegram == Telegram(
destination_address=GroupAddress("1/2/3"),
payload=GroupValueWrite(DPTArray((0x00,))),
)
rv_1.set(63)
assert xknx.telegrams.qsize() == 1
telegram = xknx.telegrams.get_nowait()
assert telegram == Telegram(
destination_address=GroupAddress("1/2/3"),
payload=GroupValueWrite(DPTArray((0x3F,))),
)
rv_2.set(0)
assert xknx.telegrams.qsize() == 1
telegram = xknx.telegrams.get_nowait()
assert telegram == Telegram(
destination_address=GroupAddress("1/2/3"),
payload=GroupValueWrite(DPTArray((0x00, 0x00))),
)
rv_2.set(63)
assert xknx.telegrams.qsize() == 1
telegram = xknx.telegrams.get_nowait()
assert telegram == Telegram(
destination_address=GroupAddress("1/2/3"),
payload=GroupValueWrite(DPTArray((0x00, 0x3F))),
)
def test_process(self) -> None:
"""Test process telegram."""
xknx = XKNX()
rv_0 = RemoteValueRaw(xknx, payload_length=0, group_address="1/0/0")
rv_1 = RemoteValueRaw(xknx, payload_length=1, group_address="1/1/1")
rv_2 = RemoteValueRaw(xknx, payload_length=2, group_address="1/2/2")
telegram = Telegram(
destination_address=GroupAddress("1/0/0"),
payload=GroupValueWrite(DPTBinary(0)),
)
rv_0.process(telegram)
assert rv_0.value == 0
telegram = Telegram(
destination_address=GroupAddress("1/1/1"),
payload=GroupValueWrite(DPTArray((0x64,))),
)
rv_1.process(telegram)
assert rv_1.value == 100
telegram = Telegram(
destination_address=GroupAddress("1/2/2"),
payload=GroupValueWrite(DPTArray((0x12, 0x34))),
)
rv_2.process(telegram)
assert rv_2.value == 4660
def test_to_process_error(self) -> None:
"""Test process erroneous telegram."""
xknx = XKNX()
rv_0 = RemoteValueRaw(xknx, payload_length=0, group_address="1/0/0")
rv_1 = RemoteValueRaw(xknx, payload_length=1, group_address="1/1/1")
rv_2 = RemoteValueRaw(xknx, payload_length=2, group_address="1/2/2")
telegram = Telegram(
destination_address=GroupAddress("1/0/0"),
payload=GroupValueWrite(DPTArray((0x01,))),
)
assert rv_0.process(telegram) is False
telegram = Telegram(
destination_address=GroupAddress("1/0/0"),
payload=GroupValueWrite(DPTArray((0x64, 0x65))),
)
assert rv_0.process(telegram) is False
assert rv_0.value is None
telegram = Telegram(
destination_address=GroupAddress("1/1/1"),
payload=GroupValueWrite(DPTBinary(1)),
)
assert rv_1.process(telegram) is False
telegram = Telegram(
destination_address=GroupAddress("1/1/1"),
payload=GroupValueWrite(DPTArray((0x64, 0x65))),
)
assert rv_1.process(telegram) is False
assert rv_1.value is None
telegram = Telegram(
destination_address=GroupAddress("1/2/2"),
payload=GroupValueWrite(DPTBinary(1)),
)
assert rv_2.process(telegram) is False
telegram = Telegram(
destination_address=GroupAddress("1/2/2"),
payload=GroupValueWrite(DPTArray((0x64,))),
)
assert rv_2.process(telegram) is False
assert rv_2.value is None
def test_to_knx_error(self) -> None:
"""Test to_knx function with wrong parameters."""
xknx = XKNX()
rv_0 = RemoteValueRaw(xknx, payload_length=0, group_address="1/0/0")
rv_1 = RemoteValueRaw(xknx, payload_length=1, group_address="1/1/1")
rv_2 = RemoteValueRaw(xknx, payload_length=2, group_address="1/2/2")
with pytest.raises(ConversionError):
rv_0.to_knx(-1)
with pytest.raises(ConversionError):
rv_0.to_knx(64)
with pytest.raises(ConversionError):
rv_0.to_knx(5.5)
with pytest.raises(ConversionError):
rv_0.to_knx("a")
with pytest.raises(ConversionError):
rv_1.to_knx(-1)
with pytest.raises(ConversionError):
rv_1.to_knx(256)
with pytest.raises(ConversionError):
rv_1.to_knx(5.5)
with pytest.raises(ConversionError):
rv_1.to_knx("a")
with pytest.raises(ConversionError):
rv_2.to_knx(-1)
with pytest.raises(ConversionError):
rv_2.to_knx(65536)
with pytest.raises(ConversionError):
rv_2.to_knx(5.5)
with pytest.raises(ConversionError):
rv_2.to_knx("a")
|
