path: root/test/functional/terminal/channel_spec.lua

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

local clear = n.clear
local eq = t.eq
local eval = n.eval
local command = n.command
local pcall_err = t.pcall_err
local feed = n.feed
local poke_eventloop = n.poke_eventloop
local is_os = t.is_os
local api = n.api
local async_meths = n.async_meths
local testprg = n.testprg
local assert_alive = n.assert_alive

describe('terminal channel is closed and later released if', function()
  local screen

  before_each(function()
    clear()
    screen = Screen.new()
  end)

  it('opened by nvim_open_term() and deleted by :bdelete!', function()
    command([[let id = nvim_open_term(0, {})]])
    local chans = eval('len(nvim_list_chans())')
    -- Channel hasn't been released yet.
    eq(
      "Vim(call):Can't send data to closed stream",
      pcall_err(command, [[bdelete! | call chansend(id, 'test')]])
    )
    -- Channel has been released after processing free_channel_event().
    eq(chans - 1, eval('len(nvim_list_chans())'))

    command('autocmd BufWipeout * ++once let id2 = nvim_open_term(str2nr(expand("<abuf>")), {})')
    -- Channel hasn't been released yet.
    eq(
      "Vim(call):Can't send data to closed stream",
      pcall_err(command, [[bdelete! | call chansend(id2, 'test')]])
    )
    -- Channel has been released after processing free_channel_event().
    eq(chans - 1, eval('len(nvim_list_chans())'))
  end)

  it('opened by nvim_open_term(), closed by chanclose(), and deleted by pressing a key', function()
    command('let id = nvim_open_term(0, {})')
    local chans = eval('len(nvim_list_chans())')
    -- Channel has been closed but not released.
    eq(
      "Vim(call):Can't send data to closed stream",
      pcall_err(command, [[call chanclose(id) | call chansend(id, 'test')]])
    )
    screen:expect({ any = '%[Terminal closed%]' })
    eq(chans, eval('len(nvim_list_chans())'))
    feed('i<CR>') -- Delete terminal.
    poke_eventloop() -- Process pending input.
    -- Channel has been released after processing free_channel_event().
    eq(chans - 1, eval('len(nvim_list_chans())'))
  end)

  it('opened by nvim_open_term(), closed by chanclose(), and deleted by :bdelete', function()
    command('let id = nvim_open_term(0, {})')
    local chans = eval('len(nvim_list_chans())')
    -- Channel has been closed but not released.
    eq(
      "Vim(call):Can't send data to closed stream",
      pcall_err(command, [[call chanclose(id) | call chansend(id, 'test')]])
    )
    screen:expect({ any = '%[Terminal closed%]' })
    eq(chans, eval('len(nvim_list_chans())'))
    -- Channel still hasn't been released yet.
    eq(
      "Vim(call):Can't send data to closed stream",
      pcall_err(command, [[bdelete | call chansend(id, 'test')]])
    )
    -- Channel has been released after processing free_channel_event().
    eq(chans - 1, eval('len(nvim_list_chans())'))
  end)

  it('opened by jobstart(…,{term=true}), exited, and deleted by pressing a key', function()
    command([[let id = jobstart('echo',{'term':v:true})]])
    local chans = eval('len(nvim_list_chans())')
    -- Wait for process to exit.
    screen:expect({ any = '%[Process exited 0%]' })
    -- Process has exited but channel has't been released.
    eq(
      "Vim(call):Can't send data to closed stream",
      pcall_err(command, [[call chansend(id, 'test')]])
    )
    eq(chans, eval('len(nvim_list_chans())'))
    feed('i<CR>') -- Delete terminal.
    poke_eventloop() -- Process pending input.
    poke_eventloop() -- Process term_delayed_free().
    -- Channel has been released after processing free_channel_event().
    eq(chans - 1, eval('len(nvim_list_chans())'))
  end)

  -- This indirectly covers #16264
  it('opened by jobstart(…,{term=true}), exited, and deleted by :bdelete', function()
    command([[let id = jobstart('echo', {'term':v:true})]])
    local chans = eval('len(nvim_list_chans())')
    -- Wait for process to exit.
    screen:expect({ any = '%[Process exited 0%]' })
    -- Process has exited but channel hasn't been released.
    eq(
      "Vim(call):Can't send data to closed stream",
      pcall_err(command, [[call chansend(id, 'test')]])
    )
    eq(chans, eval('len(nvim_list_chans())'))
    -- Channel still hasn't been released yet.
    eq(
      "Vim(call):Can't send data to closed stream",
      pcall_err(command, [[bdelete | call chansend(id, 'test')]])
    )
    poke_eventloop() -- Process term_delayed_free().
    -- Channel has been released after processing free_channel_event().
    eq(chans - 1, eval('len(nvim_list_chans())'))
  end)
end)

it('chansend sends lines to terminal channel in proper order', function()
  clear({ args = { '--cmd', 'set laststatus=2' } })
  local screen = Screen.new(100, 20)
  screen._default_attr_ids = nil
  local shells = is_os('win') and { 'cmd.exe', 'pwsh.exe -nop', 'powershell.exe -nop' } or { 'sh' }
  for _, sh in ipairs(shells) do
    command([[let id = jobstart(']] .. sh .. [[', {'term':v:true})]])
    screen:sleep(50) -- Wait some time for the shell to start.
    command([[call chansend(id, ['echo "hello"', 'echo "world"', ''])]])
    -- With PowerShell the command may be highlighted, so specify attr_ids = {}.
    screen:expect({ any = [[echo "hello".*echo "world"]], attr_ids = {} })
    command('bdelete!')
    screen:expect({ any = '%[No Name%]' })
  end
end)

--- @param event string
--- @param extra_tests fun(table, table)?
local function test_autocmd_no_crash(event, extra_tests)
  local env = {}
  -- Use REPFAST for immediately output after start.
  local term_args = { testprg('shell-test'), 'REPFAST', '50', 'TEST' }

  before_each(function()
    clear()
    env.screen = Screen.new(60, 4)
    command([[file Xoldbuf | call setline(1, 'OLDBUF') | enew]])
    -- Wait before :bwipe to avoid closing PTY master before the child calls setsid(),
    -- as that will cause SIGHUP to be also sent to the parent.
    -- Use vim.uv.sleep() which blocks the event loop.
    n.exec([[
      func Wipe()
        lua vim.uv.sleep(5)
        bwipe!
      endfunc
    ]])
  end)

  local input_prompt_screen = [[
                                                                |
    {1:~                                                           }|*2
    ^                                                            |
  ]]
  local oldbuf_screen = [[
    ^OLDBUF                                                      |
    {1:~                                                           }|*2
                                                                |
  ]]

  it('processes job exit event when using jobstart(…,{term=true})', function()
    api.nvim_create_autocmd(event, { command = "call input('')" })
    async_meths.nvim_call_function('jobstart', { term_args, { term = true } })
    env.screen:expect(input_prompt_screen)
    vim.uv.sleep(20)
    feed('<CR>')
    env.screen:expect([[
      ^0: TEST                                                     |
      1: TEST                                                     |
      2: TEST                                                     |
                                                                  |
    ]])
    feed('i')
    env.screen:expect([[
      48: TEST                                                    |
      49: TEST                                                    |
      ^[Process exited 0]                                          |
      {5:-- TERMINAL --}                                              |
    ]])
    feed('<CR>')
    env.screen:expect(oldbuf_screen)
    assert_alive()
  end)

  it('wipes buffer and processes events when using jobstart(…,{term=true})', function()
    api.nvim_create_autocmd(event, { command = "call Wipe() | call input('')" })
    async_meths.nvim_call_function('jobstart', { term_args, { term = true } })
    env.screen:expect(input_prompt_screen)
    vim.uv.sleep(20)
    feed('<CR>')
    env.screen:expect(oldbuf_screen)
    assert_alive()
    eq('Xoldbuf', eval('bufname()'))
    eq(0, eval([[exists('b:term_title')]]))
  end)

  it('processes :bwipe from TermClose when using jobstart(…,{term=true})', function()
    local term_buf = api.nvim_get_current_buf()
    api.nvim_create_autocmd('TermClose', { command = ('bwipe! %d'):format(term_buf) })
    api.nvim_create_autocmd(event, { command = "call input('')", nested = true })
    async_meths.nvim_call_function('jobstart', { term_args, { term = true } })
    env.screen:expect(input_prompt_screen)
    vim.uv.sleep(20)
    feed('<CR>')
    env.screen:expect(oldbuf_screen)
    assert_alive()
    eq('Xoldbuf', eval('bufname()'))
    eq(0, eval([[exists('b:term_title')]]))
  end)

  it('only wipes buffer when using jobstart(…,{term=true})', function()
    api.nvim_create_autocmd(event, { command = 'call Wipe()' })
    async_meths.nvim_call_function('jobstart', { term_args, { term = true } })
    env.screen:expect(oldbuf_screen)
    assert_alive()
    eq('Xoldbuf', eval('bufname()'))
    eq(0, eval([[exists('b:term_title')]]))
  end)

  if extra_tests then
    extra_tests(env, term_args)
  end
end

describe('no crash when TermOpen autocommand', function()
  test_autocmd_no_crash('TermOpen', function(env, term_args)
    it('wipes buffer and processes events when using nvim_open_term()', function()
      api.nvim_create_autocmd('TermOpen', { command = "call Wipe() | call input('')" })
      async_meths.nvim_open_term(0, {})
      env.screen:expect([[
                                                                    |
        {1:~                                                           }|*2
        ^                                                            |
      ]])
      feed('<CR>')
      env.screen:expect([[
        ^OLDBUF                                                      |
        {1:~                                                           }|*2
                                                                    |
      ]])
      assert_alive()
    end)

    it('wipes buffer when using jobstart(…,{term=true}) during Nvim exit', function()
      n.expect_exit(n.exec_lua, function()
        vim.schedule(function()
          vim.fn.jobstart(term_args, { term = true })
        end)
        vim.api.nvim_create_autocmd('TermOpen', { command = 'call Wipe()' })
        vim.cmd('qall!')
      end)
    end)
  end)
end)

describe('no crash when BufFilePre autocommand', function()
  test_autocmd_no_crash('BufFilePre')
end)

describe('no crash when BufFilePost autocommand', function()
  test_autocmd_no_crash('BufFilePost')
end)

describe('nvim_open_term', function()
  local screen

  before_each(function()
    clear()
    screen = Screen.new(8, 10)
  end)

  it('with force_crlf=true converts newlines', function()
    local win = api.nvim_get_current_win()
    local buf = api.nvim_create_buf(false, true)
    local term = api.nvim_open_term(buf, { force_crlf = true })
    api.nvim_win_set_buf(win, buf)
    api.nvim_chan_send(term, 'here\nthere\nfoo\r\nbar\n\ntest')
    screen:expect([[
      ^here        |
      there       |
      foo         |
      bar         |
                  |
      test        |
                  |*4
    ]])
    api.nvim_chan_send(term, '\nfirst')
    screen:expect([[
      ^here        |
      there       |
      foo         |
      bar         |
                  |
      test        |
      first       |
                  |*3
    ]])
  end)

  it('with force_crlf=false does not convert newlines', function()
    local win = api.nvim_get_current_win()
    local buf = api.nvim_create_buf(false, true)
    local term = api.nvim_open_term(buf, { force_crlf = false })
    api.nvim_win_set_buf(win, buf)
    api.nvim_chan_send(term, 'here\nthere')
    screen:expect([[
      ^here        |
          there   |
                  |*8
    ]])
  end)
end)