test_supervisor.py 文件源码

def supervisor():
    """A running supervisor with two connected status clients."""

    info = {
        'expected_services':
        [
            {
                "short_name": "service1",
                "long_name": "Service 1"
            },

            {
                "short_name": "service2",
                "long_name": "Service 2"
            }
        ],
        'port': 'unused'  # Bind an unused port for testing, the value will appear on visor.port after visor.loaded is set
    }

    visor = IOTileSupervisor(info)

    visor.start()
    signaled = visor.loaded.wait(2.0)
    if not signaled:
        raise ValueError("Could not start supervisor service")

    port = visor.port
    client1 = ServiceStatusClient('ws://127.0.0.1:%d/services' % port)

    yield visor, client1

    client1.stop()
    visor.stop()