summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorAlexey Stepanov <penguinolog@users.noreply.github.com>2023-04-25 19:20:33 +0200
committerGitHub <noreply@github.com>2023-04-25 19:20:33 +0200
commit1cc1d63358f7759785b39079eca069ed41078b6d (patch)
tree80261451a20ea64dac2185060908668a97b15f36
parent6bf74be23f3179d87d4f40d45d2a4e8718ab26f5 (diff)
downloadurwid-1cc1d63358f7759785b39079eca069ed41078b6d.tar.gz
Fix incorrect type cast in vterm (`apply_mapping` should return `bytes`) (#545)
* Fix incorrect type cast in vterm (`apply_mapping` should return `bytes`) Add `time.sleep(0.1)` to the event loop tests: in the worst scenario on windows and slow machine function in parallel thread/async can wait up to 80 milliseconds (tested) Add type annotations to the `vterm` and `test_vterm` to simplify error lookup. * Fix `DeprecationWarning` in doctests & examples * Add `pytest` configuration in `pyproject.toml` without migration * `Signals.emit()` rework: stop `user_args` join with `weak_args` Partial: #544 Partial: #512 Partial: #406 * drop `sleep`: not enough effective with pytest * set timer for errors raise to 0: faster raise, faster test done --------- Co-authored-by: Aleksei Stepanov <alekseis@nvidia.com>
-rwxr-xr-xexamples/calc.py5
-rw-r--r--pyproject.toml6
-rw-r--r--urwid/signals.py97
-rw-r--r--urwid/tests/test_event_loops.py73
-rw-r--r--urwid/tests/test_vterm.py59
-rw-r--r--urwid/vterm.py42
-rw-r--r--urwid/widget.py3
-rwxr-xr-xurwid/wimp.py6
8 files changed, 181 insertions, 110 deletions
diff --git a/examples/calc.py b/examples/calc.py
index b83cc07..c8e3eeb 100755
--- a/examples/calc.py
+++ b/examples/calc.py
@@ -491,9 +491,10 @@ class CellColumn( urwid.WidgetWrap ):
return self.content[-1].get_result()
+class HelpColumn(urwid.Widget):
+ _selectable = True
+ _sizing = frozenset([urwid.BOX])
-
-class HelpColumn(urwid.BoxWidget):
help_text = [
('title', "Column Calculator"),
"",
diff --git a/pyproject.toml b/pyproject.toml
index ca21353..a417089 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -49,6 +49,12 @@ universal = 0
profile = "black"
line_length = 120
+[tool.pytest.ini_options]
+minversion = "6.0"
+addopts = "-vvvv --doctest-modules -s --cov=urwid"
+doctest_optionflags = ["ELLIPSIS", "IGNORE_EXCEPTION_DETAIL"]
+testpaths = ["urwid"]
+
[tool.coverage.run]
omit = ["urwid/tests/*", ".tox/*", "setup.py"]
branch = true
diff --git a/urwid/signals.py b/urwid/signals.py
index dfbf47f..d118d9f 100644
--- a/urwid/signals.py
+++ b/urwid/signals.py
@@ -23,8 +23,10 @@
from __future__ import annotations
import itertools
+import typing
import warnings
import weakref
+from collections.abc import Callable, Collection, Container, Iterable
class MetaSignals(type):
@@ -32,15 +34,16 @@ class MetaSignals(type):
register the list of signals in the class variable signals,
including signals in superclasses.
"""
- def __init__(cls, name: str, bases, d):
+ def __init__(cls, name: str, bases: tuple[type, ...], d: dict[str, typing.Any]) -> None:
signals = d.get("signals", [])
for superclass in cls.__bases__:
signals.extend(getattr(superclass, 'signals', []))
- signals = list({x:None for x in signals}.keys())
+ signals = list({x: None for x in signals}.keys())
d["signals"] = signals
register_signal(cls, signals)
super().__init__(name, bases, d)
+
def setdefaultattr(obj, name, value):
# like dict.setdefault() for object attributes
if hasattr(obj, name):
@@ -63,7 +66,7 @@ class Signals:
def __init__(self):
self._supported = {}
- def register(self, sig_cls, signals):
+ def register(self, sig_cls, signals: Container[str]) -> None:
"""
:param sig_class: the class of an object that will be sending signals
:type sig_class: class
@@ -76,7 +79,16 @@ class Signals:
"""
self._supported[sig_cls] = signals
- def connect(self, obj, name: str, callback, user_arg=None, weak_args=None, user_args=None) -> Key:
+ def connect(
+ self,
+ obj,
+ name: str,
+ callback: Callable[..., typing.Any],
+ user_arg: typing.Any = None,
+ *,
+ weak_args: Iterable[typing.Any] = (),
+ user_args: Iterable[typing.Any] = (),
+ ) -> Key:
"""
:param obj: the object sending a signal
:type obj: object
@@ -159,7 +171,7 @@ class Signals:
)
sig_cls = obj.__class__
- if not name in self._supported.get(sig_cls, []):
+ if name not in self._supported.get(sig_cls, ()):
raise NameError(f"No such signal {name!r} for object {obj!r}")
# Just generate an arbitrary (but unique) key
@@ -189,11 +201,27 @@ class Signals:
return key
- def _prepare_user_args(self, weak_args, user_args, callback = None):
+ def _prepare_user_args(
+ self,
+ weak_args: Iterable[typing.Any] = (),
+ user_args: Iterable[typing.Any] = (),
+ callback: Callable[..., typing.Any] | None = None,
+ ) -> tuple[Collection[weakref.ReferenceType], Collection[typing.Any]]:
# Turn weak_args into weakrefs and prepend them to user_args
- return [weakref.ref(a, callback) for a in (weak_args or [])] + (user_args or [])
-
- def disconnect(self, obj, name: str, callback, user_arg=None, weak_args=None, user_args=None):
+ w_args = tuple(weakref.ref(w_arg, callback) for w_arg in weak_args)
+ args = tuple(user_args) or ()
+ return (w_args, args)
+
+ def disconnect(
+ self,
+ obj,
+ name: str,
+ callback: Callable[..., typing.Any],
+ user_arg: typing.Any = None,
+ *,
+ weak_args: Iterable[typing.Any] = (),
+ user_args: Iterable[typing.Any] = (),
+ ) -> None:
"""
:param obj: the object to disconnect the signal from
:type obj: object
@@ -227,7 +255,7 @@ class Signals:
if h[1:] == (callback, user_arg, user_args):
return self.disconnect_by_key(obj, name, h[0])
- def disconnect_by_key(self, obj, name: str, key: Key):
+ def disconnect_by_key(self, obj, name: str, key: Key) -> None:
"""
:param obj: the object to disconnect the signal from
:type obj: object
@@ -248,7 +276,7 @@ class Signals:
handlers = signals.get(name, [])
handlers[:] = [h for h in handlers if h[0] is not key]
- def emit(self, obj, name: str, *args):
+ def emit(self, obj, name: str, *args) -> bool:
"""
:param obj: the object sending a signal
:type obj: object
@@ -264,38 +292,33 @@ class Signals:
result = False
signals = getattr(obj, self._signal_attr, {})
handlers = signals.get(name, [])
- for key, callback, user_arg, user_args in handlers:
- result |= self._call_callback(callback, user_arg, user_args, args)
+ for key, callback, user_arg, (weak_args, user_args) in handlers:
+ result |= self._call_callback(callback, user_arg, weak_args, user_args, args)
return result
- def _call_callback(self, callback, user_arg, user_args, emit_args):
- if user_args:
- args_to_pass = []
- for arg in user_args:
- if isinstance(arg, weakref.ReferenceType):
- arg = arg()
- if arg is None:
- # If the weakref is None, the referenced object
- # was cleaned up. We just skip the entire
- # callback in this case. The weakref cleanup
- # handler will have removed the callback when
- # this happens, so no need to actually remove
- # the callback here.
- return False
- args_to_pass.append(arg)
-
- args_to_pass.extend(emit_args)
- else:
- # Optimization: Don't create a new list when there are
- # no user_args
- args_to_pass = emit_args
+ def _call_callback(
+ self,
+ callback,
+ user_arg: typing.Any,
+ weak_args: Collection[weakref.ReferenceType],
+ user_args: Collection[typing.Any],
+ emit_args: Iterable[typing.Any],
+ ) -> bool:
+ args_to_pass = []
+ for w_arg in weak_args:
+ real_arg = w_arg()
+ if real_arg is not None:
+ args_to_pass.append(real_arg)
+ else:
+ # de-referenced
+ return False
# The deprecated user_arg argument was added to the end
# instead of the beginning.
- if user_arg is not None:
- args_to_pass = itertools.chain(args_to_pass, (user_arg,))
+ args = itertools.chain(args_to_pass, user_args, emit_args, (user_arg,) if user_arg is not None else ())
+
+ return bool(callback(*args))
- return bool(callback(*args_to_pass))
_signals = Signals()
emit_signal = _signals.emit
diff --git a/urwid/tests/test_event_loops.py b/urwid/tests/test_event_loops.py
index b8ce257..a15bc72 100644
--- a/urwid/tests/test_event_loops.py
+++ b/urwid/tests/test_event_loops.py
@@ -3,6 +3,7 @@ from __future__ import annotations
import asyncio
import os
import sys
+import typing
import unittest
import urwid
@@ -11,38 +12,47 @@ import urwid
class EventLoopTestMixin:
def test_event_loop(self):
rd, wr = os.pipe()
- evl = self.evl
+ evl: urwid.EventLoop = self.evl
out = []
+
def step1():
out.append("writing")
os.write(wr, b"hi")
+
def step2():
out.append(os.read(rd, 2).decode('ascii'))
raise urwid.ExitMainLoop
- handle = evl.alarm(0, step1)
- handle = evl.watch_file(rd, step2)
+
+ _handle = evl.alarm(0, step1)
+ _handle = evl.watch_file(rd, step2)
+
evl.run()
self.assertEqual(out, ["writing", "hi"])
def test_remove_alarm(self):
- evl = self.evl
+ evl: urwid.EventLoop = self.evl
handle = evl.alarm(50, lambda: None)
+
def step1():
self.assertTrue(evl.remove_alarm(handle))
self.assertFalse(evl.remove_alarm(handle))
raise urwid.ExitMainLoop
+
evl.alarm(0, step1)
evl.run()
def test_remove_watch_file(self):
- evl = self.evl
+ evl: urwid.EventLoop = self.evl
fd_r, fd_w = os.pipe()
+
try:
handle = evl.watch_file(fd_r, lambda: None)
+
def step1():
self.assertTrue(evl.remove_watch_file(handle))
self.assertFalse(evl.remove_watch_file(handle))
raise urwid.ExitMainLoop
+
evl.alarm(0, step1)
evl.run()
finally:
@@ -52,36 +62,43 @@ class EventLoopTestMixin:
_expected_idle_handle = 1
def test_run(self):
- evl = self.evl
+ evl: urwid.EventLoop = self.evl
out = []
rd, wr = os.pipe()
self.assertEqual(os.write(wr, b"data"), 4)
+
def say_hello():
out.append("hello")
+
def say_waiting():
out.append("waiting")
- def exit_clean():
+
+ def exit_clean() -> typing.NoReturn:
out.append("clean exit")
raise urwid.ExitMainLoop
- def exit_error():
+
+ def exit_error() -> typing.NoReturn:
1/0
- handle = evl.alarm(0.01, exit_clean)
- handle = evl.alarm(0.005, say_hello)
+
+ _handle = evl.alarm(0.01, exit_clean)
+ _handle = evl.alarm(0.005, say_hello)
idle_handle = evl.enter_idle(say_waiting)
if self._expected_idle_handle is not None:
self.assertEqual(idle_handle, 1)
+
evl.run()
self.assertTrue("waiting" in out, out)
self.assertTrue("hello" in out, out)
self.assertTrue("clean exit" in out, out)
handle = evl.watch_file(rd, exit_clean)
del out[:]
+
evl.run()
self.assertEqual(["clean exit"], out)
self.assertTrue(evl.remove_watch_file(handle))
- handle = evl.alarm(0, exit_error)
+ _handle = evl.alarm(0, exit_error)
self.assertRaises(ZeroDivisionError, evl.run)
- handle = evl.watch_file(rd, exit_error)
+ _handle = evl.watch_file(rd, exit_error)
self.assertRaises(ZeroDivisionError, evl.run)
@@ -101,7 +118,7 @@ else:
def test_error(self):
evl = self.evl
- evl.alarm(0.5, lambda: 1 / 0) # Simulate error in event loop
+ evl.alarm(0, lambda: 1 / 0) # Simulate error in event loop
self.assertRaises(ZeroDivisionError, evl.run)
@@ -142,17 +159,22 @@ else:
out = []
rd, wr = os.pipe()
self.assertEqual(os.write(wr, b"data"), 4)
+
def step2():
out.append(os.read(rd, 2).decode('ascii'))
+
def say_hello():
out.append("hello")
+
def say_waiting():
out.append("waiting")
+
def test_remove_alarm():
handle = evl.alarm(50, lambda: None)
self.assertTrue(evl.remove_alarm(handle))
self.assertFalse(evl.remove_alarm(handle))
out.append("remove_alarm ok")
+
def test_remove_watch_file():
fd_r, fd_w = os.pipe()
try:
@@ -163,16 +185,19 @@ else:
os.close(fd_r)
os.close(fd_w)
out.append("remove_watch_file ok")
- def exit_clean():
+
+ def exit_clean() -> typing.NoReturn:
out.append("clean exit")
raise urwid.ExitMainLoop
- def exit_error():
+
+ def exit_error() -> typing.NoReturn:
1/0
- handle = evl.watch_file(rd, step2)
- handle = evl.alarm(0.1, exit_clean)
- handle = evl.alarm(0.05, say_hello)
- handle = evl.alarm(0.06, test_remove_alarm)
- handle = evl.alarm(0.07, test_remove_watch_file)
+
+ _handle = evl.watch_file(rd, step2)
+ _handle = evl.alarm(0.1, exit_clean)
+ _handle = evl.alarm(0.05, say_hello)
+ _handle = evl.alarm(0.06, test_remove_alarm)
+ _handle = evl.alarm(0.07, test_remove_watch_file)
self.assertEqual(evl.enter_idle(say_waiting), 1)
evl.run()
self.assertTrue("da" in out, out)
@@ -183,7 +208,7 @@ else:
def test_error(self):
evl = self.evl
- evl.alarm(0.5, lambda: 1 / 0) # Simulate error in event loop
+ evl.alarm(0, lambda: 1 / 0) # Simulate error in event loop
self.assertRaises(ZeroDivisionError, evl.run)
@@ -195,7 +220,7 @@ class AsyncioEventLoopTest(unittest.TestCase, EventLoopTestMixin):
def test_error(self):
evl = self.evl
- evl.alarm(0.5, lambda: 1 / 0) # Simulate error in event loop
+ evl.alarm(0, lambda: 1 / 0) # Simulate error in event loop
self.assertRaises(ZeroDivisionError, evl.run)
@unittest.skipIf(
@@ -206,7 +231,7 @@ class AsyncioEventLoopTest(unittest.TestCase, EventLoopTestMixin):
evl = self.evl
async def error_coro():
- result = 1 / 0 # Simulate error in coroutine
+ result = 1 / 0 # Simulate error in coroutine
return result
asyncio.ensure_future(error_coro(), loop=asyncio.get_event_loop_policy().get_event_loop())
@@ -226,5 +251,5 @@ else:
def test_error(self):
evl = self.evl
- evl.alarm(0.5, lambda: 1 / 0) # Simulate error in event loop
+ evl.alarm(0, lambda: 1 / 0) # Simulate error in event loop
self.assertRaises(ZeroDivisionError, evl.run)
diff --git a/urwid/tests/test_vterm.py b/urwid/tests/test_vterm.py
index 6ca6fc9..c75136a 100644
--- a/urwid/tests/test_vterm.py
+++ b/urwid/tests/test_vterm.py
@@ -23,10 +23,11 @@ from __future__ import annotations
import errno
import os
import sys
+import typing
import unittest
from itertools import dropwhile
-from urwid import signals, vterm
+from urwid import Widget, signals, vterm
from urwid.decoration import BoxAdapter
from urwid.listbox import ListBox
@@ -34,10 +35,10 @@ from urwid.listbox import ListBox
class DummyCommand:
QUITSTRING = b'|||quit|||'
- def __init__(self):
+ def __init__(self) -> None:
self.reader, self.writer = os.pipe()
- def __call__(self):
+ def __call__(self) -> None:
# reset
stdout = getattr(sys.stdout, 'buffer', sys.stdout)
stdout.write(b'\x1bc')
@@ -49,7 +50,7 @@ class DummyCommand:
stdout.write(data)
stdout.flush()
- def read(self, size):
+ def read(self, size: int) -> bytes:
while True:
try:
return os.read(self.reader, size)
@@ -57,27 +58,27 @@ class DummyCommand:
if e.errno != errno.EINTR:
raise
- def write(self, data):
+ def write(self, data: bytes) -> None:
os.write(self.writer, data)
- def quit(self):
+ def quit(self) -> None:
self.write(self.QUITSTRING)
class TermTest(unittest.TestCase):
- def setUp(self):
+ def setUp(self) -> None:
self.command = DummyCommand()
self.term = vterm.Terminal(self.command)
self.resize(80, 24)
- def tearDown(self):
+ def tearDown(self) -> None:
self.command.quit()
- def connect_signal(self, signal):
+ def connect_signal(self, signal: str):
self._sig_response = None
- def _set_signal_response(widget, *args, **kwargs):
+ def _set_signal_response(widget: Widget, *args, **kwargs) -> None:
self._sig_response = (args, kwargs)
self._set_signal_response = _set_signal_response
@@ -86,39 +87,51 @@ class TermTest(unittest.TestCase):
def expect_signal(self, *args, **kwargs):
self.assertEqual(self._sig_response, (args, kwargs))
- def disconnect_signal(self, signal):
+ def disconnect_signal(self, signal: str) -> None:
signals.disconnect_signal(self.term, signal, self._set_signal_response)
def caught_beep(self, obj):
self.beeped = True
- def resize(self, width, height, soft=False):
+ def resize(self, width: int, height: int, soft: bool = False) -> None:
self.termsize = (width, height)
if not soft:
self.term.render(self.termsize, focus=False)
- def write(self, data):
+ def write(self, data: str) -> None:
data = data.encode('iso8859-1')
self.command.write(data.replace(br'\e', b'\x1b'))
- def flush(self):
+ def flush(self) -> None:
self.write(chr(0x7f))
- def read(self, raw=False, focus=False):
+ @typing.overload
+ def read(self, raw: bool = False, focus: bool = ...) -> bytes:
+ ...
+
+ @typing.overload
+ def read(self, raw: bool = True, focus: bool = ...) -> list[list[bytes, typing.Any, typing.Any]]:
+ ...
+
+ def read(self, raw: bool = False, focus: bool = False) -> bytes | list[list[bytes, typing.Any, typing.Any]]:
self.term.wait_and_feed()
rendered = self.term.render(self.termsize, focus=focus)
if raw:
is_empty = lambda c: c == (None, None, b' ')
content = list(rendered.content())
- lines = [list(dropwhile(is_empty, reversed(line)))
- for line in content]
- return [list(reversed(line)) for line in lines if len(line)]
+ lines = (tuple(dropwhile(is_empty, reversed(line))) for line in content)
+ return [list(reversed(line)) for line in lines if line]
else:
content = rendered.text
- lines = [line.rstrip() for line in content]
+ lines = (line.rstrip() for line in content)
return b'\n'.join(lines).rstrip()
- def expect(self, what, desc=None, raw=False, focus=False):
+ def expect(
+ self,
+ what: str | list[tuple[typing.Any, str | None, bytes]],
+ desc: str | None = None, raw: bool = False,
+ focus: bool = False,
+ ) -> None:
if not isinstance(what, list):
what = what.encode('iso8859-1')
got = self.read(raw=raw, focus=focus)
@@ -126,7 +139,7 @@ class TermTest(unittest.TestCase):
desc = ''
else:
desc += '\n'
- desc += 'Expected:\n%r\nGot:\n%r' % (what, got)
+ desc += f'Expected:\n{what!r}\nGot:\n{got!r}'
self.assertEqual(got, what, desc)
def test_simplestring(self):
@@ -226,6 +239,7 @@ class TermTest(unittest.TestCase):
self.expect(' x5a98765')
def test_scrolling_region_simple(self):
+ # TODO(Aleksei): Issue #544
self.write('\\e[10;20r\\e[10f1\n2\n3\n4\n5\n6\n7\n8\n9\n10\n11\n12\\e[faa')
self.expect('aa' + '\n' * 9 + '2\n3\n4\n5\n6\n7\n8\n9\n10\n11\n12')
@@ -242,10 +256,12 @@ class TermTest(unittest.TestCase):
self.expect('\ntest')
def test_cursor_scrolling_region(self):
+ # TODO(Aleksei): Issue #544
self.write('\\e[?6h\\e[10;20r\\e[10f1\n2\n3\n4\n5\n6\n7\n8\n9\n10\n11\n12\\e[faa')
self.expect('\n' * 9 + 'aa\n3\n4\n5\n6\n7\n8\n9\n10\n11\n12')
def test_scrolling_region_simple_with_focus(self):
+ # TODO(Aleksei): Issue #544
self.write('\\e[10;20r\\e[10f1\n2\n3\n4\n5\n6\n7\n8\n9\n10\n11\n12\\e[faa')
self.expect('aa' + '\n' * 9 + '2\n3\n4\n5\n6\n7\n8\n9\n10\n11\n12', focus=True)
@@ -262,6 +278,7 @@ class TermTest(unittest.TestCase):
self.expect('\ntest', focus=True)
def test_cursor_scrolling_region_with_focus(self):
+ # TODO(Aleksei): Issue #544
self.write('\\e[?6h\\e[10;20r\\e[10f1\n2\n3\n4\n5\n6\n7\n8\n9\n10\n11\n12\\e[faa')
self.expect('\n' * 9 + 'aa\n3\n4\n5\n6\n7\n8\n9\n10\n11\n12', focus=True)
diff --git a/urwid/vterm.py b/urwid/vterm.py
index 6922c7f..f3b382f 100644
--- a/urwid/vterm.py
+++ b/urwid/vterm.py
@@ -34,7 +34,7 @@ import sys
import time
import traceback
import typing
-from collections.abc import Iterable, Mapping, Sequence
+from collections.abc import Iterable, Mapping, Sequence, Callable
try:
import fcntl
@@ -202,25 +202,25 @@ class TermCharset:
self.active = g
self.current = self.MAPPING.get(self._g[g], None)
- def set_sgr_ibmpc(self):
+ def set_sgr_ibmpc(self) -> None:
"""
Set graphics rendition mapping to IBM PC CP437.
"""
self._sgr_mapping = True
- def reset_sgr_ibmpc(self):
+ def reset_sgr_ibmpc(self) -> None:
"""
Reset graphics rendition mapping to IBM PC CP437.
"""
self._sgr_mapping = False
self.activate(g=self.active)
- def apply_mapping(self, char):
+ def apply_mapping(self, char: bytes) -> bytes:
if self._sgr_mapping or self._g[self.active] == 'ibmpc':
dec_pos = DEC_SPECIAL_CHARS.find(char.decode('cp437'))
if dec_pos >= 0:
self.current = '0'
- return str(ALT_DEC_SPECIAL_CHARS[dec_pos])
+ return ALT_DEC_SPECIAL_CHARS[dec_pos].encode("cp437")
else:
self.current = 'U'
return char
@@ -272,7 +272,7 @@ class TermCanvas(Canvas):
self.coords["cursor"] = (0, 0, None)
- self.term_cursor = [0, 0] # do not allow to shoot in the leg at `set_term_cursor`
+ self.term_cursor = (0, 0) # do not allow to shoot in the leg at `set_term_cursor`
self.cursor: tuple[int, int] | None = None
self.reset()
@@ -627,8 +627,7 @@ class TermCanvas(Canvas):
byte -- an integer ordinal
"""
- if (self.modes.main_charset == CHARSET_UTF8 or
- util._target_encoding == 'utf8'):
+ if self.modes.main_charset == CHARSET_UTF8 or util._target_encoding == 'utf8':
if byte >= 0xc0:
# start multibyte sequence
self.utf8_eat_bytes = self.get_utf8_len(byte)
@@ -656,7 +655,7 @@ class TermCanvas(Canvas):
self.process_char(char)
- def process_char(self, char: int | bytes):
+ def process_char(self, char: int | bytes) -> None:
"""
Process a single character (single- and multi-byte).
@@ -703,7 +702,7 @@ class TermCanvas(Canvas):
else:
self.push_cursor(char)
- def set_char(self, char, x: int | None = None, y: int | None = None) -> None:
+ def set_char(self, char: bytes, x: int | None = None, y: int | None = None) -> None:
"""
Set character of either the current cursor position
or a position given by 'x' and/or 'y' to 'char'.
@@ -792,16 +791,16 @@ class TermCanvas(Canvas):
relative_y = relative_x = True
if relative_x:
- x = self.term_cursor[0] + x
+ x += self.term_cursor[0]
if relative_y:
- y = self.term_cursor[1] + y
+ y += self.term_cursor[1]
elif self.modes.constrain_scrolling:
y += self.scrollregion_start
self.set_term_cursor(x, y)
- def push_char(self, char, x: int, y: int) -> None:
+ def push_char(self, char: bytes | None, x: int, y: int) -> None:
"""
Push one character to current position and advance cursor to x/y.
"""
@@ -814,7 +813,7 @@ class TermCanvas(Canvas):
self.set_term_cursor(x, y)
- def push_cursor(self, char=None):
+ def push_cursor(self, char: bytes | None = None) -> None:
"""
Move cursor one character forward wrapping lines as needed.
If 'char' is given, put the character into the former position.
@@ -906,7 +905,7 @@ class TermCanvas(Canvas):
for row in range(self.height):
self.term[row] = self.empty_line('E')
- def blank_line(self, row) -> None:
+ def blank_line(self, row: int) -> None:
"""
Blank a single line at the specified row, without modifying other lines.
"""
@@ -1367,7 +1366,7 @@ class Terminal(Widget):
def __init__(
self,
- command: Sequence[str] | None,
+ command: Sequence[str] | Callable[[], ...] | None,
env: Mapping[str, str] | Iterable[Sequence[str]] | None = None,
main_loop: event_loop.EventLoop | None = None,
escape_sequence: str | None = None,
@@ -1376,10 +1375,10 @@ class Terminal(Widget):
"""
A terminal emulator within a widget.
- ``command`` is the command to execute inside the terminal, provided as a
- list of the command followed by its arguments. If 'command' is None,
- the command is the current user's shell. You can also provide a callable
- instead of a command, which will be executed in the subprocess.
+ ``command`` is the command to execute inside the terminal,
+ provided as a list of the command followed by its arguments.
+ If 'command' is None, the command is the current user's shell.
+ You can also provide a callable instead of a command, which will be executed in the subprocess.
``env`` can be used to pass custom environment variables. If omitted,
os.environ is used.
@@ -1661,8 +1660,7 @@ class Terminal(Widget):
self.last_key = key
self._invalidate()
return None
- elif (self.last_key == self.escape_sequence
- and key != self.escape_sequence):
+ elif self.last_key == self.escape_sequence and key != self.escape_sequence:
# hand down keypress directly after ungrab.
self.last_key = key
return key
diff --git a/urwid/widget.py b/urwid/widget.py
index 71ade15..4cb29fc 100644
--- a/urwid/widget.py
+++ b/urwid/widget.py
@@ -801,12 +801,13 @@ class Divider(Widget):
return canv
-class SolidFill(BoxWidget):
+class SolidFill(Widget):
"""
A box widget that fills an area with a single character
"""
_selectable = False
ignore_focus = True
+ _sizing = frozenset([BOX])
def __init__(self, fill_char: str = " ") -> None:
"""
diff --git a/urwid/wimp.py b/urwid/wimp.py
index 7376dce..3a6c8c7 100755
--- a/urwid/wimp.py
+++ b/urwid/wimp.py
@@ -227,17 +227,17 @@ class CheckBox(WidgetWrap):
do_callback -- False to suppress signal from this change
>>> changes = []
- >>> def callback_a(cb, state, user_data):
+ >>> def callback_a(user_data, cb, state):
... changes.append("A %r %r" % (state, user_data))
>>> def callback_b(cb, state):
... changes.append("B %r" % state)
>>> cb = CheckBox('test', False, False)
- >>> key1 = connect_signal(cb, 'change', callback_a, "user_a")
+ >>> key1 = connect_signal(cb, 'change', callback_a, user_args=("user_a",))
>>> key2 = connect_signal(cb, 'change', callback_b)
>>> cb.set_state(True) # both callbacks will be triggered
>>> cb.state
True
- >>> disconnect_signal(cb, 'change', callback_a, "user_a")
+ >>> disconnect_signal(cb, 'change', callback_a, user_args=("user_a",))
>>> cb.state = False
>>> cb.state
False