summaryrefslogtreecommitdiffstats
path: root/tests/test_python
diff options
context:
space:
mode:
Diffstat (limited to '')
-rw-r--r--tests/test_python/empty0
-rwxr-xr-xtests/test_python/test.sh13
-rw-r--r--tests/test_python/test_cmdline.py149
-rw-r--r--tests/test_python/test_config_merging.py270
-rw-r--r--tests/test_python/test_config_reload.py319
-rw-r--r--tests/test_python/test_configuration.py877
-rw-r--r--tests/test_python/test_lib.py733
-rw-r--r--tests/test_python/test_lib_config.py52
-rw-r--r--tests/test_python/test_listers.py159
-rw-r--r--tests/test_python/test_logging.py467
-rw-r--r--tests/test_python/test_provided_config_files.py201
-rw-r--r--tests/test_python/test_segments.py1846
-rw-r--r--tests/test_python/test_selectors.py36
-rw-r--r--tests/test_python/test_watcher.py245
14 files changed, 5367 insertions, 0 deletions
diff --git a/tests/test_python/empty b/tests/test_python/empty
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/tests/test_python/empty
diff --git a/tests/test_python/test.sh b/tests/test_python/test.sh
new file mode 100755
index 0000000..f042237
--- /dev/null
+++ b/tests/test_python/test.sh
@@ -0,0 +1,13 @@
+#!/bin/sh
+. tests/shlib/common.sh
+
+enter_suite python final
+
+for file in "$ROOT"/tests/test_python/test_*.py ; do
+ test_name="${file##*/test_}"
+ if ! "$PYTHON" "$file" --verbose --catch ; then
+ fail "${test_name%.py}" F "Failed test(s) from $file"
+ fi
+done
+
+exit_suite
diff --git a/tests/test_python/test_cmdline.py b/tests/test_python/test_cmdline.py
new file mode 100644
index 0000000..470a7b4
--- /dev/null
+++ b/tests/test_python/test_cmdline.py
@@ -0,0 +1,149 @@
+# vim:fileencoding=utf-8:noet
+
+'''Tests for shell.py parser'''
+
+from __future__ import (unicode_literals, division, absolute_import, print_function)
+
+import sys
+
+if sys.version_info < (3,):
+ from io import BytesIO as StrIO
+else:
+ from io import StringIO as StrIO
+
+from powerline.commands.main import get_argparser, finish_args
+
+from tests.modules import TestCase
+from tests.modules.lib import replace_attr
+
+
+class TestParser(TestCase):
+ def test_main_err(self):
+ parser = get_argparser()
+ out = StrIO()
+ err = StrIO()
+
+ def flush():
+ out.truncate(0)
+ err.truncate(0)
+
+ with replace_attr(sys, 'stdout', out, 'stderr', err):
+ for raising_args, raising_reg in [
+ ([], 'too few arguments|the following arguments are required: ext'),
+ (['-r'], 'expected one argument'),
+ (['shell', '-r'], 'expected one argument'),
+ (['shell', '-w'], 'expected one argument'),
+ (['shell', '-c'], 'expected one argument'),
+ (['shell', '-t'], 'expected one argument'),
+ (['shell', '-p'], 'expected one argument'),
+ (['shell', '-R'], 'expected one argument'),
+ (['shell', '--renderer-module'], 'expected one argument'),
+ (['shell', '--width'], 'expected one argument'),
+ (['shell', '--last-exit-code'], 'expected one argument'),
+ (['shell', '--last-pipe-status'], 'expected one argument'),
+ (['shell', '--config-override'], 'expected one argument'),
+ (['shell', '--theme-override'], 'expected one argument'),
+ (['shell', '--config-path'], 'expected one argument'),
+ (['shell', '--renderer-arg'], 'expected one argument'),
+ (['shell', '--jobnum'], 'expected one argument'),
+ (['-r', '.zsh'], 'too few arguments|the following arguments are required: ext'),
+ (['shell', '--last-exit-code', 'i'], 'invalid int_or_sig value'),
+ (['shell', '--last-pipe-status', '1 i'], 'invalid <lambda> value'),
+ ]:
+ self.assertRaises(SystemExit, parser.parse_args, raising_args)
+ self.assertFalse(out.getvalue())
+ self.assertRegexpMatches(err.getvalue(), raising_reg)
+ flush()
+
+ def test_main_normal(self):
+ parser = get_argparser()
+ out = StrIO()
+ err = StrIO()
+ with replace_attr(sys, 'stdout', out, 'stderr', err):
+ for argv, expargs in [
+ (['shell', 'left'], {'ext': ['shell'], 'side': 'left'}),
+ (['shell', 'left', '-r', '.zsh'], {'ext': ['shell'], 'renderer_module': '.zsh', 'side': 'left'}),
+ ([
+ 'shell',
+ 'left',
+ '-r', '.zsh',
+ '--last-exit-code', '10',
+ '--last-pipe-status', '10 20 30',
+ '--jobnum=10',
+ '-w', '100',
+ '-c', 'common.term_truecolor=true',
+ '-c', 'common.spaces=4',
+ '-t', 'default.segment_data.hostname.before=H:',
+ '-p', '.',
+ '-p', '..',
+ '-R', 'smth={"abc":"def"}',
+ ], {
+ 'ext': ['shell'],
+ 'side': 'left',
+ 'renderer_module': '.zsh',
+ 'last_exit_code': 10,
+ 'last_pipe_status': [10, 20, 30],
+ 'jobnum': 10,
+ 'width': 100,
+ 'config_override': {'common': {'term_truecolor': True, 'spaces': 4}},
+ 'theme_override': {
+ 'default': {
+ 'segment_data': {
+ 'hostname': {
+ 'before': 'H:'
+ }
+ }
+ }
+ },
+ 'config_path': ['.', '..'],
+ 'renderer_arg': {'smth': {'abc': 'def'}},
+ }),
+ (['shell', 'left', '-R', 'arg=true'], {
+ 'ext': ['shell'],
+ 'side': 'left',
+ 'renderer_arg': {'arg': True},
+ }),
+ (['shell', 'left', '-R', 'arg=true', '-R', 'arg='], {
+ 'ext': ['shell'],
+ 'side': 'left',
+ 'renderer_arg': {},
+ }),
+ (['shell', 'left', '-R', 'arg='], {'ext': ['shell'], 'renderer_arg': {}, 'side': 'left'}),
+ (['shell', 'left', '-t', 'default.segment_info={"hostname": {}}'], {
+ 'ext': ['shell'],
+ 'side': 'left',
+ 'theme_override': {
+ 'default': {
+ 'segment_info': {
+ 'hostname': {}
+ }
+ }
+ },
+ }),
+ (['shell', 'left', '-c', 'common={ }'], {
+ 'ext': ['shell'],
+ 'side': 'left',
+ 'config_override': {'common': {}},
+ }),
+ (['shell', 'left', '--last-pipe-status='], {
+ 'ext': ['shell'],
+ 'side': 'left',
+ 'last_pipe_status': [],
+ }),
+ ]:
+ args = parser.parse_args(argv)
+ finish_args(parser, {}, args)
+ for key, val in expargs.items():
+ self.assertEqual(getattr(args, key), val)
+ for key, val in args.__dict__.items():
+ if key not in expargs:
+ self.assertFalse(val, msg='key {0} is {1} while it should be something false'.format(key, val))
+ self.assertFalse(err.getvalue() + out.getvalue(), msg='unexpected output: {0!r} {1!r}'.format(
+ err.getvalue(),
+ out.getvalue(),
+ ))
+
+
+if __name__ == '__main__':
+ from tests.modules import main
+ main()
diff --git a/tests/test_python/test_config_merging.py b/tests/test_python/test_config_merging.py
new file mode 100644
index 0000000..3f4fa2a
--- /dev/null
+++ b/tests/test_python/test_config_merging.py
@@ -0,0 +1,270 @@
+# vim:fileencoding=utf-8:noet
+from __future__ import (unicode_literals, division, absolute_import, print_function)
+
+import os
+import json
+
+from subprocess import check_call
+from operator import add
+from shutil import rmtree
+
+from powerline.lib.dict import mergedicts_copy as mdc
+from powerline import Powerline
+
+from tests.modules import TestCase
+from tests.modules.lib.config_mock import select_renderer, UT
+
+
+CONFIG_DIR = 'tests/config'
+
+
+root_config = lambda: {
+ 'common': {
+ 'interval': None,
+ 'watcher': 'auto',
+ },
+ 'ext': {
+ 'test': {
+ 'theme': 'default',
+ 'colorscheme': 'default',
+ },
+ },
+}
+
+
+colors_config = lambda: {
+ 'colors': {
+ 'c1': 1,
+ 'c2': 2,
+ },
+ 'gradients': {
+ },
+}
+
+
+colorscheme_config = lambda: {
+ 'groups': {
+ 'g': {'fg': 'c1', 'bg': 'c2', 'attrs': []},
+ }
+}
+
+
+theme_config = lambda: {
+ 'segment_data': {
+ 's': {
+ 'before': 'b',
+ },
+ },
+ 'segments': {
+ 'left': [
+ {
+ 'type': 'string',
+ 'name': 's',
+ 'contents': 't',
+ 'highlight_groups': ['g'],
+ },
+ ],
+ 'right': [],
+ }
+}
+
+top_theme_config = lambda: {
+ 'dividers': {
+ 'left': {
+ 'hard': '#>',
+ 'soft': '|>',
+ },
+ 'right': {
+ 'hard': '<#',
+ 'soft': '<|',
+ },
+ },
+ 'spaces': 0,
+}
+
+
+main_tree = lambda: {
+ '1/config': root_config(),
+ '1/colors': colors_config(),
+ '1/colorschemes/default': colorscheme_config(),
+ '1/themes/test/default': theme_config(),
+ '1/themes/' + UT: top_theme_config(),
+ '1/themes/other1': mdc(top_theme_config(), {
+ 'dividers': {
+ 'left': {
+ 'hard': '!>',
+ }
+ }
+ }),
+ '1/themes/other2': mdc(top_theme_config(), {
+ 'dividers': {
+ 'left': {
+ 'hard': '>>',
+ }
+ }
+ }),
+}
+
+
+def mkdir_recursive(directory):
+ if os.path.isdir(directory):
+ return
+ mkdir_recursive(os.path.dirname(directory))
+ os.mkdir(directory)
+
+
+class TestPowerline(Powerline):
+ def get_config_paths(self):
+ return tuple(sorted([
+ os.path.join(CONFIG_DIR, d)
+ for d in os.listdir(CONFIG_DIR)
+ ]))
+
+
+class WithConfigTree(object):
+ __slots__ = ('tree', 'p', 'p_kwargs')
+
+ def __init__(self, tree, p_kwargs={'run_once': True}):
+ self.tree = tree
+ self.p = None
+ self.p_kwargs = p_kwargs
+
+ def __enter__(self, *args):
+ os.mkdir(CONFIG_DIR)
+ for k, v in self.tree.items():
+ fname = os.path.join(CONFIG_DIR, k) + '.json'
+ mkdir_recursive(os.path.dirname(fname))
+ with open(fname, 'w') as F:
+ json.dump(v, F)
+ select_renderer(simpler_renderer=True)
+ self.p = TestPowerline(
+ ext='test',
+ renderer_module='tests.modules.lib.config_mock',
+ **self.p_kwargs
+ )
+ if os.environ.get('POWERLINE_RUN_LINT_DURING_TESTS'):
+ try:
+ check_call(['scripts/powerline-lint'] + reduce(add, (
+ ['-p', d] for d in self.p.get_config_paths()
+ )))
+ except:
+ self.__exit__()
+ raise
+ return self.p.__enter__(*args)
+
+ def __exit__(self, *args):
+ try:
+ rmtree(CONFIG_DIR)
+ finally:
+ if self.p:
+ self.p.__exit__(*args)
+
+
+class TestMerging(TestCase):
+ def assertRenderEqual(self, p, output, **kwargs):
+ self.assertEqual(p.render(**kwargs).replace(' ', ' '), output)
+
+ def test_not_merged_config(self):
+ with WithConfigTree(main_tree()) as p:
+ self.assertRenderEqual(p, '{12} bt{2-}#>{--}')
+
+ def test_root_config_merging(self):
+ with WithConfigTree(mdc(main_tree(), {
+ '2/config': {
+ 'common': {
+ 'default_top_theme': 'other1',
+ }
+ },
+ })) as p:
+ self.assertRenderEqual(p, '{12} bt{2-}!>{--}')
+ with WithConfigTree(mdc(main_tree(), {
+ '2/config': {
+ 'common': {
+ 'default_top_theme': 'other1',
+ }
+ },
+ '3/config': {
+ 'common': {
+ 'default_top_theme': 'other2',
+ }
+ },
+ })) as p:
+ self.assertRenderEqual(p, '{12} bt{2-}>>{--}')
+
+ def test_top_theme_merging(self):
+ with WithConfigTree(mdc(main_tree(), {
+ '2/themes/' + UT: {
+ 'spaces': 1,
+ },
+ '3/themes/' + UT: {
+ 'dividers': {
+ 'left': {
+ 'hard': '>>',
+ }
+ }
+ },
+ })) as p:
+ self.assertRenderEqual(p, '{12} bt {2-}>>{--}')
+
+ def test_colors_config_merging(self):
+ with WithConfigTree(mdc(main_tree(), {
+ '2/colors': {
+ 'colors': {
+ 'c1': 3,
+ }
+ },
+ })) as p:
+ self.assertRenderEqual(p, '{32} bt{2-}#>{--}')
+ with WithConfigTree(mdc(main_tree(), {
+ '2/colors': {
+ 'colors': {
+ 'c1': 3,
+ }
+ },
+ '3/colors': {
+ 'colors': {
+ 'c1': 4,
+ }
+ },
+ })) as p:
+ self.assertRenderEqual(p, '{42} bt{2-}#>{--}')
+ with WithConfigTree(mdc(main_tree(), {
+ '2/colors': {
+ 'colors': {
+ 'c1': 3,
+ }
+ },
+ '3/colors': {
+ 'colors': {
+ 'c2': 4,
+ }
+ },
+ })) as p:
+ self.assertRenderEqual(p, '{34} bt{4-}#>{--}')
+
+ def test_colorschemes_merging(self):
+ with WithConfigTree(mdc(main_tree(), {
+ '2/colorschemes/default': {
+ 'groups': {
+ 'g': {'fg': 'c2', 'bg': 'c1', 'attrs': []},
+ }
+ },
+ })) as p:
+ self.assertRenderEqual(p, '{21} bt{1-}#>{--}')
+
+ def test_theme_merging(self):
+ with WithConfigTree(mdc(main_tree(), {
+ '2/themes/test/default': {
+ 'segment_data': {
+ 's': {
+ 'after': 'a',
+ }
+ }
+ },
+ })) as p:
+ self.assertRenderEqual(p, '{12} bta{2-}#>{--}')
+
+
+if __name__ == '__main__':
+ from tests.modules import main
+ main()
diff --git a/tests/test_python/test_config_reload.py b/tests/test_python/test_config_reload.py
new file mode 100644
index 0000000..a418d49
--- /dev/null
+++ b/tests/test_python/test_config_reload.py
@@ -0,0 +1,319 @@
+# vim:fileencoding=utf-8:noet
+from __future__ import (unicode_literals, division, absolute_import, print_function)
+
+from time import sleep
+from copy import deepcopy
+from functools import wraps
+
+from tests.modules import TestCase
+from tests.modules.lib.config_mock import get_powerline, add_watcher_events, UT
+
+
+config = {
+ 'config': {
+ 'common': {
+ 'interval': 0,
+ 'watcher': 'test',
+ },
+ 'ext': {
+ 'test': {
+ 'theme': 'default',
+ 'colorscheme': 'default',
+ },
+ },
+ },
+ 'colors': {
+ 'colors': {
+ "col1": 1,
+ "col2": 2,
+ "col3": 3,
+ "col4": 4,
+ },
+ 'gradients': {
+ },
+ },
+ 'colorschemes/test/default': {
+ 'groups': {
+ 'str1': {'fg': 'col1', 'bg': 'col2', 'attrs': ['bold']},
+ 'str2': {'fg': 'col3', 'bg': 'col4', 'attrs': ['underline']},
+ },
+ },
+ 'colorschemes/test/2': {
+ 'groups': {
+ 'str1': {'fg': 'col2', 'bg': 'col3', 'attrs': ['bold']},
+ 'str2': {'fg': 'col1', 'bg': 'col4', 'attrs': ['underline']},
+ },
+ },
+ 'themes/test/default': {
+ 'segments': {
+ "left": [
+ {
+ "type": "string",
+ "contents": "s",
+ "highlight_groups": ["str1"],
+ },
+ {
+ "type": "string",
+ "contents": "g",
+ "highlight_groups": ["str2"],
+ },
+ ],
+ "right": [
+ ],
+ },
+ },
+ 'themes/' + UT: {
+ 'dividers': {
+ "left": {
+ "hard": ">>",
+ "soft": ">",
+ },
+ "right": {
+ "hard": "<<",
+ "soft": "<",
+ },
+ },
+ 'spaces': 0,
+ },
+ 'themes/other': {
+ 'dividers': {
+ "left": {
+ "hard": ">>",
+ "soft": ">",
+ },
+ "right": {
+ "hard": "<<",
+ "soft": "<",
+ },
+ },
+ 'spaces': 1,
+ },
+ 'themes/test/2': {
+ 'segments': {
+ "left": [
+ {
+ "type": "string",
+ "contents": "t",
+ "highlight_groups": ["str1"],
+ },
+ {
+ "type": "string",
+ "contents": "b",
+ "highlight_groups": ["str2"],
+ },
+ ],
+ "right": [
+ ],
+ },
+ },
+}
+
+
+def with_new_config(func):
+ @wraps(func)
+ def f(self):
+ return func(self, deepcopy(config))
+ return f
+
+
+class TestConfigReload(TestCase):
+ def assertAccessEvents(self, p, *args):
+ events = set()
+ for event in args:
+ if ':' not in event:
+ events.add('check:' + event)
+ events.add('load:' + event)
+ else:
+ events.add(event)
+ self.assertEqual(set(p._pop_events()), events)
+
+ @with_new_config
+ def test_noreload(self, config):
+ with get_powerline(config, run_once=True) as p:
+ self.assertEqual(p.render(), '<1 2 1> s<2 4 False>>><3 4 4>g<4 False False>>><None None None>')
+ self.assertAccessEvents(p, 'config', 'colors', 'check:colorschemes/default', 'check:colorschemes/test/__main__', 'colorschemes/test/default', 'themes/test/default', 'themes/' + UT, 'check:themes/test/__main__')
+ config['config']['common']['spaces'] = 1
+ add_watcher_events(p, 'config', wait=False, interval=0.05)
+ # When running once thread should not start
+ self.assertEqual(p.render(), '<1 2 1> s<2 4 False>>><3 4 4>g<4 False False>>><None None None>')
+ self.assertAccessEvents(p)
+ self.assertEqual(p.logger._pop_msgs(), [])
+
+ @with_new_config
+ def test_reload_main(self, config):
+ with get_powerline(config, run_once=False) as p:
+ self.assertEqual(p.render(), '<1 2 1> s<2 4 False>>><3 4 4>g<4 False False>>><None None None>')
+ self.assertAccessEvents(p, 'config', 'colors', 'check:colorschemes/default', 'check:colorschemes/test/__main__', 'colorschemes/test/default', 'themes/test/default', 'themes/' + UT, 'check:themes/test/__main__')
+
+ config['config']['common']['default_top_theme'] = 'other'
+ add_watcher_events(p, 'config')
+ p.render()
+ self.assertEqual(p.render(), '<1 2 1> s <2 4 False>>><3 4 4>g <4 False False>>><None None None>')
+ self.assertAccessEvents(p, 'config', 'themes/other', 'check:themes/test/__main__', 'themes/test/default')
+ self.assertEqual(p.logger._pop_msgs(), [])
+
+ config['config']['ext']['test']['theme'] = 'nonexistent'
+ add_watcher_events(p, 'config')
+ self.assertEqual(p.render(), '<1 2 1> s <2 4 False>>><3 4 4>g <4 False False>>><None None None>')
+ self.assertAccessEvents(p, 'config', 'check:themes/test/nonexistent', 'themes/other', 'check:themes/test/__main__')
+ # It should normally handle file missing error
+ self.assertEqual(p.logger._pop_msgs(), [
+ 'exception:test:powerline:Failed to load theme: themes/test/__main__',
+ 'exception:test:powerline:Failed to load theme: themes/test/nonexistent',
+ 'exception:test:powerline:Failed to create renderer: themes/test/nonexistent'
+ ])
+
+ config['config']['ext']['test']['theme'] = 'default'
+ add_watcher_events(p, 'config')
+ self.assertEqual(p.render(), '<1 2 1> s <2 4 False>>><3 4 4>g <4 False False>>><None None None>')
+ self.assertAccessEvents(p, 'config', 'themes/test/default', 'themes/other', 'check:themes/test/__main__')
+ self.assertEqual(p.logger._pop_msgs(), [])
+
+ config['config']['ext']['test']['colorscheme'] = 'nonexistent'
+ add_watcher_events(p, 'config')
+ self.assertEqual(p.render(), '<1 2 1> s <2 4 False>>><3 4 4>g <4 False False>>><None None None>')
+ self.assertAccessEvents(p, 'config', 'check:colorschemes/nonexistent', 'check:colorschemes/test/__main__', 'check:colorschemes/test/nonexistent')
+ # It should normally handle file missing error
+ self.assertEqual(p.logger._pop_msgs(), [
+ 'exception:test:powerline:Failed to load colorscheme: colorschemes/nonexistent',
+ 'exception:test:powerline:Failed to load colorscheme: colorschemes/test/__main__',
+ 'exception:test:powerline:Failed to load colorscheme: colorschemes/test/nonexistent',
+ 'exception:test:powerline:Failed to create renderer: colorschemes/test/nonexistent'
+ ])
+
+ config['config']['ext']['test']['colorscheme'] = '2'
+ add_watcher_events(p, 'config')
+ self.assertEqual(p.render(), '<2 3 1> s <3 4 False>>><1 4 4>g <4 False False>>><None None None>')
+ self.assertAccessEvents(p, 'config', 'check:colorschemes/2', 'check:colorschemes/test/__main__', 'colorschemes/test/2')
+ self.assertEqual(p.logger._pop_msgs(), [])
+
+ config['config']['ext']['test']['theme'] = '2'
+ add_watcher_events(p, 'config')
+ self.assertEqual(p.render(), '<2 3 1> t <3 4 False>>><1 4 4>b <4 False False>>><None None None>')
+ self.assertAccessEvents(p, 'config', 'themes/test/2', 'themes/other', 'check:themes/test/__main__')
+ self.assertEqual(p.logger._pop_msgs(), [])
+
+ self.assertEqual(p.renderer.local_themes, None)
+ config['config']['ext']['test']['local_themes'] = 'something'
+ add_watcher_events(p, 'config')
+ self.assertEqual(p.render(), '<2 3 1> t <3 4 False>>><1 4 4>b <4 False False>>><None None None>')
+ self.assertAccessEvents(p, 'config')
+ self.assertEqual(p.logger._pop_msgs(), [])
+ self.assertEqual(p.renderer.local_themes, 'something')
+
+ @with_new_config
+ def test_reload_unexistent(self, config):
+ with get_powerline(config, run_once=False) as p:
+ self.assertEqual(p.render(), '<1 2 1> s<2 4 False>>><3 4 4>g<4 False False>>><None None None>')
+ self.assertAccessEvents(p, 'config', 'colors', 'check:colorschemes/default', 'check:colorschemes/test/__main__', 'colorschemes/test/default', 'themes/test/default', 'themes/' + UT, 'check:themes/test/__main__')
+
+ config['config']['ext']['test']['colorscheme'] = 'nonexistentraise'
+ add_watcher_events(p, 'config')
+ # It may appear that p.logger._pop_msgs() is called after given
+ # exception is added to the mesagges, but before config_loader
+ # exception was added (this one:
+ # “exception:test:config_loader:Error while running condition
+ # function for key colorschemes/test/nonexistentraise:
+ # fcf:colorschemes/test/nonexistentraise”).
+ # sleep(0.1)
+ self.assertEqual(p.render(), '<1 2 1> s<2 4 False>>><3 4 4>g<4 False False>>><None None None>')
+ # For colorschemes/{test/,}*raise find_config_file raises
+ # IOError, but it does not do so for check:colorschemes/test/__main__,
+ # so powerline is trying to load this, but not other
+ # colorschemes/*
+ self.assertAccessEvents(p, 'config', 'check:colorschemes/test/__main__', 'check:colorschemes/nonexistentraise', 'check:colorschemes/test/nonexistentraise')
+ self.assertIn('exception:test:powerline:Failed to create renderer: fcf:colorschemes/test/nonexistentraise', p.logger._pop_msgs())
+
+ config['colorschemes/nonexistentraise'] = {}
+ config['colorschemes/test/nonexistentraise'] = {
+ 'groups': {
+ 'str1': {'fg': 'col1', 'bg': 'col3', 'attrs': ['bold']},
+ 'str2': {'fg': 'col2', 'bg': 'col4', 'attrs': ['underline']},
+ },
+ }
+ while not p._will_create_renderer():
+ sleep(0.1)
+ self.assertEqual(p.render(), '<1 3 1> s<3 4 False>>><2 4 4>g<4 False False>>><None None None>')
+ # Same as above
+ self.assertAccessEvents(p, 'colorschemes/nonexistentraise', 'colorschemes/test/nonexistentraise', 'check:colorschemes/test/__main__')
+ self.assertEqual(p.logger._pop_msgs(), [])
+
+ @with_new_config
+ def test_reload_colors(self, config):
+ with get_powerline(config, run_once=False) as p:
+ self.assertEqual(p.render(), '<1 2 1> s<2 4 False>>><3 4 4>g<4 False False>>><None None None>')
+ self.assertAccessEvents(p, 'config', 'colors', 'check:colorschemes/default', 'check:colorschemes/test/__main__', 'colorschemes/test/default', 'themes/test/default', 'themes/' + UT, 'check:themes/test/__main__')
+
+ config['colors']['colors']['col1'] = 5
+ add_watcher_events(p, 'colors')
+ self.assertEqual(p.render(), '<5 2 1> s<2 4 False>>><3 4 4>g<4 False False>>><None None None>')
+ self.assertAccessEvents(p, 'colors')
+ self.assertEqual(p.logger._pop_msgs(), [])
+
+ @with_new_config
+ def test_reload_colorscheme(self, config):
+ with get_powerline(config, run_once=False) as p:
+ self.assertEqual(p.render(), '<1 2 1> s<2 4 False>>><3 4 4>g<4 False False>>><None None None>')
+ self.assertAccessEvents(p, 'config', 'colors', 'check:colorschemes/default', 'check:colorschemes/test/__main__', 'colorschemes/test/default', 'themes/test/default', 'themes/' + UT, 'check:themes/test/__main__')
+
+ config['colorschemes/test/default']['groups']['str1']['bg'] = 'col3'
+ add_watcher_events(p, 'colorschemes/test/default')
+ self.assertEqual(p.render(), '<1 3 1> s<3 4 False>>><3 4 4>g<4 False False>>><None None None>')
+ self.assertAccessEvents(p, 'check:colorschemes/default', 'check:colorschemes/test/__main__', 'colorschemes/test/default')
+ self.assertEqual(p.logger._pop_msgs(), [])
+
+ @with_new_config
+ def test_reload_theme(self, config):
+ with get_powerline(config, run_once=False) as p:
+ self.assertEqual(p.render(), '<1 2 1> s<2 4 False>>><3 4 4>g<4 False False>>><None None None>')
+ self.assertAccessEvents(p, 'config', 'colors', 'check:colorschemes/default', 'check:colorschemes/test/__main__', 'colorschemes/test/default', 'themes/test/default', 'themes/' + UT, 'check:themes/test/__main__')
+
+ config['themes/test/default']['segments']['left'][0]['contents'] = 'col3'
+ add_watcher_events(p, 'themes/test/default')
+ self.assertEqual(p.render(), '<1 2 1> col3<2 4 False>>><3 4 4>g<4 False False>>><None None None>')
+ self.assertAccessEvents(p, 'themes/test/default', 'themes/' + UT, 'check:themes/test/__main__')
+ self.assertEqual(p.logger._pop_msgs(), [])
+
+ @with_new_config
+ def test_reload_top_theme(self, config):
+ with get_powerline(config, run_once=False) as p:
+ self.assertEqual(p.render(), '<1 2 1> s<2 4 False>>><3 4 4>g<4 False False>>><None None None>')
+ self.assertAccessEvents(p, 'config', 'colors', 'check:colorschemes/default', 'check:colorschemes/test/__main__', 'colorschemes/test/default', 'themes/test/default', 'themes/' + UT, 'check:themes/test/__main__')
+
+ config['themes/' + UT]['dividers']['left']['hard'] = '|>'
+ add_watcher_events(p, 'themes/' + UT)
+ self.assertEqual(p.render(), '<1 2 1> s<2 4 False>|><3 4 4>g<4 False False>|><None None None>')
+ self.assertAccessEvents(p, 'themes/test/default', 'themes/' + UT, 'check:themes/test/__main__')
+ self.assertEqual(p.logger._pop_msgs(), [])
+
+ @with_new_config
+ def test_reload_theme_main(self, config):
+ config['config']['common']['interval'] = None
+ with get_powerline(config, run_once=False) as p:
+ self.assertEqual(p.render(), '<1 2 1> s<2 4 False>>><3 4 4>g<4 False False>>><None None None>')
+ self.assertAccessEvents(p, 'config', 'colors', 'check:colorschemes/default', 'check:colorschemes/test/__main__', 'colorschemes/test/default', 'themes/test/default', 'themes/' + UT, 'check:themes/test/__main__')
+
+ config['themes/test/default']['segments']['left'][0]['contents'] = 'col3'
+ add_watcher_events(p, 'themes/test/default', wait=False)
+ self.assertEqual(p.render(), '<1 2 1> col3<2 4 False>>><3 4 4>g<4 False False>>><None None None>')
+ self.assertAccessEvents(p, 'themes/test/default', 'themes/' + UT, 'check:themes/test/__main__')
+ self.assertEqual(p.logger._pop_msgs(), [])
+ self.assertTrue(p._watcher._calls)
+
+ @with_new_config
+ def test_run_once_no_theme_reload(self, config):
+ config['config']['common']['interval'] = None
+ with get_powerline(config, run_once=True) as p:
+ self.assertEqual(p.render(), '<1 2 1> s<2 4 False>>><3 4 4>g<4 False False>>><None None None>')
+ self.assertAccessEvents(p, 'config', 'colors', 'check:colorschemes/default', 'check:colorschemes/test/__main__', 'colorschemes/test/default', 'themes/test/default', 'themes/' + UT, 'check:themes/test/__main__')
+
+ config['themes/test/default']['segments']['left'][0]['contents'] = 'col3'
+ add_watcher_events(p, 'themes/test/default', wait=False)
+ self.assertEqual(p.render(), '<1 2 1> s<2 4 False>>><3 4 4>g<4 False False>>><None None None>')
+ self.assertAccessEvents(p)
+ self.assertEqual(p.logger._pop_msgs(), [])
+
+
+if __name__ == '__main__':
+ from tests.modules import main
+ main()
diff --git a/tests/test_python/test_configuration.py b/tests/test_python/test_configuration.py
new file mode 100644
index 0000000..aa9e844
--- /dev/null
+++ b/tests/test_python/test_configuration.py
@@ -0,0 +1,877 @@
+# vim:fileencoding=utf-8:noet
+from __future__ import (unicode_literals, division, absolute_import, print_function)
+
+import sys
+import os
+
+from functools import wraps
+from copy import deepcopy
+
+import tests.modules.vim as vim_module
+
+from tests.modules import TestCase
+from tests.modules.lib.config_mock import (get_powerline, get_powerline_raw,
+ swap_attributes, UT)
+from tests.modules.lib import Args, replace_item
+
+
+def highlighted_string(s, group, **kwargs):
+ ret = {
+ 'type': 'string',
+ 'contents': s,
+ 'highlight_groups': [group],
+ }
+ ret.update(kwargs)
+ return ret
+
+
+config = {
+ 'config': {
+ 'common': {
+ 'interval': 0,
+ 'watcher': 'test',
+ },
+ 'ext': {
+ 'test': {
+ 'theme': 'default',
+ 'colorscheme': 'default',
+ },
+ 'vim': {
+ 'theme': 'default',
+ 'colorscheme': 'default',
+ },
+ 'shell': {
+ 'theme': 'default',
+ 'colorscheme': 'default',
+ },
+ 'wm': {
+ 'theme': 'default',
+ 'colorscheme': 'default',
+ },
+ },
+ },
+ 'colors': {
+ 'colors': {
+ 'col1': 1,
+ 'col2': 2,
+ 'col3': 3,
+ 'col4': 4,
+ 'col5': 5,
+ 'col6': 6,
+ 'col7': 7,
+ 'col8': 8,
+ 'col9': 9,
+ 'col10': 10,
+ 'col11': 11,
+ 'col12': 12,
+ },
+ 'gradients': {
+ },
+ },
+ 'colorschemes/test/__main__': {
+ 'groups': {
+ 'm1': 'g1',
+ 'm2': 'g3',
+ 'm3': {'fg': 'col11', 'bg': 'col12', 'attrs': []},
+ }
+ },
+ 'colorschemes/default': {
+ 'groups': {
+ 'g1': {'fg': 'col5', 'bg': 'col6', 'attrs': []},
+ 'g2': {'fg': 'col7', 'bg': 'col8', 'attrs': []},
+ 'g3': {'fg': 'col9', 'bg': 'col10', 'attrs': []},
+ }
+ },
+ 'colorschemes/test/default': {
+ 'groups': {
+ 'str1': {'fg': 'col1', 'bg': 'col2', 'attrs': ['bold']},
+ 'str2': {'fg': 'col3', 'bg': 'col4', 'attrs': ['underline']},
+ 'd1': 'g2',
+ 'd2': 'm2',
+ 'd3': 'm3',
+ },
+ },
+ 'colorschemes/vim/default': {
+ 'groups': {
+ 'environment': {'fg': 'col3', 'bg': 'col4', 'attrs': ['underline']},
+ },
+ },
+ 'colorschemes/wm/default': {
+ 'groups': {
+ 'hl1': {'fg': 'col1', 'bg': 'col2', 'attrs': ['underline']},
+ 'hl2': {'fg': 'col2', 'bg': 'col1', 'attrs': []},
+ 'hl3': {'fg': 'col3', 'bg': 'col1', 'attrs': ['underline']},
+ 'hl4': {'fg': 'col2', 'bg': 'col4', 'attrs': []},
+ },
+ },
+ 'themes/test/default': {
+ 'segments': {
+ 'left': [
+ highlighted_string('s', 'str1', width='auto'),
+ highlighted_string('g', 'str2'),
+ ],
+ 'right': [
+ highlighted_string('f', 'str2', width='auto', align='r'),
+ ],
+ },
+ },
+ 'themes/' + UT: {
+ 'dividers': {
+ 'left': {
+ 'hard': '>>',
+ 'soft': '>',
+ },
+ 'right': {
+ 'hard': '<<',
+ 'soft': '<',
+ },
+ },
+ 'spaces': 0,
+ },
+ 'themes/test/__main__': {
+ 'dividers': {
+ 'right': {
+ 'soft': '|',
+ },
+ },
+ },
+ 'themes/vim/default': {
+ 'segments': {
+ 'left': [
+ {
+ 'function': 'powerline.segments.common.env.environment',
+ 'args': {
+ 'variable': 'TEST',
+ },
+ },
+ ],
+ },
+ },
+ 'themes/shell/default': {
+ 'default_module': 'powerline.segments.common',
+ 'segments': {
+ 'left': [
+ highlighted_string('s', 'g1', width='auto'),
+ ],
+ },
+ },
+ 'themes/wm/default': {
+ 'default_module': 'powerline.segments.common',
+ 'segments': {
+ 'left': [
+ highlighted_string('A', 'hl1'),
+ highlighted_string('B', 'hl2'),
+ ],
+ 'right': [
+ highlighted_string('C', 'hl3'),
+ highlighted_string('D', 'hl4'),
+ ],
+ },
+ },
+}
+
+
+def with_new_config(func):
+ @wraps(func)
+ def f(self):
+ return func(self, deepcopy(config))
+ return f
+
+
+def add_args(func):
+ @wraps(func)
+ def f(self):
+ new_config = deepcopy(config)
+ with get_powerline(new_config, run_once=True, simpler_renderer=True) as p:
+ func(self, p, new_config)
+ return f
+
+
+class TestRender(TestCase):
+ def assertRenderEqual(self, p, output, **kwargs):
+ self.assertEqual(p.render(**kwargs).replace(' ', ' '), output)
+
+ def assertRenderLinesEqual(self, p, output, **kwargs):
+ self.assertEqual([l.replace(' ', ' ') for l in p.render_above_lines(**kwargs)], output)
+
+
+class TestLines(TestRender):
+ @add_args
+ def test_without_above(self, p, config):
+ self.assertRenderEqual(p, '{121} s{24}>>{344}g{34}>{34}|{344}f {--}')
+ self.assertRenderEqual(p, '{121} s {24}>>{344}g{34}>{34}|{344}f {--}', width=10)
+ # self.assertRenderEqual(p, '{121} s {24}>>{344}g{34}>{34}|{344} f {--}', width=11)
+ self.assertEqual(list(p.render_above_lines()), [])
+
+ @with_new_config
+ def test_with_above(self, config):
+ old_segments = deepcopy(config['themes/test/default']['segments'])
+ config['themes/test/default']['segments']['above'] = [old_segments]
+ with get_powerline(config, run_once=True, simpler_renderer=True) as p:
+ self.assertRenderLinesEqual(p, [
+ '{121} s{24}>>{344}g{34}>{34}|{344}f {--}',
+ ])
+ self.assertRenderLinesEqual(p, [
+ '{121} s {24}>>{344}g{34}>{34}|{344}f {--}',
+ ], width=10)
+
+ config['themes/test/default']['segments']['above'] = [old_segments] * 2
+ with get_powerline(config, run_once=True, simpler_renderer=True) as p:
+ self.assertRenderLinesEqual(p, [
+ '{121} s{24}>>{344}g{34}>{34}|{344}f {--}',
+ '{121} s{24}>>{344}g{34}>{34}|{344}f {--}',
+ ])
+ self.assertRenderLinesEqual(p, [
+ '{121} s {24}>>{344}g{34}>{34}|{344}f {--}',
+ '{121} s {24}>>{344}g{34}>{34}|{344}f {--}',
+ ], width=10)
+
+
+class TestSegments(TestRender):
+ @add_args
+ def test_display(self, p, config):
+ config['themes/test/default']['segments']['left'][0]['display'] = False
+ config['themes/test/default']['segments']['left'][1]['display'] = True
+ config['themes/test/default']['segments']['right'][0]['display'] = False
+ self.assertRenderEqual(p, '{344} g{4-}>>{--}')
+
+
+class TestColorschemesHierarchy(TestRender):
+ @add_args
+ def test_group_redirects(self, p, config):
+ config['themes/test/default']['segments'] = {
+ 'left': [
+ highlighted_string('a', 'd1', draw_hard_divider=False),
+ highlighted_string('b', 'd2', draw_hard_divider=False),
+ highlighted_string('c', 'd3', draw_hard_divider=False),
+ highlighted_string('A', 'm1', draw_hard_divider=False),
+ highlighted_string('B', 'm2', draw_hard_divider=False),
+ highlighted_string('C', 'm3', draw_hard_divider=False),
+ highlighted_string('1', 'g1', draw_hard_divider=False),
+ highlighted_string('2', 'g2', draw_hard_divider=False),
+ highlighted_string('3', 'g3', draw_hard_divider=False),
+ ],
+ 'right': [],
+ }
+ self.assertRenderEqual(p, '{78} a{910}b{1112}c{56}A{910}B{1112}C{56}1{78}2{910}3{--}')
+ self.assertEqual(p.logger._pop_msgs(), [])
+
+ @add_args
+ def test_group_redirects_no_main(self, p, config):
+ del config['colorschemes/test/__main__']
+ config['themes/test/default']['segments'] = {
+ 'left': [
+ highlighted_string('a', 'd1', draw_hard_divider=False),
+ highlighted_string('1', 'g1', draw_hard_divider=False),
+ highlighted_string('2', 'g2', draw_hard_divider=False),
+ highlighted_string('3', 'g3', draw_hard_divider=False),
+ ],
+ 'right': [],
+ }
+ self.assertRenderEqual(p, '{78} a{56}1{78}2{910}3{--}')
+ self.assertEqual(p.logger._pop_msgs(), [])
+
+ @add_args
+ def test_group_redirects_no_top_default(self, p, config):
+ del config['colorschemes/default']
+ config['themes/test/default']['segments'] = {
+ 'left': [
+ highlighted_string('c', 'd3', draw_soft_divider=False),
+ highlighted_string('C', 'm3', draw_hard_divider=False),
+ ],
+ 'right': [],
+ }
+ self.assertRenderEqual(p, '{1112} c{1112}C{--}')
+ self.assertEqual(p.logger._pop_msgs(), [])
+
+ @add_args
+ def test_group_redirects_no_test_default(self, p, config):
+ del config['colorschemes/test/default']
+ config['themes/test/default']['segments'] = {
+ 'left': [
+ highlighted_string('A', 'm1', draw_hard_divider=False),
+ highlighted_string('B', 'm2', draw_hard_divider=False),
+ highlighted_string('C', 'm3', draw_hard_divider=False),
+ highlighted_string('1', 'g1', draw_hard_divider=False),
+ highlighted_string('2', 'g2', draw_hard_divider=False),
+ highlighted_string('3', 'g3', draw_hard_divider=False),
+ ],
+ 'right': [],
+ }
+ self.assertRenderEqual(p, '{56} A{910}B{1112}C{56}1{78}2{910}3{--}')
+ self.assertEqual(p.logger._pop_msgs(), [])
+
+ @add_args
+ def test_group_redirects_only_main(self, p, config):
+ del config['colorschemes/default']
+ del config['colorschemes/test/default']
+ config['themes/test/default']['segments'] = {
+ 'left': [
+ highlighted_string('C', 'm3', draw_hard_divider=False),
+ ],
+ 'right': [],
+ }
+ # Powerline is not able to work without default colorscheme
+ # somewhere, thus it will output error string
+ self.assertRenderEqual(p, 'colorschemes/test/default')
+ self.assertEqual(p.logger._pop_msgs(), [
+ 'exception:test:powerline:Failed to load colorscheme: colorschemes/default',
+ 'exception:test:powerline:Failed to load colorscheme: colorschemes/test/default',
+ 'exception:test:powerline:Failed to create renderer: colorschemes/test/default',
+ 'exception:test:powerline:Failed to render: colorschemes/test/default',
+ ])
+
+ @add_args
+ def test_group_redirects_only_top_default(self, p, config):
+ del config['colorschemes/test/__main__']
+ del config['colorschemes/test/default']
+ config['themes/test/default']['segments'] = {
+ 'left': [
+ highlighted_string('1', 'g1', draw_hard_divider=False),
+ highlighted_string('2', 'g2', draw_hard_divider=False),
+ highlighted_string('3', 'g3', draw_hard_divider=False),
+ ],
+ 'right': [],
+ }
+ self.assertRenderEqual(p, '{56} 1{78}2{910}3{--}')
+ self.assertEqual(p.logger._pop_msgs(), [])
+
+ @add_args
+ def test_group_redirects_only_test_default(self, p, config):
+ del config['colorschemes/default']
+ del config['colorschemes/test/__main__']
+ config['themes/test/default']['segments'] = {
+ 'left': [
+ highlighted_string('s', 'str1', draw_hard_divider=False),
+ ],
+ 'right': [],
+ }
+ self.assertRenderEqual(p, '{121} s{--}')
+ self.assertEqual(p.logger._pop_msgs(), [])
+
+
+class TestThemeHierarchy(TestRender):
+ @add_args
+ def test_hierarchy(self, p, config):
+ self.assertRenderEqual(p, '{121} s{24}>>{344}g{34}>{34}|{344}f {--}')
+
+ @add_args
+ def test_no_main(self, p, config):
+ del config['themes/test/__main__']
+ self.assertRenderEqual(p, '{121} s{24}>>{344}g{34}>{34}<{344}f {--}')
+ self.assertEqual(p.logger._pop_msgs(), [])
+
+ @add_args
+ def test_no_powerline(self, p, config):
+ config['themes/test/__main__']['dividers'] = config['themes/' + UT]['dividers']
+ config['themes/test/__main__']['spaces'] = 1
+ del config['themes/' + UT]
+ self.assertRenderEqual(p, '{121} s {24}>>{344}g {34}>{34}<{344} f {--}')
+ self.assertEqual(p.logger._pop_msgs(), [])
+
+ @add_args
+ def test_no_default(self, p, config):
+ del config['themes/test/default']
+ self.assertRenderEqual(p, 'themes/test/default')
+ self.assertEqual(p.logger._pop_msgs(), [
+ 'exception:test:powerline:Failed to load theme: themes/test/default',
+ 'exception:test:powerline:Failed to create renderer: themes/test/default',
+ 'exception:test:powerline:Failed to render: themes/test/default',
+ ])
+
+ @add_args
+ def test_only_default(self, p, config):
+ config['themes/test/default']['dividers'] = config['themes/' + UT]['dividers']
+ config['themes/test/default']['spaces'] = 1
+ del config['themes/test/__main__']
+ del config['themes/' + UT]
+ self.assertRenderEqual(p, '{121} s {24}>>{344}g {34}>{34}<{344} f {--}')
+
+ @add_args
+ def test_only_main(self, p, config):
+ del config['themes/test/default']
+ del config['themes/' + UT]
+ self.assertRenderEqual(p, 'themes/test/default')
+ self.assertEqual(p.logger._pop_msgs(), [
+ 'exception:test:powerline:Failed to load theme: themes/' + UT,
+ 'exception:test:powerline:Failed to load theme: themes/test/default',
+ 'exception:test:powerline:Failed to create renderer: themes/test/default',
+ 'exception:test:powerline:Failed to render: themes/test/default',
+ ])
+
+ @add_args
+ def test_only_powerline(self, p, config):
+ del config['themes/test/default']
+ del config['themes/test/__main__']
+ self.assertRenderEqual(p, 'themes/test/default')
+ self.assertEqual(p.logger._pop_msgs(), [
+ 'exception:test:powerline:Failed to load theme: themes/test/__main__',
+ 'exception:test:powerline:Failed to load theme: themes/test/default',
+ 'exception:test:powerline:Failed to create renderer: themes/test/default',
+ 'exception:test:powerline:Failed to render: themes/test/default',
+ ])
+
+ @add_args
+ def test_nothing(self, p, config):
+ del config['themes/test/default']
+ del config['themes/' + UT]
+ del config['themes/test/__main__']
+ self.assertRenderEqual(p, 'themes/test/default')
+ self.assertEqual(p.logger._pop_msgs(), [
+ 'exception:test:powerline:Failed to load theme: themes/' + UT,
+ 'exception:test:powerline:Failed to load theme: themes/test/__main__',
+ 'exception:test:powerline:Failed to load theme: themes/test/default',
+ 'exception:test:powerline:Failed to create renderer: themes/test/default',
+ 'exception:test:powerline:Failed to render: themes/test/default',
+ ])
+
+
+class TestDisplayCondition(TestRender):
+ @add_args
+ def test_include_modes(self, p, config):
+ config['themes/test/default']['segments'] = {
+ 'left': [
+ highlighted_string('s1', 'g1', include_modes=['m1']),
+ highlighted_string('s2', 'g1', include_modes=['m1', 'm2']),
+ highlighted_string('s3', 'g1', include_modes=['m3']),
+ ]
+ }
+ self.assertRenderEqual(p, '{--}')
+ self.assertRenderEqual(p, '{56} s1{56}>{56}s2{6-}>>{--}', mode='m1')
+ self.assertRenderEqual(p, '{56} s2{6-}>>{--}', mode='m2')
+ self.assertRenderEqual(p, '{56} s3{6-}>>{--}', mode='m3')
+
+ @add_args
+ def test_exclude_modes(self, p, config):
+ config['themes/test/default']['segments'] = {
+ 'left': [
+ highlighted_string('s1', 'g1', exclude_modes=['m1']),
+ highlighted_string('s2', 'g1', exclude_modes=['m1', 'm2']),
+ highlighted_string('s3', 'g1', exclude_modes=['m3']),
+ ]
+ }
+ self.assertRenderEqual(p, '{56} s1{56}>{56}s2{56}>{56}s3{6-}>>{--}')
+ self.assertRenderEqual(p, '{56} s3{6-}>>{--}', mode='m1')
+ self.assertRenderEqual(p, '{56} s1{56}>{56}s3{6-}>>{--}', mode='m2')
+ self.assertRenderEqual(p, '{56} s1{56}>{56}s2{6-}>>{--}', mode='m3')
+
+ @add_args
+ def test_exinclude_modes(self, p, config):
+ config['themes/test/default']['segments'] = {
+ 'left': [
+ highlighted_string('s1', 'g1', exclude_modes=['m1'], include_modes=['m2']),
+ highlighted_string('s2', 'g1', exclude_modes=['m1', 'm2'], include_modes=['m3']),
+ highlighted_string('s3', 'g1', exclude_modes=['m3'], include_modes=['m3']),
+ ]
+ }
+ self.assertRenderEqual(p, '{--}')
+ self.assertRenderEqual(p, '{--}', mode='m1')
+ self.assertRenderEqual(p, '{56} s1{6-}>>{--}', mode='m2')
+ self.assertRenderEqual(p, '{56} s2{6-}>>{--}', mode='m3')
+
+ @add_args
+ def test_exinclude_function_nonexistent_module(self, p, config):
+ config['themes/test/default']['segments'] = {
+ 'left': [
+ highlighted_string('s1', 'g1', exclude_function='xxx_nonexistent_module.foo'),
+ highlighted_string('s2', 'g1', exclude_function='xxx_nonexistent_module.foo', include_function='xxx_nonexistent_module.bar'),
+ highlighted_string('s3', 'g1', include_function='xxx_nonexistent_module.bar'),
+ ]
+ }
+ self.assertRenderEqual(p, '{56} s1{56}>{56}s2{56}>{56}s3{6-}>>{--}')
+
+ @add_args
+ def test_exinclude_function(self, p, config):
+ config['themes/test/default']['segments'] = {
+ 'left': [
+ highlighted_string('s1', 'g1', exclude_function='mod.foo'),
+ highlighted_string('s2', 'g1', exclude_function='mod.foo', include_function='mod.bar'),
+ highlighted_string('s3', 'g1', include_function='mod.bar'),
+ ]
+ }
+ launched = set()
+ fool = [None]
+ barl = [None]
+
+ def foo(*args, **kwargs):
+ launched.add('foo')
+ self.assertEqual(set(kwargs.keys()), set(('pl', 'segment_info', 'mode')))
+ self.assertEqual(args, ())
+ return fool[0]
+
+ def bar(*args, **kwargs):
+ launched.add('bar')
+ self.assertEqual(set(kwargs.keys()), set(('pl', 'segment_info', 'mode')))
+ self.assertEqual(args, ())
+ return barl[0]
+
+ with replace_item(sys.modules, 'mod', Args(foo=foo, bar=bar)):
+ fool[0] = True
+ barl[0] = True
+ self.assertRenderEqual(p, '{56} s3{6-}>>{--}')
+ self.assertEqual(launched, set(('foo', 'bar')))
+
+ fool[0] = False
+ barl[0] = True
+ self.assertRenderEqual(p, '{56} s1{56}>{56}s2{56}>{56}s3{6-}>>{--}')
+ self.assertEqual(launched, set(('foo', 'bar')))
+
+ fool[0] = False
+ barl[0] = False
+ self.assertRenderEqual(p, '{56} s1{6-}>>{--}')
+ self.assertEqual(launched, set(('foo', 'bar')))
+
+ fool[0] = True
+ barl[0] = False
+ self.assertRenderEqual(p, '{--}')
+ self.assertEqual(launched, set(('foo', 'bar')))
+
+ @add_args
+ def test_exinclude_modes_override_functions(self, p, config):
+ config['themes/test/default']['segments'] = {
+ 'left': [
+ highlighted_string('s1', 'g1', exclude_function='mod.foo', exclude_modes=['m2']),
+ highlighted_string('s2', 'g1', exclude_function='mod.foo', include_modes=['m2']),
+ highlighted_string('s3', 'g1', include_function='mod.foo', exclude_modes=['m2']),
+ highlighted_string('s4', 'g1', include_function='mod.foo', include_modes=['m2']),
+ ]
+ }
+ fool = [None]
+
+ def foo(*args, **kwargs):
+ return fool[0]
+
+ with replace_item(sys.modules, 'mod', Args(foo=foo)):
+ fool[0] = True
+ self.assertRenderEqual(p, '{56} s4{6-}>>{--}', mode='m2')
+ self.assertRenderEqual(p, '{56} s3{56}>{56}s4{6-}>>{--}', mode='m1')
+
+ fool[0] = False
+ self.assertRenderEqual(p, '{56} s2{56}>{56}s4{6-}>>{--}', mode='m2')
+ self.assertRenderEqual(p, '{56} s1{6-}>>{--}', mode='m1')
+
+
+class TestOuterPadding(TestRender):
+ @add_args
+ def test_outer_padding_left(self, p, config):
+ config['themes/' + UT]['outer_padding'] = 5
+ self.assertRenderEqual(p, '{121} s{24}>>{344}g{4-}>>{--}', side='left')
+
+ @add_args
+ def test_outer_padding_right(self, p, config):
+ config['themes/' + UT]['outer_padding'] = 5
+ self.assertRenderEqual(p, '{4-}<<{344}f {--}', side='right')
+
+ @add_args
+ def test_outer_padding_ten(self, p, config):
+ config['themes/' + UT]['outer_padding'] = 10
+ self.assertRenderEqual(p, '{121} s {24}>>{344}g{34}>{34}|{344} f {--}', width=30)
+
+ @add_args
+ def test_outer_padding_zero(self, p, config):
+ config['themes/' + UT]['outer_padding'] = 0
+ self.assertRenderEqual(p, '{121}s {24}>>{344}g{34}>{34}|{344} f{--}', width=30)
+
+
+class TestSegmentAttributes(TestRender):
+ @add_args
+ def test_no_attributes(self, p, config):
+ def m1(divider=',', **kwargs):
+ return divider.join(kwargs.keys()) + divider
+ config['themes/test/default']['segments'] = {
+ 'left': [
+ {
+ 'function': 'bar.m1'
+ }
+ ]
+ }
+ with replace_item(sys.modules, 'bar', Args(m1=m1)):
+ self.assertRenderEqual(p, '{56} pl,{6-}>>{--}')
+
+ @add_args
+ def test_segment_datas(self, p, config):
+ def m1(divider=',', **kwargs):
+ return divider.join(kwargs.keys()) + divider
+ m1.powerline_segment_datas = {
+ UT: {
+ 'args': {
+ 'divider': ';'
+ }
+ },
+ 'ascii': {
+ 'args': {
+ 'divider': '--'
+ }
+ }
+ }
+ config['themes/test/default']['segments'] = {
+ 'left': [
+ {
+ 'function': 'bar.m1'
+ }
+ ]
+ }
+ with replace_item(sys.modules, 'bar', Args(m1=m1)):
+ self.assertRenderEqual(p, '{56} pl;{6-}>>{--}')
+
+ @add_args
+ def test_expand(self, p, config):
+ def m1(divider=',', **kwargs):
+ return divider.join(kwargs.keys()) + divider
+
+ def expand(pl, amount, segment, **kwargs):
+ return ('-' * amount) + segment['contents']
+
+ m1.expand = expand
+ config['themes/test/default']['segments'] = {
+ 'left': [
+ {
+ 'function': 'bar.m1',
+ 'width': 'auto'
+ }
+ ]
+ }
+ with replace_item(sys.modules, 'bar', Args(m1=m1)):
+ self.assertRenderEqual(p, '{56} ----pl,{6-}>>{--}', width=10)
+
+ @add_args
+ def test_truncate(self, p, config):
+ def m1(divider=',', **kwargs):
+ return divider.join(kwargs.keys()) + divider
+
+ def truncate(pl, amount, segment, **kwargs):
+ return segment['contents'][:-amount]
+
+ m1.truncate = truncate
+ config['themes/test/default']['segments'] = {
+ 'left': [
+ {
+ 'function': 'bar.m1'
+ }
+ ]
+ }
+ with replace_item(sys.modules, 'bar', Args(m1=m1)):
+ self.assertRenderEqual(p, '{56} p{6-}>>{--}', width=4)
+
+
+class TestSegmentData(TestRender):
+ @add_args
+ def test_segment_data(self, p, config):
+ def m1(**kwargs):
+ return 'S'
+
+ def m2(**kwargs):
+ return 'S'
+ sys.modules['bar'] = Args(m1=m1, m2=m2)
+ config['themes/' + UT]['segment_data'] = {
+ 'm1': {
+ 'before': '1'
+ },
+ 'bar.m2': {
+ 'before': '2'
+ },
+ 'n': {
+ 'before': '3'
+ },
+ 'm2': {
+ 'before': '4'
+ },
+ }
+ config['themes/test/default']['segments'] = {
+ 'left': [
+ {
+ 'function': 'bar.m1'
+ },
+ {
+ 'function': 'bar.m1',
+ 'name': 'n'
+ },
+ {
+ 'function': 'bar.m2',
+ 'name': 'n'
+ },
+ {
+ 'function': 'bar.m2'
+ }
+ ]
+ }
+ self.assertRenderEqual(p, '{56} 1S{56}>{56}3S{610}>>{910}3S{910}>{910}2S{10-}>>{--}')
+
+
+class TestShellEscapes(TestCase):
+ @with_new_config
+ def test_escapes(self, config):
+ from powerline.shell import ShellPowerline
+ import powerline as powerline_module
+ with swap_attributes(config, powerline_module):
+ with get_powerline_raw(config, ShellPowerline, args=Args(config_path=[''])) as powerline:
+ self.assertEqual(powerline.render(segment_info={}, side='left'), '\x1b[0;38;5;5;48;5;6m\xa0s\x1b[0;38;5;6;49;22m>>\x1b[0m')
+
+ @with_new_config
+ def test_tmux_escapes(self, config):
+ from powerline.shell import ShellPowerline
+ import powerline as powerline_module
+ config['config']['common']['additional_escapes'] = 'tmux'
+ with swap_attributes(config, powerline_module):
+ with get_powerline_raw(config, ShellPowerline, args=Args(config_path=[''])) as powerline:
+ self.assertEqual(powerline.render(segment_info={}, side='left'), '\x1bPtmux;\x1b\x1b[0;38;5;5;48;5;6m\x1b\\\xa0s\x1bPtmux;\x1b\x1b[0;38;5;6;49;22m\x1b\\>>\x1bPtmux;\x1b\x1b[0m\x1b\\')
+
+ @with_new_config
+ def test_screen_escapes(self, config):
+ from powerline.shell import ShellPowerline
+ import powerline as powerline_module
+ config['config']['common']['additional_escapes'] = 'screen'
+ with swap_attributes(config, powerline_module):
+ with get_powerline_raw(config, ShellPowerline, args=Args(config_path=[''])) as powerline:
+ self.assertEqual(powerline.render(segment_info={}, side='left'), '\x1bP\x1b\x1b[0;38;5;5;48;5;6m\x1b\\\xa0s\x1bP\x1b\x1b[0;38;5;6;49;22m\x1b\\>>\x1bP\x1b\x1b[0m\x1b\\')
+
+ @with_new_config
+ def test_fbterm_escapes(self, config):
+ from powerline.shell import ShellPowerline
+ import powerline as powerline_module
+ config['config']['common']['term_escape_style'] = 'fbterm'
+ with swap_attributes(config, powerline_module):
+ with get_powerline_raw(config, ShellPowerline, args=Args(config_path=[''])) as powerline:
+ self.assertEqual(powerline.render(segment_info={}, side='left'), '\x1b[0m\x1b[1;5}\x1b[2;6}\xa0s\x1b[0m\x1b[1;6}\x1b[49m\x1b[22m>>\x1b[0m')
+
+ @with_new_config
+ def test_fbterm_tmux_escapes(self, config):
+ from powerline.shell import ShellPowerline
+ import powerline as powerline_module
+ config['config']['common']['term_escape_style'] = 'fbterm'
+ config['config']['common']['additional_escapes'] = 'tmux'
+ with swap_attributes(config, powerline_module):
+ with get_powerline_raw(config, ShellPowerline, args=Args(config_path=[''])) as powerline:
+ self.assertEqual(powerline.render(segment_info={}, side='left'), '\x1bPtmux;\x1b\x1b[0m\x1b\x1b[1;5}\x1b\x1b[2;6}\x1b\\\xa0s\x1bPtmux;\x1b\x1b[0m\x1b\x1b[1;6}\x1b\x1b[49m\x1b\x1b[22m\x1b\\>>\x1bPtmux;\x1b\x1b[0m\x1b\\')
+
+ @with_new_config
+ def test_fbterm_screen_escapes(self, config):
+ from powerline.shell import ShellPowerline
+ import powerline as powerline_module
+ config['config']['common']['term_escape_style'] = 'fbterm'
+ config['config']['common']['additional_escapes'] = 'screen'
+ with swap_attributes(config, powerline_module):
+ with get_powerline_raw(config, ShellPowerline, args=Args(config_path=[''])) as powerline:
+ self.assertEqual(powerline.render(segment_info={}, side='left'), '\x1bP\x1b\x1b[0m\x1b\x1b[1;5}\x1b\x1b[2;6}\x1b\\\xa0s\x1bP\x1b\x1b[0m\x1b\x1b[1;6}\x1b\x1b[49m\x1b\x1b[22m\x1b\\>>\x1bP\x1b\x1b[0m\x1b\\')
+
+ @with_new_config
+ def test_term_truecolor_escapes(self, config):
+ from powerline.shell import ShellPowerline
+ import powerline as powerline_module
+ config['config']['common']['term_truecolor'] = True
+ with swap_attributes(config, powerline_module):
+ with get_powerline_raw(config, ShellPowerline, args=Args(config_path=[''])) as powerline:
+ self.assertEqual(powerline.render(segment_info={}, side='left'), '\x1b[0;38;2;192;0;192;48;2;0;128;128m\xa0s\x1b[0;38;2;0;128;128;49;22m>>\x1b[0m')
+
+ @with_new_config
+ def test_term_truecolor_fbterm_escapes(self, config):
+ from powerline.shell import ShellPowerline
+ import powerline as powerline_module
+ config['config']['common']['term_escape_style'] = 'fbterm'
+ config['config']['common']['term_truecolor'] = True
+ with swap_attributes(config, powerline_module):
+ with get_powerline_raw(config, ShellPowerline, args=Args(config_path=[''])) as powerline:
+ self.assertEqual(powerline.render(segment_info={}, side='left'), '\x1b[0m\x1b[1;5}\x1b[2;6}\xa0s\x1b[0m\x1b[1;6}\x1b[49m\x1b[22m>>\x1b[0m')
+
+ @with_new_config
+ def test_term_truecolor_tmux_escapes(self, config):
+ from powerline.shell import ShellPowerline
+ import powerline as powerline_module
+ config['config']['common']['term_truecolor'] = True
+ config['config']['common']['additional_escapes'] = 'tmux'
+ with swap_attributes(config, powerline_module):
+ with get_powerline_raw(config, ShellPowerline, args=Args(config_path=[''])) as powerline:
+ self.assertEqual(powerline.render(segment_info={}, side='left'), '\x1bPtmux;\x1b\x1b[0;38;2;192;0;192;48;2;0;128;128m\x1b\\\xa0s\x1bPtmux;\x1b\x1b[0;38;2;0;128;128;49;22m\x1b\\>>\x1bPtmux;\x1b\x1b[0m\x1b\\')
+
+ @with_new_config
+ def test_term_truecolor_screen_escapes(self, config):
+ from powerline.shell import ShellPowerline
+ import powerline as powerline_module
+ config['config']['common']['term_truecolor'] = True
+ config['config']['common']['additional_escapes'] = 'screen'
+ with swap_attributes(config, powerline_module):
+ with get_powerline_raw(config, ShellPowerline, args=Args(config_path=[''])) as powerline:
+ self.assertEqual(powerline.render(segment_info={}, side='left'), '\x1bP\x1b\x1b[0;38;2;192;0;192;48;2;0;128;128m\x1b\\\xa0s\x1bP\x1b\x1b[0;38;2;0;128;128;49;22m\x1b\\>>\x1bP\x1b\x1b[0m\x1b\\')
+
+
+class TestVim(TestCase):
+ def test_environ_update(self):
+ # Regression test: test that segment obtains environment from vim, not
+ # from os.environ.
+ with vim_module._with('globals', powerline_config_paths=['/']):
+ from powerline.vim import VimPowerline
+ import powerline as powerline_module
+ with swap_attributes(config, powerline_module):
+ with vim_module._with('environ', TEST='abc'):
+ with get_powerline_raw(config, VimPowerline) as powerline:
+ window = vim_module.current.window
+ window_id = 1
+ winnr = window.number
+ self.assertEqual(powerline.render(window, window_id, winnr), b'%#Pl_3_8404992_4_192_underline#\xc2\xa0abc%#Pl_4_192_NONE_None_NONE#>>')
+ vim_module._environ['TEST'] = 'def'
+ self.assertEqual(powerline.render(window, window_id, winnr), b'%#Pl_3_8404992_4_192_underline#\xc2\xa0def%#Pl_4_192_NONE_None_NONE#>>')
+
+ def test_local_themes(self):
+ # Regression test: VimPowerline.add_local_theme did not work properly.
+ from powerline.vim import VimPowerline
+ import powerline as powerline_module
+ with swap_attributes(config, powerline_module):
+ with get_powerline_raw(config, VimPowerline, replace_gcp=True) as powerline:
+ powerline.add_local_theme('tests.modules.matchers.always_true', {
+ 'segment_data': {
+ 'foo': {
+ 'contents': '“bar”'
+ }
+ },
+ 'segments': {
+ 'left': [
+ {
+ 'type': 'string',
+ 'name': 'foo',
+ 'highlight_groups': ['g1']
+ }
+ ]
+ }
+ })
+ window = vim_module.current.window
+ window_id = 1
+ winnr = window.number
+ self.assertEqual(powerline.render(window, window_id, winnr), b'%#Pl_5_12583104_6_32896_NONE#\xc2\xa0\xe2\x80\x9cbar\xe2\x80\x9d%#Pl_6_32896_NONE_None_NONE#>>')
+
+ @classmethod
+ def setUpClass(cls):
+ sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(os.path.dirname(__file__)), 'vim_sys_path')))
+
+ @classmethod
+ def tearDownClass(cls):
+ sys.path.pop(0)
+
+
+class TestLemonbar(TestRender):
+ def test_lemonbar(self):
+ import powerline as powerline_module
+ with swap_attributes(config, powerline_module):
+ with get_powerline_raw(config, powerline_module.Powerline, replace_gcp=True, ext='wm', renderer_module='lemonbar') as powerline:
+ self.assertRenderEqual(
+ powerline,
+ '%{l}%{F#ffc00000}%{B#ff008000}%{+u} A%{F-B--u}%{F#ff008000}%{B#ffc00000}>>%{F-B--u}%{F#ff008000}%{B#ffc00000}B%{F-B--u}%{F#ffc00000}>>%{F-B--u}%{r}%{F#ffc00000}<<%{F-B--u}%{F#ff804000}%{B#ffc00000}%{+u}C%{F-B--u}%{F#ff0000c0}%{B#ffc00000}<<%{F-B--u}%{F#ff008000}%{B#ff0000c0}D %{F-B--u}'
+ )
+
+ @with_new_config
+ def test_lemonbar_escape(self, config):
+ import powerline as powerline_module
+ config['themes/wm/default']['segments']['left'] = (
+ highlighted_string('%{asd}', 'hl1'),
+ highlighted_string('10% %', 'hl2'),
+ )
+ with swap_attributes(config, powerline_module):
+ with get_powerline_raw(config, powerline_module.Powerline, replace_gcp=True, ext='wm', renderer_module='lemonbar') as powerline:
+ self.assertRenderEqual(
+ powerline,
+ '%{l}%{F#ffc00000}%{B#ff008000}%{+u} %%{}{asd}%{F-B--u}%{F#ff008000}%{B#ffc00000}>>%{F-B--u}%{F#ff008000}%{B#ffc00000}10%%{} %%{}%{F-B--u}%{F#ffc00000}>>%{F-B--u}%{r}%{F#ffc00000}<<%{F-B--u}%{F#ff804000}%{B#ffc00000}%{+u}C%{F-B--u}%{F#ff0000c0}%{B#ffc00000}<<%{F-B--u}%{F#ff008000}%{B#ff0000c0}D %{F-B--u}'
+ )
+
+
+if __name__ == '__main__':
+ from tests.modules import main
+ main()
diff --git a/tests/test_python/test_lib.py b/tests/test_python/test_lib.py
new file mode 100644
index 0000000..4c85436
--- /dev/null
+++ b/tests/test_python/test_lib.py
@@ -0,0 +1,733 @@
+# vim:fileencoding=utf-8:noet
+from __future__ import (unicode_literals, division, absolute_import, print_function)
+
+import threading
+import os
+import sys
+import re
+import shutil
+import unicodedata
+
+from time import sleep
+from subprocess import call, PIPE
+
+from powerline.lib import add_divider_highlight_group
+from powerline.lib.dict import mergedicts, REMOVE_THIS_KEY
+from powerline.lib.humanize_bytes import humanize_bytes
+from powerline.lib.vcs import guess, get_fallback_create_watcher
+from powerline.lib.threaded import ThreadedSegment, KwThreadedSegment
+from powerline.lib.monotonic import monotonic
+from powerline.lib.vcs.git import git_directory
+from powerline.lib.shell import run_cmd
+
+import powerline.lib.unicode as plu
+
+from tests.modules.lib import Pl, replace_attr
+from tests.modules import TestCase, SkipTest
+
+
+try:
+ __import__('bzrlib')
+except ImportError:
+ use_bzr = False
+else:
+ use_bzr = True
+
+try:
+ __import__('hglib')
+except ImportError:
+ use_mercurial = False
+else:
+ use_mercurial = True
+
+
+GIT_REPO = 'git_repo'
+HG_REPO = 'hg_repo'
+BZR_REPO = 'bzr_repo'
+
+
+def thread_number():
+ return len(threading.enumerate())
+
+
+class TestShell(TestCase):
+ def test_run_cmd(self):
+ pl = Pl()
+ self.assertEqual(run_cmd(pl, ['xxx_nonexistent_command_xxx']), None)
+ self.assertEqual(len(pl.exceptions), 1)
+ pl = Pl()
+ self.assertEqual(run_cmd(pl, ['echo', ' test ']), 'test')
+ self.assertFalse(pl)
+ self.assertEqual(run_cmd(pl, ['echo', ' test '], strip=True), 'test')
+ self.assertFalse(pl)
+ self.assertEqual(run_cmd(pl, ['echo', ' test '], strip=False), ' test \n')
+ self.assertFalse(pl)
+ self.assertEqual(run_cmd(pl, ['cat'], stdin='test'), 'test')
+ self.assertFalse(pl)
+ self.assertEqual(run_cmd(pl, ['sh', '-c', 'cat >&2'], stdin='test'), '')
+ self.assertFalse(pl)
+
+
+class TestThreaded(TestCase):
+ def test_threaded_segment(self):
+ log = []
+ pl = Pl()
+ updates = [(None,)]
+ lock = threading.Lock()
+ event = threading.Event()
+ block_event = threading.Event()
+
+ class TestSegment(ThreadedSegment):
+ interval = 10
+
+ def set_state(self, **kwargs):
+ event.clear()
+ log.append(('set_state', kwargs))
+ return super(TestSegment, self).set_state(**kwargs)
+
+ def update(self, update_value):
+ block_event.wait()
+ event.set()
+ # Make sleep first to prevent some race conditions
+ log.append(('update', update_value))
+ with lock:
+ ret = updates[0]
+ if isinstance(ret, Exception):
+ raise ret
+ else:
+ return ret[0]
+
+ def render(self, update, **kwargs):
+ log.append(('render', update, kwargs))
+ if isinstance(update, Exception):
+ raise update
+ else:
+ return update
+
+ # Non-threaded tests
+ segment = TestSegment()
+ block_event.set()
+ updates[0] = (None,)
+ self.assertEqual(segment(pl=pl), None)
+ self.assertEqual(thread_number(), 1)
+ self.assertEqual(log, [
+ ('set_state', {}),
+ ('update', None),
+ ('render', None, {'pl': pl, 'update_first': True}),
+ ])
+ log[:] = ()
+
+ segment = TestSegment()
+ block_event.set()
+ updates[0] = ('abc',)
+ self.assertEqual(segment(pl=pl), 'abc')
+ self.assertEqual(thread_number(), 1)
+ self.assertEqual(log, [
+ ('set_state', {}),
+ ('update', None),
+ ('render', 'abc', {'pl': pl, 'update_first': True}),
+ ])
+ log[:] = ()
+
+ segment = TestSegment()
+ block_event.set()
+ updates[0] = ('abc',)
+ self.assertEqual(segment(pl=pl, update_first=False), 'abc')
+ self.assertEqual(thread_number(), 1)
+ self.assertEqual(log, [
+ ('set_state', {}),
+ ('update', None),
+ ('render', 'abc', {'pl': pl, 'update_first': False}),
+ ])
+ log[:] = ()
+
+ segment = TestSegment()
+ block_event.set()
+ updates[0] = ValueError('abc')
+ self.assertEqual(segment(pl=pl), None)
+ self.assertEqual(thread_number(), 1)
+ self.assertEqual(len(pl.exceptions), 1)
+ self.assertEqual(log, [
+ ('set_state', {}),
+ ('update', None),
+ ])
+ log[:] = ()
+ pl.exceptions[:] = ()
+
+ segment = TestSegment()
+ block_event.set()
+ updates[0] = (TypeError('def'),)
+ self.assertRaises(TypeError, segment, pl=pl)
+ self.assertEqual(thread_number(), 1)
+ self.assertEqual(log, [
+ ('set_state', {}),
+ ('update', None),
+ ('render', updates[0][0], {'pl': pl, 'update_first': True}),
+ ])
+ log[:] = ()
+
+ # Threaded tests
+ segment = TestSegment()
+ block_event.clear()
+ kwargs = {'pl': pl, 'update_first': False, 'other': 1}
+ with lock:
+ updates[0] = ('abc',)
+ segment.startup(**kwargs)
+ ret = segment(**kwargs)
+ self.assertEqual(thread_number(), 2)
+ block_event.set()
+ event.wait()
+ segment.shutdown_event.set()
+ segment.thread.join()
+ self.assertEqual(ret, None)
+ self.assertEqual(log, [
+ ('set_state', {'update_first': False, 'other': 1}),
+ ('render', None, {'pl': pl, 'update_first': False, 'other': 1}),
+ ('update', None),
+ ])
+ log[:] = ()
+
+ segment = TestSegment()
+ block_event.set()
+ kwargs = {'pl': pl, 'update_first': True, 'other': 1}
+ with lock:
+ updates[0] = ('def',)
+ segment.startup(**kwargs)
+ ret = segment(**kwargs)
+ self.assertEqual(thread_number(), 2)
+ segment.shutdown_event.set()
+ segment.thread.join()
+ self.assertEqual(ret, 'def')
+ self.assertEqual(log, [
+ ('set_state', {'update_first': True, 'other': 1}),
+ ('update', None),
+ ('render', 'def', {'pl': pl, 'update_first': True, 'other': 1}),
+ ])
+ log[:] = ()
+
+ segment = TestSegment()
+ block_event.set()
+ kwargs = {'pl': pl, 'update_first': True, 'interval': 0.2}
+ with lock:
+ updates[0] = ('abc',)
+ segment.startup(**kwargs)
+ start = monotonic()
+ ret1 = segment(**kwargs)
+ with lock:
+ updates[0] = ('def',)
+ self.assertEqual(thread_number(), 2)
+ sleep(0.5)
+ ret2 = segment(**kwargs)
+ segment.shutdown_event.set()
+ segment.thread.join()
+ end = monotonic()
+ duration = end - start
+ self.assertEqual(ret1, 'abc')
+ self.assertEqual(ret2, 'def')
+ self.assertEqual(log[:5], [
+ ('set_state', {'update_first': True, 'interval': 0.2}),
+ ('update', None),
+ ('render', 'abc', {'pl': pl, 'update_first': True, 'interval': 0.2}),
+ ('update', 'abc'),
+ ('update', 'def'),
+ ])
+ num_runs = len([e for e in log if e[0] == 'update'])
+ self.assertAlmostEqual(duration / 0.2, num_runs, delta=1)
+ log[:] = ()
+
+ segment = TestSegment()
+ block_event.set()
+ kwargs = {'pl': pl, 'update_first': True, 'interval': 0.2}
+ with lock:
+ updates[0] = ('ghi',)
+ segment.startup(**kwargs)
+ start = monotonic()
+ ret1 = segment(**kwargs)
+ with lock:
+ updates[0] = TypeError('jkl')
+ self.assertEqual(thread_number(), 2)
+ sleep(0.5)
+ ret2 = segment(**kwargs)
+ segment.shutdown_event.set()
+ segment.thread.join()
+ end = monotonic()
+ duration = end - start
+ self.assertEqual(ret1, 'ghi')
+ self.assertEqual(ret2, None)
+ self.assertEqual(log[:5], [
+ ('set_state', {'update_first': True, 'interval': 0.2}),
+ ('update', None),
+ ('render', 'ghi', {'pl': pl, 'update_first': True, 'interval': 0.2}),
+ ('update', 'ghi'),
+ ('update', 'ghi'),
+ ])
+ num_runs = len([e for e in log if e[0] == 'update'])
+ self.assertAlmostEqual(duration / 0.2, num_runs, delta=1)
+ self.assertEqual(num_runs - 1, len(pl.exceptions))
+ log[:] = ()
+
+ def test_kw_threaded_segment(self):
+ log = []
+ pl = Pl()
+ event = threading.Event()
+
+ class TestSegment(KwThreadedSegment):
+ interval = 10
+
+ @staticmethod
+ def key(_key=(None,), **kwargs):
+ log.append(('key', _key, kwargs))
+ return _key
+
+ def compute_state(self, key):
+ event.set()
+ sleep(0.1)
+ log.append(('compute_state', key))
+ ret = key
+ if isinstance(ret, Exception):
+ raise ret
+ else:
+ return ret[0]
+
+ def render_one(self, state, **kwargs):
+ log.append(('render_one', state, kwargs))
+ if isinstance(state, Exception):
+ raise state
+ else:
+ return state
+
+ # Non-threaded tests
+ segment = TestSegment()
+ event.clear()
+ self.assertEqual(segment(pl=pl), None)
+ self.assertEqual(thread_number(), 1)
+ self.assertEqual(log, [
+ ('key', (None,), {'pl': pl}),
+ ('compute_state', (None,)),
+ ('render_one', None, {'pl': pl}),
+ ])
+ log[:] = ()
+
+ segment = TestSegment()
+ kwargs = {'pl': pl, '_key': ('abc',), 'update_first': False}
+ event.clear()
+ self.assertEqual(segment(**kwargs), 'abc')
+ kwargs.update(_key=('def',))
+ self.assertEqual(segment(**kwargs), 'def')
+ self.assertEqual(thread_number(), 1)
+ self.assertEqual(log, [
+ ('key', ('abc',), {'pl': pl}),
+ ('compute_state', ('abc',)),
+ ('render_one', 'abc', {'pl': pl, '_key': ('abc',)}),
+ ('key', ('def',), {'pl': pl}),
+ ('compute_state', ('def',)),
+ ('render_one', 'def', {'pl': pl, '_key': ('def',)}),
+ ])
+ log[:] = ()
+
+ segment = TestSegment()
+ kwargs = {'pl': pl, '_key': ValueError('xyz'), 'update_first': False}
+ event.clear()
+ self.assertEqual(segment(**kwargs), None)
+ self.assertEqual(thread_number(), 1)
+ self.assertEqual(log, [
+ ('key', kwargs['_key'], {'pl': pl}),
+ ('compute_state', kwargs['_key']),
+ ])
+ log[:] = ()
+
+ segment = TestSegment()
+ kwargs = {'pl': pl, '_key': (ValueError('abc'),), 'update_first': False}
+ event.clear()
+ self.assertRaises(ValueError, segment, **kwargs)
+ self.assertEqual(thread_number(), 1)
+ self.assertEqual(log, [
+ ('key', kwargs['_key'], {'pl': pl}),
+ ('compute_state', kwargs['_key']),
+ ('render_one', kwargs['_key'][0], {'pl': pl, '_key': kwargs['_key']}),
+ ])
+ log[:] = ()
+
+ # Threaded tests
+ segment = TestSegment()
+ kwargs = {'pl': pl, 'update_first': False, '_key': ('_abc',)}
+ event.clear()
+ segment.startup(**kwargs)
+ ret = segment(**kwargs)
+ self.assertEqual(thread_number(), 2)
+ segment.shutdown_event.set()
+ segment.thread.join()
+ self.assertEqual(ret, None)
+ self.assertEqual(log[:2], [
+ ('key', kwargs['_key'], {'pl': pl}),
+ ('render_one', None, {'pl': pl, '_key': kwargs['_key']}),
+ ])
+ self.assertLessEqual(len(log), 3)
+ if len(log) > 2:
+ self.assertEqual(log[2], ('compute_state', kwargs['_key']))
+ log[:] = ()
+
+ segment = TestSegment()
+ kwargs = {'pl': pl, 'update_first': True, '_key': ('_abc',)}
+ event.clear()
+ segment.startup(**kwargs)
+ ret1 = segment(**kwargs)
+ kwargs.update(_key=('_def',))
+ ret2 = segment(**kwargs)
+ self.assertEqual(thread_number(), 2)
+ segment.shutdown_event.set()
+ segment.thread.join()
+ self.assertEqual(ret1, '_abc')
+ self.assertEqual(ret2, '_def')
+ self.assertEqual(log, [
+ ('key', ('_abc',), {'pl': pl}),
+ ('compute_state', ('_abc',)),
+ ('render_one', '_abc', {'pl': pl, '_key': ('_abc',)}),
+ ('key', ('_def',), {'pl': pl}),
+ ('compute_state', ('_def',)),
+ ('render_one', '_def', {'pl': pl, '_key': ('_def',)}),
+ ])
+ log[:] = ()
+
+
+class TestLib(TestCase):
+ 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'}})
+ mergedicts(d, {'abc': {'def': REMOVE_THIS_KEY}})
+ self.assertEqual(d, {'abc': {'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({str('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')
+
+
+width_data = {
+ 'N': 1, # Neutral
+ 'Na': 1, # Narrow
+ 'A': 1, # Ambiguous
+ 'H': 1, # Half-width
+ 'W': 2, # Wide
+ 'F': 2, # Fullwidth
+}
+
+
+class TestUnicode(TestCase):
+ def assertStringsIdentical(self, s1, s2):
+ self.assertTrue(type(s1) is type(s2), msg='string types differ')
+ self.assertEqual(s1, s2)
+
+ def test_unicode(self):
+ self.assertTrue(type('abc') is plu.unicode)
+
+ def test_unichr(self):
+ self.assertStringsIdentical('\U0010FFFF', plu.unichr(0x10FFFF))
+ self.assertStringsIdentical('\uFFFF', plu.unichr(0xFFFF))
+ self.assertStringsIdentical('\x20', plu.unichr(0x20))
+
+ def test_u(self):
+ self.assertStringsIdentical('Test', plu.u('Test'))
+ self.assertStringsIdentical('Test', plu.u(b'Test'))
+ self.assertStringsIdentical('«»', plu.u(b'\xC2\xAB\xC2\xBB'))
+ self.assertRaises(UnicodeDecodeError, plu.u, b'\xFF')
+
+ def test_tointiter(self):
+ self.assertEqual([1, 2, 3], list(plu.tointiter(b'\x01\x02\x03')))
+
+ def test_decode_error(self):
+ self.assertStringsIdentical('<FF>', b'\xFF'.decode('utf-8', 'powerline_decode_error'))
+ self.assertStringsIdentical('abc', b'abc'.decode('utf-8', 'powerline_decode_error'))
+
+ def test_register_strwidth_error(self):
+ ename = plu.register_strwidth_error(lambda s: 3)
+ self.assertStringsIdentical(b'???', 'A'.encode('latin1', ename))
+ self.assertStringsIdentical(b'abc', 'abc'.encode('latin1', ename))
+
+ def test_out_u(self):
+ self.assertStringsIdentical('abc', plu.out_u('abc'))
+ self.assertStringsIdentical('abc', plu.out_u(b'abc'))
+ self.assertRaises(TypeError, plu.out_u, None)
+
+ def test_safe_unicode(self):
+ self.assertStringsIdentical('abc', plu.safe_unicode('abc'))
+ self.assertStringsIdentical('abc', plu.safe_unicode(b'abc'))
+ self.assertStringsIdentical('«»', plu.safe_unicode(b'\xc2\xab\xc2\xbb'))
+ with replace_attr(plu, 'get_preferred_output_encoding', lambda: 'latin1'):
+ self.assertStringsIdentical('ÿ', plu.safe_unicode(b'\xFF'))
+ self.assertStringsIdentical('None', plu.safe_unicode(None))
+
+ class FailingStr(object):
+ def __str__(self):
+ raise NotImplementedError('Fail!')
+
+ self.assertStringsIdentical('Fail!', plu.safe_unicode(FailingStr()))
+
+ def test_FailedUnicode(self):
+ self.assertTrue(isinstance(plu.FailedUnicode('abc'), plu.unicode))
+ self.assertEqual('abc', plu.FailedUnicode('abc'))
+
+ def test_string(self):
+ self.assertStringsIdentical(str('abc'), plu.string('abc'))
+ self.assertStringsIdentical(str('abc'), plu.string(b'abc'))
+
+ def test_surrogate_pair_to_character(self):
+ self.assertEqual(0x1F48E, plu.surrogate_pair_to_character(0xD83D, 0xDC8E))
+
+ def test_strwidth_ucs_4(self):
+ self.assertEqual(4, plu.strwidth_ucs_4(width_data, 'abcd'))
+ self.assertEqual(4, plu.strwidth_ucs_4(width_data, 'AB'))
+ if sys.maxunicode < 0x10FFFF:
+ raise SkipTest('Can only test strwidth_ucs_4 in UCS-4 Pythons')
+
+ self.assertEqual(1, plu.strwidth_ucs_4(width_data, '\U0001F063'))
+
+ def test_strwidth_ucs_2(self):
+ self.assertEqual(4, plu.strwidth_ucs_2(width_data, 'abcd'))
+ self.assertEqual(4, plu.strwidth_ucs_2(width_data, 'AB'))
+ if not sys.maxunicode < 0x10FFFF:
+ raise SkipTest('Can only test strwidth_ucs_2 in UCS-2 Pythons')
+ self.assertEqual(1, plu.strwidth_ucs_2(width_data, '\ud83c\udc30'))
+
+
+class TestVCS(TestCase):
+ def do_branch_rename_test(self, repo, q):
+ st = monotonic()
+ while monotonic() - st < 1:
+ # Give inotify time to deliver events
+ ans = repo.branch()
+ if hasattr(q, '__call__'):
+ if q(ans):
+ break
+ else:
+ if ans == q:
+ break
+ sleep(0.01)
+ if hasattr(q, '__call__'):
+ self.assertTrue(q(ans))
+ else:
+ self.assertEqual(ans, q)
+
+ def test_git(self):
+ create_watcher = get_fallback_create_watcher()
+ repo = guess(path=GIT_REPO, create_watcher=create_watcher)
+ self.assertNotEqual(repo, None)
+ self.assertEqual(repo.branch(), 'master')
+ self.assertEqual(repo.status(), None)
+ 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'))
+ # Test changing branch
+ self.assertEqual(repo.branch(), 'master')
+ try:
+ call(['git', 'branch', 'branch1'], cwd=GIT_REPO)
+ call(['git', 'checkout', '-q', 'branch1'], cwd=GIT_REPO)
+ self.do_branch_rename_test(repo, 'branch1')
+ call(['git', 'branch', 'branch2'], cwd=GIT_REPO)
+ call(['git', 'checkout', '-q', 'branch2'], cwd=GIT_REPO)
+ self.do_branch_rename_test(repo, 'branch2')
+ call(['git', 'checkout', '-q', '--detach', 'branch1'], cwd=GIT_REPO)
+ self.do_branch_rename_test(repo, lambda b: re.match(r'^[a-f0-9]+$', b))
+ finally:
+ call(['git', 'checkout', '-q', 'master'], cwd=GIT_REPO)
+ # Test stashing
+ self.assertEqual(repo.stash(), 0)
+
+ def stash_save():
+ with open(os.path.join(GIT_REPO, 'file'), 'w') as f:
+ f.write('abc')
+ return call(['git', 'stash', '-u'], cwd=GIT_REPO, stdout=PIPE)
+
+ def stash_drop():
+ return call(['git', 'stash', 'drop'], cwd=GIT_REPO, stdout=PIPE)
+
+ def stash_list():
+ return call(['git', 'stash', 'list'], cwd=GIT_REPO, stdout=PIPE)
+
+ try:
+ stash_save()
+ self.assertEqual(repo.stash(), 1)
+ stash_save()
+ self.assertEqual(repo.stash(), 2)
+ stash_drop()
+ self.assertEqual(repo.stash(), 1)
+ stash_drop()
+ self.assertEqual(repo.stash(), 0)
+ finally:
+ while stash_list():
+ stash_drop()
+
+ def test_git_sym(self):
+ create_watcher = get_fallback_create_watcher()
+ dotgit = os.path.join(GIT_REPO, '.git')
+ spacegit = os.path.join(GIT_REPO, ' .git ')
+ os.rename(dotgit, spacegit)
+ try:
+ with open(dotgit, 'w') as F:
+ F.write('gitdir: .git \n')
+ gitdir = git_directory(GIT_REPO)
+ self.assertTrue(os.path.isdir(gitdir))
+ self.assertEqual(gitdir, os.path.abspath(spacegit))
+ repo = guess(path=GIT_REPO, create_watcher=create_watcher)
+ self.assertEqual(repo.branch(), 'master')
+ finally:
+ os.remove(dotgit)
+ os.rename(spacegit, dotgit)
+
+ def test_mercurial(self):
+ if not use_mercurial:
+ raise SkipTest('Mercurial is not available')
+ create_watcher = get_fallback_create_watcher()
+ repo = guess(path=HG_REPO, create_watcher=create_watcher)
+ self.assertNotEqual(repo, None)
+ self.assertEqual(repo.branch(), 'default')
+ self.assertEqual(repo.status(), None)
+ 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 test_bzr(self):
+ if not use_bzr:
+ raise SkipTest('Bazaar is not available')
+ create_watcher = get_fallback_create_watcher()
+ repo = guess(path=BZR_REPO, create_watcher=create_watcher)
+ self.assertNotEqual(repo, None, 'No bzr repo found. Do you have bzr installed?')
+ self.assertEqual(repo.branch(), 'test_powerline')
+ self.assertEqual(repo.status(), None)
+ with open(os.path.join(BZR_REPO, 'file'), 'w') as f:
+ f.write('abc')
+ self.assertEqual(repo.status(), ' U')
+ self.assertEqual(repo.status('file'), '? ')
+ call(['bzr', 'add', '-q', '.'], cwd=BZR_REPO, stdout=PIPE)
+ self.assertEqual(repo.status(), 'D ')
+ self.assertEqual(repo.status('file'), '+N')
+ call(['bzr', 'commit', '-q', '-m', 'initial commit'], cwd=BZR_REPO)
+ self.assertEqual(repo.status(), None)
+ with open(os.path.join(BZR_REPO, 'file'), 'w') as f:
+ f.write('def')
+ self.assertEqual(repo.status(), 'D ')
+ self.assertEqual(repo.status('file'), ' M')
+ self.assertEqual(repo.status('notexist'), None)
+ with open(os.path.join(BZR_REPO, 'ignored'), 'w') as f:
+ f.write('abc')
+ self.assertEqual(repo.status('ignored'), '? ')
+ # Test changing the .bzrignore file should update status
+ with open(os.path.join(BZR_REPO, '.bzrignore'), 'w') as f:
+ f.write('ignored')
+ self.assertEqual(repo.status('ignored'), None)
+ # Test changing the dirstate file should invalidate the cache for
+ # all files in the repo
+ with open(os.path.join(BZR_REPO, 'file2'), 'w') as f:
+ f.write('abc')
+ call(['bzr', 'add', 'file2'], cwd=BZR_REPO, stdout=PIPE)
+ call(['bzr', 'commit', '-q', '-m', 'file2 added'], cwd=BZR_REPO)
+ with open(os.path.join(BZR_REPO, 'file'), 'a') as f:
+ f.write('hello')
+ with open(os.path.join(BZR_REPO, 'file2'), 'a') as f:
+ f.write('hello')
+ self.assertEqual(repo.status('file'), ' M')
+ self.assertEqual(repo.status('file2'), ' M')
+ call(['bzr', 'commit', '-q', '-m', 'multi'], cwd=BZR_REPO)
+ self.assertEqual(repo.status('file'), None)
+ self.assertEqual(repo.status('file2'), None)
+
+ # Test changing branch
+ call(['bzr', 'nick', 'branch1'], cwd=BZR_REPO, stdout=PIPE, stderr=PIPE)
+ self.do_branch_rename_test(repo, 'branch1')
+
+ # Test branch name/status changes when swapping repos
+ for x in ('b1', 'b2'):
+ d = os.path.join(BZR_REPO, x)
+ os.mkdir(d)
+ call(['bzr', 'init', '-q'], cwd=d)
+ call(['bzr', 'nick', '-q', x], cwd=d)
+ repo = guess(path=d, create_watcher=create_watcher)
+ self.assertEqual(repo.branch(), x)
+ self.assertFalse(repo.status())
+ if x == 'b1':
+ open(os.path.join(d, 'dirty'), 'w').close()
+ self.assertTrue(repo.status())
+ os.rename(os.path.join(BZR_REPO, 'b1'), os.path.join(BZR_REPO, 'b'))
+ os.rename(os.path.join(BZR_REPO, 'b2'), os.path.join(BZR_REPO, 'b1'))
+ os.rename(os.path.join(BZR_REPO, 'b'), os.path.join(BZR_REPO, 'b2'))
+ for x, y in (('b1', 'b2'), ('b2', 'b1')):
+ d = os.path.join(BZR_REPO, x)
+ repo = guess(path=d, create_watcher=create_watcher)
+ self.do_branch_rename_test(repo, y)
+ if x == 'b1':
+ self.assertFalse(repo.status())
+ else:
+ self.assertTrue(repo.status())
+
+ @classmethod
+ def setUpClass(cls):
+ cls.powerline_old_cwd = os.getcwd()
+ os.chdir(os.path.dirname(os.path.dirname(__file__)))
+ call(['git', 'init', '--quiet', GIT_REPO])
+ assert os.path.isdir(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:
+ cls.powerline_old_HGRCPATH = os.environ.get('HGRCPATH')
+ os.environ['HGRCPATH'] = ''
+ 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')
+ if use_bzr:
+ call(['bzr', 'init', '--quiet', BZR_REPO])
+ call(['bzr', 'config', 'email=Foo <bar@example.org>'], cwd=BZR_REPO)
+ call(['bzr', 'config', 'nickname=test_powerline'], cwd=BZR_REPO)
+ call(['bzr', 'config', 'create_signatures=0'], cwd=BZR_REPO)
+
+ @classmethod
+ def tearDownClass(cls):
+ for repo_dir in [GIT_REPO] + ([HG_REPO] if use_mercurial else []) + ([BZR_REPO] if use_bzr else []):
+ shutil.rmtree(repo_dir)
+ if use_mercurial:
+ if cls.powerline_old_HGRCPATH is None:
+ os.environ.pop('HGRCPATH')
+ else:
+ os.environ['HGRCPATH'] = cls.powerline_old_HGRCPATH
+ os.chdir(cls.powerline_old_cwd)
+
+
+if __name__ == '__main__':
+ from tests.modules import main
+ main()
diff --git a/tests/test_python/test_lib_config.py b/tests/test_python/test_lib_config.py
new file mode 100644
index 0000000..053462a
--- /dev/null
+++ b/tests/test_python/test_lib_config.py
@@ -0,0 +1,52 @@
+# vim:fileencoding=utf-8:noet
+from __future__ import (unicode_literals, division, absolute_import, print_function)
+
+import os
+
+from powerline.lib.config import ConfigLoader
+
+from tests.modules import TestCase
+from tests.modules.lib.fsconfig import FSTree
+
+
+FILE_ROOT = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'cfglib')
+
+
+class LoadedList(list):
+ def pop_all(self):
+ try:
+ return self[:]
+ finally:
+ self[:] = ()
+
+
+loaded = LoadedList()
+
+
+def on_load(key):
+ loaded.append(key)
+
+
+def check_file(path):
+ if os.path.exists(path):
+ return path
+ else:
+ raise IOError
+
+
+class TestLoaderCondition(TestCase):
+ def test_update_missing(self):
+ loader = ConfigLoader(run_once=True)
+ fpath = os.path.join(FILE_ROOT, 'file.json')
+ self.assertRaises(IOError, loader.load, fpath)
+ loader.register_missing(check_file, on_load, fpath)
+ loader.update() # This line must not raise IOError
+ with FSTree({'file': {'test': 1}}, root=FILE_ROOT):
+ loader.update()
+ self.assertEqual(loader.load(fpath), {'test': 1})
+ self.assertEqual(loaded.pop_all(), [fpath])
+
+
+if __name__ == '__main__':
+ from tests.modules import main
+ main()
diff --git a/tests/test_python/test_listers.py b/tests/test_python/test_listers.py
new file mode 100644
index 0000000..f2368df
--- /dev/null
+++ b/tests/test_python/test_listers.py
@@ -0,0 +1,159 @@
+# vim:fileencoding=utf-8:noet
+from __future__ import (unicode_literals, division, absolute_import, print_function)
+
+import powerline.listers.i3wm as i3wm
+
+from tests.modules.lib import Args, replace_attr, Pl
+from tests.modules import TestCase
+
+
+class TestI3WM(TestCase):
+ workspaces = [
+ Args(name='1: w1', output='LVDS1', focused=False, urgent=False, visible=False),
+ Args(name='2: w2', output='LVDS1', focused=False, urgent=False, visible=True),
+ Args(name='3: w3', output='HDMI1', focused=False, urgent=True, visible=True),
+ Args(name='4: w4', output='DVI01', focused=True, urgent=True, visible=True),
+ ]
+
+ @staticmethod
+ def get_workspaces():
+ return iter(TestI3WM.workspaces)
+
+ @staticmethod
+ def get_outputs(pl):
+ return iter([
+ {'name': 'LVDS1'},
+ {'name': 'HDMI1'},
+ {'name': 'DVI01'},
+ ])
+
+ def test_output_lister(self):
+ pl = Pl()
+ with replace_attr(i3wm, 'get_connected_xrandr_outputs', self.get_outputs):
+ self.assertEqual(
+ list(i3wm.output_lister(pl=pl, segment_info={'a': 1})),
+ [
+ ({'a': 1, 'output': 'LVDS1'}, {'draw_inner_divider': None}),
+ ({'a': 1, 'output': 'HDMI1'}, {'draw_inner_divider': None}),
+ ({'a': 1, 'output': 'DVI01'}, {'draw_inner_divider': None}),
+ ]
+ )
+
+ def test_workspace_lister(self):
+ pl = Pl()
+ with replace_attr(i3wm, 'get_i3_connection', lambda: Args(get_workspaces=self.get_workspaces)):
+ self.assertEqual(
+ list(i3wm.workspace_lister(pl=pl, segment_info={'a': 1})),
+ [
+ ({
+ 'a': 1,
+ 'output': 'LVDS1',
+ 'workspace': self.workspaces[0],
+ }, {'draw_inner_divider': None}),
+ ({
+ 'a': 1,
+ 'output': 'LVDS1',
+ 'workspace': self.workspaces[1],
+ }, {'draw_inner_divider': None}),
+ ({
+ 'a': 1,
+ 'output': 'HDMI1',
+ 'workspace': self.workspaces[2],
+ }, {'draw_inner_divider': None}),
+ ({
+ 'a': 1,
+ 'output': 'DVI01',
+ 'workspace': self.workspaces[3],
+ }, {'draw_inner_divider': None}),
+ ]
+ )
+
+ self.assertEqual(
+ list(i3wm.workspace_lister(pl=pl, segment_info={'a': 1}, output='LVDS1')),
+ [
+ ({
+ 'a': 1,
+ 'output': 'LVDS1',
+ 'workspace': self.workspaces[0],
+ }, {'draw_inner_divider': None}),
+ ({
+ 'a': 1,
+ 'output': 'LVDS1',
+ 'workspace': self.workspaces[1],
+ }, {'draw_inner_divider': None}),
+ ]
+ )
+
+ self.assertEqual(
+ list(i3wm.workspace_lister(
+ pl=pl,
+ segment_info={'a': 1, 'output': 'LVDS1'}
+ )),
+ [
+ ({
+ 'a': 1,
+ 'output': 'LVDS1',
+ 'workspace': self.workspaces[0],
+ }, {'draw_inner_divider': None}),
+ ({
+ 'a': 1,
+ 'output': 'LVDS1',
+ 'workspace': self.workspaces[1],
+ }, {'draw_inner_divider': None}),
+ ]
+ )
+
+ self.assertEqual(
+ list(i3wm.workspace_lister(
+ pl=pl,
+ segment_info={'a': 1, 'output': 'LVDS1'},
+ output=False
+ )),
+ [
+ ({
+ 'a': 1,
+ 'output': 'LVDS1',
+ 'workspace': self.workspaces[0],
+ }, {'draw_inner_divider': None}),
+ ({
+ 'a': 1,
+ 'output': 'LVDS1',
+ 'workspace': self.workspaces[1],
+ }, {'draw_inner_divider': None}),
+ ({
+ 'a': 1,
+ 'output': 'HDMI1',
+ 'workspace': self.workspaces[2],
+ }, {'draw_inner_divider': None}),
+ ({
+ 'a': 1,
+ 'output': 'DVI01',
+ 'workspace': self.workspaces[3],
+ }, {'draw_inner_divider': None}),
+ ]
+ )
+
+ self.assertEqual(
+ list(i3wm.workspace_lister(
+ pl=pl,
+ segment_info={'a': 1},
+ only_show=['focused', 'urgent']
+ )),
+ [
+ ({
+ 'a': 1,
+ 'output': 'HDMI1',
+ 'workspace': self.workspaces[2],
+ }, {'draw_inner_divider': None}),
+ ({
+ 'a': 1,
+ 'output': 'DVI01',
+ 'workspace': self.workspaces[3],
+ }, {'draw_inner_divider': None}),
+ ]
+ )
+
+
+if __name__ == '__main__':
+ from tests.modules import main
+ main()
diff --git a/tests/test_python/test_logging.py b/tests/test_python/test_logging.py
new file mode 100644
index 0000000..d7cfe4a
--- /dev/null
+++ b/tests/test_python/test_logging.py
@@ -0,0 +1,467 @@
+# vim:fileencoding=utf-8:noet
+
+'''Tests for various logging features'''
+
+from __future__ import (unicode_literals, division, absolute_import, print_function)
+
+import sys
+import re
+import codecs
+import os
+
+from io import StringIO
+from shutil import rmtree
+
+from powerline import finish_common_config, create_logger
+
+from tests.modules import TestCase
+from tests.modules.lib import replace_attr
+
+
+TIMESTAMP_RE = r'\d{4}-\d\d-\d\d \d\d:\d\d:\d\d,\d{3}'
+
+
+class TestRE(TestCase):
+ def assertMatches(self, text, regexp):
+ self.assertTrue(
+ re.match(regexp, text),
+ '{0!r} did not match {1!r}'.format(text, regexp),
+ )
+
+
+def close_handlers(logger):
+ for handler in logger.handlers:
+ handler.close()
+
+
+class TestHandlers(TestRE):
+ def test_stderr_handler_is_default(self):
+ out = StringIO()
+ err = StringIO()
+
+ with replace_attr(sys, 'stdout', out, 'stderr', err):
+ common_config = finish_common_config('utf-8', {})
+ logger, pl, get_module_attr = create_logger(common_config)
+ pl.error('Foo')
+ close_handlers(logger)
+ self.assertMatches(err.getvalue(), '^' + TIMESTAMP_RE + ':ERROR:__unknown__:Foo\n$')
+ self.assertEqual(out.getvalue(), '')
+
+ def test_stream_override(self):
+ out = StringIO()
+ err = StringIO()
+ stream = StringIO()
+
+ with replace_attr(sys, 'stdout', out, 'stderr', err):
+ common_config = finish_common_config('utf-8', {})
+ logger, pl, get_module_attr = create_logger(common_config, stream=stream)
+ pl.error('Foo')
+ close_handlers(logger)
+ self.assertMatches(stream.getvalue(), '^' + TIMESTAMP_RE + ':ERROR:__unknown__:Foo\n$')
+ self.assertEqual(err.getvalue(), '')
+ self.assertEqual(out.getvalue(), '')
+
+ def test_explicit_none(self):
+ out = StringIO()
+ err = StringIO()
+ stream = StringIO()
+
+ with replace_attr(sys, 'stdout', out, 'stderr', err):
+ common_config = finish_common_config('utf-8', {'log_file': [None]})
+ logger, pl, get_module_attr = create_logger(common_config, stream=stream)
+ pl.error('Foo')
+ close_handlers(logger)
+ self.assertMatches(stream.getvalue(), '^' + TIMESTAMP_RE + ':ERROR:__unknown__:Foo\n$')
+ self.assertEqual(err.getvalue(), '')
+ self.assertEqual(out.getvalue(), '')
+
+ def test_explicit_stream_handler(self):
+ out = StringIO()
+ err = StringIO()
+ stream = StringIO()
+
+ with replace_attr(sys, 'stdout', out, 'stderr', err):
+ common_config = finish_common_config('utf-8', {'log_file': [['logging.StreamHandler', [[]]]]})
+ logger, pl, get_module_attr = create_logger(common_config, stream=stream)
+ pl.error('Foo')
+ close_handlers(logger)
+ self.assertEqual(stream.getvalue(), '')
+ self.assertMatches(err.getvalue(), '^' + TIMESTAMP_RE + ':ERROR:__unknown__:Foo\n$')
+ self.assertEqual(out.getvalue(), '')
+
+ def test_explicit_stream_handler_implicit_stream(self):
+ out = StringIO()
+ err = StringIO()
+ stream = StringIO()
+
+ with replace_attr(sys, 'stdout', out, 'stderr', err):
+ common_config = finish_common_config('utf-8', {'log_file': [['logging.StreamHandler', []]]})
+ logger, pl, get_module_attr = create_logger(common_config, stream=stream)
+ pl.error('Foo')
+ close_handlers(logger)
+ self.assertMatches(stream.getvalue(), '^' + TIMESTAMP_RE + ':ERROR:__unknown__:Foo\n$')
+ self.assertEqual(err.getvalue(), '')
+ self.assertEqual(out.getvalue(), '')
+
+ def test_file_handler(self):
+ out = StringIO()
+ err = StringIO()
+ stream = StringIO()
+ file_name = 'test_logging-test_file_handler'
+
+ with replace_attr(sys, 'stdout', out, 'stderr', err):
+ common_config = finish_common_config('utf-8', {'log_file': file_name})
+ try:
+ logger, pl, get_module_attr = create_logger(common_config, stream=stream)
+ pl.error('Foo')
+ close_handlers(logger)
+ with codecs.open(file_name, encoding='utf-8') as fp:
+ self.assertMatches(fp.read(), '^' + TIMESTAMP_RE + ':ERROR:__unknown__:Foo\n$')
+ finally:
+ os.unlink(file_name)
+ self.assertEqual(stream.getvalue(), '')
+ self.assertEqual(err.getvalue(), '')
+ self.assertEqual(out.getvalue(), '')
+
+ def test_file_handler_create_dir(self):
+ out = StringIO()
+ err = StringIO()
+ stream = StringIO()
+ file_name = 'test_logging-test_file_handler_create_dir/file'
+
+ self.assertFalse(os.path.isdir(os.path.dirname(file_name)))
+
+ with replace_attr(sys, 'stdout', out, 'stderr', err):
+ common_config = finish_common_config('utf-8', {'log_file': file_name})
+ try:
+ logger, pl, get_module_attr = create_logger(common_config, stream=stream)
+ pl.error('Foo')
+ close_handlers(logger)
+ self.assertTrue(os.path.isdir(os.path.dirname(file_name)))
+ with codecs.open(file_name, encoding='utf-8') as fp:
+ self.assertMatches(fp.read(), '^' + TIMESTAMP_RE + ':ERROR:__unknown__:Foo\n$')
+ finally:
+ rmtree(os.path.dirname(file_name))
+ self.assertEqual(stream.getvalue(), '')
+ self.assertEqual(err.getvalue(), '')
+ self.assertEqual(out.getvalue(), '')
+
+ def test_multiple_files(self):
+ out = StringIO()
+ err = StringIO()
+ stream = StringIO()
+ file_name_1 = 'test_logging-test_multiple_files-1'
+ file_name_2 = file_name_1[:-1] + '2'
+
+ with replace_attr(sys, 'stdout', out, 'stderr', err):
+ common_config = finish_common_config('utf-8', {'log_file': [file_name_1, file_name_2]})
+ try:
+ try:
+ logger, pl, get_module_attr = create_logger(common_config, stream=stream)
+ pl.error('Foo')
+ close_handlers(logger)
+ for file_name in (file_name_1, file_name_2):
+ with codecs.open(file_name, encoding='utf-8') as fp:
+ self.assertMatches(fp.read(), '^' + TIMESTAMP_RE + ':ERROR:__unknown__:Foo\n$')
+ finally:
+ os.unlink(file_name_1)
+ finally:
+ os.unlink(file_name_2)
+ self.assertEqual(stream.getvalue(), '')
+ self.assertEqual(err.getvalue(), '')
+ self.assertEqual(out.getvalue(), '')
+
+ def test_multiple_files_and_stream(self):
+ out = StringIO()
+ err = StringIO()
+ stream = StringIO()
+ file_name_1 = 'test_logging-test_multiple_files_and_stream-1'
+ file_name_2 = file_name_1[:-1] + '2'
+
+ with replace_attr(sys, 'stdout', out, 'stderr', err):
+ common_config = finish_common_config('utf-8', {'log_file': [file_name_1, file_name_2, None]})
+ try:
+ try:
+ logger, pl, get_module_attr = create_logger(common_config, stream=stream)
+ pl.error('Foo')
+ close_handlers(logger)
+ for file_name in (file_name_1, file_name_2):
+ with codecs.open(file_name, encoding='utf-8') as fp:
+ self.assertMatches(fp.read(), '^' + TIMESTAMP_RE + ':ERROR:__unknown__:Foo\n$')
+ finally:
+ os.unlink(file_name_1)
+ finally:
+ os.unlink(file_name_2)
+ self.assertMatches(stream.getvalue(), '^' + TIMESTAMP_RE + ':ERROR:__unknown__:Foo\n$')
+ self.assertEqual(err.getvalue(), '')
+ self.assertEqual(out.getvalue(), '')
+
+ def test_handler_args(self):
+ out = StringIO()
+ err = StringIO()
+ stream = StringIO()
+ file_name = 'test_logging-test_handler_args'
+
+ with replace_attr(sys, 'stdout', out, 'stderr', err):
+ common_config = finish_common_config('utf-8', {'log_file': [
+ ['RotatingFileHandler', [[file_name]]]
+ ]})
+ try:
+ logger, pl, get_module_attr = create_logger(common_config, stream=stream)
+ pl.error('Foo')
+ close_handlers(logger)
+ with codecs.open(file_name, encoding='utf-8') as fp:
+ self.assertMatches(fp.read(), '^' + TIMESTAMP_RE + ':ERROR:__unknown__:Foo\n$')
+ finally:
+ os.unlink(file_name)
+ self.assertEqual(stream.getvalue(), '')
+ self.assertEqual(err.getvalue(), '')
+ self.assertEqual(out.getvalue(), '')
+
+ def test_handler_args_kwargs(self):
+ out = StringIO()
+ err = StringIO()
+ stream = StringIO()
+ file_name = 'test_logging-test_handler_args_kwargs'
+
+ with replace_attr(sys, 'stdout', out, 'stderr', err):
+ common_config = finish_common_config('utf-8', {'log_file': [
+ ['RotatingFileHandler', [[file_name], {'maxBytes': 1, 'backupCount': 1}]]
+ ]})
+ try:
+ try:
+ logger, pl, get_module_attr = create_logger(common_config, stream=stream)
+ pl.error('Foo')
+ pl.error('Bar')
+ close_handlers(logger)
+ with codecs.open(file_name, encoding='utf-8') as fp:
+ self.assertMatches(fp.read(), '^' + TIMESTAMP_RE + ':ERROR:__unknown__:Bar\n$')
+ with codecs.open(file_name + '.1', encoding='utf-8') as fp:
+ self.assertMatches(fp.read(), '^' + TIMESTAMP_RE + ':ERROR:__unknown__:Foo\n$')
+ finally:
+ os.unlink(file_name + '.1')
+ finally:
+ os.unlink(file_name)
+ self.assertEqual(stream.getvalue(), '')
+ self.assertEqual(err.getvalue(), '')
+ self.assertEqual(out.getvalue(), '')
+
+ def test_logger_level(self):
+ out = StringIO()
+ err = StringIO()
+ stream = StringIO()
+ stream1 = StringIO()
+ stream2 = StringIO()
+
+ with replace_attr(sys, 'stdout', out, 'stderr', err):
+ common_config = finish_common_config('utf-8', {'log_file': [
+ ['logging.StreamHandler', [[stream1]], 'WARNING'],
+ ['logging.StreamHandler', [[stream2]], 'ERROR'],
+ ]})
+ logger, pl, get_module_attr = create_logger(common_config, stream=stream)
+ pl.warn('Foo')
+ pl.error('Bar')
+ close_handlers(logger)
+ self.assertMatches(stream1.getvalue(), (
+ '^' + TIMESTAMP_RE + ':WARNING:__unknown__:Foo\n'
+ + TIMESTAMP_RE + ':ERROR:__unknown__:Bar\n$'
+ ))
+ self.assertMatches(stream2.getvalue(), '^' + TIMESTAMP_RE + ':ERROR:__unknown__:Bar\n$')
+ self.assertEqual(stream.getvalue(), '')
+ self.assertEqual(err.getvalue(), '')
+ self.assertEqual(out.getvalue(), '')
+
+ def test_logger_level_not_overriding_default(self):
+ out = StringIO()
+ err = StringIO()
+ stream = StringIO()
+ stream1 = StringIO()
+
+ with replace_attr(sys, 'stdout', out, 'stderr', err):
+ common_config = finish_common_config('utf-8', {'log_file': [
+ ['logging.StreamHandler', [[stream1]], 'DEBUG'],
+ ]})
+ logger, pl, get_module_attr = create_logger(common_config, stream=stream)
+ pl.debug('Foo')
+ pl.error('Bar')
+ close_handlers(logger)
+ self.assertMatches(stream1.getvalue(), '^' + TIMESTAMP_RE + ':ERROR:__unknown__:Bar\n$')
+ self.assertEqual(stream.getvalue(), '')
+ self.assertEqual(err.getvalue(), '')
+ self.assertEqual(out.getvalue(), '')
+
+ def test_top_log_level(self):
+ out = StringIO()
+ err = StringIO()
+ stream = StringIO()
+ stream1 = StringIO()
+
+ with replace_attr(sys, 'stdout', out, 'stderr', err):
+ common_config = finish_common_config('utf-8', {'log_file': [
+ ['logging.StreamHandler', [[stream1]], 'DEBUG'],
+ ], 'log_level': 'DEBUG'})
+ logger, pl, get_module_attr = create_logger(common_config, stream=stream)
+ pl.debug('Foo')
+ pl.error('Bar')
+ close_handlers(logger)
+ self.assertMatches(stream1.getvalue(), (
+ '^' + TIMESTAMP_RE + ':DEBUG:__unknown__:Foo\n'
+ + TIMESTAMP_RE + ':ERROR:__unknown__:Bar\n$'
+ ))
+ self.assertEqual(stream.getvalue(), '')
+ self.assertEqual(err.getvalue(), '')
+ self.assertEqual(out.getvalue(), '')
+
+ def test_logger_format(self):
+ out = StringIO()
+ err = StringIO()
+ stream = StringIO()
+ stream1 = StringIO()
+
+ with replace_attr(sys, 'stdout', out, 'stderr', err):
+ common_config = finish_common_config('utf-8', {'log_file': [
+ ['logging.StreamHandler', [[stream1]], 'WARNING', 'FOO'],
+ ]})
+ logger, pl, get_module_attr = create_logger(common_config, stream=stream)
+ pl.warn('Foo')
+ pl.error('Bar')
+ close_handlers(logger)
+ self.assertEqual(stream1.getvalue(), 'FOO\nFOO\n')
+ self.assertEqual(stream.getvalue(), '')
+ self.assertEqual(err.getvalue(), '')
+ self.assertEqual(out.getvalue(), '')
+
+ def test_top_log_format(self):
+ out = StringIO()
+ err = StringIO()
+ stream = StringIO()
+ stream1 = StringIO()
+ stream2 = StringIO()
+
+ with replace_attr(sys, 'stdout', out, 'stderr', err):
+ common_config = finish_common_config('utf-8', {'log_file': [
+ ['logging.StreamHandler', [[stream1]], 'WARNING', 'FOO'],
+ ['logging.StreamHandler', [[stream2]], 'WARNING'],
+ ], 'log_format': 'BAR'})
+ logger, pl, get_module_attr = create_logger(common_config, stream=stream)
+ pl.warn('Foo')
+ pl.error('Bar')
+ close_handlers(logger)
+ self.assertEqual(stream2.getvalue(), 'BAR\nBAR\n')
+ self.assertEqual(stream1.getvalue(), 'FOO\nFOO\n')
+ self.assertEqual(stream.getvalue(), '')
+ self.assertEqual(err.getvalue(), '')
+ self.assertEqual(out.getvalue(), '')
+
+
+class TestPowerlineLogger(TestRE):
+ def test_args_formatting(self):
+ stream = StringIO()
+
+ common_config = finish_common_config('utf-8', {})
+ logger, pl, get_module_attr = create_logger(common_config, stream=stream)
+ pl.warn('foo {0}', 'Test')
+ pl.warn('bar {0!r}', 'Test')
+ close_handlers(logger)
+ self.assertMatches(stream.getvalue(), (
+ '^' + TIMESTAMP_RE + ':WARNING:__unknown__:foo Test\n'
+ + TIMESTAMP_RE + ':WARNING:__unknown__:bar u?\'Test\'\n$'
+ ))
+
+ def test_prefix_formatting(self):
+ stream = StringIO()
+
+ common_config = finish_common_config('utf-8', {})
+ logger, pl, get_module_attr = create_logger(common_config, stream=stream)
+ pl.prefix = '1'
+ pl.warn('foo')
+ pl.prefix = '2'
+ pl.warn('bar')
+ close_handlers(logger)
+ self.assertMatches(stream.getvalue(), (
+ '^' + TIMESTAMP_RE + ':WARNING:__unknown__:1:foo\n'
+ + TIMESTAMP_RE + ':WARNING:__unknown__:2:bar\n$'
+ ))
+
+ def test_kwargs_formatting(self):
+ stream = StringIO()
+
+ common_config = finish_common_config('utf-8', {})
+ logger, pl, get_module_attr = create_logger(common_config, stream=stream)
+ pl.warn('foo {arg}', arg='Test')
+ pl.warn('bar {arg!r}', arg='Test')
+ close_handlers(logger)
+ self.assertMatches(stream.getvalue(), (
+ '^' + TIMESTAMP_RE + ':WARNING:__unknown__:foo Test\n'
+ + TIMESTAMP_RE + ':WARNING:__unknown__:bar u?\'Test\'\n$'
+ ))
+
+ def test_args_kwargs_formatting(self):
+ stream = StringIO()
+
+ common_config = finish_common_config('utf-8', {})
+ logger, pl, get_module_attr = create_logger(common_config, stream=stream)
+ pl.warn('foo {0!r} {arg}', 'Test0', arg='Test')
+ pl.warn('bar {0} {arg!r}', 'Test0', arg='Test')
+ close_handlers(logger)
+ self.assertMatches(stream.getvalue(), (
+ '^' + TIMESTAMP_RE + ':WARNING:__unknown__:foo u?\'Test0\' Test\n'
+ + TIMESTAMP_RE + ':WARNING:__unknown__:bar Test0 u?\'Test\'\n$'
+ ))
+
+ def test_exception_formatting(self):
+ stream = StringIO()
+
+ common_config = finish_common_config('utf-8', {})
+ logger, pl, get_module_attr = create_logger(common_config, stream=stream)
+ try:
+ raise ValueError('foo')
+ except ValueError:
+ pl.exception('Message')
+ close_handlers(logger)
+ self.assertMatches(stream.getvalue(), (
+ '^' + TIMESTAMP_RE + ':ERROR:__unknown__:Message\n'
+ + 'Traceback \\(most recent call last\\):\n'
+ + '(?: File ".*?", line \\d+, in \\w+\n [^\n]*\n)+'
+ + 'ValueError: foo\n$'
+ ))
+
+ def test_levels(self):
+ stream = StringIO()
+
+ common_config = finish_common_config('utf-8', {'log_level': 'DEBUG'})
+ logger, pl, get_module_attr = create_logger(common_config, stream=stream)
+ pl.debug('1')
+ pl.info('2')
+ pl.warn('3')
+ pl.error('4')
+ pl.critical('5')
+ close_handlers(logger)
+ self.assertMatches(stream.getvalue(), (
+ '^' + TIMESTAMP_RE + ':DEBUG:__unknown__:1\n'
+ + TIMESTAMP_RE + ':INFO:__unknown__:2\n'
+ + TIMESTAMP_RE + ':WARNING:__unknown__:3\n'
+ + TIMESTAMP_RE + ':ERROR:__unknown__:4\n'
+ + TIMESTAMP_RE + ':CRITICAL:__unknown__:5\n$'
+ ))
+
+
+old_cwd = None
+
+
+def setUpModule():
+ global old_cwd
+ global __file__
+ old_cwd = os.getcwd()
+ __file__ = os.path.abspath(__file__)
+ os.chdir(os.path.dirname(os.path.dirname(__file__)))
+
+
+def tearDownModule():
+ global old_cwd
+ os.chdir(old_cwd)
+
+
+if __name__ == '__main__':
+ from tests.modules import main
+ main()
diff --git a/tests/test_python/test_provided_config_files.py b/tests/test_python/test_provided_config_files.py
new file mode 100644
index 0000000..fd8b16e
--- /dev/null
+++ b/tests/test_python/test_provided_config_files.py
@@ -0,0 +1,201 @@
+# vim:fileencoding=utf-8:noet
+
+'''Dynamic configuration files tests.'''
+
+from __future__ import (unicode_literals, division, absolute_import, print_function)
+
+import sys
+import os
+import json
+import logging
+
+import tests.modules.vim as vim_module
+
+from tests.modules.lib import Args, urllib_read, replace_attr
+from tests.modules import TestCase
+
+from powerline import NotInterceptedError
+from powerline.segments.common import wthr
+
+
+VBLOCK = chr(ord('V') - 0x40)
+SBLOCK = chr(ord('S') - 0x40)
+
+
+class FailingLogger(logging.Logger):
+ def exception(self, *args, **kwargs):
+ super(FailingLogger, self).exception(*args, **kwargs)
+ raise NotInterceptedError('Unexpected exception occurred')
+
+
+def get_logger(stream=None):
+ log_format = '%(asctime)s:%(levelname)s:%(message)s'
+ formatter = logging.Formatter(log_format)
+
+ level = logging.WARNING
+ handler = logging.StreamHandler(stream)
+ handler.setLevel(level)
+ handler.setFormatter(formatter)
+
+ logger = FailingLogger('powerline')
+ logger.setLevel(level)
+ logger.addHandler(handler)
+ return logger
+
+
+class TestVimConfig(TestCase):
+ def test_vim(self):
+ from powerline.vim import VimPowerline
+ cfg_path = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), 'powerline', 'config_files')
+ buffers = (
+ (('bufoptions',), {'buftype': 'help'}),
+ (('bufname', '[Command Line]'), {}),
+ (('bufoptions',), {'buftype': 'quickfix'}),
+ (('bufname', 'NERD_tree_1'), {}),
+ (('bufname', '__Gundo__'), {}),
+ (('bufname', '__Gundo_Preview__'), {}),
+ # No Command-T tests here: requires +ruby or emulation
+ # No tabline here: tablines are tested separately
+ )
+ with open(os.path.join(cfg_path, 'config.json'), 'r') as f:
+ local_themes_raw = json.load(f)['ext']['vim']['local_themes']
+ # Don’t run tests on external/plugin segments
+ local_themes = dict((k, v) for (k, v) in local_themes_raw.items())
+ # See end of the buffers definition above for `- 2`
+ self.assertEqual(len(buffers), len(local_themes) - 2)
+ outputs = {}
+ i = 0
+
+ with vim_module._with('split'):
+ with VimPowerline(logger=get_logger()) as powerline:
+ def check_output(mode, args, kwargs):
+ if mode == 'nc':
+ window = vim_module.windows[0]
+ window_id = 2
+ else:
+ vim_module._start_mode(mode)
+ window = vim_module.current.window
+ window_id = 1
+ winnr = window.number
+ out = powerline.render(window, window_id, winnr)
+ if out in outputs:
+ self.fail('Duplicate in set #{0} ({1}) for mode {2!r} (previously defined in set #{3} ({4!r}) for mode {5!r})'.format(i, (args, kwargs), mode, *outputs[out]))
+ outputs[out] = (i, (args, kwargs), mode)
+
+ with vim_module._with('bufname', '/tmp/foo.txt'):
+ out = powerline.render(vim_module.current.window, 1, vim_module.current.window.number, is_tabline=True)
+ outputs[out] = (-1, (None, None), 'tab')
+ with vim_module._with('globals', powerline_config_paths=[cfg_path]):
+ 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?', '!']:
+ check_output(mode, None, None)
+ for args, kwargs in buffers:
+ i += 1
+ if mode in exclude:
+ continue
+ with vim_module._with(*args, **kwargs):
+ check_output(mode, args, kwargs)
+ finally:
+ vim_module._start_mode('n')
+
+ @classmethod
+ def setUpClass(cls):
+ sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(os.path.dirname(__file__)), 'vim_sys_path')))
+
+ @classmethod
+ def tearDownClass(cls):
+ sys.path.pop(0)
+
+
+class TestConfig(TestCase):
+ def test_tmux(self):
+ from powerline.segments import common
+ from imp import reload
+ reload(common)
+ from powerline.shell import ShellPowerline
+ with replace_attr(common, 'urllib_read', urllib_read):
+ with ShellPowerline(Args(ext=['tmux']), logger=get_logger(), run_once=False) as powerline:
+ powerline.render()
+ with ShellPowerline(Args(ext=['tmux']), logger=get_logger(), run_once=False) as powerline:
+ powerline.render()
+
+ def test_zsh(self):
+ from powerline.shell import ShellPowerline
+ args = Args(last_pipe_status=[1, 0], jobnum=0, ext=['shell'], renderer_module='.zsh')
+ segment_info = {'args': args}
+ with ShellPowerline(args, logger=get_logger(), run_once=False) as powerline:
+ powerline.render(segment_info=segment_info)
+ with ShellPowerline(args, logger=get_logger(), run_once=False) as powerline:
+ powerline.render(segment_info=segment_info)
+ segment_info['local_theme'] = 'select'
+ with ShellPowerline(args, logger=get_logger(), run_once=False) as powerline:
+ powerline.render(segment_info=segment_info)
+ segment_info['local_theme'] = 'continuation'
+ segment_info['parser_state'] = 'if cmdsubst'
+ with ShellPowerline(args, logger=get_logger(), run_once=False) as powerline:
+ powerline.render(segment_info=segment_info)
+
+ def test_bash(self):
+ from powerline.shell import ShellPowerline
+ args = Args(last_exit_code=1, last_pipe_status=[], jobnum=0, ext=['shell'], renderer_module='.bash', config_override={'ext': {'shell': {'theme': 'default_leftonly'}}})
+ with ShellPowerline(args, logger=get_logger(), run_once=False) as powerline:
+ powerline.render(segment_info={'args': args})
+ with ShellPowerline(args, logger=get_logger(), run_once=False) as powerline:
+ powerline.render(segment_info={'args': args})
+
+ def test_ipython(self):
+ from powerline.ipython import IPythonPowerline
+
+ class IpyPowerline(IPythonPowerline):
+ config_paths = None
+ config_overrides = None
+ theme_overrides = {}
+
+ segment_info = Args(prompt_count=1)
+
+ with IpyPowerline(logger=get_logger(), renderer_module='.pre_5') as powerline:
+ for prompt_type in ['in', 'in2']:
+ powerline.render(is_prompt=True, matcher_info=prompt_type, segment_info=segment_info)
+ powerline.render(is_prompt=True, matcher_info=prompt_type, segment_info=segment_info)
+ with IpyPowerline(logger=get_logger(), renderer_module='.pre_5') as powerline:
+ for prompt_type in ['out', 'rewrite']:
+ powerline.render(is_prompt=False, matcher_info=prompt_type, segment_info=segment_info)
+ powerline.render(is_prompt=False, matcher_info=prompt_type, segment_info=segment_info)
+
+ def test_wm(self):
+ from powerline.segments import common
+ from imp import reload
+ reload(common)
+ from powerline import Powerline
+ with replace_attr(wthr, 'urllib_read', urllib_read):
+ Powerline(logger=get_logger(), ext='wm', renderer_module='pango_markup', run_once=True).render()
+ reload(common)
+
+
+old_cwd = None
+saved_get_config_paths = None
+
+
+def setUpModule():
+ global old_cwd
+ global saved_get_config_paths
+ import powerline
+ saved_get_config_paths = powerline.get_config_paths
+ path = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), 'powerline', 'config_files')
+ powerline.get_config_paths = lambda: [path]
+ old_cwd = os.getcwd()
+
+
+def tearDownModule():
+ global old_cwd
+ global saved_get_config_paths
+ import powerline
+ powerline.get_config_paths = saved_get_config_paths
+ os.chdir(old_cwd)
+ old_cwd = None
+
+
+if __name__ == '__main__':
+ from tests.modules import main
+ main()
diff --git a/tests/test_python/test_segments.py b/tests/test_python/test_segments.py
new file mode 100644
index 0000000..c7a9870
--- /dev/null
+++ b/tests/test_python/test_segments.py
@@ -0,0 +1,1846 @@
+# vim:fileencoding=utf-8:noet
+from __future__ import (unicode_literals, division, absolute_import, print_function)
+
+import sys
+import os
+
+from functools import partial
+from collections import namedtuple
+from time import sleep
+from platform import python_implementation
+
+from powerline.segments import shell, tmux, pdb, i3wm
+from powerline.lib.vcs import get_fallback_create_watcher
+from powerline.lib.unicode import out_u
+
+import tests.modules.vim as vim_module
+
+from tests.modules.lib import (Args, urllib_read, replace_attr, new_module,
+ replace_module_module, replace_env, Pl)
+from tests.modules import TestCase, SkipTest
+
+
+def get_dummy_guess(**kwargs):
+ if 'directory' in kwargs:
+ def guess(path, create_watcher):
+ return Args(branch=lambda: out_u(os.path.basename(path)), **kwargs)
+ else:
+ def guess(path, create_watcher):
+ return Args(branch=lambda: out_u(os.path.basename(path)), directory=path, **kwargs)
+ return guess
+
+
+class TestShell(TestCase):
+ def test_last_status(self):
+ pl = Pl()
+ segment_info = {'args': Args(last_exit_code=10)}
+ self.assertEqual(shell.last_status(pl=pl, segment_info=segment_info), [
+ {'contents': '10', 'highlight_groups': ['exit_fail']}
+ ])
+ segment_info['args'].last_exit_code = 137
+ self.assertEqual(shell.last_status(pl=pl, segment_info=segment_info), [
+ {'contents': 'SIGKILL', 'highlight_groups': ['exit_fail']}
+ ])
+ self.assertEqual(shell.last_status(pl=pl, segment_info=segment_info, signal_names=False), [
+ {'contents': '137', 'highlight_groups': ['exit_fail']}
+ ])
+ segment_info['args'].last_exit_code = 0
+ self.assertEqual(shell.last_status(pl=pl, segment_info=segment_info), None)
+ segment_info['args'].last_exit_code = None
+ self.assertEqual(shell.last_status(pl=pl, segment_info=segment_info), None)
+ segment_info['args'].last_exit_code = 'sigsegv'
+ self.assertEqual(shell.last_status(pl=pl, segment_info=segment_info), [
+ {'contents': 'sigsegv', 'highlight_groups': ['exit_fail']}
+ ])
+ segment_info['args'].last_exit_code = 'sigsegv+core'
+ self.assertEqual(shell.last_status(pl=pl, segment_info=segment_info), [
+ {'contents': 'sigsegv+core', 'highlight_groups': ['exit_fail']}
+ ])
+
+ def test_last_pipe_status(self):
+ pl = Pl()
+ segment_info = {'args': Args(last_pipe_status=[], last_exit_code=0)}
+ self.assertEqual(shell.last_pipe_status(pl=pl, segment_info=segment_info), None)
+ segment_info['args'].last_pipe_status = [0, 0, 0]
+ self.assertEqual(shell.last_pipe_status(pl=pl, segment_info=segment_info), None)
+ segment_info['args'].last_pipe_status = [0, 0]
+ self.assertEqual(shell.last_pipe_status(pl=pl, segment_info=segment_info), None)
+ segment_info['args'].last_pipe_status = [0]
+ self.assertEqual(shell.last_pipe_status(pl=pl, segment_info=segment_info), None)
+ segment_info['args'].last_pipe_status = [0, 2, 0]
+ self.assertEqual(shell.last_pipe_status(pl=pl, segment_info=segment_info), [
+ {'contents': '0', 'highlight_groups': ['exit_success'], 'draw_inner_divider': True},
+ {'contents': '2', 'highlight_groups': ['exit_fail'], 'draw_inner_divider': True},
+ {'contents': '0', 'highlight_groups': ['exit_success'], 'draw_inner_divider': True},
+ ])
+ segment_info['args'].last_pipe_status = [2, 0, 0]
+ self.assertEqual(shell.last_pipe_status(pl=pl, segment_info=segment_info), [
+ {'contents': '2', 'highlight_groups': ['exit_fail'], 'draw_inner_divider': True},
+ {'contents': '0', 'highlight_groups': ['exit_success'], 'draw_inner_divider': True},
+ {'contents': '0', 'highlight_groups': ['exit_success'], 'draw_inner_divider': True},
+ ])
+ segment_info['args'].last_pipe_status = [137, 0, 0]
+ self.assertEqual(shell.last_pipe_status(pl=pl, segment_info=segment_info), [
+ {'contents': 'SIGKILL', 'highlight_groups': ['exit_fail'], 'draw_inner_divider': True},
+ {'contents': '0', 'highlight_groups': ['exit_success'], 'draw_inner_divider': True},
+ {'contents': '0', 'highlight_groups': ['exit_success'], 'draw_inner_divider': True},
+ ])
+
+ self.assertEqual(shell.last_pipe_status(pl=pl, segment_info=segment_info, signal_names=False), [
+ {'contents': '137', 'highlight_groups': ['exit_fail'], 'draw_inner_divider': True},
+ {'contents': '0', 'highlight_groups': ['exit_success'], 'draw_inner_divider': True},
+ {'contents': '0', 'highlight_groups': ['exit_success'], 'draw_inner_divider': True},
+ ])
+
+ segment_info['args'].last_pipe_status = [0, 0, 2]
+ self.assertEqual(shell.last_pipe_status(pl=pl, segment_info=segment_info), [
+ {'contents': '0', 'highlight_groups': ['exit_success'], 'draw_inner_divider': True},
+ {'contents': '0', 'highlight_groups': ['exit_success'], 'draw_inner_divider': True},
+ {'contents': '2', 'highlight_groups': ['exit_fail'], 'draw_inner_divider': True},
+ ])
+ segment_info['args'].last_pipe_status = [2]
+ self.assertEqual(shell.last_pipe_status(pl=pl, segment_info=segment_info), [
+ {'contents': '2', 'highlight_groups': ['exit_fail'], 'draw_inner_divider': True},
+ ])
+ segment_info['args'].last_pipe_status = [0, 'sigsegv', 'sigsegv+core']
+ self.assertEqual(shell.last_pipe_status(pl=pl, segment_info=segment_info), [
+ {'contents': '0', 'highlight_groups': ['exit_success'], 'draw_inner_divider': True},
+ {'contents': 'sigsegv', 'highlight_groups': ['exit_fail'], 'draw_inner_divider': True},
+ {'contents': 'sigsegv+core', 'highlight_groups': ['exit_fail'], 'draw_inner_divider': True}
+ ])
+ segment_info['args'].last_pipe_status = [0, 'sigsegv', 0]
+ self.assertEqual(shell.last_pipe_status(pl=pl, segment_info=segment_info), [
+ {'contents': '0', 'highlight_groups': ['exit_success'], 'draw_inner_divider': True},
+ {'contents': 'sigsegv', 'highlight_groups': ['exit_fail'], 'draw_inner_divider': True},
+ {'contents': '0', 'highlight_groups': ['exit_success'], 'draw_inner_divider': True}
+ ])
+ segment_info['args'].last_pipe_status = [0, 'sigsegv+core', 0]
+ self.assertEqual(shell.last_pipe_status(pl=pl, segment_info=segment_info), [
+ {'contents': '0', 'highlight_groups': ['exit_success'], 'draw_inner_divider': True},
+ {'contents': 'sigsegv+core', 'highlight_groups': ['exit_fail'], 'draw_inner_divider': True},
+ {'contents': '0', 'highlight_groups': ['exit_success'], 'draw_inner_divider': True}
+ ])
+ segment_info['args'].last_pipe_status = []
+ segment_info['args'].last_exit_code = 5
+ self.assertEqual(shell.last_pipe_status(pl=pl, segment_info=segment_info), [
+ {'contents': '5', 'highlight_groups': ['exit_fail'], 'draw_inner_divider': True},
+ ])
+
+ def test_jobnum(self):
+ pl = Pl()
+ segment_info = {'args': Args(jobnum=0)}
+ self.assertEqual(shell.jobnum(pl=pl, segment_info=segment_info), None)
+ self.assertEqual(shell.jobnum(pl=pl, segment_info=segment_info, show_zero=False), None)
+ self.assertEqual(shell.jobnum(pl=pl, segment_info=segment_info, show_zero=True), '0')
+ segment_info = {'args': Args(jobnum=1)}
+ self.assertEqual(shell.jobnum(pl=pl, segment_info=segment_info), '1')
+ self.assertEqual(shell.jobnum(pl=pl, segment_info=segment_info, show_zero=False), '1')
+ self.assertEqual(shell.jobnum(pl=pl, segment_info=segment_info, show_zero=True), '1')
+
+ def test_continuation(self):
+ pl = Pl()
+ self.assertEqual(shell.continuation(pl=pl, segment_info={}), [{
+ 'contents': '',
+ 'width': 'auto',
+ 'highlight_groups': ['continuation:current', 'continuation'],
+ }])
+ segment_info = {'parser_state': 'if cmdsubst'}
+ self.assertEqual(shell.continuation(pl=pl, segment_info=segment_info), [
+ {
+ 'contents': 'if',
+ 'draw_inner_divider': True,
+ 'highlight_groups': ['continuation:current', 'continuation'],
+ 'width': 'auto',
+ 'align': 'l',
+ },
+ ])
+ self.assertEqual(shell.continuation(pl=pl, segment_info=segment_info, right_align=True), [
+ {
+ 'contents': 'if',
+ 'draw_inner_divider': True,
+ 'highlight_groups': ['continuation:current', 'continuation'],
+ 'width': 'auto',
+ 'align': 'r',
+ },
+ ])
+ self.assertEqual(shell.continuation(pl=pl, segment_info=segment_info, omit_cmdsubst=False), [
+ {
+ 'contents': 'if',
+ 'draw_inner_divider': True,
+ 'highlight_groups': ['continuation'],
+ },
+ {
+ 'contents': 'cmdsubst',
+ 'draw_inner_divider': True,
+ 'highlight_groups': ['continuation:current', 'continuation'],
+ 'width': 'auto',
+ 'align': 'l',
+ },
+ ])
+ self.assertEqual(shell.continuation(pl=pl, segment_info=segment_info, omit_cmdsubst=False, right_align=True), [
+ {
+ 'contents': 'if',
+ 'draw_inner_divider': True,
+ 'highlight_groups': ['continuation'],
+ 'width': 'auto',
+ 'align': 'r',
+ },
+ {
+ 'contents': 'cmdsubst',
+ 'draw_inner_divider': True,
+ 'highlight_groups': ['continuation:current', 'continuation'],
+ },
+ ])
+ self.assertEqual(shell.continuation(pl=pl, segment_info=segment_info, omit_cmdsubst=True, right_align=True), [
+ {
+ 'contents': 'if',
+ 'draw_inner_divider': True,
+ 'highlight_groups': ['continuation:current', 'continuation'],
+ 'width': 'auto',
+ 'align': 'r',
+ },
+ ])
+ self.assertEqual(shell.continuation(pl=pl, segment_info=segment_info, omit_cmdsubst=True, right_align=True, renames={'if': 'IF'}), [
+ {
+ 'contents': 'IF',
+ 'draw_inner_divider': True,
+ 'highlight_groups': ['continuation:current', 'continuation'],
+ 'width': 'auto',
+ 'align': 'r',
+ },
+ ])
+ self.assertEqual(shell.continuation(pl=pl, segment_info=segment_info, omit_cmdsubst=True, right_align=True, renames={'if': None}), [
+ {
+ 'contents': '',
+ 'highlight_groups': ['continuation:current', 'continuation'],
+ 'width': 'auto',
+ 'align': 'r',
+ },
+ ])
+ segment_info = {'parser_state': 'then then then cmdsubst'}
+ self.assertEqual(shell.continuation(pl=pl, segment_info=segment_info), [
+ {
+ 'contents': 'then',
+ 'draw_inner_divider': True,
+ 'highlight_groups': ['continuation'],
+ },
+ {
+ 'contents': 'then',
+ 'draw_inner_divider': True,
+ 'highlight_groups': ['continuation'],
+ },
+ {
+ 'contents': 'then',
+ 'draw_inner_divider': True,
+ 'highlight_groups': ['continuation:current', 'continuation'],
+ 'width': 'auto',
+ 'align': 'l',
+ },
+ ])
+
+ def test_cwd(self):
+ new_os = new_module('os', path=os.path, sep='/')
+ pl = Pl()
+ cwd = [None]
+
+ def getcwd():
+ wd = cwd[0]
+ if isinstance(wd, Exception):
+ raise wd
+ else:
+ return wd
+
+ segment_info = {'getcwd': getcwd, 'home': None}
+ with replace_attr(shell, 'os', new_os):
+ cwd[0] = '/abc/def/ghi/foo/bar'
+ self.assertEqual(shell.cwd(pl=pl, segment_info=segment_info), [
+ {'contents': '/', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': True},
+ {'contents': 'abc', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': True},
+ {'contents': 'def', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': True},
+ {'contents': 'ghi', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': True},
+ {'contents': 'foo', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': True},
+ {'contents': 'bar', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': True, 'highlight_groups': ['cwd:current_folder', 'cwd']},
+ ])
+ segment_info['home'] = '/abc/def/ghi'
+ self.assertEqual(shell.cwd(pl=pl, segment_info=segment_info), [
+ {'contents': '~', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': True},
+ {'contents': 'foo', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': True},
+ {'contents': 'bar', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': True, 'highlight_groups': ['cwd:current_folder', 'cwd']},
+ ])
+ segment_info.update(shortened_path='~foo/ghi')
+ self.assertEqual(shell.cwd(pl=pl, segment_info=segment_info), [
+ {'contents': '~foo', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': True},
+ {'contents': 'ghi', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': True, 'highlight_groups': ['cwd:current_folder', 'cwd']},
+ ])
+ self.assertEqual(shell.cwd(pl=pl, segment_info=segment_info, use_shortened_path=False), [
+ {'contents': '~', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': True},
+ {'contents': 'foo', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': True},
+ {'contents': 'bar', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': True, 'highlight_groups': ['cwd:current_folder', 'cwd']},
+ ])
+ segment_info.pop('shortened_path')
+ self.assertEqual(shell.cwd(pl=pl, segment_info=segment_info, dir_limit_depth=3), [
+ {'contents': '~', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': True},
+ {'contents': 'foo', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': True},
+ {'contents': 'bar', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': True, 'highlight_groups': ['cwd:current_folder', 'cwd']}
+ ])
+ self.assertEqual(shell.cwd(pl=pl, segment_info=segment_info, dir_limit_depth=3, shorten_home=False), [
+ {'contents': '...', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': True},
+ {'contents': 'ghi', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': True},
+ {'contents': 'foo', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': True},
+ {'contents': 'bar', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': True, 'highlight_groups': ['cwd:current_folder', 'cwd']}
+ ])
+ self.assertEqual(shell.cwd(pl=pl, segment_info=segment_info, dir_limit_depth=1), [
+ {'contents': '...', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': True},
+ {'contents': 'bar', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': True, 'highlight_groups': ['cwd:current_folder', 'cwd']}
+ ])
+ self.assertEqual(shell.cwd(pl=pl, segment_info=segment_info, dir_limit_depth=1, ellipsis='---'), [
+ {'contents': '---', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': True},
+ {'contents': 'bar', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': True, 'highlight_groups': ['cwd:current_folder', 'cwd']}
+ ])
+ self.assertEqual(shell.cwd(pl=pl, segment_info=segment_info, dir_limit_depth=1, ellipsis=None), [
+ {'contents': 'bar', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': True, 'highlight_groups': ['cwd:current_folder', 'cwd']}
+ ])
+ self.assertEqual(shell.cwd(pl=pl, segment_info=segment_info, dir_limit_depth=1, use_path_separator=True), [
+ {'contents': '.../', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': False},
+ {'contents': 'bar', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': False, 'highlight_groups': ['cwd:current_folder', 'cwd']}
+ ])
+ self.assertEqual(shell.cwd(pl=pl, segment_info=segment_info, dir_limit_depth=1, use_path_separator=True, ellipsis='---'), [
+ {'contents': '---/', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': False},
+ {'contents': 'bar', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': False, 'highlight_groups': ['cwd:current_folder', 'cwd']}
+ ])
+ self.assertEqual(shell.cwd(pl=pl, segment_info=segment_info, dir_limit_depth=1, use_path_separator=True, ellipsis=None), [
+ {'contents': 'bar', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': False, 'highlight_groups': ['cwd:current_folder', 'cwd']}
+ ])
+ self.assertEqual(shell.cwd(pl=pl, segment_info=segment_info, dir_limit_depth=2, dir_shorten_len=2), [
+ {'contents': '~', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': True},
+ {'contents': 'fo', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': True},
+ {'contents': 'bar', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': True, 'highlight_groups': ['cwd:current_folder', 'cwd']}
+ ])
+ self.assertEqual(shell.cwd(pl=pl, segment_info=segment_info, dir_limit_depth=2, dir_shorten_len=2, use_path_separator=True), [
+ {'contents': '~/', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': False},
+ {'contents': 'fo/', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': False},
+ {'contents': 'bar', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': False, 'highlight_groups': ['cwd:current_folder', 'cwd']}
+ ])
+ cwd[0] = '/etc'
+ self.assertEqual(shell.cwd(pl=pl, segment_info=segment_info, use_path_separator=False), [
+ {'contents': '/', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': True},
+ {'contents': 'etc', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': True, 'highlight_groups': ['cwd:current_folder', 'cwd']},
+ ])
+ self.assertEqual(shell.cwd(pl=pl, segment_info=segment_info, use_path_separator=True), [
+ {'contents': '/', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': False},
+ {'contents': 'etc', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': False, 'highlight_groups': ['cwd:current_folder', 'cwd']},
+ ])
+ cwd[0] = '/'
+ self.assertEqual(shell.cwd(pl=pl, segment_info=segment_info, use_path_separator=False), [
+ {'contents': '/', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': True, 'highlight_groups': ['cwd:current_folder', 'cwd']},
+ ])
+ self.assertEqual(shell.cwd(pl=pl, segment_info=segment_info, use_path_separator=True), [
+ {'contents': '/', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': False, 'highlight_groups': ['cwd:current_folder', 'cwd']},
+ ])
+ ose = OSError()
+ ose.errno = 2
+ cwd[0] = ose
+ self.assertEqual(shell.cwd(pl=pl, segment_info=segment_info, dir_limit_depth=2, dir_shorten_len=2), [
+ {'contents': '[not found]', 'divider_highlight_group': 'cwd:divider', 'highlight_groups': ['cwd:current_folder', 'cwd'], 'draw_inner_divider': True}
+ ])
+ cwd[0] = OSError()
+ self.assertRaises(OSError, shell.cwd, pl=pl, segment_info=segment_info, dir_limit_depth=2, dir_shorten_len=2)
+ cwd[0] = ValueError()
+ self.assertRaises(ValueError, shell.cwd, pl=pl, segment_info=segment_info, dir_limit_depth=2, dir_shorten_len=2)
+
+
+class TestTmux(TestCase):
+ def test_attached_clients(self):
+ def get_tmux_output(pl, cmd, *args):
+ if cmd == 'list-panes':
+ return 'session_name\n'
+ elif cmd == 'list-clients':
+ return '/dev/pts/2: 0 [191x51 xterm-256color] (utf8)\n/dev/pts/3: 0 [191x51 xterm-256color] (utf8)'
+
+ pl = Pl()
+ with replace_attr(tmux, 'get_tmux_output', get_tmux_output):
+ self.assertEqual(tmux.attached_clients(pl=pl), '2')
+ self.assertEqual(tmux.attached_clients(pl=pl, minimum=3), None)
+
+
+class TestCommon(TestCase):
+ @classmethod
+ def setUpClass(cls):
+ module = __import__(str('powerline.segments.common.{0}'.format(cls.module_name)))
+ cls.module = getattr(module.segments.common, str(cls.module_name))
+
+
+class TestNet(TestCommon):
+ module_name = 'net'
+
+ def test_hostname(self):
+ pl = Pl()
+ with replace_env('SSH_CLIENT', '192.168.0.12 40921 22') as segment_info:
+ with replace_module_module(self.module, 'socket', gethostname=lambda: 'abc'):
+ self.assertEqual(self.module.hostname(pl=pl, segment_info=segment_info), 'abc')
+ self.assertEqual(self.module.hostname(pl=pl, segment_info=segment_info, only_if_ssh=True), 'abc')
+ with replace_module_module(self.module, 'socket', gethostname=lambda: 'abc.mydomain'):
+ self.assertEqual(self.module.hostname(pl=pl, segment_info=segment_info), 'abc.mydomain')
+ self.assertEqual(self.module.hostname(pl=pl, segment_info=segment_info, exclude_domain=True), 'abc')
+ self.assertEqual(self.module.hostname(pl=pl, segment_info=segment_info, only_if_ssh=True), 'abc.mydomain')
+ self.assertEqual(self.module.hostname(pl=pl, segment_info=segment_info, only_if_ssh=True, exclude_domain=True), 'abc')
+ segment_info['environ'].pop('SSH_CLIENT')
+ with replace_module_module(self.module, 'socket', gethostname=lambda: 'abc'):
+ self.assertEqual(self.module.hostname(pl=pl, segment_info=segment_info), 'abc')
+ self.assertEqual(self.module.hostname(pl=pl, segment_info=segment_info, only_if_ssh=True), None)
+ with replace_module_module(self.module, 'socket', gethostname=lambda: 'abc.mydomain'):
+ self.assertEqual(self.module.hostname(pl=pl, segment_info=segment_info), 'abc.mydomain')
+ self.assertEqual(self.module.hostname(pl=pl, segment_info=segment_info, exclude_domain=True), 'abc')
+ self.assertEqual(self.module.hostname(pl=pl, segment_info=segment_info, only_if_ssh=True, exclude_domain=True), None)
+
+ def test_external_ip(self):
+ pl = Pl()
+ with replace_attr(self.module, 'urllib_read', urllib_read):
+ self.assertEqual(self.module.external_ip(pl=pl), [{'contents': '127.0.0.1', 'divider_highlight_group': 'background:divider'}])
+
+ def test_internal_ip(self):
+ try:
+ import netifaces
+ except ImportError:
+ raise SkipTest('netifaces module is not available')
+ pl = Pl()
+ addr = {
+ 'enp2s0': {
+ netifaces.AF_INET: [{'addr': '192.168.100.200'}],
+ netifaces.AF_INET6: [{'addr': 'feff::5446:5eff:fe5a:7777%enp2s0'}]
+ },
+ 'lo': {
+ netifaces.AF_INET: [{'addr': '127.0.0.1'}],
+ netifaces.AF_INET6: [{'addr': '::1'}]
+ },
+ 'teredo': {
+ netifaces.AF_INET6: [{'addr': 'feff::5446:5eff:fe5a:7777'}]
+ },
+ }
+ interfaces = ['lo', 'enp2s0', 'teredo']
+ with replace_module_module(
+ self.module, 'netifaces',
+ interfaces=(lambda: interfaces),
+ ifaddresses=(lambda interface: addr[interface]),
+ AF_INET=netifaces.AF_INET,
+ AF_INET6=netifaces.AF_INET6,
+ ):
+ self.assertEqual(self.module.internal_ip(pl=pl), '192.168.100.200')
+ self.assertEqual(self.module.internal_ip(pl=pl, interface='auto'), '192.168.100.200')
+ self.assertEqual(self.module.internal_ip(pl=pl, interface='lo'), '127.0.0.1')
+ self.assertEqual(self.module.internal_ip(pl=pl, interface='teredo'), None)
+ self.assertEqual(self.module.internal_ip(pl=pl, ipv=4), '192.168.100.200')
+ self.assertEqual(self.module.internal_ip(pl=pl, interface='auto', ipv=4), '192.168.100.200')
+ self.assertEqual(self.module.internal_ip(pl=pl, interface='lo', ipv=4), '127.0.0.1')
+ self.assertEqual(self.module.internal_ip(pl=pl, interface='teredo', ipv=4), None)
+ self.assertEqual(self.module.internal_ip(pl=pl, ipv=6), 'feff::5446:5eff:fe5a:7777%enp2s0')
+ self.assertEqual(self.module.internal_ip(pl=pl, interface='auto', ipv=6), 'feff::5446:5eff:fe5a:7777%enp2s0')
+ self.assertEqual(self.module.internal_ip(pl=pl, interface='lo', ipv=6), '::1')
+ self.assertEqual(self.module.internal_ip(pl=pl, interface='teredo', ipv=6), 'feff::5446:5eff:fe5a:7777')
+ interfaces[1:2] = ()
+ self.assertEqual(self.module.internal_ip(pl=pl, ipv=6), 'feff::5446:5eff:fe5a:7777')
+ interfaces[1:2] = ()
+ self.assertEqual(self.module.internal_ip(pl=pl, ipv=6), '::1')
+ interfaces[:] = ()
+ self.assertEqual(self.module.internal_ip(pl=pl, ipv=6), None)
+
+ gateways = {
+ 'default': {
+ netifaces.AF_INET: ('192.168.100.1', 'enp2s0'),
+ netifaces.AF_INET6: ('feff::5446:5eff:fe5a:0001', 'enp2s0')
+ }
+ }
+
+ with replace_module_module(
+ self.module, 'netifaces',
+ interfaces=(lambda: interfaces),
+ ifaddresses=(lambda interface: addr[interface]),
+ gateways=(lambda: gateways),
+ AF_INET=netifaces.AF_INET,
+ AF_INET6=netifaces.AF_INET6,
+ ):
+ # default gateway has specified address family
+ self.assertEqual(self.module.internal_ip(pl=pl, interface='default_gateway', ipv=4), '192.168.100.200')
+ self.assertEqual(self.module.internal_ip(pl=pl, interface='default_gateway', ipv=6), 'feff::5446:5eff:fe5a:7777%enp2s0')
+ # default gateway doesn't have specified address family
+ gateways['default'] = {}
+ self.assertEqual(self.module.internal_ip(pl=pl, interface='default_gateway', ipv=4), None)
+ self.assertEqual(self.module.internal_ip(pl=pl, interface='default_gateway', ipv=6), None)
+
+# TODO: fix network load
+#def test_network_load(self):
+# def gb(interface):
+# return None
+#
+# f = [gb]
+#
+# def _get_bytes(interface):
+# return f[0](interface)
+#
+# pl = Pl()
+#
+# with replace_attr(self.module, '_get_bytes', _get_bytes):
+# self.module.network_load.startup(pl=pl)
+# try:
+# self.assertEqual(self.module.network_load(pl=pl, interface='eth0'), None)
+# sleep(self.module.network_load.interval)
+# self.assertEqual(self.module.network_load(pl=pl, interface='eth0'), None)
+# while 'prev' not in self.module.network_load.interfaces.get('eth0', {}):
+# sleep(0.1)
+# self.assertEqual(self.module.network_load(pl=pl, interface='eth0'), None)
+#
+# l = [0, 0]
+#
+# def gb2(interface):
+# l[0] += 1200
+# l[1] += 2400
+# return tuple(l)
+# f[0] = gb2
+#
+# while not self.module.network_load.interfaces.get('eth0', {}).get('prev', (None, None))[1]:
+# sleep(0.1)
+# self.assertEqual(self.module.network_load(pl=pl, interface='eth0'), [
+# {'divider_highlight_group': 'network_load:divider', 'contents': 'DL 1 KiB/s', 'highlight_groups': ['network_load_recv', 'network_load']},
+# {'divider_highlight_group': 'network_load:divider', 'contents': 'UL 2 KiB/s', 'highlight_groups': ['network_load_sent', 'network_load']},
+# ])
+# self.assertEqual(self.module.network_load(pl=pl, interface='eth0', recv_format='r {value}', sent_format='s {value}'), [
+# {'divider_highlight_group': 'network_load:divider', 'contents': 'r 1 KiB/s', 'highlight_groups': ['network_load_recv', 'network_load']},
+# {'divider_highlight_group': 'network_load:divider', 'contents': 's 2 KiB/s', 'highlight_groups': ['network_load_sent', 'network_load']},
+# ])
+# self.assertEqual(self.module.network_load(pl=pl, recv_format='r {value}', sent_format='s {value}', suffix='bps', interface='eth0'), [
+# {'divider_highlight_group': 'network_load:divider', 'contents': 'r 1 Kibps', 'highlight_groups': ['network_load_recv', 'network_load']},
+# {'divider_highlight_group': 'network_load:divider', 'contents': 's 2 Kibps', 'highlight_groups': ['network_load_sent', 'network_load']},
+# ])
+# self.assertEqual(self.module.network_load(pl=pl, recv_format='r {value}', sent_format='s {value}', si_prefix=True, interface='eth0'), [
+# {'divider_highlight_group': 'network_load:divider', 'contents': 'r 1 kB/s', 'highlight_groups': ['network_load_recv', 'network_load']},
+# {'divider_highlight_group': 'network_load:divider', 'contents': 's 2 kB/s', 'highlight_groups': ['network_load_sent', 'network_load']},
+# ])
+# self.assertEqual(self.module.network_load(pl=pl, recv_format='r {value}', sent_format='s {value}', recv_max=0, interface='eth0'), [
+# {'divider_highlight_group': 'network_load:divider', 'contents': 'r 1 KiB/s', 'highlight_groups': ['network_load_recv_gradient', 'network_load_gradient', 'network_load_recv', 'network_load'], 'gradient_level': 100},
+# {'divider_highlight_group': 'network_load:divider', 'contents': 's 2 KiB/s', 'highlight_groups': ['network_load_sent', 'network_load']},
+# ])
+#
+# class ApproxEqual(object):
+# def __eq__(self, i):
+# return abs(i - 50.0) < 1
+#
+# self.assertEqual(self.module.network_load(pl=pl, recv_format='r {value}', sent_format='s {value}', sent_max=4800, interface='eth0'), [
+# {'divider_highlight_group': 'network_load:divider', 'contents': 'r 1 KiB/s', 'highlight_groups': ['network_load_recv', 'network_load']},
+# {'divider_highlight_group': 'network_load:divider', 'contents': 's 2 KiB/s', 'highlight_groups': ['network_load_sent_gradient', 'network_load_gradient', 'network_load_sent', 'network_load'], 'gradient_level': ApproxEqual()},
+# ])
+# finally:
+# self.module.network_load.shutdown()
+
+
+class TestEnv(TestCommon):
+ module_name = 'env'
+
+ def test_user(self):
+ new_os = new_module('os', getpid=lambda: 1)
+
+ class Process(object):
+ def __init__(self, pid):
+ pass
+
+ def username(self):
+ return 'def@DOMAIN.COM'
+
+ if hasattr(self.module, 'psutil') and not callable(self.module.psutil.Process.username):
+ username = property(username)
+
+ segment_info = {'environ': {}}
+
+ def user(*args, **kwargs):
+ return self.module.user(pl=pl, segment_info=segment_info, *args, **kwargs)
+
+ struct_passwd = namedtuple('struct_passwd', ('pw_name',))
+ new_psutil = new_module('psutil', Process=Process)
+ new_pwd = new_module('pwd', getpwuid=lambda uid: struct_passwd(pw_name='def@DOMAIN.COM'))
+ new_getpass = new_module('getpass', getuser=lambda: 'def@DOMAIN.COM')
+ pl = Pl()
+ with replace_attr(self.module, 'pwd', new_pwd):
+ with replace_attr(self.module, 'getpass', new_getpass):
+ with replace_attr(self.module, 'os', new_os):
+ with replace_attr(self.module, 'psutil', new_psutil):
+ with replace_attr(self.module, '_geteuid', lambda: 5):
+ self.assertEqual(user(), [
+ {'contents': 'def@DOMAIN.COM', 'highlight_groups': ['user']}
+ ])
+ self.assertEqual(user(hide_user='abc'), [
+ {'contents': 'def@DOMAIN.COM', 'highlight_groups': ['user']}
+ ])
+ self.assertEqual(user(hide_domain=False), [
+ {'contents': 'def@DOMAIN.COM', 'highlight_groups': ['user']}
+ ])
+ self.assertEqual(user(hide_user='def@DOMAIN.COM'), None)
+ self.assertEqual(user(hide_domain=True), [
+ {'contents': 'def', 'highlight_groups': ['user']}
+ ])
+ with replace_attr(self.module, '_geteuid', lambda: 0):
+ self.assertEqual(user(), [
+ {'contents': 'def', 'highlight_groups': ['superuser', 'user']}
+ ])
+
+ def test_cwd(self):
+ new_os = new_module('os', path=os.path, sep='/')
+ pl = Pl()
+ cwd = [None]
+
+ def getcwd():
+ wd = cwd[0]
+ if isinstance(wd, Exception):
+ raise wd
+ else:
+ return wd
+
+ segment_info = {'getcwd': getcwd, 'home': None}
+ with replace_attr(self.module, 'os', new_os):
+ cwd[0] = '/abc/def/ghi/foo/bar'
+ self.assertEqual(self.module.cwd(pl=pl, segment_info=segment_info), [
+ {'contents': '/', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': True},
+ {'contents': 'abc', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': True},
+ {'contents': 'def', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': True},
+ {'contents': 'ghi', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': True},
+ {'contents': 'foo', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': True},
+ {'contents': 'bar', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': True, 'highlight_groups': ['cwd:current_folder', 'cwd']},
+ ])
+ segment_info['home'] = '/abc/def/ghi'
+ self.assertEqual(self.module.cwd(pl=pl, segment_info=segment_info), [
+ {'contents': '~', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': True},
+ {'contents': 'foo', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': True},
+ {'contents': 'bar', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': True, 'highlight_groups': ['cwd:current_folder', 'cwd']},
+ ])
+ self.assertEqual(self.module.cwd(pl=pl, segment_info=segment_info, dir_limit_depth=3), [
+ {'contents': '~', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': True},
+ {'contents': 'foo', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': True},
+ {'contents': 'bar', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': True, 'highlight_groups': ['cwd:current_folder', 'cwd']}
+ ])
+ self.assertEqual(self.module.cwd(pl=pl, segment_info=segment_info, dir_limit_depth=3, shorten_home=False), [
+ {'contents': '...', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': True},
+ {'contents': 'ghi', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': True},
+ {'contents': 'foo', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': True},
+ {'contents': 'bar', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': True, 'highlight_groups': ['cwd:current_folder', 'cwd']}
+ ])
+ self.assertEqual(self.module.cwd(pl=pl, segment_info=segment_info, dir_limit_depth=1), [
+ {'contents': '...', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': True},
+ {'contents': 'bar', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': True, 'highlight_groups': ['cwd:current_folder', 'cwd']}
+ ])
+ self.assertEqual(self.module.cwd(pl=pl, segment_info=segment_info, dir_limit_depth=1, ellipsis='---'), [
+ {'contents': '---', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': True},
+ {'contents': 'bar', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': True, 'highlight_groups': ['cwd:current_folder', 'cwd']}
+ ])
+ self.assertEqual(self.module.cwd(pl=pl, segment_info=segment_info, dir_limit_depth=1, ellipsis=None), [
+ {'contents': 'bar', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': True, 'highlight_groups': ['cwd:current_folder', 'cwd']}
+ ])
+ self.assertEqual(self.module.cwd(pl=pl, segment_info=segment_info, dir_limit_depth=1, use_path_separator=True), [
+ {'contents': '.../', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': False},
+ {'contents': 'bar', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': False, 'highlight_groups': ['cwd:current_folder', 'cwd']}
+ ])
+ self.assertEqual(self.module.cwd(pl=pl, segment_info=segment_info, dir_limit_depth=1, use_path_separator=True, ellipsis='---'), [
+ {'contents': '---/', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': False},
+ {'contents': 'bar', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': False, 'highlight_groups': ['cwd:current_folder', 'cwd']}
+ ])
+ self.assertEqual(self.module.cwd(pl=pl, segment_info=segment_info, dir_limit_depth=1, use_path_separator=True, ellipsis=None), [
+ {'contents': 'bar', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': False, 'highlight_groups': ['cwd:current_folder', 'cwd']}
+ ])
+ self.assertEqual(self.module.cwd(pl=pl, segment_info=segment_info, dir_limit_depth=2, dir_shorten_len=2), [
+ {'contents': '~', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': True},
+ {'contents': 'fo', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': True},
+ {'contents': 'bar', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': True, 'highlight_groups': ['cwd:current_folder', 'cwd']}
+ ])
+ self.assertEqual(self.module.cwd(pl=pl, segment_info=segment_info, dir_limit_depth=2, dir_shorten_len=2, use_path_separator=True), [
+ {'contents': '~/', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': False},
+ {'contents': 'fo/', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': False},
+ {'contents': 'bar', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': False, 'highlight_groups': ['cwd:current_folder', 'cwd']}
+ ])
+ cwd[0] = '/etc'
+ self.assertEqual(self.module.cwd(pl=pl, segment_info=segment_info, use_path_separator=False), [
+ {'contents': '/', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': True},
+ {'contents': 'etc', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': True, 'highlight_groups': ['cwd:current_folder', 'cwd']},
+ ])
+ self.assertEqual(self.module.cwd(pl=pl, segment_info=segment_info, use_path_separator=True), [
+ {'contents': '/', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': False},
+ {'contents': 'etc', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': False, 'highlight_groups': ['cwd:current_folder', 'cwd']},
+ ])
+ cwd[0] = '/'
+ self.assertEqual(self.module.cwd(pl=pl, segment_info=segment_info, use_path_separator=False), [
+ {'contents': '/', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': True, 'highlight_groups': ['cwd:current_folder', 'cwd']},
+ ])
+ self.assertEqual(self.module.cwd(pl=pl, segment_info=segment_info, use_path_separator=True), [
+ {'contents': '/', 'divider_highlight_group': 'cwd:divider', 'draw_inner_divider': False, 'highlight_groups': ['cwd:current_folder', 'cwd']},
+ ])
+ ose = OSError()
+ ose.errno = 2
+ cwd[0] = ose
+ self.assertEqual(self.module.cwd(pl=pl, segment_info=segment_info, dir_limit_depth=2, dir_shorten_len=2), [
+ {'contents': '[not found]', 'divider_highlight_group': 'cwd:divider', 'highlight_groups': ['cwd:current_folder', 'cwd'], 'draw_inner_divider': True}
+ ])
+ cwd[0] = OSError()
+ self.assertRaises(OSError, self.module.cwd, pl=pl, segment_info=segment_info, dir_limit_depth=2, dir_shorten_len=2)
+ cwd[0] = ValueError()
+ self.assertRaises(ValueError, self.module.cwd, pl=pl, segment_info=segment_info, dir_limit_depth=2, dir_shorten_len=2)
+
+ def test_virtualenv(self):
+ pl = Pl()
+ with replace_env('VIRTUAL_ENV', '/abc/def/ghi') as segment_info:
+ self.assertEqual(self.module.virtualenv(pl=pl, segment_info=segment_info), 'ghi')
+ self.assertEqual(self.module.virtualenv(pl=pl, segment_info=segment_info, ignore_conda=True), 'ghi')
+ self.assertEqual(self.module.virtualenv(pl=pl, segment_info=segment_info, ignore_venv=True), None)
+ self.assertEqual(self.module.virtualenv(pl=pl, segment_info=segment_info, ignore_venv=True, ignore_conda=True), None)
+ self.assertEqual(self.module.virtualenv(pl=pl, segment_info=segment_info, ignored_names=["aaa"]), "ghi")
+ self.assertEqual(self.module.virtualenv(pl=pl, segment_info=segment_info, ignored_names=["ghi"]), "def")
+ self.assertEqual(self.module.virtualenv(pl=pl, segment_info=segment_info, ignored_names=["def", "ghi"]), "abc")
+ self.assertEqual(self.module.virtualenv(pl=pl, segment_info=segment_info, ignored_names=["abc", "def", "ghi"]), None)
+
+ segment_info['environ'].pop('VIRTUAL_ENV')
+ self.assertEqual(self.module.virtualenv(pl=pl, segment_info=segment_info), None)
+ self.assertEqual(self.module.virtualenv(pl=pl, segment_info=segment_info, ignore_conda=True), None)
+ self.assertEqual(self.module.virtualenv(pl=pl, segment_info=segment_info, ignore_venv=True), None)
+ self.assertEqual(self.module.virtualenv(pl=pl, segment_info=segment_info, ignore_venv=True, ignore_conda=True), None)
+
+ with replace_env('CONDA_DEFAULT_ENV', 'foo') as segment_info:
+ self.assertEqual(self.module.virtualenv(pl=pl, segment_info=segment_info), 'foo')
+ self.assertEqual(self.module.virtualenv(pl=pl, segment_info=segment_info, ignored_names=["foo"]), None)
+ self.assertEqual(self.module.virtualenv(pl=pl, segment_info=segment_info, ignore_conda=True), None)
+ self.assertEqual(self.module.virtualenv(pl=pl, segment_info=segment_info, ignore_venv=True), 'foo')
+ self.assertEqual(self.module.virtualenv(pl=pl, segment_info=segment_info, ignore_venv=True, ignore_conda=True), None)
+
+ segment_info['environ'].pop('CONDA_DEFAULT_ENV')
+ self.assertEqual(self.module.virtualenv(pl=pl, segment_info=segment_info), None)
+ self.assertEqual(self.module.virtualenv(pl=pl, segment_info=segment_info, ignore_conda=True), None)
+ self.assertEqual(self.module.virtualenv(pl=pl, segment_info=segment_info, ignore_venv=True), None)
+ self.assertEqual(self.module.virtualenv(pl=pl, segment_info=segment_info, ignore_venv=True, ignore_conda=True), None)
+
+ with replace_env('CONDA_DEFAULT_ENV', 'foo', environ={'VIRTUAL_ENV': '/sbc/def/ghi'}) as segment_info:
+ self.assertEqual(self.module.virtualenv(pl=pl, segment_info=segment_info), 'ghi')
+ self.assertEqual(self.module.virtualenv(pl=pl, segment_info=segment_info, ignore_conda=True), 'ghi')
+ self.assertEqual(self.module.virtualenv(pl=pl, segment_info=segment_info, ignore_venv=True), 'foo')
+ self.assertEqual(self.module.virtualenv(pl=pl, segment_info=segment_info, ignore_venv=True, ignore_conda=True), None)
+
+ segment_info['environ'].pop('CONDA_DEFAULT_ENV')
+ self.assertEqual(self.module.virtualenv(pl=pl, segment_info=segment_info), 'ghi')
+ self.assertEqual(self.module.virtualenv(pl=pl, segment_info=segment_info, ignore_conda=True), 'ghi')
+ self.assertEqual(self.module.virtualenv(pl=pl, segment_info=segment_info, ignore_venv=True), None)
+ self.assertEqual(self.module.virtualenv(pl=pl, segment_info=segment_info, ignore_venv=True, ignore_conda=True), None)
+
+ with replace_env('VIRTUAL_ENV', '/abc/def/venv') as segment_info:
+ self.assertEqual(self.module.virtualenv(pl=pl, segment_info=segment_info), 'def')
+
+ def test_environment(self):
+ pl = Pl()
+ variable = 'FOO'
+ value = 'bar'
+ with replace_env(variable, value) as segment_info:
+ self.assertEqual(self.module.environment(pl=pl, segment_info=segment_info, variable=variable), value)
+ segment_info['environ'].pop(variable)
+ self.assertEqual(self.module.environment(pl=pl, segment_info=segment_info, variable=variable), None)
+
+
+class TestVcs(TestCommon):
+ module_name = 'vcs'
+
+ def test_branch(self):
+ pl = Pl()
+ create_watcher = get_fallback_create_watcher()
+ segment_info = {'getcwd': os.getcwd}
+ branch = partial(self.module.branch, pl=pl, create_watcher=create_watcher)
+ with replace_attr(self.module, 'guess', get_dummy_guess(status=lambda: None, directory='/tmp/tests')):
+ with replace_attr(self.module, 'tree_status', lambda repo, pl: None):
+ self.assertEqual(branch(segment_info=segment_info, status_colors=False), [{
+ 'highlight_groups': ['branch'],
+ 'contents': 'tests',
+ 'divider_highlight_group': None
+ }])
+ self.assertEqual(branch(segment_info=segment_info, status_colors=True), [{
+ 'contents': 'tests',
+ 'highlight_groups': ['branch_clean', 'branch'],
+ 'divider_highlight_group': None
+ }])
+ with replace_attr(self.module, 'guess', get_dummy_guess(status=lambda: 'D ', directory='/tmp/tests')):
+ with replace_attr(self.module, 'tree_status', lambda repo, pl: 'D '):
+ self.assertEqual(branch(segment_info=segment_info, status_colors=False), [{
+ 'highlight_groups': ['branch'],
+ 'contents': 'tests',
+ 'divider_highlight_group': None
+ }])
+ self.assertEqual(branch(segment_info=segment_info, status_colors=True), [{
+ 'contents': 'tests',
+ 'highlight_groups': ['branch_dirty', 'branch'],
+ 'divider_highlight_group': None
+ }])
+ self.assertEqual(branch(segment_info=segment_info, status_colors=False), [{
+ 'highlight_groups': ['branch'],
+ 'contents': 'tests',
+ 'divider_highlight_group': None
+ }])
+ with replace_attr(self.module, 'guess', lambda path, create_watcher: None):
+ self.assertEqual(branch(segment_info=segment_info, status_colors=False), None)
+ with replace_attr(self.module, 'guess', get_dummy_guess(status=lambda: 'U')):
+ with replace_attr(self.module, 'tree_status', lambda repo, pl: 'U'):
+ self.assertEqual(branch(segment_info=segment_info, status_colors=False, ignore_statuses=['U']), [{
+ 'highlight_groups': ['branch'],
+ 'contents': 'tests',
+ 'divider_highlight_group': None
+ }])
+ self.assertEqual(branch(segment_info=segment_info, status_colors=True, ignore_statuses=['DU']), [{
+ 'highlight_groups': ['branch_dirty', 'branch'],
+ 'contents': 'tests',
+ 'divider_highlight_group': None
+ }])
+ self.assertEqual(branch(segment_info=segment_info, status_colors=True), [{
+ 'highlight_groups': ['branch_dirty', 'branch'],
+ 'contents': 'tests',
+ 'divider_highlight_group': None
+ }])
+ self.assertEqual(branch(segment_info=segment_info, status_colors=True, ignore_statuses=['U']), [{
+ 'highlight_groups': ['branch_clean', 'branch'],
+ 'contents': 'tests',
+ 'divider_highlight_group': None
+ }])
+
+ def test_stash(self):
+ pl = Pl()
+ create_watcher = get_fallback_create_watcher()
+ stash = partial(self.module.stash, pl=pl, create_watcher=create_watcher, segment_info={'getcwd': os.getcwd})
+
+ def forge_stash(n):
+ return replace_attr(self.module, 'guess', get_dummy_guess(stash=lambda: n, directory='/tmp/tests'))
+
+ with forge_stash(0):
+ self.assertEqual(stash(), None)
+ with forge_stash(1):
+ self.assertEqual(stash(), [{
+ 'highlight_groups': ['stash'],
+ 'contents': '1',
+ 'divider_highlight_group': None
+ }])
+ with forge_stash(2):
+ self.assertEqual(stash(), [{
+ 'highlight_groups': ['stash'],
+ 'contents': '2',
+ 'divider_highlight_group': None
+ }])
+
+
+class TestTime(TestCommon):
+ module_name = 'time'
+
+ def test_date(self):
+ pl = Pl()
+ with replace_attr(self.module, 'datetime', Args(strptime=lambda timezone, fmt: Args(tzinfo=timezone), now=lambda tz:Args(strftime=lambda fmt: fmt + (tz if tz else '')))):
+ self.assertEqual(self.module.date(pl=pl), [{'contents': '%Y-%m-%d', 'highlight_groups': ['date'], 'divider_highlight_group': None}])
+ self.assertEqual(self.module.date(pl=pl, timezone='+0900'), [{'contents': '%Y-%m-%d+0900', 'highlight_groups': ['date'], 'divider_highlight_group': None}])
+ self.assertEqual(self.module.date(pl=pl, format='%H:%M', istime=True), [{'contents': '%H:%M', 'highlight_groups': ['time', 'date'], 'divider_highlight_group': 'time:divider'}])
+ self.assertEqual(self.module.date(pl=pl, format='%H:%M', istime=True, timezone='-0900'), [{'contents': '%H:%M-0900', 'highlight_groups': ['time', 'date'], 'divider_highlight_group': 'time:divider'}])
+ unicode_date = self.module.date(pl=pl, format='\u231a', istime=True)
+ expected_unicode_date = [{'contents': '\u231a', 'highlight_groups': ['time', 'date'], 'divider_highlight_group': 'time:divider'}]
+ if python_implementation() == 'PyPy' and sys.version_info >= (3,):
+ if unicode_date != expected_unicode_date:
+ raise SkipTest('Dates do not match, see https://bitbucket.org/pypy/pypy/issues/2161/pypy3-strftime-does-not-accept-unicode')
+ self.assertEqual(unicode_date, expected_unicode_date)
+
+ def test_fuzzy_time(self):
+ time = Args(hour=0, minute=45)
+ pl = Pl()
+ hour_str = ['12', 'one', 'two', 'three', 'four', 'five', 'six', 'seven', 'eight', 'nine', 'ten', 'eleven']
+ minute_str = {'0': 'o\'clock', '5': 'five past', '10': 'ten past','15':
+ 'quarter past','20': 'twenty past', '25': 'twenty-five past',
+ '30': 'half past', '35': 'twenty-five to','40': 'twenty to', '45':
+ 'quarter to', '50': 'ten to', '55': 'five to'}
+ special_case_str = {
+ '(23, 58)': '~ midnight',
+ '(23, 59)': '~ midnight',
+ '(0, 0)': 'midnight',
+ '(0, 1)': '~ midnight',
+ '(0, 2)': '~ midnight',
+ '(12, 0)': 'twelve o\'clock'}
+ with replace_attr(self.module, 'datetime', Args(strptime=lambda timezone, fmt: Args(tzinfo=timezone), now=lambda tz: time)):
+ self.assertEqual(self.module.fuzzy_time(pl=pl, hour_str=hour_str, minute_str=minute_str, special_case_str=special_case_str), 'quarter to one')
+ self.assertEqual(self.module.fuzzy_time(pl=pl), 'quarter to one')
+ time.hour = 23
+ time.minute = 59
+ self.assertEqual(self.module.fuzzy_time(pl=pl, hour_str=hour_str, minute_str=minute_str, special_case_str=special_case_str), '~ midnight')
+ self.assertEqual(self.module.fuzzy_time(pl=pl), 'round about midnight')
+ time.hour = 11
+ time.minute = 33
+ self.assertEqual(self.module.fuzzy_time(pl=pl, hour_str=hour_str, minute_str=minute_str, special_case_str=special_case_str),'twenty-five to 12')
+ self.assertEqual(self.module.fuzzy_time(pl=pl), 'twenty-five to twelve')
+ time.hour = 12
+ time.minute = 0
+ self.assertEqual(self.module.fuzzy_time(pl=pl, hour_str=hour_str, minute_str=minute_str, special_case_str=special_case_str), 'twelve o\'clock')
+ self.assertEqual(self.module.fuzzy_time(pl=pl), 'noon')
+ time.hour = 11
+ time.minute = 33
+ self.assertEqual(self.module.fuzzy_time(pl=pl, unicode_text=False, hour_str=hour_str, minute_str=minute_str, special_case_str=special_case_str), 'twenty-five to 12')
+ self.assertEqual(self.module.fuzzy_time(pl=pl, unicode_text=False), 'twenty-five to twelve')
+ time.hour = 12
+ time.minute = 0
+ self.assertEqual(self.module.fuzzy_time(pl=pl, unicode_text=False, hour_str=hour_str, minute_str=minute_str, special_case_str=special_case_str), 'twelve o\'clock')
+ self.assertEqual(self.module.fuzzy_time(pl=pl, unicode_text=False), 'noon')
+ time.hour = 11
+ time.minute = 33
+ self.assertEqual(self.module.fuzzy_time(pl=pl, unicode_text=True, hour_str=hour_str, minute_str=minute_str, special_case_str=special_case_str), 'twenty‐five to 12')
+ self.assertEqual(self.module.fuzzy_time(pl=pl, unicode_text=True), 'twenty‐five to twelve')
+ time.hour = 12
+ time.minute = 0
+ self.assertEqual(self.module.fuzzy_time(pl=pl, unicode_text=True, hour_str=hour_str, minute_str=minute_str, special_case_str=special_case_str), 'twelve o’clock')
+ self.assertEqual(self.module.fuzzy_time(pl=pl, unicode_text=True),'noon')
+
+
+class TestSys(TestCommon):
+ module_name = 'sys'
+
+ def test_uptime(self):
+ pl = Pl()
+ with replace_attr(self.module, '_get_uptime', lambda: 259200):
+ self.assertEqual(self.module.uptime(pl=pl), [{'contents': '3d 0h 00m', 'divider_highlight_group': 'background:divider'}])
+ with replace_attr(self.module, '_get_uptime', lambda: 93784):
+ self.assertEqual(self.module.uptime(pl=pl), [{'contents': '1d 2h 03m', 'divider_highlight_group': 'background:divider'}])
+ self.assertEqual(self.module.uptime(pl=pl, shorten_len=4), [{'contents': '1d 2h 03m 04s', 'divider_highlight_group': 'background:divider'}])
+ with replace_attr(self.module, '_get_uptime', lambda: 65536):
+ self.assertEqual(self.module.uptime(pl=pl), [{'contents': '18h 12m 16s', 'divider_highlight_group': 'background:divider'}])
+ self.assertEqual(self.module.uptime(pl=pl, shorten_len=2), [{'contents': '18h 12m', 'divider_highlight_group': 'background:divider'}])
+ self.assertEqual(self.module.uptime(pl=pl, shorten_len=1), [{'contents': '18h', 'divider_highlight_group': 'background:divider'}])
+
+ def _get_uptime():
+ raise NotImplementedError
+
+ with replace_attr(self.module, '_get_uptime', _get_uptime):
+ self.assertEqual(self.module.uptime(pl=pl), None)
+
+ def test_system_load(self):
+ pl = Pl()
+ with replace_module_module(self.module, 'os', getloadavg=lambda: (7.5, 3.5, 1.5)):
+ with replace_attr(self.module, '_cpu_count', lambda: 2):
+ self.assertEqual(self.module.system_load(pl=pl), [
+ {'contents': '7.5 ', 'highlight_groups': ['system_load_gradient', 'system_load'], 'divider_highlight_group': 'background:divider', 'gradient_level': 100},
+ {'contents': '3.5 ', 'highlight_groups': ['system_load_gradient', 'system_load'], 'divider_highlight_group': 'background:divider', 'gradient_level': 75.0},
+ {'contents': '1.5', 'highlight_groups': ['system_load_gradient', 'system_load'], 'divider_highlight_group': 'background:divider', 'gradient_level': 0}
+ ])
+ self.assertEqual(self.module.system_load(pl=pl, format='{avg:.0f}', threshold_good=0, threshold_bad=1), [
+ {'contents': '8 ', 'highlight_groups': ['system_load_gradient', 'system_load'], 'divider_highlight_group': 'background:divider', 'gradient_level': 100},
+ {'contents': '4 ', 'highlight_groups': ['system_load_gradient', 'system_load'], 'divider_highlight_group': 'background:divider', 'gradient_level': 100},
+ {'contents': '2', 'highlight_groups': ['system_load_gradient', 'system_load'], 'divider_highlight_group': 'background:divider', 'gradient_level': 75.0}
+ ])
+ self.assertEqual(self.module.system_load(pl=pl, short=True), [
+ {'contents': '7.5', 'highlight_groups': ['system_load_gradient', 'system_load'], 'divider_highlight_group': 'background:divider', 'gradient_level': 100},
+ ])
+ self.assertEqual(self.module.system_load(pl=pl, format='{avg:.0f}', threshold_good=0, threshold_bad=1, short=True), [
+ {'contents': '8', 'highlight_groups': ['system_load_gradient', 'system_load'], 'divider_highlight_group': 'background:divider', 'gradient_level': 100},
+ ])
+
+ def test_cpu_load_percent(self):
+ try:
+ __import__('psutil')
+ except ImportError as e:
+ raise SkipTest('Failed to import psutil: {0}'.format(e))
+ pl = Pl()
+ with replace_module_module(self.module, 'psutil', cpu_percent=lambda **kwargs: 52.3):
+ self.assertEqual(self.module.cpu_load_percent(pl=pl), [{
+ 'contents': '52%',
+ 'gradient_level': 52.3,
+ 'highlight_groups': ['cpu_load_percent_gradient', 'cpu_load_percent'],
+ }])
+ self.assertEqual(self.module.cpu_load_percent(pl=pl, format='{0:.1f}%'), [{
+ 'contents': '52.3%',
+ 'gradient_level': 52.3,
+ 'highlight_groups': ['cpu_load_percent_gradient', 'cpu_load_percent'],
+ }])
+
+
+class TestWthr(TestCommon):
+ module_name = 'wthr'
+
+ def test_weather(self):
+ pl = Pl()
+ with replace_attr(self.module, 'urllib_read', urllib_read):
+ self.assertEqual(self.module.weather(pl=pl), [
+ {'divider_highlight_group': 'background:divider', 'highlight_groups': ['weather_condition_sunny', 'weather_conditions', 'weather'], 'contents': 'SUN '},
+ {'divider_highlight_group': 'background:divider', 'highlight_groups': ['weather_temp_gradient', 'weather_temp', 'weather'], 'contents': '20°C', 'gradient_level': 71.42857142857143}
+ ])
+ self.assertEqual(self.module.weather(pl=pl, temp_coldest=0, temp_hottest=100), [
+ {'divider_highlight_group': 'background:divider', 'highlight_groups': ['weather_condition_sunny', 'weather_conditions', 'weather'], 'contents': 'SUN '},
+ {'divider_highlight_group': 'background:divider', 'highlight_groups': ['weather_temp_gradient', 'weather_temp', 'weather'], 'contents': '20°C', 'gradient_level': 20}
+ ])
+ self.assertEqual(self.module.weather(pl=pl, temp_coldest=-100, temp_hottest=-50), [
+ {'divider_highlight_group': 'background:divider', 'highlight_groups': ['weather_condition_sunny', 'weather_conditions', 'weather'], 'contents': 'SUN '},
+ {'divider_highlight_group': 'background:divider', 'highlight_groups': ['weather_temp_gradient', 'weather_temp', 'weather'], 'contents': '20°C', 'gradient_level': 100}
+ ])
+ self.assertEqual(self.module.weather(pl=pl, icons={'sunny': 'o'}), [
+ {'divider_highlight_group': 'background:divider', 'highlight_groups': ['weather_condition_sunny', 'weather_conditions', 'weather'], 'contents': 'o '},
+ {'divider_highlight_group': 'background:divider', 'highlight_groups': ['weather_temp_gradient', 'weather_temp', 'weather'], 'contents': '20°C', 'gradient_level': 71.42857142857143}
+ ])
+ # Test is disabled as no request has more than 1 weather condition associated currently
+ # self.assertEqual(self.module.weather(pl=pl, icons={'windy': 'x'}), [
+ # {'divider_highlight_group': 'background:divider', 'highlight_groups': ['weather_condition_blustery', 'weather_condition_windy', 'weather_conditions', 'weather'], 'contents': 'x '},
+ # {'divider_highlight_group': 'background:divider', 'highlight_groups': ['weather_temp_gradient', 'weather_temp', 'weather'], 'contents': '14°C', 'gradient_level': 62.857142857142854}
+ # ])
+ self.assertEqual(self.module.weather(pl=pl, unit='F'), [
+ {'divider_highlight_group': 'background:divider', 'highlight_groups': ['weather_condition_sunny', 'weather_conditions', 'weather'], 'contents': 'SUN '},
+ {'divider_highlight_group': 'background:divider', 'highlight_groups': ['weather_temp_gradient', 'weather_temp', 'weather'], 'contents': '68°F', 'gradient_level': 100}
+ ])
+ self.assertEqual(self.module.weather(pl=pl, unit='K'), [
+ {'divider_highlight_group': 'background:divider', 'highlight_groups': ['weather_condition_sunny', 'weather_conditions', 'weather'], 'contents': 'SUN '},
+ {'divider_highlight_group': 'background:divider', 'highlight_groups': ['weather_temp_gradient', 'weather_temp', 'weather'], 'contents': '293K', 'gradient_level': 100}
+ ])
+ self.assertEqual(self.module.weather(pl=pl, temp_format='{temp:.1e}C'), [
+ {'divider_highlight_group': 'background:divider', 'highlight_groups': ['weather_condition_sunny', 'weather_conditions', 'weather'], 'contents': 'SUN '},
+ {'divider_highlight_group': 'background:divider', 'highlight_groups': ['weather_temp_gradient', 'weather_temp', 'weather'], 'contents': '2.0e+01C', 'gradient_level': 71.42857142857143}
+ ])
+ with replace_attr(self.module, 'urllib_read', urllib_read):
+ self.module.weather.startup(pl=pl, location_query='Meppen,06,DE')
+ self.assertEqual(self.module.weather(pl=pl), [
+ {'divider_highlight_group': 'background:divider', 'highlight_groups': ['weather_condition_sunny', 'weather_conditions', 'weather'], 'contents': 'SUN '},
+ {'divider_highlight_group': 'background:divider', 'highlight_groups': ['weather_temp_gradient', 'weather_temp', 'weather'], 'contents': '20°C', 'gradient_level': 71.42857142857143}
+ ])
+ self.assertEqual(self.module.weather(pl=pl, location_query='Moscow,RU'), [
+ {'divider_highlight_group': 'background:divider', 'highlight_groups': ['weather_condition_sunny', 'weather_conditions', 'weather'], 'contents': 'SUN '},
+ {'divider_highlight_group': 'background:divider', 'highlight_groups': ['weather_temp_gradient', 'weather_temp', 'weather'], 'contents': '10°C', 'gradient_level': 57.142857142857146}
+ ])
+ self.module.weather.shutdown()
+
+
+class TestI3WM(TestCase):
+ def test_workspaces(self):
+ class Conn(object):
+ def get_tree(self):
+ return self
+
+ def descendents(self):
+ nodes_unfocused = [Args(focused = False)]
+ nodes_focused = [Args(focused = True)]
+
+ workspace_scratch = lambda: Args(name='__i3_scratch')
+ workspace_noscratch = lambda: Args(name='2: w2')
+ return [
+ Args(scratchpad_state='fresh', urgent=False, workspace=workspace_scratch, nodes=nodes_unfocused),
+ Args(scratchpad_state='changed', urgent=True, workspace=workspace_noscratch, nodes=nodes_focused),
+ Args(scratchpad_state='fresh', urgent=False, workspace=workspace_scratch, nodes=nodes_unfocused),
+ Args(scratchpad_state=None, urgent=False, workspace=workspace_noscratch, nodes=nodes_unfocused),
+ Args(scratchpad_state='fresh', urgent=False, workspace=workspace_scratch, nodes=nodes_focused),
+ Args(scratchpad_state=None, urgent=True, workspace=workspace_noscratch, nodes=nodes_unfocused),
+ ]
+
+ def workspaces(self):
+ return iter([
+ Args(name='1: w1', output='LVDS1', focused=False, urgent=False, visible=False, num=1, leaves=lambda: []),
+ Args(name='2: w2', output='LVDS1', focused=False, urgent=False, visible=True, num=2, leaves=lambda: []),
+ Args(name='3: w3', output='HDMI1', focused=False, urgent=True, visible=True, num=3, leaves=lambda: []),
+ Args(name='4: w4', output='DVI01', focused=True, urgent=True, visible=True, num=None, leaves=lambda: [])
+ ])
+
+ def get_workspaces(self):
+ return iter([
+ Args(name='1: w1', output='LVDS1', focused=False, urgent=False, visible=False, num=1, leaves=lambda: []),
+ Args(name='2: w2', output='LVDS1', focused=False, urgent=False, visible=True, num=2, leaves=lambda: []),
+ Args(name='3: w3', output='HDMI1', focused=False, urgent=True, visible=True, num=3, leaves=lambda: []),
+ Args(name='4: w4', output='DVI01', focused=True, urgent=True, visible=True, num=None, leaves=lambda: [])
+ ])
+
+ def get_outputs(self):
+ return iter([
+ Args(name='LVDS1', active=True),
+ Args(name='HDMI1', active=True),
+ Args(name='DVI01', active=True),
+ Args(name='HDMI2', active=False),
+ ])
+
+ pl = Pl()
+ with replace_attr(i3wm, 'get_i3_connection', lambda: Conn()):
+ segment_info = {}
+
+ self.assertEqual(i3wm.workspaces(pl=pl, segment_info=segment_info), [
+ {'contents': '1: w1', 'highlight_groups': ['workspace']},
+ {'contents': '2: w2', 'highlight_groups': ['workspace:visible', 'w_visible', 'workspace']},
+ {'contents': '3: w3', 'highlight_groups': ['workspace:urgent', 'w_urgent', 'workspace:visible', 'w_visible', 'workspace']},
+ {'contents': '4: w4', 'highlight_groups': ['workspace:focused', 'w_focused', 'workspace:urgent', 'w_urgent', 'workspace:visible', 'w_visible', 'workspace']},
+ ])
+ self.assertEqual(i3wm.workspaces(pl=pl, segment_info=segment_info, only_show=None), [
+ {'contents': '1: w1', 'highlight_groups': ['workspace']},
+ {'contents': '2: w2', 'highlight_groups': ['workspace:visible', 'w_visible', 'workspace']},
+ {'contents': '3: w3', 'highlight_groups': ['workspace:urgent', 'w_urgent', 'workspace:visible', 'w_visible', 'workspace']},
+ {'contents': '4: w4', 'highlight_groups': ['workspace:focused', 'w_focused', 'workspace:urgent', 'w_urgent', 'workspace:visible', 'w_visible', 'workspace']},
+ ])
+ self.assertEqual(i3wm.workspaces(pl=pl, segment_info=segment_info, only_show=['focused', 'urgent']), [
+ {'contents': '3: w3', 'highlight_groups': ['workspace:urgent', 'w_urgent', 'workspace:visible', 'w_visible', 'workspace']},
+ {'contents': '4: w4', 'highlight_groups': ['workspace:focused', 'w_focused', 'workspace:urgent', 'w_urgent', 'workspace:visible', 'w_visible', 'workspace']},
+ ])
+ self.assertEqual(i3wm.workspaces(pl=pl, segment_info=segment_info, only_show=['visible']), [
+ {'contents': '2: w2', 'highlight_groups': ['workspace:visible', 'w_visible', 'workspace']},
+ {'contents': '3: w3', 'highlight_groups': ['workspace:urgent', 'w_urgent', 'workspace:visible', 'w_visible', 'workspace']},
+ {'contents': '4: w4', 'highlight_groups': ['workspace:focused', 'w_focused', 'workspace:urgent', 'w_urgent', 'workspace:visible', 'w_visible', 'workspace']},
+ ])
+ self.assertEqual(i3wm.workspaces(pl=pl, segment_info=segment_info, only_show=['visible'], strip=3), [
+ {'contents': 'w2', 'highlight_groups': ['workspace:visible', 'w_visible', 'workspace']},
+ {'contents': 'w3', 'highlight_groups': ['workspace:urgent', 'w_urgent', 'workspace:visible', 'w_visible', 'workspace']},
+ {'contents': 'w4', 'highlight_groups': ['workspace:focused', 'w_focused', 'workspace:urgent', 'w_urgent', 'workspace:visible', 'w_visible', 'workspace']},
+ ])
+ self.assertEqual(i3wm.workspaces(pl=pl, segment_info=segment_info, only_show=['focused', 'urgent'], output='DVI01'), [
+ {'contents': '4: w4', 'highlight_groups': ['workspace:focused', 'w_focused', 'workspace:urgent', 'w_urgent', 'workspace:visible', 'w_visible', 'workspace']},
+ ])
+ self.assertEqual(i3wm.workspaces(pl=pl, segment_info=segment_info, only_show=['visible'], output='HDMI1'), [
+ {'contents': '3: w3', 'highlight_groups': ['workspace:urgent', 'w_urgent', 'workspace:visible', 'w_visible', 'workspace']},
+ ])
+ self.assertEqual(i3wm.workspaces(pl=pl, segment_info=segment_info, only_show=['visible'], strip=3, output='LVDS1'), [
+ {'contents': 'w2', 'highlight_groups': ['workspace:visible', 'w_visible', 'workspace']},
+ ])
+ segment_info['output'] = 'LVDS1'
+ self.assertEqual(i3wm.workspaces(pl=pl, segment_info=segment_info, only_show=['visible'], output='HDMI1'), [
+ {'contents': '3: w3', 'highlight_groups': ['workspace:urgent', 'w_urgent', 'workspace:visible', 'w_visible', 'workspace']},
+ ])
+ self.assertEqual(i3wm.workspaces(pl=pl, segment_info=segment_info, only_show=['visible'], strip=3), [
+ {'contents': 'w2', 'highlight_groups': ['workspace:visible', 'w_visible', 'workspace']},
+ ])
+
+ def test_workspace(self):
+ class Conn(object):
+ def get_tree(self):
+ return self
+
+ def descendents(self):
+ nodes_unfocused = [Args(focused = False)]
+ nodes_focused = [Args(focused = True)]
+
+ workspace_scratch = lambda: Args(name='__i3_scratch')
+ workspace_noscratch = lambda: Args(name='2: w2')
+ return [
+ Args(scratchpad_state='fresh', urgent=False, workspace=workspace_scratch, nodes=nodes_unfocused),
+ Args(scratchpad_state='changed', urgent=True, workspace=workspace_noscratch, nodes=nodes_focused),
+ Args(scratchpad_state='fresh', urgent=False, workspace=workspace_scratch, nodes=nodes_unfocused),
+ Args(scratchpad_state=None, urgent=False, workspace=workspace_noscratch, nodes=nodes_unfocused),
+ Args(scratchpad_state='fresh', urgent=False, workspace=workspace_scratch, nodes=nodes_focused),
+ Args(scratchpad_state=None, urgent=True, workspace=workspace_noscratch, nodes=nodes_unfocused),
+ ]
+
+ def workspaces(self):
+ return iter([
+ Args(name='1: w1', output='LVDS1', focused=False, urgent=False, visible=False, num=1, leaves=lambda: []),
+ Args(name='2: w2', output='LVDS1', focused=False, urgent=False, visible=True, num=2, leaves=lambda: []),
+ Args(name='3: w3', output='HDMI1', focused=False, urgent=True, visible=True, num=3, leaves=lambda: []),
+ Args(name='4: w4', output='DVI01', focused=True, urgent=True, visible=True, num=None, leaves=lambda: [])
+ ])
+
+ def get_workspaces(self):
+ return iter([
+ Args(name='1: w1', output='LVDS1', focused=False, urgent=False, visible=False, num=1, leaves=lambda: []),
+ Args(name='2: w2', output='LVDS1', focused=False, urgent=False, visible=True, num=2, leaves=lambda: []),
+ Args(name='3: w3', output='HDMI1', focused=False, urgent=True, visible=True, num=3, leaves=lambda: []),
+ Args(name='4: w4', output='DVI01', focused=True, urgent=True, visible=True, num=None, leaves=lambda: [])
+ ])
+
+ def get_outputs(self):
+ return iter([
+ Args(name='LVDS1', active=True),
+ Args(name='HDMI1', active=True),
+ Args(name='DVI01', active=True),
+ Args(name='HDMI2', active=False),
+ ])
+
+ pl = Pl()
+ with replace_attr(i3wm, 'get_i3_connection', lambda: Conn()):
+ segment_info = {}
+
+ self.assertEqual(i3wm.workspace(pl=pl, segment_info=segment_info, workspace='1: w1'), [
+ {'contents': '1: w1', 'highlight_groups': ['workspace']},
+ ])
+ self.assertEqual(i3wm.workspace(pl=pl, segment_info=segment_info, workspace='3: w3', strip=True), [
+ {'contents': 'w3', 'highlight_groups': ['workspace:urgent', 'w_urgent', 'workspace:visible', 'w_visible', 'workspace']},
+ ])
+ self.assertEqual(i3wm.workspace(pl=pl, segment_info=segment_info, workspace='9: w9'), None)
+ self.assertEqual(i3wm.workspace(pl=pl, segment_info=segment_info), [
+ {'contents': '4: w4', 'highlight_groups': ['workspace:focused', 'w_focused', 'workspace:urgent', 'w_urgent', 'workspace:visible', 'w_visible', 'workspace']},
+ ])
+ segment_info['workspace'] = next(Conn().get_workspaces())
+ self.assertEqual(i3wm.workspace(pl=pl, segment_info=segment_info, workspace='4: w4'), [
+ {'contents': '4: w4', 'highlight_groups': ['workspace:focused', 'w_focused', 'workspace:urgent', 'w_urgent', 'workspace:visible', 'w_visible', 'workspace']},
+ ])
+ self.assertEqual(i3wm.workspace(pl=pl, segment_info=segment_info, strip=True), [
+ {'contents': 'w1', 'highlight_groups': ['workspace']},
+ ])
+
+ def test_mode(self):
+ pl = Pl()
+ self.assertEqual(i3wm.mode(pl=pl, segment_info={'mode': 'default'}), None)
+ self.assertEqual(i3wm.mode(pl=pl, segment_info={'mode': 'test'}), 'test')
+ self.assertEqual(i3wm.mode(pl=pl, segment_info={'mode': 'default'}, names={'default': 'test'}), 'test')
+ self.assertEqual(i3wm.mode(pl=pl, segment_info={'mode': 'test'}, names={'default': 'test', 'test': 't'}), 't')
+
+ def test_scratchpad(self):
+ class Conn(object):
+ def get_tree(self):
+ return self
+
+ def descendants(self):
+ nodes_unfocused = [Args(focused = False)]
+ nodes_focused = [Args(focused = True)]
+
+ workspace_scratch = lambda: Args(name='__i3_scratch')
+ workspace_noscratch = lambda: Args(name='2: www')
+ return [
+ Args(scratchpad_state='fresh', urgent=False, workspace=workspace_scratch, nodes=nodes_unfocused),
+ Args(scratchpad_state='changed', urgent=True, workspace=workspace_noscratch, nodes=nodes_focused),
+ Args(scratchpad_state='fresh', urgent=False, workspace=workspace_scratch, nodes=nodes_unfocused),
+ Args(scratchpad_state=None, urgent=False, workspace=workspace_noscratch, nodes=nodes_unfocused),
+ Args(scratchpad_state='fresh', urgent=False, workspace=workspace_scratch, nodes=nodes_focused),
+ Args(scratchpad_state=None, urgent=True, workspace=workspace_noscratch, nodes=nodes_unfocused),
+ ]
+
+ pl = Pl()
+ with replace_attr(i3wm, 'get_i3_connection', lambda: Conn()):
+ self.assertEqual(i3wm.scratchpad(pl=pl), [
+ {'contents': 'O', 'highlight_groups': ['scratchpad']},
+ {'contents': 'X', 'highlight_groups': ['scratchpad:urgent', 'scratchpad:focused', 'scratchpad:visible', 'scratchpad']},
+ {'contents': 'O', 'highlight_groups': ['scratchpad']},
+ {'contents': 'X', 'highlight_groups': ['scratchpad:visible', 'scratchpad']},
+ {'contents': 'O', 'highlight_groups': ['scratchpad:focused', 'scratchpad']},
+ {'contents': 'X', 'highlight_groups': ['scratchpad:urgent', 'scratchpad:visible', 'scratchpad']},
+ ])
+ self.assertEqual(i3wm.scratchpad(pl=pl, icons={'changed': '-', 'fresh': 'o'}), [
+ {'contents': 'o', 'highlight_groups': ['scratchpad']},
+ {'contents': '-', 'highlight_groups': ['scratchpad:urgent', 'scratchpad:focused', 'scratchpad:visible', 'scratchpad']},
+ {'contents': 'o', 'highlight_groups': ['scratchpad']},
+ {'contents': '-', 'highlight_groups': ['scratchpad:visible', 'scratchpad']},
+ {'contents': 'o', 'highlight_groups': ['scratchpad:focused', 'scratchpad']},
+ {'contents': '-', 'highlight_groups': ['scratchpad:urgent', 'scratchpad:visible', 'scratchpad']},
+ ])
+
+
+class TestMail(TestCommon):
+ module_name = 'mail'
+
+ def test_email_imap_alert(self):
+ # TODO
+ pass
+
+
+class TestPlayers(TestCommon):
+ module_name = 'players'
+
+ def test_now_playing(self):
+ # TODO
+ pass
+
+
+class TestBat(TestCommon):
+ module_name = 'bat'
+
+ def test_battery(self):
+ pl = Pl()
+
+ def _get_battery_status(pl):
+ return 86, False
+
+ with replace_attr(self.module, '_get_battery_status', _get_battery_status):
+ self.assertEqual(self.module.battery(pl=pl), [{
+ 'contents': ' 86%',
+ 'highlight_groups': ['battery_gradient', 'battery'],
+ 'gradient_level': 14,
+ }])
+ self.assertEqual(self.module.battery(pl=pl, format='{capacity:.2f}'), [{
+ 'contents': '0.86',
+ 'highlight_groups': ['battery_gradient', 'battery'],
+ 'gradient_level': 14,
+ }])
+ self.assertEqual(self.module.battery(pl=pl, steps=7), [{
+ 'contents': ' 86%',
+ 'highlight_groups': ['battery_gradient', 'battery'],
+ 'gradient_level': 14,
+ }])
+ self.assertEqual(self.module.battery(pl=pl, gamify=True), [
+ {
+ 'contents': ' ',
+ 'draw_inner_divider': False,
+ 'highlight_groups': ['battery_offline', 'battery_ac_state', 'battery_gradient', 'battery'],
+ 'gradient_level': 0
+ },
+ {
+ 'contents': 'OOOO',
+ 'draw_inner_divider': False,
+ 'highlight_groups': ['battery_full', 'battery_gradient', 'battery'],
+ 'gradient_level': 0
+ },
+ {
+ 'contents': 'O',
+ 'draw_inner_divider': False,
+ 'highlight_groups': ['battery_empty', 'battery_gradient', 'battery'],
+ 'gradient_level': 100
+ }
+ ])
+ self.assertEqual(self.module.battery(pl=pl, gamify=True, full_heart='+', empty_heart='-', steps='10'), [
+ {
+ 'contents': ' ',
+ 'draw_inner_divider': False,
+ 'highlight_groups': ['battery_offline', 'battery_ac_state', 'battery_gradient', 'battery'],
+ 'gradient_level': 0
+ },
+ {
+ 'contents': '++++++++',
+ 'draw_inner_divider': False,
+ 'highlight_groups': ['battery_full', 'battery_gradient', 'battery'],
+ 'gradient_level': 0
+ },
+ {
+ 'contents': '--',
+ 'draw_inner_divider': False,
+ 'highlight_groups': ['battery_empty', 'battery_gradient', 'battery'],
+ 'gradient_level': 100
+ }
+ ])
+
+ def test_battery_with_ac_online(self):
+ pl = Pl()
+
+ def _get_battery_status(pl):
+ return 86, True
+
+ with replace_attr(self.module, '_get_battery_status', _get_battery_status):
+ self.assertEqual(self.module.battery(pl=pl, online='C', offline=' '), [
+ {
+ 'contents': 'C 86%',
+ 'highlight_groups': ['battery_gradient', 'battery'],
+ 'gradient_level': 14,
+ }])
+
+ def test_battery_with_ac_offline(self):
+ pl = Pl()
+
+ def _get_battery_status(pl):
+ return 86, False
+
+ with replace_attr(self.module, '_get_battery_status', _get_battery_status):
+ self.assertEqual(self.module.battery(pl=pl, online='C', offline=' '), [
+ {
+ 'contents': ' 86%',
+ 'highlight_groups': ['battery_gradient', 'battery'],
+ 'gradient_level': 14,
+ }])
+
+
+class TestVim(TestCase):
+ def test_mode(self):
+ pl = Pl()
+ segment_info = vim_module._get_segment_info()
+ self.assertEqual(self.vim.mode(pl=pl, segment_info=segment_info), 'NORMAL')
+ self.assertEqual(self.vim.mode(pl=pl, segment_info=segment_info, override={'i': 'INS'}), 'NORMAL')
+ self.assertEqual(self.vim.mode(pl=pl, segment_info=segment_info, override={'n': 'NORM'}), 'NORM')
+ with vim_module._with('mode', 'i') as segment_info:
+ self.assertEqual(self.vim.mode(pl=pl, segment_info=segment_info), 'INSERT')
+ with vim_module._with('mode', 'i\0') as segment_info:
+ self.assertEqual(self.vim.mode(pl=pl, segment_info=segment_info), 'INSERT')
+ with vim_module._with('mode', chr(ord('V') - 0x40)) as segment_info:
+ self.assertEqual(self.vim.mode(pl=pl, segment_info=segment_info), 'V-BLCK')
+ self.assertEqual(self.vim.mode(pl=pl, segment_info=segment_info, override={'^V': 'VBLK'}), 'VBLK')
+
+ def test_visual_range(self):
+ pl = Pl()
+ vr = partial(self.vim.visual_range, pl=pl)
+ vim_module.current.window.cursor = [0, 0]
+ try:
+ with vim_module._with('mode', 'i') as segment_info:
+ self.assertEqual(vr(segment_info=segment_info), '')
+ with vim_module._with('mode', '^V') as segment_info:
+ self.assertEqual(vr(segment_info=segment_info), '1 x 1')
+ with vim_module._with('vpos', line=5, col=5, off=0):
+ self.assertEqual(vr(segment_info=segment_info), '5 x 5')
+ with vim_module._with('vpos', line=5, col=4, off=0):
+ self.assertEqual(vr(segment_info=segment_info), '5 x 4')
+ with vim_module._with('mode', '^S') as segment_info:
+ self.assertEqual(vr(segment_info=segment_info), '1 x 1')
+ with vim_module._with('vpos', line=5, col=5, off=0):
+ self.assertEqual(vr(segment_info=segment_info), '5 x 5')
+ with vim_module._with('vpos', line=5, col=4, off=0):
+ self.assertEqual(vr(segment_info=segment_info), '5 x 4')
+ with vim_module._with('mode', 'V') as segment_info:
+ self.assertEqual(vr(segment_info=segment_info), 'L:1')
+ with vim_module._with('vpos', line=5, col=5, off=0):
+ self.assertEqual(vr(segment_info=segment_info), 'L:5')
+ with vim_module._with('vpos', line=5, col=4, off=0):
+ self.assertEqual(vr(segment_info=segment_info), 'L:5')
+ with vim_module._with('mode', 'S') as segment_info:
+ self.assertEqual(vr(segment_info=segment_info), 'L:1')
+ with vim_module._with('vpos', line=5, col=5, off=0):
+ self.assertEqual(vr(segment_info=segment_info), 'L:5')
+ with vim_module._with('vpos', line=5, col=4, off=0):
+ self.assertEqual(vr(segment_info=segment_info), 'L:5')
+ with vim_module._with('mode', 'v') as segment_info:
+ self.assertEqual(vr(segment_info=segment_info), 'C:1')
+ with vim_module._with('vpos', line=5, col=5, off=0):
+ self.assertEqual(vr(segment_info=segment_info), 'L:5')
+ with vim_module._with('vpos', line=5, col=4, off=0):
+ self.assertEqual(vr(segment_info=segment_info), 'L:5')
+ with vim_module._with('mode', 's') as segment_info:
+ self.assertEqual(vr(segment_info=segment_info), 'C:1')
+ with vim_module._with('vpos', line=5, col=5, off=0):
+ self.assertEqual(vr(segment_info=segment_info), 'L:5')
+ with vim_module._with('vpos', line=5, col=4, off=0):
+ self.assertEqual(vr(segment_info=segment_info), 'L:5')
+ finally:
+ vim_module._close(1)
+
+ def test_modified_indicator(self):
+ pl = Pl()
+ segment_info = vim_module._get_segment_info()
+ self.assertEqual(self.vim.modified_indicator(pl=pl, segment_info=segment_info), None)
+ segment_info['buffer'][0] = 'abc'
+ try:
+ self.assertEqual(self.vim.modified_indicator(pl=pl, segment_info=segment_info), '+')
+ self.assertEqual(self.vim.modified_indicator(pl=pl, segment_info=segment_info, text='-'), '-')
+ finally:
+ vim_module._bw(segment_info['bufnr'])
+
+ def test_paste_indicator(self):
+ pl = Pl()
+ segment_info = vim_module._get_segment_info()
+ self.assertEqual(self.vim.paste_indicator(pl=pl, segment_info=segment_info), None)
+ with vim_module._with('options', paste=1):
+ self.assertEqual(self.vim.paste_indicator(pl=pl, segment_info=segment_info), 'PASTE')
+ self.assertEqual(self.vim.paste_indicator(pl=pl, segment_info=segment_info, text='P'), 'P')
+
+ def test_readonly_indicator(self):
+ pl = Pl()
+ segment_info = vim_module._get_segment_info()
+ self.assertEqual(self.vim.readonly_indicator(pl=pl, segment_info=segment_info), None)
+ with vim_module._with('bufoptions', readonly=1):
+ self.assertEqual(self.vim.readonly_indicator(pl=pl, segment_info=segment_info), 'RO')
+ self.assertEqual(self.vim.readonly_indicator(pl=pl, segment_info=segment_info, text='L'), 'L')
+
+ def test_file_scheme(self):
+ pl = Pl()
+ segment_info = vim_module._get_segment_info()
+ self.assertEqual(self.vim.file_scheme(pl=pl, segment_info=segment_info), None)
+ with vim_module._with('buffer', '/tmp/’’/abc') as segment_info:
+ self.assertEqual(self.vim.file_scheme(pl=pl, segment_info=segment_info), None)
+ with vim_module._with('buffer', 'zipfile:/tmp/abc.zip::abc/abc.vim') as segment_info:
+ self.assertEqual(self.vim.file_scheme(pl=pl, segment_info=segment_info), 'zipfile')
+
+ def test_file_directory(self):
+ pl = Pl()
+ segment_info = vim_module._get_segment_info()
+ self.assertEqual(self.vim.file_directory(pl=pl, segment_info=segment_info), None)
+ with replace_env('HOME', '/home/foo', os.environ):
+ with vim_module._with('buffer', '/tmp/’’/abc') as segment_info:
+ self.assertEqual(self.vim.file_directory(pl=pl, segment_info=segment_info), '/tmp/’’/')
+ with vim_module._with('buffer', b'/tmp/\xFF\xFF/abc') as segment_info:
+ self.assertEqual(self.vim.file_directory(pl=pl, segment_info=segment_info), '/tmp/<ff><ff>/')
+ with vim_module._with('buffer', '/tmp/abc') as segment_info:
+ self.assertEqual(self.vim.file_directory(pl=pl, segment_info=segment_info), '/tmp/')
+ os.environ['HOME'] = '/tmp'
+ self.assertEqual(self.vim.file_directory(pl=pl, segment_info=segment_info), '~/')
+ with vim_module._with('buffer', 'zipfile:/tmp/abc.zip::abc/abc.vim') as segment_info:
+ self.assertEqual(self.vim.file_directory(pl=pl, segment_info=segment_info, remove_scheme=False), 'zipfile:/tmp/abc.zip::abc/')
+ self.assertEqual(self.vim.file_directory(pl=pl, segment_info=segment_info, remove_scheme=True), '/tmp/abc.zip::abc/')
+ self.assertEqual(self.vim.file_directory(pl=pl, segment_info=segment_info), '/tmp/abc.zip::abc/')
+ os.environ['HOME'] = '/tmp'
+ self.assertEqual(self.vim.file_directory(pl=pl, segment_info=segment_info, remove_scheme=False), 'zipfile:/tmp/abc.zip::abc/')
+ self.assertEqual(self.vim.file_directory(pl=pl, segment_info=segment_info, remove_scheme=True), '/tmp/abc.zip::abc/')
+ self.assertEqual(self.vim.file_directory(pl=pl, segment_info=segment_info), '/tmp/abc.zip::abc/')
+
+ def test_file_name(self):
+ pl = Pl()
+ segment_info = vim_module._get_segment_info()
+ self.assertEqual(self.vim.file_name(pl=pl, segment_info=segment_info), None)
+ self.assertEqual(self.vim.file_name(pl=pl, segment_info=segment_info, display_no_file=True), [
+ {'contents': '[No file]', 'highlight_groups': ['file_name_no_file', 'file_name']}
+ ])
+ self.assertEqual(self.vim.file_name(pl=pl, segment_info=segment_info, display_no_file=True, no_file_text='X'), [
+ {'contents': 'X', 'highlight_groups': ['file_name_no_file', 'file_name']}
+ ])
+ with vim_module._with('buffer', '/tmp/abc') as segment_info:
+ self.assertEqual(self.vim.file_name(pl=pl, segment_info=segment_info), 'abc')
+ with vim_module._with('buffer', '/tmp/’’') as segment_info:
+ self.assertEqual(self.vim.file_name(pl=pl, segment_info=segment_info), '’’')
+ with vim_module._with('buffer', b'/tmp/\xFF\xFF') as segment_info:
+ self.assertEqual(self.vim.file_name(pl=pl, segment_info=segment_info), '<ff><ff>')
+
+ def test_file_size(self):
+ pl = Pl()
+ segment_info = vim_module._get_segment_info()
+ self.assertEqual(self.vim.file_size(pl=pl, segment_info=segment_info), '0 B')
+ with vim_module._with(
+ 'buffer',
+ os.path.join(
+ os.path.dirname(os.path.dirname(__file__)), 'empty')
+ ) as segment_info:
+ self.assertEqual(self.vim.file_size(pl=pl, segment_info=segment_info), '0 B')
+
+ def test_file_opts(self):
+ pl = Pl()
+ segment_info = vim_module._get_segment_info()
+ self.assertEqual(self.vim.file_format(pl=pl, segment_info=segment_info), [
+ {'divider_highlight_group': 'background:divider', 'contents': 'unix'}
+ ])
+ self.assertEqual(self.vim.file_encoding(pl=pl, segment_info=segment_info), [
+ {'divider_highlight_group': 'background:divider', 'contents': 'utf-8'}
+ ])
+ self.assertEqual(self.vim.file_type(pl=pl, segment_info=segment_info), None)
+ with vim_module._with('bufoptions', filetype='python'):
+ self.assertEqual(self.vim.file_type(pl=pl, segment_info=segment_info), [
+ {'divider_highlight_group': 'background:divider', 'contents': 'python'}
+ ])
+
+ def test_window_title(self):
+ pl = Pl()
+ segment_info = vim_module._get_segment_info()
+ self.assertEqual(self.vim.window_title(pl=pl, segment_info=segment_info), None)
+ with vim_module._with('wvars', quickfix_title='Abc'):
+ self.assertEqual(self.vim.window_title(pl=pl, segment_info=segment_info), 'Abc')
+
+ def test_line_percent(self):
+ pl = Pl()
+ segment_info = vim_module._get_segment_info()
+ segment_info['buffer'][0:-1] = [str(i) for i in range(100)]
+ try:
+ self.assertEqual(self.vim.line_percent(pl=pl, segment_info=segment_info), '1')
+ vim_module._set_cursor(50, 0)
+ self.assertEqual(self.vim.line_percent(pl=pl, segment_info=segment_info), '50')
+ self.assertEqual(self.vim.line_percent(pl=pl, segment_info=segment_info, gradient=True), [
+ {'contents': '50', 'highlight_groups': ['line_percent_gradient', 'line_percent'], 'gradient_level': 50 * 100.0 / 101}
+ ])
+ finally:
+ vim_module._bw(segment_info['bufnr'])
+
+ def test_line_count(self):
+ pl = Pl()
+ segment_info = vim_module._get_segment_info()
+ segment_info['buffer'][0:-1] = [str(i) for i in range(99)]
+ try:
+ self.assertEqual(self.vim.line_count(pl=pl, segment_info=segment_info), '100')
+ vim_module._set_cursor(50, 0)
+ self.assertEqual(self.vim.line_count(pl=pl, segment_info=segment_info), '100')
+ finally:
+ vim_module._bw(segment_info['bufnr'])
+
+ def test_position(self):
+ pl = Pl()
+ segment_info = vim_module._get_segment_info()
+ try:
+ segment_info['buffer'][0:-1] = [str(i) for i in range(99)]
+ vim_module._set_cursor(49, 0)
+ self.assertEqual(self.vim.position(pl=pl, segment_info=segment_info), '50%')
+ self.assertEqual(self.vim.position(pl=pl, segment_info=segment_info, gradient=True), [
+ {'contents': '50%', 'highlight_groups': ['position_gradient', 'position'], 'gradient_level': 50.0}
+ ])
+ vim_module._set_cursor(0, 0)
+ self.assertEqual(self.vim.position(pl=pl, segment_info=segment_info), 'Top')
+ vim_module._set_cursor(97, 0)
+ self.assertEqual(self.vim.position(pl=pl, segment_info=segment_info, position_strings={'top': 'Comienzo', 'bottom': 'Final', 'all': 'Todo'}), 'Final')
+ segment_info['buffer'][0:-1] = [str(i) for i in range(2)]
+ vim_module._set_cursor(0, 0)
+ self.assertEqual(self.vim.position(pl=pl, segment_info=segment_info, position_strings={'top': 'Comienzo', 'bottom': 'Final', 'all': 'Todo'}), 'Todo')
+ self.assertEqual(self.vim.position(pl=pl, segment_info=segment_info, gradient=True), [
+ {'contents': 'All', 'highlight_groups': ['position_gradient', 'position'], 'gradient_level': 0.0}
+ ])
+ finally:
+ vim_module._bw(segment_info['bufnr'])
+
+ def test_cursor_current(self):
+ pl = Pl()
+ segment_info = vim_module._get_segment_info()
+ self.assertEqual(self.vim.line_current(pl=pl, segment_info=segment_info), '1')
+ self.assertEqual(self.vim.col_current(pl=pl, segment_info=segment_info), '1')
+ self.assertEqual(self.vim.virtcol_current(pl=pl, segment_info=segment_info), [{
+ 'highlight_groups': ['virtcol_current_gradient', 'virtcol_current', 'col_current'], 'contents': '1', 'gradient_level': 100.0 / 80,
+ }])
+ self.assertEqual(self.vim.virtcol_current(pl=pl, segment_info=segment_info, gradient=False), [{
+ 'highlight_groups': ['virtcol_current', 'col_current'], 'contents': '1',
+ }])
+
+ def test_modified_buffers(self):
+ pl = Pl()
+ self.assertEqual(self.vim.modified_buffers(pl=pl), None)
+
+ def test_branch(self):
+ pl = Pl()
+ create_watcher = get_fallback_create_watcher()
+ branch = partial(self.vim.branch, pl=pl, create_watcher=create_watcher)
+ with vim_module._with('buffer', '/foo') as segment_info:
+ with replace_attr(self.vcs, 'guess', get_dummy_guess(status=lambda: None)):
+ with replace_attr(self.vcs, 'tree_status', lambda repo, pl: None):
+ self.assertEqual(branch(segment_info=segment_info, status_colors=False), [
+ {'divider_highlight_group': 'branch:divider', 'highlight_groups': ['branch'], 'contents': 'foo'}
+ ])
+ self.assertEqual(branch(segment_info=segment_info, status_colors=True), [
+ {'divider_highlight_group': 'branch:divider', 'highlight_groups': ['branch_clean', 'branch'], 'contents': 'foo'}
+ ])
+ with replace_attr(self.vcs, 'guess', get_dummy_guess(status=lambda: 'DU')):
+ with replace_attr(self.vcs, 'tree_status', lambda repo, pl: 'DU'):
+ self.assertEqual(branch(segment_info=segment_info, status_colors=False), [
+ {'divider_highlight_group': 'branch:divider', 'highlight_groups': ['branch'], 'contents': 'foo'}
+ ])
+ self.assertEqual(branch(segment_info=segment_info, status_colors=True), [
+ {'divider_highlight_group': 'branch:divider', 'highlight_groups': ['branch_dirty', 'branch'], 'contents': 'foo'}
+ ])
+ with replace_attr(self.vcs, 'guess', get_dummy_guess(status=lambda: 'U')):
+ with replace_attr(self.vcs, 'tree_status', lambda repo, pl: 'U'):
+ self.assertEqual(branch(segment_info=segment_info, status_colors=False, ignore_statuses=['U']), [
+ {'divider_highlight_group': 'branch:divider', 'highlight_groups': ['branch'], 'contents': 'foo'}
+ ])
+ self.assertEqual(branch(segment_info=segment_info, status_colors=True, ignore_statuses=['DU']), [
+ {'divider_highlight_group': 'branch:divider', 'highlight_groups': ['branch_dirty', 'branch'], 'contents': 'foo'}
+ ])
+ self.assertEqual(branch(segment_info=segment_info, status_colors=True), [
+ {'divider_highlight_group': 'branch:divider', 'highlight_groups': ['branch_dirty', 'branch'], 'contents': 'foo'}
+ ])
+ self.assertEqual(branch(segment_info=segment_info, status_colors=True, ignore_statuses=['U']), [
+ {'divider_highlight_group': 'branch:divider', 'highlight_groups': ['branch_clean', 'branch'], 'contents': 'foo'}
+ ])
+
+ def test_stash(self):
+ pl = Pl()
+ create_watcher = get_fallback_create_watcher()
+ with vim_module._with('buffer', '/foo') as segment_info:
+ stash = partial(self.vim.stash, pl=pl, create_watcher=create_watcher, segment_info=segment_info)
+
+ def forge_stash(n):
+ return replace_attr(self.vcs, 'guess', get_dummy_guess(stash=lambda: n))
+
+ with forge_stash(0):
+ self.assertEqual(stash(), None)
+ with forge_stash(1):
+ self.assertEqual(stash(), [{
+ 'divider_highlight_group': 'stash:divider',
+ 'highlight_groups': ['stash'],
+ 'contents': '1'
+ }])
+ with forge_stash(2):
+ self.assertEqual(stash(), [{
+ 'divider_highlight_group': 'stash:divider',
+ 'highlight_groups': ['stash'],
+ 'contents': '2'
+ }])
+
+ def test_file_vcs_status(self):
+ pl = Pl()
+ create_watcher = get_fallback_create_watcher()
+ file_vcs_status = partial(self.vim.file_vcs_status, pl=pl, create_watcher=create_watcher)
+ with vim_module._with('buffer', '/foo') as segment_info:
+ with replace_attr(self.vim, 'guess', get_dummy_guess(status=lambda file: 'M')):
+ self.assertEqual(file_vcs_status(segment_info=segment_info), [
+ {'highlight_groups': ['file_vcs_status_M', 'file_vcs_status'], 'contents': 'M'}
+ ])
+ with replace_attr(self.vim, 'guess', get_dummy_guess(status=lambda file: None)):
+ self.assertEqual(file_vcs_status(segment_info=segment_info), None)
+ with vim_module._with('buffer', '/bar') as segment_info:
+ with vim_module._with('bufoptions', buftype='nofile'):
+ with replace_attr(self.vim, 'guess', get_dummy_guess(status=lambda file: 'M')):
+ self.assertEqual(file_vcs_status(segment_info=segment_info), None)
+
+ def test_trailing_whitespace(self):
+ pl = Pl()
+ with vim_module._with('buffer', 'tws') as segment_info:
+ trailing_whitespace = partial(self.vim.trailing_whitespace, pl=pl, segment_info=segment_info)
+ self.assertEqual(trailing_whitespace(), None)
+ self.assertEqual(trailing_whitespace(), None)
+ vim_module.current.buffer[0] = ' '
+ self.assertEqual(trailing_whitespace(), [{
+ 'highlight_groups': ['trailing_whitespace', 'warning'],
+ 'contents': '1',
+ }])
+ self.assertEqual(trailing_whitespace(), [{
+ 'highlight_groups': ['trailing_whitespace', 'warning'],
+ 'contents': '1',
+ }])
+ vim_module.current.buffer[0] = ''
+ self.assertEqual(trailing_whitespace(), None)
+ self.assertEqual(trailing_whitespace(), None)
+
+ def test_tabnr(self):
+ pl = Pl()
+ segment_info = vim_module._get_segment_info()
+ self.assertEqual(self.vim.tabnr(pl=pl, segment_info=segment_info, show_current=True), '1')
+ self.assertEqual(self.vim.tabnr(pl=pl, segment_info=segment_info, show_current=False), None)
+
+ def test_tab(self):
+ pl = Pl()
+ segment_info = vim_module._get_segment_info()
+ self.assertEqual(self.vim.tab(pl=pl, segment_info=segment_info), [{
+ 'contents': None,
+ 'literal_contents': (0, '%1T'),
+ }])
+ self.assertEqual(self.vim.tab(pl=pl, segment_info=segment_info, end=True), [{
+ 'contents': None,
+ 'literal_contents': (0, '%T'),
+ }])
+
+ def test_bufnr(self):
+ pl = Pl()
+ segment_info = vim_module._get_segment_info()
+ self.assertEqual(self.vim.bufnr(pl=pl, segment_info=segment_info, show_current=True), str(segment_info['bufnr']))
+ self.assertEqual(self.vim.bufnr(pl=pl, segment_info=segment_info, show_current=False), None)
+
+ def test_winnr(self):
+ pl = Pl()
+ segment_info = vim_module._get_segment_info()
+ self.assertEqual(self.vim.winnr(pl=pl, segment_info=segment_info, show_current=True), str(segment_info['winnr']))
+ self.assertEqual(self.vim.winnr(pl=pl, segment_info=segment_info, show_current=False), None)
+
+ def test_segment_info(self):
+ pl = Pl()
+ with vim_module._with('tabpage'):
+ with vim_module._with('buffer', '1') as segment_info:
+ self.assertEqual(self.vim.tab_modified_indicator(pl=pl, segment_info=segment_info), None)
+ vim_module.current.buffer[0] = ' '
+ self.assertEqual(self.vim.tab_modified_indicator(pl=pl, segment_info=segment_info), [{
+ 'contents': '+',
+ 'highlight_groups': ['tab_modified_indicator', 'modified_indicator'],
+ }])
+ vim_module._undo()
+ self.assertEqual(self.vim.tab_modified_indicator(pl=pl, segment_info=segment_info), None)
+ old_buffer = vim_module.current.buffer
+ vim_module._new('2')
+ segment_info = vim_module._get_segment_info()
+ self.assertEqual(self.vim.tab_modified_indicator(pl=pl, segment_info=segment_info), None)
+ old_buffer[0] = ' '
+ self.assertEqual(self.vim.modified_indicator(pl=pl, segment_info=segment_info), None)
+ self.assertEqual(self.vim.tab_modified_indicator(pl=pl, segment_info=segment_info), [{
+ 'contents': '+',
+ 'highlight_groups': ['tab_modified_indicator', 'modified_indicator'],
+ }])
+
+ def test_csv_col_current(self):
+ pl = Pl()
+ segment_info = vim_module._get_segment_info()
+
+ def csv_col_current(**kwargs):
+ self.vim.csv_cache and self.vim.csv_cache.clear()
+ return self.vim.csv_col_current(pl=pl, segment_info=segment_info, **kwargs)
+
+ buffer = segment_info['buffer']
+ try:
+ self.assertEqual(csv_col_current(), None)
+ buffer.options['filetype'] = 'csv'
+ self.assertEqual(csv_col_current(), None)
+ buffer[:] = ['1;2;3', '4;5;6']
+ vim_module._set_cursor(1, 1)
+ self.assertEqual(csv_col_current(), [{
+ 'contents': '1', 'highlight_groups': ['csv:column_number', 'csv']
+ }])
+ vim_module._set_cursor(2, 3)
+ self.assertEqual(csv_col_current(), [{
+ 'contents': '2', 'highlight_groups': ['csv:column_number', 'csv']
+ }])
+ vim_module._set_cursor(2, 3)
+ self.assertEqual(csv_col_current(display_name=True), [{
+ 'contents': '2', 'highlight_groups': ['csv:column_number', 'csv']
+ }, {
+ 'contents': ' (2)', 'highlight_groups': ['csv:column_name', 'csv']
+ }])
+ buffer[:0] = ['Foo;Bar;Baz']
+ vim_module._set_cursor(2, 3)
+ self.assertEqual(csv_col_current(), [{
+ 'contents': '2', 'highlight_groups': ['csv:column_number', 'csv']
+ }, {
+ 'contents': ' (Bar)', 'highlight_groups': ['csv:column_name', 'csv']
+ }])
+ if sys.version_info < (2, 7):
+ raise SkipTest('csv module in Python-2.6 does not handle multiline csv files well')
+ buffer[len(buffer):] = ['1;"bc', 'def', 'ghi', 'jkl";3']
+ vim_module._set_cursor(5, 1)
+ self.assertEqual(csv_col_current(), [{
+ 'contents': '2', 'highlight_groups': ['csv:column_number', 'csv']
+ }, {
+ 'contents': ' (Bar)', 'highlight_groups': ['csv:column_name', 'csv']
+ }])
+ vim_module._set_cursor(7, 6)
+ self.assertEqual(csv_col_current(), [{
+ 'contents': '3', 'highlight_groups': ['csv:column_number', 'csv']
+ }, {
+ 'contents': ' (Baz)', 'highlight_groups': ['csv:column_name', 'csv']
+ }])
+ self.assertEqual(csv_col_current(name_format=' ({column_name:.1})'), [{
+ 'contents': '3', 'highlight_groups': ['csv:column_number', 'csv']
+ }, {
+ 'contents': ' (B)', 'highlight_groups': ['csv:column_name', 'csv']
+ }])
+ self.assertEqual(csv_col_current(display_name=True, name_format=' ({column_name:.1})'), [{
+ 'contents': '3', 'highlight_groups': ['csv:column_number', 'csv']
+ }, {
+ 'contents': ' (B)', 'highlight_groups': ['csv:column_name', 'csv']
+ }])
+ self.assertEqual(csv_col_current(display_name=False, name_format=' ({column_name:.1})'), [{
+ 'contents': '3', 'highlight_groups': ['csv:column_number', 'csv']
+ }])
+ self.assertEqual(csv_col_current(display_name=False), [{
+ 'contents': '3', 'highlight_groups': ['csv:column_number', 'csv']
+ }])
+ finally:
+ vim_module._bw(segment_info['bufnr'])
+
+ @classmethod
+ def setUpClass(cls):
+ sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(os.path.dirname(__file__)), 'vim_sys_path')))
+ from powerline.segments import vim
+ cls.vim = vim
+ from powerline.segments.common import vcs
+ cls.vcs = vcs
+
+ @classmethod
+ def tearDownClass(cls):
+ sys.path.pop(0)
+
+
+class TestPDB(TestCase):
+ def test_current_line(self):
+ pl = Pl()
+ self.assertEqual(pdb.current_line(pl=pl, segment_info={'curframe': Args(f_lineno=10)}), '10')
+
+ def test_current_file(self):
+ pl = Pl()
+ cf = lambda **kwargs: pdb.current_file(
+ pl=pl,
+ segment_info={'curframe': Args(f_code=Args(co_filename='/tmp/abc.py'))},
+ **kwargs
+ )
+ self.assertEqual(cf(), 'abc.py')
+ self.assertEqual(cf(basename=True), 'abc.py')
+ self.assertEqual(cf(basename=False), '/tmp/abc.py')
+
+ def test_current_code_name(self):
+ pl = Pl()
+ ccn = lambda **kwargs: pdb.current_code_name(
+ pl=pl,
+ segment_info={'curframe': Args(f_code=Args(co_name='<module>'))},
+ **kwargs
+ )
+ self.assertEqual(ccn(), '<module>')
+
+ def test_current_context(self):
+ pl = Pl()
+ cc = lambda **kwargs: pdb.current_context(
+ pl=pl,
+ segment_info={'curframe': Args(f_code=Args(co_name='<module>', co_filename='/tmp/abc.py'))},
+ **kwargs
+ )
+ self.assertEqual(cc(), 'abc.py')
+
+ def test_stack_depth(self):
+ pl = Pl()
+ sd = lambda **kwargs: pdb.stack_depth(
+ pl=pl,
+ segment_info={'pdb': Args(stack=[1, 2, 3]), 'initial_stack_length': 1},
+ **kwargs
+ )
+ self.assertEqual(sd(), '2')
+ self.assertEqual(sd(full_stack=False), '2')
+ self.assertEqual(sd(full_stack=True), '3')
+
+
+old_cwd = None
+
+
+def setUpModule():
+ global old_cwd
+ global __file__
+ old_cwd = os.getcwd()
+ __file__ = os.path.abspath(__file__)
+ os.chdir(os.path.dirname(os.path.dirname(__file__)))
+
+
+def tearDownModule():
+ global old_cwd
+ os.chdir(old_cwd)
+
+
+if __name__ == '__main__':
+ from tests.modules import main
+ main()
diff --git a/tests/test_python/test_selectors.py b/tests/test_python/test_selectors.py
new file mode 100644
index 0000000..74ace8d
--- /dev/null
+++ b/tests/test_python/test_selectors.py
@@ -0,0 +1,36 @@
+# vim:fileencoding=utf-8:noet
+from __future__ import (unicode_literals, division, absolute_import, print_function)
+
+import os
+import sys
+
+from functools import partial
+
+import tests.modules.vim as vim_module
+
+from tests.modules.lib import Pl
+from tests.modules import TestCase
+
+
+class TestVim(TestCase):
+ def test_single_tab(self):
+ pl = Pl()
+ single_tab = partial(self.vim.single_tab, pl=pl, segment_info=None, mode=None)
+ with vim_module._with('tabpage'):
+ self.assertEqual(single_tab(), False)
+ self.assertEqual(single_tab(), True)
+
+ @classmethod
+ def setUpClass(cls):
+ sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(os.path.dirname(__file__)), 'vim_sys_path')))
+ from powerline.selectors import vim
+ cls.vim = vim
+
+ @classmethod
+ def tearDownClass(cls):
+ sys.path.pop(0)
+
+
+if __name__ == '__main__':
+ from tests.modules import main
+ main()
diff --git a/tests/test_python/test_watcher.py b/tests/test_python/test_watcher.py
new file mode 100644
index 0000000..a246d0b
--- /dev/null
+++ b/tests/test_python/test_watcher.py
@@ -0,0 +1,245 @@
+# vim:fileencoding=utf-8:noet
+from __future__ import (unicode_literals, division, absolute_import, print_function)
+
+import shutil
+import os
+
+from time import sleep
+from functools import partial
+from errno import ENOENT
+
+from powerline.lib.watcher import create_file_watcher, create_tree_watcher, INotifyError
+from powerline.lib.watcher.uv import UvNotFound
+from powerline import get_fallback_logger
+from powerline.lib.monotonic import monotonic
+
+from tests.modules import TestCase, SkipTest
+
+
+INOTIFY_DIR = 'inotify' + os.path.basename(os.environ.get('PYTHON', ''))
+
+
+def clear_dir(dir):
+ for root, dirs, files in list(os.walk(dir, topdown=False)):
+ for f in files:
+ os.remove(os.path.join(root, f))
+ for d in dirs:
+ os.rmdir(os.path.join(root, d))
+
+
+def set_watcher_tests(l):
+ byte_tests = (('bytes', True), ('unicode', False))
+
+ for btn, use_bytes in byte_tests:
+ def test_inotify_file_watcher(self, use_bytes=use_bytes):
+ try:
+ w = create_file_watcher(pl=get_fallback_logger(), watcher_type='inotify')
+ except INotifyError:
+ raise SkipTest('This test is not suitable for a stat based file watcher')
+ self.do_test_file_watcher(w, use_bytes)
+
+ def test_uv_file_watcher(self, use_bytes=use_bytes):
+ raise SkipTest('Uv watcher tests are not stable')
+ try:
+ w = create_file_watcher(pl=get_fallback_logger(), watcher_type='uv')
+ except UvNotFound:
+ raise SkipTest('Pyuv is not available')
+ self.do_test_file_watcher(w, use_bytes)
+
+ def test_inotify_tree_watcher(self, use_bytes=use_bytes):
+ try:
+ tw = create_tree_watcher(get_fallback_logger(), watcher_type='inotify')
+ except INotifyError:
+ raise SkipTest('INotify is not available')
+ self.do_test_tree_watcher(tw, use_bytes)
+
+ def test_uv_tree_watcher(self, use_bytes=use_bytes):
+ raise SkipTest('Uv watcher tests are not stable')
+ try:
+ tw = create_tree_watcher(get_fallback_logger(), 'uv')
+ except UvNotFound:
+ raise SkipTest('Pyuv is not available')
+ self.do_test_tree_watcher(tw, use_bytes)
+
+ def test_inotify_file_watcher_is_watching(self, use_bytes=use_bytes):
+ try:
+ w = create_file_watcher(pl=get_fallback_logger(), watcher_type='inotify')
+ except INotifyError:
+ raise SkipTest('INotify is not available')
+ self.do_test_file_watcher_is_watching(w, use_bytes)
+
+ def test_stat_file_watcher_is_watching(self, use_bytes=use_bytes):
+ w = create_file_watcher(pl=get_fallback_logger(), watcher_type='stat')
+ self.do_test_file_watcher_is_watching(w, use_bytes)
+
+ def test_uv_file_watcher_is_watching(self, use_bytes=use_bytes):
+ try:
+ w = create_file_watcher(pl=get_fallback_logger(), watcher_type='uv')
+ except UvNotFound:
+ raise SkipTest('Pyuv is not available')
+ self.do_test_file_watcher_is_watching(w, use_bytes)
+
+ for wt in ('uv', 'inotify'):
+ l['test_{0}_file_watcher_{1}'.format(wt, btn)] = locals()['test_{0}_file_watcher'.format(wt)]
+ l['test_{0}_tree_watcher_{1}'.format(wt, btn)] = locals()['test_{0}_tree_watcher'.format(wt)]
+ l['test_{0}_file_watcher_is_watching_{1}'.format(wt, btn)] = (
+ locals()['test_{0}_file_watcher_is_watching'.format(wt)])
+ l['test_{0}_file_watcher_is_watching_{1}'.format('stat', btn)] = (
+ locals()['test_{0}_file_watcher_is_watching'.format('stat')])
+
+
+class TestFilesystemWatchers(TestCase):
+ def do_test_for_change(self, watcher, path):
+ st = monotonic()
+ while monotonic() - st < 1:
+ if watcher(path):
+ return
+ sleep(0.1)
+ self.fail('The change to {0} was not detected'.format(path))
+
+ def do_test_file_watcher(self, w, use_bytes=False):
+ try:
+ f1, f2, f3 = map(lambda x: os.path.join(INOTIFY_DIR, 'file%d' % x), (1, 2, 3))
+ ne = os.path.join(INOTIFY_DIR, 'notexists')
+ if use_bytes:
+ f1 = f1.encode('utf-8')
+ f2 = f2.encode('utf-8')
+ f3 = f3.encode('utf-8')
+ ne = ne.encode('utf-8')
+ with open(f1, 'wb'):
+ with open(f2, 'wb'):
+ with open(f3, 'wb'):
+ pass
+ self.assertRaises(OSError, w, ne)
+ self.assertTrue(w(f1))
+ self.assertTrue(w(f2))
+ os.utime(f1, None), os.utime(f2, None)
+ self.do_test_for_change(w, f1)
+ self.do_test_for_change(w, f2)
+ # Repeat once
+ os.utime(f1, None), os.utime(f2, None)
+ self.do_test_for_change(w, f1)
+ self.do_test_for_change(w, f2)
+ # Check that no false changes are reported
+ self.assertFalse(w(f1), 'Spurious change detected')
+ self.assertFalse(w(f2), 'Spurious change detected')
+ # Check that open the file with 'w' triggers a change
+ with open(f1, 'wb'):
+ with open(f2, 'wb'):
+ pass
+ self.do_test_for_change(w, f1)
+ self.do_test_for_change(w, f2)
+ # Check that writing to a file with 'a' triggers a change
+ with open(f1, 'ab') as f:
+ f.write(b'1')
+ self.do_test_for_change(w, f1)
+ # Check that deleting a file registers as a change
+ os.unlink(f1)
+ self.do_test_for_change(w, f1)
+ # Test that changing the inode of a file does not cause it to stop
+ # being watched
+ os.rename(f3, f2)
+ self.do_test_for_change(w, f2)
+ self.assertFalse(w(f2), 'Spurious change detected')
+ os.utime(f2, None)
+ self.do_test_for_change(w, f2)
+ finally:
+ clear_dir(INOTIFY_DIR)
+
+ def do_test_tree_watcher(self, tw, use_bytes=False):
+ try:
+ inotify_dir = INOTIFY_DIR
+ subdir = os.path.join(inotify_dir, 'subdir')
+ t1 = os.path.join(inotify_dir, 'tree1')
+ ts1 = os.path.join(subdir, 'tree1')
+ suffix = '1'
+ f = os.path.join(subdir, 'f')
+ if use_bytes:
+ inotify_dir = inotify_dir.encode('utf-8')
+ subdir = subdir.encode('utf-8')
+ t1 = t1.encode('utf-8')
+ ts1 = ts1.encode('utf-8')
+ suffix = suffix.encode('utf-8')
+ f = f.encode('utf-8')
+ os.mkdir(subdir)
+ try:
+ if tw.watch(inotify_dir).is_dummy:
+ raise SkipTest('No tree watcher available')
+ except UvNotFound:
+ raise SkipTest('Pyuv is not available')
+ except INotifyError:
+ raise SkipTest('INotify is not available')
+ self.assertTrue(tw(inotify_dir))
+ self.assertFalse(tw(inotify_dir))
+ changed = partial(self.do_test_for_change, tw, inotify_dir)
+ open(t1, 'w').close()
+ changed()
+ open(ts1, 'w').close()
+ changed()
+ os.unlink(ts1)
+ changed()
+ os.rmdir(subdir)
+ changed()
+ os.mkdir(subdir)
+ changed()
+ os.rename(subdir, subdir + suffix)
+ changed()
+ shutil.rmtree(subdir + suffix)
+ changed()
+ os.mkdir(subdir)
+ open(f, 'w').close()
+ changed()
+ with open(f, 'a') as s:
+ s.write(' ')
+ changed()
+ os.rename(f, f + suffix)
+ changed()
+ finally:
+ clear_dir(inotify_dir)
+
+ def do_test_file_watcher_is_watching(self, w, use_bytes=False):
+ try:
+ f1, f2, f3 = map(lambda x: os.path.join(INOTIFY_DIR, 'file%d' % x), (1, 2, 3))
+ ne = os.path.join(INOTIFY_DIR, 'notexists')
+ if use_bytes:
+ f1 = f1.encode('utf-8')
+ f2 = f2.encode('utf-8')
+ f3 = f3.encode('utf-8')
+ ne = ne.encode('utf-8')
+ with open(f1, 'wb'):
+ with open(f2, 'wb'):
+ with open(f3, 'wb'):
+ pass
+ self.assertRaises(OSError, w, ne)
+ try:
+ w(ne)
+ except OSError as e:
+ self.assertEqual(e.errno, ENOENT)
+ self.assertTrue(w(f1))
+ self.assertFalse(w.is_watching(ne))
+ self.assertTrue(w.is_watching(f1))
+ self.assertFalse(w.is_watching(f2))
+ finally:
+ clear_dir(INOTIFY_DIR)
+
+ set_watcher_tests(locals())
+
+
+old_cwd = None
+
+
+def setUpModule():
+ global old_cwd
+ old_cwd = os.getcwd()
+ os.chdir(os.path.dirname(os.path.dirname(__file__)))
+ os.mkdir(INOTIFY_DIR)
+
+
+def tearDownModule():
+ shutil.rmtree(INOTIFY_DIR)
+ os.chdir(old_cwd)
+
+
+if __name__ == '__main__':
+ from tests.modules import main
+ main()