vendor/github.com/ugorji/go/codec/test.py

   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
  14 import cbor, msgpack, msgpackrpc, sys, os, threading
  15
  16 def 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
  67 def 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
  80 def 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
  98 def 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
 104 def 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
 111 def 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
 124 if __name__ == "__main__":
 125     doMain(sys.argv[1:])
 126