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
|
# Copyright 2022 Scott K Logan
# Licensed under the Apache License, Version 2.0
import asyncio
import os
from aioraven.data import MeterList
from aioraven.device import RAVEnConnectionError
from aioraven.device import RAVEnNotOpenError
from aioraven.serial import RAVEnSerialDevice
import pytest
from .mock_device import mock_pty_device
if os.name == 'nt':
pytest.skip(
'Pseudo-terminals are not supported on non-Unix platforms',
allow_module_level=True)
@pytest.mark.asyncio
async def test_serial_data():
"""Verify simple device query behavior."""
responses = {
b'<Command><Name>get_meter_list</Name></Command>':
b'<MeterList>'
b' <DeviceMacId>0x0123456789abcdef</DeviceMacId>'
b'</MeterList>',
}
async with mock_pty_device(responses) as mock_device:
async with RAVEnSerialDevice(mock_device) as dut:
actual = await dut.get_meter_list()
assert actual == MeterList(
device_mac_id=bytes.fromhex('0123456789ABCDEF'),
meter_mac_ids=[])
@pytest.mark.asyncio
async def test_serial_disconnect():
"""Verify behavior when a device is unexpectedly disconnected."""
responses = {
b'<Command><Name>get_meter_list</Name></Command>':
b'<MeterList>'
b' <DeviceMacId>0x0123456789abcdef</DeviceMacId>'
b'</MeterList>',
}
async with mock_pty_device(responses) as mock_device:
dut = RAVEnSerialDevice(mock_device)
await dut.open()
assert await dut.get_meter_list()
async with dut:
with pytest.raises(RAVEnConnectionError):
await dut.get_meter_list()
with pytest.raises(RAVEnConnectionError):
await dut.get_meter_list()
@pytest.mark.asyncio
async def test_serial_incomplete():
"""Verify behavior when a partial fragment is received."""
responses = {
b'<Command><Name>get_meter_list</Name></Command>':
b'<MeterList>'
b' <DeviceMacId>0x0123456789abcdef</DeviceMacId>'
b'</MeterList>'
b'<DeviceInfo>'
}
async with mock_pty_device(responses) as mock_device:
dut = RAVEnSerialDevice(mock_device)
await dut.open()
assert await dut.get_meter_list()
task = asyncio.create_task(dut.get_meter_list())
await asyncio.wait((task,), timeout=0.05)
async with dut:
with pytest.raises(RAVEnConnectionError):
assert not await task
@pytest.mark.asyncio
async def test_serial_not_open():
"""Verify behavior when reading from an unopened device."""
async with mock_pty_device({}) as mock_device:
dut = RAVEnSerialDevice(mock_device)
with pytest.raises(RAVEnNotOpenError):
await dut.get_device_info()
@pytest.mark.asyncio
async def test_serial_parse_error():
"""Verify behavior when invalid syntax is received."""
responses = {
b'<Command><Name>get_device_info</Name></Command>':
b'</DeviceInfo>',
b'<Command><Name>get_meter_list</Name></Command>':
b'<MeterList>'
b' <DeviceMacId>0x0123456789abcdef</DeviceMacId>'
b'</MeterList>',
}
async with mock_pty_device(responses) as mock_device:
async with RAVEnSerialDevice(mock_device) as dut:
with pytest.raises(RAVEnConnectionError):
await dut.get_device_info()
actual = await dut.get_meter_list()
assert actual == MeterList(
device_mac_id=bytes.fromhex('0123456789ABCDEF'),
meter_mac_ids=[])
@pytest.mark.asyncio
async def test_serial_repr():
"""Verify representation of a serial device."""
async with mock_pty_device({}) as mock_device:
async with RAVEnSerialDevice(mock_device) as dut:
assert 'RAVEnSerialDevice' in str(dut)
@pytest.mark.asyncio
async def test_serial_timeout_recovery():
"""Verify device recovery after a timeout."""
responses = {
b'<Command><Name>get_device_info</Name></Command>':
b'<DeviceInfo>'
b' <DeviceMacId>0x0123456789abcdef</DeviceMacId>'
b'</DeviceInfo>',
b'<Command><Name>get_meter_list</Name></Command>': b'',
}
async with mock_pty_device(responses) as mock_device:
async with RAVEnSerialDevice(mock_device) as dut:
with pytest.raises(asyncio.TimeoutError):
await asyncio.wait_for(dut.get_meter_list(), 0.05)
assert await dut.get_device_info()
@pytest.mark.asyncio
async def test_serial_abort():
"""Verify behavior of device abort."""
async with mock_pty_device({}) as mock_device:
async with RAVEnSerialDevice(mock_device) as dut:
await dut.abort()
await dut.abort()
|
