Add canonical Python 3.15 optional native integration sources (#9368)

Copy curses frontend sources and canonical curses, DBM, Tcl and DTrace
integration tests from CPython v3.15.0 through update_lib. Preserve
upstream optional-native/resource/configuration gates and canonical file
modes, without adding native implementations or custom skips.

Validate source identity, hooks and default/frozen compile-only checks.
Native runtime integration and probe execution remain unvalidated.

Assisted-by: OpenAI Codex: model version unavailable
This commit is contained in:
youknowdot authored and GitHub committed 2026-10-26 10:10:47 +09:00
1 parent 163b8c557c
commit dca9dfdf38
12 files changed
+4319 -73

No files matched your search

+98
View File
@@ -0,0 +1,98 @@
"""curses
The main package for curses support for Python. Normally used by importing
the package, and perhaps a particular module inside it.
import curses
from curses import textpad
curses.initscr()
...
"""
from _curses import *
import os as _os
import sys as _sys
# Some constants, most notably the ACS_* ones, are only added to the C
# _curses module's dictionary after initscr() is called. (Some
# versions of SGI's curses don't define values for those constants
# until initscr() has been called.) This wrapper function calls the
# underlying C initscr(), and then copies the constants from the
# _curses module to the curses package's dictionary. Don't do 'from
# curses import *' if you'll be needing the ACS_* constants.
def initscr():
import _curses, curses
# we call setupterm() here because it raises an error
# instead of calling exit() in error cases.
setupterm(term=_os.environ.get("TERM", "unknown"),
fd=_sys.__stdout__.fileno())
stdscr = _curses.initscr()
for key, value in _curses.__dict__.items():
if key.startswith('ACS_') or key in ('LINES', 'COLS'):
setattr(curses, key, value)
return stdscr
# This is a similar wrapper for start_color(), which adds the COLORS and
# COLOR_PAIRS variables which are only available after start_color() is
# called.
def start_color():
import _curses, curses
_curses.start_color()
curses.COLORS = _curses.COLORS
curses.COLOR_PAIRS = _curses.COLOR_PAIRS
# Import Python has_key() implementation if _curses doesn't contain has_key()
try:
has_key
except NameError:
from .has_key import has_key # noqa: F401
# Wrapper for the entire curses-based application. Runs a function which
# should be the rest of your curses-based application. If the application
# raises an exception, wrapper() will restore the terminal to a sane state so
# you can read the resulting traceback.
def wrapper(func, /, *args, **kwds):
"""Wrapper function that initializes curses and calls another function,
restoring normal keyboard/screen behavior on error.
The callable object 'func' is then passed the main window 'stdscr'
as its first argument, followed by any other arguments passed to
wrapper().
"""
try:
# Initialize curses
stdscr = initscr()
# Turn off echoing of keys, and enter cbreak mode,
# where no buffering is performed on keyboard input
noecho()
cbreak()
# In keypad mode, escape sequences for special keys
# (like the cursor keys) will be interpreted and
# a special value like curses.KEY_LEFT will be returned
stdscr.keypad(1)
# Start color, too. Harmless if the terminal doesn't have
# color; user can test with has_color() later on. The try/catch
# works around a minor bit of over-conscientiousness in the curses
# module -- the error return from C start_color() is ignorable,
# unless they are raised by the interpreter due to other issues.
try:
start_color()
except _curses.error:
pass
return func(stdscr, *args, **kwds)
finally:
# Set everything back to normal
if 'stdscr' in locals():
stdscr.keypad(0)
echo()
nocbreak()
endwin()
+99
View File
@@ -0,0 +1,99 @@
"""Constants and membership tests for ASCII characters"""
NUL = 0x00 # ^@
SOH = 0x01 # ^A
STX = 0x02 # ^B
ETX = 0x03 # ^C
EOT = 0x04 # ^D
ENQ = 0x05 # ^E
ACK = 0x06 # ^F
BEL = 0x07 # ^G
BS = 0x08 # ^H
TAB = 0x09 # ^I
HT = 0x09 # ^I
LF = 0x0a # ^J
NL = 0x0a # ^J
VT = 0x0b # ^K
FF = 0x0c # ^L
CR = 0x0d # ^M
SO = 0x0e # ^N
SI = 0x0f # ^O
DLE = 0x10 # ^P
DC1 = 0x11 # ^Q
DC2 = 0x12 # ^R
DC3 = 0x13 # ^S
DC4 = 0x14 # ^T
NAK = 0x15 # ^U
SYN = 0x16 # ^V
ETB = 0x17 # ^W
CAN = 0x18 # ^X
EM = 0x19 # ^Y
SUB = 0x1a # ^Z
ESC = 0x1b # ^[
FS = 0x1c # ^\
GS = 0x1d # ^]
RS = 0x1e # ^^
US = 0x1f # ^_
SP = 0x20 # space
DEL = 0x7f # delete
controlnames = [
"NUL", "SOH", "STX", "ETX", "EOT", "ENQ", "ACK", "BEL",
"BS", "HT", "LF", "VT", "FF", "CR", "SO", "SI",
"DLE", "DC1", "DC2", "DC3", "DC4", "NAK", "SYN", "ETB",
"CAN", "EM", "SUB", "ESC", "FS", "GS", "RS", "US",
"SP"
]
def _ctoi(c):
if isinstance(c, str):
return ord(c)
else:
return c
def isalnum(c): return isalpha(c) or isdigit(c)
def isalpha(c): return isupper(c) or islower(c)
def isascii(c): return 0 <= _ctoi(c) <= 127 # ?
def isblank(c): return _ctoi(c) in (9, 32)
def iscntrl(c): return 0 <= _ctoi(c) <= 31 or _ctoi(c) == 127
def isdigit(c): return 48 <= _ctoi(c) <= 57
def isgraph(c): return 33 <= _ctoi(c) <= 126
def islower(c): return 97 <= _ctoi(c) <= 122
def isprint(c): return 32 <= _ctoi(c) <= 126
def ispunct(c): return isgraph(c) and not isalnum(c)
def isspace(c): return _ctoi(c) in (9, 10, 11, 12, 13, 32)
def isupper(c): return 65 <= _ctoi(c) <= 90
def isxdigit(c): return isdigit(c) or \
(65 <= _ctoi(c) <= 70) or (97 <= _ctoi(c) <= 102)
def isctrl(c): return 0 <= _ctoi(c) < 32
def ismeta(c): return _ctoi(c) > 127
def ascii(c):
if isinstance(c, str):
return chr(_ctoi(c) & 0x7f)
else:
return _ctoi(c) & 0x7f
def ctrl(c):
if isinstance(c, str):
return chr(_ctoi(c) & 0x1f)
else:
return _ctoi(c) & 0x1f
def alt(c):
if isinstance(c, str):
return chr(_ctoi(c) | 0x80)
else:
return _ctoi(c) | 0x80
def unctrl(c):
bits = _ctoi(c)
if bits == 0x7f:
rep = "^?"
elif isprint(bits & 0x7f):
rep = chr(bits & 0x7f)
else:
rep = "^" + chr(((bits & 0x7f) | 0x20) + 0x20)
if bits & 0x80:
return "!" + rep
return rep
+192
View File
@@ -0,0 +1,192 @@
#
# Emulation of has_key() function for platforms that don't use ncurses
#
import _curses
# Table mapping curses keys to the terminfo capability name
_capability_names = frozendict({
_curses.KEY_A1: 'ka1',
_curses.KEY_A3: 'ka3',
_curses.KEY_B2: 'kb2',
_curses.KEY_BACKSPACE: 'kbs',
_curses.KEY_BEG: 'kbeg',
_curses.KEY_BTAB: 'kcbt',
_curses.KEY_C1: 'kc1',
_curses.KEY_C3: 'kc3',
_curses.KEY_CANCEL: 'kcan',
_curses.KEY_CATAB: 'ktbc',
_curses.KEY_CLEAR: 'kclr',
_curses.KEY_CLOSE: 'kclo',
_curses.KEY_COMMAND: 'kcmd',
_curses.KEY_COPY: 'kcpy',
_curses.KEY_CREATE: 'kcrt',
_curses.KEY_CTAB: 'kctab',
_curses.KEY_DC: 'kdch1',
_curses.KEY_DL: 'kdl1',
_curses.KEY_DOWN: 'kcud1',
_curses.KEY_EIC: 'krmir',
_curses.KEY_END: 'kend',
_curses.KEY_ENTER: 'kent',
_curses.KEY_EOL: 'kel',
_curses.KEY_EOS: 'ked',
_curses.KEY_EXIT: 'kext',
_curses.KEY_F0: 'kf0',
_curses.KEY_F1: 'kf1',
_curses.KEY_F10: 'kf10',
_curses.KEY_F11: 'kf11',
_curses.KEY_F12: 'kf12',
_curses.KEY_F13: 'kf13',
_curses.KEY_F14: 'kf14',
_curses.KEY_F15: 'kf15',
_curses.KEY_F16: 'kf16',
_curses.KEY_F17: 'kf17',
_curses.KEY_F18: 'kf18',
_curses.KEY_F19: 'kf19',
_curses.KEY_F2: 'kf2',
_curses.KEY_F20: 'kf20',
_curses.KEY_F21: 'kf21',
_curses.KEY_F22: 'kf22',
_curses.KEY_F23: 'kf23',
_curses.KEY_F24: 'kf24',
_curses.KEY_F25: 'kf25',
_curses.KEY_F26: 'kf26',
_curses.KEY_F27: 'kf27',
_curses.KEY_F28: 'kf28',
_curses.KEY_F29: 'kf29',
_curses.KEY_F3: 'kf3',
_curses.KEY_F30: 'kf30',
_curses.KEY_F31: 'kf31',
_curses.KEY_F32: 'kf32',
_curses.KEY_F33: 'kf33',
_curses.KEY_F34: 'kf34',
_curses.KEY_F35: 'kf35',
_curses.KEY_F36: 'kf36',
_curses.KEY_F37: 'kf37',
_curses.KEY_F38: 'kf38',
_curses.KEY_F39: 'kf39',
_curses.KEY_F4: 'kf4',
_curses.KEY_F40: 'kf40',
_curses.KEY_F41: 'kf41',
_curses.KEY_F42: 'kf42',
_curses.KEY_F43: 'kf43',
_curses.KEY_F44: 'kf44',
_curses.KEY_F45: 'kf45',
_curses.KEY_F46: 'kf46',
_curses.KEY_F47: 'kf47',
_curses.KEY_F48: 'kf48',
_curses.KEY_F49: 'kf49',
_curses.KEY_F5: 'kf5',
_curses.KEY_F50: 'kf50',
_curses.KEY_F51: 'kf51',
_curses.KEY_F52: 'kf52',
_curses.KEY_F53: 'kf53',
_curses.KEY_F54: 'kf54',
_curses.KEY_F55: 'kf55',
_curses.KEY_F56: 'kf56',
_curses.KEY_F57: 'kf57',
_curses.KEY_F58: 'kf58',
_curses.KEY_F59: 'kf59',
_curses.KEY_F6: 'kf6',
_curses.KEY_F60: 'kf60',
_curses.KEY_F61: 'kf61',
_curses.KEY_F62: 'kf62',
_curses.KEY_F63: 'kf63',
_curses.KEY_F7: 'kf7',
_curses.KEY_F8: 'kf8',
_curses.KEY_F9: 'kf9',
_curses.KEY_FIND: 'kfnd',
_curses.KEY_HELP: 'khlp',
_curses.KEY_HOME: 'khome',
_curses.KEY_IC: 'kich1',
_curses.KEY_IL: 'kil1',
_curses.KEY_LEFT: 'kcub1',
_curses.KEY_LL: 'kll',
_curses.KEY_MARK: 'kmrk',
_curses.KEY_MESSAGE: 'kmsg',
_curses.KEY_MOVE: 'kmov',
_curses.KEY_NEXT: 'knxt',
_curses.KEY_NPAGE: 'knp',
_curses.KEY_OPEN: 'kopn',
_curses.KEY_OPTIONS: 'kopt',
_curses.KEY_PPAGE: 'kpp',
_curses.KEY_PREVIOUS: 'kprv',
_curses.KEY_PRINT: 'kprt',
_curses.KEY_REDO: 'krdo',
_curses.KEY_REFERENCE: 'kref',
_curses.KEY_REFRESH: 'krfr',
_curses.KEY_REPLACE: 'krpl',
_curses.KEY_RESTART: 'krst',
_curses.KEY_RESUME: 'kres',
_curses.KEY_RIGHT: 'kcuf1',
_curses.KEY_SAVE: 'ksav',
_curses.KEY_SBEG: 'kBEG',
_curses.KEY_SCANCEL: 'kCAN',
_curses.KEY_SCOMMAND: 'kCMD',
_curses.KEY_SCOPY: 'kCPY',
_curses.KEY_SCREATE: 'kCRT',
_curses.KEY_SDC: 'kDC',
_curses.KEY_SDL: 'kDL',
_curses.KEY_SELECT: 'kslt',
_curses.KEY_SEND: 'kEND',
_curses.KEY_SEOL: 'kEOL',
_curses.KEY_SEXIT: 'kEXT',
_curses.KEY_SF: 'kind',
_curses.KEY_SFIND: 'kFND',
_curses.KEY_SHELP: 'kHLP',
_curses.KEY_SHOME: 'kHOM',
_curses.KEY_SIC: 'kIC',
_curses.KEY_SLEFT: 'kLFT',
_curses.KEY_SMESSAGE: 'kMSG',
_curses.KEY_SMOVE: 'kMOV',
_curses.KEY_SNEXT: 'kNXT',
_curses.KEY_SOPTIONS: 'kOPT',
_curses.KEY_SPREVIOUS: 'kPRV',
_curses.KEY_SPRINT: 'kPRT',
_curses.KEY_SR: 'kri',
_curses.KEY_SREDO: 'kRDO',
_curses.KEY_SREPLACE: 'kRPL',
_curses.KEY_SRIGHT: 'kRIT',
_curses.KEY_SRSUME: 'kRES',
_curses.KEY_SSAVE: 'kSAV',
_curses.KEY_SSUSPEND: 'kSPD',
_curses.KEY_STAB: 'khts',
_curses.KEY_SUNDO: 'kUND',
_curses.KEY_SUSPEND: 'kspd',
_curses.KEY_UNDO: 'kund',
_curses.KEY_UP: 'kcuu1'
})
def has_key(ch):
if isinstance(ch, str):
ch = ord(ch)
# Figure out the correct capability name for the keycode.
capability_name = _capability_names.get(ch)
if capability_name is None:
return False
#Check the current terminal description for that capability;
#if present, return true, else return false.
if _curses.tigetstr(capability_name):
return True
else:
return False
if __name__ == '__main__':
# Compare the output of this implementation and the ncurses has_key,
# on platforms where has_key is already available
try:
L = []
_curses.initscr()
for key in _capability_names.keys():
system = _curses.has_key(key)
python = has_key(key)
if system != python:
L.append( 'Mismatch for key %s, system=%i, Python=%i'
% (_curses.keyname( key ), system, python) )
finally:
_curses.endwin()
for i in L: print(i)
+6
View File
@@ -0,0 +1,6 @@
"""curses.panel
Module for using panels with curses.
"""
from _curses_panel import *
+236
View File
@@ -0,0 +1,236 @@
"""Simple textbox editing widget with Emacs-like keybindings."""
import curses
import curses.ascii
def rectangle(win, uly, ulx, lry, lrx):
"""Draw a rectangle with corners at the provided upper-left
and lower-right coordinates.
"""
win.vline(uly+1, ulx, curses.ACS_VLINE, lry - uly - 1)
win.hline(uly, ulx+1, curses.ACS_HLINE, lrx - ulx - 1)
win.hline(lry, ulx+1, curses.ACS_HLINE, lrx - ulx - 1)
win.vline(uly+1, lrx, curses.ACS_VLINE, lry - uly - 1)
win.addch(uly, ulx, curses.ACS_ULCORNER)
win.addch(uly, lrx, curses.ACS_URCORNER)
win.addch(lry, lrx, curses.ACS_LRCORNER)
win.addch(lry, ulx, curses.ACS_LLCORNER)
class Textbox:
"""Editing widget using the interior of a window object.
Supports the following Emacs-like key bindings:
Ctrl-A Go to left edge of window.
Ctrl-B Cursor left, wrapping to previous line if appropriate.
Ctrl-D Delete character under cursor.
Ctrl-E Go to right edge (stripspaces off) or end of line
(stripspaces on).
Ctrl-F Cursor right, wrapping to next line when appropriate.
Ctrl-G Terminate, returning the window contents.
Ctrl-H Delete character backward.
Ctrl-J Terminate if the window is 1 line, otherwise move to start
of next line.
Ctrl-K If line is blank, delete it, otherwise clear to end of line.
Ctrl-L Refresh screen.
Ctrl-N Cursor down; move down one line.
Ctrl-O Insert a blank line at cursor location.
Ctrl-P Cursor up; move up one line.
Move operations do nothing if the cursor is at an edge where the
movement is not possible. The following synonyms are supported where
possible:
KEY_LEFT = Ctrl-B, KEY_RIGHT = Ctrl-F, KEY_UP = Ctrl-P,
KEY_DOWN = Ctrl-N, KEY_BACKSPACE = Ctrl-h
"""
def __init__(self, win, insert_mode=False):
self.win = win
self.insert_mode = insert_mode
self._update_max_yx()
self.stripspaces = 1
self.lastcmd = None
win.keypad(1)
def _update_max_yx(self):
maxy, maxx = self.win.getmaxyx()
self.maxy = maxy - 1
self.maxx = maxx - 1
def _decode(self, ch):
# The text of a chtype cell or input byte, decoded with the window's
# encoding. A_CHARTEXT keeps the character byte, dropping the attributes.
return bytes([ch & curses.A_CHARTEXT]).decode(self.win.encoding, 'replace')
def _char_at(self, *yx):
# The text of the cell at the given position (default: the cursor).
# instr() re-encodes it to the window's encoding; inch() cannot
# represent a non-ASCII 8-bit-locale character on a wide build.
return self.win.instr(*yx, 1).decode(self.win.encoding, 'replace')
def _cell_at(self, *yx):
# The cell at the given position (default: the cursor) as a chtype
# addch() can write back with its rendition. inch() mangles a non-ASCII
# character on a wide build, so take the byte from instr() and the
# attributes from inch().
return self.win.instr(*yx, 1)[0] | self.win.inch(*yx) & curses.A_ATTRIBUTES
def _isprint(self, cell):
# Whether a chtype cell holds a printable character; _decode() drops the
# attribute bits.
return self._decode(cell).isprintable()
def _printable_key(self, ch):
# Whether the integer keystroke is a printable character, not a key
# code. 0..255 are character bytes (decoded with the window's encoding);
# larger values are function and navigation keys.
return ch <= 0xff and self._decode(ch).isprintable()
def _end_of_line(self, y):
"""Go to the location of the first blank on the given line,
returning the index of the last non-blank character."""
self._update_max_yx()
last = self.maxx
while True:
if self._char_at(y, last) != ' ':
last = min(self.maxx, last+1)
break
elif last == 0:
break
last = last - 1
return last
def _insert_printable_char(self, ch):
self._update_max_yx()
(y, x) = self.win.getyx()
backyx = None
while True:
if self.insert_mode:
oldch = self._cell_at()
if y >= self.maxy and x >= self.maxx:
# Use insch() in the lower-right cell: addch() there would move
# the cursor out of the window, raising an error and scrolling
# a scrollable window. Pass it as text: insch() does not decode
# an int byte through the locale on a wide build.
self.win.insch(self._decode(ch), ch & curses.A_ATTRIBUTES)
break
self.win.addch(ch)
if not self.insert_mode or not self._isprint(oldch):
break
ch = oldch
(y, x) = self.win.getyx()
# Remember where to put the cursor back since we are in insert_mode
if backyx is None:
backyx = y, x
if backyx is not None:
self.win.move(*backyx)
def do_command(self, ch):
"Process a single editing command."
self._update_max_yx()
(y, x) = self.win.getyx()
self.lastcmd = ch
if self._printable_key(ch):
self._insert_printable_char(ch)
elif ch == curses.ascii.SOH: # ^a
self.win.move(y, 0)
elif ch in (curses.ascii.STX,curses.KEY_LEFT,
curses.ascii.BS,
curses.KEY_BACKSPACE,
curses.ascii.DEL):
if x > 0:
self.win.move(y, x-1)
elif y == 0:
pass
elif self.stripspaces:
self.win.move(y-1, self._end_of_line(y-1))
else:
self.win.move(y-1, self.maxx)
if ch in (curses.ascii.BS, curses.KEY_BACKSPACE, curses.ascii.DEL):
self.win.delch()
elif ch == curses.ascii.EOT: # ^d
self.win.delch()
elif ch == curses.ascii.ENQ: # ^e
if self.stripspaces:
self.win.move(y, self._end_of_line(y))
else:
self.win.move(y, self.maxx)
elif ch in (curses.ascii.ACK, curses.KEY_RIGHT): # ^f
if x < self.maxx:
self.win.move(y, x+1)
elif y == self.maxy:
pass
else:
self.win.move(y+1, 0)
elif ch == curses.ascii.BEL: # ^g
return 0
elif ch == curses.ascii.NL: # ^j
if self.maxy == 0:
return 0
elif y < self.maxy:
self.win.move(y+1, 0)
elif ch == curses.ascii.VT: # ^k
if x == 0 and self._end_of_line(y) == 0:
self.win.deleteln()
else:
# first undo the effect of self._end_of_line
self.win.move(y, x)
self.win.clrtoeol()
elif ch == curses.ascii.FF: # ^l
self.win.refresh()
elif ch in (curses.ascii.SO, curses.KEY_DOWN): # ^n
if y < self.maxy:
self.win.move(y+1, x)
if x > self._end_of_line(y+1):
self.win.move(y+1, self._end_of_line(y+1))
elif ch == curses.ascii.SI: # ^o
self.win.insertln()
elif ch in (curses.ascii.DLE, curses.KEY_UP): # ^p
if y > 0:
self.win.move(y-1, x)
if x > self._end_of_line(y-1):
self.win.move(y-1, self._end_of_line(y-1))
return 1
def gather(self):
"Collect and return the contents of the window."
result = ""
self._update_max_yx()
for y in range(self.maxy+1):
self.win.move(y, 0)
stop = self._end_of_line(y)
if stop == 0 and self.stripspaces:
continue
for x in range(self.maxx+1):
if self.stripspaces and x > stop:
break
result = result + self._char_at(y, x)
if self.maxy > 0:
result = result + "\n"
return result
def edit(self, validate=None):
"Edit in the widget window and collect the results."
while 1:
ch = self.win.getch()
if validate:
ch = validate(ch)
if not ch:
continue
if not self.do_command(ch):
break
self.win.refresh()
return self.gather()
if __name__ == '__main__':
def test_editbox(stdscr):
ncols, nlines = 9, 4
uly, ulx = 15, 20
stdscr.addstr(uly-2, ulx, "Use Ctrl-G to end editing.")
win = curses.newwin(nlines, ncols, uly, ulx)
rectangle(stdscr, uly-1, ulx-1, uly + nlines, ulx + ncols)
stdscr.refresh()
return Textbox(win).edit()
str = curses.wrapper(test_editbox)
print('Contents of text box:', repr(str))
Vendored Executable
+46
View File
@@ -0,0 +1,46 @@
#!/usr/bin/env python3
#
# $Id: ncurses.py 36559 2004-07-18 05:56:09Z tim_one $
#
# Interactive test suite for the curses module.
# This script displays various things and the user should verify whether
# they display correctly.
#
import curses
from curses import textpad
def test_textpad(stdscr, insert_mode=False):
ncols, nlines = 8, 3
uly, ulx = 3, 2
if insert_mode:
mode = 'insert mode'
else:
mode = 'overwrite mode'
stdscr.addstr(uly-3, ulx, "Use Ctrl-G to end editing (%s)." % mode)
stdscr.addstr(uly-2, ulx, "Be sure to try typing in the lower-right corner.")
win = curses.newwin(nlines, ncols, uly, ulx)
textpad.rectangle(stdscr, uly-1, ulx-1, uly + nlines, ulx + ncols)
stdscr.refresh()
box = textpad.Textbox(win, insert_mode)
contents = box.edit()
stdscr.addstr(uly+ncols+2, 0, "Text entered in the box\n")
stdscr.addstr(repr(contents))
stdscr.addstr('\n')
stdscr.addstr('Press any key')
stdscr.getch()
for i in range(3):
stdscr.move(uly+ncols+2 + i, 0)
stdscr.clrtoeol()
def main(stdscr):
stdscr.clear()
test_textpad(stdscr, False)
test_textpad(stdscr, True)
if __name__ == '__main__':
curses.wrapper(main)
+9 -20
View File
@@ -10,32 +10,21 @@ function basename:string(path:string)
return last_token;
}
probe process.mark("function__entry")
{
funcname = user_string($arg2);
if (funcname == "start") {
tracing = 1;
}
}
probe process.mark("function__entry"), process.mark("function__return")
probe @PYTHON_SYSTEMTAP_PROBE@("function__entry"),
@PYTHON_SYSTEMTAP_PROBE@("function__return")
{
filename = user_string($arg1);
funcname = user_string($arg2);
lineno = $arg3;
if (tracing) {
if (funcname == "start") {
if ($$name == "function__entry") {
tracing = 1;
} else {
tracing = 0;
}
} else if (tracing) {
printf("%d\t%s:%s:%s:%d\n", gettimeofday_us(), $$name,
basename(filename), funcname, lineno);
}
}
probe process.mark("function__return")
{
funcname = user_string($arg2);
if (funcname == "start") {
tracing = 0;
}
}
+2046
View File
File diff suppressed because it is too large. Load diff
+222
View File
@@ -0,0 +1,222 @@
import os
import unittest
from test import support
from test.support import cpython_only, import_helper
from test.support.os_helper import (TESTFN, TESTFN_NONASCII, FakePath,
create_empty_file, temp_dir, unlink)
gdbm = import_helper.import_module("dbm.gnu") # skip if not supported
filename = TESTFN
class TestGdbm(unittest.TestCase):
@staticmethod
def setUpClass():
if support.verbose:
try:
from _gdbm import _GDBM_VERSION as version
except ImportError:
pass
else:
print(f"gdbm version: {version}")
def setUp(self):
self.g = None
def tearDown(self):
if self.g is not None:
self.g.close()
unlink(filename)
@cpython_only
def test_disallow_instantiation(self):
# Ensure that the type disallows instantiation (bpo-43916)
self.g = gdbm.open(filename, 'c')
support.check_disallow_instantiation(self, type(self.g))
def test_key_methods(self):
self.g = gdbm.open(filename, 'c')
self.assertEqual(self.g.keys(), [])
self.g['a'] = 'b'
self.g['12345678910'] = '019237410982340912840198242'
self.g[b'bytes'] = b'data'
key_set = set(self.g.keys())
self.assertEqual(key_set, set([b'a', b'bytes', b'12345678910']))
self.assertIn('a', self.g)
self.assertIn(b'a', self.g)
self.assertEqual(self.g[b'bytes'], b'data')
key = self.g.firstkey()
while key:
self.assertIn(key, key_set)
key_set.remove(key)
key = self.g.nextkey(key)
# get() and setdefault() work as in the dict interface
self.assertEqual(self.g.get(b'a'), b'b')
self.assertIsNone(self.g.get(b'xxx'))
self.assertEqual(self.g.get(b'xxx', b'foo'), b'foo')
with self.assertRaises(KeyError):
self.g['xxx']
self.assertEqual(self.g.setdefault(b'xxx', b'foo'), b'foo')
self.assertEqual(self.g[b'xxx'], b'foo')
def test_error_conditions(self):
# Try to open a non-existent database.
unlink(filename)
self.assertRaises(gdbm.error, gdbm.open, filename, 'r')
# Try to access a closed database.
self.g = gdbm.open(filename, 'c')
self.g.close()
self.assertRaises(gdbm.error, lambda: self.g['a'])
# try pass an invalid open flag
self.assertRaises(gdbm.error, lambda: gdbm.open(filename, 'rx').close())
def test_flags(self):
# Test the flag parameter open() by trying all supported flag modes.
all = set(gdbm.open_flags)
# Test standard flags (presumably "crwn").
modes = all - set('fsu')
for mode in sorted(modes): # put "c" mode first
self.g = gdbm.open(filename, mode)
self.g.close()
# Test additional flags (presumably "fsu").
flags = all - set('crwn')
for mode in modes:
for flag in flags:
self.g = gdbm.open(filename, mode + flag)
self.g.close()
def test_reorganize(self):
self.g = gdbm.open(filename, 'c')
size0 = os.path.getsize(filename)
# bpo-33901: on macOS with gdbm 1.15, an empty database uses 16 MiB
# and adding an entry of 10,000 B has no effect on the file size.
# Add size0 bytes to make sure that the file size changes.
value_size = max(size0, 10000)
self.g['x'] = 'x' * value_size
size1 = os.path.getsize(filename)
self.assertGreater(size1, size0)
del self.g['x']
# 'size' is supposed to be the same even after deleting an entry.
self.assertEqual(os.path.getsize(filename), size1)
self.g.reorganize()
size2 = os.path.getsize(filename)
self.assertLess(size2, size1)
self.assertGreaterEqual(size2, size0)
def test_context_manager(self):
with gdbm.open(filename, 'c') as db:
db["gdbm context manager"] = "context manager"
with gdbm.open(filename, 'r') as db:
self.assertEqual(list(db.keys()), [b"gdbm context manager"])
with self.assertRaises(gdbm.error) as cm:
db.keys()
self.assertEqual(str(cm.exception),
"GDBM object has already been closed")
def test_bool_empty(self):
with gdbm.open(filename, 'c') as db:
self.assertFalse(bool(db))
def test_bool_not_empty(self):
with gdbm.open(filename, 'c') as db:
db['a'] = 'b'
self.assertTrue(bool(db))
def test_bool_on_closed_db_raises(self):
with gdbm.open(filename, 'c') as db:
db['a'] = 'b'
self.assertRaises(gdbm.error, bool, db)
def test_bytes(self):
with gdbm.open(filename, 'c') as db:
db[b'bytes key \xbd'] = b'bytes value \xbd'
with gdbm.open(filename, 'r') as db:
self.assertEqual(list(db.keys()), [b'bytes key \xbd'])
self.assertTrue(b'bytes key \xbd' in db)
self.assertEqual(db[b'bytes key \xbd'], b'bytes value \xbd')
def test_unicode(self):
with gdbm.open(filename, 'c') as db:
db['Unicode key \U0001f40d'] = 'Unicode value \U0001f40d'
with gdbm.open(filename, 'r') as db:
self.assertEqual(list(db.keys()), ['Unicode key \U0001f40d'.encode()])
self.assertTrue('Unicode key \U0001f40d'.encode() in db)
self.assertTrue('Unicode key \U0001f40d' in db)
self.assertEqual(db['Unicode key \U0001f40d'.encode()],
'Unicode value \U0001f40d'.encode())
self.assertEqual(db['Unicode key \U0001f40d'],
'Unicode value \U0001f40d'.encode())
def test_write_readonly_file(self):
with gdbm.open(filename, 'c') as db:
db[b'bytes key'] = b'bytes value'
with gdbm.open(filename, 'r') as db:
with self.assertRaises(gdbm.error):
del db[b'not exist key']
with self.assertRaises(gdbm.error):
del db[b'bytes key']
with self.assertRaises(gdbm.error):
db[b'not exist key'] = b'not exist value'
@unittest.skipUnless(TESTFN_NONASCII,
'requires OS support of non-ASCII encodings')
def test_nonascii_filename(self):
filename = TESTFN_NONASCII
self.addCleanup(unlink, filename)
with gdbm.open(filename, 'c') as db:
db[b'key'] = b'value'
self.assertTrue(os.path.exists(filename))
with gdbm.open(filename, 'r') as db:
self.assertEqual(list(db.keys()), [b'key'])
self.assertTrue(b'key' in db)
self.assertEqual(db[b'key'], b'value')
def test_nonexisting_file(self):
nonexisting_file = 'nonexisting-file'
with self.assertRaises(gdbm.error) as cm:
gdbm.open(nonexisting_file)
self.assertIn(nonexisting_file, str(cm.exception))
self.assertEqual(cm.exception.filename, nonexisting_file)
def test_open_with_pathlib_path(self):
gdbm.open(FakePath(filename), "c").close()
def test_open_with_bytes_path(self):
gdbm.open(os.fsencode(filename), "c").close()
def test_open_with_pathlib_bytes_path(self):
gdbm.open(FakePath(os.fsencode(filename)), "c").close()
def test_clear(self):
kvs = [('foo', 'bar'), ('1234', '5678')]
with gdbm.open(filename, 'c') as db:
for k, v in kvs:
db[k] = v
self.assertIn(k, db)
self.assertEqual(len(db), len(kvs))
db.clear()
for k, v in kvs:
self.assertNotIn(k, db)
self.assertEqual(len(db), 0)
@support.run_with_locale(
'LC_ALL',
'fr_FR.iso88591', 'ja_JP.sjis', 'zh_CN.gbk',
'fr_FR.utf8', 'en_US.utf8',
'',
)
def test_localized_error(self):
with temp_dir() as d:
create_empty_file(os.path.join(d, 'test'))
self.assertRaises(gdbm.error, gdbm.open, filename, 'r')
if __name__ == '__main__':
unittest.main()
+165
View File
@@ -0,0 +1,165 @@
from test.support import import_helper
from test.support import os_helper
import_helper.import_module("dbm.ndbm") #skip if not supported
import os
import unittest
import dbm.ndbm
from dbm.ndbm import error
class DbmTestCase(unittest.TestCase):
def setUp(self):
self.filename = os_helper.TESTFN
self.d = dbm.ndbm.open(self.filename, 'c')
self.d.close()
def tearDown(self):
for suffix in ['', '.pag', '.dir', '.db']:
os_helper.unlink(self.filename + suffix)
def test_keys(self):
self.d = dbm.ndbm.open(self.filename, 'c')
self.assertEqual(self.d.keys(), [])
self.d['a'] = 'b'
self.d[b'bytes'] = b'data'
self.d['12345678910'] = '019237410982340912840198242'
self.d.keys()
self.assertIn('a', self.d)
self.assertIn(b'a', self.d)
self.assertEqual(self.d[b'bytes'], b'data')
# get() and setdefault() work as in the dict interface
self.assertEqual(self.d.get(b'a'), b'b')
self.assertIsNone(self.d.get(b'xxx'))
self.assertEqual(self.d.get(b'xxx', b'foo'), b'foo')
with self.assertRaises(KeyError):
self.d['xxx']
self.assertEqual(self.d.setdefault(b'xxx', b'foo'), b'foo')
self.assertEqual(self.d[b'xxx'], b'foo')
self.d.close()
def test_empty_value(self):
if dbm.ndbm.library == 'Berkeley DB':
self.skipTest("Berkeley DB doesn't distinguish the empty value "
"from the absent one")
self.d = dbm.ndbm.open(self.filename, 'c')
self.assertEqual(self.d.keys(), [])
self.d['empty'] = ''
self.assertEqual(self.d.keys(), [b'empty'])
self.assertIn(b'empty', self.d)
self.assertEqual(self.d[b'empty'], b'')
self.assertEqual(self.d.get(b'empty'), b'')
self.assertEqual(self.d.setdefault(b'empty'), b'')
self.d.close()
def test_modes(self):
for mode in ['r', 'rw', 'w', 'n']:
try:
self.d = dbm.ndbm.open(self.filename, mode)
self.d.close()
except error:
self.fail()
def test_context_manager(self):
with dbm.ndbm.open(self.filename, 'c') as db:
db["ndbm context manager"] = "context manager"
with dbm.ndbm.open(self.filename, 'r') as db:
self.assertEqual(list(db.keys()), [b"ndbm context manager"])
with self.assertRaises(dbm.ndbm.error) as cm:
db.keys()
self.assertEqual(str(cm.exception),
"DBM object has already been closed")
def test_bytes(self):
with dbm.ndbm.open(self.filename, 'c') as db:
db[b'bytes key \xbd'] = b'bytes value \xbd'
with dbm.ndbm.open(self.filename, 'r') as db:
self.assertEqual(list(db.keys()), [b'bytes key \xbd'])
self.assertTrue(b'bytes key \xbd' in db)
self.assertEqual(db[b'bytes key \xbd'], b'bytes value \xbd')
def test_unicode(self):
with dbm.ndbm.open(self.filename, 'c') as db:
db['Unicode key \U0001f40d'] = 'Unicode value \U0001f40d'
with dbm.ndbm.open(self.filename, 'r') as db:
self.assertEqual(list(db.keys()), ['Unicode key \U0001f40d'.encode()])
self.assertTrue('Unicode key \U0001f40d'.encode() in db)
self.assertTrue('Unicode key \U0001f40d' in db)
self.assertEqual(db['Unicode key \U0001f40d'.encode()],
'Unicode value \U0001f40d'.encode())
self.assertEqual(db['Unicode key \U0001f40d'],
'Unicode value \U0001f40d'.encode())
def test_write_readonly_file(self):
with dbm.ndbm.open(self.filename, 'c') as db:
db[b'bytes key'] = b'bytes value'
with dbm.ndbm.open(self.filename, 'r') as db:
with self.assertRaises(error):
del db[b'not exist key']
with self.assertRaises(error):
del db[b'bytes key']
with self.assertRaises(error):
db[b'not exist key'] = b'not exist value'
@unittest.skipUnless(os_helper.TESTFN_NONASCII,
'requires OS support of non-ASCII encodings')
def test_nonascii_filename(self):
filename = os_helper.TESTFN_NONASCII
for suffix in ['', '.pag', '.dir', '.db']:
self.addCleanup(os_helper.unlink, filename + suffix)
with dbm.ndbm.open(filename, 'c') as db:
db[b'key'] = b'value'
self.assertTrue(any(os.path.exists(filename + suffix)
for suffix in ['', '.pag', '.dir', '.db']))
with dbm.ndbm.open(filename, 'r') as db:
self.assertEqual(list(db.keys()), [b'key'])
self.assertTrue(b'key' in db)
self.assertEqual(db[b'key'], b'value')
def test_nonexisting_file(self):
nonexisting_file = 'nonexisting-file'
with self.assertRaises(dbm.ndbm.error) as cm:
dbm.ndbm.open(nonexisting_file)
self.assertIn(nonexisting_file, str(cm.exception))
self.assertEqual(cm.exception.filename, nonexisting_file)
def test_open_with_pathlib_path(self):
dbm.ndbm.open(os_helper.FakePath(self.filename), "c").close()
def test_open_with_bytes_path(self):
dbm.ndbm.open(os.fsencode(self.filename), "c").close()
def test_open_with_pathlib_bytes_path(self):
dbm.ndbm.open(os_helper.FakePath(os.fsencode(self.filename)), "c").close()
def test_bool_empty(self):
with dbm.ndbm.open(self.filename, 'c') as db:
self.assertFalse(bool(db))
def test_bool_not_empty(self):
with dbm.ndbm.open(self.filename, 'c') as db:
db['a'] = 'b'
self.assertTrue(bool(db))
def test_bool_on_closed_db_raises(self):
with dbm.ndbm.open(self.filename, 'c') as db:
db['a'] = 'b'
self.assertRaises(dbm.ndbm.error, bool, db)
def test_clear(self):
kvs = [('foo', 'bar'), ('1234', '5678')]
with dbm.ndbm.open(self.filename, 'c') as db:
for k, v in kvs:
db[k] = v
self.assertIn(k, db)
self.assertEqual(len(db), len(kvs))
db.clear()
for k, v in kvs:
self.assertNotIn(k, db)
self.assertEqual(len(db), 0)
if __name__ == '__main__':
unittest.main()
+362 -53
View File
@@ -1,24 +1,57 @@
import dis
import os.path
import re
import shlex
import signal
import subprocess
import sys
import sysconfig
import tempfile
import types
import unittest
from test import support
from test.support import findfile, MS_WINDOWS
from test.support import os_helper
if not support.has_subprocess_support:
raise unittest.SkipTest("test module requires subprocess")
if not sysconfig.get_config_var('WITH_DTRACE'):
raise unittest.SkipTest(
"CPython must be configured with the --with-dtrace option."
)
def abspath(filename):
return os.path.abspath(findfile(filename, subdir="dtracedata"))
def get_probe_binary():
binary = sys.executable
if sysconfig.get_config_var("Py_ENABLE_SHARED"):
lib_dir = sysconfig.get_config_var("LIBDIR")
if not lib_dir or sysconfig.is_python_build():
lib_dir = os.path.abspath(os.path.dirname(sys.executable))
lib_names = []
for name in (
sysconfig.get_config_var("INSTSONAME"),
sysconfig.get_config_var("LDLIBRARY"),
):
if name and name not in lib_names:
lib_names.append(name)
if lib_dir:
for name in lib_names:
libpython_path = os.path.join(lib_dir, name)
if os.path.exists(libpython_path):
binary = libpython_path
break
return binary
def normalize_trace_output(output):
"""Normalize DTrace output for comparison.
@@ -33,27 +66,93 @@ def normalize_trace_output(output):
result = [
row.split("\t")
for row in output.splitlines()
if row and not row.startswith('#')
if row and not row.startswith('#') and not row.startswith('@')
]
result.sort(key=lambda row: int(row[0]))
result = [row[1] for row in result]
return "\n".join(result)
# Normalize paths to basenames (bpftrace outputs full paths)
normalized = []
for line in result:
# Replace full paths with just the filename
line = re.sub(r'/[^:]+/([^/:]+\.py)', r'\1', line)
normalized.append(line)
return "\n".join(normalized)
except (IndexError, ValueError):
raise AssertionError(
"tracer produced unparsable output:\n{}".format(output)
)
USE_PROCESS_GROUP = (hasattr(os, "setsid") and hasattr(os, "killpg"))
TERMINATE_TIMEOUT = 10
def create_process_group(*args, **kwargs):
if USE_PROCESS_GROUP:
kwargs['start_new_session'] = True
return subprocess.Popen(*args, **kwargs)
def terminate_process_group(proc):
if USE_PROCESS_GROUP:
try:
os.killpg(proc.pid, signal.SIGTERM)
except ProcessLookupError:
pass
else:
proc.terminate()
try:
proc.communicate(timeout=TERMINATE_TIMEOUT)
except subprocess.TimeoutExpired:
if USE_PROCESS_GROUP:
try:
os.killpg(proc.pid, signal.SIGKILL)
except ProcessLookupError:
pass
else:
proc.kill()
proc.communicate(timeout=TERMINATE_TIMEOUT) # Clean up
def run_readelf(cmd):
# Force the C locale to disable localization.
env = dict(os.environ, LC_ALL="C")
try:
proc = subprocess.Popen(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
env=env,
)
except OSError:
raise unittest.SkipTest("Couldn't find readelf on the path")
with proc:
stdout, stderr = proc.communicate()
if proc.returncode:
raise AssertionError(
f"Command {shlex.join(cmd)!r} failed "
f"with exit code {proc.returncode}: "
f"stdout={stdout!r} stderr={stderr!r}"
)
return stdout
class TraceBackend:
EXTENSION = None
COMMAND = None
COMMAND_ARGS = []
def run_case(self, name, optimize_python=None):
actual_output = normalize_trace_output(self.trace_python(
script_file=abspath(name + self.EXTENSION),
python_file=abspath(name + ".py"),
optimize_python=optimize_python))
try:
actual_output = normalize_trace_output(self.trace_python(
script_file=abspath(name + self.EXTENSION),
python_file=abspath(name + ".py"),
optimize_python=optimize_python))
except subprocess.TimeoutExpired:
raise AssertionError(f"{self.COMMAND[0]} timed out")
with open(abspath(name + self.EXTENSION + ".expected")) as f:
expected_output = f.read().rstrip()
@@ -66,25 +165,42 @@ class TraceBackend:
command += ["-c", subcommand]
return command
def trace(self, script_file, subcommand=None):
def trace(self, script_file, subcommand=None, *, timeout=None,
check_returncode=False):
command = self.generate_trace_command(script_file, subcommand)
stdout, _ = subprocess.Popen(command,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
universal_newlines=True).communicate()
proc = create_process_group(command,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
universal_newlines=True)
try:
stdout, _ = proc.communicate(timeout=timeout)
except subprocess.TimeoutExpired:
terminate_process_group(proc)
raise
if check_returncode and proc.returncode:
raise AssertionError(
f"Command {shlex.join(command)!r} failed "
f"with exit code {proc.returncode}: output={stdout!r}"
)
return stdout
def trace_python(self, script_file, python_file, optimize_python=None):
python_flags = []
if optimize_python:
python_flags.extend(["-O"] * optimize_python)
subcommand = " ".join([sys.executable] + python_flags + [python_file])
return self.trace(script_file, subcommand)
subcommand = shlex.join([sys.executable] + python_flags + [python_file])
return self.trace(script_file, subcommand, timeout=60,
check_returncode=True)
def assert_usable(self):
try:
output = self.trace(abspath("assert_usable" + self.EXTENSION))
output = self.trace(abspath("assert_usable" + self.EXTENSION),
timeout=10)
output = output.strip()
except subprocess.TimeoutExpired:
raise unittest.SkipTest(
f"{self.COMMAND[0]} timed out during usability check"
)
except (FileNotFoundError, NotADirectoryError, PermissionError) as fnfe:
output = str(fnfe)
if output != "probe: success":
@@ -96,11 +212,223 @@ class TraceBackend:
class DTraceBackend(TraceBackend):
EXTENSION = ".d"
COMMAND = ["dtrace", "-q", "-s"]
if sys.platform == "sunos5":
COMMAND.insert(2, "-Z")
class SystemTapBackend(TraceBackend):
EXTENSION = ".stp"
COMMAND = ["stap", "-g"]
PROBE_PLACEHOLDER = "@PYTHON_SYSTEMTAP_PROBE@"
@staticmethod
def quote_systemtap_string(value):
return value.replace("\\", "\\\\").replace('"', '\\"')
def python_probe(self):
executable = self.quote_systemtap_string(sys.executable)
probe_binary = get_probe_binary()
if probe_binary == sys.executable:
return f'process("{executable}").mark'
# Python built with --enable-shared
probe_binary = self.quote_systemtap_string(probe_binary)
return f'process("{executable}").library("{probe_binary}").mark'
def render_script(self, filename):
with open(filename) as fp:
script = fp.read()
return script.replace(self.PROBE_PLACEHOLDER, self.python_probe())
def trace(self, script_file, subcommand=None, *, timeout=None,
check_returncode=False):
with tempfile.NamedTemporaryFile(
mode="w", encoding="utf-8", suffix=self.EXTENSION, delete=False
) as script:
script.write(self.render_script(script_file))
generated_script_file = script.name
try:
return super().trace(
generated_script_file,
subcommand,
timeout=timeout,
check_returncode=check_returncode,
)
finally:
os_helper.unlink(generated_script_file)
class BPFTraceBackend(TraceBackend):
EXTENSION = ".bt"
COMMAND = ["bpftrace"]
# Inline bpftrace programs for each test case
PROGRAMS = {
"call_stack": """
usdt:{python}:python:function__entry {{
printf("%lld\\tfunction__entry:%s:%s:%d\\n",
nsecs, str(arg0), str(arg1), arg2);
}}
usdt:{python}:python:function__return {{
printf("%lld\\tfunction__return:%s:%s:%d\\n",
nsecs, str(arg0), str(arg1), arg2);
}}
""",
"gc": """
usdt:{python}:python:function__entry {{
if (str(arg1) == "start") {{ @tracing = 1; }}
}}
usdt:{python}:python:function__return {{
if (str(arg1) == "start") {{ @tracing = 0; }}
}}
usdt:{python}:python:gc__start {{
if (@tracing) {{
printf("%lld\\tgc__start:%d\\n", nsecs, arg0);
}}
}}
usdt:{python}:python:gc__done {{
if (@tracing) {{
printf("%lld\\tgc__done:%lld\\n", nsecs, arg0);
}}
}}
END {{ clear(@tracing); }}
""",
}
# Which test scripts to filter by filename (None = use @tracing flag)
FILTER_BY_FILENAME = {"call_stack": "call_stack.py"}
@staticmethod
def _filter_probe_rows(output):
return "\n".join(
line for line in output.splitlines()
if line.partition("\t")[0].isdigit()
)
# Expected outputs for each test case
# Note: bpftrace captures <module> entry/return and may have slight timing
# differences compared to SystemTap due to probe firing order
EXPECTED = {
"call_stack": """function__entry:call_stack.py:<module>:0
function__entry:call_stack.py:start:23
function__entry:call_stack.py:function_1:1
function__entry:call_stack.py:function_3:9
function__return:call_stack.py:function_3:10
function__return:call_stack.py:function_1:2
function__entry:call_stack.py:function_2:5
function__entry:call_stack.py:function_1:1
function__entry:call_stack.py:function_3:9
function__return:call_stack.py:function_3:10
function__return:call_stack.py:function_1:2
function__return:call_stack.py:function_2:6
function__entry:call_stack.py:function_3:9
function__return:call_stack.py:function_3:10
function__entry:call_stack.py:function_4:13
function__return:call_stack.py:function_4:14
function__entry:call_stack.py:function_5:18
function__return:call_stack.py:function_5:21
function__return:call_stack.py:start:28
function__return:call_stack.py:<module>:30""",
"gc": """gc__start:0
gc__done:0
gc__start:1
gc__done:0
gc__start:2
gc__done:0
gc__start:2
gc__done:1""",
}
def run_case(self, name, optimize_python=None):
if name not in self.PROGRAMS:
raise unittest.SkipTest(f"No bpftrace program for {name}")
python_file = abspath(name + ".py")
python_flags = []
if optimize_python:
python_flags.extend(["-O"] * optimize_python)
subcommand = [sys.executable] + python_flags + [python_file]
program = self.PROGRAMS[name].format(python=get_probe_binary())
try:
proc = create_process_group(
["bpftrace", "-e", program, "-c", shlex.join(subcommand)],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
universal_newlines=True,
)
stdout, stderr = proc.communicate(timeout=60)
except subprocess.TimeoutExpired:
terminate_process_group(proc)
raise AssertionError("bpftrace timed out")
except (FileNotFoundError, PermissionError) as e:
raise unittest.SkipTest(f"bpftrace not available: {e}")
if proc.returncode != 0:
raise AssertionError(
f"bpftrace failed with code {proc.returncode}:\n{stderr}"
)
stdout = self._filter_probe_rows(stdout)
# Filter output by filename if specified (bpftrace captures everything)
if name in self.FILTER_BY_FILENAME:
filter_filename = self.FILTER_BY_FILENAME[name]
filtered_lines = [
line for line in stdout.splitlines()
if filter_filename in line
]
stdout = "\n".join(filtered_lines)
actual_output = normalize_trace_output(stdout)
expected_output = self.EXPECTED[name].strip()
return (expected_output, actual_output)
def assert_usable(self):
# Check if bpftrace is available and can attach to USDT probes
program = f'usdt:{get_probe_binary()}:python:function__entry {{ printf("probe: success\\n"); exit(); }}'
try:
proc = create_process_group(
["bpftrace", "-e", program, "-c",
shlex.join([sys.executable, "-c", "pass"])],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
universal_newlines=True,
)
stdout, stderr = proc.communicate(timeout=10)
except subprocess.TimeoutExpired:
terminate_process_group(proc)
raise unittest.SkipTest("bpftrace timed out during usability check")
except OSError as e:
raise unittest.SkipTest(f"bpftrace not available: {e}")
# Check for permission errors (bpftrace usually requires root)
if proc.returncode != 0:
raise unittest.SkipTest(
f"bpftrace(1) failed with code {proc.returncode}: {stderr}"
)
if "probe: success" not in stdout:
raise unittest.SkipTest(
f"bpftrace(1) failed: stdout={stdout!r} stderr={stderr!r}"
)
class BPFTraceOutputTests(unittest.TestCase):
def test_filter_probe_rows_ignores_warnings(self):
output = """stdin:1-19: WARNING: found external warnings
HINT: include/vmlinux.h:1439:3: warning: declaration does not declare anything
4623214882928\tgc__start:0
4623214885575\tgc__done:0
"""
self.assertEqual(
BPFTraceBackend._filter_probe_rows(output),
"4623214882928\tgc__start:0\n4623214885575\tgc__done:0",
)
@unittest.skipIf(MS_WINDOWS, "Tests not compliant with trace on Windows.")
@@ -127,7 +455,8 @@ class TraceTests:
def test_verify_call_opcodes(self):
"""Ensure our call stack test hits all function call opcodes"""
opcodes = set(["CALL_FUNCTION", "CALL_FUNCTION_EX", "CALL_FUNCTION_KW"])
# Modern Python uses CALL, CALL_KW, and CALL_FUNCTION_EX
opcodes = set(["CALL", "CALL_FUNCTION_EX", "CALL_KW"])
with open(abspath("call_stack.py")) as f:
code_string = f.read()
@@ -152,9 +481,6 @@ class TraceTests:
def test_gc(self):
self.run_case("gc")
def test_line(self):
self.run_case("line")
class DTraceNormalTests(TraceTests, unittest.TestCase):
backend = DTraceBackend()
@@ -175,38 +501,28 @@ class SystemTapOptimizedTests(TraceTests, unittest.TestCase):
backend = SystemTapBackend()
optimize_python = 2
class BPFTraceNormalTests(TraceTests, unittest.TestCase):
backend = BPFTraceBackend()
optimize_python = 0
class BPFTraceOptimizedTests(TraceTests, unittest.TestCase):
backend = BPFTraceBackend()
optimize_python = 2
class CheckDtraceProbes(unittest.TestCase):
@classmethod
def setUpClass(cls):
if sysconfig.get_config_var('WITH_DTRACE'):
readelf_major_version, readelf_minor_version = cls.get_readelf_version()
if support.verbose:
print(f"readelf version: {readelf_major_version}.{readelf_minor_version}")
else:
raise unittest.SkipTest("CPython must be configured with the --with-dtrace option.")
readelf_major_version, readelf_minor_version = cls.get_readelf_version()
if support.verbose:
print(f"readelf version: {readelf_major_version}.{readelf_minor_version}")
@staticmethod
def get_readelf_version():
try:
cmd = ["readelf", "--version"]
proc = subprocess.Popen(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
universal_newlines=True,
)
with proc:
version, stderr = proc.communicate()
if proc.returncode:
raise Exception(
f"Command {' '.join(cmd)!r} failed "
f"with exit code {proc.returncode}: "
f"stdout={version!r} stderr={stderr!r}"
)
except OSError:
raise unittest.SkipTest("Couldn't find readelf on the path")
version = run_readelf(["readelf", "--version"])
# Regex to parse:
# 'GNU readelf (GNU Binutils) 2.40.0\n' -> 2.40
@@ -217,14 +533,7 @@ class CheckDtraceProbes(unittest.TestCase):
return int(match.group(1)), int(match.group(2))
def get_readelf_output(self):
command = ["readelf", "-n", sys.executable]
stdout, _ = subprocess.Popen(
command,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
universal_newlines=True,
).communicate()
return stdout
return run_readelf(["readelf", "-n", get_probe_binary()])
def test_check_probes(self):
readelf_output = self.get_readelf_output()
@@ -235,6 +544,8 @@ class CheckDtraceProbes(unittest.TestCase):
"Name: audit",
"Name: gc__start",
"Name: gc__done",
"Name: function__entry",
"Name: function__return",
]
for probe_name in available_probe_names:
@@ -247,8 +558,6 @@ class CheckDtraceProbes(unittest.TestCase):
# Missing probes will be added in the future.
missing_probe_names = [
"Name: function__entry",
"Name: function__return",
"Name: line",
]
+838
View File
@@ -0,0 +1,838 @@
import unittest
import subprocess
import sys
import os
from test import support
from test.support import import_helper
from test.support import os_helper
# Skip this test if the _tkinter module wasn't built.
_tkinter = import_helper.import_module('_tkinter')
import tkinter
from tkinter import Tcl
from _tkinter import TclError
try:
from _testcapi import INT_MAX, PY_SSIZE_T_MAX
except ImportError:
INT_MAX = PY_SSIZE_T_MAX = sys.maxsize
tcl_version = tuple(map(int, _tkinter.TCL_VERSION.split('.')))
class TkinterTest(unittest.TestCase):
def testFlattenLen(self):
# Object without length.
self.assertRaises(TypeError, _tkinter._flatten, True)
# Object with length, but not sequence.
self.assertRaises(TypeError, _tkinter._flatten, {})
# Sequence or set, but not tuple or list.
# (issue44608: there were leaks in the following cases)
self.assertRaises(TypeError, _tkinter._flatten, 'string')
self.assertRaises(TypeError, _tkinter._flatten, {'set'})
class TclTest(unittest.TestCase):
def setUp(self):
self.interp = Tcl()
self.wantobjects = self.interp.tk.wantobjects()
def passValue(self, value):
return self.interp.call('set', '_', value)
def testEval(self):
tcl = self.interp
tcl.eval('set a 1')
self.assertEqual(tcl.eval('set a'),'1')
def test_eval_null_in_result(self):
tcl = self.interp
self.assertEqual(tcl.eval('set a "a\\0b"'), 'a\x00b')
def test_eval_surrogates_in_result(self):
tcl = self.interp
result = tcl.eval(r'set a "<\ud83d\udcbb>"')
if sys.platform == 'win32' and tcl_version >= (9, 0):
self.assertEqual('<\ud83d\udcbb>', result)
else:
self.assertEqual('<\U0001f4bb>', result)
def testEvalException(self):
tcl = self.interp
self.assertRaises(TclError,tcl.eval,'set a')
def testEvalException2(self):
tcl = self.interp
self.assertRaises(TclError,tcl.eval,'this is wrong')
def test_eval_returns_tcl_obj(self):
tcl = self.interp.tk
tcl.eval(r'set a "\u20ac \ud83d\udcbb \0 \udcab"; regexp -about $a')
a = tcl.eval('set a')
expected = '\u20ac \U0001f4bb \0 \udced\udcb2\udcab'
self.assertEqual(a, expected)
def testCall(self):
tcl = self.interp
tcl.call('set','a','1')
self.assertEqual(tcl.call('set','a'),'1')
def test_call_passing_null(self):
tcl = self.interp
tcl.call('set', 'a', 'a\0b') # ASCII-only
self.assertEqual(tcl.getvar('a'), 'a\x00b')
self.assertEqual(tcl.call('set', 'a'), 'a\x00b')
self.assertEqual(tcl.eval('set a'), 'a\x00b')
tcl.call('set', 'a', '\u20ac\0') # non-ASCII
self.assertEqual(tcl.getvar('a'), '\u20ac\x00')
self.assertEqual(tcl.call('set', 'a'), '\u20ac\x00')
self.assertEqual(tcl.eval('set a'), '\u20ac\x00')
def testCallException(self):
tcl = self.interp
self.assertRaises(TclError,tcl.call,'set','a')
def testCallException2(self):
tcl = self.interp
self.assertRaises(TclError,tcl.call,'this','is','wrong')
def test_call_returns_tcl_obj(self):
tcl = self.interp.tk
tcl.eval(r'set a "\u20ac \ud83d\udcbb \0 \udcab"; regexp -about $a')
a = tcl.call('set', 'a')
expected = '\u20ac \U0001f4bb \0 \udced\udcb2\udcab'
if self.wantobjects:
self.assertEqual(str(a), expected)
self.assertEqual(a.string, expected)
self.assertEqual(a.typename, 'regexp')
else:
self.assertEqual(a, expected)
def testSetVar(self):
tcl = self.interp
tcl.setvar('a','1')
self.assertEqual(tcl.eval('set a'),'1')
def test_setvar_passing_null(self):
tcl = self.interp
tcl.setvar('a', 'a\0b') # ASCII-only
self.assertEqual(tcl.getvar('a'), 'a\x00b')
self.assertEqual(tcl.call('set', 'a'), 'a\x00b')
self.assertEqual(tcl.eval('set a'), 'a\x00b')
tcl.setvar('a', '\u20ac\0') # non-ASCII
self.assertEqual(tcl.getvar('a'), '\u20ac\x00')
self.assertEqual(tcl.call('set', 'a'), '\u20ac\x00')
self.assertEqual(tcl.eval('set a'), '\u20ac\x00')
def testSetVarArray(self):
tcl = self.interp
tcl.setvar('a(1)','1')
self.assertEqual(tcl.eval('set a(1)'),'1')
def testGetVar(self):
tcl = self.interp
tcl.eval('set a 1')
self.assertEqual(tcl.getvar('a'),'1')
def testGetVarArray(self):
tcl = self.interp
tcl.eval('set a(1) 1')
self.assertEqual(tcl.getvar('a(1)'),'1')
def testGetVarException(self):
tcl = self.interp
self.assertRaises(TclError,tcl.getvar,'a')
def testGetVarArrayException(self):
tcl = self.interp
self.assertRaises(TclError,tcl.getvar,'a(1)')
def test_getvar_returns_tcl_obj(self):
tcl = self.interp.tk
tcl.eval(r'set a "\u20ac \ud83d\udcbb \0 \udcab"; regexp -about $a')
a = tcl.getvar('a')
expected = '\u20ac \U0001f4bb \0 \udced\udcb2\udcab'
if self.wantobjects:
self.assertEqual(str(a), expected)
self.assertEqual(a.string, expected)
self.assertEqual(a.typename, 'regexp')
else:
self.assertEqual(a, expected)
def testUnsetVar(self):
tcl = self.interp
tcl.setvar('a',1)
self.assertEqual(tcl.eval('info exists a'),'1')
tcl.unsetvar('a')
self.assertEqual(tcl.eval('info exists a'),'0')
def testUnsetVarArray(self):
tcl = self.interp
tcl.setvar('a(1)',1)
tcl.setvar('a(2)',2)
self.assertEqual(tcl.eval('info exists a(1)'),'1')
self.assertEqual(tcl.eval('info exists a(2)'),'1')
tcl.unsetvar('a(1)')
self.assertEqual(tcl.eval('info exists a(1)'),'0')
self.assertEqual(tcl.eval('info exists a(2)'),'1')
def testUnsetVarException(self):
tcl = self.interp
self.assertRaises(TclError,tcl.unsetvar,'a')
def get_integers(self):
return (0, 1, -1,
2**31-1, -2**31, 2**31, -2**31-1,
2**63-1, -2**63, 2**63, -2**63-1,
2**1000, -2**1000)
def test_getint(self):
tcl = self.interp.tk
for i in self.get_integers():
self.assertEqual(tcl.getint(' %d ' % i), i)
self.assertEqual(tcl.getint(' %#o ' % i), i)
# Numbers starting with 0 are parsed as decimal in Tcl 9.0
# and as octal in older versions.
self.assertEqual(tcl.getint((' %#o ' % i).replace('o', '')),
i if tcl_version < (9, 0) else int('%o' % i))
self.assertEqual(tcl.getint(' %#x ' % i), i)
self.assertEqual(tcl.getint(42), 42)
self.assertRaises(TypeError, tcl.getint)
self.assertRaises(TypeError, tcl.getint, '42', '10')
self.assertRaises(TypeError, tcl.getint, b'42')
self.assertRaises(TypeError, tcl.getint, 42.0)
self.assertRaises(TclError, tcl.getint, 'a')
self.assertRaises((TypeError, ValueError, TclError),
tcl.getint, '42\0')
self.assertRaises((UnicodeEncodeError, ValueError, TclError),
tcl.getint, '42\ud800')
def test_getdouble(self):
tcl = self.interp.tk
self.assertEqual(tcl.getdouble(' 42 '), 42.0)
self.assertEqual(tcl.getdouble(' 42.5 '), 42.5)
self.assertEqual(tcl.getdouble(42.5), 42.5)
self.assertEqual(tcl.getdouble(42), 42.0)
self.assertRaises(TypeError, tcl.getdouble)
self.assertRaises(TypeError, tcl.getdouble, '42.5', '10')
self.assertRaises(TypeError, tcl.getdouble, b'42.5')
self.assertRaises(TclError, tcl.getdouble, 'a')
self.assertRaises((TypeError, ValueError, TclError),
tcl.getdouble, '42.5\0')
self.assertRaises((UnicodeEncodeError, ValueError, TclError),
tcl.getdouble, '42.5\ud800')
def test_getboolean(self):
tcl = self.interp.tk
self.assertIs(tcl.getboolean('on'), True)
self.assertIs(tcl.getboolean('1'), True)
self.assertIs(tcl.getboolean(42), True)
self.assertIs(tcl.getboolean(0), False)
self.assertRaises(TypeError, tcl.getboolean)
self.assertRaises(TypeError, tcl.getboolean, 'on', '1')
self.assertRaises(TypeError, tcl.getboolean, b'on')
self.assertRaises(TypeError, tcl.getboolean, 1.0)
self.assertRaises(TclError, tcl.getboolean, 'a')
self.assertRaises((TypeError, ValueError, TclError),
tcl.getboolean, 'on\0')
self.assertRaises((UnicodeEncodeError, ValueError, TclError),
tcl.getboolean, 'on\ud800')
def testEvalFile(self):
tcl = self.interp
filename = os_helper.TESTFN_ASCII
self.addCleanup(os_helper.unlink, filename)
with open(filename, 'w') as f:
f.write("""set a 1
set b 2
set c [ expr $a + $b ]
""")
tcl.evalfile(filename)
self.assertEqual(tcl.eval('set a'),'1')
self.assertEqual(tcl.eval('set b'),'2')
self.assertEqual(tcl.eval('set c'),'3')
def test_evalfile_null_in_result(self):
tcl = self.interp
filename = os_helper.TESTFN_ASCII
self.addCleanup(os_helper.unlink, filename)
with open(filename, 'w') as f:
f.write("""
set a "a\0b"
set b "a\\0b"
""")
tcl.evalfile(filename)
self.assertEqual(tcl.eval('set a'), 'a\x00b')
self.assertEqual(tcl.eval('set b'), 'a\x00b')
def test_evalfile_surrogates_in_result(self):
tcl = self.interp
encoding = tcl.call('encoding', 'system')
self.addCleanup(tcl.call, 'encoding', 'system', encoding)
tcl.call('encoding', 'system', 'utf-8')
filename = os_helper.TESTFN_ASCII
self.addCleanup(os_helper.unlink, filename)
with open(filename, 'wb') as f:
f.write(b"""
set a "<\xed\xa0\xbd\xed\xb2\xbb>"
""")
if tcl_version >= (9, 0):
self.assertRaises(TclError, tcl.evalfile, filename)
else:
tcl.evalfile(filename)
self.assertEqual(tcl.eval('set a'), '<\U0001f4bb>')
with open(filename, 'wb') as f:
f.write(b"""
set b "<\\ud83d\\udcbb>"
""")
tcl.evalfile(filename)
result = tcl.eval('set b')
if sys.platform == 'win32' and tcl_version >= (9, 0):
self.assertEqual('<\ud83d\udcbb>', result)
else:
self.assertEqual('<\U0001f4bb>', result)
def testEvalFileException(self):
tcl = self.interp
filename = "doesnotexists"
try:
os.remove(filename)
except Exception as e:
pass
self.assertRaises(TclError,tcl.evalfile,filename)
def testPackageRequireException(self):
tcl = self.interp
self.assertRaises(TclError,tcl.eval,'package require DNE')
@unittest.skipUnless(sys.platform == 'win32', 'Requires Windows')
def testLoadWithUNC(self):
# Build a UNC path from the regular path.
# Something like
# \\%COMPUTERNAME%\c$\python27\python.exe
fullname = os.path.abspath(sys.executable)
if fullname[1] != ':':
raise unittest.SkipTest('Absolute path should have drive part')
unc_name = r'\\%s\%s$\%s' % (os.environ['COMPUTERNAME'],
fullname[0],
fullname[3:])
if not os.path.exists(unc_name):
raise unittest.SkipTest('Cannot connect to UNC Path')
with os_helper.EnvironmentVarGuard() as env:
env.unset("TCL_LIBRARY")
stdout = subprocess.check_output(
[unc_name, '-c', 'import tkinter; print(tkinter)'])
self.assertIn(b'tkinter', stdout)
def test_exprstring(self):
tcl = self.interp
tcl.call('set', 'a', 3)
tcl.call('set', 'b', 6)
def check(expr, expected):
result = tcl.exprstring(expr)
self.assertEqual(result, expected)
self.assertIsInstance(result, str)
self.assertRaises(TypeError, tcl.exprstring)
self.assertRaises(TypeError, tcl.exprstring, '8.2', '+6')
self.assertRaises(TypeError, tcl.exprstring, b'8.2 + 6')
self.assertRaises(TclError, tcl.exprstring, 'spam')
check('', '0')
check('8.2 + 6', '14.2')
check('3.1 + $a', '6.1')
check('2 + "$a.$b"', '5.6')
check('4*[llength "6 2"]', '8')
check('{word one} < "word $a"', '0')
check('4*2 < 7', '0')
check('hypot($a, 4)', '5.0')
check('5 / 4', '1')
check('5 / 4.0', '1.25')
check('5 / ( [string length "abcd"] + 0.0 )', '1.25')
check('20.0/5.0', '4.0')
check('"0x03" > "2"', '1')
check('[string length "a\xbd\u20ac"]', '3')
check(r'[string length "a\xbd\u20ac"]', '3')
check('"abc"', 'abc')
check('"a\xbd\u20ac"', 'a\xbd\u20ac')
check(r'"a\xbd\u20ac"', 'a\xbd\u20ac')
check(r'"a\0b"', 'a\x00b')
check('2**64', str(2**64))
def test_exprdouble(self):
tcl = self.interp
tcl.call('set', 'a', 3)
tcl.call('set', 'b', 6)
def check(expr, expected):
result = tcl.exprdouble(expr)
self.assertEqual(result, expected)
self.assertIsInstance(result, float)
self.assertRaises(TypeError, tcl.exprdouble)
self.assertRaises(TypeError, tcl.exprdouble, '8.2', '+6')
self.assertRaises(TypeError, tcl.exprdouble, b'8.2 + 6')
self.assertRaises(TclError, tcl.exprdouble, 'spam')
check('', 0.0)
check('8.2 + 6', 14.2)
check('3.1 + $a', 6.1)
check('2 + "$a.$b"', 5.6)
check('4*[llength "6 2"]', 8.0)
check('{word one} < "word $a"', 0.0)
check('4*2 < 7', 0.0)
check('hypot($a, 4)', 5.0)
check('5 / 4', 1.0)
check('5 / 4.0', 1.25)
check('5 / ( [string length "abcd"] + 0.0 )', 1.25)
check('20.0/5.0', 4.0)
check('"0x03" > "2"', 1.0)
check('[string length "a\xbd\u20ac"]', 3.0)
check(r'[string length "a\xbd\u20ac"]', 3.0)
self.assertRaises(TclError, tcl.exprdouble, '"abc"')
check('2**64', float(2**64))
def test_exprlong(self):
tcl = self.interp
tcl.call('set', 'a', 3)
tcl.call('set', 'b', 6)
def check(expr, expected):
result = tcl.exprlong(expr)
self.assertEqual(result, expected)
self.assertIsInstance(result, int)
self.assertRaises(TypeError, tcl.exprlong)
self.assertRaises(TypeError, tcl.exprlong, '8.2', '+6')
self.assertRaises(TypeError, tcl.exprlong, b'8.2 + 6')
self.assertRaises(TclError, tcl.exprlong, 'spam')
check('', 0)
check('8.2 + 6', 14)
check('3.1 + $a', 6)
check('2 + "$a.$b"', 5)
check('4*[llength "6 2"]', 8)
check('{word one} < "word $a"', 0)
check('4*2 < 7', 0)
check('hypot($a, 4)', 5)
check('5 / 4', 1)
check('5 / 4.0', 1)
check('5 / ( [string length "abcd"] + 0.0 )', 1)
check('20.0/5.0', 4)
check('"0x03" > "2"', 1)
check('[string length "a\xbd\u20ac"]', 3)
check(r'[string length "a\xbd\u20ac"]', 3)
self.assertRaises(TclError, tcl.exprlong, '"abc"')
self.assertRaises(TclError, tcl.exprlong, '2**64')
def test_exprboolean(self):
tcl = self.interp
tcl.call('set', 'a', 3)
tcl.call('set', 'b', 6)
def check(expr, expected):
result = tcl.exprboolean(expr)
self.assertEqual(result, expected)
self.assertIsInstance(result, int)
self.assertNotIsInstance(result, bool)
self.assertRaises(TypeError, tcl.exprboolean)
self.assertRaises(TypeError, tcl.exprboolean, '8.2', '+6')
self.assertRaises(TypeError, tcl.exprboolean, b'8.2 + 6')
self.assertRaises(TclError, tcl.exprboolean, 'spam')
check('', False)
for value in ('0', 'false', 'no', 'off'):
check(value, False)
check('"%s"' % value, False)
check('{%s}' % value, False)
for value in ('1', 'true', 'yes', 'on'):
check(value, True)
check('"%s"' % value, True)
check('{%s}' % value, True)
check('8.2 + 6', True)
check('3.1 + $a', True)
check('2 + "$a.$b"', True)
check('4*[llength "6 2"]', True)
check('{word one} < "word $a"', False)
check('4*2 < 7', False)
check('hypot($a, 4)', True)
check('5 / 4', True)
check('5 / 4.0', True)
check('5 / ( [string length "abcd"] + 0.0 )', True)
check('20.0/5.0', True)
check('"0x03" > "2"', True)
check('[string length "a\xbd\u20ac"]', True)
check(r'[string length "a\xbd\u20ac"]', True)
self.assertRaises(TclError, tcl.exprboolean, '"abc"')
check('2**64', True)
def test_booleans(self):
tcl = self.interp
def check(expr, expected):
result = tcl.call('expr', expr)
if tcl.wantobjects():
self.assertEqual(result, expected)
self.assertIsInstance(result, int)
else:
self.assertIn(result, (expr, str(int(expected))))
self.assertIsInstance(result, str)
check('true', True)
check('yes', True)
check('on', True)
check('false', False)
check('no', False)
check('off', False)
check('1 < 2', True)
check('1 > 2', False)
def test_expr_bignum(self):
tcl = self.interp
for i in self.get_integers():
result = tcl.call('expr', str(i))
if self.wantobjects:
self.assertEqual(result, i)
self.assertIsInstance(result, int)
else:
self.assertEqual(result, str(i))
self.assertIsInstance(result, str)
def test_passing_values(self):
passValue = self.passValue
self.assertEqual(passValue(True), True if self.wantobjects else '1')
self.assertEqual(passValue(False), False if self.wantobjects else '0')
self.assertEqual(passValue('string'), 'string')
self.assertEqual(passValue('string\u20ac'), 'string\u20ac')
self.assertEqual(passValue('string\U0001f4bb'), 'string\U0001f4bb')
self.assertEqual(passValue('str\x00ing'), 'str\x00ing')
self.assertEqual(passValue('str\x00ing\xbd'), 'str\x00ing\xbd')
self.assertEqual(passValue('str\x00ing\u20ac'), 'str\x00ing\u20ac')
self.assertEqual(passValue('str\x00ing\U0001f4bb'),
'str\x00ing\U0001f4bb')
if sys.platform != 'win32':
self.assertEqual(passValue('<\udce2\udc82\udcac>'),
'<\u20ac>')
self.assertEqual(passValue('<\udced\udca0\udcbd\udced\udcb2\udcbb>'),
'<\U0001f4bb>')
self.assertEqual(passValue(b'str\x00ing'),
b'str\x00ing' if self.wantobjects else 'str\x00ing')
self.assertEqual(passValue(b'str\xc0\x80ing'),
b'str\xc0\x80ing' if self.wantobjects else 'str\xc0\x80ing')
self.assertEqual(passValue(b'str\xbding'),
b'str\xbding' if self.wantobjects else 'str\xbding')
for i in self.get_integers():
self.assertEqual(passValue(i), i if self.wantobjects else str(i))
for f in (0.0, 1.0, -1.0, 1/3,
sys.float_info.min, sys.float_info.max,
-sys.float_info.min, -sys.float_info.max):
if self.wantobjects:
self.assertEqual(passValue(f), f)
else:
self.assertEqual(float(passValue(f)), f)
if self.wantobjects:
f = passValue(float('nan'))
self.assertNotEqual(f, f)
self.assertEqual(passValue(float('inf')), float('inf'))
self.assertEqual(passValue(-float('inf')), -float('inf'))
else:
self.assertEqual(float(passValue(float('inf'))), float('inf'))
self.assertEqual(float(passValue(-float('inf'))), -float('inf'))
# XXX NaN representation can be not parsable by float()
self.assertEqual(passValue((1, '2', (3.4,))),
(1, '2', (3.4,)) if self.wantobjects else '1 2 3.4')
self.assertEqual(passValue(['a', ['b', 'c']]),
('a', ('b', 'c')) if self.wantobjects else 'a {b c}')
def test_set_object_concurrent_mutation_in_sequence_conversion(self):
# Prevent SIGSEV when the object to convert is concurrently mutated.
# See: https://github.com/python/cpython/issues/143310.
string = "value"
class Value:
def __str__(self):
values.clear()
return string
class List(list):
pass
expect = (string, "pad") if self.wantobjects else f"{string} pad"
self.assertEqual(self.passValue(values := [Value(), "pad"]), expect)
self.assertEqual(self.passValue(values := List([Value(), "pad"])), expect)
def test_user_command(self):
result = None
def testfunc(arg):
nonlocal result
result = arg
return arg
self.interp.createcommand('testfunc', testfunc)
self.addCleanup(self.interp.tk.deletecommand, 'testfunc')
def check(value, expected1=None, expected2=None, *, eq=self.assertEqual):
expected = value
if self.wantobjects >= 2:
if expected2 is not None:
expected = expected2
expected_type = type(expected)
else:
if expected1 is not None:
expected = expected1
expected_type = str
nonlocal result
result = None
r = self.interp.call('testfunc', value)
self.assertIsInstance(result, expected_type)
eq(result, expected)
self.assertIsInstance(r, expected_type)
eq(r, expected)
def float_eq(actual, expected):
self.assertAlmostEqual(float(actual), expected,
delta=abs(expected) * 1e-10)
check(True, '1', 1)
check(False, '0', 0)
check('string')
check('string\xbd')
check('string\u20ac')
check('string\U0001f4bb')
if sys.platform != 'win32':
check('<\udce2\udc82\udcac>', '<\u20ac>', '<\u20ac>')
check('<\udced\udca0\udcbd\udced\udcb2\udcbb>', '<\U0001f4bb>', '<\U0001f4bb>')
check('')
check(b'string', 'string')
check(b'string\xe2\x82\xac', 'string\xe2\x82\xac')
check(b'string\xbd', 'string\xbd')
check(b'', '')
check('str\x00ing')
check('str\x00ing\xbd')
check('str\x00ing\u20ac')
check(b'str\x00ing', 'str\x00ing')
check(b'str\xc0\x80ing', 'str\xc0\x80ing')
check(b'str\xc0\x80ing\xe2\x82\xac', 'str\xc0\x80ing\xe2\x82\xac')
for i in self.get_integers():
check(i, str(i))
for f in (0.0, 1.0, -1.0):
check(f, repr(f))
for f in (1/3.0, sys.float_info.min, sys.float_info.max,
-sys.float_info.min, -sys.float_info.max):
check(f, eq=float_eq)
check(float('inf'), eq=float_eq)
check(-float('inf'), eq=float_eq)
# XXX NaN representation can be not parsable by float()
check((), '', '')
check((1, (2,), (3, 4), '5 6', ()),
'1 2 {3 4} {5 6} {}',
(1, (2,), (3, 4), '5 6', ''))
check([1, [2,], [3, 4], '5 6', []],
'1 2 {3 4} {5 6} {}',
(1, (2,), (3, 4), '5 6', ''))
def test_passing_tcl_obj(self):
tcl = self.interp.tk
a = None
def testfunc(arg):
nonlocal a
a = arg
self.interp.createcommand('testfunc', testfunc)
self.addCleanup(self.interp.tk.deletecommand, 'testfunc')
tcl.eval(r'set a "\u20ac \ud83d\udcbb \0 \udcab"; regexp -about $a')
tcl.eval(r'testfunc $a')
expected = '\u20ac \U0001f4bb \0 \udced\udcb2\udcab'
if self.wantobjects >= 2:
self.assertEqual(str(a), expected)
self.assertEqual(a.string, expected)
self.assertEqual(a.typename, 'regexp')
else:
self.assertEqual(a, expected)
def test_splitlist(self):
splitlist = self.interp.tk.splitlist
call = self.interp.tk.call
self.assertRaises(TypeError, splitlist)
self.assertRaises(TypeError, splitlist, 'a', 'b')
self.assertRaises(TypeError, splitlist, 2)
testcases = [
('2', ('2',)),
('', ()),
('{}', ('',)),
('""', ('',)),
('a\n b\t\r c\n ', ('a', 'b', 'c')),
(b'a\n b\t\r c\n ', ('a', 'b', 'c')),
('a \u20ac', ('a', '\u20ac')),
('a \U0001f4bb', ('a', '\U0001f4bb')),
(b'a \xe2\x82\xac', ('a', '\u20ac')),
(b'a \xf0\x9f\x92\xbb', ('a', '\U0001f4bb')),
(b'a \xed\xa0\xbd\xed\xb2\xbb', ('a', '\U0001f4bb')),
(b'a\xc0\x80b c\xc0\x80d', ('a\x00b', 'c\x00d')),
('a {b c}', ('a', 'b c')),
(r'a b\ c', ('a', 'b c')),
(('a', 'b c'), ('a', 'b c')),
('a 2', ('a', '2')),
(('a', 2), ('a', 2)),
('a 3.4', ('a', '3.4')),
(('a', 3.4), ('a', 3.4)),
((), ()),
([], ()),
(['a', ['b', 'c']], ('a', ['b', 'c'])),
(call('list', 1, '2', (3.4,)),
(1, '2', (3.4,)) if self.wantobjects else
('1', '2', '3.4')),
]
if not self.wantobjects:
expected = ('12', '\u20ac', '\xe2\x82\xac', '3.4')
else:
expected = (12, '\u20ac', b'\xe2\x82\xac', (3.4,))
testcases += [
(call('dict', 'create', 12, '\u20ac', b'\xe2\x82\xac', (3.4,)),
expected),
]
dbg_info = ('want objects? %s, Tcl version: %s, Tcl patchlevel: %s'
% (self.wantobjects, tcl_version, self.interp.info_patchlevel()))
for arg, res in testcases:
self.assertEqual(splitlist(arg), res,
'arg=%a, %s' % (arg, dbg_info))
self.assertRaises(TclError, splitlist, '{')
def test_splitdict(self):
splitdict = tkinter._splitdict
tcl = self.interp.tk
arg = '-a {1 2 3} -something foo status {}'
self.assertEqual(splitdict(tcl, arg, False),
{'-a': '1 2 3', '-something': 'foo', 'status': ''})
self.assertEqual(splitdict(tcl, arg),
{'a': '1 2 3', 'something': 'foo', 'status': ''})
arg = ('-a', (1, 2, 3), '-something', 'foo', 'status', '{}')
self.assertEqual(splitdict(tcl, arg, False),
{'-a': (1, 2, 3), '-something': 'foo', 'status': '{}'})
self.assertEqual(splitdict(tcl, arg),
{'a': (1, 2, 3), 'something': 'foo', 'status': '{}'})
self.assertRaises(RuntimeError, splitdict, tcl, '-a b -c ')
self.assertRaises(RuntimeError, splitdict, tcl, ('-a', 'b', '-c'))
arg = tcl.call('list',
'-a', (1, 2, 3), '-something', 'foo', 'status', ())
self.assertEqual(splitdict(tcl, arg),
{'a': (1, 2, 3) if self.wantobjects else '1 2 3',
'something': 'foo', 'status': ''})
arg = tcl.call('dict', 'create',
'-a', (1, 2, 3), '-something', 'foo', 'status', ())
if not self.wantobjects:
expected = {'a': '1 2 3', 'something': 'foo', 'status': ''}
else:
expected = {'a': (1, 2, 3), 'something': 'foo', 'status': ''}
self.assertEqual(splitdict(tcl, arg), expected)
def test_join(self):
join = tkinter._join
tcl = self.interp.tk
def unpack(s):
return tcl.call('lindex', s, 0)
def check(value):
self.assertEqual(unpack(join([value])), value)
self.assertEqual(unpack(join([value, 0])), value)
self.assertEqual(unpack(unpack(join([[value]]))), value)
self.assertEqual(unpack(unpack(join([[value, 0]]))), value)
self.assertEqual(unpack(unpack(join([[value], 0]))), value)
self.assertEqual(unpack(unpack(join([[value, 0], 0]))), value)
check('')
check('spam')
check('sp am')
check('sp\tam')
check('sp\nam')
check(' \t\n')
check('{spam}')
check('{sp am}')
check('"spam"')
check('"sp am"')
check('{"spam"}')
check('"{spam}"')
check('sp\\am')
check('"sp\\am"')
check('"{}" "{}"')
check('"\\')
check('"{')
check('"}')
check('\n\\')
check('\n{')
check('\n}')
check('\\\n')
check('{\n')
check('}\n')
@support.cpython_only
def test_new_tcl_obj(self):
support.check_disallow_instantiation(self, _tkinter.Tcl_Obj)
support.check_disallow_instantiation(self, _tkinter.TkttType)
support.check_disallow_instantiation(self, _tkinter.TkappType)
class BigmemTclTest(unittest.TestCase):
def setUp(self):
self.interp = Tcl()
@support.cpython_only
@unittest.skipUnless(INT_MAX < PY_SSIZE_T_MAX, "needs UINT_MAX < SIZE_MAX")
@support.bigmemtest(size=INT_MAX + 1, memuse=5, dry_run=False)
def test_huge_string_call(self, size):
value = ' ' * size
self.assertRaises(OverflowError, self.interp.call, 'string', 'index', value, 0)
@support.cpython_only
@unittest.skipUnless(INT_MAX < PY_SSIZE_T_MAX, "needs UINT_MAX < SIZE_MAX")
@support.bigmemtest(size=INT_MAX + 1, memuse=3, dry_run=False)
def test_huge_string_builtins(self, size):
tk = self.interp.tk
value = '1' + ' ' * size
self.assertRaises(OverflowError, tk.getint, value)
self.assertRaises(OverflowError, tk.getdouble, value)
self.assertRaises(OverflowError, tk.getboolean, value)
self.assertRaises(OverflowError, tk.eval, value)
self.assertRaises(OverflowError, tk.evalfile, value)
self.assertRaises(OverflowError, tk.record, value)
self.assertRaises(OverflowError, tk.adderrorinfo, value)
self.assertRaises(OverflowError, tk.setvar, value, 'x', 'a')
self.assertRaises(OverflowError, tk.setvar, 'x', value, 'a')
self.assertRaises(OverflowError, tk.unsetvar, value)
self.assertRaises(OverflowError, tk.unsetvar, 'x', value)
self.assertRaises(OverflowError, tk.adderrorinfo, value)
self.assertRaises(OverflowError, tk.exprstring, value)
self.assertRaises(OverflowError, tk.exprlong, value)
self.assertRaises(OverflowError, tk.exprboolean, value)
self.assertRaises(OverflowError, tk.splitlist, value)
self.assertRaises(OverflowError, tk.createcommand, value, max)
self.assertRaises(OverflowError, tk.deletecommand, value)
@support.cpython_only
@unittest.skipUnless(INT_MAX < PY_SSIZE_T_MAX, "needs UINT_MAX < SIZE_MAX")
@support.bigmemtest(size=INT_MAX + 1, memuse=6, dry_run=False)
def test_huge_string_builtins2(self, size):
# These commands require larger memory for possible error messages
tk = self.interp.tk
value = '1' + ' ' * size
self.assertRaises(OverflowError, tk.evalfile, value)
self.assertRaises(OverflowError, tk.unsetvar, value)
self.assertRaises(OverflowError, tk.unsetvar, 'x', value)
def setUpModule():
wantobjects = support.get_resource_value('wantobjects')
if wantobjects is not None:
unittest.enterModuleContext(
support.swap_attr(tkinter, 'wantobjects', int(wantobjects)))
if support.verbose:
tcl = Tcl()
print('patchlevel =', tcl.call('info', 'patchlevel'), flush=True)
if __name__ == "__main__":
unittest.main()