path: root/test/functional/core/channels_spec.lua

local t = require('test.testutil')
local n = require('test.functional.testnvim')()

local describe, it, before_each, finally = t.describe, t.it, t.before_each, t.finally
local clear, eq, eval, next_msg, ok, source = n.clear, t.eq, n.eval, n.next_msg, t.ok, n.source
local command, fn, api = n.command, n.fn, n.api
local feed = n.feed
local exec_lua = n.exec_lua
local matches = t.matches
local sleep = vim.uv.sleep
local get_session, set_session = n.get_session, n.set_session
local nvim_prog = n.nvim_prog
local is_os = t.is_os
local retry = t.retry
local expect_twostreams = n.expect_twostreams
local assert_alive = n.assert_alive
local pcall_err = t.pcall_err
local skip = t.skip

describe('channels', function()
  local init = [[
    function! Normalize(data) abort
      " Windows: remove ^M
      return type([]) == type(a:data)
        \ ? mapnew(a:data, 'substitute(v:val, "\r", "", "g")')
        \ : a:data
    endfunction
    function! OnEvent(id, data, event) dict
      call rpcnotify(1, a:event, a:id, a:data)
    endfunction
  ]]
  before_each(function()
    clear()
    source(init)
  end)

  it('can connect to socket', function()
    local server = n.new_session(true)
    set_session(server)
    local address = fn.serverlist()[1]
    local client = n.new_session(true)
    set_session(client)
    source(init)

    api.nvim_set_var('address', address)
    command("let g:id = sockconnect('pipe', address, {'on_data':'OnEvent'})")
    local id = eval('g:id')
    ok(id > 0)

    command("call chansend(g:id, msgpackdump([[2,'nvim_set_var',['code',23]]]))")
    set_session(server)
    retry(nil, 1000, function()
      eq(23, api.nvim_get_var('code'))
    end)
    set_session(client)

    command("call chansend(g:id, msgpackdump([[0,0,'nvim_eval',['2+3']]]))")

    local res = eval('msgpackdump([[1,0,v:null,5]])')
    eq({ '\148\001\n\192\005' }, res)
    eq({ 'notification', 'data', { id, res } }, next_msg())
    command("call chansend(g:id, msgpackdump([[2,'nvim_command',['quit']]]))")
    eq({ 'notification', 'data', { id, { '' } } }, next_msg())
  end)

  it('emits ChanClose when RPC socket reaches EOF', function()
    local client = get_session()
    local server = n.new_session(true)
    finally(function()
      set_session(client)
      if server then
        server:close()
      end
    end)
    set_session(server)
    local address = fn.serverlist()[1]
    set_session(client)

    api.nvim_set_var('address', address)
    exec_lua(function()
      _G.closed_event = nil
      vim.api.nvim_create_autocmd('ChanClose', {
        callback = function()
          _G.closed_event = vim.deepcopy(vim.v.event.info)
        end,
      })
    end)
    command("let g:id = sockconnect('pipe', address, {'rpc': v:true})")
    local id = eval('g:id')
    ok(id > 0)
    eq(5, eval("rpcrequest(g:id, 'nvim_eval', '2+3')"))

    server:close()
    server = nil
    set_session(client)
    retry(nil, 3000, function()
      local info = exec_lua(function()
        return _G.closed_event
      end)
      eq(id, info.id)
      eq('socket', info.stream)
      eq('rpc', info.mode)
    end)
  end)

  it('dont crash due to garbage in rpc #23781', function()
    local client = get_session()
    local server = n.new_session(true)
    set_session(server)
    local address = fn.serverlist()[1]
    set_session(client)

    api.nvim_set_var('address', address)
    command("let g:id = sockconnect('pipe', address, {'on_data':'OnEvent'})")
    local id = eval('g:id')
    ok(id > 0)

    command("call chansend(g:id, 'F')")
    eq({ 'notification', 'data', { id, { '' } } }, next_msg())
    set_session(server)
    assert_alive()

    set_session(client)
    command('call chanclose(g:id)')
    command("let g:id = sockconnect('pipe', address, {'on_data':'OnEvent'})")
    id = eval('g:id')
    ok(id > 0)

    command("call chansend(g:id, msgpackdump([[2, 'redraw', 'F']], 'B')[:-4])")
    set_session(server)
    assert_alive()
    set_session(client)
    command("call chansend(g:id, 'F')")
    eq({ 'notification', 'data', { id, { '' } } }, next_msg())

    set_session(server)
    assert_alive()
    set_session(client)
    command('call chanclose(g:id)')
    server:close()
  end)

  it('can use stdio channel', function()
    source([[
      let g:job_opts = {
      \ 'on_stdout': function('OnEvent'),
      \ 'on_stderr': function('OnEvent'),
      \ 'on_exit': function('OnEvent'),
      \ }
    ]])
    api.nvim_set_var('nvim_prog', nvim_prog)
    api.nvim_set_var(
      'code',
      [[
      function! OnEvent(id, data, event) dict
        let text = string([a:id, a:data, a:event])
        call chansend(g:x, text)

        if a:data == ['']
          call chansend(v:stderr, "*dies*")
          quit
        endif
      endfunction
      let g:x = stdioopen({'on_stdin':'OnEvent'})
      call chansend(x, "hello")
    ]]
    )
    command(
      "let g:id = jobstart([ g:nvim_prog, '-u', 'NONE', '-i', 'NONE', '--cmd', 'set noswapfile', '--headless', '--cmd', g:code], g:job_opts)"
    )
    local id = eval('g:id')
    ok(id > 0)

    eq({ 'notification', 'stdout', { id, { 'hello' } } }, next_msg())

    command("call chansend(id, 'howdy')")
    eq({ 'notification', 'stdout', { id, { "[1, ['howdy'], 'stdin']" } } }, next_msg())

    command('call chansend(id, 0z686f6c61)')
    eq({ 'notification', 'stdout', { id, { "[1, ['hola'], 'stdin']" } } }, next_msg())

    command("call chanclose(id, 'stdin')")
    expect_twostreams({
      { 'notification', 'stdout', { id, { "[1, [''], 'stdin']" } } },
      { 'notification', 'stdout', { id, { '' } } },
    }, {
      { 'notification', 'stderr', { id, { '*dies*' } } },
      { 'notification', 'stderr', { id, { '' } } },
    })
    eq({ 'notification', 'exit', { 3, 0 } }, next_msg())
  end)

  it('can use stdio channel and on_print callback', function()
    source([[
      let g:job_opts = {
      \ 'on_stdout': function('OnEvent'),
      \ 'on_stderr': function('OnEvent'),
      \ 'on_exit': function('OnEvent'),
      \ }
    ]])
    api.nvim_set_var('nvim_prog', nvim_prog)
    api.nvim_set_var(
      'code',
      [[
      function! OnStdin(id, data, event) dict
        echo string([a:id, a:data, a:event])
        if a:data == ['']
          quit
        endif
      endfunction
      function! OnPrint(text) dict
        call chansend(g:x, ['OnPrint:' .. a:text])
      endfunction
      let g:x = stdioopen({'on_stdin': funcref('OnStdin'), 'on_print':'OnPrint'})
      call chansend(x, "hello")
    ]]
    )
    command(
      "let g:id = jobstart([ g:nvim_prog, '-u', 'NONE', '-i', 'NONE', '--cmd', 'set noswapfile', '--headless', '--cmd', g:code], g:job_opts)"
    )
    local id = eval('g:id')
    ok(id > 0)

    eq({ 'notification', 'stdout', { id, { 'hello' } } }, next_msg())

    command("call chansend(id, 'howdy')")
    eq({ 'notification', 'stdout', { id, { "OnPrint:[1, ['howdy'], 'stdin']" } } }, next_msg())
  end)

  -- Helper to accumulate PTY stdout data until expected string is received.
  -- PTY reads are non-atomic and may deliver data in chunks.
  local function expect_stdout(id, expected)
    local accumulated = ''
    while #accumulated < #expected do
      local msg = next_msg(1000)
      if not msg then
        break
      end
      eq('notification', msg[1])
      eq('stdout', msg[2])
      eq(id, msg[3][1])
      accumulated = accumulated .. table.concat(msg[3][2], '\n')
      -- Windows: strip terminal escapes injected by ConPTY (CSI/OSC).
      accumulated = accumulated:gsub('\27%[[%d;?]*[%a~]', ''):gsub('\27%][^\7]*\7', '')
    end
    eq(expected, accumulated)
  end

  it('can use stdio channel with pty', function()
    source([[
      let g:job_opts = {
      \ 'on_stdout': function('OnEvent'),
      \ 'on_exit': function('OnEvent'),
      \ 'pty': v:true,
      \ }
    ]])
    api.nvim_set_var('nvim_prog', nvim_prog)
    api.nvim_set_var(
      'code',
      [[
      function! OnEvent(id, data, event) dict
        let text = string([a:id, a:data, a:event])
        call chansend(g:x, text)
      endfunction
      let g:x = stdioopen({'on_stdin':'OnEvent'})
    ]]
    )
    command(
      "let g:id = jobstart([ g:nvim_prog, '-u', 'NONE', '-i', 'NONE', '--cmd', 'set noswapfile', '--headless', '--cmd', g:code], g:job_opts)"
    )
    local id = eval('g:id')
    ok(id > 0)

    -- Windows ConPTY does not echo input to the master, unlike POSIX ptys.
    local echo_text = is_os('win') and '' or 'TEXT\r\n'
    local echo_blobs = is_os('win') and '' or 'Blobs!\r\n'

    command("call chansend(id, 'TEXT\n')")
    expect_stdout(id, echo_text .. "[1, ['TEXT', ''], 'stdin']")

    command('call chansend(id, 0z426c6f6273210a)')
    expect_stdout(id, echo_blobs .. "[1, ['Blobs!', ''], 'stdin']")

    -- Windows: the assertions below depend on POSIX pty canonical-mode behavior (echo, backspace handling, EOT-as-EOF).
    if is_os('win') then
      return
    end

    command("call chansend(id, 'neovan')")
    expect_stdout(id, 'neovan')
    command("call chansend(id, '\127\127im\n')")
    expect_stdout(id, "\b \b\b \bim\r\n[1, ['neovim', ''], 'stdin']")

    command("call chansend(id, 'incomplet\004')")

    local bsdlike = is_os('bsd') or is_os('mac')
    local extra = bsdlike and '^D\008\008' or ''
    expect_stdout(id, 'incomplet' .. extra .. "[1, ['incomplet'], 'stdin']")

    command("call chansend(id, '\004')")
    expect_stdout(id, extra .. "[1, [''], 'stdin']")

    -- channel is still open
    command("call chansend(id, 'hi again!\n')")
    expect_stdout(id, 'hi again!\r\n')
  end)

  it('stdio channel can use rpc and stderr simultaneously', function