open-vault/vendor/github.com/ugorji/go/codec/test.py

127 lines
3.9 KiB
Python
Raw Normal View History

2015-06-29 21:50:55 +00:00
#!/usr/bin/env python
# This will create golden files in a directory passed to it.
# A Test calls this internally to create the golden files
# So it can process them (so we don't have to checkin the files).
# Ensure msgpack-python and cbor are installed first, using:
2015-09-22 15:44:21 +00:00
# sudo apt-get install python-dev
# sudo apt-get install python-pip
# pip install --user msgpack-python msgpack-rpc-python cbor
2015-06-29 21:50:55 +00:00
2016-04-26 00:18:04 +00:00
# Ensure all "string" keys are utf strings (else encoded as bytes)
2015-06-29 21:50:55 +00:00
import cbor, msgpack, msgpackrpc, sys, os, threading
def get_test_data_list():
# get list with all primitive types, and a combo type
l0 = [
-8,
-1616,
-32323232,
-6464646464646464,
192,
1616,
32323232,
6464646464646464,
192,
-3232.0,
-6464646464.0,
3232.0,
2016-04-26 00:18:04 +00:00
6464.0,
2015-06-29 21:50:55 +00:00
6464646464.0,
False,
True,
2016-04-26 00:18:04 +00:00
u"null",
2015-06-29 21:50:55 +00:00
None,
u"some&day>some<day",
2015-06-29 21:50:55 +00:00
1328176922000002000,
2016-04-26 00:18:04 +00:00
u"",
2015-06-29 21:50:55 +00:00
-2206187877999998000,
2016-04-26 00:18:04 +00:00
u"bytestring",
2015-06-29 21:50:55 +00:00
270,
2016-04-26 00:18:04 +00:00
u"none",
2015-06-29 21:50:55 +00:00
-2013855847999995777,
#-6795364578871345152,
]
l1 = [
{ "true": True,
"false": False },
2016-04-26 00:18:04 +00:00
{ "true": u"True",
2015-06-29 21:50:55 +00:00
"false": False,
"uint16(1616)": 1616 },
{ "list": [1616, 32323232, True, -3232.0, {"TRUE":True, "FALSE":False}, [True, False] ],
"int32":32323232, "bool": True,
2016-04-26 00:18:04 +00:00
"LONG STRING": u"123456789012345678901234567890123456789012345678901234567890",
"SHORT STRING": u"1234567890" },
{ True: "true", 138: False, "false": 200 }
2015-06-29 21:50:55 +00:00
]
l = []
l.extend(l0)
l.append(l0)
2016-04-26 00:18:04 +00:00
l.append(1)
2015-06-29 21:50:55 +00:00
l.extend(l1)
return l
def build_test_data(destdir):
l = get_test_data_list()
for i in range(len(l)):
# packer = msgpack.Packer()
serialized = msgpack.dumps(l[i])
f = open(os.path.join(destdir, str(i) + '.msgpack.golden'), 'wb')
f.write(serialized)
f.close()
serialized = cbor.dumps(l[i])
f = open(os.path.join(destdir, str(i) + '.cbor.golden'), 'wb')
f.write(serialized)
f.close()
def doRpcServer(port, stopTimeSec):
class EchoHandler(object):
def Echo123(self, msg1, msg2, msg3):
return ("1:%s 2:%s 3:%s" % (msg1, msg2, msg3))
def EchoStruct(self, msg):
return ("%s" % msg)
2017-09-05 22:06:47 +00:00
addr = msgpackrpc.Address('127.0.0.1', port)
2015-06-29 21:50:55 +00:00
server = msgpackrpc.Server(EchoHandler())
server.listen(addr)
# run thread to stop it after stopTimeSec seconds if > 0
if stopTimeSec > 0:
def myStopRpcServer():
server.stop()
t = threading.Timer(stopTimeSec, myStopRpcServer)
t.start()
server.start()
def doRpcClientToPythonSvc(port):
2017-09-05 22:06:47 +00:00
address = msgpackrpc.Address('127.0.0.1', port)
2015-06-29 21:50:55 +00:00
client = msgpackrpc.Client(address, unpack_encoding='utf-8')
print client.call("Echo123", "A1", "B2", "C3")
print client.call("EchoStruct", {"A" :"Aa", "B":"Bb", "C":"Cc"})
def doRpcClientToGoSvc(port):
# print ">>>> port: ", port, " <<<<<"
2017-09-05 22:06:47 +00:00
address = msgpackrpc.Address('127.0.0.1', port)
2015-06-29 21:50:55 +00:00
client = msgpackrpc.Client(address, unpack_encoding='utf-8')
print client.call("TestRpcInt.Echo123", ["A1", "B2", "C3"])
print client.call("TestRpcInt.EchoStruct", {"A" :"Aa", "B":"Bb", "C":"Cc"})
def doMain(args):
if len(args) == 2 and args[0] == "testdata":
build_test_data(args[1])
elif len(args) == 3 and args[0] == "rpc-server":
doRpcServer(int(args[1]), int(args[2]))
elif len(args) == 2 and args[0] == "rpc-client-python-service":
doRpcClientToPythonSvc(int(args[1]))
elif len(args) == 2 and args[0] == "rpc-client-go-service":
doRpcClientToGoSvc(int(args[1]))
else:
print("Usage: test.py " +
"[testdata|rpc-server|rpc-client-python-service|rpc-client-go-service] ...")
if __name__ == "__main__":
doMain(sys.argv[1:])