test_util.py 文件源码

def run_black_peer_server_as_process_and_stub(port, radiostation_port=conf.PORT_RADIOSTATION, group_id=None):
    process = run_black_peer_server_as_process(port, radiostation_port, group_id)
    channel = grpc.insecure_channel('localhost:' + str(port))
    stub = loopchain_pb2_grpc.PeerServiceStub(channel)
    return process, stub