gtsocial-umbx

Unnamed repository; edit this file 'description' to name the repository.
Log | Files | Refs | README | LICENSE

test.py (4544B)

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