test_interoperability.py 文件源码

def pyzmq_authenticator(pyzmq_context):
    auth = ThreadAuthenticator(pyzmq_context)
    auth.start()
    auth.allow('127.0.0.1')
    auth.configure_curve(domain='*', location=zmq.auth.CURVE_ALLOW_ANY)
    yield
    auth.stop()