File: test.py

package info (click to toggle)
golang-github-ugorji-go-codec 1.1.7-2
  • links: PTS, VCS
  • area: main
  • in suites: bookworm, bullseye, sid
  • size: 2,136 kB
  • sloc: sh: 373; python: 97; makefile: 5
file content (124 lines) | stat: -rwxr-xr-x 4,057 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
#!/usr/bin/env python

# This will create golden files in a directory passed to it.
# A Test calls this internally to create the golden files
# So it can process them (so we don't have to checkin the files).

# Ensure msgpack-python and cbor are installed first, using:
#   sudo apt-get install python-dev
#   sudo apt-get install python-pip
#   pip install --user msgpack-python msgpack-rpc-python cbor

# Ensure all "string" keys are utf strings (else encoded as bytes)

from __future__ import print_function
import cbor, msgpack, msgpackrpc, sys, os, threading

def get_test_data_list():
    # get list with all primitive types, and a combo type
    l0 = [ 
        -8,
         -1616,
         -32323232,
         -6464646464646464,
         192,
         1616,
         32323232,
         6464646464646464,
         192,
         -3232.0,
         -6464646464.0,
         3232.0,
         6464.0,
         6464646464.0,
         False,
         True,
         u"null",
         None,
         u"some&day>some<day",
         1328176922000002000,
         u"",
         -2206187877999998000,
         u"bytestring",
         270,
         u"none",
        -2013855847999995777,
         #-6795364578871345152,
         ]
    l1 = [
        { "true": True,
          "false": False },
        { "true": u"True",
          "false": False,
          "uint16(1616)": 1616 },
        { "list": [1616, 32323232, True, -3232.0, {"TRUE":True, "FALSE":False}, [True, False] ],
          "int32":32323232, "bool": True, 
          "LONG STRING": u"123456789012345678901234567890123456789012345678901234567890",
          "SHORT STRING": u"1234567890" },
        { True: "true", 138: False, "false": 200 }
        ]
    
    l = []
    l.extend(l0)
    l.append(l0)
    l.append(1)
    l.extend(l1)
    return l

def build_test_data(destdir):
    l = get_test_data_list()
    for i in range(len(l)):
        # packer = msgpack.Packer()
        serialized = msgpack.dumps(l[i])
        with open(os.path.join(destdir, str(i) + '.msgpack.golden'), 'wb') as f:
            f.write(serialized)
        serialized = cbor.dumps(l[i])
        with open(os.path.join(destdir, str(i) + '.cbor.golden'), 'wb') as f:
            f.write(serialized)

def doRpcServer(port, stopTimeSec):
    class EchoHandler(object):
        def Echo123(self, msg1, msg2, msg3):
            return ("1:%s 2:%s 3:%s" % (msg1, msg2, msg3))
        def EchoStruct(self, msg):
            return ("%s" % msg)
    
    addr = msgpackrpc.Address('127.0.0.1', port)
    server = msgpackrpc.Server(EchoHandler())
    server.listen(addr)
    # run thread to stop it after stopTimeSec seconds if > 0
    if stopTimeSec > 0:
        def myStopRpcServer():
            server.stop()
        t = threading.Timer(stopTimeSec, myStopRpcServer)
        t.start()
    server.start()

def doRpcClientToPythonSvc(port):
    address = msgpackrpc.Address('127.0.0.1', port)
    client = msgpackrpc.Client(address, unpack_encoding='utf-8')
    print(client.call("Echo123", "A1", "B2", "C3"))
    print(client.call("EchoStruct", {"A" :"Aa", "B":"Bb", "C":"Cc"}))
   
def doRpcClientToGoSvc(port):
    # print(">>>> port: ", port, " <<<<<")
    address = msgpackrpc.Address('127.0.0.1', port)
    client = msgpackrpc.Client(address, unpack_encoding='utf-8')
    print(client.call("TestRpcInt.Echo123", ["A1", "B2", "C3"]))
    print(client.call("TestRpcInt.EchoStruct", {"A" :"Aa", "B":"Bb", "C":"Cc"}))

def doMain(args):
    if len(args) == 2 and args[0] == "testdata":
        build_test_data(args[1])
    elif len(args) == 3 and args[0] == "rpc-server":
        doRpcServer(int(args[1]), int(args[2]))
    elif len(args) == 2 and args[0] == "rpc-client-python-service":
        doRpcClientToPythonSvc(int(args[1]))
    elif len(args) == 2 and args[0] == "rpc-client-go-service":
        doRpcClientToGoSvc(int(args[1]))
    else:
        print("Usage: test.py " + 
              "[testdata|rpc-server|rpc-client-python-service|rpc-client-go-service] ...")
    
if __name__ == "__main__":
    doMain(sys.argv[1:])