def test_reads(circuit_pair, data_type, data_count, data, metadata):
cli_circuit, srv_circuit = circuit_pair
cli_channel, srv_channel = make_channels(*circuit_pair, data_type,
data_count)
req = ca.ReadNotifyRequest(data_count=data_count, data_type=data_type,
ioid=0, sid=0)
buffers_to_send = cli_circuit.send(req)
# Socket transport would happen here. Calling bytes() simulates
# serialization over the socket.
commands, _ = srv_circuit.recv(*(_np_hack(buf) for buf in buffers_to_send))
for command in commands:
srv_circuit.process_command(command)
res = ca.ReadNotifyResponse(data=data, metadata=metadata,
data_count=data_count, data_type=data_type,
ioid=0, status=1)
buffers_to_send = srv_circuit.send(res)
# Socket transport would happen here. Calling bytes() simulates
# serialization over the socket.
commands, _ = cli_circuit.recv(*(_np_hack(buf) for buf in buffers_to_send))
res_received, = commands
cli_circuit.process_command(res_received)
if isinstance(data, array.ArrayType):
# Before comparing array.array (which exposes the byteorder naively)
# with a numpy.ndarray (which buries the byteorder in dtype), flip
# the byte order to little-endian.
expected = copy.deepcopy(data)
expected.byteswap()
assert_array_almost_equal(res_received.data, expected)
elif isinstance(data, bytes):
assert data == _np_hack(res_received.data)
else:
try:
assert_array_equal(res_received.data, data) # for strings
except AssertionError:
assert_array_almost_equal(res_received.data, data) # for floats
评论列表
文章目录
