82ceb62887
The testing on Travis only relies on tmux which also works locally fine. No need to support two interfaces for the same thing.
230 lines
6.8 KiB
Python
230 lines
6.8 KiB
Python
# encoding: utf-8
|
||
|
||
import os
|
||
import re
|
||
import shutil
|
||
import subprocess
|
||
import tempfile
|
||
import textwrap
|
||
import time
|
||
|
||
from test.constant import (ARR_D, ARR_L, ARR_R, ARR_U, BS, ESC, PYTHON3,
|
||
SEQUENCES)
|
||
|
||
|
||
def wait_until_file_exists(file_path, times=None, interval=0.01):
|
||
while times is None or times:
|
||
if os.path.exists(file_path):
|
||
return True
|
||
time.sleep(interval)
|
||
if times is not None:
|
||
times -= 1
|
||
return False
|
||
|
||
|
||
def read_text_file(filename):
|
||
"""Reads the contens of a text file."""
|
||
if PYTHON3:
|
||
return open(filename, 'r', encoding='utf-8').read()
|
||
else:
|
||
return open(filename, 'r').read()
|
||
|
||
|
||
def is_process_running(pid):
|
||
"""Returns true if a process with pid is running, false otherwise."""
|
||
# from
|
||
# http://stackoverflow.com/questions/568271/how-to-check-if-there-exists-a-process-with-a-given-pid
|
||
try:
|
||
os.kill(pid, 0)
|
||
except OSError:
|
||
return False
|
||
else:
|
||
return True
|
||
|
||
|
||
def silent_call(cmd):
|
||
"""Calls 'cmd' and returns the exit value."""
|
||
return subprocess.call(cmd, stderr=subprocess.PIPE, stdout=subprocess.PIPE)
|
||
|
||
|
||
def create_directory(dirname):
|
||
"""Creates 'dirname' and its parents if it does not exist."""
|
||
try:
|
||
os.makedirs(dirname)
|
||
except OSError:
|
||
pass
|
||
|
||
|
||
class TempFileManager(object):
|
||
|
||
def __init__(self, name=''):
|
||
self._temp_dir = tempfile.mkdtemp(prefix='UltiSnipsTest_' + name)
|
||
|
||
def name_temp(self, file_path):
|
||
return os.path.join(self._temp_dir, file_path)
|
||
|
||
def write_temp(self, file_path, content):
|
||
abs_path = self.name_temp(file_path)
|
||
create_directory(os.path.dirname(abs_path))
|
||
if PYTHON3:
|
||
with open(abs_path, 'w', encoding='utf-8') as f:
|
||
f.write(content)
|
||
else:
|
||
with open(abs_path, 'w') as f:
|
||
f.write(content)
|
||
return abs_path
|
||
|
||
def unique_name_temp(self, suffix='', prefix=''):
|
||
file_handler, abspath = tempfile.mkstemp(
|
||
suffix, prefix, self._temp_dir)
|
||
os.close(file_handler)
|
||
os.remove(abspath)
|
||
return abspath
|
||
|
||
def clear_temp(self):
|
||
shutil.rmtree(self._temp_dir)
|
||
create_directory(self._temp_dir)
|
||
|
||
|
||
class VimInterface(TempFileManager):
|
||
|
||
def __init__(self, name=''):
|
||
TempFileManager.__init__(self, name)
|
||
|
||
def get_buffer_data(self):
|
||
buffer_path = self.unique_name_temp(prefix='buffer_')
|
||
self.send(ESC + ':w! %s\n' % buffer_path)
|
||
if wait_until_file_exists(buffer_path, 50):
|
||
return read_text_file(buffer_path)[:-1]
|
||
|
||
def send(self, s):
|
||
raise NotImplementedError()
|
||
|
||
def launch(self, config=[]):
|
||
pid_file = self.name_temp('vim.pid')
|
||
done_file = self.name_temp('loading_done')
|
||
if os.path.exists(done_file):
|
||
os.remove(done_file)
|
||
|
||
post_config = []
|
||
post_config.append('%s << EOF' % ('py3' if PYTHON3 else 'py'))
|
||
post_config.append('import vim')
|
||
post_config.append(
|
||
"with open('%s', 'w') as pid_file: pid_file.write(vim.eval('getpid()'))" %
|
||
pid_file)
|
||
post_config.append(
|
||
"with open('%s', 'w') as done_file: pass" %
|
||
done_file)
|
||
post_config.append('EOF')
|
||
|
||
config_path = self.write_temp('vim_config.vim',
|
||
textwrap.dedent(os.linesep.join(config + post_config) + '\n'))
|
||
|
||
# Note the space to exclude it from shell history.
|
||
self.send(""" vim -u %s\r\n""" % config_path)
|
||
wait_until_file_exists(done_file)
|
||
self._vim_pid = int(open(pid_file, 'r').read())
|
||
|
||
def leave_with_wait(self):
|
||
self.send(3 * ESC + ':qa!\n')
|
||
while is_process_running(self._vim_pid):
|
||
time.sleep(.05)
|
||
|
||
|
||
class VimInterfaceTmux(VimInterface):
|
||
|
||
def __init__(self, session):
|
||
VimInterface.__init__(self, 'Tmux')
|
||
self.session = session
|
||
self._check_version()
|
||
|
||
def send(self, s):
|
||
# I did not find any documentation on what needs escaping when sending
|
||
# to tmux, but it seems like this is all that is needed for now.
|
||
s = s.replace(';', r'\;')
|
||
|
||
if PYTHON3:
|
||
s = s.encode('utf-8')
|
||
silent_call(['tmux', 'send-keys', '-t', self.session, '-l', s])
|
||
|
||
def _check_version(self):
|
||
stdout, _ = subprocess.Popen(['tmux', '-V'],
|
||
stdout=subprocess.PIPE, stderr=subprocess.PIPE).communicate()
|
||
if PYTHON3:
|
||
stdout = stdout.decode('utf-8')
|
||
m = re.match(r"tmux (\d+).(\d+)", stdout)
|
||
if not m or not (int(m.group(1)), int(m.group(2))) >= (1, 8):
|
||
raise RuntimeError(
|
||
'Need at least tmux 1.8, you have %s.' %
|
||
stdout.strip())
|
||
|
||
|
||
class VimInterfaceWindows(VimInterface):
|
||
BRACES = re.compile('([}{])')
|
||
WIN_ESCAPES = ['+', '^', '%', '~', '[', ']', '<', '>', '(', ')']
|
||
WIN_REPLACES = [
|
||
(BS, '{BS}'),
|
||
(ARR_L, '{LEFT}'),
|
||
(ARR_R, '{RIGHT}'),
|
||
(ARR_U, '{UP}'),
|
||
(ARR_D, '{DOWN}'),
|
||
('\t', '{TAB}'),
|
||
('\n', '~'),
|
||
(ESC, '{ESC}'),
|
||
|
||
# On my system ` waits for a second keystroke, so `+SPACE = "`". On
|
||
# most systems, `+Space = "` ". I work around this, by sending the host
|
||
# ` as `+_+BS. Awkward, but the only way I found to get this working.
|
||
('`', '`_{BS}'),
|
||
('´', '´_{BS}'),
|
||
('{^}', '{^}_{BS}'),
|
||
]
|
||
|
||
def __init__(self):
|
||
self.seq_buf = []
|
||
# import windows specific modules
|
||
import win32com.client
|
||
import win32gui
|
||
self.win32gui = win32gui
|
||
self.shell = win32com.client.Dispatch('WScript.Shell')
|
||
|
||
def is_focused(self, title=None):
|
||
cur_title = self.win32gui.GetWindowText(
|
||
self.win32gui.GetForegroundWindow())
|
||
if (title or '- GVIM') in cur_title:
|
||
return True
|
||
return False
|
||
|
||
def focus(self, title=None):
|
||
if not self.shell.AppActivate(title or '- GVIM'):
|
||
raise Exception('Failed to switch to GVim window')
|
||
time.sleep(1)
|
||
|
||
def convert_keys(self, keys):
|
||
keys = self.BRACES.sub(r"{\1}", keys)
|
||
for k in self.WIN_ESCAPES:
|
||
keys = keys.replace(k, '{%s}' % k)
|
||
for f, r in self.WIN_REPLACES:
|
||
keys = keys.replace(f, r)
|
||
return keys
|
||
|
||
def send(self, keys):
|
||
self.seq_buf.append(keys)
|
||
seq = ''.join(self.seq_buf)
|
||
|
||
for f in SEQUENCES:
|
||
if f.startswith(seq) and f != seq:
|
||
return
|
||
self.seq_buf = []
|
||
|
||
seq = self.convert_keys(seq)
|
||
|
||
if not self.is_focused():
|
||
time.sleep(2)
|
||
self.focus()
|
||
if not self.is_focused():
|
||
# This is the only way I can find to stop test execution
|
||
raise KeyboardInterrupt('Failed to focus GVIM')
|
||
|
||
self.shell.SendKeys(seq)
|