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