Start adding tests

Tests are temporary disabled on all branches except tests (does not really work
though, needs travis.yml in all branches).
This commit is contained in:
ZyX 2013-02-24 14:03:14 +04:00 committed by Kim Silkebækken
parent 2a39ed83d8
commit 584755765b
9 changed files with 943 additions and 3 deletions

12
.travis.yml Normal file
View File

@ -0,0 +1,12 @@
language: python
python:
- "2.7"
- "3.2"
- "3.3"
install: "pip install ."
script: python setup.py test
branches:
only:
- tests
# vim: et

View File

@ -44,7 +44,7 @@ def keyvaluesplit(s):
raise ValueError('Option names must not start with `_\'')
idx = s.index('=')
o = s[:idx]
val = json.loads(s[idx+1:])
val = json.loads(s[idx + 1:])
return (o, val)

View File

@ -24,14 +24,14 @@ setup(
'scripts/powerline',
],
keywords='',
packages=find_packages(),
packages=find_packages(exclude=('tests',)),
include_package_data=True,
zip_safe=False,
test_suite='powerline',
install_requires=[],
extras_require={
'docs': [
'Sphinx',
],
},
test_suite='tests',
)

0
tests/__init__.py Normal file
View File

0
tests/empty Normal file
View File

View File

@ -0,0 +1,74 @@
# -*- coding: utf-8 -*-
'''Dynamic configuration files tests.'''
from unittest import TestCase
import tests.vim as vim_module
import sys
import os
import json
VBLOCK = chr(ord('V') - 0x40)
SBLOCK = chr(ord('S') - 0x40)
class TestConfig(TestCase):
def test_vim(self):
from powerline.vim import VimPowerline
cfg_path = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'powerline', 'config_files')
vim_module._g['powerline_config_path'] = cfg_path
buffers = ((lambda: vim_module._set_bufoption('buftype', 'help'), lambda: vim_module._set_bufoption('buftype', '')),
(lambda: vim_module._edit('[Command Line]'), lambda: vim_module._bw()))
with open(os.path.join(cfg_path, 'config.json'), 'r') as f:
self.assertEqual(len(buffers), len(json.load(f)['ext']['vim']['local_themes']))
outputs = {}
i = 0
mode = None
powerline = VimPowerline()
def check_output(*args):
out = powerline.renderer.render(*args + (0 if mode == 'nc' else 1,))
if out in outputs:
self.fail('Duplicate in set #{0} for mode {1!r} (previously defined in set #{2} for mode {3!r})'.format(i, mode, *outputs[out]))
outputs[out] = (i, mode)
try:
exclude = set(('no', 'v', 'V', VBLOCK, 's', 'S', SBLOCK, 'R', 'Rv', 'c', 'cv', 'ce', 'r', 'rm', 'r?', '!'))
try:
for mode in ['n', 'nc', 'no', 'v', 'V', VBLOCK, 's', 'S', SBLOCK, 'i', 'R', 'Rv', 'c', 'cv', 'ce', 'r', 'rm', 'r?', '!']:
if mode != 'nc':
vim_module._start_mode(mode)
check_output(1, 0)
for setup, teardown in buffers:
i += 1
if mode in exclude:
continue
setup()
try:
check_output(1, 0)
finally:
teardown()
finally:
vim_module._start_mode('n')
finally:
vim_module._g.pop('powerline_config_path')
old_cwd = None
def setUpModule():
global old_cwd
old_cwd = os.getcwd()
sys.modules['vim'] = vim_module._get_module()
from powerline.segments import vim
globals()['vim'] = vim
def tearDownModule():
global old_cwd
sys.modules.pop('vim')
os.chdir(old_cwd)
old_cwd = None

106
tests/test_lib.py Normal file
View File

@ -0,0 +1,106 @@
from powerline.lib import mergedicts, underscore_to_camelcase, add_divider_highlight_group, humanize_bytes
from powerline.lib.vcs import guess
from unittest import TestCase
from subprocess import call, PIPE
import os
import sys
class TestLib(TestCase):
def test_underscore_to_camelcase(self):
self.assertEqual(underscore_to_camelcase('abc_def_ghi'), 'AbcDefGhi')
self.assertEqual(underscore_to_camelcase('abc_def__ghi'), 'AbcDef_Ghi')
def test_mergedicts(self):
d = {}
mergedicts(d, {'abc': {'def': 'ghi'}})
self.assertEqual(d, {'abc': {'def': 'ghi'}})
mergedicts(d, {'abc': {'def': {'ghi': 'jkl'}}})
self.assertEqual(d, {'abc': {'def': {'ghi': 'jkl'}}})
mergedicts(d, {})
self.assertEqual(d, {'abc': {'def': {'ghi': 'jkl'}}})
mergedicts(d, {'abc': {'mno': 'pqr'}})
self.assertEqual(d, {'abc': {'def': {'ghi': 'jkl'}, 'mno': 'pqr'}})
def test_add_divider_highlight_group(self):
def decorated_function_name(**kwargs):
return str(kwargs)
func = add_divider_highlight_group('hl_group')(decorated_function_name)
self.assertEqual(func.__name__, 'decorated_function_name')
self.assertEqual(func(kw={}), [{'contents': repr({'kw': {}}), 'divider_highlight_group': 'hl_group'}])
def test_humanize_bytes(self):
self.assertEqual(humanize_bytes(0), '0 B')
self.assertEqual(humanize_bytes(1), '1 B')
self.assertEqual(humanize_bytes(1, suffix='bit'), '1 bit')
self.assertEqual(humanize_bytes(1000, si_prefix=True), '1 kB')
self.assertEqual(humanize_bytes(1024, si_prefix=True), '1 kB')
self.assertEqual(humanize_bytes(1000000000, si_prefix=True), '1.00 GB')
self.assertEqual(humanize_bytes(1000000000, si_prefix=False), '953.7 MiB')
use_mercurial = sys.version_info < (3, 0)
class TestVCS(TestCase):
def test_git(self):
repo = guess(path='git_repo')
self.assertNotEqual(repo, None)
self.assertEqual(repo.branch(), 'master')
self.assertEqual(repo.status(), ' ')
self.assertEqual(repo.status('file'), None)
with open(os.path.join('git_repo', 'file'), 'w') as f:
f.write('abc')
f.flush()
self.assertEqual(repo.status(), ' U')
self.assertEqual(repo.status('file'), '??')
call(['git', 'add', '.'], cwd='git_repo')
self.assertEqual(repo.status(), ' I ')
self.assertEqual(repo.status('file'), 'A ')
f.write('def')
f.flush()
self.assertEqual(repo.status(), 'DI ')
self.assertEqual(repo.status('file'), 'AM')
os.remove(os.path.join('git_repo', 'file'))
if use_mercurial:
def test_mercurial(self):
repo = guess(path='hg_repo')
self.assertNotEqual(repo, None)
self.assertEqual(repo.branch(), 'default')
with open(os.path.join('hg_repo', 'file'), 'w') as f:
f.write('abc')
f.flush()
self.assertEqual(repo.status(), ' U')
self.assertEqual(repo.status('file'), 'U')
call(['hg', 'add', '.'], cwd='hg_repo', stdout=PIPE)
self.assertEqual(repo.status(), 'D ')
self.assertEqual(repo.status('file'), 'A')
os.remove(os.path.join('hg_repo', 'file'))
def setUpModule():
global old_cwd
old_cwd = os.getcwd()
os.chdir(os.path.dirname(__file__))
call(['git', 'init', '--quiet', 'git_repo'])
call(['git', 'config', '--local', 'user.name', 'Foo'], cwd='git_repo')
call(['git', 'config', '--local', 'user.email', 'bar@example.org'], cwd='git_repo')
call(['git', 'commit', '--allow-empty', '--message', 'Initial commit', '--quiet'], cwd='git_repo')
if use_mercurial:
call(['hg', 'init', 'hg_repo'])
with open(os.path.join('hg_repo', '.hg', 'hgrc'), 'w') as hgrc:
hgrc.write('[ui]\n')
hgrc.write('username = Foo <bar@example.org>\n')
def tearDownModule():
global old_cwd
for repo_dir in ['git_repo'] + (['hg_repo'] if use_mercurial else []):
for root, dirs, files in list(os.walk(repo_dir, topdown=False)):
for file in files:
os.remove(os.path.join(root, file))
for dir in dirs:
os.rmdir(os.path.join(root, dir))
os.rmdir(repo_dir)
os.chdir(old_cwd) # NOQA

381
tests/test_segments.py Normal file
View File

@ -0,0 +1,381 @@
# -*- coding: utf-8 -*-
from unittest import TestCase
from powerline.segments import shell, common
import tests.vim as vim_module
import sys
import os
import imp
class Args(object):
theme_option = None
config = None
config_path = None
ext = ['shell']
renderer_module = None
def __init__(self, **kwargs):
self.__dict__.update(kwargs)
vim = None
class TestShell(TestCase):
def test_last_status(self):
self.assertEqual(shell.last_status(Args(last_exit_code=10)),
[{'contents': '10', 'highlight_group': 'exit_fail'}])
self.assertEqual(shell.last_status(Args(last_exit_code=None)), None)
def test_last_pipe_status(self):
self.assertEqual(shell.last_pipe_status(Args(last_pipe_status=[])), None)
self.assertEqual(shell.last_pipe_status(Args(last_pipe_status=[0, 0, 0])), None)
self.assertEqual(shell.last_pipe_status(Args(last_pipe_status=[0, 2, 0])),
[{'contents': '0', 'highlight_group': 'exit_success'},
{'contents': '2', 'highlight_group': 'exit_fail'},
{'contents': '0', 'highlight_group': 'exit_success'}])
class TestCommon(TestCase):
def test_hostname(self):
os.environ['SSH_CLIENT'] = '192.168.0.12 40921 22'
socket = imp.new_module('socket')
socket.gethostname = lambda: 'abc'
sys.modules['socket'] = socket
try:
self.assertEqual(common.hostname(), 'abc')
self.assertEqual(common.hostname(only_if_ssh=True), 'abc')
os.environ.pop('SSH_CLIENT')
self.assertEqual(common.hostname(), 'abc')
self.assertEqual(common.hostname(only_if_ssh=True), None)
finally:
sys.modules.pop('socket')
def test_user(self):
new_os = imp.new_module('os')
new_os.environ = {'USER': 'def'}
common.os = new_os
try:
self.assertEqual(common.user(), [{'contents': 'def', 'highlight_group': 'user'}])
new_os.geteuid = lambda: 1
self.assertEqual(common.user(), [{'contents': 'def', 'highlight_group': 'user'}])
new_os.geteuid = lambda: 0
self.assertEqual(common.user(), [{'contents': 'def', 'highlight_group': ['superuser', 'user']}])
finally:
common.os = os
def test_branch(self):
vcslib = imp.new_module('powerline.lib.vcs')
vcslib.guess = lambda path: Args(branch=lambda: os.path.basename(path))
sys.modules['powerline.lib.vcs'] = vcslib
try:
self.assertEqual(common.branch(), 'tests')
vcslib.guess = lambda path: None
self.assertEqual(common.branch(), None)
finally:
sys.modules.pop('powerline.lib.vcs')
def test_cwd(self):
new_os = imp.new_module('os')
new_os.path = os.path
new_os.environ = {}
new_os.sep = '/'
new_os.getcwd = lambda: '/abc/def/ghi/foo/bar'
common.os = new_os
try:
self.assertEqual(common.cwd(),
[{'contents': '/', 'divider_highlight_group': 'cwd:divider'},
{'contents': 'abc', 'divider_highlight_group': 'cwd:divider'},
{'contents': 'def', 'divider_highlight_group': 'cwd:divider'},
{'contents': 'ghi', 'divider_highlight_group': 'cwd:divider'},
{'contents': 'foo', 'divider_highlight_group': 'cwd:divider'},
{'contents': 'bar', 'divider_highlight_group': 'cwd:divider', 'highlight_group': ['cwd:current_folder', 'cwd']}])
new_os.getcwdu = lambda: '/abc/def/ghi/foo/bar'
self.assertEqual(common.cwd(),
[{'contents': '/', 'divider_highlight_group': 'cwd:divider'},
{'contents': 'abc', 'divider_highlight_group': 'cwd:divider'},
{'contents': 'def', 'divider_highlight_group': 'cwd:divider'},
{'contents': 'ghi', 'divider_highlight_group': 'cwd:divider'},
{'contents': 'foo', 'divider_highlight_group': 'cwd:divider'},
{'contents': 'bar', 'divider_highlight_group': 'cwd:divider', 'highlight_group': ['cwd:current_folder', 'cwd']}])
new_os.environ['HOME'] = '/abc/def/ghi'
self.assertEqual(common.cwd(),
[{'contents': '~', 'divider_highlight_group': 'cwd:divider'},
{'contents': 'foo', 'divider_highlight_group': 'cwd:divider'},
{'contents': 'bar', 'divider_highlight_group': 'cwd:divider', 'highlight_group': ['cwd:current_folder', 'cwd']}])
self.assertEqual(common.cwd(dir_limit_depth=3),
[{'contents': '~', 'divider_highlight_group': 'cwd:divider'},
{'contents': 'foo', 'divider_highlight_group': 'cwd:divider'},
{'contents': 'bar', 'divider_highlight_group': 'cwd:divider', 'highlight_group': ['cwd:current_folder', 'cwd']}])
self.assertEqual(common.cwd(dir_limit_depth=1),
[{'contents': '', 'divider_highlight_group': 'cwd:divider'},
{'contents': 'bar', 'divider_highlight_group': 'cwd:divider', 'highlight_group': ['cwd:current_folder', 'cwd']}])
self.assertEqual(common.cwd(dir_limit_depth=2, dir_shorten_len=2),
[{'contents': '~', 'divider_highlight_group': 'cwd:divider'},
{'contents': 'fo', 'divider_highlight_group': 'cwd:divider'},
{'contents': 'bar', 'divider_highlight_group': 'cwd:divider', 'highlight_group': ['cwd:current_folder', 'cwd']}])
ose = OSError()
ose.errno = 2
def raises(exc):
raise exc
new_os.getcwdu = lambda: raises(ose)
self.assertEqual(common.cwd(dir_limit_depth=2, dir_shorten_len=2),
[{'contents': '[not found]', 'divider_highlight_group': 'cwd:divider', 'highlight_group': ['cwd:current_folder', 'cwd']}])
new_os.getcwdu = lambda: raises(OSError())
with self.assertRaises(OSError):
common.cwd(dir_limit_depth=2, dir_shorten_len=2),
new_os.getcwdu = lambda: raises(ValueError())
with self.assertRaises(ValueError):
common.cwd(dir_limit_depth=2, dir_shorten_len=2),
finally:
common.os = os
def test_date(self):
datetime = imp.new_module('datetime')
datetime.datetime = Args(now=lambda: Args(strftime=lambda fmt: fmt))
sys.modules['datetime'] = datetime
try:
self.assertEqual(common.date(), [{'contents': '%Y-%m-%d', 'highlight_group': ['date'], 'divider_highlight_group': None}])
self.assertEqual(common.date(format='%H:%M', istime=True), [{'contents': '%H:%M', 'highlight_group': ['time', 'date'], 'divider_highlight_group': 'time:divider'}])
finally:
sys.modules.pop('datetime')
def test_fuzzy_time(self):
datetime = imp.new_module('datetime')
time = Args(hour=0, minute=45)
datetime.datetime = Args(now=lambda: time)
sys.modules['datetime'] = datetime
try:
self.assertEqual(common.fuzzy_time(), 'quarter to one')
time.hour = 23
time.minute = 59
self.assertEqual(common.fuzzy_time(), 'round about midnight')
time.minute = 33
self.assertEqual(common.fuzzy_time(), 'twenty-five to twelve')
time.minute = 60
self.assertEqual(common.fuzzy_time(), 'twelve o\'clock')
finally:
sys.modules.pop('datetime')
def test_external_ip(self):
old_urllib_read = common.urllib_read
common.urllib_read = lambda url: '127.0.0.1'
try:
self.assertEqual(common.external_ip(), [{'contents': '127.0.0.1', 'divider_highlight_group': 'background:divider'}])
finally:
common.urllib_read = old_urllib_read
def test_uptime(self):
# TODO
pass
def test_weather(self):
# TODO
pass
def test_system_load(self):
new_os = imp.new_module('os')
new_os.getloadavg = lambda: (7.5, 3.5, 1.5)
multiprocessing = imp.new_module('multiprocessing')
multiprocessing.cpu_count = lambda: 2
common.os = new_os
sys.modules['multiprocessing'] = multiprocessing
try:
self.assertEqual(common.system_load(),
[{'contents': '7.5 ', 'highlight_group': ['system_load_ugly', 'system_load'], 'draw_divider': True, 'divider_highlight_group': 'background:divider'},
{'contents': '3.5 ', 'highlight_group': ['system_load_bad', 'system_load'], 'draw_divider': False, 'divider_highlight_group': 'background:divider'},
{'contents': '1.5', 'highlight_group': ['system_load_good', 'system_load'], 'draw_divider': False, 'divider_highlight_group': 'background:divider'}])
self.assertEqual(common.system_load(format='{avg:.0f}', threshold_good=0, threshold_bad=1),
[{'contents': '8 ', 'highlight_group': ['system_load_ugly', 'system_load'], 'draw_divider': True, 'divider_highlight_group': 'background:divider'},
{'contents': '4 ', 'highlight_group': ['system_load_ugly', 'system_load'], 'draw_divider': False, 'divider_highlight_group': 'background:divider'},
{'contents': '2', 'highlight_group': ['system_load_bad', 'system_load'], 'draw_divider': False, 'divider_highlight_group': 'background:divider'}])
finally:
common.os = os
sys.modules.pop('multiprocessing')
def test_cpu_load_percent(self):
psutil = imp.new_module('psutil')
psutil.cpu_percent = lambda **kwargs: 52.3
sys.modules['psutil'] = psutil
try:
self.assertEqual(common.cpu_load_percent(), '52%')
finally:
sys.modules.pop('psutil')
def test_network_load(self):
# TODO
pass
def test_virtualenv(self):
os.environ['VIRTUAL_ENV'] = '/abc/def/ghi'
self.assertEqual(common.virtualenv(), 'ghi')
os.environ.pop('VIRTUAL_ENV')
self.assertEqual(common.virtualenv(), None)
def test_email_imap_alert(self):
# TODO
pass
def test_now_playing(self):
# TODO
pass
class TestVim(TestCase):
def test_mode(self):
segment_info = vim_module._get_segment_info()
self.assertEqual(vim.mode(segment_info=segment_info), 'NORMAL')
self.assertEqual(vim.mode(segment_info=segment_info, override={'i': 'INS'}), 'NORMAL')
self.assertEqual(vim.mode(segment_info=segment_info, override={'n': 'NORM'}), 'NORM')
try:
vim_module._start_mode('i')
segment_info = vim_module._get_segment_info()
self.assertEqual(vim.mode(segment_info=segment_info), 'INSERT')
vim_module._start_mode(chr(ord('V') - 0x40))
segment_info = vim_module._get_segment_info()
self.assertEqual(vim.mode(segment_info=segment_info), 'V·BLCK')
self.assertEqual(vim.mode(segment_info=segment_info, override={'^V': 'VBLK'}), 'VBLK')
finally:
vim_module._start_mode('n')
def test_modified_indicator(self):
segment_info = vim_module._get_segment_info()
self.assertEqual(vim.modified_indicator(segment_info=segment_info), None)
segment_info['buffer'][0] = 'abc'
try:
self.assertEqual(vim.modified_indicator(segment_info=segment_info), '+')
self.assertEqual(vim.modified_indicator(segment_info=segment_info, text='-'), '-')
finally:
vim_module._undo()
def test_paste_indicator(self):
segment_info = vim_module._get_segment_info()
self.assertEqual(vim.paste_indicator(segment_info=segment_info), None)
vim_module._options['paste'] = 1
try:
self.assertEqual(vim.paste_indicator(segment_info=segment_info), 'PASTE')
self.assertEqual(vim.paste_indicator(segment_info=segment_info, text='P'), 'P')
finally:
vim_module._options['paste'] = 0
def test_readonly_indicator(self):
segment_info = vim_module._get_segment_info()
self.assertEqual(vim.readonly_indicator(segment_info=segment_info), None)
vim_module._buf_options[vim_module._buffer()]['readonly'] = 1
try:
self.assertEqual(vim.readonly_indicator(segment_info=segment_info), '')
self.assertEqual(vim.readonly_indicator(segment_info=segment_info, text='L'), 'L')
finally:
vim_module._buf_options[vim_module._buffer()]['readonly'] = 0
def test_file_directory(self):
segment_info = vim_module._get_segment_info()
self.assertEqual(vim.file_directory(segment_info=segment_info), None)
os.environ['HOME'] = '/home/foo'
vim_module._edit('/tmp/abc')
segment_info = vim_module._get_segment_info()
try:
self.assertEqual(vim.file_directory(segment_info=segment_info), '/tmp/')
os.environ['HOME'] = '/tmp'
self.assertEqual(vim.file_directory(segment_info=segment_info), '~/')
finally:
vim_module._bw(segment_info['bufnr'])
def test_file_name(self):
segment_info = vim_module._get_segment_info()
self.assertEqual(vim.file_name(segment_info=segment_info), None)
self.assertEqual(vim.file_name(segment_info=segment_info, display_no_file=True),
[{'contents': '[No file]', 'highlight_group': ['file_name_no_file', 'file_name']}])
self.assertEqual(vim.file_name(segment_info=segment_info, display_no_file=True, no_file_text='X'),
[{'contents': 'X', 'highlight_group': ['file_name_no_file', 'file_name']}])
vim_module._edit('/tmp/abc')
segment_info = vim_module._get_segment_info()
try:
self.assertEqual(vim.file_name(segment_info=segment_info), 'abc')
finally:
vim_module._bw(segment_info['bufnr'])
vim_module._edit('/tmp/')
segment_info = vim_module._get_segment_info()
try:
self.assertEqual(vim.file_name(segment_info=segment_info), '')
finally:
vim_module._bw(segment_info['bufnr'])
def test_file_size(self):
segment_info = vim_module._get_segment_info()
self.assertEqual(vim.file_size(segment_info=segment_info), None)
vim_module._edit(os.path.join(os.path.dirname(__file__), 'empty'))
segment_info = vim_module._get_segment_info()
try:
self.assertEqual(vim.file_size(segment_info=segment_info), '0 B')
finally:
vim_module._bw(segment_info['bufnr'])
def test_file_opts(self):
segment_info = vim_module._get_segment_info()
self.assertEqual(vim.file_format(segment_info=segment_info),
[{'divider_highlight_group': 'background:divider', 'contents': 'unix'}])
self.assertEqual(vim.file_encoding(segment_info=segment_info),
[{'divider_highlight_group': 'background:divider', 'contents': 'utf-8'}])
self.assertEqual(vim.file_type(segment_info=segment_info), None)
vim_module._set_filetype('python')
try:
self.assertEqual(vim.file_type(segment_info=segment_info),
[{'divider_highlight_group': 'background:divider', 'contents': 'python'}])
finally:
vim_module._set_filetype('')
def test_line_percent(self):
segment_info = vim_module._get_segment_info()
segment_info['buffer'][0:-1] = [str(i) for i in range(100)]
try:
self.assertEqual(vim.line_percent(segment_info=segment_info), '0')
vim_module._set_cursor(50, 0)
self.assertEqual(vim.line_percent(segment_info=segment_info), '49')
self.assertEqual(vim.line_percent(segment_info=segment_info, gradient=True),
[{'contents': '49', 'highlight_group': ['line_percent_gradient', 'line_percent'], 'gradient_level': 49}])
finally:
vim_module._bw(segment_info['bufnr'])
def test_cursor_current(self):
segment_info = vim_module._get_segment_info()
self.assertEqual(vim.line_current(segment_info=segment_info), '1')
self.assertEqual(vim.col_current(segment_info=segment_info), '1')
self.assertEqual(vim.virtcol_current(segment_info=segment_info),
[{'highlight_group': ['virtcol_current', 'col_current'], 'contents': '1'}])
def test_modified_buffers(self):
self.assertEqual(vim.modified_buffers(), None)
def test_branch(self):
# TODO
pass
def test_file_vcs_status(self):
# TODO
pass
def test_repository_status(self):
# TODO
pass
old_cwd = None
def setUpModule():
global old_cwd
old_cwd = os.getcwd()
os.chdir(os.path.dirname(__file__))
sys.modules['vim'] = vim_module._get_module()
from powerline.segments import vim
globals()['vim'] = vim
def tearDownModule():
global old_cwd
sys.modules.pop('vim')
os.chdir(old_cwd)

367
tests/vim.py Normal file
View File

@ -0,0 +1,367 @@
_log = []
_g = {}
_window = 0
_mode = 'n'
_buf_purge_events = set()
_options = {
'paste': 0,
}
_last_bufnr = 0
_highlights = {}
buffers = {}
windows = []
def _buffer():
return windows[_window - 1].buffer.number
def _logged(func):
from functools import wraps
@wraps(func)
def f(*args):
_log.append((func.__name__, args))
return func(*args)
return f
@_logged
def command(cmd):
if cmd.startswith('let g:'):
import re
varname, value = re.compile(r'^let g:(\w+)\s*=\s*(.*)').match(cmd).groups()
_g[varname] = value
elif cmd.startswith('hi '):
sp = cmd.split()
_highlights[sp[1]] = sp[2:]
else:
raise NotImplementedError
@_logged
def eval(expr):
if expr.startswith('g:'):
return _g[expr[2:]]
elif expr.startswith('&'):
return _options[expr[1:]]
elif expr.startswith('PowerlineRegisterCachePurgerEvent'):
_buf_purge_events.add(expr[expr.find('"') + 1:expr.rfind('"') - 1])
return "0"
raise NotImplementedError
@_logged
def bindeval(expr):
if expr == 'g:':
return _g
import re
match = re.compile(r'^function\("([^"\\]+)"\)$').match(expr)
if match:
return globals()['_emul_' + match.group(1)]
else:
raise NotImplementedError
@_logged
def _emul_mode(*args):
if args and args[0]:
return _mode
else:
return _mode[0]
@_logged
def _emul_getbufvar(bufnr, varname):
if varname[0] == '&':
if bufnr not in _buf_options:
return ''
try:
return _buf_options[bufnr][varname[1:]]
except KeyError:
try:
return _options[varname[1:]]
except KeyError:
return ''
raise NotImplementedError
@_logged
def _emul_getwinvar(winnr, varname):
return _win_scopes[winnr][varname]
@_logged
def _emul_setwinvar(winnr, varname, value):
_win_scopes[winnr][varname] = value
@_logged
def _emul_virtcol(expr):
if expr == '.':
return windows[_window - 1].cursor[1] + 1
raise NotImplementedError
@_logged
def _emul_fnamemodify(path, modstring):
import os
_modifiers = {
'~': lambda path: path.replace(os.environ['HOME'], '~') if path.startswith(os.environ['HOME']) else path,
'.': lambda path: (lambda tpath: path if tpath[:3] == '..' + os.sep else tpath)(os.path.relpath(path)),
't': lambda path: os.path.basename(path),
'h': lambda path: os.path.dirname(path),
}
for mods in modstring.split(':')[1:]:
path = _modifiers[mods](path)
return path
@_logged
def _emul_expand(expr):
if expr == '<abuf>':
return _buffer()
raise NotImplementedError
@_logged
def _emul_bufnr(expr):
if expr == '$':
return _last_bufnr
raise NotImplementedError
@_logged
def _emul_exists(varname):
if varname.startswith('g:'):
return varname[2:] in _g
raise NotImplementedError
_window_ids = [None]
_window_id = 0
_win_scopes = [None]
_win_options = [None]
class _Window(object):
def __init__(self, buffer=None, cursor=(1, 0), width=80):
global _window_id
self.cursor = cursor
self.width = width
if buffer:
if type(buffer) is _Buffer:
self.buffer = buffer
else:
self.buffer = _Buffer(**buffer)
else:
self.buffer = _Buffer()
windows.append(self)
_window_id += 1
_window_ids.append(_window_id)
_win_scopes.append({})
_win_options.append({})
def __repr__(self):
return '<window ' + str(windows.index(self)) + '>'
_buf_scopes = {}
_buf_options = {}
_buf_lines = {}
_undostate = {}
_undo_written = {}
class _Buffer(object):
def __init__(self, name=None):
global _last_bufnr
import os
_last_bufnr += 1
bufnr = _last_bufnr
self.number = bufnr
self.name = os.path.abspath(name) if name else None
_buf_scopes[bufnr] = {}
_buf_options[bufnr] = {
'modified': 0,
'readonly': 0,
'fileformat': 'unix',
'filetype': '',
'buftype': '',
'fileencoding': 'utf-8',
}
_buf_lines[bufnr] = ['']
from copy import copy
_undostate[bufnr] = [copy(_buf_lines[bufnr])]
_undo_written[bufnr] = len(_undostate[bufnr])
buffers[bufnr] = self
def __getitem__(self, line):
return _buf_lines[self.number][line]
def __setitem__(self, line, value):
_buf_options[self.number]['modified'] = 1
_buf_lines[self.number][line] = value
from copy import copy
_undostate[self.number].append(copy(_buf_lines[self.number]))
def __setslice__(self, *args):
_buf_options[self.number]['modified'] = 1
_buf_lines[self.number].__setslice__(*args)
from copy import copy
_undostate[self.number].append(copy(_buf_lines[self.number]))
def __getslice__(self, *args):
return _buf_lines[self.number].__getslice__(*args)
def __len__(self):
return len(_buf_lines[self.number])
def __repr__(self):
return '<buffer ' + str(self.name) + '>'
def __del__(self):
bufnr = self.number
if _buf_options:
_buf_options.pop(bufnr)
_buf_lines.pop(bufnr)
_undostate.pop(bufnr)
_undo_written.pop(bufnr)
_buf_scopes.pop(bufnr)
_module = None
def _get_module():
global _module
if _module:
return _module
import imp
_module = imp.new_module('vim')
for varname, value in globals().items():
if varname[0] != '_':
setattr(_module, varname, value)
_new()
return _module
def _get_segment_info():
mode_translations = {
chr(ord('V') - 0x40): '^V',
chr(ord('S') - 0x40): '^S',
}
mode = _mode
mode = mode_translations.get(mode, mode)
return {
'window': windows[_window - 1],
'buffer': buffers[_buffer()],
'bufnr': _buffer(),
'window_id': _window_ids[_window],
'mode': mode,
}
def _launch_event(event):
pass
def _start_mode(mode):
global _mode
if mode == 'i':
_launch_event('InsertEnter')
elif _mode == 'i':
_launch_event('InsertLeave')
_mode = mode
def _undo():
if len(_undostate[_buffer()]) == 1:
return
_undostate[_buffer()].pop(-1)
_buf_lines[_buffer()] = _undostate[_buffer()][-1]
if _undo_written[_buffer()] == len(_undostate[_buffer()]):
_buf_options[_buffer()]['modified'] = 0
def _edit(name=None):
global _last_bufnr
if _buffer() and buffers[_buffer()].name is None:
buf = buffers[_buffer()]
buf.name = name
else:
buf = _Buffer(name)
windows[_window - 1].buffer = buf
def _new(name=None):
global _window
_Window(buffer={'name': name})
_window = len(windows)
def _del_window(winnr):
win = windows.pop(winnr - 1)
_win_scopes.pop(winnr)
_win_options.pop(winnr)
_window_ids.pop(winnr)
return win
def _close(winnr, wipe=True):
global _window
win = _del_window(winnr)
if _window == winnr:
_window = len(windows)
if wipe:
for w in windows:
if w.buffer.number == win.buffer.number:
break
else:
_bw(win.buffer.number)
if not windows:
_Window()
def _bw(bufnr=None):
bufnr = bufnr or _buffer()
winnr = 1
for win in windows:
if win.buffer.number == bufnr:
_close(winnr, wipe=False)
winnr += 1
buffers.pop(bufnr)
if not buffers:
_Buffer()
_b(max(buffers.keys()))
def _b(bufnr):
windows[_window - 1].buffer = buffers[bufnr]
def _set_filetype(ft):
_buf_options[_buffer()]['filetype'] = ft
_launch_event('FileType')
def _set_cursor(line, col):
windows[_window - 1].cursor = (line, col)
if _mode == 'n':
_launch_event('CursorMoved')
elif _mode == 'i':
_launch_event('CursorMovedI')
def _get_buffer():
return buffers[_buffer()]
def _set_bufoption(option, value, bufnr=None):
_buf_options[bufnr or _buffer()][option] = value