289 lines
10 KiB
Python
289 lines
10 KiB
Python
#
|
|
# Licensed to the Apache Software Foundation (ASF) under one
|
|
# or more contributor license agreements. See the NOTICE file
|
|
# distributed with this work for additional information
|
|
# regarding copyright ownership. The ASF licenses this file
|
|
# to you under the Apache License, Version 2.0 (the
|
|
# "License"); you may not use this file except in compliance
|
|
# with the License. You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing,
|
|
# software distributed under the License is distributed on an
|
|
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
|
|
# KIND, either express or implied. See the License for the
|
|
# specific language governing permissions and limitations
|
|
# under the License.
|
|
#
|
|
|
|
import _import_local_thrift # noqa
|
|
from thrift.protocol import TCompactProtocol
|
|
from thrift.transport import TTransport
|
|
import unittest
|
|
|
|
CLEAR = 0
|
|
FIELD_WRITE = 1
|
|
VALUE_WRITE = 2
|
|
CONTAINER_WRITE = 3
|
|
BOOL_WRITE = 4
|
|
FIELD_READ = 5
|
|
CONTAINER_READ = 6
|
|
VALUE_READ = 7
|
|
BOOL_READ = 8
|
|
|
|
|
|
def testNaked(type, data):
|
|
buf = TTransport.TMemoryBuffer()
|
|
transport = TTransport.TBufferedTransportFactory().getTransport(buf)
|
|
protocol = TCompactProtocol.TCompactProtocol(transport)
|
|
|
|
if type.capitalize() == 'Byte':
|
|
protocol.state = VALUE_WRITE
|
|
protocol.writeByte(data)
|
|
|
|
elif type.capitalize() == 'I16':
|
|
protocol.state = CONTAINER_WRITE
|
|
protocol.writeI16(data)
|
|
|
|
elif type.capitalize() == 'I32':
|
|
protocol.state = CONTAINER_WRITE
|
|
protocol.writeI32(data)
|
|
|
|
elif type.capitalize() == 'I64':
|
|
protocol.state = CONTAINER_WRITE
|
|
protocol.writeI64(data)
|
|
|
|
elif type.capitalize() == 'String':
|
|
protocol.state = CONTAINER_WRITE
|
|
protocol.writeString(data)
|
|
|
|
elif type.capitalize() == 'Double':
|
|
protocol.state = VALUE_WRITE
|
|
protocol.writeDouble(data)
|
|
|
|
elif type.capitalize() == 'Binary':
|
|
protocol.state = FIELD_WRITE
|
|
protocol.writeBinary(data)
|
|
|
|
elif type.capitalize() == 'Bool':
|
|
protocol.state = CONTAINER_WRITE
|
|
protocol.writeBool(True)
|
|
|
|
transport.flush()
|
|
data_r = buf.getvalue()
|
|
buf = TTransport.TMemoryBuffer(data_r)
|
|
transport = TTransport.TBufferedTransportFactory().getTransport(buf)
|
|
protocol = TCompactProtocol.TCompactProtocol(transport)
|
|
if type.capitalize() == 'Byte':
|
|
protocol.state = VALUE_READ
|
|
return protocol.readByte()
|
|
|
|
elif type.capitalize() == 'I16':
|
|
protocol.state = CONTAINER_READ
|
|
return protocol.readI16()
|
|
|
|
elif type.capitalize() == 'I32':
|
|
protocol.state = CONTAINER_READ
|
|
return protocol.readI32()
|
|
|
|
elif type.capitalize() == 'I64':
|
|
protocol.state = CONTAINER_READ
|
|
return protocol.readI64()
|
|
|
|
elif type.capitalize() == 'String':
|
|
protocol.state = VALUE_READ
|
|
return protocol.readString()
|
|
|
|
elif type.capitalize() == 'Double':
|
|
protocol.state = VALUE_READ
|
|
return protocol.readDouble()
|
|
|
|
elif type.capitalize() == 'Binary':
|
|
protocol.state = FIELD_READ
|
|
return protocol.readBinary()
|
|
|
|
elif type.capitalize() == 'Bool':
|
|
protocol.state = CONTAINER_READ
|
|
return protocol.readBool()
|
|
|
|
|
|
def testField(type, data):
|
|
TType = {"Bool": 2, "Byte": 3, "Binary": 5, "I16": 6, "I32": 8, "I64": 10, "Double": 11, "String": 12}
|
|
buf = TTransport.TMemoryBuffer()
|
|
transport = TTransport.TBufferedTransportFactory().getTransport(buf)
|
|
protocol = TCompactProtocol.TCompactProtocol(transport)
|
|
protocol.writeStructBegin('struct')
|
|
protocol.writeFieldBegin("field", TType[type.capitalize()], 10)
|
|
if type.capitalize() == 'Byte':
|
|
protocol.writeByte(data)
|
|
|
|
elif type.capitalize() == 'I16':
|
|
protocol.writeI16(data)
|
|
|
|
elif type.capitalize() == 'I32':
|
|
protocol.writeI32(data)
|
|
|
|
elif type.capitalize() == 'I64':
|
|
protocol.writeI64(data)
|
|
|
|
elif type.capitalize() == 'String':
|
|
protocol.writeString(data)
|
|
|
|
elif type.capitalize() == 'Double':
|
|
protocol.writeDouble(data)
|
|
|
|
elif type.capitalize() == 'Binary':
|
|
protocol.writeBinary(data)
|
|
|
|
elif type.capitalize() == 'Bool':
|
|
protocol.writeBool(data)
|
|
|
|
protocol.writeFieldEnd()
|
|
protocol.writeStructEnd()
|
|
|
|
transport.flush()
|
|
data_r = buf.getvalue()
|
|
|
|
buf = TTransport.TMemoryBuffer(data_r)
|
|
transport = TTransport.TBufferedTransportFactory().getTransport(buf)
|
|
protocol = TCompactProtocol.TCompactProtocol(transport)
|
|
protocol.readStructBegin()
|
|
protocol.readFieldBegin()
|
|
if type.capitalize() == 'Byte':
|
|
return protocol.readByte()
|
|
|
|
elif type.capitalize() == 'I16':
|
|
return protocol.readI16()
|
|
|
|
elif type.capitalize() == 'I32':
|
|
return protocol.readI32()
|
|
|
|
elif type.capitalize() == 'I64':
|
|
return protocol.readI32()
|
|
|
|
elif type.capitalize() == 'String':
|
|
return protocol.readString()
|
|
|
|
elif type.capitalize() == 'Double':
|
|
return protocol.readDouble()
|
|
|
|
elif type.capitalize() == 'Binary':
|
|
return protocol.readBinary()
|
|
|
|
elif type.capitalize() == 'Bool':
|
|
return protocol.readBool()
|
|
|
|
protocol.readFieldEnd()
|
|
protocol.readStructEnd()
|
|
|
|
|
|
def testMessage(data):
|
|
message = {}
|
|
message['name'] = data[0]
|
|
message['type'] = data[1]
|
|
message['seqid'] = data[2]
|
|
|
|
buf = TTransport.TMemoryBuffer()
|
|
transport = TTransport.TBufferedTransportFactory().getTransport(buf)
|
|
protocol = TCompactProtocol.TCompactProtocol(transport)
|
|
protocol.writeMessageBegin(message['name'], message['type'], message['seqid'])
|
|
protocol.writeMessageEnd()
|
|
|
|
transport.flush()
|
|
data_r = buf.getvalue()
|
|
|
|
buf = TTransport.TMemoryBuffer(data_r)
|
|
transport = TTransport.TBufferedTransportFactory().getTransport(buf)
|
|
protocol = TCompactProtocol.TCompactProtocol(transport)
|
|
result = protocol.readMessageBegin()
|
|
protocol.readMessageEnd()
|
|
return result
|
|
|
|
|
|
class TestTCompactProtocol(unittest.TestCase):
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
unittest.TestCase.__init__(self, *args, **kwargs)
|
|
|
|
def test_TCompactProtocol_write_read(self):
|
|
try:
|
|
testNaked('Byte', 123)
|
|
for i in range(0, 128):
|
|
self.assertEqual(i, testField('Byte', i))
|
|
self.assertEqual(-i, testField('Byte', -i))
|
|
|
|
self.assertEqual(0, testNaked("I16", 0))
|
|
self.assertEqual(1, testNaked("I16", 1))
|
|
self.assertEqual(15000, testNaked("I16", 15000))
|
|
self.assertEqual(0x7fff, testNaked('I16', 0x7fff))
|
|
self.assertEqual(-1, testNaked('I16', -1))
|
|
self.assertEqual(-15000, testNaked('I16', -15000))
|
|
self.assertEqual(-0x7fff, testNaked('I16', -0x7fff))
|
|
self.assertEqual(32767, testNaked('I16', 32767))
|
|
|
|
self.assertEqual(0, testField('I16', 0))
|
|
self.assertEqual(1, testField('I16', 1))
|
|
self.assertEqual(7, testField('I16', 7))
|
|
self.assertEqual(150, testField('I16', 150))
|
|
self.assertEqual(15000, testField('I16', 15000))
|
|
self.assertEqual(0x7fff, testField('I16', 0x7fff))
|
|
self.assertEqual(-1, testField('I16', -1))
|
|
self.assertEqual(-7, testField('I16', -7))
|
|
self.assertEqual(-150, testField('I16', -150))
|
|
self.assertEqual(-15000, testField('I16', -15000))
|
|
self.assertEqual(-0xfff, testField('I16', -0xfff))
|
|
|
|
self.assertEqual(0, testNaked('I32', 0))
|
|
self.assertEqual(1, testNaked('I32', 1))
|
|
self.assertEqual(15000, testNaked('I32', 15000))
|
|
self.assertEqual(0xfff, testNaked('I32', 0xfff))
|
|
self.assertEqual(-1, testNaked('I32', -1))
|
|
self.assertEqual(-15000, testNaked('I32', -15000))
|
|
self.assertEqual(-0xfff, testNaked('I32', -0xfff))
|
|
self.assertEqual(2147483647, testNaked('I32', 2147483647))
|
|
self.assertEqual(-2147483647, testNaked('I32', -2147483647))
|
|
|
|
self.assertEqual(0, testField('I32', 0))
|
|
self.assertEqual(1, testField('I32', 1))
|
|
self.assertEqual(7, testField('I32', 7))
|
|
self.assertEqual(150, testField('I32', 150))
|
|
self.assertEqual(15000, testField('I32', 15000))
|
|
self.assertEqual(31337, testField('I32', 31337))
|
|
self.assertEqual(0xffff, testField('I32', 0xffff))
|
|
self.assertEqual(0xffffff, testField('I32', 0xffffff))
|
|
self.assertEqual(-1, testField('I32', -1))
|
|
self.assertEqual(-7, testField('I32', -7))
|
|
self.assertEqual(-150, testField('I32', -150))
|
|
self.assertEqual(-15000, testField('I32', -15000))
|
|
self.assertEqual(-0xffff, testField('I32', -0xffff))
|
|
self.assertEqual(-0xffffff, testField('I32', -0xffffff))
|
|
|
|
self.assertEqual(9223372036854775807, testNaked("I64", 9223372036854775807))
|
|
self.assertEqual(-9223372036854775807, testNaked('I64', -9223372036854775807))
|
|
self.assertEqual(-0, testNaked('I64', 0))
|
|
self.assertEqual(True, testNaked('Bool', True))
|
|
self.assertEqual(3.14159261, testNaked('Double', 3.14159261))
|
|
self.assertEqual("hello thrift", testNaked('String', "hello thrift"))
|
|
self.assertEqual(True, testField('Bool', True))
|
|
self.assertEqual(3.14159261, testField('Double', 3.14159261))
|
|
self.assertEqual("hello thrift", testField('String', "hello thrift"))
|
|
TMessage = {"T_CALL": 1, "T_REPLY": 2, "T_EXCEPTION": 3, "T_ONEWAY": 4}
|
|
test_data = [("short message name", TMessage["T_CALL"], 0),
|
|
("1", TMessage["T_REPLY"], 12345),
|
|
("loooooooooooooooooooong", TMessage["T_EXCEPTION"], 1 << 16),
|
|
("one way push", TMessage["T_ONEWAY"], 12),
|
|
("JANKY", TMessage["T_CALL"], 0)]
|
|
for dt in test_data:
|
|
result = testMessage(dt)
|
|
self.assertEqual(result[0], dt[0])
|
|
self.assertEqual(result[1], dt[1])
|
|
self.assertEqual(result[2], dt[2])
|
|
except Exception as e:
|
|
print("Assertion fail")
|
|
raise e
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|