" Test for channel functions.
scriptencoding utf-8
if !has('channel')
finish
endif
" This test requires the Python command to run the test server.
" This most likely only works on Unix and Windows.
if has('unix')
" We also need the job feature or the pkill command to make sure the server
" can be stopped.
if !(executable('python') && (has('job') || executable('pkill')))
finish
endif
let s:python = 'python'
elseif has('win32')
" Use Python Launcher for Windows (py.exe) if available.
if executable('py.exe')
let s:python = 'py.exe'
elseif executable('python.exe')
let s:python = 'python.exe'
else
finish
endif
else
" Can't run this test.
finish
endif
let s:chopt = {}
" Run "testfunc" after sarting the server and stop the server afterwards.
func s:run_server(testfunc, ...)
" The Python program writes the port number in Xportnr.
call delete("Xportnr")
if a:0 == 1
let arg = ' ' . a:1
else
let arg = ''
endif
let cmd = s:python . " test_channel.py" . arg
try
if has('job')
let s:job = job_start(cmd, {"stoponexit": "hup"})
call job_setoptions(s:job, {"stoponexit": "kill"})
elseif has('win32')
exe 'silent !start cmd /c start "test_channel" ' . cmd
else
exe 'silent !' . cmd . '&'
endif
" Wait for up to 2 seconds for the port number to be there.
let l = []
for i in range(200)
try
let l = readfile("Xportnr")
catch
endtry
if len(l) >= 1
break
endif
sleep 10m
endfor
call delete("Xportnr")
if len(l) == 0
" Can't make the connection, give up.
call assert_false(1, "Can't start test_channel.py")
return -1
endif
let port = l[0]
call call(function(a:testfunc), [port])
catch
call assert_false(1, "Caught exception: " . v:exception)
finally
call s:kill_server()
endtry
endfunc
func s:kill_server()
if has('job')
if exists('s:job')
call job_stop(s:job)
unlet s:job
endif
elseif has('win32')
call system('taskkill /IM ' . s:python . ' /T /F /FI "WINDOWTITLE eq test_channel"')
else
call system("pkill -f test_channel.py")
endif
endfunc
let s:responseMsg = ''
func s:RequestHandler(handle, msg)
let s:responseHandle = a:handle
let s:responseMsg = a:msg
endfunc
" Wait for up to a second for "expr" to become true.
func s:waitFor(expr)
for i in range(100)
try
if eval(a:expr)
return
endif
catch
endtry
sleep 10m
endfor
endfunc
func s:communicate(port)
let handle = ch_open('localhost:' . a:port, s:chopt)
if ch_status(handle) == "fail"
call assert_false(1, "Can't open channel")
return
endif
if has('job')
" check that getjob without a job is handled correctly
call assert_equal('no process', string(ch_getjob(handle)))
endif
let dict = ch_info(handle)
call assert_true(dict.id != 0)
call assert_equal('open', dict.status)
call assert_equal(a:port, string(dict.port))
call assert_equal('open', dict.sock_status)
call assert_equal('socket', dict.sock_io)
" Simple string request and reply.
call assert_equal('got it', ch_evalexpr(handle, 'hello!'))
" Malformed command should be ignored.
call assert_equal('ok', ch_evalexpr(handle, 'malformed1'))
call assert_equal('ok', ch_evalexpr(handle, 'malformed2'))
call assert_equal('ok', ch_evalexpr(handle, 'malformed3'))
" split command should work
call assert_equal('ok', ch_evalexpr(handle, 'split'))
call s:waitFor('exists("g:split")')
call assert_equal(123, g:split)
" Request that triggers sending two ex commands. These will usually be
" handled before getting the response, but it's not guaranteed, thus wait a
" tiny bit for the commands to get executed.
call assert_equal('ok', ch_evalexpr(handle, 'make change'))
call s:waitFor('"added2" == getline("$")')
call assert_equal('added1', getline(line('$') - 1))
call assert_equal('added2', getline('$'))
" Request command "foo bar", which fails silently.
call assert_equal('