10 changed files with 1036 additions and 7 deletions
@ -0,0 +1,55 @@
|
||||
# MIT License |
||||
# |
||||
# Copyright (c) 2026 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. |
||||
|
||||
__all__ = ['PTP_TestResult', 'PTP_ScanResult', 'PTP_Node'] |
||||
|
||||
from dataclasses import dataclass |
||||
from typing import Optional,List,Tuple |
||||
|
||||
@dataclass |
||||
class PTP_Node(): |
||||
nodeId: str |
||||
filename: str |
||||
lineNumber:int = 0 |
||||
|
||||
@dataclass |
||||
class PTP_TestResult(): |
||||
nodeId: str |
||||
outcome: str |
||||
duration: float |
||||
longrepr: Optional[str] |
||||
sections: List[Tuple[str,str]] |
||||
location: tuple[str, int | None, str] |
||||
|
||||
@dataclass |
||||
class PTP_ScanResult(): |
||||
nodeId: str |
||||
path:str |
||||
lineno:int |
||||
testname:str |
||||
|
||||
def get_node(self) -> PTP_Node: |
||||
return PTP_Node( |
||||
nodeId=self.nodeId, |
||||
filename=self.path, |
||||
lineNumber=self.lineno |
||||
) |
||||
@ -0,0 +1,157 @@
|
||||
# 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. |
||||
|
||||
__all__ = ['PTP_Engine'] |
||||
|
||||
import os |
||||
import sys |
||||
import json |
||||
import threading |
||||
import subprocess |
||||
from pathlib import Path |
||||
from dataclasses import dataclass, asdict, fields |
||||
from typing import Dict, Any, Optional, List |
||||
|
||||
import TermTk as ttk |
||||
|
||||
from _030.pytest_data import PTP_TestResult, PTP_ScanResult |
||||
|
||||
def _strip_result(result: str) -> str: |
||||
lines = result.splitlines() |
||||
indent = min(len(line) - len(line.lstrip()) for line in lines if line.strip()) |
||||
result = "\n".join(line[indent:] for line in lines) |
||||
return result.lstrip('\n') |
||||
|
||||
class PTP_Engine(): |
||||
__slots__ = ( |
||||
'_scan_lock', '_test_lock', |
||||
'itemScanned','testResultReady', |
||||
'errorReported', |
||||
'endScan', 'endTest') |
||||
|
||||
def __init__(self): |
||||
self.itemScanned = ttk.pyTTkSignal(PTP_ScanResult) |
||||
self.testResultReady = ttk.pyTTkSignal(PTP_TestResult) |
||||
self.errorReported = ttk.pyTTkSignal(str) |
||||
self.endScan = ttk.pyTTkSignal() |
||||
self.endTest = ttk.pyTTkSignal() |
||||
self._scan_lock = threading.RLock() |
||||
self._test_lock = threading.RLock() |
||||
|
||||
def scan(self) -> Dict: |
||||
threading.Thread(target=self._scan_thread).start() |
||||
|
||||
def _scan_thread(self): |
||||
with self._scan_lock: |
||||
dirname = os.path.dirname(__file__) |
||||
script_file = Path(dirname) / 'scripts' / '_main_scan.py' |
||||
script = script_file.read_text() |
||||
# Run the script in a subprocess and capture output |
||||
process = subprocess.Popen( |
||||
[sys.executable, "-c", script, dirname], |
||||
stdout=subprocess.PIPE, |
||||
stderr=subprocess.PIPE, |
||||
text=True, |
||||
bufsize=1, # Line buffered |
||||
universal_newlines=True |
||||
) |
||||
|
||||
def read_stdout(): |
||||
for line in iter(process.stdout.readline, ''): |
||||
line = line.strip() |
||||
if line: |
||||
try: |
||||
result_dict = json.loads(line) |
||||
result = PTP_ScanResult(**result_dict) |
||||
self.itemScanned.emit(result) |
||||
except (json.JSONDecodeError, KeyError) as e: |
||||
ttk.TTkLog.error(f"Error parsing test result: {line}") |
||||
process.stdout.close() |
||||
|
||||
def read_stderr(): |
||||
for line in iter(process.stderr.readline, ''): |
||||
self.errorReported.emit(line) |
||||
process.stderr.close() |
||||
|
||||
# Start threads to read stdout and stderr |
||||
stdout_thread = threading.Thread(target=read_stdout) |
||||
stderr_thread = threading.Thread(target=read_stderr) |
||||
|
||||
stdout_thread.start() |
||||
stderr_thread.start() |
||||
|
||||
# Wait for process to complete |
||||
process.wait() |
||||
|
||||
self.endScan.emit() |
||||
|
||||
|
||||
def run_all_tests(self): |
||||
self.run_tests('.') |
||||
|
||||
def run_tests(self, test_path:str): |
||||
threading.Thread(target=self._run_tests_thread, args=(test_path,)).start() |
||||
|
||||
def _run_tests_thread(self, test_path:str): |
||||
with self._test_lock: |
||||
dirname = os.path.dirname(__file__) |
||||
script_file = Path(dirname) / 'scripts' / '_main_tests.py' |
||||
script = script_file.read_text() |
||||
|
||||
# Run the script in a subprocess with streaming output |
||||
process = subprocess.Popen( |
||||
[sys.executable, "-c", script, dirname, test_path], |
||||
stdout=subprocess.PIPE, |
||||
stderr=subprocess.PIPE, |
||||
text=True, |
||||
bufsize=1, # Line buffered |
||||
universal_newlines=True |
||||
) |
||||
|
||||
def read_stdout(): |
||||
for line in iter(process.stdout.readline, ''): |
||||
line = line.strip() |
||||
if line: |
||||
try: |
||||
result_dict = json.loads(line) |
||||
result = PTP_TestResult(**result_dict) |
||||
self.testResultReady.emit(result) |
||||
except (json.JSONDecodeError, KeyError) as e: |
||||
ttk.TTkLog.error(f"Error parsing test result: {line}") |
||||
process.stdout.close() |
||||
|
||||
def read_stderr(): |
||||
for line in iter(process.stderr.readline, ''): |
||||
self.errorReported.emit(line) |
||||
process.stderr.close() |
||||
|
||||
# Start threads to read stdout and stderr |
||||
stdout_thread = threading.Thread(target=read_stdout) |
||||
stderr_thread = threading.Thread(target=read_stderr) |
||||
|
||||
stdout_thread.start() |
||||
stderr_thread.start() |
||||
|
||||
# Wait for process to complete |
||||
process.wait() |
||||
|
||||
self.endTest.emit() |
||||
@ -0,0 +1,338 @@
|
||||
# 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. |
||||
|
||||
''' PyTest Tree Widget Components |
||||
|
||||
This module provides specialized tree widget components for displaying pytest |
||||
test hierarchies with visual status indicators and interactive controls for |
||||
running individual tests or test groups. |
||||
''' |
||||
|
||||
from __future__ import annotations |
||||
|
||||
__all__ = [ |
||||
'PTP_Tree', |
||||
'PTP_Action', |
||||
'PTP_TreeItem', 'PTP_TreeItemPath', 'PTP_TreeItemMethod'] |
||||
|
||||
from enum import IntEnum |
||||
from dataclasses import dataclass |
||||
from typing import Optional |
||||
|
||||
import TermTk as ttk |
||||
|
||||
from _030.pytest_data import PTP_Node |
||||
from ttkode.app.ttkode import TTKodeFileWidgetItem |
||||
|
||||
class _testStatus(IntEnum): |
||||
''' Test status enumeration for visual indicators |
||||
|
||||
Represents the current state of a test item in the tree. |
||||
''' |
||||
Pass = 0x01 |
||||
Fail = 0x02 |
||||
Undefined = 0x03 |
||||
|
||||
_statusMarks = { |
||||
_testStatus.Pass: ttk.TTkString('✔ ', ttk.TTkColor.GREEN), |
||||
_testStatus.Fail: ttk.TTkString('x ', ttk.TTkColor.RED), |
||||
_testStatus.Undefined: ttk.TTkString('○ ', ttk.TTkColor.fg('#888888')), |
||||
} |
||||
|
||||
def _toMark(status:_testStatus) -> ttk.TTkString: |
||||
''' Convert test status to colored visual mark |
||||
|
||||
:param status: the test status |
||||
:type status: _testStatus |
||||
|
||||
:return: colored status mark (✔, x, or ○) |
||||
:rtype: :py:class:`TTkString` |
||||
''' |
||||
return _statusMarks.get(status, ttk.TTkString('- ')) |
||||
|
||||
class PTP_TreeItem(ttk.TTkTreeWidgetItem): |
||||
''' Base tree item for pytest test nodes |
||||
|
||||
Extends :py:class:`TTkTreeWidgetItem` to add test-specific functionality including |
||||
status tracking and visual indicators. |
||||
|
||||
:param name: display name for the tree item |
||||
:type name: str |
||||
:param test_id: unique pytest node ID |
||||
:type test_id: str |
||||
:param node: associated pytest node data |
||||
:type node: PTP_Node |
||||
:param testStatus: initial test status |
||||
:type testStatus: _testStatus |
||||
''' |
||||
__slots__ = ('_testStatus', '_ptp_node', '_test_id') |
||||
def __init__(self, name:str, test_id:str, node:PTP_Node, testStatus:_testStatus=_testStatus.Undefined, **kwargs): |
||||
self._testStatus = testStatus |
||||
self._ptp_node = node |
||||
self._test_id = test_id |
||||
super().__init__([ttk.TTkString(name)], **kwargs) |
||||
|
||||
def test_id(self) -> str: |
||||
''' Get the pytest node ID for this item |
||||
|
||||
:return: the test ID path |
||||
:rtype: str |
||||
''' |
||||
return self._test_id |
||||
|
||||
def node(self) -> PTP_Node: |
||||
''' Get the pytest node data |
||||
|
||||
:return: the associated node data |
||||
:rtype: PTP_Node |
||||
''' |
||||
return self._ptp_node |
||||
|
||||
def data(self, col, role = None) -> ttk.TTkString: |
||||
''' Get display data with status mark prefix |
||||
|
||||
:param col: column index |
||||
:type col: int |
||||
:param role: data role (unused) |
||||
:type role: Optional[Any] |
||||
|
||||
:return: formatted string with status mark |
||||
:rtype: :py:class:`TTkString` |
||||
''' |
||||
return _toMark(self._testStatus) + super().data(col, role) |
||||
|
||||
def testStatus(self) -> _testStatus: |
||||
''' Get the current test status |
||||
|
||||
:return: the test status |
||||
:rtype: _testStatus |
||||
''' |
||||
return self._testStatus |
||||
|
||||
def clearTestStatus(self, clearParent:bool=False, clearChildren:bool=False) -> None: |
||||
''' Clear test status to undefined |
||||
|
||||
:param clearParent: whether to recursively clear parent status |
||||
:type clearParent: bool |
||||
:param clearChildren: whether to recursively clear children status |
||||
:type clearChildren: bool |
||||
''' |
||||
if self._testStatus is _testStatus.Undefined: |
||||
return |
||||
self._testStatus = _testStatus.Undefined |
||||
if clearParent: |
||||
if isinstance(self._parent, PTP_TreeItem): |
||||
self._parent.clearTestStatus(clearParent=True) |
||||
self.dataChanged.emit() |
||||
|
||||
def setTestStatus(self, status:_testStatus) -> None: |
||||
''' Set the test status and update parent if needed |
||||
|
||||
:param status: the new test status |
||||
:type status: _testStatus |
||||
''' |
||||
if status == self._testStatus: |
||||
return |
||||
self._testStatus = status |
||||
if isinstance(self._parent, PTP_TreeItemPath): |
||||
self._parent._updateTestStatus(status=status) |
||||
self.dataChanged.emit() |
||||
|
||||
class PTP_TreeItemPath(PTP_TreeItem): |
||||
''' Tree item representing a test path or directory |
||||
|
||||
Aggregates status from child test items - shows failure if any child fails. |
||||
''' |
||||
def _updateTestStatus(self, status:_testStatus): |
||||
''' Update status based on child test statuses |
||||
|
||||
:param status: the new status to propagate |
||||
:type status: _testStatus |
||||
''' |
||||
fail = any(_c.testStatus() == _testStatus.Fail for _c in self.children() if isinstance(_c, PTP_TreeItem)) |
||||
self._testStatus = _testStatus.Fail if fail else status |
||||
if isinstance(self._parent, PTP_TreeItemPath): |
||||
self._parent._updateTestStatus(status=status) |
||||
self.dataChanged.emit() |
||||
|
||||
def clearTestStatus(self, clearParent:bool=False, clearChildren:bool=False) -> None: |
||||
if clearChildren: |
||||
for _c in self.children(): |
||||
if isinstance(_c, PTP_TreeItem): |
||||
_c.clearTestStatus(clearChildren=True) |
||||
super().clearTestStatus(clearParent=clearParent) |
||||
|
||||
class PTP_TreeItemMethod(PTP_TreeItem): |
||||
''' Tree item representing an individual test method |
||||
|
||||
Leaf node in the test tree hierarchy. |
||||
''' |
||||
pass |
||||
|
||||
|
||||
@dataclass |
||||
class _PTP_Highlight(): |
||||
''' Internal state for mouse hover highlighting |
||||
|
||||
:param pos: vertical position of the highlight |
||||
:type pos: int |
||||
:param run: whether mouse is over the run button area |
||||
:type run: bool |
||||
:param item: the highlighted tree item |
||||
:type item: PTP_TreeItem |
||||
''' |
||||
pos:int |
||||
run:bool |
||||
item:PTP_TreeItem |
||||
|
||||
@dataclass |
||||
class PTP_Action(): |
||||
''' Action data emitted when a test run button is clicked |
||||
|
||||
:param item: the test item to run |
||||
:type item: PTP_TreeItem |
||||
''' |
||||
item:PTP_TreeItem |
||||
|
||||
class PTP_TreeWidget(ttk.TTkTreeWidget): |
||||
''' Custom tree widget with interactive run buttons |
||||
|
||||
Extends :py:class:`TTkTreeWidget` to add per-item run buttons that appear |
||||
on mouse hover, allowing individual test or test group execution. |
||||
|
||||
:: |
||||
|
||||
✔ test_module.py ...[ ]▶ |
||||
○ test_pending |
||||
x test_failed |
||||
|
||||
''' |
||||
__slots__ = ('_PTP_highight', 'actionPressed') |
||||
|
||||
_PTP_highight:Optional[_PTP_Highlight] |
||||
actionPressed: ttk.pyTTkSignal |
||||
''' |
||||
This signal is emitted when a run button is clicked. |
||||
|
||||
:param action: the action containing the item to run |
||||
:type action: PTP_Action |
||||
''' |
||||
|
||||
def __init__(self, **kwargs): |
||||
self._PTP_highight = None |
||||
self.actionPressed = ttk.pyTTkSignal(PTP_Action) |
||||
super().__init__(**kwargs) |
||||
self.mergeStyle({'default':{'hoveredColor':ttk.TTkColor.bg('#666666')}}) |
||||
|
||||
def mousePressEvent(self, evt:ttk.TTkMouseEvent) -> bool: |
||||
''' Handle mouse press events for run button activation |
||||
|
||||
:param evt: the mouse event |
||||
:type evt: :py:class:`TTkMouseEvent` |
||||
|
||||
:return: True if event was handled |
||||
:rtype: bool |
||||
''' |
||||
y,x = evt.y, evt.x |
||||
w = self.width() |
||||
if (_item:=self.itemAt(y)) and x>=w-3: |
||||
self.actionPressed.emit(PTP_Action(item=_item)) |
||||
self.update() |
||||
return True |
||||
return super().mousePressEvent(evt) |
||||
|
||||
def mouseMoveEvent(self, evt:ttk.TTkMouseEvent) -> bool: |
||||
''' Handle mouse movement to update hover highlighting |
||||
|
||||
:param evt: the mouse event |
||||
:type evt: :py:class:`TTkMouseEvent` |
||||
|
||||
:return: True if event was handled |
||||
:rtype: bool |
||||
''' |
||||
y,x = evt.y, evt.x |
||||
w = self.width() |
||||
if _item:=self.itemAt(y): |
||||
self._PTP_highight = _PTP_Highlight( |
||||
pos=evt.y, |
||||
run=x>=w-3, |
||||
item=_item |
||||
) |
||||
self.update() |
||||
elif self._PTP_highight is not None: |
||||
self._PTP_highight = None |
||||
self.update() |
||||
return super().mouseMoveEvent(evt) |
||||
|
||||
def leaveEvent(self, evt:ttk.TTkMouseEvent) -> bool: |
||||
''' Clear highlighting when mouse leaves widget |
||||
|
||||
:param evt: the mouse event |
||||
:type evt: :py:class:`TTkMouseEvent` |
||||
|
||||
:return: True if event was handled |
||||
:rtype: bool |
||||
''' |
||||
if self._PTP_highight is not None: |
||||
self._PTP_highight = None |
||||
self.update() |
||||
super().leaveEvent(evt) |
||||
|
||||
def paintEvent(self, canvas:ttk.TTkCanvas) -> None: |
||||
''' Paint the tree with interactive run buttons on hover |
||||
|
||||
:param canvas: the canvas to paint on |
||||
:type canvas: :py:class:`TTkCanvas` |
||||
''' |
||||
super().paintEvent(canvas) |
||||
style = self.currentStyle() |
||||
hoveredColor=style['hoveredColor'] |
||||
if _ph:=self._PTP_highight: |
||||
w = self.width() |
||||
if _ph.run: |
||||
canvas.drawText(text='[ ]', pos=(w-3,_ph.pos), color=hoveredColor+ttk.TTkColor.YELLOW+ttk.TTkColor.BOLD) |
||||
canvas.drawText(text= '▶' , pos=(w-2,_ph.pos), color=hoveredColor+ttk.TTkColor.RED) |
||||
else: |
||||
canvas.drawText(text='[ ]', pos=(w-3,_ph.pos), color=hoveredColor+ttk.TTkColor.YELLOW+ttk.TTkColor.BOLD) |
||||
canvas.drawText(text= '▶' , pos=(w-2,_ph.pos), color=hoveredColor+ttk.TTkColor.GREEN) |
||||
# canvas.drawText(text= '▷' , pos=(w-3,_ph.pos), color=hoveredColor+ttk.TTkColor.GREEN) |
||||
|
||||
|
||||
class PTP_Tree(ttk.TTkTree): |
||||
''' Pytest tree container with scrollable view |
||||
|
||||
Wraps :py:class:`PTP_TreeWidget` in a scrollable :py:class:`TTkTree` container, |
||||
exposing the actionPressed signal for test execution. |
||||
''' |
||||
__slots__ = ('actionPressed') |
||||
actionPressed: ttk.pyTTkSignal |
||||
''' |
||||
This signal is emitted when a run button is clicked. |
||||
|
||||
:param action: the action containing the item to run |
||||
:type action: PTP_Action |
||||
''' |
||||
|
||||
def __init__(self, **kwargs): |
||||
tw = PTP_TreeWidget(**kwargs) |
||||
super().__init__(treeWidget=tw, **kwargs) |
||||
self.actionPressed = tw.actionPressed |
||||
@ -0,0 +1,303 @@
|
||||
# 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. |
||||
|
||||
''' PyTest Widget Plugin |
||||
|
||||
This module provides a pytest integration widget for ttkode that allows |
||||
scanning, running, and viewing test results in a tree structure. |
||||
''' |
||||
|
||||
__all__ = ['PTP_PyTestWidget'] |
||||
|
||||
from typing import Dict, List, Any, Optional |
||||
|
||||
from pygments import highlight |
||||
from pygments.lexers import PythonLexer |
||||
from pygments.formatters import TerminalTrueColorFormatter |
||||
|
||||
import TermTk as ttk |
||||
|
||||
from ttkode.proxy import ttkodeProxy |
||||
from ttkode.app.ttkode import TTKodeFileWidgetItem |
||||
|
||||
from .pytest_tree import ( |
||||
_testStatus, |
||||
PTP_TreeItem, PTP_TreeItemPath, PTP_TreeItemMethod, |
||||
PTP_Tree, PTP_Action) |
||||
from .pytest_engine import PTP_Engine, PTP_TestResult, PTP_ScanResult |
||||
from .pytest_data import PTP_Node |
||||
|
||||
def _strip_result(result: str) -> str: |
||||
''' Strip common leading whitespace from multi-line test result strings |
||||
|
||||
:param result: the test result string to strip |
||||
:type result: str |
||||
|
||||
:return: the stripped result string |
||||
:rtype: str |
||||
''' |
||||
lines = result.splitlines() |
||||
indent = min(len(line) - len(line.lstrip()) for line in lines if line.strip()) |
||||
result = "\n".join(line[indent:] for line in lines) |
||||
return result.lstrip('\n') |
||||
|
||||
_out_map = { |
||||
'passed': ttk.TTkString('PASS', ttk.TTkColor.GREEN)+ttk.TTkColor.RST, |
||||
'failed': ttk.TTkString('FAIL', ttk.TTkColor.RED)+ttk.TTkColor.RST |
||||
} |
||||
_error_color = ttk.TTkColor.fgbg('#FFFF00',"#FF0000") |
||||
|
||||
class PTP_PyTestWidget(ttk.TTkContainer): |
||||
''' PTP_PyTestWidget: |
||||
|
||||
A widget for integrating pytest test execution and result viewing in ttkode. |
||||
Provides a tree view of discovered tests, action buttons for scanning and running, |
||||
and displays test results with syntax-highlighted error messages. |
||||
|
||||
:: |
||||
|
||||
┌─────────────────────────┐ |
||||
│ [Scan] [Run] │ |
||||
├─────────────────────────┤ |
||||
│ ▼ tests/ │ |
||||
│ ▼ test_module.py │ |
||||
│ ✓ test_pass │ |
||||
│ ✗ test_fail │ |
||||
│ ? test_pending │ |
||||
└─────────────────────────┘ |
||||
|
||||
:param testResults: the text edit widget to display test results |
||||
:type testResults: :py:class:`TTkTextEdit` |
||||
''' |
||||
__slots__ = ( |
||||
'_test_engine', |
||||
'_res_tree','_test_results') |
||||
_res_tree:PTP_Tree |
||||
_test_results:ttk.TTkTextEdit |
||||
|
||||
def __init__(self, testResults=ttk.TTkTextEdit, **kwargs): |
||||
self._test_results = testResults |
||||
self._test_engine = PTP_Engine() |
||||
super().__init__(**kwargs) |
||||
self.setLayout(layout:=ttk.TTkGridLayout()) |
||||
|
||||
layout.addWidget(btn_scan:=ttk.TTkButton(text="Scan", border=False), 0,0) |
||||
layout.addWidget(btn_run:=ttk.TTkButton(text="Run", border=False), 0,1) |
||||
layout.addWidget(res_tree:=PTP_Tree(dragDropMode=ttk.TTkK.DragDropMode.AllowDrag), 1,0,1,2) |
||||
res_tree.setHeaderLabels(["Tests"]) |
||||
|
||||
testResults.setText('Info Tests') |
||||
|
||||
self._res_tree = res_tree |
||||
|
||||
btn_scan.clicked.connect(self._scan) |
||||
btn_run.clicked.connect(self._run_tests) |
||||
self._test_engine.testResultReady.connect(self._test_updated) |
||||
|
||||
res_tree.actionPressed.connect(self._tree_action_pressed) |
||||
res_tree.itemActivated.connect(self._tree_item_activated) |
||||
|
||||
@ttk.pyTTkSlot(PTP_Action) |
||||
def _tree_action_pressed(self, action:PTP_Action) -> None: |
||||
''' Handle action button presses in the test tree |
||||
|
||||
Clears previous test status and runs tests for the selected item. |
||||
|
||||
:param action: the tree action containing the test item to run |
||||
:type action: PTP_Action |
||||
''' |
||||
item = action.item |
||||
if isinstance(item, PTP_TreeItem): |
||||
item.clearTestStatus(clearChildren=True, clearParent=True) |
||||
self._test_results.clear() |
||||
self._test_engine.run_tests(item.test_id()) |
||||
|
||||
@ttk.pyTTkSlot(PTP_TreeItem, int) |
||||
def _tree_item_activated(self, item:PTP_TreeItem, _): |
||||
''' Handle test tree item activation |
||||
|
||||
When a test method is activated, opens the corresponding file at the test line number. |
||||
|
||||
:param item: the activated tree item |
||||
:type item: PTP_TreeItem |
||||
:param _: column index (unused) |
||||
:type _: int |
||||
''' |
||||
ttk.TTkLog.debug(item) |
||||
if isinstance(item, PTP_TreeItemMethod): |
||||
file = item.node().filename |
||||
line = item.node().lineNumber |
||||
ttkodeProxy.openFile(file, line) |
||||
|
||||
def _get_node_from_path(self, _n:ttk.TTkTreeWidgetItem, _p:str) -> Optional[PTP_TreeItem]: |
||||
''' Recursively search for a tree node by test ID path |
||||
|
||||
:param _n: the tree node to search from |
||||
:type _n: :py:class:`TTkTreeWidgetItem` |
||||
:param _p: the test ID path to find |
||||
:type _p: str |
||||
|
||||
:return: the matching tree item or None if not found |
||||
:rtype: Optional[PTP_TreeItem] |
||||
''' |
||||
for _c in _n.children(): |
||||
if isinstance(_c, PTP_TreeItem) and _c.test_id() == _p: |
||||
return _c |
||||
if _cc:= self._get_node_from_path(_n=_c, _p=_p): |
||||
return _cc |
||||
return None |
||||
|
||||
@ttk.pyTTkSlot() |
||||
def _scan(self): |
||||
''' Scan the workspace for pytest tests |
||||
|
||||
Clears the current tree and results, then initiates a pytest collection |
||||
to discover all available tests and populate the tree structure. |
||||
''' |
||||
self._res_tree.clear() |
||||
self._test_results.clear() |
||||
|
||||
def _get_or_add_path_in_node(_n:ttk.TTkTreeWidgetItem, _p:str, _name:str, _node:PTP_ScanResult) -> PTP_TreeItem: |
||||
for _c in _n.children(): |
||||
if isinstance(_c, PTP_TreeItem) and _c.test_id() == _p: |
||||
return _c |
||||
_ptp_node = _node.get_node() |
||||
if _node.nodeId==_p: # This is the leaf test |
||||
_n.addChild(_c:=PTP_TreeItemMethod(test_id=_p, name=_name, node=_ptp_node, expanded=True)) |
||||
else: |
||||
_n.addChild(_c:=PTP_TreeItemPath(test_id=_p, name=_name, node=_ptp_node, expanded=True)) |
||||
return _c |
||||
|
||||
@ttk.pyTTkSlot(PTP_ScanResult) |
||||
def _add_node(_node:PTP_ScanResult)->None: |
||||
self._test_results.append(_node.nodeId) |
||||
_node_id_split = _node.nodeId.split('::') |
||||
_full_test_path = _node_id_split[0] |
||||
_leaves = _node_id_split[1:] |
||||
_tree_node: ttk.TTkTreeWidgetItem = self._res_tree.invisibleRootItem() |
||||
_full_composite_test_path = '' |
||||
for _test_path in _full_test_path.split('/'): |
||||
if _full_composite_test_path: |
||||
_full_composite_test_path = '/'.join([_full_composite_test_path,_test_path]) |
||||
else: |
||||
_full_composite_test_path = _test_path |
||||
_tree_node = _get_or_add_path_in_node(_n=_tree_node, _p=_full_composite_test_path, _name=_test_path, _node=_node) |
||||
for _leaf in _leaves: |
||||
_full_composite_test_path = '::'.join([_full_composite_test_path,_leaf]) |
||||
_tree_node = _get_or_add_path_in_node(_n=_tree_node, _p=_full_composite_test_path, _name=_leaf, _node=_node) |
||||
# _tree_node = _tree_node.addChild(TestTreeItemMethod([ttk.TTkString(_leaf)],test_call=_node.nodeId)) |
||||
|
||||
@ttk.pyTTkSlot(str) |
||||
def _add_error(error:str) -> None: |
||||
self._test_results.append(ttk.TTkString(error, ttk.TTkColor.RED)) |
||||
|
||||
@ttk.pyTTkSlot() |
||||
def _end_scan() -> None: |
||||
self._res_tree.resizeColumnToContents(0) |
||||
self._test_engine.itemScanned.disconnect(_add_node) |
||||
|
||||
self._test_engine.errorReported.connect(_add_error) |
||||
self._test_engine.endScan.connect(_end_scan) |
||||
self._test_engine.itemScanned.connect(_add_node) |
||||
|
||||
self._test_engine.scan() |
||||
|
||||
def _mark_node(self, _nodeid:str, outcome:str) -> None: |
||||
''' Mark a test node and its ancestors with the test outcome status |
||||
|
||||
Updates the visual status of tree items based on test results. |
||||
Parent nodes inherit failure status from children. |
||||
|
||||
:param _nodeid: the test node ID to mark |
||||
:type _nodeid: str |
||||
:param outcome: the test outcome ("passed", "failed", "undefined") |
||||
:type outcome: str |
||||
''' |
||||
status = { |
||||
'passed' : _testStatus.Pass, |
||||
'failed' : _testStatus.Fail, |
||||
}.get(outcome, _testStatus.Undefined) |
||||
def _recurse_node(_n:ttk.TTkTreeWidgetItem): |
||||
for _c in _n.children(): |
||||
if isinstance(_c, PTP_TreeItemPath): |
||||
if _nodeid.startswith(_c.test_id()): |
||||
_recurse_node(_c) |
||||
elif isinstance(_c, PTP_TreeItemMethod) and _nodeid.startswith(_c.test_id()): |
||||
_c.setTestStatus(status) |
||||
|
||||
_recurse_node(self._res_tree.invisibleRootItem()) |
||||
|
||||
def _clear_nodes(self, node:Optional[PTP_TreeItem] = None) -> None: |
||||
''' Clear test status for all nodes in the tree |
||||
|
||||
Resets all test items to undefined status. |
||||
|
||||
:param node: the root node to clear from, defaults to tree root |
||||
:type node: Optional[PTP_TreeItem] |
||||
''' |
||||
if node: |
||||
node.clearTestStatus(clearChildren=True, clearParent=True) |
||||
else: |
||||
for _c in self._res_tree.invisibleRootItem().children(): |
||||
if isinstance(_c, PTP_TreeItem): |
||||
_c.clearTestStatus(clearChildren=True) |
||||
|
||||
@ttk.pyTTkSlot(PTP_TestResult) |
||||
def _test_updated(self, test:PTP_TestResult) -> None: |
||||
''' Handle test result updates from the test engine |
||||
|
||||
Updates tree node status and displays formatted test results including |
||||
outcome, duration, captured output, and syntax-highlighted error messages. |
||||
|
||||
:param test: the test result data |
||||
:type test: PTP_TestResult |
||||
''' |
||||
self._mark_node(_nodeid=test.nodeId, outcome=test.outcome) |
||||
|
||||
_outcome = _out_map.get(test.outcome, test.outcome) |
||||
self._test_results.append(f"{test.nodeId}: " + _outcome + f" ({test.duration:.2f}s)") |
||||
|
||||
for (_s_name,_s_content) in test.sections: |
||||
self._test_results.append(ttk.TTkString(_s_name, ttk.TTkColor.GREEN)) |
||||
self._test_results.append(_s_content) |
||||
|
||||
if test.outcome == 'failed': |
||||
self._test_results.append(ttk.TTkString("◢◤◢◤◢◤◢◤◢◤◢◤◢◤◢◤◢◤◢◤◢◤◢◤◢◤◢◤◢◤ ↳ Error: ◢◤◢◤◢◤◢◤◢◤◢◤◢◤◢◤◢◤◢◤◢◤◢◤◢◤◢◤◢◤", ttk.TTkColor.RED)) |
||||
_code = test.longrepr |
||||
_code_str = ttk.TTkString(highlight(_code, PythonLexer(), TerminalTrueColorFormatter(style='material'))) |
||||
_code_h_err = [ |
||||
_l.setColor(_error_color) if _l._text.startswith('E') else _l |
||||
for _l in _code_str.split('\n')] |
||||
self._test_results.append(ttk.TTkString('\n').join(_code_h_err)) |
||||
self._test_results.append(ttk.TTkString("◢◤◢◤◢◤◢◤◢◤◢◤◢◤◢◤◢◤◢◤◢◤◢◤◢◤◢◤◢◤◢◤◢◤◢◤◢◤◢◤◢◤◢◤◢◤◢◤◢◤◢◤◢◤◢◤◢◤◢◤◢◤◢◤◢◤◢◤◢◤", ttk.TTkColor.RED)) |
||||
|
||||
|
||||
@ttk.pyTTkSlot() |
||||
def _run_tests(self) -> None: |
||||
''' Run all discovered tests |
||||
|
||||
Clears existing test statuses and results, then executes all tests |
||||
that were previously discovered by the scan operation. |
||||
''' |
||||
self._clear_nodes() |
||||
self._test_results.clear() |
||||
self._test_engine.run_all_tests() |
||||
@ -0,0 +1,100 @@
|
||||
# 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 |
||||
import json |
||||
import pytest |
||||
|
||||
from dataclasses import asdict |
||||
from typing import Dict, Any |
||||
|
||||
from _030.pytest_data import PTP_TestResult, PTP_ScanResult |
||||
|
||||
class _Highlighter(): |
||||
_highlight = None |
||||
|
||||
class _Dummy(): |
||||
_tw = _Highlighter() |
||||
|
||||
class ResultCollector_Logreport: |
||||
def pytest_configure(self, config:pytest.Config): |
||||
config.pluginmanager.register(_Dummy(), "terminalreporter") |
||||
# config.option.verbose = 2 |
||||
|
||||
|
||||
def pytest_runtest_logreport(self, report:pytest.TestReport) -> None: |
||||
if report.when == "call": |
||||
result = PTP_TestResult( |
||||
nodeId = report.nodeid, |
||||
outcome = report.outcome, |
||||
duration = report.duration, |
||||
sections = list(report.sections), |
||||
longrepr = str(report.longrepr) if report.failed else None, |
||||
location = report.location |
||||
) |
||||
|
||||
# Print detailed output for failed tests (like terminal plugin does) |
||||
# if report.failed and report.longrepr: |
||||
# print(f"\n{'='*70}") |
||||
# print(f"FAILED: {report.nodeid}") |
||||
# print(f"{'='*70}") |
||||
# print(report.longrepr) |
||||
# print() |
||||
|
||||
# # Print captured output sections |
||||
# if report.sections: |
||||
# for section_name, section_content in report.sections: |
||||
# print(f"\n{'-'*70} {section_name} {'-'*70}") |
||||
# print(section_content) |
||||
|
||||
# print('REPORT: -------') |
||||
# print(report) |
||||
# print('SECTIONS: -------') |
||||
# for s in report.sections: |
||||
# print('sec',s[0]) |
||||
# print('cont',s[1]) |
||||
# print('LONGREPR: -------') |
||||
# print(report.longrepr) |
||||
# print('LOCATION: -------') |
||||
# print(1,report.location[0]) |
||||
# print(2,report.location[1]) |
||||
# print(3,report.location[2]) |
||||
# # Print result immediately for real-time streaming |
||||
# print('RET: -------') |
||||
|
||||
print(json.dumps(asdict(result)), flush=True) |
||||
|
||||
|
||||
class ResultCollector_ItemCollected: |
||||
def __init__(self): |
||||
self.results:Dict[str,Any] = [] |
||||
|
||||
def pytest_itemcollected(self, item:pytest.Item) -> None: |
||||
# Called during --collect-only |
||||
relfspath, lineno, testname = item.location |
||||
result = PTP_ScanResult( |
||||
nodeId=item.nodeid, |
||||
path=relfspath, |
||||
lineno=lineno, |
||||
testname=testname |
||||
) |
||||
print(json.dumps(asdict(result)), flush=True) |
||||
@ -0,0 +1,35 @@
|
||||
# MIT License |
||||
# |
||||
# Copyright (c) 2026 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. |
||||
|
||||
def main() -> None: |
||||
import sys, pytest |
||||
|
||||
dirname = sys.argv[1] |
||||
|
||||
sys.path.append(f'{dirname}/..') |
||||
sys.path.append(f'{dirname}/scripts') |
||||
|
||||
from _glue_lib import ResultCollector_ItemCollected |
||||
|
||||
collector = ResultCollector_ItemCollected() |
||||
pytest.main(['--collect-only', '-p', 'no:terminal', '.'], plugins=[collector]) |
||||
main() |
||||
@ -0,0 +1,39 @@
|
||||
# MIT License |
||||
# |
||||
# Copyright (c) 2026 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. |
||||
|
||||
def main() -> None: |
||||
import sys, pytest |
||||
|
||||
dirname = sys.argv[1] |
||||
test_path = sys.argv[2] |
||||
|
||||
# dirname = "apps/ttkode/ttkode/plugins/_030" |
||||
# test_path = "tests/pytest/test_005_tree.py::test_tree_show_hide" |
||||
|
||||
sys.path.append(f'{dirname}/..') |
||||
sys.path.append(f'{dirname}/scripts') |
||||
|
||||
from _glue_lib import ResultCollector_Logreport |
||||
|
||||
collector = ResultCollector_Logreport() |
||||
pytest.main(['-p', 'no:terminal', test_path], plugins=[collector]) |
||||
main() |
||||
Loading…
Reference in new issue