test_processor.py 文件源码

def processor_client():
    """Returns a client (grpc) fixture that is passed into processor
    tests """
    sys.stdout = ThreadPrinter()
    sys.argv = ["", '{}']
    proc = MockProcessor("MyProcessor", 1)
    proc.start()
    t_end = time.time() + 5
    # wait for our collector to print its preamble
    while len(sys.stdout.lines) == 0 and time.time() < t_end:
        time.sleep(.1)
    resp = json.loads(sys.stdout.lines[0])
    client = ProcessorStub(
        grpc.insecure_channel(resp["ListenAddress"]))
    yield client
    proc.stop()