1#!/usr/bin/env python 2 3# This will create golden files in a directory passed to it. 4# A Test calls this internally to create the golden files 5# So it can process them (so we don't have to checkin the files). 6 7# Ensure msgpack-python and cbor are installed first, using: 8# sudo apt-get install python-dev 9# sudo apt-get install python-pip 10# pip install --user msgpack-python msgpack-rpc-python cbor 11 12import cbor, msgpack, msgpackrpc, sys, os, threading 13 14def get_test_data_list(): 15 # get list with all primitive types, and a combo type 16 l0 = [ 17 -8, 18 -1616, 19 -32323232, 20 -6464646464646464, 21 192, 22 1616, 23 32323232, 24 6464646464646464, 25 192, 26 -3232.0, 27 -6464646464.0, 28 3232.0, 29 6464646464.0, 30 False, 31 True, 32 None, 33 u"someday", 34 u"", 35 u"bytestring", 36 1328176922000002000, 37 -2206187877999998000, 38 270, 39 -2013855847999995777, 40 #-6795364578871345152, 41 ] 42 l1 = [ 43 { "true": True, 44 "false": False }, 45 { "true": "True", 46 "false": False, 47 "uint16(1616)": 1616 }, 48 { "list": [1616, 32323232, True, -3232.0, {"TRUE":True, "FALSE":False}, [True, False] ], 49 "int32":32323232, "bool": True, 50 "LONG STRING": "123456789012345678901234567890123456789012345678901234567890", 51 "SHORT STRING": "1234567890" }, 52 { True: "true", 8: False, "false": 0 } 53 ] 54 55 l = [] 56 l.extend(l0) 57 l.append(l0) 58 l.extend(l1) 59 return l 60 61def build_test_data(destdir): 62 l = get_test_data_list() 63 for i in range(len(l)): 64 # packer = msgpack.Packer() 65 serialized = msgpack.dumps(l[i]) 66 f = open(os.path.join(destdir, str(i) + '.msgpack.golden'), 'wb') 67 f.write(serialized) 68 f.close() 69 serialized = cbor.dumps(l[i]) 70 f = open(os.path.join(destdir, str(i) + '.cbor.golden'), 'wb') 71 f.write(serialized) 72 f.close() 73 74def doRpcServer(port, stopTimeSec): 75 class EchoHandler(object): 76 def Echo123(self, msg1, msg2, msg3): 77 return ("1:%s 2:%s 3:%s" % (msg1, msg2, msg3)) 78 def EchoStruct(self, msg): 79 return ("%s" % msg) 80 81 addr = msgpackrpc.Address('localhost', port) 82 server = msgpackrpc.Server(EchoHandler()) 83 server.listen(addr) 84 # run thread to stop it after stopTimeSec seconds if > 0 85 if stopTimeSec > 0: 86 def myStopRpcServer(): 87 server.stop() 88 t = threading.Timer(stopTimeSec, myStopRpcServer) 89 t.start() 90 server.start() 91 92def doRpcClientToPythonSvc(port): 93 address = msgpackrpc.Address('localhost', port) 94 client = msgpackrpc.Client(address, unpack_encoding='utf-8') 95 print client.call("Echo123", "A1", "B2", "C3") 96 print client.call("EchoStruct", {"A" :"Aa", "B":"Bb", "C":"Cc"}) 97 98def doRpcClientToGoSvc(port): 99 # print ">>>> port: ", port, " <<<<<" 100 address = msgpackrpc.Address('localhost', port) 101 client = msgpackrpc.Client(address, unpack_encoding='utf-8') 102 print client.call("TestRpcInt.Echo123", ["A1", "B2", "C3"]) 103 print client.call("TestRpcInt.EchoStruct", {"A" :"Aa", "B":"Bb", "C":"Cc"}) 104 105def doMain(args): 106 if len(args) == 2 and args[0] == "testdata": 107 build_test_data(args[1]) 108 elif len(args) == 3 and args[0] == "rpc-server": 109 doRpcServer(int(args[1]), int(args[2])) 110 elif len(args) == 2 and args[0] == "rpc-client-python-service": 111 doRpcClientToPythonSvc(int(args[1])) 112 elif len(args) == 2 and args[0] == "rpc-client-go-service": 113 doRpcClientToGoSvc(int(args[1])) 114 else: 115 print("Usage: test.py " + 116 "[testdata|rpc-server|rpc-client-python-service|rpc-client-go-service] ...") 117 118if __name__ == "__main__": 119 doMain(sys.argv[1:]) 120 121