mirror of
https://codeberg.org/forgejo/forgejo.git
synced 2024-11-23 08:47:42 -05:00
b6a95a8cb3
* Dropped unused codekit config * Integrated dynamic and static bindata for public * Ignore public bindata * Add a general generate make task * Integrated flexible public assets into web command * Updated vendoring, added all missiong govendor deps * Made the linter happy with the bindata and dynamic code * Moved public bindata definition to modules directory * Ignoring the new bindata path now * Updated to the new public modules import path * Updated public bindata command and drop the new prefix
126 lines
3.9 KiB
Python
Executable file
126 lines
3.9 KiB
Python
Executable file
#!/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)
|
|
|
|
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"someday",
|
|
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])
|
|
f = open(os.path.join(destdir, str(i) + '.msgpack.golden'), 'wb')
|
|
f.write(serialized)
|
|
f.close()
|
|
serialized = cbor.dumps(l[i])
|
|
f = open(os.path.join(destdir, str(i) + '.cbor.golden'), 'wb')
|
|
f.write(serialized)
|
|
f.close()
|
|
|
|
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('localhost', 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('localhost', 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('localhost', 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:])
|
|
|