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
202
203
204
|
import asyncio
import unittest
from siobrultech_protocols.gem.const import CMD_DELAY_NEXT_PACKET
from siobrultech_protocols.gem.protocol import (
ApiCall,
ApiType,
BidirectionalProtocol,
ConnectionLostMessage,
ConnectionMadeMessage,
PacketProtocolMessage,
PacketReceivedMessage,
ProtocolStateException,
)
from tests.gem.mock_transport import MockTransport
from tests.gem.packet_test_data import assert_packet, read_packet
TestCall = ApiCall[str, str](
gem_formatter=lambda x: x,
gem_parser=lambda x: x if x.endswith("\n") else None,
ecm_formatter=lambda x: [
(x + "1").encode(),
(x + "2").encode(),
(x + "3").encode(),
],
ecm_parser=lambda x: x.decode() if x.endswith(b"\n") else None,
)
class TestBidirectionalProtocol(unittest.IsolatedAsyncioTestCase):
def setUp(self):
self._queue: asyncio.Queue[PacketProtocolMessage] = asyncio.Queue()
self._transport = MockTransport()
self._protocol = BidirectionalProtocol(self._queue, api_type=ApiType.GEM)
self._protocol.connection_made(self._transport)
self._result: asyncio.Future[str] = asyncio.get_event_loop().create_future()
message = self._queue.get_nowait()
assert isinstance(message, ConnectionMadeMessage)
assert message.protocol is self._protocol
def tearDown(self) -> None:
if not self._transport.closed:
exc = Exception("Test")
self._protocol.connection_lost(exc=exc)
message = self._queue.get_nowait()
assert isinstance(message, ConnectionLostMessage)
assert message.protocol is self._protocol
assert message.exc is exc
self._protocol.close() # Close after connection_lost is not required, but at least should not crash
def testClose(self):
self._protocol.close()
assert self._transport.closed
def testBeginApi(self):
self._protocol.begin_api_request()
self.assertEqual(self._transport.writes, [CMD_DELAY_NEXT_PACKET.encode()])
def testBeginApiWithoutDelay(self):
self._protocol.send_packet_delay = False
self._protocol.begin_api_request()
self.assertEqual(self._transport.writes, [])
def testSendWithoutBeginFails(self):
with self.assertRaises(ProtocolStateException):
self._protocol.invoke_api(TestCall, "request", self._result)
def testSendRequest(self):
self._protocol.begin_api_request()
self._transport.writes.clear()
self._protocol.invoke_api(TestCall, "request", self._result)
self.assertEqual(self._transport.writes, ["request".encode()])
async def testEcmApiCall(self):
self._protocol.api_type = ApiType.ECM
self._protocol.begin_api_request()
self._protocol.invoke_api(TestCall, "request", self._result)
self._protocol.data_received(b"\xfc")
self._protocol.data_received(b"\xfc")
self._protocol.data_received(b"\xfcRESPONSE\n")
response = await self.get_response()
self.assertEqual(
self._transport.writes, [b"request1", b"request2", b"request3", b"\xfc"]
)
self.assertEqual(response, "RESPONSE\n")
async def testFailureDuringEcmApiCallDoesNotPreventNextCall(self):
self._protocol.api_type = ApiType.ECM
self._protocol.begin_api_request()
self._protocol.invoke_api(TestCall, "request", self._result)
self._protocol.data_received(b"\xfc")
self._protocol.data_received(b"X")
with self.assertRaises(Exception):
await self.get_response()
self._protocol.end_api_request()
self.assertEqual(self._transport.writes, [b"request1", b"request2"])
self._transport.writes.clear()
self._result = asyncio.get_event_loop().create_future()
self._protocol.begin_api_request()
self._protocol.invoke_api(TestCall, "request2", self._result)
self._protocol.data_received(b"\xfc")
self._protocol.data_received(b"\xfc")
self._protocol.data_received(b"\xfcRESPONSE\n")
response = await self.get_response()
self._protocol.end_api_request()
self.assertEqual(
self._transport.writes, [b"request21", b"request22", b"request23", b"\xfc"]
)
self.assertEqual(response, "RESPONSE\n")
async def testPacketRacingWithApi(self):
"""Tests that the protocol can handle a packet coming in right after it has
requested a packet delay from the GEM."""
self._protocol.begin_api_request()
self._protocol.data_received(read_packet("BIN32-ABS.bin"))
self._protocol.invoke_api(TestCall, "REQUEST", self._result)
self._protocol.data_received(b"RESPONSE\n")
response = await self.get_response()
self._protocol.end_api_request()
self.assertEqual(response, "RESPONSE\n")
self.assertPacket("BIN32-ABS.bin")
async def testPacketInterleavingWithApi(self):
"""Tests that the protocol can handle a packet coming in in the middle of the API response.
(I don't know whether this can happen in practice.)"""
self._protocol.begin_api_request()
self._protocol.data_received(read_packet("BIN32-ABS.bin"))
self._protocol.invoke_api(TestCall, "REQUEST", self._result)
self._protocol.data_received(b"RES")
self._protocol.data_received(read_packet("BIN32-ABS.bin"))
self._protocol.data_received(b"PONSE\n")
response = await self.get_response()
self._protocol.end_api_request()
self.assertEqual(response, "RESPONSE\n")
self.assertPacket("BIN32-ABS.bin")
self.assertPacket("BIN32-ABS.bin")
def testDeviceIgnoresApi(self):
"""Tests that the protocol fails appropriately if a device ignores API calls and just keeps sending packets."""
self._protocol.begin_api_request()
self._protocol.data_received(read_packet("BIN32-ABS.bin"))
self._protocol.invoke_api(TestCall, "REQUEST", self._result)
self._protocol.data_received(read_packet("BIN32-ABS.bin"))
assert not self._result.done()
self._protocol.end_api_request()
self.assertPacket("BIN32-ABS.bin")
self.assertPacket("BIN32-ABS.bin")
async def testApiCallWithPacketInProgress(self):
"""Tests that the protocol can handle a packet that's partially arrived when it
requested a packet delay from the GEM."""
packet = read_packet("BIN32-ABS.bin")
bytes_sent_before_packet_delay_command = 32
self._protocol.data_received(packet[0:bytes_sent_before_packet_delay_command])
self._protocol.begin_api_request()
self._protocol.data_received(packet[bytes_sent_before_packet_delay_command:])
self._protocol.invoke_api(TestCall, "REQUEST", self._result)
self._protocol.data_received(b"RESPONSE\n")
response = await self.get_response()
self._protocol.end_api_request()
self.assertEqual(response, "RESPONSE\n")
self.assertPacket("BIN32-ABS.bin")
async def testApiCallToIdleGem(self):
"""Tests that the protocol can handle no packets arriving after it has
requested a packet delay from the GEM."""
self._protocol.begin_api_request()
self._protocol.invoke_api(TestCall, "REQUEST", self._result)
self._protocol.data_received(b"RESPONSE\n")
response = await self.get_response()
self._protocol.end_api_request()
self.assertEqual(response, "RESPONSE\n")
self.assertNoPacket()
def testEndAfterBegin(self):
"""Checks for the case where user-code may fail, and we just call end_api_request
after calling begin_api_request."""
self._protocol.begin_api_request()
self._protocol.end_api_request()
async def get_response(self) -> str:
return await asyncio.wait_for(self._result, 0)
def assertNoPacket(self):
self.assertTrue(self._queue.empty())
def assertPacket(self, expected_packet: str):
message = self._queue.get_nowait()
assert isinstance(message, PacketReceivedMessage)
assert message.protocol is self._protocol
assert_packet(expected_packet, message.packet)
if __name__ == "__main__":
unittest.main()
|
