test_serialization.py 文件源码

def test_writes(circuit_pair, data_type, data_count, data, metadata):

    cli_circuit, srv_circuit = circuit_pair
    cli_channel, srv_channel = make_channels(*circuit_pair, 5, 1)

    req = ca.WriteNotifyRequest(data=data, metadata=metadata,
                                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)
    req_received, = commands
    res = ca.WriteNotifyResponse(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))
    for command in commands:
        cli_circuit.process_command(command)

    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(req_received.data, expected)
    elif isinstance(data, bytes):
        assert data == _np_hack(req_received.data)
    else:
        try:
            assert_array_equal(req_received.data, data)  # for strings
        except AssertionError:
            assert_array_almost_equal(req_received.data, data)  # for floats