test_subscription.py 文件源码

def test_bad_subscription(io_loop, mocker):
    framework_info = {}

    handler = mocker.Mock()
    handlers = {Event.SUBSCRIBED: handler,
                Event.HEARTBEAT: handler,
                Event.OFFERS: handler,
                Event.SHUTDOWN: handler}

    sub = Subscription(framework_info, 'zk://localhost:2181',
                       '/api/v1/scheduler', handlers, timeout=1, loop=io_loop)

    assert sub.state.current_state == States.CLOSED
    yield sub.start()
    yield gen.sleep(5)
    assert sub.state.current_state == States.CLOSED