test_util.py 文件源码

def run_peer_server_as_process_and_stub(port, radiostation_port=conf.PORT_RADIOSTATION, group_id=None, score=None):
    process = run_peer_server_as_process(port, radiostation_port, group_id, score)
    channel = grpc.insecure_channel('localhost:' + str(port))
    stub = loopchain_pb2_grpc.PeerServiceStub(channel)
    util.request_server_in_time(stub.GetStatus, loopchain_pb2.StatusRequest(request=""))
    return process, stub