test_util.py 文件源码

def run_radio_station_as_process_and_stub(port):
    process = run_radio_station_as_process(port)
    channel = grpc.insecure_channel('localhost:' + str(port))
    stub = loopchain_pb2_grpc.RadioStationStub(channel)
    util.request_server_in_time(stub.GetStatus, loopchain_pb2.StatusRequest(request=""))
    return process, stub