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
from aioraven.data import MeterList
from aioraven.device import RAVEnConnectionError
from aioraven.device import RAVEnNotOpenError
from aioraven.streams import open_connection
from aioraven.streams import RAVEnNetworkDevice
import pytest
from .mock_device import mock_device
@pytest.mark.asyncio
async def test_tcp_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_device(responses) as (host, port):
async with RAVEnNetworkDevice(host, port) 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_tcp_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_device(responses) as (host, port):
dut = RAVEnNetworkDevice(host, port)
await dut.open()
assert await dut.get_meter_list()
async with dut:
assert not await dut.get_meter_list()
assert not await dut.get_meter_list()
@pytest.mark.asyncio
async def test_tcp_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_device(responses) as (host, port):
dut = RAVEnNetworkDevice(host, port)
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):
await task
assert not await dut.get_meter_list()
@pytest.mark.asyncio
async def test_tcp_not_open():
"""Verify behavior when reading from an unopened device."""
async with mock_device() as (host, port):
dut = RAVEnNetworkDevice(host, port)
with pytest.raises(RAVEnNotOpenError):
await dut.get_device_info()
@pytest.mark.asyncio
async def test_tcp_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_device(responses) as (host, port):
async with RAVEnNetworkDevice(host, port) 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_tcp_repr():
"""Verify representation of a network device."""
async with mock_device() as (host, port):
async with RAVEnNetworkDevice(host, port) as dut:
assert 'RAVEnNetworkDevice' in str(dut)
reader, writer = await open_connection(host, port)
try:
assert 'RAVEnReader' in str(reader)
assert 'RAVEnWriter' in str(writer)
finally:
writer.close()
@pytest.mark.asyncio
async def test_tcp_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_device(responses) as (host, port):
async with RAVEnNetworkDevice(host, port) 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_tcp_abort():
"""Verify behavior of device abort."""
async with mock_device({}) as (host, port):
async with RAVEnNetworkDevice(host, port) as dut:
await dut.abort()
await dut.abort()
|
