test.py 3.9 KB

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