test_asyncio_kernel.py 文件源码

def test_stream_pty_raises_error_with_abnormal_status(mocker):
    mock_req_obj = asynctest.MagicMock(spec=Request)
    mock_exception = aiohttp.ClientResponseError(
        None, None, code=400,
        message='emulated-handshake-error')
    mock_req_obj.connect_websocket = \
        asynctest.MagicMock(side_effect=mock_exception)
    with asynctest.patch('ai.backend.client.asyncio.kernel.Request',
                         return_value=mock_req_obj) as mock_req_cls:
        with pytest.raises(BackendClientError):
            await Kernel('mykernel').stream_pty()