Browse Source

chore: improve the error handling and the quit routine (#494)

pull/496/head
Pier CeccoPierangioliEugenio 5 months ago committed by GitHub
parent
commit
d644604e37
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
  1. 8
      Makefile
  2. 15
      docs/MDNotes/internals/threads.md
  3. 17
      libs/pyTermTk/TermTk/TTkCore/TTkTerm/input_thread.py
  4. 9
      libs/pyTermTk/TermTk/TTkCore/TTkTerm/term_base.py
  5. 4
      libs/pyTermTk/TermTk/TTkCore/drivers/_unused_.unix_thread.py
  6. 14
      libs/pyTermTk/TermTk/TTkCore/drivers/term_unix_common.py
  7. 14
      libs/pyTermTk/TermTk/TTkCore/drivers/term_windows.py
  8. 5
      libs/pyTermTk/TermTk/TTkCore/helper.py
  9. 31
      libs/pyTermTk/TermTk/TTkCore/log.py
  10. 10
      libs/pyTermTk/TermTk/TTkCore/timer_pyodide.py
  11. 34
      libs/pyTermTk/TermTk/TTkCore/timer_unix.py
  12. 158
      libs/pyTermTk/TermTk/TTkCore/ttk.py
  13. 30
      tests/pytest/mock_input.py
  14. 11
      tests/pytest/mock_term.py
  15. 55
      tests/pytest/test_005_stderr.py
  16. 24
      tests/sandbox/Makefile
  17. 36
      tests/t.generic/test.threading.01.exceptionhook.py
  18. 70
      tests/t.ui/test.ui.035.error.handling.py
  19. 10
      tools/check.import.sh

8
Makefile

@ -133,14 +133,14 @@ test: .venv
# tests/pytest/test_001_demo.py -r test.input.bin
# Play the test stream
# tests/pytest/test_001_demo.py -p test.input.bin
mkdir -p tmp
wget -O tmp/test.input.001.bin https://github.com/ceccopierangiolieugenio/binaryRepo/raw/master/pyTermTk/tests/test.input.001.bin
wget -O tmp/test.input.002.bin https://github.com/ceccopierangiolieugenio/binaryRepo/raw/master/pyTermTk/tests/test.input.002.bin
wget -O tmp/test.input.003.bin https://github.com/ceccopierangiolieugenio/binaryRepo/raw/master/pyTermTk/tests/test.input.003.bin
tools/check.import.sh
. .venv/bin/activate ; \
flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics --exclude .venv,build,tmp,experiments ;
. .venv/bin/activate ; \
pytest tests/pytest
mkdir -p tmp
wget -O tmp/test.input.001.bin https://github.com/ceccopierangiolieugenio/binaryRepo/raw/master/pyTermTk/tests/test.input.001.bin
wget -O tmp/test.input.002.bin https://github.com/ceccopierangiolieugenio/binaryRepo/raw/master/pyTermTk/tests/test.input.002.bin
wget -O tmp/test.input.003.bin https://github.com/ceccopierangiolieugenio/binaryRepo/raw/master/pyTermTk/tests/test.input.003.bin
. .venv/bin/activate ; \
pytest tests/pytest/run_*

15
docs/MDNotes/internals/threads.md

@ -0,0 +1,15 @@
Summary of the main pyTermTk Threading (26/Oct/2025) [8148cf0a]
```text
MainThread (mainlop)
│ └─▶ Wait on inputQueue ─> Signal ┬─▶ inputEvent
│ ▲ └─▶ pasteEvent
│ │
│ (push to inputQueue)
├─▶ TTkInput Thread │
│ └─▶ Wait on select(Stdin)
└─▶ TTk (Draw)
└─▶ Wait on timeout (~65 fps) ──▶ Refresh Screen
```

17
libs/pyTermTk/TermTk/TTkCore/TTkTerm/input_thread.py

@ -27,6 +27,8 @@ from time import time
import threading, queue
from typing import Optional
from ..drivers import TTkInputDriver
from TermTk.TTkCore.log import TTkLog
@ -40,10 +42,11 @@ from TermTk.TTkCore.TTkTerm.inputmouse import TTkMouseEvent
class TTkInput:
inputEvent = pyTTkSignal(TTkKeyEvent, TTkMouseEvent)
pasteEvent = pyTTkSignal(str)
exceptionRaised = pyTTkSignal(Exception)
_pasteBuffer = ""
_bracketedPaste = False
_readInput = None
_inputThread = None
_inputThread:Optional[threading.Thread] = None
_inputQueue = None
_leftLastTime = 0
_midLastTime = 0
@ -69,6 +72,8 @@ class TTkInput:
TTkTerm.setMouse(False, False)
if TTkInput._readInput:
TTkInput._readInput.close()
if TTkInput._inputThread and TTkInput._inputThread.is_alive():
TTkInput._inputThread.join()
@staticmethod
def stop() -> None:
@ -106,9 +111,13 @@ class TTkInput:
@staticmethod
def _run():
for stdinRead in TTkInput._readInput.read():
outq = TTkInput.key_process(stdinRead)
TTkInput._inputQueue.put(outq)
try:
for stdinRead in TTkInput._readInput.read():
outq = TTkInput.key_process(stdinRead)
TTkInput._inputQueue.put(outq)
except Exception as e:
TTkInput._inputQueue.put(None)
TTkInput.exceptionRaised.emit(e)
TTkInput._inputQueue.put(None)
@staticmethod

9
libs/pyTermTk/TermTk/TTkCore/TTkTerm/term_base.py

@ -23,6 +23,8 @@
__all__ = ['TTkTermBase']
import os
from enum import Flag
from typing import Callable, TextIO
class TTkTermBase():
'''TTkTermBase'''
@ -79,7 +81,8 @@ class TTkTermBase():
def hide():
TTkTermBase.push(TTkTermBase.Cursor.HIDE)
class Sigmask():
class Sigmask(Flag):
NONE = 0x0000
CTRL_C = 0x0001
CTRL_S = 0x0002
CTRL_Z = 0x0004
@ -158,4 +161,6 @@ class TTkTermBase():
setEcho = lambda *args: None
CRNL = lambda *args: None
getTerminalSize = lambda *args: (80,24)
registerResizeCb = lambda *args: None
registerResizeCb:Callable[[Callable[[int,int],None]],None] = lambda w,h: None
getStdErr:Callable[[None],TextIO] = lambda : TextIO()
setStdErr:Callable[[TextIO],None] = lambda tio: None

4
libs/pyTermTk/TermTk/TTkCore/drivers/unix_thread.py → libs/pyTermTk/TermTk/TTkCore/drivers/_unused_.unix_thread.py

@ -20,6 +20,10 @@
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
##################################
# Experimental Driver - Unused #
##################################
__all__ = []
import sys, os

14
libs/pyTermTk/TermTk/TTkCore/drivers/term_unix_common.py

@ -25,6 +25,8 @@ __all__ = ['_TTkTerm']
import sys, os, signal
from threading import Thread, Lock
from typing import Callable, TextIO
try: import termios
except Exception as e:
print(f'ERROR: {e}')
@ -122,9 +124,19 @@ class _TTkTerm(TTkTermBase):
Thread(target=_TTkTerm._sigWinChThreaded).start()
@staticmethod
def _registerResizeCb(callback):
def _registerResizeCb(callback:Callable[[int,int],None]) -> None:
_TTkTerm._sigWinChCb = callback
# Dummy call to retrieve the terminal size
_TTkTerm._sigWinCh(signal.SIGWINCH, None)
signal.signal(signal.SIGWINCH, _TTkTerm._sigWinCh)
TTkTermBase.registerResizeCb = _registerResizeCb
@staticmethod
def _setStdErr(ioRedirect:TextIO) -> None:
sys.stderr = ioRedirect
TTkTermBase.setStdErr = _setStdErr
@staticmethod
def _getStdErr() -> TextIO:
return sys.stderr
TTkTermBase.getStdErr = _getStdErr

14
libs/pyTermTk/TermTk/TTkCore/drivers/term_windows.py

@ -25,6 +25,8 @@ __all__ = ['TTkTerm']
import sys, os
from threading import Thread, Lock
from typing import TextIO
from ..TTkTerm.term_base import TTkTermBase
from TermTk.TTkCore.log import TTkLog
from .windows import *
@ -71,4 +73,14 @@ class TTkTerm(TTkTermBase):
def _registerResizeCb(callback):
TTkTerm._sigWinChCb = callback
TTkInputDriver.windowResized.connect(TTkTerm._sigWinCh)
TTkTermBase.registerResizeCb = _registerResizeCb
TTkTermBase.registerResizeCb = _registerResizeCb
@staticmethod
def _setStdErr(ioRedirect:TextIO) -> None:
sys.stderr = ioRedirect
TTkTermBase.setStdErr = _setStdErr
@staticmethod
def _getStdErr() -> TextIO:
return sys.stderr
TTkTermBase.getStdErr = _getStdErr

5
libs/pyTermTk/TermTk/TTkCore/helper.py

@ -93,6 +93,11 @@ class TTkHelper:
TTkHelper._rootWidget = widget
TTkHelper._rootCanvas.enableDoubleBuffer()
@staticmethod
def cleanRootWidget() -> None:
TTkHelper._rootCanvas = None
TTkHelper._rootWidget = None
quitEvent: pyTTkSignal = pyTTkSignal()
@staticmethod

31
libs/pyTermTk/TermTk/TTkCore/log.py

@ -20,16 +20,14 @@
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
__all__ = ['TTkLog', 'ttk_capture_stderr']
__all__ = ['TTkLog']
# This code is inspired by
# https://github.com/ceccopierangiolieugenio/pyCuT/blob/master/cupy/CuTCore/CuDebug.py
import sys
import inspect
import logging
import contextlib
from collections.abc import Callable, Set
from typing import Callable
class _TTkContext:
__slots__ = ['file', 'line', 'function']
@ -113,28 +111,3 @@ class TTkLog:
@staticmethod
def installMessageHandler(mh: Callable):
TTkLog._messageHandler.append(mh)
class TTkStderrHandler:
def write(self, text):
TTkLog.error(text)
return len(text)
def flush(self):
pass
def getvalue(self):
raise NotImplementedError()
@contextlib.contextmanager
def ttk_capture_stderr():
_stderr_bk = sys.stderr
try:
sys.stderr = TTkStderrHandler()
yield
except Exception as e:
sys.stderr = _stderr_bk
TTkLog.critical(f"Caught an exception: {e}")
print(f"Caught an exception: {e}",sys.stderr)
finally:
sys.stderr = _stderr_bk

10
libs/pyTermTk/TermTk/TTkCore/timer_pyodide.py

@ -22,7 +22,7 @@
__all__ = ['TTkTimer']
from typing import Optional
from typing import Optional,Callable
from TermTk.TTkCore.helper import TTkHelper
from TermTk.TTkCore.signal import pyTTkSlot, pyTTkSignal
@ -39,7 +39,10 @@ class TTkTimer():
'_delay', '_delayLock', '_quit',
'_stopTime')
def __init__(self, name:Optional[str]=None):
def __init__(
self,
name:Optional[str]=None,
excepthook:Optional[Callable[[Exception],None]]=None):
# Define Signals
self.timeout = pyTTkSignal()
self._running = True
@ -81,3 +84,6 @@ class TTkTimer():
if self._timer:
pyodideProxy.stopTimeout(self._timer)
self._timer = None
def join(self) -> None:
pass

34
libs/pyTermTk/TermTk/TTkCore/timer_unix.py

@ -22,7 +22,7 @@
__all__ = ['TTkTimer']
from typing import Optional
from typing import Optional,Callable
import threading
@ -32,8 +32,16 @@ from TermTk.TTkCore.helper import TTkHelper
class TTkTimer(threading.Thread):
__slots__ = (
'timeout', '_delay',
'_timer', '_quit', '_start')
def __init__(self, name:Optional[str]=None):
'_timer', '_quit', '_start',
'_excepthook'
)
_delay:float
_excepthook:Optional[Callable[[Exception],None]]
def __init__(
self,
name:Optional[str]=None,
excepthook:Optional[Callable[[Exception],None]]=None):
self._excepthook = excepthook
self.timeout = pyTTkSignal()
self._delay = 0
self._quit = threading.Event()
@ -50,14 +58,21 @@ class TTkTimer(threading.Thread):
self._start.set()
def run(self):
while not self._quit.is_set():
self._start.wait()
self._start.clear()
if not self._timer.wait(self._delay):
self.timeout.emit()
try:
while not self._quit.is_set():
self._start.wait()
self._start.clear()
if not self._timer.wait(self._delay):
self.timeout.emit()
except Exception as e:
TTkHelper.quitEvent.disconnect(self.quit)
if self._excepthook:
self._excepthook(e)
else:
raise e
@pyTTkSlot(float)
def start(self, sec=0.0):
def start(self, sec:float=0.0):
self._delay = sec
self._timer.set()
self._timer.clear()
@ -68,4 +83,3 @@ class TTkTimer(threading.Thread):
@pyTTkSlot()
def stop(self):
self._timer.set()

158
libs/pyTermTk/TermTk/TTkCore/ttk.py

@ -27,6 +27,9 @@ import signal
import time
import threading
import platform
import contextlib
from typing import Optional, Callable, TextIO, List
from TermTk.TTkCore.drivers import TTkSignalDriver
from TermTk.TTkCore.TTkTerm.input import TTkInput
@ -35,7 +38,7 @@ from TermTk.TTkCore.TTkTerm.inputmouse import TTkMouseEvent
from TermTk.TTkCore.TTkTerm.term import TTkTerm
from TermTk.TTkCore.signal import pyTTkSignal, pyTTkSlot
from TermTk.TTkCore.constant import TTkK
from TermTk.TTkCore.log import TTkLog, ttk_capture_stderr
from TermTk.TTkCore.log import TTkLog
from TermTk.TTkCore.cfg import TTkCfg, TTkGlbl
from TermTk.TTkCore.helper import TTkHelper
from TermTk.TTkCore.timer import TTkTimer
@ -45,6 +48,33 @@ from TermTk.TTkWidgets.about import TTkAbout
from TermTk.TTkWidgets.widget import TTkWidget
from TermTk.TTkWidgets.container import TTkContainer
class _TTkStderrHandler(TextIO):
def write(self, text):
TTkLog.error(text)
return len(text)
def flush(self):
pass
def getvalue(self):
raise NotImplementedError()
@contextlib.contextmanager
def _ttk_capture_stderr():
_stderr_bk = TTkTerm.getStdErr()
try:
TTkTerm.setStdErr(_TTkStderrHandler())
yield
except Exception as e:
TTkTerm.setStdErr(_stderr_bk)
TTkHelper.quit()
raise e
finally:
TTkTerm.setStdErr(_stderr_bk)
class _MouseCursor():
__slots__ = ('_cursor','_color', '_pos', 'updated')
def __init__(self):
@ -73,6 +103,7 @@ class _MouseCursor():
self._pos = (mevt.x, mevt.y)
self.updated.emit()
class TTk(TTkContainer):
__slots__ = (
'_termMouse', '_termDirectMouse',
@ -82,11 +113,15 @@ class TTk(TTkContainer):
'_drawMutex',
'_paintEvent',
'_lastMultiTap',
'_exceptions',
'paintExecuted')
_timer:TTkTimer
_exceptions:List[Exception]
def __init__(self, *,
title:str='TermTk',
sigmask:TTkTerm.Sigmask=TTkK.NONE,
sigmask:TTkTerm.Sigmask=TTkTerm.Sigmask.NONE,
mouseTrack:bool=False,
mouseCursor:bool=False,
**kwargs) -> None:
@ -97,7 +132,8 @@ class TTk(TTkContainer):
if ('TERMTK_FILE_LOG' in os.environ and (_logFile := os.environ['TERMTK_FILE_LOG'])):
TTkLog.use_default_file_logging(_logFile)
self._timer = None
self._timer = TTkTimer(name='TTk (Draw)', excepthook=self._timer_exception_hook)
self._exceptions = []
self._title = title
self._sigmask = sigmask
self.paintExecuted = pyTTkSignal()
@ -106,8 +142,11 @@ class TTk(TTkContainer):
self._mouseCursor = None
self._showMouseCursor = os.environ.get("TERMTK_MOUSE",mouseCursor)
super().__init__(**kwargs)
TTkInput.inputEvent.connect(self._processInput)
TTkInput.pasteEvent.connect(self._processPaste)
TTkInput.exceptionRaised.connect(self._input_exception_hook)
TTkSignalDriver.sigStop.connect(self._SIGSTOP)
TTkSignalDriver.sigCont.connect(self._SIGCONT)
TTkSignalDriver.sigInt.connect( self._SIGINT)
@ -139,11 +178,11 @@ class TTk(TTkContainer):
self.frame = 0
self.time = curtime
def mainloop(self):
with ttk_capture_stderr():
self._mainloop_1()
def mainloop(self) -> None:
with _ttk_capture_stderr():
self._mainloop()
def _mainloop_1(self):
def _mainloop(self) -> None:
try:
'''Enters the main event loop and waits until :meth:`~quit` is called or the main widget is destroyed.'''
TTkLog.debug( "" )
@ -167,7 +206,6 @@ class TTk(TTkContainer):
TTkTerm.registerResizeCb(self._win_resize_cb)
self._timer = TTkTimer(name='TTk (Draw)')
self._timer.timeout.connect(self._time_event)
self._timer.start(0.1)
self.show()
@ -186,30 +224,66 @@ class TTk(TTkContainer):
self._mouseCursor.updated.connect(self.update)
self.paintChildCanvas = self._mouseCursorPaintChildCanvas
self._mainLoop_2()
if platform.system() != 'Emscripten':
TTkInput.start()
finally:
if platform.system() != 'Emscripten':
TTkHelper.quitEvent.emit()
if self._timer:
self._timer.timeout.disconnect(self._time_event)
self._timer.quit()
self._paintEvent.set()
# self._timer.join()
self._quit_timer()
self._timer.join()
TTkSignalDriver.exit()
self.quit()
TTkTerm.exit()
for e in self._exceptions:
raise e
@pyTTkSlot(Exception)
def _timer_exception_hook(self, e:Exception) -> None:
# This exception is raised during the '_timer' thread
# responsible of the drawing routine
# any failure in this loop should clean the environment and
# quit the app, I ensure in this routine to stop
# The '_timer' thread and te 'TTkInput' thread
# This will results in the mainloop proceed and run the
# 'finally:' block
self._quit_timer()
TTkInput.inputEvent.clear()
TTkInput.close() # Shoul close it and wait for join
# The exception will be raised in the main loop
# Once it is ensured that the env is clean
# And the terminal goes back to its original state
self._exceptions.append(e)
@pyTTkSlot(Exception)
def _input_exception_hook(self, e:Exception) -> None:
# This exception is raised during the 'TTkInput' thread
# responsible of collecting the mouse/keyboard/paste events
# any failure in this loop should clean the environment and
# quit the app, I ensure in this routine to stop
# The '_timer' thread and te 'TTkInput' thread
# This will results in the mainloop proceed and run the
# 'finally:' block
self._quit_timer()
TTkInput.inputEvent.clear()
TTkInput.close() # Shoul close it and wait for join
self._timer.join()
# The exception will be raised in the main loop
# Once it is ensured that the env is clean
# And the terminal goes back to its original state
self._exceptions.append(e)
def _quit_timer(self) -> None:
self._timer.timeout.disconnect(self._time_event)
self._timer.quit()
self._paintEvent.set()
def _mouseCursorPaintChildCanvas(self) -> None:
super().paintChildCanvas()
ch = self._mouseCursor._cursor
pos = self._mouseCursor._pos
color = self._mouseCursor._color
self.getCanvas().drawChar(char=ch, pos=pos, color=color)
def _mainLoop_2(self):
if platform.system() == 'Emscripten':
return
TTkInput.start()
if self._mouseCursor:
ch = self._mouseCursor._cursor
pos = self._mouseCursor._pos
color = self._mouseCursor._color
self.getCanvas().drawChar(char=ch, pos=pos, color=color)
@pyTTkSlot(str)
def _processPaste(self, txt:str):
@ -219,12 +293,11 @@ class TTk(TTkContainer):
@pyTTkSlot(TTkKeyEvent, TTkMouseEvent)
def _processInput(self, kevt, mevt):
self._drawMutex.acquire()
if kevt is not None:
self._key_event(kevt)
if mevt is not None:
self._mouse_event(mevt)
self._drawMutex.release()
with self._drawMutex:
if kevt is not None:
self._key_event(kevt)
if mevt is not None:
self._mouse_event(mevt)
def _mouse_event(self, mevt):
# Upload the global mouse position
@ -310,21 +383,19 @@ class TTk(TTkContainer):
self._paintEvent.clear()
w,h = TTkTerm.getTerminalSize()
self._drawMutex.acquire()
self.setGeometry(0,0,w,h)
self._fps()
TTkHelper.paintAll()
self.paintExecuted.emit()
self._drawMutex.release()
with self._drawMutex:
self.setGeometry(0,0,w,h)
self._fps()
TTkHelper.paintAll()
self.paintExecuted.emit()
self._timer.start(1/TTkCfg.maxFps)
def _win_resize_cb(self, width, height):
TTkGlbl.term_w = int(width)
TTkGlbl.term_h = int(height)
self._drawMutex.acquire()
self.setGeometry(0,0,TTkGlbl.term_w,TTkGlbl.term_h)
TTkHelper.rePaintAll()
self._drawMutex.release()
with self._drawMutex:
self.setGeometry(0,0,TTkGlbl.term_w,TTkGlbl.term_h)
TTkHelper.rePaintAll()
TTkLog.info(f"Resize: w:{TTkGlbl.term_w}, h:{TTkGlbl.term_h}")
@pyTTkSlot()
@ -348,8 +419,13 @@ class TTk(TTkContainer):
@pyTTkSlot()
def _quit(self):
'''Tells the application to exit with a return code.'''
if self._timer:
self._timer.timeout.disconnect(self._time_event)
TTkHelper.cleanRootWidget()
TTkInput.inputEvent.disconnect(self._processInput)
TTkInput.pasteEvent.disconnect(self._processPaste)
TTkSignalDriver.sigStop.disconnect(self._SIGSTOP)
TTkSignalDriver.sigCont.disconnect(self._SIGCONT)
TTkSignalDriver.sigInt.disconnect( self._SIGINT)
self._quit_timer()
TTkInput.inputEvent.clear()
TTkInput.close()

30
tests/pytest/mock_input.py

@ -22,7 +22,25 @@
# Thanks to: https://stackoverflow.com/questions/43162722/mocking-a-module-import-in-pytest
class mock_signal():
@staticmethod
def connect(*args,**argv):
pass
@staticmethod
def disconnect(*args,**argv):
pass
@staticmethod
def emit(*args,**argv):
pass
@staticmethod
def clear():
pass
class Mock_TTkInput():
exceptionRaised = mock_signal
inputEvent = mock_signal
pasteEvent = mock_signal
@staticmethod
def init(mouse, directMouse):pass
@staticmethod
@ -37,15 +55,3 @@ class Mock_TTkInput():
def get_key( callback=None): pass
@staticmethod
def start(): pass
class inputEvent():
def connect(*args):
pass
def clear():
pass
class pasteEvent():
def connect(*args):
pass
def clear():
pass

11
tests/pytest/mock_term.py

@ -23,6 +23,7 @@
# Thanks to: https://stackoverflow.com/questions/43162722/mocking-a-module-import-in-pytest
import sys
from enum import Flag
class Mock_TTkTerm():
CLEAR = None
@ -63,11 +64,13 @@ class Mock_TTkTerm():
@staticmethod
def hide(): pass
class Sigmask():
class Sigmask(Flag):
NONE = 0x0000
CTRL_C = 0x0001
CTRL_S = 0x0002
CTRL_Z = 0x0004
CTRL_Q = 0x0008
CTRL_Y = 0x0010
@staticmethod
def push(*args):
@ -83,3 +86,9 @@ class Mock_TTkTerm():
@staticmethod
def getTerminalSize():
return 250,70
@staticmethod
def getStdErr():
return sys.stderr
@staticmethod
def setStdErr(err):
sys.stderr = err

55
tests/pytest/test_005_stderr.py

@ -21,51 +21,36 @@
# SOFTWARE.
import sys, os, io
import logging
import pytest
from typing import Union, Optional
sys.path.append(os.path.join(sys.path[0],'../../libs/pyTermTk'))
import TermTk as ttk
class FakeStderr():
@pytest.fixture(autouse=True)
def reset_fake_stderr():
"""Reset FakeStderr before each test"""
FakeStderr.value = []
yield
FakeStderr.value = []
class FakeStderr:
value = []
def message_handler(mode, context, message):
FakeStderr.value.append(message)
msgType = "NONE"
if mode == ttk.TTkLog.InfoMsg: msgType = "[INFO]"
elif mode == ttk.TTkLog.WarningMsg: msgType = "[WARNING]"
elif mode == ttk.TTkLog.CriticalMsg: msgType = "[CRITICAL]"
elif mode == ttk.TTkLog.FatalMsg: msgType = "[FATAL]"
elif mode == ttk.TTkLog.ErrorMsg: msgType = "[ERROR]"
if mode == ttk.TTkLog.InfoMsg:
msgType = "[INFO]"
elif mode == ttk.TTkLog.WarningMsg:
msgType = "[WARNING]"
elif mode == ttk.TTkLog.CriticalMsg:
msgType = "[CRITICAL]"
elif mode == ttk.TTkLog.FatalMsg:
msgType = "[FATAL]"
elif mode == ttk.TTkLog.ErrorMsg:
msgType = "[ERROR]"
print(f"{msgType} {context.file} {message}")
def test_stderr_01():
ttk.TTkLog.installMessageHandler(message_handler)
print('Test',file=sys.stderr)
with ttk.ttk_capture_stderr():
print('XXXXX Test',file=sys.stderr)
with open('pippo','r') as f:
f.read()
raise ValueError('YYYYY Test')
print('After Test')
def test_ttk_capture_stderr():
ttk.TTkLog.installMessageHandler(message_handler)
with ttk.ttk_capture_stderr() as fake_stderr:
print("This is an error message", file=sys.stderr)
output = FakeStderr.value
assert "This is an error message" in output
def test_ttk_capture_stderr_exception_handling():
ttk.TTkLog.installMessageHandler(message_handler)
with ttk.ttk_capture_stderr() as fake_stderr:
raise ValueError("Test exception")
# After the exception, sys.stderr should be restored
output = FakeStderr.value
assert any("Test exception" in _o for _o in output)
assert isinstance(sys.stderr, io.TextIOWrapper) or sys.stderr is sys.__stderr__

24
tests/sandbox/Makefile

@ -12,6 +12,7 @@ www:
www/w2ui
wget -P www/pyodide/ https://ceccopierangiolieugenio.github.io/pyTermTk-Docs/sandbox/www/pyodide/pyodide.js
wget -P www/pyodide/ https://ceccopierangiolieugenio.github.io/pyTermTk-Docs/sandbox/www/pyodide/pyodide-lock.json
wget -P www/pyodide/ https://ceccopierangiolieugenio.github.io/pyTermTk-Docs/sandbox/www/pyodide/python_stdlib.zip
wget -P www/pyodide/ https://ceccopierangiolieugenio.github.io/pyTermTk-Docs/sandbox/www/pyodide/pyodide.asm.js
# wget -P www/pyodide/ https://ceccopierangiolieugenio.github.io/pyTermTk-Docs/sandbox/www/pyodide/repodata.json
@ -29,13 +30,13 @@ www:
wget -P www/fontawesome/ https://ceccopierangiolieugenio.github.io/pyTermTk-Docs/sandbox/www/fontawesome/regular.min.css
wget -P www/fontawesome/ https://ceccopierangiolieugenio.github.io/pyTermTk-Docs/sandbox/www/fontawesome/fontawesome.min.css
# wget -P www/webfonts/ https://ceccopierangiolieugenio.github.io/pyTermTk-Docs/sandbox/www/webfonts/fa-regular-400.woff2
wget -P www/webfonts/ https://ceccopierangiolieugenio.github.io/pyTermTk-Docs/sandbox/www/webfonts/fa-regular-400.woff2
# wget -P www/fonts/nerdfonts/ https://ceccopierangiolieugenio.github.io/pyTermTk-Docs/sandbox/www/nerdfonts/HurmitNerdFontMono-Regular.otf
wget -P www/fonts/nerdfonts/ https://ceccopierangiolieugenio.github.io/pyTermTk-Docs/sandbox/www/fonts/nerdfonts/DejaVuSansMNerdFont-Regular.ttf
wget -P www/w2ui/ https://ceccopierangiolieugenio.github.io/pyTermTk-Docs/sandbox/www/w2ui/w2ui-2.0.min.js
wget -P www/w2ui/ https://ceccopierangiolieugenio.github.io/pyTermTk-Docs/sandbox/www/w2ui/w2ui-2.0.min.css
wget -P www/w2ui/ https://ceccopierangiolieugenio.github.io/pyTermTk-Docs/sandbox/www/w2ui/w2ui.es6.min.js
wget -P www/codemirror/ https://ceccopierangiolieugenio.github.io/pyTermTk-Docs/sandbox/www/codemirror/codemirror.js
wget -P www/codemirror/ https://ceccopierangiolieugenio.github.io/pyTermTk-Docs/sandbox/www/codemirror/codemirror.css
@ -72,18 +73,7 @@ buildSandbox: www
$( cd ../../ ; tools/prepareBuild.sh release ; )
find ../../tmp/TermTk/ -name "*.py" | sed 's,.*tmp/,,' | sort | xargs tar cvzf bin/TermTk.tgz -C ../../tmp
find ../../tutorial -name '*.py' -o -name '*.json' | sort | xargs tar cvzf bin/tutorial.tgz
find ../../demo/paint.py ../../demo/ttkode.py ../../demo/demo.py ../../demo/showcase/*.* | sort | xargs tar cvzf bin/demo.tgz
find ../../tests/ansi.images.json ../../tests/t.ui/*.* | sort | xargs tar cvzf bin/tests.tgz
buildTestSandbox: www
rm -rf bin
mkdir -p bin
$( cd ../../ ; tools/prepareBuild.sh release ; )
find ../../TermTk/ -name "*.py" | sort | xargs tar cvzf bin/TermTk.tgz
find ../../tutorial -name "*.py" | sort | xargs tar cvzf bin/tutorial.tgz
find ../../demo/paint.py ../../demo/ttkode.py ../../demo/demo.py ../../demo/showcase/*.* | sort | xargs tar cvzf bin/demo.tgz
find ../../tests/ansi.images.json ../../tests/t.ui/*.* | sort | xargs tar cvzf bin/tests.tgz
find ../../libs/pyTermTk/TermTk -name "*.py" | sed 's,.*/pyTermTk/,,' | sort | xargs tar cvzf bin/TermTk.tgz -C ../../libs/pyTermTk
find ../../tutorial -name '*.py' -o -name '*.json' | sed 's,.*/tutorial/,tutorial/,' | sort | xargs tar cvzf bin/tutorial.tgz -C ../..
find ../../demo/paint.py ../../demo/ttkode.py ../../demo/demo.py ../../demo/showcase/*.* | sed 's,.*/demo/,demo/,' | sort | xargs tar cvzf bin/demo.tgz -C ../..
find ../../tests/ansi.images.json ../../tests/t.ui/*.* | sed 's,.*/tests/,tests/,' | sort | xargs tar cvzf bin/tests.tgz -C ../..

36
tests/t.generic/test.threading.01.exceptionhook.py

@ -0,0 +1,36 @@
import threading
import traceback
def custom_thread_excepthook(args):
"""
Args is a named tuple with:
- exc_type: Exception class
- exc_value: Exception instance
- exc_traceback: Traceback object
- thread: Thread object where exception occurred
"""
print(f"Thread {args.thread.name} crashed!")
print(f"Exception: {args.exc_type.__name__}: {args.exc_value}")
# Print full traceback like default handler
print("\nFull traceback:")
traceback.print_exception(args.exc_type, args.exc_value, args.exc_traceback, colorize=True)
# Or to match default format exactly:
# ßsys.stderr.write(f"\n\nPIPPO\n\nException in thread {args.thread.name}:\n")
# traceback.print_exception(args.exc_type, args.exc_value, args.exc_traceback, file=sys.stderr)
# Set custom handler
threading.excepthook = custom_thread_excepthook
# Example thread that will crash
def worker():
raise ValueError("Something went wrong!")
thread = threading.Thread(name='Pippo', target=worker)
try:
thread.start()
thread.join()
except Exception as e:
print(e)
print('EUGENIO')

70
tests/t.ui/test.ui.035.error.handling.py

@ -0,0 +1,70 @@
#!/usr/bin/env python3
# MIT License
#
# Copyright (c) 2025 Eugenio Parodi <ceccopierangiolieugenio AT googlemail DOT com>
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
import sys, os
import threading
sys.path.append(os.path.join(sys.path[0],'../../libs/pyTermTk'))
import TermTk as ttk
ttk.TTkLog.use_default_file_logging()
@ttk.pyTTkSlot()
def _raise():
raise Exception('_raise FAIL!!!')
class Fail(ttk.TTkWidget):
fail_state:bool = False
def mousePressEvent(self, evt):
Fail.fail_state = True
self.update()
return True
def paintEvent(self, canvas):
if Fail.fail_state:
raise Exception('Fail FAIL!!!')
canvas.fill(color=ttk.TTkColor.BG_RED)
canvas.drawText(text='X', pos=(2,1))
timer = ttk.TTkTimer()
timer.timeout.connect(_raise)
root = ttk.TTk()
# simulate error
# Fail in the main Thread
ttk.TTkButton(parent=root,text=' X ',border=True).clicked.connect(_raise)
# Fail in the draw Thread
Fail(parent=root,pos=(0,3), size=(5,3))
# Generic Failure on a generic thread
ttk.TTkButton(parent=root,pos=(0,6),text=' X ',border=True).clicked.connect(lambda:threading.Thread(target=_raise).start())
# Generic Failure on a TTkTimer
ttk.TTkButton(parent=root,pos=(0,9),text=' X ',border=True).clicked.connect(lambda:timer.start())
# Generic Quit
ttk.TTkButton(parent=root,pos=(0,13),text=' X ',border=True).clicked.connect(root.quit)
ttk.TTkButton(parent=root,pos=(0,16),text=' X ',border=True).clicked.connect(ttk.TTkHelper.quit)
win=ttk.TTkWindow(parent=root, pos=(5,0), size=(100,30), border=True, layout=ttk.TTkGridLayout())
ttk.TTkLogViewer(parent=win)
root.mainloop()

10
tools/check.import.sh

@ -31,6 +31,7 @@ __check(){
-e "ttk.py:import queue" \
-e "ttk.py:import threading" \
-e "ttk.py:import platform" \
-e "ttk.py:import contextlib" \
-e "clipboard.py:import importlib.util" \
-e "filebuffer.py:import threading" \
-e "texedit.py:from math import log10, floor" \
@ -47,6 +48,7 @@ __check(){
-e "savetools.py:import json" \
-e "TTkCore/constant.py:from enum import IntEnum" |
grep -v \
-e "TTkTerm/term_base.py:from enum import Flag" \
-e "TTkTerm/input_mono.py:from time import time" \
-e "TTkTerm/input_mono.py:import platform" \
-e "TTkTerm/input_mono.py:from ..drivers import TTkInputDriver" \
@ -61,10 +63,10 @@ __check(){
-e "TTkGui/textdocument_highlight_pygments.py:from pygments" |
grep -v \
-e "TTkTerm/term.py:from ..drivers import *" \
-e "drivers/unix_thread.py:import sys, os" \
-e "drivers/unix_thread.py:from select import select" \
-e "drivers/unix_thread.py:import threading" \
-e "drivers/unix_thread.py:import queue" \
-e "drivers/_unused_.unix_thread.py:import sys, os" \
-e "drivers/_unused_.unix_thread.py:from select import select" \
-e "drivers/_unused_.unix_thread.py:import threading" \
-e "drivers/_unused_.unix_thread.py:import queue" \
-e "drivers/unix.py:import sys, os, re" \
-e "drivers/unix.py:import signal" \
-e "drivers/unix.py:from select import select" \

Loading…
Cancel
Save