test_websocket.py 文件源码

def test_websocket_wrapper_authentication_error():
    loop = asyncio.get_event_loop()

    @websocket("/", authenticate=True)
    def handler(ws, params, **kwargs):
        ws.send_str("hello, world!")

    @asyncio.coroutine
    def start_server(loop):
        app = Application()
        app.router.add_route(*handler.route)
        srv = yield from loop.create_server(app.make_handler(), "localhost", 9000)
        return srv

    @asyncio.coroutine
    def go(loop):
        srv = yield from start_server(loop)
        with pytest.raises(WSServerHandshakeError):
            client = yield from ws_connect("http://localhost:9000")
            yield from client.close()

        srv.close()
        yield from srv.wait_closed()

    loop.run_until_complete(go(loop))