Linux server1.dn-server.com 4.18.0-553.89.1.lve.el8.x86_64 #1 SMP Wed Dec 10 13:58:50 UTC 2025 x86_64
LiteSpeed
Server IP : 195.201.204.189 & Your IP : 216.73.217.145
Domains :
Cant Read [ /etc/named.conf ]
User : beriska1
Terminal
Auto Root
Create File
Create Folder
Localroot Suggester
Backdoor Destroyer
Readme
/
opt /
alt /
python311 /
lib64 /
python3.11 /
test /
Delete
Unzip
Name
Size
Permission
Date
Action
__pycache__
[ DIR ]
drwxr-xr-x
2026-05-05 23:07
audiodata
[ DIR ]
drwxr-xr-x
2026-05-05 23:07
certdata
[ DIR ]
drwxr-xr-x
2026-05-05 23:07
cjkencodings
[ DIR ]
drwxr-xr-x
2026-05-05 23:07
configdata
[ DIR ]
drwxr-xr-x
2026-05-05 23:07
crashers
[ DIR ]
drwxr-xr-x
2026-05-05 23:07
data
[ DIR ]
drwxr-xr-x
2026-05-05 23:07
decimaltestdata
[ DIR ]
drwxr-xr-x
2026-05-05 23:07
dtracedata
[ DIR ]
drwxr-xr-x
2026-05-05 23:07
encoded_modules
[ DIR ]
drwxr-xr-x
2026-05-05 23:07
imghdrdata
[ DIR ]
drwxr-xr-x
2026-05-05 23:07
leakers
[ DIR ]
drwxr-xr-x
2026-05-05 23:07
libregrtest
[ DIR ]
drwxr-xr-x
2026-05-05 23:07
regrtestdata
[ DIR ]
drwxr-xr-x
2026-04-17 14:59
sndhdrdata
[ DIR ]
drwxr-xr-x
2026-05-05 23:07
subprocessdata
[ DIR ]
drwxr-xr-x
2026-05-05 23:07
support
[ DIR ]
drwxr-xr-x
2026-05-05 23:07
test_asyncio
[ DIR ]
drwxr-xr-x
2026-05-05 23:07
test_capi
[ DIR ]
drwxr-xr-x
2026-05-05 23:07
test_concurrent_futures
[ DIR ]
drwxr-xr-x
2026-05-05 23:07
test_cppext
[ DIR ]
drwxr-xr-x
2026-05-05 23:07
test_dataclasses
[ DIR ]
drwxr-xr-x
2026-05-05 23:07
test_doctest
[ DIR ]
drwxr-xr-x
2026-05-05 23:07
test_email
[ DIR ]
drwxr-xr-x
2026-05-05 23:07
test_future_stmt
[ DIR ]
drwxr-xr-x
2026-05-05 23:07
test_gdb
[ DIR ]
drwxr-xr-x
2026-05-05 23:07
test_import
[ DIR ]
drwxr-xr-x
2026-05-05 23:07
test_importlib
[ DIR ]
drwxr-xr-x
2026-05-05 23:07
test_inspect
[ DIR ]
drwxr-xr-x
2026-05-05 23:07
test_json
[ DIR ]
drwxr-xr-x
2026-05-05 23:07
test_module
[ DIR ]
drwxr-xr-x
2026-05-05 23:07
test_multiprocessing_fork
[ DIR ]
drwxr-xr-x
2026-05-05 23:07
test_multiprocessing_forkserver
[ DIR ]
drwxr-xr-x
2026-05-05 23:07
test_multiprocessing_spawn
[ DIR ]
drwxr-xr-x
2026-05-05 23:07
test_peg_generator
[ DIR ]
drwxr-xr-x
2026-05-05 23:07
test_pydoc
[ DIR ]
drwxr-xr-x
2026-05-05 23:07
test_sqlite3
[ DIR ]
drwxr-xr-x
2026-05-05 23:07
test_tomllib
[ DIR ]
drwxr-xr-x
2026-05-05 23:07
test_tools
[ DIR ]
drwxr-xr-x
2026-05-05 23:07
test_warnings
[ DIR ]
drwxr-xr-x
2026-05-05 23:07
test_zoneinfo
[ DIR ]
drwxr-xr-x
2026-05-05 23:07
tokenizedata
[ DIR ]
drwxr-xr-x
2026-05-05 23:07
tracedmodules
[ DIR ]
drwxr-xr-x
2026-05-05 23:07
typinganndata
[ DIR ]
drwxr-xr-x
2026-05-05 23:07
xmltestdata
[ DIR ]
drwxr-xr-x
2026-05-05 23:07
ziptestdata
[ DIR ]
drwxr-xr-x
2026-05-05 23:07
Sine-1000Hz-300ms.aif
60.25
KB
-rw-r--r--
2026-03-03 04:22
__init__.py
47
B
-rw-r--r--
2026-03-03 04:22
__main__.py
67
B
-rw-r--r--
2026-03-03 04:22
_test_atexit.py
3.61
KB
-rw-r--r--
2026-03-03 04:22
_test_eintr.py
17.75
KB
-rw-r--r--
2026-03-03 04:22
_test_embed_set_config.py
8.68
KB
-rw-r--r--
2026-03-03 04:22
_test_embed_structseq.py
1.83
KB
-rw-r--r--
2026-03-03 04:22
_test_multiprocessing.py
204.49
KB
-rw-r--r--
2026-03-03 04:22
_test_venv_multiprocessing.py
762
B
-rw-r--r--
2026-03-03 04:22
archiver_tests.py
6.1
KB
-rw-r--r--
2026-03-03 04:22
audiotest.au
27.48
KB
-rw-r--r--
2026-03-03 04:22
audiotests.py
12.13
KB
-rw-r--r--
2026-03-03 04:22
audit-tests.py
11.55
KB
-rw-r--r--
2026-03-03 04:22
autotest.py
214
B
-rw-r--r--
2026-03-03 04:22
bisect_cmd.py
5.34
KB
-rwxr-xr-x
2026-03-03 04:22
clinic.test.c
118.1
KB
-rw-r--r--
2026-03-03 04:22
cmath_testcases.txt
141.05
KB
-rw-r--r--
2026-03-03 04:22
curses_tests.py
1.23
KB
-rwxr-xr-x
2026-03-03 04:22
datetimetester.py
257.92
KB
-rw-r--r--
2026-03-03 04:22
dis_module.py
76
B
-rw-r--r--
2026-03-03 04:22
empty.vbs
70
B
-rw-r--r--
2026-03-03 04:22
exception_hierarchy.txt
2.33
KB
-rw-r--r--
2026-03-03 04:22
floating_points.txt
15.92
KB
-rw-r--r--
2026-03-03 04:22
fork_wait.py
2.18
KB
-rw-r--r--
2026-03-03 04:22
formatfloat_testcases.txt
7.45
KB
-rw-r--r--
2026-03-03 04:22
ieee754.txt
3.15
KB
-rw-r--r--
2026-03-03 04:22
imp_dummy.py
63
B
-rw-r--r--
2026-03-03 04:22
list_tests.py
16.91
KB
-rw-r--r--
2026-03-03 04:22
lock_tests.py
35.49
KB
-rw-r--r--
2026-03-03 04:22
mailcap.txt
1.24
KB
-rw-r--r--
2026-03-03 04:22
mapping_tests.py
21.83
KB
-rw-r--r--
2026-03-03 04:22
math_testcases.txt
23.19
KB
-rw-r--r--
2026-03-03 04:22
memory_watchdog.py
859
B
-rw-r--r--
2026-03-03 04:22
mime.types
47.37
KB
-rw-r--r--
2026-03-03 04:22
mock_socket.py
3.7
KB
-rw-r--r--
2026-03-03 04:22
mp_fork_bomb.py
448
B
-rw-r--r--
2026-03-03 04:22
mp_preload.py
351
B
-rw-r--r--
2026-03-03 04:22
multibytecodec_support.py
14.19
KB
-rw-r--r--
2026-03-03 04:22
pickletester.py
143.84
KB
-rw-r--r--
2026-03-03 04:22
profilee.py
2.97
KB
-rw-r--r--
2026-03-03 04:22
pstats.pck
65.05
KB
-rw-r--r--
2026-03-03 04:22
pyclbr_input.py
648
B
-rw-r--r--
2026-03-03 04:22
pythoninfo.py
28.32
KB
-rw-r--r--
2026-03-03 04:22
randv2_32.pck
7.34
KB
-rw-r--r--
2026-03-03 04:22
randv2_64.pck
7.19
KB
-rw-r--r--
2026-03-03 04:22
randv3.pck
7.82
KB
-rw-r--r--
2026-03-03 04:22
re_tests.py
25.94
KB
-rwxr-xr-x
2026-03-03 04:22
recursion.tar
516
B
-rw-r--r--
2026-03-03 04:22
regrtest.py
1.28
KB
-rwxr-xr-x
2026-03-03 04:22
relimport.py
27
B
-rw-r--r--
2026-03-03 04:22
reperf.py
538
B
-rw-r--r--
2026-03-03 04:22
seq_tests.py
14.96
KB
-rw-r--r--
2026-03-03 04:22
signalinterproctester.py
3.08
KB
-rw-r--r--
2026-03-03 04:22
ssl_servers.py
7.12
KB
-rw-r--r--
2026-03-03 04:22
ssltests.py
1.03
KB
-rw-r--r--
2026-03-03 04:22
string_tests.py
69.05
KB
-rw-r--r--
2026-03-03 04:22
test___all__.py
5.39
KB
-rw-r--r--
2026-03-03 04:22
test__locale.py
8.09
KB
-rw-r--r--
2026-03-03 04:22
test__opcode.py
4.14
KB
-rw-r--r--
2026-03-03 04:22
test__osx_support.py
13.62
KB
-rw-r--r--
2026-03-03 04:22
test__xxsubinterpreters.py
78.84
KB
-rw-r--r--
2026-03-03 04:22
test_abc.py
23.65
KB
-rw-r--r--
2026-03-03 04:22
test_abstract_numbers.py
5.81
KB
-rw-r--r--
2026-03-03 04:22
test_aifc.py
17.84
KB
-rw-r--r--
2026-03-03 04:22
test_argparse.py
187.05
KB
-rw-r--r--
2026-03-03 04:22
test_array.py
55.27
KB
-rw-r--r--
2026-03-03 04:22
test_asdl_parser.py
4.44
KB
-rw-r--r--
2026-03-03 04:22
test_ast.py
118.11
KB
-rw-r--r--
2026-03-03 04:22
test_asyncgen.py
46.94
KB
-rw-r--r--
2026-03-03 04:22
test_asynchat.py
9.46
KB
-rw-r--r--
2026-03-03 04:22
test_asyncore.py
26.11
KB
-rw-r--r--
2026-03-03 04:22
test_atexit.py
3.21
KB
-rw-r--r--
2026-03-03 04:22
test_audioop.py
28.32
KB
-rw-r--r--
2026-03-03 04:22
test_audit.py
6.91
KB
-rw-r--r--
2026-03-03 04:22
test_augassign.py
7.68
KB
-rw-r--r--
2026-03-03 04:22
test_base64.py
35.08
KB
-rw-r--r--
2026-03-03 04:22
test_baseexception.py
7.77
KB
-rw-r--r--
2026-03-03 04:22
test_bdb.py
43.7
KB
-rw-r--r--
2026-03-03 04:22
test_bigaddrspace.py
2.83
KB
-rw-r--r--
2026-03-03 04:22
test_bigmem.py
45.01
KB
-rw-r--r--
2026-03-03 04:22
test_binascii.py
19.37
KB
-rw-r--r--
2026-03-03 04:22
test_binop.py
14.14
KB
-rw-r--r--
2026-03-03 04:22
test_bisect.py
15.71
KB
-rw-r--r--
2026-03-03 04:22
test_bool.py
13.02
KB
-rw-r--r--
2026-03-03 04:22
test_buffer.py
160.38
KB
-rw-r--r--
2026-03-03 04:22
test_bufio.py
2.58
KB
-rw-r--r--
2026-03-03 04:22
test_builtin.py
92.55
KB
-rw-r--r--
2026-03-03 04:22
test_bytes.py
79.63
KB
-rw-r--r--
2026-03-03 04:22
test_bz2.py
42.44
KB
-rw-r--r--
2026-03-03 04:22
test_c_locale_coercion.py
20.99
KB
-rw-r--r--
2026-03-03 04:22
test_calendar.py
51.01
KB
-rw-r--r--
2026-03-03 04:22
test_call.py
26.57
KB
-rw-r--r--
2026-03-03 04:22
test_cgi.py
22.27
KB
-rw-r--r--
2026-03-03 04:22
test_cgitb.py
2.62
KB
-rw-r--r--
2026-03-03 04:22
test_charmapcodec.py
1.77
KB
-rw-r--r--
2026-03-03 04:22
test_check_c_globals.py
1.02
KB
-rw-r--r--
2026-03-03 04:22
test_class.py
19.33
KB
-rw-r--r--
2026-03-03 04:22
test_clinic.py
76.86
KB
-rw-r--r--
2026-03-03 04:22
test_cmath.py
24.14
KB
-rw-r--r--
2026-03-03 04:22
test_cmd.py
6.49
KB
-rw-r--r--
2026-03-03 04:22
test_cmd_line.py
39.49
KB
-rw-r--r--
2026-03-03 04:22
test_cmd_line_script.py
34.91
KB
-rw-r--r--
2026-03-03 04:22
test_code.py
23.29
KB
-rw-r--r--
2026-03-03 04:22
test_code_module.py
5.53
KB
-rw-r--r--
2026-03-03 04:22
test_codeccallbacks.py
49.22
KB
-rw-r--r--
2026-03-03 04:22
test_codecencodings_cn.py
3.86
KB
-rw-r--r--
2026-03-03 04:22
test_codecencodings_hk.py
701
B
-rw-r--r--
2026-03-03 04:22
test_codecencodings_iso2022.py
3.65
KB
-rw-r--r--
2026-03-03 04:22
test_codecencodings_jp.py
4.79
KB
-rw-r--r--
2026-03-03 04:22
test_codecencodings_kr.py
2.96
KB
-rw-r--r--
2026-03-03 04:22
test_codecencodings_tw.py
681
B
-rw-r--r--
2026-03-03 04:22
test_codecmaps_cn.py
746
B
-rw-r--r--
2026-03-03 04:22
test_codecmaps_hk.py
386
B
-rw-r--r--
2026-03-03 04:22
test_codecmaps_jp.py
1.7
KB
-rw-r--r--
2026-03-03 04:22
test_codecmaps_kr.py
1.16
KB
-rw-r--r--
2026-03-03 04:22
test_codecmaps_tw.py
705
B
-rw-r--r--
2026-03-03 04:22
test_codecs.py
139.04
KB
-rw-r--r--
2026-03-03 04:22
test_codeop.py
9.62
KB
-rw-r--r--
2026-03-03 04:22
test_collections.py
92.34
KB
-rw-r--r--
2026-03-03 04:22
test_colorsys.py
4.27
KB
-rw-r--r--
2026-03-03 04:22
test_compare.py
17.46
KB
-rw-r--r--
2026-03-03 04:22
test_compile.py
59.92
KB
-rw-r--r--
2026-03-03 04:22
test_compileall.py
48.27
KB
-rw-r--r--
2026-03-03 04:22
test_complex.py
32.89
KB
-rw-r--r--
2026-03-03 04:22
test_configparser.py
86.84
KB
-rw-r--r--
2026-03-03 04:22
test_contains.py
3.35
KB
-rw-r--r--
2026-03-03 04:22
test_context.py
30.78
KB
-rw-r--r--
2026-03-03 04:22
test_contextlib.py
41.2
KB
-rw-r--r--
2026-03-03 04:22
test_contextlib_async.py
23.66
KB
-rw-r--r--
2026-03-03 04:22
test_copy.py
26.52
KB
-rw-r--r--
2026-03-03 04:22
test_copyreg.py
4.39
KB
-rw-r--r--
2026-03-03 04:22
test_coroutines.py
67.04
KB
-rw-r--r--
2026-03-03 04:22
test_cprofile.py
6.33
KB
-rw-r--r--
2026-03-03 04:22
test_crashers.py
1.17
KB
-rw-r--r--
2026-03-03 04:22
test_crypt.py
4.19
KB
-rw-r--r--
2026-03-03 04:22
test_csv.py
59
KB
-rw-r--r--
2026-03-03 04:22
test_ctypes.py
199
B
-rw-r--r--
2026-03-03 04:22
test_curses.py
46.72
KB
-rw-r--r--
2026-03-03 04:22
test_datetime.py
2.31
KB
-rw-r--r--
2026-03-03 04:22
test_dbm.py
6.83
KB
-rw-r--r--
2026-03-03 04:22
test_dbm_dumb.py
11.08
KB
-rw-r--r--
2026-03-03 04:22
test_dbm_gnu.py
6.82
KB
-rw-r--r--
2026-03-03 04:22
test_dbm_ndbm.py
5.46
KB
-rw-r--r--
2026-03-03 04:22
test_decimal.py
216.13
KB
-rw-r--r--
2026-03-03 04:22
test_decorators.py
14.65
KB
-rw-r--r--
2026-03-03 04:22
test_defaultdict.py
6.13
KB
-rw-r--r--
2026-03-03 04:22
test_deque.py
33.16
KB
-rw-r--r--
2026-03-03 04:22
test_descr.py
195.49
KB
-rw-r--r--
2026-03-03 04:22
test_descrtut.py
10.97
KB
-rw-r--r--
2026-03-03 04:22
test_devpoll.py
4.44
KB
-rw-r--r--
2026-03-03 04:22
test_dict.py
50.44
KB
-rw-r--r--
2026-03-03 04:22
test_dict_version.py
6.1
KB
-rw-r--r--
2026-03-03 04:22
test_dictcomps.py
5.15
KB
-rw-r--r--
2026-03-03 04:22
test_dictviews.py
13.69
KB
-rw-r--r--
2026-03-03 04:22
test_difflib.py
21.47
KB
-rw-r--r--
2026-03-03 04:22
test_difflib_expect.html
100.85
KB
-rw-r--r--
2026-03-03 04:22
test_dis.py
77.33
KB
-rw-r--r--
2026-03-03 04:22
test_distutils.py
549
B
-rw-r--r--
2026-03-03 04:22
test_docxmlrpc.py
9.1
KB
-rw-r--r--
2026-03-03 04:22
test_dtrace.py
7.61
KB
-rw-r--r--
2026-03-03 04:22
test_dynamic.py
5.91
KB
-rw-r--r--
2026-03-03 04:22
test_dynamicclassattribute.py
9.57
KB
-rw-r--r--
2026-03-03 04:22
test_eintr.py
627
B
-rw-r--r--
2026-03-03 04:22
test_embed.py
68.41
KB
-rw-r--r--
2026-03-03 04:22
test_ensurepip.py
12.2
KB
-rw-r--r--
2026-03-03 04:22
test_enum.py
173.47
KB
-rw-r--r--
2026-03-03 04:22
test_enumerate.py
9.14
KB
-rw-r--r--
2026-03-03 04:22
test_eof.py
3.02
KB
-rw-r--r--
2026-03-03 04:22
test_epoll.py
9.14
KB
-rw-r--r--
2026-03-03 04:22
test_errno.py
1.04
KB
-rw-r--r--
2026-03-03 04:22
test_except_star.py
38.9
KB
-rw-r--r--
2026-03-03 04:22
test_exception_group.py
33.55
KB
-rw-r--r--
2026-03-03 04:22
test_exception_hierarchy.py
7.54
KB
-rw-r--r--
2026-03-03 04:22
test_exception_variations.py
13.74
KB
-rw-r--r--
2026-03-03 04:22
test_exceptions.py
94.1
KB
-rw-r--r--
2026-03-03 04:22
test_extcall.py
14.1
KB
-rw-r--r--
2026-03-03 04:22
test_faulthandler.py
30.3
KB
-rw-r--r--
2026-03-03 04:22
test_fcntl.py
7.43
KB
-rw-r--r--
2026-03-03 04:22
test_file.py
11.78
KB
-rw-r--r--
2026-03-03 04:22
test_file_eintr.py
10.74
KB
-rw-r--r--
2026-03-03 04:22
test_filecmp.py
10.33
KB
-rw-r--r--
2026-03-03 04:22
test_fileinput.py
37.85
KB
-rw-r--r--
2026-03-03 04:22
test_fileio.py
19.97
KB
-rw-r--r--
2026-03-03 04:22
test_fileutils.py
951
B
-rw-r--r--
2026-03-03 04:22
test_finalization.py
14.66
KB
-rw-r--r--
2026-03-03 04:22
test_float.py
67.84
KB
-rw-r--r--
2026-03-03 04:22
test_flufl.py
1.55
KB
-rw-r--r--
2026-03-03 04:22
test_fnmatch.py
10.68
KB
-rw-r--r--
2026-03-03 04:22
test_fork1.py
3.3
KB
-rw-r--r--
2026-03-03 04:22
test_format.py
28.44
KB
-rw-r--r--
2026-03-03 04:22
test_fractions.py
34.72
KB
-rw-r--r--
2026-03-03 04:22
test_frame.py
11.31
KB
-rw-r--r--
2026-03-03 04:22
test_frozen.py
2.2
KB
-rw-r--r--
2026-03-03 04:22
test_fstring.py
53.37
KB
-rw-r--r--
2026-03-03 04:22
test_ftplib.py
41.99
KB
-rw-r--r--
2026-03-03 04:22
test_funcattrs.py
14.86
KB
-rw-r--r--
2026-03-03 04:22
test_functools.py
109.04
KB
-rw-r--r--
2026-03-03 04:22
test_gc.py
47.19
KB
-rw-r--r--
2026-03-03 04:22
test_generator_stop.py
943
B
-rw-r--r--
2026-03-03 04:22
test_generators.py
64.62
KB
-rw-r--r--
2026-03-03 04:22
test_genericalias.py
17.09
KB
-rw-r--r--
2026-03-03 04:22
test_genericclass.py
9.44
KB
-rw-r--r--
2026-03-03 04:22
test_genericpath.py
22.85
KB
-rw-r--r--
2026-03-03 04:22
test_genexps.py
7.3
KB
-rw-r--r--
2026-03-03 04:22
test_getopt.py
6.59
KB
-rw-r--r--
2026-03-03 04:22
test_getpass.py
6.37
KB
-rw-r--r--
2026-03-03 04:22
test_getpath.py
44.37
KB
-rw-r--r--
2026-03-03 04:22
test_gettext.py
35.67
KB
-rw-r--r--
2026-03-03 04:22
test_glob.py
17.19
KB
-rw-r--r--
2026-03-03 04:22
test_global.py
1.2
KB
-rw-r--r--
2026-03-03 04:22
test_grammar.py
65.51
KB
-rw-r--r--
2026-03-03 04:22
test_graphlib.py
8.31
KB
-rw-r--r--
2026-03-03 04:22
test_grp.py
3.67
KB
-rw-r--r--
2026-03-03 04:22
test_gzip.py
38.78
KB
-rw-r--r--
2026-03-03 04:22
test_hash.py
12.11
KB
-rw-r--r--
2026-03-03 04:22
test_hashlib.py
46.51
KB
-rw-r--r--
2026-03-03 04:22
test_heapq.py
16.45
KB
-rw-r--r--
2026-03-03 04:22
test_hmac.py
26.06
KB
-rw-r--r--
2026-03-03 04:22
test_html.py
4.23
KB
-rw-r--r--
2026-03-03 04:22
test_htmlparser.py
45.15
KB
-rw-r--r--
2026-03-03 04:22
test_http_cookiejar.py
81.59
KB
-rw-r--r--
2026-03-03 04:22
test_http_cookies.py
21.76
KB
-rw-r--r--
2026-03-03 04:22
test_httplib.py
91.75
KB
-rw-r--r--
2026-03-03 04:22
test_httpservers.py
57.44
KB
-rw-r--r--
2026-03-03 04:22
test_idle.py
830
B
-rw-r--r--
2026-03-03 04:22
test_imaplib.py
42.95
KB
-rw-r--r--
2026-03-03 04:22
test_imghdr.py
4.81
KB
-rw-r--r--
2026-03-03 04:22
test_imp.py
19.31
KB
-rw-r--r--
2026-03-03 04:22
test_index.py
8.37
KB
-rw-r--r--
2026-03-03 04:22
test_int.py
29.62
KB
-rw-r--r--
2026-03-03 04:22
test_int_literal.py
6.89
KB
-rw-r--r--
2026-03-03 04:22
test_interpreters.py
22.57
KB
-rw-r--r--
2026-03-03 04:22
test_io.py
180.21
KB
-rw-r--r--
2026-03-03 04:22
test_ioctl.py
3.24
KB
-rw-r--r--
2026-03-03 04:22
test_ipaddress.py
125.61
KB
-rw-r--r--
2026-03-03 04:22
test_isinstance.py
12.95
KB
-rw-r--r--
2026-03-03 04:22
test_iter.py
36.77
KB
-rw-r--r--
2026-03-03 04:22
test_iterlen.py
7.1
KB
-rw-r--r--
2026-03-03 04:22
test_itertools.py
100.18
KB
-rw-r--r--
2026-03-03 04:22
test_keyword.py
1.97
KB
-rw-r--r--
2026-03-03 04:22
test_keywordonlyarg.py
6.89
KB
-rw-r--r--
2026-03-03 04:22
test_kqueue.py
8.76
KB
-rw-r--r--
2026-03-03 04:22
test_largefile.py
10.17
KB
-rw-r--r--
2026-03-03 04:22
test_launcher.py
27.26
KB
-rw-r--r--
2026-03-03 04:22
test_lib2to3.py
311
B
-rw-r--r--
2026-03-03 04:22
test_linecache.py
11.04
KB
-rw-r--r--
2026-03-03 04:22
test_list.py
9.63
KB
-rw-r--r--
2026-03-03 04:22
test_listcomps.py
3.82
KB
-rw-r--r--
2026-03-03 04:22
test_lltrace.py
3.71
KB
-rw-r--r--
2026-03-03 04:22
test_locale.py
25.21
KB
-rw-r--r--
2026-03-03 04:22
test_logging.py
228.31
KB
-rw-r--r--
2026-03-03 04:22
test_long.py
62
KB
-rw-r--r--
2026-03-03 04:22
test_longexp.py
233
B
-rw-r--r--
2026-03-03 04:22
test_lzma.py
92.78
KB
-rw-r--r--
2026-03-03 04:22
test_mailbox.py
92.51
KB
-rw-r--r--
2026-03-03 04:22
test_mailcap.py
11.49
KB
-rw-r--r--
2026-03-03 04:22
test_marshal.py
23.13
KB
-rw-r--r--
2026-03-03 04:22
test_math.py
95.44
KB
-rw-r--r--
2026-03-03 04:22
test_memoryio.py
32
KB
-rw-r--r--
2026-03-03 04:22
test_memoryview.py
21.37
KB
-rw-r--r--
2026-03-03 04:22
test_metaclass.py
6.19
KB
-rw-r--r--
2026-03-03 04:22
test_mimetypes.py
15.28
KB
-rw-r--r--
2026-03-03 04:22
test_minidom.py
69.84
KB
-rw-r--r--
2026-04-17 14:36
test_mmap.py
38.56
KB
-rw-r--r--
2026-03-03 04:22
test_modulefinder.py
12.21
KB
-rw-r--r--
2026-03-03 04:22
test_msilib.py
5.52
KB
-rw-r--r--
2026-03-03 04:22
test_multibytecodec.py
15.79
KB
-rw-r--r--
2026-03-03 04:22
test_multiprocessing_main_handling.py
11.37
KB
-rw-r--r--
2026-03-03 04:22
test_named_expressions.py
21.36
KB
-rw-r--r--
2026-03-03 04:22
test_netrc.py
11.86
KB
-rw-r--r--
2026-03-03 04:22
test_nis.py
1.28
KB
-rw-r--r--
2026-03-03 04:22
test_nntplib.py
62.7
KB
-rw-r--r--
2026-03-03 04:22
test_ntpath.py
55.96
KB
-rw-r--r--
2026-03-03 04:22
test_numeric_tower.py
8
KB
-rw-r--r--
2026-03-03 04:22
test_opcache.py
10.31
KB
-rw-r--r--
2026-03-03 04:22
test_opcodes.py
3.62
KB
-rw-r--r--
2026-03-03 04:22
test_openpty.py
600
B
-rw-r--r--
2026-03-03 04:22
test_operator.py
26.72
KB
-rw-r--r--
2026-03-03 04:22
test_optparse.py
61.02
KB
-rw-r--r--
2026-03-03 04:22
test_ordered_dict.py
35.63
KB
-rw-r--r--
2026-03-03 04:22
test_os.py
172.72
KB
-rw-r--r--
2026-03-03 04:22
test_ossaudiodev.py
7.27
KB
-rw-r--r--
2026-03-03 04:22
test_osx_env.py
1.31
KB
-rw-r--r--
2026-03-03 04:22
test_pathlib.py
114.88
KB
-rw-r--r--
2026-03-03 04:22
test_patma.py
85.55
KB
-rw-r--r--
2026-03-03 04:22
test_pdb.py
81.57
KB
-rw-r--r--
2026-03-03 04:22
test_peepholer.py
26.69
KB
-rw-r--r--
2026-03-03 04:22
test_pep646_syntax.py
7.79
KB
-rw-r--r--
2026-03-03 04:22
test_pickle.py
19.8
KB
-rw-r--r--
2026-03-03 04:22
test_picklebuffer.py
4.99
KB
-rw-r--r--
2026-03-03 04:22
test_pickletools.py
4.13
KB
-rw-r--r--
2026-03-03 04:22
test_pipes.py
6.79
KB
-rw-r--r--
2026-03-03 04:22
test_pkg.py
9.59
KB
-rw-r--r--
2026-03-03 04:22
test_pkgutil.py
22.84
KB
-rw-r--r--
2026-03-03 04:22
test_platform.py
21.49
KB
-rw-r--r--
2026-03-03 04:22
test_plistlib.py
42.62
KB
-rw-r--r--
2026-03-03 04:22
test_poll.py
7.39
KB
-rw-r--r--
2026-03-03 04:22
test_popen.py
2.11
KB
-rw-r--r--
2026-03-03 04:22
test_poplib.py
17.62
KB
-rw-r--r--
2026-03-03 04:22
test_positional_only_arg.py
17.77
KB
-rw-r--r--
2026-03-03 04:22
test_posix.py
93.48
KB
-rw-r--r--
2026-03-03 04:22
test_posixpath.py
42.27
KB
-rw-r--r--
2026-03-03 04:22
test_pow.py
6.38
KB
-rw-r--r--
2026-03-03 04:22
test_pprint.py
50.58
KB
-rw-r--r--
2026-03-03 04:22
test_print.py
7.72
KB
-rw-r--r--
2026-03-03 04:22
test_profile.py
8.69
KB
-rw-r--r--
2026-03-03 04:22
test_property.py
12.15
KB
-rw-r--r--
2026-03-03 04:22
test_pstats.py
4.31
KB
-rw-r--r--
2026-03-03 04:22
test_pty.py
16.18
KB
-rw-r--r--
2026-03-03 04:22
test_pulldom.py
12.33
KB
-rw-r--r--
2026-03-03 04:22
test_pwd.py
4.32
KB
-rw-r--r--
2026-03-03 04:22
test_py_compile.py
11.89
KB
-rw-r--r--
2026-03-03 04:22
test_pyclbr.py
10.15
KB
-rw-r--r--
2026-03-03 04:22
test_pyexpat.py
39.21
KB
-rw-r--r--
2026-03-03 04:22
test_queue.py
20.61
KB
-rw-r--r--
2026-03-03 04:22
test_quopri.py
7.87
KB
-rw-r--r--
2026-03-03 04:22
test_raise.py
13.44
KB
-rw-r--r--
2026-03-03 04:22
test_random.py
55.22
KB
-rw-r--r--
2026-03-03 04:22
test_range.py
24.64
KB
-rw-r--r--
2026-03-03 04:22
test_re.py
132.4
KB
-rw-r--r--
2026-03-03 04:22
test_readline.py
12.76
KB
-rw-r--r--
2026-03-03 04:22
test_regrtest.py
90.02
KB
-rw-r--r--
2026-03-03 04:22
test_repl.py
4.99
KB
-rw-r--r--
2026-03-03 04:22
test_reprlib.py
15.3
KB
-rw-r--r--
2026-03-03 04:22
test_resource.py
7.12
KB
-rw-r--r--
2026-03-03 04:22
test_richcmp.py
11.95
KB
-rw-r--r--
2026-03-03 04:22
test_rlcompleter.py
7.4
KB
-rw-r--r--
2026-03-03 04:22
test_robotparser.py
11.02
KB
-rw-r--r--
2026-03-03 04:22
test_runpy.py
33.95
KB
-rw-r--r--
2026-03-03 04:22
test_sax.py
54.63
KB
-rw-r--r--
2026-03-03 04:22
test_sched.py
7.38
KB
-rw-r--r--
2026-03-03 04:22
test_scope.py
21.21
KB
-rw-r--r--
2026-03-03 04:22
test_script_helper.py
5.82
KB
-rw-r--r--
2026-03-03 04:22
test_secrets.py
4.28
KB
-rw-r--r--
2026-03-03 04:22
test_select.py
3.43
KB
-rw-r--r--
2026-03-03 04:22
test_selectors.py
19.64
KB
-rw-r--r--
2026-03-03 04:22
test_set.py
71.3
KB
-rw-r--r--
2026-03-03 04:22
test_setcomps.py
3.8
KB
-rw-r--r--
2026-03-03 04:22
test_shelve.py
6.46
KB
-rw-r--r--
2026-03-03 04:22
test_shlex.py
13.48
KB
-rw-r--r--
2026-03-03 04:22
test_shutil.py
112.43
KB
-rw-r--r--
2026-03-03 04:22
test_signal.py
52.72
KB
-rw-r--r--
2026-03-03 04:22
test_site.py
29.82
KB
-rw-r--r--
2026-03-03 04:22
test_slice.py
9.23
KB
-rw-r--r--
2026-03-03 04:22
test_smtpd.py
40.72
KB
-rw-r--r--
2026-03-03 04:22
test_smtplib.py
60.1
KB
-rw-r--r--
2026-03-03 04:22
test_smtpnet.py
3.03
KB
-rw-r--r--
2026-03-03 04:22
test_sndhdr.py
1.51
KB
-rw-r--r--
2026-03-03 04:22
test_socket.py
254.01
KB
-rw-r--r--
2026-03-03 04:22
test_socketserver.py
17.71
KB
-rw-r--r--
2026-03-03 04:22
test_sort.py
13.59
KB
-rw-r--r--
2026-03-03 04:22
test_source_encoding.py
12.47
KB
-rw-r--r--
2026-03-03 04:22
test_spwd.py
2.83
KB
-rw-r--r--
2026-03-03 04:22
test_ssl.py
224.97
KB
-rw-r--r--
2026-03-03 04:22
test_stable_abi_ctypes.py
24.41
KB
-rw-r--r--
2026-03-03 04:22
test_startfile.py
1.7
KB
-rw-r--r--
2026-03-03 04:22
test_stat.py
8.92
KB
-rw-r--r--
2026-03-03 04:22
test_statistics.py
118.58
KB
-rw-r--r--
2026-03-03 04:22
test_strftime.py
7.54
KB
-rw-r--r--
2026-03-03 04:22
test_string.py
21.86
KB
-rw-r--r--
2026-03-03 04:22
test_string_literals.py
13.14
KB
-rw-r--r--
2026-03-03 04:22
test_stringprep.py
3.04
KB
-rw-r--r--
2026-03-03 04:22
test_strptime.py
36.17
KB
-rw-r--r--
2026-03-03 04:22
test_strtod.py
20.06
KB
-rw-r--r--
2026-03-03 04:22
test_struct.py
37.05
KB
-rw-r--r--
2026-03-03 04:22
test_structseq.py
7.23
KB
-rw-r--r--
2026-03-03 04:22
test_subclassinit.py
8.12
KB
-rw-r--r--
2026-03-03 04:22
test_subprocess.py
161.6
KB
-rw-r--r--
2026-03-03 04:22
test_sunau.py
6.09
KB
-rw-r--r--
2026-03-03 04:22
test_sundry.py
2.71
KB
-rw-r--r--
2026-03-03 04:22
test_super.py
9.87
KB
-rw-r--r--
2026-03-03 04:22
test_support.py
27.42
KB
-rw-r--r--
2026-03-03 04:22
test_symtable.py
9.72
KB
-rw-r--r--
2026-03-03 04:22
test_syntax.py
66.33
KB
-rw-r--r--
2026-03-03 04:22
test_sys.py
66.34
KB
-rw-r--r--
2026-03-03 04:22
test_sys_setprofile.py
13.19
KB
-rw-r--r--
2026-03-03 04:22
test_sys_settrace.py
79.97
KB
-rw-r--r--
2026-03-03 04:22
test_sysconfig.py
22.51
KB
-rw-r--r--
2026-03-03 04:22
test_syslog.py
2.76
KB
-rw-r--r--
2026-03-03 04:22
test_tabnanny.py
13.44
KB
-rw-r--r--
2026-03-03 04:22
test_tarfile.py
176.59
KB
-rw-r--r--
2026-03-03 04:22
test_tcl.py
27.26
KB
-rw-r--r--
2026-03-03 04:22
test_telnetlib.py
12.85
KB
-rw-r--r--
2026-03-03 04:22
test_tempfile.py
70.35
KB
-rw-r--r--
2026-03-03 04:22
test_termios.py
10.95
KB
-rw-r--r--
2026-03-03 04:22
test_textwrap.py
41.89
KB
-rw-r--r--
2026-03-03 04:22
test_thread.py
8.57
KB
-rw-r--r--
2026-03-03 04:22
test_threadedtempfile.py
1.93
KB
-rw-r--r--
2026-03-03 04:22
test_threading.py
64.25
KB
-rw-r--r--
2026-03-03 04:22
test_threading_local.py
6.59
KB
-rw-r--r--
2026-03-03 04:22
test_threadsignals.py
9.87
KB
-rw-r--r--
2026-03-03 04:22
test_time.py
42.44
KB
-rw-r--r--
2026-03-03 04:22
test_timeit.py
15.2
KB
-rw-r--r--
2026-03-03 04:22
test_timeout.py
10.75
KB
-rw-r--r--
2026-03-03 04:22
test_tix.py
1.05
KB
-rw-r--r--
2026-03-03 04:22
test_tk.py
546
B
-rw-r--r--
2026-03-03 04:22
test_tokenize.py
99.04
KB
-rw-r--r--
2026-03-03 04:22
test_trace.py
20.7
KB
-rw-r--r--
2026-03-03 04:22
test_traceback.py
117.68
KB
-rw-r--r--
2026-03-03 04:22
test_tracemalloc.py
39.54
KB
-rw-r--r--
2026-03-03 04:22
test_ttk_guionly.py
986
B
-rw-r--r--
2026-03-03 04:22
test_ttk_textonly.py
16.69
KB
-rw-r--r--
2026-03-03 04:22
test_tty.py
2.21
KB
-rw-r--r--
2026-03-03 04:22
test_tuple.py
19.8
KB
-rw-r--r--
2026-03-03 04:22
test_turtle.py
12.75
KB
-rw-r--r--
2026-03-03 04:22
test_type_annotations.py
6.22
KB
-rw-r--r--
2026-03-03 04:22
test_type_cache.py
5.92
KB
-rw-r--r--
2026-03-03 04:22
test_type_comments.py
10.54
KB
-rw-r--r--
2026-03-03 04:22
test_typechecks.py
2.55
KB
-rw-r--r--
2026-03-03 04:22
test_types.py
77.85
KB
-rw-r--r--
2026-03-03 04:22
test_typing.py
302
KB
-rw-r--r--
2026-03-03 04:22
test_ucn.py
9.52
KB
-rw-r--r--
2026-03-03 04:22
test_unary.py
1.63
KB
-rw-r--r--
2026-03-03 04:22
test_unicode.py
125.4
KB
-rw-r--r--
2026-03-03 04:22
test_unicode_file.py
5.72
KB
-rw-r--r--
2026-03-03 04:22
test_unicode_file_functions.py
6.91
KB
-rw-r--r--
2026-03-03 04:22
test_unicode_identifiers.py
997
B
-rw-r--r--
2026-03-03 04:22
test_unicodedata.py
16.26
KB
-rw-r--r--
2026-03-03 04:22
test_unittest.py
226
B
-rw-r--r--
2026-03-03 04:22
test_univnewlines.py
3.88
KB
-rw-r--r--
2026-03-03 04:22
test_unpack.py
3.51
KB
-rw-r--r--
2026-03-03 04:22
test_unpack_ex.py
9.89
KB
-rw-r--r--
2026-03-03 04:22
test_unparse.py
24.68
KB
-rw-r--r--
2026-03-03 04:22
test_urllib.py
70.53
KB
-rw-r--r--
2026-03-03 04:22
test_urllib2.py
79
KB
-rw-r--r--
2026-03-03 04:22
test_urllib2_localnet.py
25.59
KB
-rw-r--r--
2026-03-03 04:22
test_urllib2net.py
13.96
KB
-rw-r--r--
2026-03-03 04:22
test_urllib_response.py
2
KB
-rw-r--r--
2026-03-03 04:22
test_urllibnet.py
9.41
KB
-rw-r--r--
2026-03-03 04:22
test_urlparse.py
78.84
KB
-rw-r--r--
2026-03-03 04:22
test_userdict.py
7.56
KB
-rw-r--r--
2026-03-03 04:22
test_userlist.py
1.97
KB
-rw-r--r--
2026-03-03 04:22
test_userstring.py
2.52
KB
-rw-r--r--
2026-03-03 04:22
test_utf8_mode.py
10.31
KB
-rw-r--r--
2026-03-03 04:22
test_utf8source.py
1.08
KB
-rw-r--r--
2026-03-03 04:22
test_uu.py
9.2
KB
-rw-r--r--
2026-03-03 04:22
test_uuid.py
40.19
KB
-rw-r--r--
2026-03-03 04:22
test_venv.py
35.28
KB
-rw-r--r--
2026-03-03 04:22
test_wait3.py
1.75
KB
-rw-r--r--
2026-03-03 04:22
test_wait4.py
1.15
KB
-rw-r--r--
2026-03-03 04:22
test_wave.py
6.49
KB
-rw-r--r--
2026-03-03 04:22
test_weakref.py
74.32
KB
-rw-r--r--
2026-03-03 04:22
test_weakset.py
16.3
KB
-rw-r--r--
2026-03-03 04:22
test_webbrowser.py
10.67
KB
-rw-r--r--
2026-03-03 04:22
test_winconsoleio.py
6.69
KB
-rw-r--r--
2026-03-03 04:22
test_winreg.py
22.2
KB
-rw-r--r--
2026-03-03 04:22
test_winsound.py
4.61
KB
-rw-r--r--
2026-03-03 04:22
test_with.py
26.02
KB
-rw-r--r--
2026-03-03 04:22
test_wsgiref.py
30.25
KB
-rw-r--r--
2026-03-03 04:22
test_xdrlib.py
2.25
KB
-rw-r--r--
2026-03-03 04:22
test_xml_dom_minicompat.py
4.18
KB
-rw-r--r--
2026-03-03 04:22
test_xml_etree.py
160.54
KB
-rw-r--r--
2026-03-03 04:22
test_xml_etree_c.py
8.73
KB
-rw-r--r--
2026-03-03 04:22
test_xmlrpc.py
58.08
KB
-rw-r--r--
2026-03-03 04:22
test_xmlrpc_net.py
954
B
-rw-r--r--
2026-03-03 04:22
test_xxlimited.py
2.46
KB
-rw-r--r--
2026-03-03 04:22
test_xxtestfuzz.py
690
B
-rw-r--r--
2026-03-03 04:22
test_yield_from.py
30.09
KB
-rw-r--r--
2026-03-03 04:22
test_zipapp.py
16
KB
-rw-r--r--
2026-03-03 04:22
test_zipfile.py
160.41
KB
-rw-r--r--
2026-03-03 04:22
test_zipfile64.py
5.86
KB
-rw-r--r--
2026-03-03 04:22
test_zipimport.py
33.87
KB
-rw-r--r--
2026-03-03 04:22
test_zipimport_support.py
10.57
KB
-rw-r--r--
2026-03-03 04:22
test_zlib.py
35.67
KB
-rw-r--r--
2026-03-03 04:22
testcodec.py
1.02
KB
-rw-r--r--
2026-03-03 04:22
testtar.tar
425
KB
-rw-r--r--
2026-03-03 04:22
testtar.tar.xz
172
B
-rw-r--r--
2026-03-03 04:22
tf_inherit_check.py
714
B
-rw-r--r--
2026-03-03 04:22
time_hashlib.py
2.87
KB
-rw-r--r--
2026-03-03 04:22
win_console_handler.py
1.38
KB
-rw-r--r--
2026-03-03 04:22
xmltests.py
499
B
-rw-r--r--
2026-03-03 04:22
zip_cp437_header.zip
270
B
-rw-r--r--
2026-03-03 04:22
zipdir.zip
374
B
-rw-r--r--
2026-03-03 04:22
zipdir_backslash.zip
192
B
-rw-r--r--
2026-03-03 04:22
Save
Rename
""" Tests for the threading module. """ import test.support from test.support import threading_helper, requires_subprocess from test.support import verbose, cpython_only, os_helper from test.support.import_helper import import_module from test.support.script_helper import assert_python_ok, assert_python_failure import random import sys import _thread import threading import time import unittest import weakref import os import subprocess import signal import textwrap import traceback from unittest import mock from test import lock_tests from test import support threading_helper.requires_working_threading(module=True) # Between fork() and exec(), only async-safe functions are allowed (issues # #12316 and #11870), and fork() from a worker thread is known to trigger # problems with some operating systems (issue #3863): skip problematic tests # on platforms known to behave badly. platforms_to_skip = ('netbsd5', 'hp-ux11') # Is Python built with Py_DEBUG macro defined? Py_DEBUG = hasattr(sys, 'gettotalrefcount') def skip_unless_reliable_fork(test): if not support.has_fork_support: return unittest.skip("requires working os.fork()")(test) if sys.platform in platforms_to_skip: return unittest.skip("due to known OS bug related to thread+fork")(test) if support.HAVE_ASAN_FORK_BUG: return unittest.skip("libasan has a pthread_create() dead lock related to thread+fork")(test) return test def restore_default_excepthook(testcase): testcase.addCleanup(setattr, threading, 'excepthook', threading.excepthook) threading.excepthook = threading.__excepthook__ # A trivial mutable counter. class Counter(object): def __init__(self): self.value = 0 def inc(self): self.value += 1 def dec(self): self.value -= 1 def get(self): return self.value class TestThread(threading.Thread): def __init__(self, name, testcase, sema, mutex, nrunning): threading.Thread.__init__(self, name=name) self.testcase = testcase self.sema = sema self.mutex = mutex self.nrunning = nrunning def run(self): delay = random.random() / 10000.0 if verbose: print('task %s will run for %.1f usec' % (self.name, delay * 1e6)) with self.sema: with self.mutex: self.nrunning.inc() if verbose: print(self.nrunning.get(), 'tasks are running') self.testcase.assertLessEqual(self.nrunning.get(), 3) time.sleep(delay) if verbose: print('task', self.name, 'done') with self.mutex: self.nrunning.dec() self.testcase.assertGreaterEqual(self.nrunning.get(), 0) if verbose: print('%s is finished. %d tasks are running' % (self.name, self.nrunning.get())) class BaseTestCase(unittest.TestCase): def setUp(self): self._threads = threading_helper.threading_setup() def tearDown(self): threading_helper.threading_cleanup(*self._threads) test.support.reap_children() class ThreadTests(BaseTestCase): maxDiff = 9999 @cpython_only def test_name(self): def func(): pass thread = threading.Thread(name="myname1") self.assertEqual(thread.name, "myname1") # Convert int name to str thread = threading.Thread(name=123) self.assertEqual(thread.name, "123") # target name is ignored if name is specified thread = threading.Thread(target=func, name="myname2") self.assertEqual(thread.name, "myname2") with mock.patch.object(threading, '_counter', return_value=2): thread = threading.Thread(name="") self.assertEqual(thread.name, "Thread-2") with mock.patch.object(threading, '_counter', return_value=3): thread = threading.Thread() self.assertEqual(thread.name, "Thread-3") with mock.patch.object(threading, '_counter', return_value=5): thread = threading.Thread(target=func) self.assertEqual(thread.name, "Thread-5 (func)") def test_args_argument(self): # bpo-45735: Using list or tuple as *args* in constructor could # achieve the same effect. num_list = [1] num_tuple = (1,) str_list = ["str"] str_tuple = ("str",) list_in_tuple = ([1],) tuple_in_list = [(1,)] test_cases = ( (num_list, lambda arg: self.assertEqual(arg, 1)), (num_tuple, lambda arg: self.assertEqual(arg, 1)), (str_list, lambda arg: self.assertEqual(arg, "str")), (str_tuple, lambda arg: self.assertEqual(arg, "str")), (list_in_tuple, lambda arg: self.assertEqual(arg, [1])), (tuple_in_list, lambda arg: self.assertEqual(arg, (1,))) ) for args, target in test_cases: with self.subTest(target=target, args=args): t = threading.Thread(target=target, args=args) t.start() t.join() @cpython_only def test_disallow_instantiation(self): # Ensure that the type disallows instantiation (bpo-43916) lock = threading.Lock() test.support.check_disallow_instantiation(self, type(lock)) # Create a bunch of threads, let each do some work, wait until all are # done. def test_various_ops(self): # This takes about n/3 seconds to run (about n/3 clumps of tasks, # times about 1 second per clump). NUMTASKS = 10 # no more than 3 of the 10 can run at once sema = threading.BoundedSemaphore(value=3) mutex = threading.RLock() numrunning = Counter() threads = [] for i in range(NUMTASKS): t = TestThread("<thread %d>"%i, self, sema, mutex, numrunning) threads.append(t) self.assertIsNone(t.ident) self.assertRegex(repr(t), r'^<TestThread\(.*, initial\)>$') t.start() if hasattr(threading, 'get_native_id'): native_ids = set(t.native_id for t in threads) | {threading.get_native_id()} self.assertNotIn(None, native_ids) self.assertEqual(len(native_ids), NUMTASKS + 1) if verbose: print('waiting for all tasks to complete') for t in threads: t.join() self.assertFalse(t.is_alive()) self.assertNotEqual(t.ident, 0) self.assertIsNotNone(t.ident) self.assertRegex(repr(t), r'^<TestThread\(.*, stopped -?\d+\)>$') if verbose: print('all tasks done') self.assertEqual(numrunning.get(), 0) def test_ident_of_no_threading_threads(self): # The ident still must work for the main thread and dummy threads. self.assertIsNotNone(threading.current_thread().ident) def f(): ident.append(threading.current_thread().ident) done.set() done = threading.Event() ident = [] with threading_helper.wait_threads_exit(): tid = _thread.start_new_thread(f, ()) done.wait() self.assertEqual(ident[0], tid) # Kill the "immortal" _DummyThread del threading._active[ident[0]] # run with a small(ish) thread stack size (256 KiB) def test_various_ops_small_stack(self): if verbose: print('with 256 KiB thread stack size...') try: threading.stack_size(262144) except _thread.error: raise unittest.SkipTest( 'platform does not support changing thread stack size') self.test_various_ops() threading.stack_size(0) # run with a large thread stack size (1 MiB) def test_various_ops_large_stack(self): if verbose: print('with 1 MiB thread stack size...') try: threading.stack_size(0x100000) except _thread.error: raise unittest.SkipTest( 'platform does not support changing thread stack size') self.test_various_ops() threading.stack_size(0) def test_foreign_thread(self): # Check that a "foreign" thread can use the threading module. def f(mutex): # Calling current_thread() forces an entry for the foreign # thread to get made in the threading._active map. threading.current_thread() mutex.release() mutex = threading.Lock() mutex.acquire() with threading_helper.wait_threads_exit(): tid = _thread.start_new_thread(f, (mutex,)) # Wait for the thread to finish. mutex.acquire() self.assertIn(tid, threading._active) self.assertIsInstance(threading._active[tid], threading._DummyThread) #Issue 29376 self.assertTrue(threading._active[tid].is_alive()) self.assertRegex(repr(threading._active[tid]), '_DummyThread') del threading._active[tid] # PyThreadState_SetAsyncExc() is a CPython-only gimmick, not (currently) # exposed at the Python level. This test relies on ctypes to get at it. def test_PyThreadState_SetAsyncExc(self): ctypes = import_module("ctypes") set_async_exc = ctypes.pythonapi.PyThreadState_SetAsyncExc set_async_exc.argtypes = (ctypes.c_ulong, ctypes.py_object) class AsyncExc(Exception): pass exception = ctypes.py_object(AsyncExc) # First check it works when setting the exception from the same thread. tid = threading.get_ident() self.assertIsInstance(tid, int) self.assertGreater(tid, 0) try: result = set_async_exc(tid, exception) # The exception is async, so we might have to keep the VM busy until # it notices. while True: pass except AsyncExc: pass else: # This code is unreachable but it reflects the intent. If we wanted # to be smarter the above loop wouldn't be infinite. self.fail("AsyncExc not raised") try: self.assertEqual(result, 1) # one thread state modified except UnboundLocalError: # The exception was raised too quickly for us to get the result. pass # `worker_started` is set by the thread when it's inside a try/except # block waiting to catch the asynchronously set AsyncExc exception. # `worker_saw_exception` is set by the thread upon catching that # exception. worker_started = threading.Event() worker_saw_exception = threading.Event() class Worker(threading.Thread): def run(self): self.id = threading.get_ident() self.finished = False try: while True: worker_started.set() time.sleep(0.1) except AsyncExc: self.finished = True worker_saw_exception.set() t = Worker() t.daemon = True # so if this fails, we don't hang Python at shutdown t.start() if verbose: print(" started worker thread") # Try a thread id that doesn't make sense. if verbose: print(" trying nonsensical thread id") result = set_async_exc(-1, exception) self.assertEqual(result, 0) # no thread states modified # Now raise an exception in the worker thread. if verbose: print(" waiting for worker thread to get started") ret = worker_started.wait() self.assertTrue(ret) if verbose: print(" verifying worker hasn't exited") self.assertFalse(t.finished) if verbose: print(" attempting to raise asynch exception in worker") result = set_async_exc(t.id, exception) self.assertEqual(result, 1) # one thread state modified if verbose: print(" waiting for worker to say it caught the exception") worker_saw_exception.wait(timeout=support.SHORT_TIMEOUT) self.assertTrue(t.finished) if verbose: print(" all OK -- joining worker") if t.finished: t.join() # else the thread is still running, and we have no way to kill it def test_limbo_cleanup(self): # Issue 7481: Failure to start thread should cleanup the limbo map. def fail_new_thread(*args): raise threading.ThreadError() _start_new_thread = threading._start_new_thread threading._start_new_thread = fail_new_thread try: t = threading.Thread(target=lambda: None) self.assertRaises(threading.ThreadError, t.start) self.assertFalse( t in threading._limbo, "Failed to cleanup _limbo map on failure of Thread.start().") finally: threading._start_new_thread = _start_new_thread def test_finalize_running_thread(self): # Issue 1402: the PyGILState_Ensure / _Release functions may be called # very late on python exit: on deallocation of a running thread for # example. import_module("ctypes") rc, out, err = assert_python_failure("-c", """if 1: import ctypes, sys, time, _thread # This lock is used as a simple event variable. ready = _thread.allocate_lock() ready.acquire() # Module globals are cleared before __del__ is run # So we save the functions in class dict class C: ensure = ctypes.pythonapi.PyGILState_Ensure release = ctypes.pythonapi.PyGILState_Release def __del__(self): state = self.ensure() self.release(state) def waitingThread(): x = C() ready.release() time.sleep(100) _thread.start_new_thread(waitingThread, ()) ready.acquire() # Be sure the other thread is waiting. sys.exit(42) """) self.assertEqual(rc, 42) def test_finalize_with_trace(self): # Issue1733757 # Avoid a deadlock when sys.settrace steps into threading._shutdown assert_python_ok("-c", """if 1: import sys, threading # A deadlock-killer, to prevent the # testsuite to hang forever def killer(): import os, time time.sleep(2) print('program blocked; aborting') os._exit(2) t = threading.Thread(target=killer) t.daemon = True t.start() # This is the trace function def func(frame, event, arg): threading.current_thread() return func sys.settrace(func) """) def test_join_nondaemon_on_shutdown(self): # Issue 1722344 # Raising SystemExit skipped threading._shutdown rc, out, err = assert_python_ok("-c", """if 1: import threading from time import sleep def child(): sleep(1) # As a non-daemon thread we SHOULD wake up and nothing # should be torn down yet print("Woke up, sleep function is:", sleep) threading.Thread(target=child).start() raise SystemExit """) self.assertEqual(out.strip(), b"Woke up, sleep function is: <built-in function sleep>") self.assertEqual(err, b"") def test_enumerate_after_join(self): # Try hard to trigger #1703448: a thread is still returned in # threading.enumerate() after it has been join()ed. enum = threading.enumerate old_interval = sys.getswitchinterval() try: for i in range(1, 100): sys.setswitchinterval(i * 0.0002) t = threading.Thread(target=lambda: None) t.start() t.join() l = enum() self.assertNotIn(t, l, "#1703448 triggered after %d trials: %s" % (i, l)) finally: sys.setswitchinterval(old_interval) def test_no_refcycle_through_target(self): class RunSelfFunction(object): def __init__(self, should_raise): # The links in this refcycle from Thread back to self # should be cleaned up when the thread completes. self.should_raise = should_raise self.thread = threading.Thread(target=self._run, args=(self,), kwargs={'yet_another':self}) self.thread.start() def _run(self, other_ref, yet_another): if self.should_raise: raise SystemExit restore_default_excepthook(self) cyclic_object = RunSelfFunction(should_raise=False) weak_cyclic_object = weakref.ref(cyclic_object) cyclic_object.thread.join() del cyclic_object self.assertIsNone(weak_cyclic_object(), msg=('%d references still around' % sys.getrefcount(weak_cyclic_object()))) raising_cyclic_object = RunSelfFunction(should_raise=True) weak_raising_cyclic_object = weakref.ref(raising_cyclic_object) raising_cyclic_object.thread.join() del raising_cyclic_object self.assertIsNone(weak_raising_cyclic_object(), msg=('%d references still around' % sys.getrefcount(weak_raising_cyclic_object()))) def test_old_threading_api(self): # Just a quick sanity check to make sure the old method names are # still present t = threading.Thread() with self.assertWarnsRegex(DeprecationWarning, r'get the daemon attribute'): t.isDaemon() with self.assertWarnsRegex(DeprecationWarning, r'set the daemon attribute'): t.setDaemon(True) with self.assertWarnsRegex(DeprecationWarning, r'get the name attribute'): t.getName() with self.assertWarnsRegex(DeprecationWarning, r'set the name attribute'): t.setName("name") e = threading.Event() with self.assertWarnsRegex(DeprecationWarning, 'use is_set()'): e.isSet() cond = threading.Condition() cond.acquire() with self.assertWarnsRegex(DeprecationWarning, 'use notify_all()'): cond.notifyAll() with self.assertWarnsRegex(DeprecationWarning, 'use active_count()'): threading.activeCount() with self.assertWarnsRegex(DeprecationWarning, 'use current_thread()'): threading.currentThread() def test_repr_daemon(self): t = threading.Thread() self.assertNotIn('daemon', repr(t)) t.daemon = True self.assertIn('daemon', repr(t)) def test_daemon_param(self): t = threading.Thread() self.assertFalse(t.daemon) t = threading.Thread(daemon=False) self.assertFalse(t.daemon) t = threading.Thread(daemon=True) self.assertTrue(t.daemon) @skip_unless_reliable_fork def test_fork_at_exit(self): # bpo-42350: Calling os.fork() after threading._shutdown() must # not log an error. code = textwrap.dedent(""" import atexit import os import sys from test.support import wait_process # Import the threading module to register its "at fork" callback import threading def exit_handler(): pid = os.fork() if not pid: print("child process ok", file=sys.stderr, flush=True) # child process else: wait_process(pid, exitcode=0) # exit_handler() will be called after threading._shutdown() atexit.register(exit_handler) """) _, out, err = assert_python_ok("-c", code) self.assertEqual(out, b'') self.assertEqual(err.rstrip(), b'child process ok') @skip_unless_reliable_fork def test_dummy_thread_after_fork(self): # Issue #14308: a dummy thread in the active list doesn't mess up # the after-fork mechanism. code = """if 1: import _thread, threading, os, time def background_thread(evt): # Creates and registers the _DummyThread instance threading.current_thread() evt.set() time.sleep(10) evt = threading.Event() _thread.start_new_thread(background_thread, (evt,)) evt.wait() assert threading.active_count() == 2, threading.active_count() if os.fork() == 0: assert threading.active_count() == 1, threading.active_count() os._exit(0) else: os.wait() """ _, out, err = assert_python_ok("-c", code) self.assertEqual(out, b'') self.assertEqual(err, b'') @skip_unless_reliable_fork def test_is_alive_after_fork(self): # Try hard to trigger #18418: is_alive() could sometimes be True on # threads that vanished after a fork. old_interval = sys.getswitchinterval() self.addCleanup(sys.setswitchinterval, old_interval) # Make the bug more likely to manifest. test.support.setswitchinterval(1e-6) for i in range(20): t = threading.Thread(target=lambda: None) t.start() pid = os.fork() if pid == 0: os._exit(11 if t.is_alive() else 10) else: t.join() support.wait_process(pid, exitcode=10) def test_main_thread(self): main = threading.main_thread() self.assertEqual(main.name, 'MainThread') self.assertEqual(main.ident, threading.current_thread().ident) self.assertEqual(main.ident, threading.get_ident()) def f(): self.assertNotEqual(threading.main_thread().ident, threading.current_thread().ident) th = threading.Thread(target=f) th.start() th.join() @skip_unless_reliable_fork @unittest.skipUnless(hasattr(os, 'waitpid'), "test needs os.waitpid()") def test_main_thread_after_fork(self): code = """if 1: import os, threading from test import support ident = threading.get_ident() pid = os.fork() if pid == 0: print("current ident", threading.get_ident() == ident) main = threading.main_thread() print("main", main.name) print("main ident", main.ident == ident) print("current is main", threading.current_thread() is main) else: support.wait_process(pid, exitcode=0) """ _, out, err = assert_python_ok("-c", code) data = out.decode().replace('\r', '') self.assertEqual(err, b"") self.assertEqual(data, "current ident True\n" "main MainThread\n" "main ident True\n" "current is main True\n") @skip_unless_reliable_fork @unittest.skipUnless(hasattr(os, 'waitpid'), "test needs os.waitpid()") def test_main_thread_after_fork_from_nonmain_thread(self): code = """if 1: import os, threading, sys from test import support def func(): ident = threading.get_ident() pid = os.fork() if pid == 0: print("current ident", threading.get_ident() == ident) main = threading.main_thread() print("main", main.name, type(main).__name__) print("main ident", main.ident == ident) print("current is main", threading.current_thread() is main) # stdout is fully buffered because not a tty, # we have to flush before exit. sys.stdout.flush() th = threading.Thread(target=func) th.start() th.join() """ _, out, err = assert_python_ok("-c", code) data = out.decode().replace('\r', '') self.assertEqual(err.decode('utf-8'), "") self.assertEqual(data, "current ident True\n" "main Thread-1 (func) Thread\n" "main ident True\n" "current is main True\n" ) @unittest.skipIf(sys.platform in platforms_to_skip, "due to known OS bug") @support.requires_fork() @unittest.skipUnless(hasattr(os, 'waitpid'), "test needs os.waitpid()") def test_main_thread_after_fork_from_foreign_thread(self, create_dummy=False): code = """if 1: import os, threading, sys, traceback, _thread from test import support def func(lock): ident = threading.get_ident() if %s: # call current_thread() before fork to allocate DummyThread current = threading.current_thread() print("current", current.name, type(current).__name__) print("ident in _active", ident in threading._active) # flush before fork, so child won't flush it again sys.stdout.flush() pid = os.fork() if pid == 0: print("current ident", threading.get_ident() == ident) main = threading.main_thread() print("main", main.name, type(main).__name__) print("main ident", main.ident == ident) print("current is main", threading.current_thread() is main) print("_dangling", [t.name for t in list(threading._dangling)]) # stdout is fully buffered because not a tty, # we have to flush before exit. sys.stdout.flush() try: threading._shutdown() os._exit(0) except: traceback.print_exc() sys.stderr.flush() os._exit(1) else: try: support.wait_process(pid, exitcode=0) except Exception: # avoid 'could not acquire lock for # <_io.BufferedWriter name='<stderr>'> at interpreter shutdown,' traceback.print_exc() sys.stderr.flush() finally: lock.release() join_lock = _thread.allocate_lock() join_lock.acquire() th = _thread.start_new_thread(func, (join_lock,)) join_lock.acquire() """ % create_dummy # "DeprecationWarning: This process is multi-threaded, use of fork() # may lead to deadlocks in the child" _, out, err = assert_python_ok("-W", "ignore::DeprecationWarning", "-c", code) data = out.decode().replace('\r', '') self.assertEqual(err.decode(), "") self.assertEqual(data, ("current Dummy-1 _DummyThread\n" if create_dummy else "") + f"ident in _active {create_dummy!s}\n" + "current ident True\n" "main MainThread _MainThread\n" "main ident True\n" "current is main True\n" "_dangling ['MainThread']\n") def test_main_thread_after_fork_from_dummy_thread(self, create_dummy=False): self.test_main_thread_after_fork_from_foreign_thread(create_dummy=True) def test_main_thread_during_shutdown(self): # bpo-31516: current_thread() should still point to the main thread # at shutdown code = """if 1: import gc, threading main_thread = threading.current_thread() assert main_thread is threading.main_thread() # sanity check class RefCycle: def __init__(self): self.cycle = self def __del__(self): print("GC:", threading.current_thread() is main_thread, threading.main_thread() is main_thread, threading.enumerate() == [main_thread]) RefCycle() gc.collect() # sanity check x = RefCycle() """ _, out, err = assert_python_ok("-c", code) data = out.decode() self.assertEqual(err, b"") self.assertEqual(data.splitlines(), ["GC: True True True"] * 2) def test_finalization_shutdown(self): # bpo-36402: Py_Finalize() calls threading._shutdown() which must wait # until Python thread states of all non-daemon threads get deleted. # # Test similar to SubinterpThreadingTests.test_threads_join_2(), but # test the finalization of the main interpreter. code = """if 1: import os import threading import time import random def random_sleep(): seconds = random.random() * 0.010 time.sleep(seconds) class Sleeper: def __del__(self): random_sleep() tls = threading.local() def f(): # Sleep a bit so that the thread is still running when # Py_Finalize() is called. random_sleep() tls.x = Sleeper() random_sleep() threading.Thread(target=f).start() random_sleep() """ rc, out, err = assert_python_ok("-c", code) self.assertEqual(err, b"") def test_tstate_lock(self): # Test an implementation detail of Thread objects. started = _thread.allocate_lock() finish = _thread.allocate_lock() started.acquire() finish.acquire() def f(): started.release() finish.acquire() time.sleep(0.01) # The tstate lock is None until the thread is started t = threading.Thread(target=f) self.assertIs(t._tstate_lock, None) t.start() started.acquire() self.assertTrue(t.is_alive()) # The tstate lock can't be acquired when the thread is running # (or suspended). tstate_lock = t._tstate_lock self.assertFalse(tstate_lock.acquire(timeout=0), False) finish.release() # When the thread ends, the state_lock can be successfully # acquired. self.assertTrue(tstate_lock.acquire(timeout=support.SHORT_TIMEOUT), False) # But is_alive() is still True: we hold _tstate_lock now, which # prevents is_alive() from knowing the thread's end-of-life C code # is done. self.assertTrue(t.is_alive()) # Let is_alive() find out the C code is done. tstate_lock.release() self.assertFalse(t.is_alive()) # And verify the thread disposed of _tstate_lock. self.assertIsNone(t._tstate_lock) t.join() def test_repr_stopped(self): # Verify that "stopped" shows up in repr(Thread) appropriately. started = _thread.allocate_lock() finish = _thread.allocate_lock() started.acquire() finish.acquire() def f(): started.release() finish.acquire() t = threading.Thread(target=f) t.start() started.acquire() self.assertIn("started", repr(t)) finish.release() # "stopped" should appear in the repr in a reasonable amount of time. # Implementation detail: as of this writing, that's trivially true # if .join() is called, and almost trivially true if .is_alive() is # called. The detail we're testing here is that "stopped" shows up # "all on its own". LOOKING_FOR = "stopped" for i in range(500): if LOOKING_FOR in repr(t): break time.sleep(0.01) self.assertIn(LOOKING_FOR, repr(t)) # we waited at least 5 seconds t.join() def test_BoundedSemaphore_limit(self): # BoundedSemaphore should raise ValueError if released too often. for limit in range(1, 10): bs = threading.BoundedSemaphore(limit) threads = [threading.Thread(target=bs.acquire) for _ in range(limit)] for t in threads: t.start() for t in threads: t.join() threads = [threading.Thread(target=bs.release) for _ in range(limit)] for t in threads: t.start() for t in threads: t.join() self.assertRaises(ValueError, bs.release) @cpython_only def test_frame_tstate_tracing(self): # Issue #14432: Crash when a generator is created in a C thread that is # destroyed while the generator is still used. The issue was that a # generator contains a frame, and the frame kept a reference to the # Python state of the destroyed C thread. The crash occurs when a trace # function is setup. def noop_trace(frame, event, arg): # no operation return noop_trace def generator(): while 1: yield "generator" def callback(): if callback.gen is None: callback.gen = generator() return next(callback.gen) callback.gen = None old_trace = sys.gettrace() sys.settrace(noop_trace) try: # Install a trace function threading.settrace(noop_trace) # Create a generator in a C thread which exits after the call import _testcapi _testcapi.call_in_temporary_c_thread(callback) # Call the generator in a different Python thread, check that the # generator didn't keep a reference to the destroyed thread state for test in range(3): # The trace function is still called here callback() finally: sys.settrace(old_trace) def test_gettrace(self): def noop_trace(frame, event, arg): # no operation return noop_trace old_trace = threading.gettrace() try: threading.settrace(noop_trace) trace_func = threading.gettrace() self.assertEqual(noop_trace,trace_func) finally: threading.settrace(old_trace) def test_getprofile(self): def fn(*args): pass old_profile = threading.getprofile() try: threading.setprofile(fn) self.assertEqual(fn, threading.getprofile()) finally: threading.setprofile(old_profile) @cpython_only def test_shutdown_locks(self): for daemon in (False, True): with self.subTest(daemon=daemon): event = threading.Event() thread = threading.Thread(target=event.wait, daemon=daemon) # Thread.start() must add lock to _shutdown_locks, # but only for non-daemon thread thread.start() tstate_lock = thread._tstate_lock if not daemon: self.assertIn(tstate_lock, threading._shutdown_locks) else: self.assertNotIn(tstate_lock, threading._shutdown_locks) # unblock the thread and join it event.set() thread.join() # Thread._stop() must remove tstate_lock from _shutdown_locks. # Daemon threads must never add it to _shutdown_locks. self.assertNotIn(tstate_lock, threading._shutdown_locks) def test_locals_at_exit(self): # bpo-19466: thread locals must not be deleted before destructors # are called rc, out, err = assert_python_ok("-c", """if 1: import threading class Atexit: def __del__(self): print("thread_dict.atexit = %r" % thread_dict.atexit) thread_dict = threading.local() thread_dict.atexit = "value" atexit = Atexit() """) self.assertEqual(out.rstrip(), b"thread_dict.atexit = 'value'") def test_boolean_target(self): # bpo-41149: A thread that had a boolean value of False would not # run, regardless of whether it was callable. The correct behaviour # is for a thread to do nothing if its target is None, and to call # the target otherwise. class BooleanTarget(object): def __init__(self): self.ran = False def __bool__(self): return False def __call__(self): self.ran = True target = BooleanTarget() thread = threading.Thread(target=target) thread.start() thread.join() self.assertTrue(target.ran) def test_leak_without_join(self): # bpo-37788: Test that a thread which is not joined explicitly # does not leak. Test written for reference leak checks. def noop(): pass with threading_helper.wait_threads_exit(): threading.Thread(target=noop).start() # Thread.join() is not called @unittest.skipUnless(Py_DEBUG, 'need debug build (Py_DEBUG)') def test_debug_deprecation(self): # bpo-44584: The PYTHONTHREADDEBUG environment variable is deprecated rc, out, err = assert_python_ok("-Wdefault", "-c", "pass", PYTHONTHREADDEBUG="1") msg = (b'DeprecationWarning: The threading debug ' b'(PYTHONTHREADDEBUG environment variable) ' b'is deprecated and will be removed in Python 3.12') self.assertIn(msg, err) def test_import_from_another_thread(self): # bpo-1596321: If the threading module is first import from a thread # different than the main thread, threading._shutdown() must handle # this case without logging an error at Python exit. code = textwrap.dedent(''' import _thread import sys event = _thread.allocate_lock() event.acquire() def import_threading(): import threading event.release() if 'threading' in sys.modules: raise Exception('threading is already imported') _thread.start_new_thread(import_threading, ()) # wait until the threading module is imported event.acquire() event.release() if 'threading' not in sys.modules: raise Exception('threading is not imported') # don't wait until the thread completes ''') rc, out, err = assert_python_ok("-c", code) self.assertEqual(out, b'') self.assertEqual(err, b'') class ThreadJoinOnShutdown(BaseTestCase): def _run_and_join(self, script): script = """if 1: import sys, os, time, threading # a thread, which waits for the main program to terminate def joiningfunc(mainthread): mainthread.join() print('end of thread') # stdout is fully buffered because not a tty, we have to flush # before exit. sys.stdout.flush() \n""" + script rc, out, err = assert_python_ok("-c", script) data = out.decode().replace('\r', '') self.assertEqual(data, "end of main\nend of thread\n") def test_1_join_on_shutdown(self): # The usual case: on exit, wait for a non-daemon thread script = """if 1: import os t = threading.Thread(target=joiningfunc, args=(threading.current_thread(),)) t.start() time.sleep(0.1) print('end of main') """ self._run_and_join(script) @skip_unless_reliable_fork def test_2_join_in_forked_process(self): # Like the test above, but from a forked interpreter script = """if 1: from test import support childpid = os.fork() if childpid != 0: # parent process support.wait_process(childpid, exitcode=0) sys.exit(0) # child process t = threading.Thread(target=joiningfunc, args=(threading.current_thread(),)) t.start() print('end of main') """ self._run_and_join(script) @skip_unless_reliable_fork def test_3_join_in_forked_from_thread(self): # Like the test above, but fork() was called from a worker thread # In the forked process, the main Thread object must be marked as stopped. script = """if 1: from test import support main_thread = threading.current_thread() def worker(): childpid = os.fork() if childpid != 0: # parent process support.wait_process(childpid, exitcode=0) sys.exit(0) # child process t = threading.Thread(target=joiningfunc, args=(main_thread,)) print('end of main') t.start() t.join() # Should not block: main_thread is already stopped w = threading.Thread(target=worker) w.start() """ self._run_and_join(script) @unittest.skipIf(sys.platform in platforms_to_skip, "due to known OS bug") def test_4_daemon_threads(self): # Check that a daemon thread cannot crash the interpreter on shutdown # by manipulating internal structures that are being disposed of in # the main thread. script = """if True: import os import random import sys import time import threading thread_has_run = set() def random_io(): '''Loop for a while sleeping random tiny amounts and doing some I/O.''' import test.test_threading as mod while True: with open(mod.__file__, 'rb') as in_f: stuff = in_f.read(200) with open(os.devnull, 'wb') as null_f: null_f.write(stuff) time.sleep(random.random() / 1995) thread_has_run.add(threading.current_thread()) def main(): count = 0 for _ in range(40): new_thread = threading.Thread(target=random_io) new_thread.daemon = True new_thread.start() count += 1 while len(thread_has_run) < count: time.sleep(0.001) # Trigger process shutdown sys.exit(0) main() """ rc, out, err = assert_python_ok('-c', script) self.assertFalse(err) @skip_unless_reliable_fork def test_reinit_tls_after_fork(self): # Issue #13817: fork() would deadlock in a multithreaded program with # the ad-hoc TLS implementation. def do_fork_and_wait(): # just fork a child process and wait it pid = os.fork() if pid > 0: support.wait_process(pid, exitcode=50) else: os._exit(50) # start a bunch of threads that will fork() child processes threads = [] for i in range(16): t = threading.Thread(target=do_fork_and_wait) threads.append(t) t.start() for t in threads: t.join() @skip_unless_reliable_fork def test_clear_threads_states_after_fork(self): # Issue #17094: check that threads states are cleared after fork() # start a bunch of threads threads = [] for i in range(16): t = threading.Thread(target=lambda : time.sleep(0.3)) threads.append(t) t.start() pid = os.fork() if pid == 0: # check that threads states have been cleared if len(sys._current_frames()) == 1: os._exit(51) else: os._exit(52) else: support.wait_process(pid, exitcode=51) for t in threads: t.join() class SubinterpThreadingTests(BaseTestCase): def pipe(self): r, w = os.pipe() self.addCleanup(os.close, r) self.addCleanup(os.close, w) if hasattr(os, 'set_blocking'): os.set_blocking(r, False) return (r, w) def test_threads_join(self): # Non-daemon threads should be joined at subinterpreter shutdown # (issue #18808) r, w = self.pipe() code = textwrap.dedent(r""" import os import random import threading import time def random_sleep(): seconds = random.random() * 0.010 time.sleep(seconds) def f(): # Sleep a bit so that the thread is still running when # Py_EndInterpreter is called. random_sleep() os.write(%d, b"x") threading.Thread(target=f).start() random_sleep() """ % (w,)) ret = test.support.run_in_subinterp(code) self.assertEqual(ret, 0) # The thread was joined properly. self.assertEqual(os.read(r, 1), b"x") def test_threads_join_2(self): # Same as above, but a delay gets introduced after the thread's # Python code returned but before the thread state is deleted. # To achieve this, we register a thread-local object which sleeps # a bit when deallocated. r, w = self.pipe() code = textwrap.dedent(r""" import os import random import threading import time def random_sleep(): seconds = random.random() * 0.010 time.sleep(seconds) class Sleeper: def __del__(self): random_sleep() tls = threading.local() def f(): # Sleep a bit so that the thread is still running when # Py_EndInterpreter is called. random_sleep() tls.x = Sleeper() os.write(%d, b"x") threading.Thread(target=f).start() random_sleep() """ % (w,)) ret = test.support.run_in_subinterp(code) self.assertEqual(ret, 0) # The thread was joined properly. self.assertEqual(os.read(r, 1), b"x") @cpython_only def test_daemon_threads_fatal_error(self): subinterp_code = f"""if 1: import os import threading import time def f(): # Make sure the daemon thread is still running when # Py_EndInterpreter is called. time.sleep({test.support.SHORT_TIMEOUT}) threading.Thread(target=f, daemon=True).start() """ script = r"""if 1: import _testcapi _testcapi.run_in_subinterp(%r) """ % (subinterp_code,) with test.support.SuppressCrashReport(): rc, out, err = assert_python_failure("-c", script) self.assertIn("Fatal Python error: Py_EndInterpreter: " "not the last thread", err.decode()) class ThreadingExceptionTests(BaseTestCase): # A RuntimeError should be raised if Thread.start() is called # multiple times. def test_start_thread_again(self): thread = threading.Thread() thread.start() self.assertRaises(RuntimeError, thread.start) thread.join() def test_joining_current_thread(self): current_thread = threading.current_thread() self.assertRaises(RuntimeError, current_thread.join); def test_joining_inactive_thread(self): thread = threading.Thread() self.assertRaises(RuntimeError, thread.join) def test_daemonize_active_thread(self): thread = threading.Thread() thread.start() self.assertRaises(RuntimeError, setattr, thread, "daemon", True) thread.join() def test_releasing_unacquired_lock(self): lock = threading.Lock() self.assertRaises(RuntimeError, lock.release) @requires_subprocess() def test_recursion_limit(self): # Issue 9670 # test that excessive recursion within a non-main thread causes # an exception rather than crashing the interpreter on platforms # like Mac OS X or FreeBSD which have small default stack sizes # for threads script = """if True: import threading def recurse(): return recurse() def outer(): try: recurse() except RecursionError: pass w = threading.Thread(target=outer) w.start() w.join() print('end of main thread') """ expected_output = "end of main thread\n" p = subprocess.Popen([sys.executable, "-c", script], stdout=subprocess.PIPE, stderr=subprocess.PIPE) stdout, stderr = p.communicate() data = stdout.decode().replace('\r', '') self.assertEqual(p.returncode, 0, "Unexpected error: " + stderr.decode()) self.assertEqual(data, expected_output) def test_print_exception(self): script = r"""if True: import threading import time running = False def run(): global running running = True while running: time.sleep(0.01) 1/0 t = threading.Thread(target=run) t.start() while not running: time.sleep(0.01) running = False t.join() """ rc, out, err = assert_python_ok("-c", script) self.assertEqual(out, b'') err = err.decode() self.assertIn("Exception in thread", err) self.assertIn("Traceback (most recent call last):", err) self.assertIn("ZeroDivisionError", err) self.assertNotIn("Unhandled exception", err) def test_print_exception_stderr_is_none_1(self): script = r"""if True: import sys import threading import time running = False def run(): global running running = True while running: time.sleep(0.01) 1/0 t = threading.Thread(target=run) t.start() while not running: time.sleep(0.01) sys.stderr = None running = False t.join() """ rc, out, err = assert_python_ok("-c", script) self.assertEqual(out, b'') err = err.decode() self.assertIn("Exception in thread", err) self.assertIn("Traceback (most recent call last):", err) self.assertIn("ZeroDivisionError", err) self.assertNotIn("Unhandled exception", err) def test_print_exception_stderr_is_none_2(self): script = r"""if True: import sys import threading import time running = False def run(): global running running = True while running: time.sleep(0.01) 1/0 sys.stderr = None t = threading.Thread(target=run) t.start() while not running: time.sleep(0.01) running = False t.join() """ rc, out, err = assert_python_ok("-c", script) self.assertEqual(out, b'') self.assertNotIn("Unhandled exception", err.decode()) def test_bare_raise_in_brand_new_thread(self): def bare_raise(): raise class Issue27558(threading.Thread): exc = None def run(self): try: bare_raise() except Exception as exc: self.exc = exc thread = Issue27558() thread.start() thread.join() self.assertIsNotNone(thread.exc) self.assertIsInstance(thread.exc, RuntimeError) # explicitly break the reference cycle to not leak a dangling thread thread.exc = None def test_multithread_modify_file_noerror(self): # See issue25872 def modify_file(): with open(os_helper.TESTFN, 'w', encoding='utf-8') as fp: fp.write(' ') traceback.format_stack() self.addCleanup(os_helper.unlink, os_helper.TESTFN) threads = [ threading.Thread(target=modify_file) for i in range(100) ] for t in threads: t.start() t.join() class ThreadRunFail(threading.Thread): def run(self): raise ValueError("run failed") class ExceptHookTests(BaseTestCase): def setUp(self): restore_default_excepthook(self) super().setUp() def test_excepthook(self): with support.captured_output("stderr") as stderr: thread = ThreadRunFail(name="excepthook thread") thread.start() thread.join() stderr = stderr.getvalue().strip() self.assertIn(f'Exception in thread {thread.name}:\n', stderr) self.assertIn('Traceback (most recent call last):\n', stderr) self.assertIn(' raise ValueError("run failed")', stderr) self.assertIn('ValueError: run failed', stderr) @support.cpython_only def test_excepthook_thread_None(self): # threading.excepthook called with thread=None: log the thread # identifier in this case. with support.captured_output("stderr") as stderr: try: raise ValueError("bug") except Exception as exc: args = threading.ExceptHookArgs([*sys.exc_info(), None]) try: threading.excepthook(args) finally: # Explicitly break a reference cycle args = None stderr = stderr.getvalue().strip() self.assertIn(f'Exception in thread {threading.get_ident()}:\n', stderr) self.assertIn('Traceback (most recent call last):\n', stderr) self.assertIn(' raise ValueError("bug")', stderr) self.assertIn('ValueError: bug', stderr) def test_system_exit(self): class ThreadExit(threading.Thread): def run(self): sys.exit(1) # threading.excepthook() silently ignores SystemExit with support.captured_output("stderr") as stderr: thread = ThreadExit() thread.start() thread.join() self.assertEqual(stderr.getvalue(), '') def test_custom_excepthook(self): args = None def hook(hook_args): nonlocal args args = hook_args try: with support.swap_attr(threading, 'excepthook', hook): thread = ThreadRunFail() thread.start() thread.join() self.assertEqual(args.exc_type, ValueError) self.assertEqual(str(args.exc_value), 'run failed') self.assertEqual(args.exc_traceback, args.exc_value.__traceback__) self.assertIs(args.thread, thread) finally: # Break reference cycle args = None def test_custom_excepthook_fail(self): def threading_hook(args): raise ValueError("threading_hook failed") err_str = None def sys_hook(exc_type, exc_value, exc_traceback): nonlocal err_str err_str = str(exc_value) with support.swap_attr(threading, 'excepthook', threading_hook), \ support.swap_attr(sys, 'excepthook', sys_hook), \ support.captured_output('stderr') as stderr: thread = ThreadRunFail() thread.start() thread.join() self.assertEqual(stderr.getvalue(), 'Exception in threading.excepthook:\n') self.assertEqual(err_str, 'threading_hook failed') def test_original_excepthook(self): def run_thread(): with support.captured_output("stderr") as output: thread = ThreadRunFail(name="excepthook thread") thread.start() thread.join() return output.getvalue() def threading_hook(args): print("Running a thread failed", file=sys.stderr) default_output = run_thread() with support.swap_attr(threading, 'excepthook', threading_hook): custom_hook_output = run_thread() threading.excepthook = threading.__excepthook__ recovered_output = run_thread() self.assertEqual(default_output, recovered_output) self.assertNotEqual(default_output, custom_hook_output) self.assertEqual(custom_hook_output, "Running a thread failed\n") class TimerTests(BaseTestCase): def setUp(self): BaseTestCase.setUp(self) self.callback_args = [] self.callback_event = threading.Event() def test_init_immutable_default_args(self): # Issue 17435: constructor defaults were mutable objects, they could be # mutated via the object attributes and affect other Timer objects. timer1 = threading.Timer(0.01, self._callback_spy) timer1.start() self.callback_event.wait() timer1.args.append("blah") timer1.kwargs["foo"] = "bar" self.callback_event.clear() timer2 = threading.Timer(0.01, self._callback_spy) timer2.start() self.callback_event.wait() self.assertEqual(len(self.callback_args), 2) self.assertEqual(self.callback_args, [((), {}), ((), {})]) timer1.join() timer2.join() def _callback_spy(self, *args, **kwargs): self.callback_args.append((args[:], kwargs.copy())) self.callback_event.set() class LockTests(lock_tests.LockTests): locktype = staticmethod(threading.Lock) class PyRLockTests(lock_tests.RLockTests): locktype = staticmethod(threading._PyRLock) @unittest.skipIf(threading._CRLock is None, 'RLock not implemented in C') class CRLockTests(lock_tests.RLockTests): locktype = staticmethod(threading._CRLock) class EventTests(lock_tests.EventTests): eventtype = staticmethod(threading.Event) class ConditionAsRLockTests(lock_tests.RLockTests): # Condition uses an RLock by default and exports its API. locktype = staticmethod(threading.Condition) def test_recursion_count(self): self.skipTest("Condition does not expose _recursion_count()") class ConditionTests(lock_tests.ConditionTests): condtype = staticmethod(threading.Condition) class SemaphoreTests(lock_tests.SemaphoreTests): semtype = staticmethod(threading.Semaphore) class BoundedSemaphoreTests(lock_tests.BoundedSemaphoreTests): semtype = staticmethod(threading.BoundedSemaphore) class BarrierTests(lock_tests.BarrierTests): barriertype = staticmethod(threading.Barrier) class MiscTestCase(unittest.TestCase): def test__all__(self): restore_default_excepthook(self) extra = {"ThreadError"} not_exported = {'currentThread', 'activeCount'} support.check__all__(self, threading, ('threading', '_thread'), extra=extra, not_exported=not_exported) class InterruptMainTests(unittest.TestCase): def check_interrupt_main_with_signal_handler(self, signum): def handler(signum, frame): 1/0 old_handler = signal.signal(signum, handler) self.addCleanup(signal.signal, signum, old_handler) with self.assertRaises(ZeroDivisionError): _thread.interrupt_main() def check_interrupt_main_noerror(self, signum): handler = signal.getsignal(signum) try: # No exception should arise. signal.signal(signum, signal.SIG_IGN) _thread.interrupt_main(signum) signal.signal(signum, signal.SIG_DFL) _thread.interrupt_main(signum) finally: # Restore original handler signal.signal(signum, handler) def test_interrupt_main_subthread(self): # Calling start_new_thread with a function that executes interrupt_main # should raise KeyboardInterrupt upon completion. def call_interrupt(): _thread.interrupt_main() t = threading.Thread(target=call_interrupt) with self.assertRaises(KeyboardInterrupt): t.start() t.join() t.join() def test_interrupt_main_mainthread(self): # Make sure that if interrupt_main is called in main thread that # KeyboardInterrupt is raised instantly. with self.assertRaises(KeyboardInterrupt): _thread.interrupt_main() def test_interrupt_main_with_signal_handler(self): self.check_interrupt_main_with_signal_handler(signal.SIGINT) self.check_interrupt_main_with_signal_handler(signal.SIGTERM) def test_interrupt_main_noerror(self): self.check_interrupt_main_noerror(signal.SIGINT) self.check_interrupt_main_noerror(signal.SIGTERM) def test_interrupt_main_invalid_signal(self): self.assertRaises(ValueError, _thread.interrupt_main, -1) self.assertRaises(ValueError, _thread.interrupt_main, signal.NSIG) self.assertRaises(ValueError, _thread.interrupt_main, 1000000) @threading_helper.reap_threads def test_can_interrupt_tight_loops(self): cont = [True] started = [False] interrupted = [False] def worker(started, cont, interrupted): iterations = 100_000_000 started[0] = True while cont[0]: if iterations: iterations -= 1 else: return pass interrupted[0] = True t = threading.Thread(target=worker,args=(started, cont, interrupted)) t.start() while not started[0]: pass cont[0] = False t.join() self.assertTrue(interrupted[0]) class AtexitTests(unittest.TestCase): def test_atexit_output(self): rc, out, err = assert_python_ok("-c", """if True: import threading def run_last(): print('parrot') threading._register_atexit(run_last) """) self.assertFalse(err) self.assertEqual(out.strip(), b'parrot') def test_atexit_called_once(self): rc, out, err = assert_python_ok("-c", """if True: import threading from unittest.mock import Mock mock = Mock() threading._register_atexit(mock) mock.assert_not_called() # force early shutdown to ensure it was called once threading._shutdown() mock.assert_called_once() """) self.assertFalse(err) def test_atexit_after_shutdown(self): # The only way to do this is by registering an atexit within # an atexit, which is intended to raise an exception. rc, out, err = assert_python_ok("-c", """if True: import threading def func(): pass def run_last(): threading._register_atexit(func) threading._register_atexit(run_last) """) self.assertTrue(err) self.assertIn("RuntimeError: can't register atexit after shutdown", err.decode()) if __name__ == "__main__": unittest.main()