From ca59bbcd5eac2ad8f873fe3e51d163e9b0bbdc8b Mon Sep 17 00:00:00 2001 From: Avram Lubkin Date: Wed, 23 May 2018 11:15:27 -0400 Subject: [PATCH 1/8] Add support for Python 3.7 --- .travis.yml | 6 +++++- blessed/sequences.py | 4 ++-- blessed/tests/test_core.py | 3 ++- setup.py | 3 +++ tox.ini | 14 ++++++++++---- 5 files changed, 22 insertions(+), 8 deletions(-) diff --git a/.travis.yml b/.travis.yml index ecb95fbf..ccc2438c 100644 --- a/.travis.yml +++ b/.travis.yml @@ -10,7 +10,11 @@ matrix: - python: 3.4 env: TOXENV=py34 TEST_QUICK=1 COVERAGE_ID=travis-ci - python: 3.5 - env: TOXENV=py35 COVERAGE_ID=travis-ci + env: TOXENV=py35 TEST_QUICK=1 COVERAGE_ID=travis-ci + - python: 3.6 + env: TOXENV=py36 COVERAGE_ID=travis-ci + - python: 3.7-dev + env: TOXENV=py37 COVERAGE_ID=travis-ci install: - pip install tox script: diff --git a/blessed/sequences.py b/blessed/sequences.py index cad01796..6bff1b55 100644 --- a/blessed/sequences.py +++ b/blessed/sequences.py @@ -129,9 +129,9 @@ def build(cls, name, capability, attribute, nparams=0, return cls(name, pattern, attribute) if match_grouped: - pattern = re.sub(r'(\d+)', _numeric_regex, _outp) + pattern = re.sub(r'(\d+)', lambda x: _numeric_regex, _outp) else: - pattern = re.sub(r'\d+', _numeric_regex, _outp) + pattern = re.sub(r'\d+', lambda x: _numeric_regex, _outp) return cls(name, pattern, attribute) diff --git a/blessed/tests/test_core.py b/blessed/tests/test_core.py index 046088fa..bda924d4 100644 --- a/blessed/tests/test_core.py +++ b/blessed/tests/test_core.py @@ -479,7 +479,8 @@ def test_termcap_repr(): given_ttype='vt220' given_capname = 'cursor_up' - expected = [r"", + expected = [r"", + r"", r""] @as_subprocess diff --git a/setup.py b/setup.py index 7c10f629..c7c50f74 100755 --- a/setup.py +++ b/setup.py @@ -59,6 +59,9 @@ def _get_long_description(fname): 'Programming Language :: Python :: 3.2', 'Programming Language :: Python :: 3.3', 'Programming Language :: Python :: 3.4', + 'Programming Language :: Python :: 3.5', + 'Programming Language :: Python :: 3.6', + 'Programming Language :: Python :: 3.7', 'Topic :: Software Development :: Libraries', 'Topic :: Software Development :: User Interfaces', 'Topic :: Terminals' diff --git a/tox.ini b/tox.ini index 4820f2fc..4b519e18 100644 --- a/tox.ini +++ b/tox.ini @@ -1,5 +1,5 @@ [tox] -envlist = about, sa, sphinx, py{26,27,34,35} +envlist = about, sa, sphinx, py{26,27,34,35,36,37} skip_missing_interpreters = true [testenv] @@ -30,14 +30,14 @@ commands = coveralls [testenv:about] deps = -rrequirements-about.txt -basepython = python3.5 +basepython = python3.6 commands = python {toxinidir}/bin/display-sighandlers.py python {toxinidir}/bin/display-terminalinfo.py python {toxinidir}/bin/display-fpathconf.py python {toxinidir}/bin/display-maxcanon.py [testenv:sa] -basepython = python3.5 +basepython = python3.6 deps = -rrequirements-analysis.txt -rrequirements-about.txt commands = python -m compileall -fq {toxinidir}/blessed @@ -49,7 +49,7 @@ commands = python -m compileall -fq {toxinidir}/blessed [testenv:sphinx] whitelist_externals = echo -basepython = python3.5 +basepython = python3.6 deps = -rrequirements-docs.txt commands = {envbindir}/sphinx-build -v -W \ -d {toxinidir}/docs/_build/doctrees \ @@ -57,6 +57,12 @@ commands = {envbindir}/sphinx-build -v -W \ {toxinidir}/docs/_build/html echo "--> open docs/_build/html/index.html for review." +[testenv:py35] +# there is not much difference of py34 vs. 35 in blessed +# library; prefer testing integration against py35, and +# just do a 'quick' on py34, if exists. +setenv = TEST_QUICK=1 + [testenv:py34] # there is not much difference of py34 vs. 35 in blessed # library; prefer testing integration against py35, and From 042573d460612bf810556a3229d4170ea93358d6 Mon Sep 17 00:00:00 2001 From: Avram Lubkin Date: Wed, 23 May 2018 11:27:49 -0400 Subject: [PATCH 2/8] Workaround for https://github.com/travis-ci/travis-ci/issues/8363 --- .travis.yml | 3 +++ 1 file changed, 3 insertions(+) diff --git a/.travis.yml b/.travis.yml index ccc2438c..6fa927a0 100644 --- a/.travis.yml +++ b/.travis.yml @@ -15,6 +15,9 @@ matrix: env: TOXENV=py36 COVERAGE_ID=travis-ci - python: 3.7-dev env: TOXENV=py37 COVERAGE_ID=travis-ci +before_install: + # work around https://github.com/travis-ci/travis-ci/issues/8363 + - pyenv global system 3.6 install: - pip install tox script: From 279b4276bcdc234844960cc62440a8e09b89cba5 Mon Sep 17 00:00:00 2001 From: Avram Lubkin Date: Wed, 23 May 2018 11:43:54 -0400 Subject: [PATCH 3/8] Pass TRAVIS envvar to tox --- tox.ini | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tox.ini b/tox.ini index 4b519e18..bf6013ad 100644 --- a/tox.ini +++ b/tox.ini @@ -5,7 +5,7 @@ skip_missing_interpreters = true [testenv] whitelist_externals = cp setenv = PYTHONIOENCODING=UTF8 -passenv = TEST_QUICK TEST_FULL +passenv = TEST_QUICK TEST_FULL TRAVIS deps = -rrequirements-tests.txt commands = {envbindir}/py.test {posargs:\ --strict --verbose --verbose --color=yes \ From 42ee7d6fab9b663288dca7fc9948d6ae7fc7f074 Mon Sep 17 00:00:00 2001 From: Avram Lubkin Date: Wed, 23 May 2018 14:42:58 -0400 Subject: [PATCH 4/8] Fix issues with static analysis --- .landscape.yml | 5 ++++- bin/display-maxcanon.py | 2 +- bin/editor.py | 6 +++--- bin/on_resize.py | 2 +- bin/progress_bar.py | 2 +- bin/worms.py | 4 ++-- blessed/formatters.py | 22 +++++++++++----------- blessed/keyboard.py | 10 +++------- blessed/sequences.py | 7 ++++--- blessed/terminal.py | 16 +++++++++------- requirements-analysis.txt | 1 + tox.ini | 1 + 12 files changed, 41 insertions(+), 37 deletions(-) diff --git a/.landscape.yml b/.landscape.yml index 8c4ccb78..30a05c1f 100644 --- a/.landscape.yml +++ b/.landscape.yml @@ -45,7 +45,10 @@ pep257: - 'D203' # 1 blank line required after class docstring - 'D204' - + # Multi-line docstring summary should start at the first line + - 'D212' + # First line should be in imperative mood + - 'D401' pep8: # style checking diff --git a/bin/display-maxcanon.py b/bin/display-maxcanon.py index 1df8a7e4..2839202f 100755 --- a/bin/display-maxcanon.py +++ b/bin/display-maxcanon.py @@ -1,6 +1,6 @@ #!/usr/bin/env python """ -This tool uses pexpect to test expected Canonical mode length. +A tool which uses pexpect to test expected Canonical mode length. All systems use the value of MAX_CANON which can be found using fpathconf(3) value PC_MAX_CANON -- with the exception of Linux diff --git a/bin/editor.py b/bin/editor.py index 877e2787..4b1c36e0 100755 --- a/bin/editor.py +++ b/bin/editor.py @@ -81,9 +81,9 @@ def readline(term, width=20): text = text[:-1] # https://utcc.utoronto.ca/~cks/space/blog/unix/HowUnixBackspaces # - # "When you hit backspace, the kernel tty line discipline rubs out your previous - # character by printing (in the simple case) Ctrl-H, a space, and then another - # Ctrl-H." + # "When you hit backspace, the kernel tty line discipline rubs out + # your previous character by printing (in the simple case) + # Ctrl-H, a space, and then another Ctrl-H." echo(u'\b \b') return text diff --git a/bin/on_resize.py b/bin/on_resize.py index 7a483376..e77bb453 100755 --- a/bin/on_resize.py +++ b/bin/on_resize.py @@ -1,6 +1,6 @@ #!/usr/bin/env python """ -This is an example application for the 'blessed' Terminal library for python. +Example application for the 'blessed' Terminal library for python. Window size changes are caught by the 'on_resize' function using a traditional signal handler. Meanwhile, blocking keyboard input is displayed to stdout. diff --git a/bin/progress_bar.py b/bin/progress_bar.py index 037f1ad2..6a2ebfba 100755 --- a/bin/progress_bar.py +++ b/bin/progress_bar.py @@ -1,6 +1,6 @@ #!/usr/bin/env python """ -This is an example application for the 'blessed' Terminal library for python. +Example application for the 'blessed' Terminal library for python. This isn't a real progress bar, just a sample "animated prompt" of sorts that demonstrates the separate move_x() and move_y() functions, made diff --git a/bin/worms.py b/bin/worms.py index 907667ec..f1353c77 100755 --- a/bin/worms.py +++ b/bin/worms.py @@ -1,6 +1,6 @@ #!/usr/bin/env python """ -This is an example application for the 'blessed' Terminal library for python. +Example application for the 'blessed' Terminal library for python. It is also an experiment in functional programming. """ @@ -25,7 +25,7 @@ import sys def echo(text): - """python 2 version of print(end='', flush=True).""" + """Python 2 version of print(end='', flush=True).""" sys.stdout.write(u'{0}'.format(text)) sys.stdout.flush() diff --git a/blessed/formatters.py b/blessed/formatters.py index 66020738..5484bb25 100644 --- a/blessed/formatters.py +++ b/blessed/formatters.py @@ -1,4 +1,4 @@ -"""This sub-module provides sequence-formatting functions.""" +"""Sub-module providing sequence-formatting functions.""" # standard imports import curses @@ -59,10 +59,10 @@ def __new__(cls, *args): :arg normal: terminating sequence for this capability (optional). :arg name: name of this terminal capability (optional). """ - assert len(args) and len(args) < 4, args + assert args and len(args) < 4, args new = six.text_type.__new__(cls, args[0]) - new._normal = len(args) > 1 and args[1] or u'' - new._name = len(args) > 2 and args[2] or u'' + new._normal = args[1] if len(args) > 1 else u'' + new._name = args[2] if len(args) > 2 else u'' return new def __call__(self, *args): @@ -84,7 +84,7 @@ def __call__(self, *args): except TypeError as err: # If the first non-int (i.e. incorrect) arg was a string, suggest # something intelligent: - if len(args) and isinstance(args[0], six.string_types): + if args and isinstance(args[0], six.string_types): raise TypeError( "A native or nonexistent capability template, %r received" " invalid argument %r: %s. You probably misspelled a" @@ -135,13 +135,13 @@ def __new__(cls, *args): :arg normal: terminating sequence for this capability (optional). :arg name: name of this terminal capability (optional). """ - assert len(args) and len(args) < 4, args + assert args and len(args) < 4, args assert isinstance(args[0], tuple), args[0] assert callable(args[0][1]), args[0][1] new = six.text_type.__new__(cls, args[0][0]) new._fmt_args = args[0][1] - new._normal = len(args) > 1 and args[1] or u'' - new._name = len(args) > 2 and args[2] or u'' + new._normal = args[1] if len(args) > 1 else u'' + new._name = args[2] if len(args) > 2 else u'' return new def __call__(self, *args): @@ -226,7 +226,7 @@ def __new__(cls, *args): """ assert 1 <= len(args) <= 2, args new = six.text_type.__new__(cls, args[0]) - new._normal = len(args) > 1 and args[1] or u'' + new._normal = args[1] if len(args) > 1 else u'' return new def __call__(self, *args): @@ -246,7 +246,7 @@ def __call__(self, *args): expected_types=six.string_types, )) postfix = u'' - if len(self) and self._normal: + if self and self._normal: postfix = self._normal _refresh = self._normal + self args = [_refresh.join(ucs_part.split(self._normal)) @@ -281,7 +281,7 @@ def __call__(self, *args): the first arg, acting in place of :class:`FormattingString` without any attributes. """ - if len(args) == 0 or isinstance(args[0], int): + if not args or isinstance(args[0], int): # As a NullCallableString, even when provided with a parameter, # such as t.color(5), we must also still be callable, fe: # diff --git a/blessed/keyboard.py b/blessed/keyboard.py index 81b82573..bbf43265 100644 --- a/blessed/keyboard.py +++ b/blessed/keyboard.py @@ -1,4 +1,4 @@ -"""This sub-module provides 'keyboard awareness'.""" +"""Sub-module providing 'keyboard awareness'.""" # std imports import curses.has_key @@ -52,8 +52,7 @@ def is_sequence(self): def __repr__(self): """Docstring overwritten.""" - return (self._name is None and - six.text_type.__repr__(self) or + return (six.text_type.__repr__(self) if self._name is None else self._name) __repr__.__doc__ = six.text_type.__doc__ @@ -246,10 +245,7 @@ class method :meth:`~.Terminal.kbhit` and similar functions. :returns: time remaining as float. If no time is remaining, then the integer ``0`` is returned. """ - if timeout is not None: - if timeout == 0: - return 0 - return max(0, timeout - (time.time() - stime)) + return max(0, timeout - (time.time() - stime)) if timeout else timeout def _read_until(term, pattern, timeout): diff --git a/blessed/sequences.py b/blessed/sequences.py index 6bff1b55..e2aef824 100644 --- a/blessed/sequences.py +++ b/blessed/sequences.py @@ -1,5 +1,5 @@ # encoding: utf-8 -"""This module provides 'sequence awareness'.""" +"""Module providing 'sequence awareness'.""" # std imports import functools import textwrap @@ -136,7 +136,7 @@ def build(cls, name, capability, attribute, nparams=0, class SequenceTextWrapper(textwrap.TextWrapper): - """This docstring overridden.""" + """Docstring overridden.""" def __init__(self, width, term, **kwargs): """ @@ -195,7 +195,8 @@ def _wrap_chunks(self, chunks): return lines def _handle_long_word(self, reversed_chunks, cur_line, cur_len, width): - """Sequence-aware :meth:`textwrap.TextWrapper._handle_long_word`. + """ + Sequence-aware :meth:`textwrap.TextWrapper._handle_long_word`. This simply ensures that word boundaries are not broken mid-sequence, as standard python textwrap would incorrectly determine the length diff --git a/blessed/terminal.py b/blessed/terminal.py index fc9c8af7..84ea99bf 100644 --- a/blessed/terminal.py +++ b/blessed/terminal.py @@ -1,5 +1,5 @@ # encoding: utf-8 -"""This module contains :class:`Terminal`, the primary API entry point.""" +"""Module containing :class:`Terminal`, the primary API entry point.""" # pylint: disable=too-many-lines # Too many lines in module (1027/1000) import codecs @@ -75,6 +75,9 @@ ) +_CUR_TERM = None # See comments at end of file + + class Terminal(object): """ An abstraction for color, style, positioning, and input in the terminal. @@ -191,9 +194,8 @@ def __init__(self, kind=None, stream=None, force_styling=False): self._normal = None # cache normal attr, preventing recursive lookups # The descriptor to direct terminal initialization sequences to. - self._init_descriptor = (stream_fd is None and - sys.__stdout__.fileno() or - stream_fd) + self._init_descriptor = (sys.__stdout__.fileno() if stream_fd is None + else stream_fd) self._kind = kind or os.environ.get('TERM', 'unknown') if self.does_styling: @@ -1009,8 +1011,8 @@ def raw(self): A context manager for :func:`tty.setraw`. Although both :meth:`break` and :meth:`raw` modes allow each keystroke - to be read immediately after it is pressed, Raw mode disables processing - of input and output. + to be read immediately after it is pressed, Raw mode disables + processing of input and output. In cbreak mode, special input characters such as ``^C`` or ``^S`` are interpreted by the terminal driver and excluded from the stdin stream. @@ -1177,6 +1179,7 @@ class WINSZ(collections.namedtuple('WINSZ', ( _BUF = '\x00' * struct.calcsize(_FMT) +#: _CUR_TERM = None #: From libcurses/doc/ncurses-intro.html (ESR, Thomas Dickey, et. al):: #: #: "After the call to setupterm(), the global variable cur_term is set to @@ -1196,4 +1199,3 @@ class WINSZ(collections.namedtuple('WINSZ', ( #: Therefore, the :attr:`Terminal.kind` of each :class:`Terminal` is #: essentially a singleton. This global variable reflects that, and a warning #: is emitted if somebody expects otherwise. -_CUR_TERM = None diff --git a/requirements-analysis.txt b/requirements-analysis.txt index f36f1409..1f5e69b4 100644 --- a/requirements-analysis.txt +++ b/requirements-analysis.txt @@ -1,3 +1,4 @@ prospector[with_pyroma] restructuredtext_lint doc8 +Pygments diff --git a/tox.ini b/tox.ini index bf6013ad..f68268f6 100644 --- a/tox.ini +++ b/tox.ini @@ -43,6 +43,7 @@ deps = -rrequirements-analysis.txt commands = python -m compileall -fq {toxinidir}/blessed {envbindir}/prospector \ --die-on-tool-error \ + --no-external-config \ {toxinidir} {envbindir}/rst-lint README.rst {envbindir}/doc8 --ignore-path docs/_build --ignore D000 docs From 465f384edd6d2ca84b0325face95b9eac4dd56e6 Mon Sep 17 00:00:00 2001 From: Avram Lubkin Date: Wed, 23 May 2018 15:00:10 -0400 Subject: [PATCH 5/8] Temporarily disable display-maxcanon.py in tox --- tox.ini | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/tox.ini b/tox.ini index f68268f6..42838f58 100644 --- a/tox.ini +++ b/tox.ini @@ -34,7 +34,8 @@ basepython = python3.6 commands = python {toxinidir}/bin/display-sighandlers.py python {toxinidir}/bin/display-terminalinfo.py python {toxinidir}/bin/display-fpathconf.py - python {toxinidir}/bin/display-maxcanon.py + # Temporarily disable until limits are added to MAX_CANON logic + # python {toxinidir}/bin/display-maxcanon.py [testenv:sa] basepython = python3.6 From f59e53607b51ec866bd9c13e1673873a774eed27 Mon Sep 17 00:00:00 2001 From: Jeff Quast Date: Tue, 19 Jun 2018 21:58:45 -0700 Subject: [PATCH 6/8] remove timing tests for folks timing tests can only be tested using integration tests such as these, but debian wishes to run all of our tests, there are not any formal contracts to prevent people from running certain tests, such as explicit definition of test groups. --- blessed/tests/test_keyboard.py | 1124 ++++++++++++++++---------------- 1 file changed, 562 insertions(+), 562 deletions(-) diff --git a/blessed/tests/test_keyboard.py b/blessed/tests/test_keyboard.py index 9df528d6..dd351969 100644 --- a/blessed/tests/test_keyboard.py +++ b/blessed/tests/test_keyboard.py @@ -5,7 +5,7 @@ import tempfile import signal import curses -import time +#import time import math import tty # NOQA import pty @@ -35,89 +35,89 @@ unichr = chr -@pytest.mark.skipif(os.environ.get('TEST_QUICK', None) is not None, - reason="TEST_QUICK specified") -def test_kbhit_interrupted(): - "kbhit() should not be interrupted with a signal handler." - pid, master_fd = pty.fork() - if pid == 0: - cov = init_subproc_coverage('test_kbhit_interrupted') - - # child pauses, writes semaphore and begins awaiting input - global got_sigwinch - got_sigwinch = False - - def on_resize(sig, action): - global got_sigwinch - got_sigwinch = True - - term = TestTerminal() - signal.signal(signal.SIGWINCH, on_resize) - read_until_semaphore(sys.__stdin__.fileno(), semaphore=SEMAPHORE) - os.write(sys.__stdout__.fileno(), SEMAPHORE) - with term.raw(): - assert term.inkey(timeout=1.05) == u'' - os.write(sys.__stdout__.fileno(), b'complete') - assert got_sigwinch - if cov is not None: - cov.stop() - cov.save() - os._exit(0) - - with echo_off(master_fd): - os.write(master_fd, SEND_SEMAPHORE) - read_until_semaphore(master_fd) - stime = time.time() - os.kill(pid, signal.SIGWINCH) - output = read_until_eof(master_fd) - - pid, status = os.waitpid(pid, 0) - assert output == u'complete' - assert os.WEXITSTATUS(status) == 0 - assert math.floor(time.time() - stime) == 1.0 - - -@pytest.mark.skipif(os.environ.get('TEST_QUICK', None) is not None, - reason="TEST_QUICK specified") -def test_kbhit_interrupted_nonetype(): - "kbhit() should also allow interruption with timeout of None." - pid, master_fd = pty.fork() - if pid == 0: - cov = init_subproc_coverage('test_kbhit_interrupted_nonetype') - - # child pauses, writes semaphore and begins awaiting input - global got_sigwinch - got_sigwinch = False - - def on_resize(sig, action): - global got_sigwinch - got_sigwinch = True - - term = TestTerminal() - signal.signal(signal.SIGWINCH, on_resize) - read_until_semaphore(sys.__stdin__.fileno(), semaphore=SEMAPHORE) - os.write(sys.__stdout__.fileno(), SEMAPHORE) - with term.raw(): - term.inkey(timeout=1) - os.write(sys.__stdout__.fileno(), b'complete') - assert got_sigwinch - if cov is not None: - cov.stop() - cov.save() - os._exit(0) - - with echo_off(master_fd): - os.write(master_fd, SEND_SEMAPHORE) - read_until_semaphore(master_fd) - stime = time.time() - time.sleep(0.05) - os.kill(pid, signal.SIGWINCH) - output = read_until_eof(master_fd) - - pid, status = os.waitpid(pid, 0) - assert output == u'complete' - assert os.WEXITSTATUS(status) == 0 - assert math.floor(time.time() - stime) == 1.0 +#@pytest.mark.skipif(os.environ.get('TEST_QUICK', None) is not None, +# reason="TEST_QUICK specified") +#def test_kbhit_interrupted(): +# "kbhit() should not be interrupted with a signal handler." +# pid, master_fd = pty.fork() +# if pid == 0: +# cov = init_subproc_coverage('test_kbhit_interrupted') +# +# # child pauses, writes semaphore and begins awaiting input +# global got_sigwinch +# got_sigwinch = False +# +# def on_resize(sig, action): +# global got_sigwinch +# got_sigwinch = True +# +# term = TestTerminal() +# signal.signal(signal.SIGWINCH, on_resize) +# read_until_semaphore(sys.__stdin__.fileno(), semaphore=SEMAPHORE) +# os.write(sys.__stdout__.fileno(), SEMAPHORE) +# with term.raw(): +# assert term.inkey(timeout=1.05) == u'' +# os.write(sys.__stdout__.fileno(), b'complete') +# assert got_sigwinch +# if cov is not None: +# cov.stop() +# cov.save() +# os._exit(0) +# +# with echo_off(master_fd): +# os.write(master_fd, SEND_SEMAPHORE) +# read_until_semaphore(master_fd) +# stime = time.time() +# os.kill(pid, signal.SIGWINCH) +# output = read_until_eof(master_fd) +# +# pid, status = os.waitpid(pid, 0) +# assert output == u'complete' +# assert os.WEXITSTATUS(status) == 0 +# assert math.floor(time.time() - stime) == 1.0 +# +# +#@pytest.mark.skipif(os.environ.get('TEST_QUICK', None) is not None, +# reason="TEST_QUICK specified") +#def test_kbhit_interrupted_nonetype(): +# "kbhit() should also allow interruption with timeout of None." +# pid, master_fd = pty.fork() +# if pid == 0: +# cov = init_subproc_coverage('test_kbhit_interrupted_nonetype') +# +# # child pauses, writes semaphore and begins awaiting input +# global got_sigwinch +# got_sigwinch = False +# +# def on_resize(sig, action): +# global got_sigwinch +# got_sigwinch = True +# +# term = TestTerminal() +# signal.signal(signal.SIGWINCH, on_resize) +# read_until_semaphore(sys.__stdin__.fileno(), semaphore=SEMAPHORE) +# os.write(sys.__stdout__.fileno(), SEMAPHORE) +# with term.raw(): +# term.inkey(timeout=1) +# os.write(sys.__stdout__.fileno(), b'complete') +# assert got_sigwinch +# if cov is not None: +# cov.stop() +# cov.save() +# os._exit(0) +# +# with echo_off(master_fd): +# os.write(master_fd, SEND_SEMAPHORE) +# read_until_semaphore(master_fd) +# stime = time.time() +# time.sleep(0.05) +# os.kill(pid, signal.SIGWINCH) +# output = read_until_eof(master_fd) +# +# pid, status = os.waitpid(pid, 0) +# assert output == u'complete' +# assert os.WEXITSTATUS(status) == 0 +# assert math.floor(time.time() - stime) == 1.0 def test_break_input_no_kb(): @@ -172,484 +172,484 @@ def child(): child() -def test_kbhit_no_kb(): - "kbhit() always immediately returns False without a keyboard." - @as_subprocess - def child(): - term = TestTerminal(stream=six.StringIO()) - stime = time.time() - assert term._keyboard_fd is None - assert not term.kbhit(timeout=1.1) - assert math.floor(time.time() - stime) == 1.0 - child() - - -def test_keystroke_0s_cbreak_noinput(): - "0-second keystroke without input; '' should be returned." - @as_subprocess - def child(): - term = TestTerminal() - with term.cbreak(): - stime = time.time() - inp = term.inkey(timeout=0) - assert (inp == u'') - assert (math.floor(time.time() - stime) == 0.0) - child() - - -def test_keystroke_0s_cbreak_noinput_nokb(): - "0-second keystroke without data in input stream and no keyboard/tty." - @as_subprocess - def child(): - term = TestTerminal(stream=six.StringIO()) - with term.cbreak(): - stime = time.time() - inp = term.inkey(timeout=0) - assert (inp == u'') - assert (math.floor(time.time() - stime) == 0.0) - child() - - -@pytest.mark.skipif(os.environ.get('TEST_QUICK', None) is not None, - reason="TEST_QUICK specified") -def test_keystroke_1s_cbreak_noinput(): - "1-second keystroke without input; '' should be returned after ~1 second." - @as_subprocess - def child(): - term = TestTerminal() - with term.cbreak(): - stime = time.time() - inp = term.inkey(timeout=1) - assert (inp == u'') - assert (math.floor(time.time() - stime) == 1.0) - child() - - -@pytest.mark.skipif(os.environ.get('TEST_QUICK', None) is not None, - reason="TEST_QUICK specified") -def test_keystroke_1s_cbreak_noinput_nokb(): - "1-second keystroke without input or keyboard." - @as_subprocess - def child(): - term = TestTerminal(stream=six.StringIO()) - with term.cbreak(): - stime = time.time() - inp = term.inkey(timeout=1) - assert (inp == u'') - assert (math.floor(time.time() - stime) == 1.0) - child() - - -def test_keystroke_0s_cbreak_with_input(): - "0-second keystroke with input; Keypress should be immediately returned." - pid, master_fd = pty.fork() - if pid == 0: - cov = init_subproc_coverage('test_keystroke_0s_cbreak_with_input') - # child pauses, writes semaphore and begins awaiting input - term = TestTerminal() - read_until_semaphore(sys.__stdin__.fileno(), semaphore=SEMAPHORE) - os.write(sys.__stdout__.fileno(), SEMAPHORE) - with term.cbreak(): - inp = term.inkey(timeout=0) - os.write(sys.__stdout__.fileno(), inp.encode('utf-8')) - if cov is not None: - cov.stop() - cov.save() - os._exit(0) - - with echo_off(master_fd): - os.write(master_fd, SEND_SEMAPHORE) - os.write(master_fd, u'x'.encode('ascii')) - read_until_semaphore(master_fd) - stime = time.time() - output = read_until_eof(master_fd) - - pid, status = os.waitpid(pid, 0) - assert output == u'x' - assert os.WEXITSTATUS(status) == 0 - assert math.floor(time.time() - stime) == 0.0 - - -def test_keystroke_cbreak_with_input_slowly(): - "0-second keystroke with input; Keypress should be immediately returned." - pid, master_fd = pty.fork() - if pid == 0: - cov = init_subproc_coverage('test_keystroke_cbreak_with_input_slowly') - # child pauses, writes semaphore and begins awaiting input - term = TestTerminal() - read_until_semaphore(sys.__stdin__.fileno(), semaphore=SEMAPHORE) - os.write(sys.__stdout__.fileno(), SEMAPHORE) - with term.cbreak(): - while True: - inp = term.inkey(timeout=0.5) - os.write(sys.__stdout__.fileno(), inp.encode('utf-8')) - if inp == 'X': - break - if cov is not None: - cov.stop() - cov.save() - os._exit(0) - - with echo_off(master_fd): - os.write(master_fd, SEND_SEMAPHORE) - os.write(master_fd, u'a'.encode('ascii')) - time.sleep(0.1) - os.write(master_fd, u'b'.encode('ascii')) - time.sleep(0.1) - os.write(master_fd, u'cdefgh'.encode('ascii')) - time.sleep(0.1) - os.write(master_fd, u'X'.encode('ascii')) - read_until_semaphore(master_fd) - stime = time.time() - output = read_until_eof(master_fd) - - pid, status = os.waitpid(pid, 0) - assert output == u'abcdefghX' - assert os.WEXITSTATUS(status) == 0 - assert math.floor(time.time() - stime) == 0.0 - - -def test_keystroke_0s_cbreak_multibyte_utf8(): - "0-second keystroke with multibyte utf-8 input; should decode immediately." - # utf-8 bytes represent "latin capital letter upsilon". - pid, master_fd = pty.fork() - if pid == 0: # child - cov = init_subproc_coverage('test_keystroke_0s_cbreak_multibyte_utf8') - term = TestTerminal() - read_until_semaphore(sys.__stdin__.fileno(), semaphore=SEMAPHORE) - os.write(sys.__stdout__.fileno(), SEMAPHORE) - with term.cbreak(): - inp = term.inkey(timeout=0) - os.write(sys.__stdout__.fileno(), inp.encode('utf-8')) - if cov is not None: - cov.stop() - cov.save() - os._exit(0) - - with echo_off(master_fd): - os.write(master_fd, SEND_SEMAPHORE) - os.write(master_fd, u'\u01b1'.encode('utf-8')) - read_until_semaphore(master_fd) - stime = time.time() - output = read_until_eof(master_fd) - pid, status = os.waitpid(pid, 0) - assert output == u'Ʊ' - assert os.WEXITSTATUS(status) == 0 - assert math.floor(time.time() - stime) == 0.0 - - -@pytest.mark.skipif(os.environ.get('TRAVIS', None) is not None, - reason="travis-ci does not handle ^C very well.") -def test_keystroke_0s_raw_input_ctrl_c(): - "0-second keystroke with raw allows receiving ^C." - pid, master_fd = pty.fork() - if pid == 0: # child - cov = init_subproc_coverage('test_keystroke_0s_raw_input_ctrl_c') - term = TestTerminal() - read_until_semaphore(sys.__stdin__.fileno(), semaphore=SEMAPHORE) - with term.raw(): - os.write(sys.__stdout__.fileno(), RECV_SEMAPHORE) - inp = term.inkey(timeout=0) - os.write(sys.__stdout__.fileno(), inp.encode('latin1')) - if cov is not None: - cov.stop() - cov.save() - os._exit(0) - - with echo_off(master_fd): - os.write(master_fd, SEND_SEMAPHORE) - # ensure child is in raw mode before sending ^C, - read_until_semaphore(master_fd) - os.write(master_fd, u'\x03'.encode('latin1')) - stime = time.time() - output = read_until_eof(master_fd) - pid, status = os.waitpid(pid, 0) - assert (output == u'\x03' or - output == u'' and not os.isatty(0)) - assert os.WEXITSTATUS(status) == 0 - assert math.floor(time.time() - stime) == 0.0 - - -def test_keystroke_0s_cbreak_sequence(): - "0-second keystroke with multibyte sequence; should decode immediately." - pid, master_fd = pty.fork() - if pid == 0: # child - cov = init_subproc_coverage('test_keystroke_0s_cbreak_sequence') - term = TestTerminal() - os.write(sys.__stdout__.fileno(), SEMAPHORE) - with term.cbreak(): - inp = term.inkey(timeout=0) - os.write(sys.__stdout__.fileno(), inp.name.encode('ascii')) - sys.stdout.flush() - if cov is not None: - cov.stop() - cov.save() - os._exit(0) - - with echo_off(master_fd): - os.write(master_fd, u'\x1b[D'.encode('ascii')) - read_until_semaphore(master_fd) - stime = time.time() - output = read_until_eof(master_fd) - pid, status = os.waitpid(pid, 0) - assert output == u'KEY_LEFT' - assert os.WEXITSTATUS(status) == 0 - assert math.floor(time.time() - stime) == 0.0 - - -@pytest.mark.skipif(os.environ.get('TEST_QUICK', None) is not None, - reason="TEST_QUICK specified") -def test_keystroke_1s_cbreak_with_input(): - "1-second keystroke w/multibyte sequence; should return after ~1 second." - pid, master_fd = pty.fork() - if pid == 0: # child - cov = init_subproc_coverage('test_keystroke_1s_cbreak_with_input') - term = TestTerminal() - os.write(sys.__stdout__.fileno(), SEMAPHORE) - with term.cbreak(): - inp = term.inkey(timeout=3) - os.write(sys.__stdout__.fileno(), inp.name.encode('utf-8')) - sys.stdout.flush() - if cov is not None: - cov.stop() - cov.save() - os._exit(0) - - with echo_off(master_fd): - read_until_semaphore(master_fd) - stime = time.time() - time.sleep(1) - os.write(master_fd, u'\x1b[C'.encode('ascii')) - output = read_until_eof(master_fd) - - pid, status = os.waitpid(pid, 0) - assert output == u'KEY_RIGHT' - assert os.WEXITSTATUS(status) == 0 - assert math.floor(time.time() - stime) == 1.0 - - -@pytest.mark.skipif(os.environ.get('TEST_QUICK', None) is not None, - reason="TEST_QUICK specified") -def test_esc_delay_cbreak_035(): - "esc_delay will cause a single ESC (\\x1b) to delay for 0.35." - pid, master_fd = pty.fork() - if pid == 0: # child - cov = init_subproc_coverage('test_esc_delay_cbreak_035') - term = TestTerminal() - os.write(sys.__stdout__.fileno(), SEMAPHORE) - with term.cbreak(): - stime = time.time() - inp = term.inkey(timeout=5) - measured_time = (time.time() - stime) * 100 - os.write(sys.__stdout__.fileno(), ( - '%s %i' % (inp.name, measured_time,)).encode('ascii')) - sys.stdout.flush() - if cov is not None: - cov.stop() - cov.save() - os._exit(0) - - with echo_off(master_fd): - read_until_semaphore(master_fd) - stime = time.time() - os.write(master_fd, u'\x1b'.encode('ascii')) - key_name, duration_ms = read_until_eof(master_fd).split() - - pid, status = os.waitpid(pid, 0) - assert key_name == u'KEY_ESCAPE' - assert os.WEXITSTATUS(status) == 0 - assert math.floor(time.time() - stime) == 0.0 - assert 34 <= int(duration_ms) <= 45, duration_ms - - -@pytest.mark.skipif(os.environ.get('TEST_QUICK', None) is not None, - reason="TEST_QUICK specified") -def test_esc_delay_cbreak_135(): - "esc_delay=1.35 will cause a single ESC (\\x1b) to delay for 1.35." - pid, master_fd = pty.fork() - if pid == 0: # child - cov = init_subproc_coverage('test_esc_delay_cbreak_135') - term = TestTerminal() - os.write(sys.__stdout__.fileno(), SEMAPHORE) - with term.cbreak(): - stime = time.time() - inp = term.inkey(timeout=5, esc_delay=1.35) - measured_time = (time.time() - stime) * 100 - os.write(sys.__stdout__.fileno(), ( - '%s %i' % (inp.name, measured_time,)).encode('ascii')) - sys.stdout.flush() - if cov is not None: - cov.stop() - cov.save() - os._exit(0) - - with echo_off(master_fd): - read_until_semaphore(master_fd) - stime = time.time() - os.write(master_fd, u'\x1b'.encode('ascii')) - key_name, duration_ms = read_until_eof(master_fd).split() - - pid, status = os.waitpid(pid, 0) - assert key_name == u'KEY_ESCAPE' - assert os.WEXITSTATUS(status) == 0 - assert math.floor(time.time() - stime) == 1.0 - assert 134 <= int(duration_ms) <= 145, int(duration_ms) - - -def test_esc_delay_cbreak_timout_0(): - """esc_delay still in effect with timeout of 0 ("nonblocking").""" - pid, master_fd = pty.fork() - if pid == 0: # child - cov = init_subproc_coverage('test_esc_delay_cbreak_timout_0') - term = TestTerminal() - os.write(sys.__stdout__.fileno(), SEMAPHORE) - with term.cbreak(): - stime = time.time() - inp = term.inkey(timeout=0) - measured_time = (time.time() - stime) * 100 - os.write(sys.__stdout__.fileno(), ( - '%s %i' % (inp.name, measured_time,)).encode('ascii')) - sys.stdout.flush() - if cov is not None: - cov.stop() - cov.save() - os._exit(0) - - with echo_off(master_fd): - os.write(master_fd, u'\x1b'.encode('ascii')) - read_until_semaphore(master_fd) - stime = time.time() - key_name, duration_ms = read_until_eof(master_fd).split() - - pid, status = os.waitpid(pid, 0) - assert key_name == u'KEY_ESCAPE' - assert os.WEXITSTATUS(status) == 0 - assert math.floor(time.time() - stime) == 0.0 - assert 34 <= int(duration_ms) <= 45, int(duration_ms) - - -def test_esc_delay_cbreak_nonprefix_sequence(): - "ESC a (\\x1ba) will return an ESC immediately" - pid, master_fd = pty.fork() - if pid is 0: # child - cov = init_subproc_coverage('test_esc_delay_cbreak_nonprefix_sequence') - term = TestTerminal() - os.write(sys.__stdout__.fileno(), SEMAPHORE) - with term.cbreak(): - stime = time.time() - esc = term.inkey(timeout=5) - inp = term.inkey(timeout=5) - measured_time = (time.time() - stime) * 100 - os.write(sys.__stdout__.fileno(), ( - '%s %s %i' % (esc.name, inp, measured_time,)).encode('ascii')) - sys.stdout.flush() - if cov is not None: - cov.stop() - cov.save() - os._exit(0) - - with echo_off(master_fd): - read_until_semaphore(master_fd) - stime = time.time() - os.write(master_fd, u'\x1ba'.encode('ascii')) - key1_name, key2, duration_ms = read_until_eof(master_fd).split() - - pid, status = os.waitpid(pid, 0) - assert key1_name == u'KEY_ESCAPE' - assert key2 == u'a' - assert os.WEXITSTATUS(status) == 0 - assert math.floor(time.time() - stime) == 0.0 - assert -1 <= int(duration_ms) <= 15, duration_ms - - -def test_esc_delay_cbreak_prefix_sequence(): - "An unfinished multibyte sequence (\\x1b[) will delay an ESC by .35 " - pid, master_fd = pty.fork() - if pid is 0: # child - cov = init_subproc_coverage('test_esc_delay_cbreak_prefix_sequence') - term = TestTerminal() - os.write(sys.__stdout__.fileno(), SEMAPHORE) - with term.cbreak(): - stime = time.time() - esc = term.inkey(timeout=5) - inp = term.inkey(timeout=5) - measured_time = (time.time() - stime) * 100 - os.write(sys.__stdout__.fileno(), ( - '%s %s %i' % (esc.name, inp, measured_time,)).encode('ascii')) - sys.stdout.flush() - if cov is not None: - cov.stop() - cov.save() - os._exit(0) - - with echo_off(master_fd): - read_until_semaphore(master_fd) - stime = time.time() - os.write(master_fd, u'\x1b['.encode('ascii')) - key1_name, key2, duration_ms = read_until_eof(master_fd).split() - - pid, status = os.waitpid(pid, 0) - assert key1_name == u'KEY_ESCAPE' - assert key2 == u'[' - assert os.WEXITSTATUS(status) == 0 - assert math.floor(time.time() - stime) == 0.0 - assert 34 <= int(duration_ms) <= 45, duration_ms - - -def test_get_location_0s(): - "0-second get_location call without response." - @as_subprocess - def child(): - term = TestTerminal(stream=six.StringIO()) - stime = time.time() - y, x = term.get_location(timeout=0) - assert (math.floor(time.time() - stime) == 0.0) - assert (y, x) == (-1, -1) - child() - - -def test_get_location_0s_under_raw(): - "0-second get_location call without response under raw mode." - @as_subprocess - def child(): - term = TestTerminal(stream=six.StringIO()) - with term.raw(): - stime = time.time() - y, x = term.get_location(timeout=0) - assert (math.floor(time.time() - stime) == 0.0) - assert (y, x) == (-1, -1) - child() - - -def test_get_location_0s_reply_via_ungetch(): - "0-second get_location call with response." - @as_subprocess - def child(): - term = TestTerminal(stream=six.StringIO()) - stime = time.time() - # monkey patch in an invalid response ! - term.ungetch(u'\x1b[10;10R') - - y, x = term.get_location(timeout=0.01) - assert (math.floor(time.time() - stime) == 0.0) - assert (y, x) == (10, 10) - child() - - -def test_get_location_0s_reply_via_ungetch_under_raw(): - "0-second get_location call with response under raw mode." - @as_subprocess - def child(): - term = TestTerminal(stream=six.StringIO()) - with term.raw(): - stime = time.time() - # monkey patch in an invalid response ! - term.ungetch(u'\x1b[10;10R') - - y, x = term.get_location(timeout=0.01) - assert (math.floor(time.time() - stime) == 0.0) - assert (y, x) == (10, 10) - child() +#def test_kbhit_no_kb(): +# "kbhit() always immediately returns False without a keyboard." +# @as_subprocess +# def child(): +# term = TestTerminal(stream=six.StringIO()) +# stime = time.time() +# assert term._keyboard_fd is None +# assert not term.kbhit(timeout=1.1) +# assert math.floor(time.time() - stime) == 1.0 +# child() +# +# +#def test_keystroke_0s_cbreak_noinput(): +# "0-second keystroke without input; '' should be returned." +# @as_subprocess +# def child(): +# term = TestTerminal() +# with term.cbreak(): +# stime = time.time() +# inp = term.inkey(timeout=0) +# assert (inp == u'') +# assert (math.floor(time.time() - stime) == 0.0) +# child() +# +# +#def test_keystroke_0s_cbreak_noinput_nokb(): +# "0-second keystroke without data in input stream and no keyboard/tty." +# @as_subprocess +# def child(): +# term = TestTerminal(stream=six.StringIO()) +# with term.cbreak(): +# stime = time.time() +# inp = term.inkey(timeout=0) +# assert (inp == u'') +# assert (math.floor(time.time() - stime) == 0.0) +# child() +# +# +#@pytest.mark.skipif(os.environ.get('TEST_QUICK', None) is not None, +# reason="TEST_QUICK specified") +#def test_keystroke_1s_cbreak_noinput(): +# "1-second keystroke without input; '' should be returned after ~1 second." +# @as_subprocess +# def child(): +# term = TestTerminal() +# with term.cbreak(): +# stime = time.time() +# inp = term.inkey(timeout=1) +# assert (inp == u'') +# assert (math.floor(time.time() - stime) == 1.0) +# child() +# +# +#@pytest.mark.skipif(os.environ.get('TEST_QUICK', None) is not None, +# reason="TEST_QUICK specified") +#def test_keystroke_1s_cbreak_noinput_nokb(): +# "1-second keystroke without input or keyboard." +# @as_subprocess +# def child(): +# term = TestTerminal(stream=six.StringIO()) +# with term.cbreak(): +# stime = time.time() +# inp = term.inkey(timeout=1) +# assert (inp == u'') +# assert (math.floor(time.time() - stime) == 1.0) +# child() +# +# +#def test_keystroke_0s_cbreak_with_input(): +# "0-second keystroke with input; Keypress should be immediately returned." +# pid, master_fd = pty.fork() +# if pid == 0: +# cov = init_subproc_coverage('test_keystroke_0s_cbreak_with_input') +# # child pauses, writes semaphore and begins awaiting input +# term = TestTerminal() +# read_until_semaphore(sys.__stdin__.fileno(), semaphore=SEMAPHORE) +# os.write(sys.__stdout__.fileno(), SEMAPHORE) +# with term.cbreak(): +# inp = term.inkey(timeout=0) +# os.write(sys.__stdout__.fileno(), inp.encode('utf-8')) +# if cov is not None: +# cov.stop() +# cov.save() +# os._exit(0) +# +# with echo_off(master_fd): +# os.write(master_fd, SEND_SEMAPHORE) +# os.write(master_fd, u'x'.encode('ascii')) +# read_until_semaphore(master_fd) +# stime = time.time() +# output = read_until_eof(master_fd) +# +# pid, status = os.waitpid(pid, 0) +# assert output == u'x' +# assert os.WEXITSTATUS(status) == 0 +# assert math.floor(time.time() - stime) == 0.0 +# +# +#def test_keystroke_cbreak_with_input_slowly(): +# "0-second keystroke with input; Keypress should be immediately returned." +# pid, master_fd = pty.fork() +# if pid == 0: +# cov = init_subproc_coverage('test_keystroke_cbreak_with_input_slowly') +# # child pauses, writes semaphore and begins awaiting input +# term = TestTerminal() +# read_until_semaphore(sys.__stdin__.fileno(), semaphore=SEMAPHORE) +# os.write(sys.__stdout__.fileno(), SEMAPHORE) +# with term.cbreak(): +# while True: +# inp = term.inkey(timeout=0.5) +# os.write(sys.__stdout__.fileno(), inp.encode('utf-8')) +# if inp == 'X': +# break +# if cov is not None: +# cov.stop() +# cov.save() +# os._exit(0) +# +# with echo_off(master_fd): +# os.write(master_fd, SEND_SEMAPHORE) +# os.write(master_fd, u'a'.encode('ascii')) +# time.sleep(0.1) +# os.write(master_fd, u'b'.encode('ascii')) +# time.sleep(0.1) +# os.write(master_fd, u'cdefgh'.encode('ascii')) +# time.sleep(0.1) +# os.write(master_fd, u'X'.encode('ascii')) +# read_until_semaphore(master_fd) +# stime = time.time() +# output = read_until_eof(master_fd) +# +# pid, status = os.waitpid(pid, 0) +# assert output == u'abcdefghX' +# assert os.WEXITSTATUS(status) == 0 +# assert math.floor(time.time() - stime) == 0.0 +# +# +#def test_keystroke_0s_cbreak_multibyte_utf8(): +# "0-second keystroke with multibyte utf-8 input; should decode immediately." +# # utf-8 bytes represent "latin capital letter upsilon". +# pid, master_fd = pty.fork() +# if pid == 0: # child +# cov = init_subproc_coverage('test_keystroke_0s_cbreak_multibyte_utf8') +# term = TestTerminal() +# read_until_semaphore(sys.__stdin__.fileno(), semaphore=SEMAPHORE) +# os.write(sys.__stdout__.fileno(), SEMAPHORE) +# with term.cbreak(): +# inp = term.inkey(timeout=0) +# os.write(sys.__stdout__.fileno(), inp.encode('utf-8')) +# if cov is not None: +# cov.stop() +# cov.save() +# os._exit(0) +# +# with echo_off(master_fd): +# os.write(master_fd, SEND_SEMAPHORE) +# os.write(master_fd, u'\u01b1'.encode('utf-8')) +# read_until_semaphore(master_fd) +# stime = time.time() +# output = read_until_eof(master_fd) +# pid, status = os.waitpid(pid, 0) +# assert output == u'Ʊ' +# assert os.WEXITSTATUS(status) == 0 +# assert math.floor(time.time() - stime) == 0.0 +# +# +#@pytest.mark.skipif(os.environ.get('TRAVIS', None) is not None, +# reason="travis-ci does not handle ^C very well.") +#def test_keystroke_0s_raw_input_ctrl_c(): +# "0-second keystroke with raw allows receiving ^C." +# pid, master_fd = pty.fork() +# if pid == 0: # child +# cov = init_subproc_coverage('test_keystroke_0s_raw_input_ctrl_c') +# term = TestTerminal() +# read_until_semaphore(sys.__stdin__.fileno(), semaphore=SEMAPHORE) +# with term.raw(): +# os.write(sys.__stdout__.fileno(), RECV_SEMAPHORE) +# inp = term.inkey(timeout=0) +# os.write(sys.__stdout__.fileno(), inp.encode('latin1')) +# if cov is not None: +# cov.stop() +# cov.save() +# os._exit(0) +# +# with echo_off(master_fd): +# os.write(master_fd, SEND_SEMAPHORE) +# # ensure child is in raw mode before sending ^C, +# read_until_semaphore(master_fd) +# os.write(master_fd, u'\x03'.encode('latin1')) +# stime = time.time() +# output = read_until_eof(master_fd) +# pid, status = os.waitpid(pid, 0) +# assert (output == u'\x03' or +# output == u'' and not os.isatty(0)) +# assert os.WEXITSTATUS(status) == 0 +# assert math.floor(time.time() - stime) == 0.0 +# +# +#def test_keystroke_0s_cbreak_sequence(): +# "0-second keystroke with multibyte sequence; should decode immediately." +# pid, master_fd = pty.fork() +# if pid == 0: # child +# cov = init_subproc_coverage('test_keystroke_0s_cbreak_sequence') +# term = TestTerminal() +# os.write(sys.__stdout__.fileno(), SEMAPHORE) +# with term.cbreak(): +# inp = term.inkey(timeout=0) +# os.write(sys.__stdout__.fileno(), inp.name.encode('ascii')) +# sys.stdout.flush() +# if cov is not None: +# cov.stop() +# cov.save() +# os._exit(0) +# +# with echo_off(master_fd): +# os.write(master_fd, u'\x1b[D'.encode('ascii')) +# read_until_semaphore(master_fd) +# stime = time.time() +# output = read_until_eof(master_fd) +# pid, status = os.waitpid(pid, 0) +# assert output == u'KEY_LEFT' +# assert os.WEXITSTATUS(status) == 0 +# assert math.floor(time.time() - stime) == 0.0 +# +# +#@pytest.mark.skipif(os.environ.get('TEST_QUICK', None) is not None, +# reason="TEST_QUICK specified") +#def test_keystroke_1s_cbreak_with_input(): +# "1-second keystroke w/multibyte sequence; should return after ~1 second." +# pid, master_fd = pty.fork() +# if pid == 0: # child +# cov = init_subproc_coverage('test_keystroke_1s_cbreak_with_input') +# term = TestTerminal() +# os.write(sys.__stdout__.fileno(), SEMAPHORE) +# with term.cbreak(): +# inp = term.inkey(timeout=3) +# os.write(sys.__stdout__.fileno(), inp.name.encode('utf-8')) +# sys.stdout.flush() +# if cov is not None: +# cov.stop() +# cov.save() +# os._exit(0) +# +# with echo_off(master_fd): +# read_until_semaphore(master_fd) +# stime = time.time() +# time.sleep(1) +# os.write(master_fd, u'\x1b[C'.encode('ascii')) +# output = read_until_eof(master_fd) +# +# pid, status = os.waitpid(pid, 0) +# assert output == u'KEY_RIGHT' +# assert os.WEXITSTATUS(status) == 0 +# assert math.floor(time.time() - stime) == 1.0 +# +# +#@pytest.mark.skipif(os.environ.get('TEST_QUICK', None) is not None, +# reason="TEST_QUICK specified") +#def test_esc_delay_cbreak_035(): +# "esc_delay will cause a single ESC (\\x1b) to delay for 0.35." +# pid, master_fd = pty.fork() +# if pid == 0: # child +# cov = init_subproc_coverage('test_esc_delay_cbreak_035') +# term = TestTerminal() +# os.write(sys.__stdout__.fileno(), SEMAPHORE) +# with term.cbreak(): +# stime = time.time() +# inp = term.inkey(timeout=5) +# measured_time = (time.time() - stime) * 100 +# os.write(sys.__stdout__.fileno(), ( +# '%s %i' % (inp.name, measured_time,)).encode('ascii')) +# sys.stdout.flush() +# if cov is not None: +# cov.stop() +# cov.save() +# os._exit(0) +# +# with echo_off(master_fd): +# read_until_semaphore(master_fd) +# stime = time.time() +# os.write(master_fd, u'\x1b'.encode('ascii')) +# key_name, duration_ms = read_until_eof(master_fd).split() +# +# pid, status = os.waitpid(pid, 0) +# assert key_name == u'KEY_ESCAPE' +# assert os.WEXITSTATUS(status) == 0 +# assert math.floor(time.time() - stime) == 0.0 +# assert 34 <= int(duration_ms) <= 45, duration_ms +# +# +#@pytest.mark.skipif(os.environ.get('TEST_QUICK', None) is not None, +# reason="TEST_QUICK specified") +#def test_esc_delay_cbreak_135(): +# "esc_delay=1.35 will cause a single ESC (\\x1b) to delay for 1.35." +# pid, master_fd = pty.fork() +# if pid == 0: # child +# cov = init_subproc_coverage('test_esc_delay_cbreak_135') +# term = TestTerminal() +# os.write(sys.__stdout__.fileno(), SEMAPHORE) +# with term.cbreak(): +# stime = time.time() +# inp = term.inkey(timeout=5, esc_delay=1.35) +# measured_time = (time.time() - stime) * 100 +# os.write(sys.__stdout__.fileno(), ( +# '%s %i' % (inp.name, measured_time,)).encode('ascii')) +# sys.stdout.flush() +# if cov is not None: +# cov.stop() +# cov.save() +# os._exit(0) +# +# with echo_off(master_fd): +# read_until_semaphore(master_fd) +# stime = time.time() +# os.write(master_fd, u'\x1b'.encode('ascii')) +# key_name, duration_ms = read_until_eof(master_fd).split() +# +# pid, status = os.waitpid(pid, 0) +# assert key_name == u'KEY_ESCAPE' +# assert os.WEXITSTATUS(status) == 0 +# assert math.floor(time.time() - stime) == 1.0 +# assert 134 <= int(duration_ms) <= 145, int(duration_ms) +# +# +#def test_esc_delay_cbreak_timout_0(): +# """esc_delay still in effect with timeout of 0 ("nonblocking").""" +# pid, master_fd = pty.fork() +# if pid == 0: # child +# cov = init_subproc_coverage('test_esc_delay_cbreak_timout_0') +# term = TestTerminal() +# os.write(sys.__stdout__.fileno(), SEMAPHORE) +# with term.cbreak(): +# stime = time.time() +# inp = term.inkey(timeout=0) +# measured_time = (time.time() - stime) * 100 +# os.write(sys.__stdout__.fileno(), ( +# '%s %i' % (inp.name, measured_time,)).encode('ascii')) +# sys.stdout.flush() +# if cov is not None: +# cov.stop() +# cov.save() +# os._exit(0) +# +# with echo_off(master_fd): +# os.write(master_fd, u'\x1b'.encode('ascii')) +# read_until_semaphore(master_fd) +# stime = time.time() +# key_name, duration_ms = read_until_eof(master_fd).split() +# +# pid, status = os.waitpid(pid, 0) +# assert key_name == u'KEY_ESCAPE' +# assert os.WEXITSTATUS(status) == 0 +# assert math.floor(time.time() - stime) == 0.0 +# assert 34 <= int(duration_ms) <= 45, int(duration_ms) +# +# +#def test_esc_delay_cbreak_nonprefix_sequence(): +# "ESC a (\\x1ba) will return an ESC immediately" +# pid, master_fd = pty.fork() +# if pid is 0: # child +# cov = init_subproc_coverage('test_esc_delay_cbreak_nonprefix_sequence') +# term = TestTerminal() +# os.write(sys.__stdout__.fileno(), SEMAPHORE) +# with term.cbreak(): +# stime = time.time() +# esc = term.inkey(timeout=5) +# inp = term.inkey(timeout=5) +# measured_time = (time.time() - stime) * 100 +# os.write(sys.__stdout__.fileno(), ( +# '%s %s %i' % (esc.name, inp, measured_time,)).encode('ascii')) +# sys.stdout.flush() +# if cov is not None: +# cov.stop() +# cov.save() +# os._exit(0) +# +# with echo_off(master_fd): +# read_until_semaphore(master_fd) +# stime = time.time() +# os.write(master_fd, u'\x1ba'.encode('ascii')) +# key1_name, key2, duration_ms = read_until_eof(master_fd).split() +# +# pid, status = os.waitpid(pid, 0) +# assert key1_name == u'KEY_ESCAPE' +# assert key2 == u'a' +# assert os.WEXITSTATUS(status) == 0 +# assert math.floor(time.time() - stime) == 0.0 +# assert -1 <= int(duration_ms) <= 15, duration_ms +# +# +#def test_esc_delay_cbreak_prefix_sequence(): +# "An unfinished multibyte sequence (\\x1b[) will delay an ESC by .35 " +# pid, master_fd = pty.fork() +# if pid is 0: # child +# cov = init_subproc_coverage('test_esc_delay_cbreak_prefix_sequence') +# term = TestTerminal() +# os.write(sys.__stdout__.fileno(), SEMAPHORE) +# with term.cbreak(): +# stime = time.time() +# esc = term.inkey(timeout=5) +# inp = term.inkey(timeout=5) +# measured_time = (time.time() - stime) * 100 +# os.write(sys.__stdout__.fileno(), ( +# '%s %s %i' % (esc.name, inp, measured_time,)).encode('ascii')) +# sys.stdout.flush() +# if cov is not None: +# cov.stop() +# cov.save() +# os._exit(0) +# +# with echo_off(master_fd): +# read_until_semaphore(master_fd) +# stime = time.time() +# os.write(master_fd, u'\x1b['.encode('ascii')) +# key1_name, key2, duration_ms = read_until_eof(master_fd).split() +# +# pid, status = os.waitpid(pid, 0) +# assert key1_name == u'KEY_ESCAPE' +# assert key2 == u'[' +# assert os.WEXITSTATUS(status) == 0 +# assert math.floor(time.time() - stime) == 0.0 +# assert 34 <= int(duration_ms) <= 45, duration_ms +# +# +#def test_get_location_0s(): +# "0-second get_location call without response." +# @as_subprocess +# def child(): +# term = TestTerminal(stream=six.StringIO()) +# stime = time.time() +# y, x = term.get_location(timeout=0) +# assert (math.floor(time.time() - stime) == 0.0) +# assert (y, x) == (-1, -1) +# child() +# +# +#def test_get_location_0s_under_raw(): +# "0-second get_location call without response under raw mode." +# @as_subprocess +# def child(): +# term = TestTerminal(stream=six.StringIO()) +# with term.raw(): +# stime = time.time() +# y, x = term.get_location(timeout=0) +# assert (math.floor(time.time() - stime) == 0.0) +# assert (y, x) == (-1, -1) +# child() +# +# +#def test_get_location_0s_reply_via_ungetch(): +# "0-second get_location call with response." +# @as_subprocess +# def child(): +# term = TestTerminal(stream=six.StringIO()) +# stime = time.time() +# # monkey patch in an invalid response ! +# term.ungetch(u'\x1b[10;10R') +# +# y, x = term.get_location(timeout=0.01) +# assert (math.floor(time.time() - stime) == 0.0) +# assert (y, x) == (10, 10) +# child() +# +# +#def test_get_location_0s_reply_via_ungetch_under_raw(): +# "0-second get_location call with response under raw mode." +# @as_subprocess +# def child(): +# term = TestTerminal(stream=six.StringIO()) +# with term.raw(): +# stime = time.time() +# # monkey patch in an invalid response ! +# term.ungetch(u'\x1b[10;10R') +# +# y, x = term.get_location(timeout=0.01) +# assert (math.floor(time.time() - stime) == 0.0) +# assert (y, x) == (10, 10) +# child() def test_keystroke_default_args(): From a744a42f5a0c5b06da26f2b2534a82e8759692fb Mon Sep 17 00:00:00 2001 From: Jeff Quast Date: Tue, 19 Jun 2018 22:20:40 -0700 Subject: [PATCH 7/8] disabling useful tests make non-contributors happy --- blessed/__init__.py | 2 +- docs/history.rst | 7 +++++++ version.json | 2 +- 3 files changed, 9 insertions(+), 2 deletions(-) diff --git a/blessed/__init__.py b/blessed/__init__.py index cf5fbb9b..1775b353 100644 --- a/blessed/__init__.py +++ b/blessed/__init__.py @@ -16,4 +16,4 @@ 'support due to http://bugs.python.org/issue10570.') __all__ = ('Terminal',) -__version__ = '1.14.2' +__version__ = '1.15.0' diff --git a/docs/history.rst b/docs/history.rst index e593377d..402c70d6 100644 --- a/docs/history.rst +++ b/docs/history.rst @@ -1,5 +1,12 @@ Version History =============== +1.15 + * disable timing integration tests for keyboard routines. + + They work perfectly fine for regression testing for contributing + developers, but people run our tests on build farms and open issues when + they fail. So we comment out these useful tests. :ghissue:`29`. + 1.14 * bugfix: :meth:`~.Terminal.wrap` misbehaved for text containing newlines, :ghissue:`74`. diff --git a/version.json b/version.json index 00a1292b..25d66792 100644 --- a/version.json +++ b/version.json @@ -1 +1 @@ -{"version": "1.14.2"} \ No newline at end of file +{"version": "1.15.0"} From da48e3149aee53ecf6a54856982b00360773eaa5 Mon Sep 17 00:00:00 2001 From: Jeff Quast Date: Tue, 19 Jun 2018 23:27:07 -0700 Subject: [PATCH 8/8] small docfix --- blessed/keyboard.py | 1 + docs/history.rst | 4 +++- 2 files changed, 4 insertions(+), 1 deletion(-) diff --git a/blessed/keyboard.py b/blessed/keyboard.py index bbf43265..4a4eab81 100644 --- a/blessed/keyboard.py +++ b/blessed/keyboard.py @@ -340,6 +340,7 @@ def _inject_curses_keynames(): _lastval += 1 setattr(curses, 'KEY_{0}'.format(key), _lastval) + _inject_curses_keynames() #: In a perfect world, terminal emulators would always send exactly what diff --git a/docs/history.rst b/docs/history.rst index 402c70d6..be7608c9 100644 --- a/docs/history.rst +++ b/docs/history.rst @@ -5,7 +5,9 @@ Version History They work perfectly fine for regression testing for contributing developers, but people run our tests on build farms and open issues when - they fail. So we comment out these useful tests. :ghissue:`29`. + they fail. So we comment out these useful tests. :ghissue:`100`. + + * Support python 3.7. :ghissue:`102`. 1.14 * bugfix: :meth:`~.Terminal.wrap` misbehaved for text containing newlines,