Merge pull request #22077 from bfredl/neolua_client

refactor(tests): integrate lua-client into core and use core for functionaltests

15 files changed, 525 insertions, 55 deletions

diff --git a/cmake.deps/cmake/BuildLuarocks.cmake b/cmake.deps/cmake/BuildLuarocks.cmake
index b84ce34d45..e50f50798a 100644
--- a/cmake.deps/cmake/BuildLuarocks.cmake
+++ b/cmake.deps/cmake/BuildLuarocks.cmake
@@ -158,23 +158,14 @@ if(USE_BUNDLED_BUSTED)
     DEPENDS busted)
   add_custom_target(luacheck DEPENDS ${LUACHECK_EXE})
 
-  # luv
-  set(LUV_DEPS luacheck)
-  if(USE_BUNDLED_LUV)
-    set(NVIM_CLIENT_DEPS luacheck luv-static lua-compat-5.3)
-  else()
-    add_custom_command(OUTPUT ${ROCKS_DIR}/luv
-      COMMAND ${LUAROCKS_BINARY} build luv ${LUV_VERSION} ${LUAROCKS_BUILDARGS}
-      DEPENDS luacheck)
-    add_custom_target(luv DEPENDS ${ROCKS_DIR}/luv)
-    set(NVIM_CLIENT_DEPS luv)
+  if (NOT USE_BUNDLED_LUAJIT)
+    # coxpcall
+    add_custom_command(OUTPUT ${ROCKS_DIR}/coxpcall
+      COMMAND ${LUAROCKS_BINARY} build coxpcall 1.16.0-1 ${LUAROCKS_BUILDARGS}
+      DEPENDS luarocks)
+    add_custom_target(coxpcall DEPENDS ${ROCKS_DIR}/coxpcall)
+  list(APPEND THIRD_PARTY_DEPS coxpcall)
   endif()
 
-  # nvim-client: https://github.com/neovim/lua-client
-  add_custom_command(OUTPUT ${ROCKS_DIR}/nvim-client
-    COMMAND ${LUAROCKS_BINARY} build nvim-client 0.2.4-1 ${LUAROCKS_BUILDARGS}
-    DEPENDS ${NVIM_CLIENT_DEPS})
-  add_custom_target(nvim-client DEPENDS ${ROCKS_DIR}/nvim-client)
-
-  list(APPEND THIRD_PARTY_DEPS busted luacheck nvim-client)
+  list(APPEND THIRD_PARTY_DEPS busted luacheck)
 endif()
diff --git a/cmake/RunTests.cmake b/cmake/RunTests.cmake
index d724f43a5f..3ac15f91ea 100644
--- a/cmake/RunTests.cmake
+++ b/cmake/RunTests.cmake
@@ -71,16 +71,10 @@ if(NOT DEFINED ENV{TEST_TIMEOUT} OR "$ENV{TEST_TIMEOUT}" STREQUAL "")
 endif()
 
 set(ENV{SYSTEM_NAME} ${CMAKE_HOST_SYSTEM_NAME})  # used by test/helpers.lua.
-
-# TODO: eventually always use NVIM_PRG as the runner
-if("${TEST_TYPE}" STREQUAL "unit")
-  set(RUNNER_PRG ${NVIM_PRG} -ll ${WORKING_DIR}/test/busted_runner.lua)
-else()
-  set(RUNNER_PRG ${BUSTED_PRG})
-endif()
+set(ENV{DEPS_PREFIX} ${DEPS_PREFIX})  # used by test/busted_runner.lua on windows
 
 execute_process(
-  COMMAND ${RUNNER_PRG} -v -o test.busted.outputHandlers.${BUSTED_OUTPUT_TYPE}
+  COMMAND ${NVIM_PRG} -ll ${WORKING_DIR}/test/busted_runner.lua -v -o test.busted.outputHandlers.${BUSTED_OUTPUT_TYPE}
     --lazy --helper=${TEST_DIR}/${TEST_TYPE}/preload.lua
     --lpath=${BUILD_DIR}/?.lua
     --lpath=${WORKING_DIR}/runtime/lua/?.lua
diff --git a/src/mpack/lmpack.c b/src/mpack/lmpack.c
index 53d7092a0c..957bac37cc 100644
--- a/src/mpack/lmpack.c
+++ b/src/mpack/lmpack.c
@@ -644,7 +644,13 @@ static void lmpack_unparse_enter(mpack_parser_t *parser, mpack_node_t *node)
       mpack_node_t *n;
 
       int has_meta = lua_getmetatable(L, -1);
-      if (packer->ext != LUA_NOREF && has_meta) {
+      int has_mtdict = false;
+      if (has_meta && packer->mtdict != LUA_NOREF) {
+          lmpack_geti(L, packer->reg, packer->mtdict); // [table, metatable, mtdict]
+          has_mtdict = lua_rawequal(L, -1, -2);
+          lua_pop(L, 1); // [table, metatable];
+      }
+      if (packer->ext != LUA_NOREF && has_meta && !has_mtdict) {
         /* check if there's a handler for this metatable */
         lmpack_geti(L, packer->reg, packer->ext);
         lua_pushvalue(L, -2);
@@ -701,14 +707,7 @@ static void lmpack_unparse_enter(mpack_parser_t *parser, mpack_node_t *node)
         }
       }
 
-      int is_array = 1;
       if (has_meta) {
-        // stack: [table, metatable]
-        if (packer->mtdict != LUA_NOREF) {
-          lmpack_geti(L, packer->reg, packer->mtdict); // [table, metatable, mtdict]
-          is_array = !lua_rawequal(L, -1, -2);
-          lua_pop(L, 1); // [table, metatable];
-        }
         lua_pop(L, 1); // [table]
       }
 
@@ -726,6 +725,7 @@ static void lmpack_unparse_enter(mpack_parser_t *parser, mpack_node_t *node)
         lua_pop(L, 1);
       }
 
+      int is_array = !has_mtdict;
       len = lmpack_objlen(L, &is_array);
       if (is_array) {
         node->tok = mpack_pack_array(len);
diff --git a/src/mpack/object.c b/src/mpack/object.c
index e2d893bc88..60f49f73aa 100644
--- a/src/mpack/object.c
+++ b/src/mpack/object.c
@@ -128,8 +128,11 @@ MPACK_API int mpack_unparse(mpack_parser_t *parser, char **buf, size_t *buflen,
   return status;
 }
 
-MPACK_API void mpack_parser_copy(mpack_parser_t *dst, mpack_parser_t *src)
+MPACK_API void mpack_parser_copy(mpack_parser_t *d, mpack_parser_t *s)
 {
+  // workaround UBSAN being NOT happy with a flexible array member with arr[N>1] initial size
+  mpack_one_parser_t *dst = (mpack_one_parser_t *)d;
+  mpack_one_parser_t *src = (mpack_one_parser_t *)s;
   mpack_uint32_t i;
   mpack_uint32_t dst_capacity = dst->capacity; 
   assert(src->capacity <= dst_capacity);
@@ -148,8 +151,9 @@ static int mpack_parser_full(mpack_parser_t *parser)
   return parser->size == parser->capacity;
 }
 
-static mpack_node_t *mpack_parser_push(mpack_parser_t *parser)
+static mpack_node_t *mpack_parser_push(mpack_parser_t *p)
 {
+  mpack_one_parser_t *parser = (mpack_one_parser_t *)p;
   mpack_node_t *top;
   assert(parser->size < parser->capacity);
   top = parser->items + parser->size + 1;
@@ -162,8 +166,9 @@ static mpack_node_t *mpack_parser_push(mpack_parser_t *parser)
   return top;
 }
 
-static mpack_node_t *mpack_parser_pop(mpack_parser_t *parser)
+static mpack_node_t *mpack_parser_pop(mpack_parser_t *p)
 {
+  mpack_one_parser_t *parser = (mpack_one_parser_t *)p;
   mpack_node_t *top, *parent;
   assert(parser->size);
   top = parser->items + parser->size;
diff --git a/test/CMakeLists.txt b/test/CMakeLists.txt
index 7e5fb07382..7b1a082f4d 100644
--- a/test/CMakeLists.txt
+++ b/test/CMakeLists.txt
@@ -47,6 +47,7 @@ if(BUSTED_PRG)
       -D BUSTED_OUTPUT_TYPE=${BUSTED_OUTPUT_TYPE}
       -D TEST_DIR=${CMAKE_CURRENT_SOURCE_DIR}
       -D BUILD_DIR=${CMAKE_BINARY_DIR}
+      -D DEPS_PREFIX=${DEPS_PREFIX}
       -D TEST_TYPE=functional
       -D CIRRUS_CI=$ENV{CIRRUS_CI}
       -P ${PROJECT_SOURCE_DIR}/cmake/RunTests.cmake
diff --git a/test/busted_runner.lua b/test/busted_runner.lua
index b195ce3cc5..62d1e611e7 100644
--- a/test/busted_runner.lua
+++ b/test/busted_runner.lua
@@ -1 +1,12 @@
+local platform = vim.loop.os_uname()
+if platform and platform.sysname:lower():find'windows' then
+  local deps_prefix = os.getenv 'DEPS_PREFIX'
+  if deps_prefix ~= nil and deps_prefix ~= "" then
+    package.path = deps_prefix.."/share/lua/5.1/?.lua;"..deps_prefix.."/share/lua/5.1/?/init.lua;"..package.path
+    package.path = deps_prefix.."/bin/lua/?.lua;"..deps_prefix.."/bin/lua/?/init.lua;"..package.path
+    package.cpath = deps_prefix.."/lib/lua/5.1/?.dll;"..package.cpath;
+    package.cpath = deps_prefix.."/bin/?.dll;"..deps_prefix.."/bin/loadall.dll;"..package.cpath;
+  end
+end
+
 require 'busted.runner'({ standalone = false })
diff --git a/test/client/msgpack_rpc_stream.lua b/test/client/msgpack_rpc_stream.lua
new file mode 100644
index 0000000000..5711616b17
--- /dev/null
+++ b/test/client/msgpack_rpc_stream.lua
@@ -0,0 +1,112 @@
+local mpack = require('mpack')
+
+-- temporary hack to be able to manipulate buffer/window/tabpage
+local Buffer = {}
+Buffer.__index = Buffer
+function Buffer.new(id) return setmetatable({id=id}, Buffer) end
+local Window = {}
+Window.__index = Window
+function Window.new(id) return setmetatable({id=id}, Window) end
+local Tabpage = {}
+Tabpage.__index = Tabpage
+function Tabpage.new(id) return setmetatable({id=id}, Tabpage) end
+
+local Response = {}
+Response.__index = Response
+
+function Response.new(msgpack_rpc_stream, request_id)
+  return setmetatable({
+    _msgpack_rpc_stream = msgpack_rpc_stream,
+    _request_id = request_id
+  }, Response)
+end
+
+function Response:send(value, is_error)
+  local data = self._msgpack_rpc_stream._session:reply(self._request_id)
+  if is_error then
+    data = data .. self._msgpack_rpc_stream._pack(value)
+    data = data .. self._msgpack_rpc_stream._pack(mpack.NIL)
+  else
+    data = data .. self._msgpack_rpc_stream._pack(mpack.NIL)
+    data = data .. self._msgpack_rpc_stream._pack(value)
+  end
+  self._msgpack_rpc_stream._stream:write(data)
+end
+
+local MsgpackRpcStream = {}
+MsgpackRpcStream.__index = MsgpackRpcStream
+
+function MsgpackRpcStream.new(stream)
+  return setmetatable({
+    _stream = stream,
+    _pack = mpack.Packer({
+      ext = {
+        [Buffer] = function(o) return 0, mpack.encode(o.id) end,
+        [Window] = function(o) return 1, mpack.encode(o.id) end,
+        [Tabpage] = function(o) return 2, mpack.encode(o.id) end
+      }
+    }),
+    _session = mpack.Session({
+      unpack = mpack.Unpacker({
+        ext = {
+          [0] = function(_c, s) return Buffer.new(mpack.decode(s)) end,
+          [1] = function(_c, s) return Window.new(mpack.decode(s)) end,
+          [2] = function(_c, s) return Tabpage.new(mpack.decode(s)) end
+        }
+      })
+    }),
+  }, MsgpackRpcStream)
+end
+
+function MsgpackRpcStream:write(method, args, response_cb)
+  local data
+  if response_cb then
+    assert(type(response_cb) == 'function')
+    data = self._session:request(response_cb)
+  else
+    data = self._session:notify()
+  end
+
+  data = data ..  self._pack(method) ..  self._pack(args)
+  self._stream:write(data)
+end
+
+function MsgpackRpcStream:read_start(request_cb, notification_cb, eof_cb)
+  self._stream:read_start(function(data)
+    if not data then
+      return eof_cb()
+    end
+    local type, id_or_cb, method_or_error, args_or_result
+    local pos = 1
+    local len = #data
+    while pos <= len do
+      type, id_or_cb, method_or_error, args_or_result, pos =
+        self._session:receive(data, pos)
+      if type == 'request' or type == 'notification' then
+        if type == 'request' then
+          request_cb(method_or_error, args_or_result, Response.new(self,
+                                                                   id_or_cb))
+        else
+          notification_cb(method_or_error, args_or_result)
+        end
+      elseif type == 'response' then
+        if method_or_error == mpack.NIL then
+          method_or_error = nil
+        else
+          args_or_result = nil
+        end
+        id_or_cb(method_or_error, args_or_result)
+      end
+    end
+  end)
+end
+
+function MsgpackRpcStream:read_stop()
+  self._stream:read_stop()
+end
+
+function MsgpackRpcStream:close(signal)
+  self._stream:close(signal)
+end
+
+return MsgpackRpcStream
diff --git a/test/client/session.lua b/test/client/session.lua
new file mode 100644
index 0000000000..0509fa88be
--- /dev/null
+++ b/test/client/session.lua
@@ -0,0 +1,192 @@
+local uv = require('luv')
+local MsgpackRpcStream = require('test.client.msgpack_rpc_stream')
+
+local Session = {}
+Session.__index = Session
+if package.loaded['jit'] then
+  -- luajit pcall is already coroutine safe
+  Session.safe_pcall = pcall
+else
+  Session.safe_pcall = require'coxpcall'.pcall
+end
+
+local function resume(co, ...)
+  local status, result = coroutine.resume(co, ...)
+
+  if coroutine.status(co) == 'dead' then
+    if not status then
+      error(result)
+    end
+    return
+  end
+
+  assert(coroutine.status(co) == 'suspended')
+  result(co)
+end
+
+local function coroutine_exec(func, ...)
+  local args = {...}
+  local on_complete
+
+  if #args > 0 and type(args[#args]) == 'function' then
+    -- completion callback
+    on_complete = table.remove(args)
+  end
+
+  resume(coroutine.create(function()
+    local status, result, flag = Session.safe_pcall(func, unpack(args))
+    if on_complete then
+      coroutine.yield(function()
+        -- run the completion callback on the main thread
+        on_complete(status, result, flag)
+      end)
+    end
+  end))
+end
+
+function Session.new(stream)
+  return setmetatable({
+    _msgpack_rpc_stream = MsgpackRpcStream.new(stream),
+    _pending_messages = {},
+    _prepare = uv.new_prepare(),
+    _timer = uv.new_timer(),
+    _is_running = false
+  }, Session)
+end
+
+function Session:next_message(timeout)
+  local function on_request(method, args, response)
+    table.insert(self._pending_messages, {'request', method, args, response})
+    uv.stop()
+  end
+
+  local function on_notification(method, args)
+    table.insert(self._pending_messages, {'notification', method, args})
+    uv.stop()
+  end
+
+  if self._is_running then
+    error('Event loop already running')
+  end
+
+  if #self._pending_messages > 0 then
+    return table.remove(self._pending_messages, 1)
+  end
+
+  self:_run(on_request, on_notification, timeout)
+  return table.remove(self._pending_messages, 1)
+end
+
+function Session:notify(method, ...)
+  self._msgpack_rpc_stream:write(method, {...})
+end
+
+function Session:request(method, ...)
+  local args = {...}
+  local err, result
+  if self._is_running then
+    err, result = self:_yielding_request(method, args)
+  else
+    err, result = self:_blocking_request(method, args)
+  end
+
+  if err then
+    return false, err
+  end
+
+  return true, result
+end
+
+function Session:run(request_cb, notification_cb, setup_cb, timeout)
+  local function on_request(method, args, response)
+    coroutine_exec(request_cb, method, args, function(status, result, flag)
+      if status then
+        response:send(result, flag)
+      else
+        response:send(result, true)
+      end
+    end)
+  end
+
+  local function on_notification(method, args)
+    coroutine_exec(notification_cb, method, args)
+  end
+
+  self._is_running = true
+
+  if setup_cb then
+    coroutine_exec(setup_cb)
+  end
+
+  while #self._pending_messages > 0 do
+    local msg = table.remove(self._pending_messages, 1)
+    if msg[1] == 'request' then
+      on_request(msg[2], msg[3], msg[4])
+    else
+      on_notification(msg[2], msg[3])
+    end
+  end
+
+  self:_run(on_request, on_notification, timeout)
+  self._is_running = false
+end
+
+function Session:stop()
+  uv.stop()
+end
+
+function Session:close(signal)
+  if not self._timer:is_closing() then self._timer:close() end
+  if not self._prepare:is_closing() then self._prepare:close() end
+  self._msgpack_rpc_stream:close(signal)
+end
+
+function Session:_yielding_request(method, args)
+  return coroutine.yield(function(co)
+    self._msgpack_rpc_stream:write(method, args, function(err, result)
+      resume(co, err, result)
+    end)
+  end)
+end
+
+function Session:_blocking_request(method, args)
+  local err, result
+
+  local function on_request(method_, args_, response)
+    table.insert(self._pending_messages, {'request', method_, args_, response})
+  end
+
+  local function on_notification(method_, args_)
+    table.insert(self._pending_messages, {'notification', method_, args_})
+  end
+
+  self._msgpack_rpc_stream:write(method, args, function(e, r)
+    err = e
+    result = r
+    uv.stop()
+  end)
+
+  self:_run(on_request, on_notification)
+  return (err or self.eof_err), result
+end
+
+function Session:_run(request_cb, notification_cb, timeout)
+  if type(timeout) == 'number' then
+    self._prepare:start(function()
+      self._timer:start(timeout, 0, function()
+        uv.stop()
+      end)
+      self._prepare:stop()
+    end)
+  end
+  self._msgpack_rpc_stream:read_start(request_cb, notification_cb, function()
+    uv.stop()
+    self.eof_err = {1, "EOF was received from Nvim. Likely the Nvim process crashed."}
+  end)
+  uv.run()
+  self._prepare:stop()
+  self._timer:stop()
+  self._msgpack_rpc_stream:read_stop()
+end
+
+return Session
diff --git a/test/client/uv_stream.lua b/test/client/uv_stream.lua
new file mode 100644
index 0000000000..cea77f0dbd
--- /dev/null
+++ b/test/client/uv_stream.lua
@@ -0,0 +1,169 @@
+local uv = require('luv')
+
+local StdioStream = {}
+StdioStream.__index = StdioStream
+
+function StdioStream.open()
+  local self = setmetatable({
+    _in = uv.new_pipe(false),
+    _out = uv.new_pipe(false)
+  }, StdioStream)
+  self._in:open(0)
+  self._out:open(1)
+  return self
+end
+
+function StdioStream:write(data)
+  self._out:write(data)
+end
+
+function StdioStream:read_start(cb)
+  self._in:read_start(function(err, chunk)
+    if err then
+      error(err)
+    end
+    cb(chunk)
+  end)
+end
+
+function StdioStream:read_stop()
+  self._in:read_stop()
+end
+
+function StdioStream:close()
+  self._in:close()
+  self._out:close()
+end
+
+local SocketStream = {}
+SocketStream.__index = SocketStream
+
+function SocketStream.open(file)
+  local socket = uv.new_pipe(false)
+  local self = setmetatable({
+    _socket = socket,
+    _stream_error = nil
+  }, SocketStream)
+  uv.pipe_connect(socket, file, function (err)
+    self._stream_error = self._stream_error or err
+  end)
+  return self
+end
+
+function SocketStream.connect(host, port)
+  local socket = uv.new_tcp()
+  local self = setmetatable({
+    _socket = socket,
+    _stream_error = nil
+  }, SocketStream)
+  uv.tcp_connect(socket, host, port, function (err)
+    self._stream_error = self._stream_error or err
+  end)
+  return self
+end
+
+
+function SocketStream:write(data)
+  if self._stream_error then
+    error(self._stream_error)
+  end
+  uv.write(self._socket, data, function(err)
+    if err then
+      error(self._stream_error or err)
+    end
+  end)
+end
+
+function SocketStream:read_start(cb)
+  if self._stream_error then
+    error(self._stream_error)
+  end
+  uv.read_start(self._socket, function(err, chunk)
+    if err then
+      error(err)
+    end
+    cb(chunk)
+  end)
+end
+
+function SocketStream:read_stop()
+  if self._stream_error then
+    error(self._stream_error)
+  end
+  uv.read_stop(self._socket)
+end
+
+function SocketStream:close()
+  uv.close(self._socket)
+end
+
+local ChildProcessStream = {}
+ChildProcessStream.__index = ChildProcessStream
+
+function ChildProcessStream.spawn(argv, env, io_extra)
+  local self = setmetatable({
+    _child_stdin = uv.new_pipe(false);
+    _child_stdout = uv.new_pipe(false);
+    _exiting = false;
+  }, ChildProcessStream)
+  local prog = argv[1]
+  local args = {}
+  for i = 2, #argv do
+    args[#args + 1] = argv[i]
+  end
+  self._proc, self._pid = uv.spawn(prog, {
+    stdio = {self._child_stdin, self._child_stdout, 2, io_extra},
+    args = args,
+    env = env,
+  }, function(status, signal)
+    self.status = status
+    self.signal = signal
+  end)
+
+  if not self._proc then
+    local err = self._pid
+    error(err)
+  end
+
+  return self
+end
+
+function ChildProcessStream:write(data)
+  self._child_stdin:write(data)
+end
+
+function ChildProcessStream:read_start(cb)
+  self._child_stdout:read_start(function(err, chunk)
+    if err then
+      error(err)
+    end
+    cb(chunk)
+  end)
+end
+
+function ChildProcessStream:read_stop()
+  self._child_stdout:read_stop()
+end
+
+function ChildProcessStream:close(signal)
+  if self._closed then
+    return
+  end
+  self._closed = true
+  self:read_stop()
+  self._child_stdin:close()
+  self._child_stdout:close()
+  if type(signal) == 'string' then
+    self._proc:kill('sig'..signal)
+  end
+  while self.status == nil do
+    uv.run 'once'
+  end
+  return self.status, self.signal
+end
+
+return {
+  StdioStream = StdioStream;
+  ChildProcessStream = ChildProcessStream;
+  SocketStream = SocketStream;
+}
diff --git a/test/functional/api/rpc_fixture.lua b/test/functional/api/rpc_fixture.lua
index 94df751363..c860a6da59 100644
--- a/test/functional/api/rpc_fixture.lua
+++ b/test/functional/api/rpc_fixture.lua
@@ -4,9 +4,8 @@
 package.path = arg[1]
 package.cpath = arg[2]
 
-local mpack = require('mpack')
-local StdioStream = require('nvim.stdio_stream')
-local Session = require('nvim.session')
+local StdioStream = require'test.client.uv_stream'.StdioStream
+local Session = require'test.client.session'
 
 local stdio_stream = StdioStream.open()
 local session = Session.new(stdio_stream)
@@ -19,7 +18,7 @@ local function on_request(method, args)
     return "done!"
   elseif method == "exit" then
     session:stop()
-    return mpack.NIL
+    return vim.NIL
   end
 end
 
diff --git a/test/functional/api/server_requests_spec.lua b/test/functional/api/server_requests_spec.lua
index e6bfc6b64f..cb273aedba 100644
--- a/test/functional/api/server_requests_spec.lua
+++ b/test/functional/api/server_requests_spec.lua
@@ -237,7 +237,7 @@ describe('server -> client', function()
         \ }
       ]])
       meths.set_var("args", {
-        helpers.test_lua_prg,
+        nvim_prog, '-ll',
         'test/functional/api/rpc_fixture.lua',
         package.path,
         package.cpath,
diff --git a/test/functional/api/vim_spec.lua b/test/functional/api/vim_spec.lua
index 8fcdd9620b..fc550f5861 100644
--- a/test/functional/api/vim_spec.lua
+++ b/test/functional/api/vim_spec.lua
@@ -59,7 +59,7 @@ describe('API', function()
 
     -- XXX: This must be the last one, else next one will fail:
     --      "Packer instance already working. Use another Packer ..."
-    matches("can't serialize object$",
+    matches("can't serialize object of type .$",
       pcall_err(request, nil))
   end)
 
diff --git a/test/functional/helpers.lua b/test/functional/helpers.lua
index 6400db9f87..cd6b535477 100644
--- a/test/functional/helpers.lua
+++ b/test/functional/helpers.lua
@@ -1,14 +1,11 @@
-require('coxpcall')
 local luv = require('luv')
 local lfs = require('lfs')
-local mpack = require('mpack')
 local global_helpers = require('test.helpers')
 
--- nvim client: Found in .deps/usr/share/lua/<version>/nvim/ if "bundled".
-local Session = require('nvim.session')
-local TcpStream = require('nvim.tcp_stream')
-local SocketStream = require('nvim.socket_stream')
-local ChildProcessStream = require('nvim.child_process_stream')
+local Session = require('test.client.session')
+local uv_stream = require('test.client.uv_stream')
+local SocketStream = uv_stream.SocketStream
+local ChildProcessStream = uv_stream.ChildProcessStream
 
 local check_cores = global_helpers.check_cores
 local check_logs = global_helpers.check_logs
@@ -23,7 +20,6 @@ local tbl_contains = global_helpers.tbl_contains
 local fail = global_helpers.fail
 
 local module = {
-  NIL = mpack.NIL,
   mkdir = lfs.mkdir,
 }
 
@@ -202,7 +198,7 @@ function module.expect_msg_seq(...)
 end
 
 local function call_and_stop_on_error(lsession, ...)
-  local status, result = copcall(...)  -- luacheck: ignore
+  local status, result = Session.safe_pcall(...)  -- luacheck: ignore
   if not status then
     lsession:stop()
     last_error = result
@@ -428,7 +424,7 @@ end
 -- Creates a new Session connected by domain socket (named pipe) or TCP.
 function module.connect(file_or_address)
   local addr, port = string.match(file_or_address, "(.*):(%d+)")
-  local stream = (addr and port) and TcpStream.open(addr, port) or
+  local stream = (addr and port) and SocketStream.connect(addr, port) or
     SocketStream.open(file_or_address)
   return Session.new(stream)
 end
diff --git a/test/functional/shada/shada_spec.lua b/test/functional/shada/shada_spec.lua
index 88a99d9b55..7b1ce5d0ca 100644
--- a/test/functional/shada/shada_spec.lua
+++ b/test/functional/shada/shada_spec.lua
@@ -126,11 +126,11 @@ describe('ShaDa support code', function()
     wshada(s .. table.concat(msgpack, e .. s) .. e)
     eq(0, exc_exec('wshada ' .. shada_fname))
     local found = 0
-    local typ = mpack.unpack(s)
+    local typ = mpack.decode(s)
     for _, v in ipairs(read_shada_file(shada_fname)) do
       if v.type == typ then
         found = found + 1
-        eq(mpack.unpack(msgpack[found]), v.timestamp)
+        eq(mpack.decode(msgpack[found]), v.timestamp)
       end
     end
     eq(#msgpack, found)
diff --git a/test/helpers.lua b/test/helpers.lua
index 82ff23bef8..d45536b42b 100644
--- a/test/helpers.lua
+++ b/test/helpers.lua
@@ -1,5 +1,5 @@
 require('test.compat')
-local shared = require('vim.shared')
+local shared = vim
 local assert = require('luassert')
 local busted = require('busted')
 local luv = require('luv')