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.216.198
Domains :
Cant Read [ /etc/named.conf ]
User : beriska1
Terminal
Auto Root
Create File
Create Folder
Localroot Suggester
Backdoor Destroyer
Readme
/
opt /
alt /
python313 /
lib64 /
python3.13 /
test /
Delete
Unzip
Name
Size
Permission
Date
Action
__pycache__
[ DIR ]
drwxr-xr-x
2026-05-05 23:06
archivetestdata
[ DIR ]
drwxr-xr-x
2026-05-05 23:06
audiodata
[ DIR ]
drwxr-xr-x
2026-05-05 23:06
certdata
[ DIR ]
drwxr-xr-x
2026-05-05 23:06
cjkencodings
[ DIR ]
drwxr-xr-x
2026-05-05 23:06
configdata
[ DIR ]
drwxr-xr-x
2026-05-05 23:06
crashers
[ DIR ]
drwxr-xr-x
2026-05-05 23:06
data
[ DIR ]
drwxr-xr-x
2026-05-05 23:06
decimaltestdata
[ DIR ]
drwxr-xr-x
2026-05-05 23:06
dtracedata
[ DIR ]
drwxr-xr-x
2026-05-05 23:06
encoded_modules
[ DIR ]
drwxr-xr-x
2026-05-05 23:06
leakers
[ DIR ]
drwxr-xr-x
2026-05-05 23:06
libregrtest
[ DIR ]
drwxr-xr-x
2026-05-05 23:06
mathdata
[ DIR ]
drwxr-xr-x
2026-05-05 23:06
regrtestdata
[ DIR ]
drwxr-xr-x
2026-03-26 17:37
subprocessdata
[ DIR ]
drwxr-xr-x
2026-05-05 23:06
support
[ DIR ]
drwxr-xr-x
2026-05-05 23:06
test_ast
[ DIR ]
drwxr-xr-x
2026-05-05 23:06
test_asyncio
[ DIR ]
drwxr-xr-x
2026-05-05 23:06
test_capi
[ DIR ]
drwxr-xr-x
2026-05-05 23:06
test_cext
[ DIR ]
drwxr-xr-x
2026-05-05 23:06
test_concurrent_futures
[ DIR ]
drwxr-xr-x
2026-05-05 23:06
test_cppext
[ DIR ]
drwxr-xr-x
2026-05-05 23:06
test_ctypes
[ DIR ]
drwxr-xr-x
2026-05-05 23:06
test_dataclasses
[ DIR ]
drwxr-xr-x
2026-05-05 23:06
test_doctest
[ DIR ]
drwxr-xr-x
2026-05-05 23:06
test_email
[ DIR ]
drwxr-xr-x
2026-05-05 23:06
test_free_threading
[ DIR ]
drwxr-xr-x
2026-05-05 23:06
test_future_stmt
[ DIR ]
drwxr-xr-x
2026-05-05 23:06
test_gdb
[ DIR ]
drwxr-xr-x
2026-05-05 23:06
test_import
[ DIR ]
drwxr-xr-x
2026-05-05 23:06
test_importlib
[ DIR ]
drwxr-xr-x
2026-05-05 23:06
test_inspect
[ DIR ]
drwxr-xr-x
2026-05-05 23:06
test_interpreters
[ DIR ]
drwxr-xr-x
2026-05-05 23:06
test_json
[ DIR ]
drwxr-xr-x
2026-05-05 23:06
test_module
[ DIR ]
drwxr-xr-x
2026-05-05 23:06
test_multiprocessing_fork
[ DIR ]
drwxr-xr-x
2026-05-05 23:06
test_multiprocessing_forkserver
[ DIR ]
drwxr-xr-x
2026-05-05 23:06
test_multiprocessing_spawn
[ DIR ]
drwxr-xr-x
2026-05-05 23:06
test_pathlib
[ DIR ]
drwxr-xr-x
2026-05-05 23:06
test_peg_generator
[ DIR ]
drwxr-xr-x
2026-05-05 23:06
test_pydoc
[ DIR ]
drwxr-xr-x
2026-05-05 23:06
test_pyrepl
[ DIR ]
drwxr-xr-x
2026-05-05 23:06
test_sqlite3
[ DIR ]
drwxr-xr-x
2026-05-05 23:06
test_tkinter
[ DIR ]
drwxr-xr-x
2026-05-05 23:06
test_tomllib
[ DIR ]
drwxr-xr-x
2026-05-05 23:06
test_tools
[ DIR ]
drwxr-xr-x
2026-05-05 23:06
test_ttk
[ DIR ]
drwxr-xr-x
2026-05-05 23:06
test_unittest
[ DIR ]
drwxr-xr-x
2026-05-05 23:06
test_warnings
[ DIR ]
drwxr-xr-x
2026-05-05 23:06
test_zipfile
[ DIR ]
drwxr-xr-x
2026-05-05 23:06
test_zoneinfo
[ DIR ]
drwxr-xr-x
2026-05-05 23:06
tkinterdata
[ DIR ]
drwxr-xr-x
2026-05-05 23:06
tokenizedata
[ DIR ]
drwxr-xr-x
2026-05-05 23:06
tracedmodules
[ DIR ]
drwxr-xr-x
2026-05-05 23:06
translationdata
[ DIR ]
drwxr-xr-x
2026-03-26 17:37
typinganndata
[ DIR ]
drwxr-xr-x
2026-05-05 23:06
wheeldata
[ DIR ]
drwxr-xr-x
2026-05-05 23:06
xmltestdata
[ DIR ]
drwxr-xr-x
2026-05-05 23:06
zipimport_data
[ DIR ]
drwxr-xr-x
2026-05-05 23:06
NormalizationTest-3.2.0.txt
1.93
MB
-rw-r--r--
2026-02-03 21:23
__init__.py
47
B
-rw-r--r--
2026-03-26 17:22
__main__.py
67
B
-rw-r--r--
2026-03-26 17:22
_test_atexit.py
5.32
KB
-rw-r--r--
2026-03-26 17:22
_test_eintr.py
17.83
KB
-rw-r--r--
2026-03-26 17:22
_test_embed_set_config.py
9.32
KB
-rw-r--r--
2026-03-26 17:22
_test_embed_structseq.py
1.99
KB
-rw-r--r--
2026-03-26 17:22
_test_monitoring_shutdown.py
704
B
-rw-r--r--
2026-03-26 17:39
_test_multiprocessing.py
228.26
KB
-rw-r--r--
2026-03-26 17:22
_test_venv_multiprocessing.py
796
B
-rw-r--r--
2026-03-26 17:22
archiver_tests.py
7.06
KB
-rw-r--r--
2026-03-26 17:22
audiotests.py
12.13
KB
-rw-r--r--
2026-03-26 17:22
audit-tests.py
14.81
KB
-rw-r--r--
2026-03-26 17:22
autotest.py
214
B
-rw-r--r--
2026-03-26 17:22
bisect_cmd.py
5.36
KB
-rwxr-xr-x
2026-03-26 17:22
clinic.test.c
161.63
KB
-rw-r--r--
2026-02-03 21:23
cov.py
1.26
KB
-rw-r--r--
2026-03-26 17:22
curses_tests.py
1.23
KB
-rwxr-xr-x
2026-03-26 17:22
datetimetester.py
274.67
KB
-rw-r--r--
2026-03-26 17:22
dis_module.py
76
B
-rw-r--r--
2026-03-26 17:22
empty.vbs
70
B
-rw-r--r--
2026-02-03 21:23
exception_hierarchy.txt
2.38
KB
-rw-r--r--
2026-02-03 21:23
fork_wait.py
2.29
KB
-rw-r--r--
2026-03-26 17:22
levenshtein_examples.json
406.44
KB
-rw-r--r--
2026-02-03 21:23
list_tests.py
17.11
KB
-rw-r--r--
2026-03-26 17:22
lock_tests.py
35.64
KB
-rw-r--r--
2026-03-26 17:22
mapping_tests.py
21.87
KB
-rw-r--r--
2026-03-26 17:22
memory_watchdog.py
711
B
-rw-r--r--
2026-03-26 17:22
mime.types
47.37
KB
-rw-r--r--
2026-02-03 21:23
mock_socket.py
3.69
KB
-rw-r--r--
2026-03-26 17:22
mp_fork_bomb.py
448
B
-rw-r--r--
2026-03-26 17:22
mp_preload.py
351
B
-rw-r--r--
2026-03-26 17:22
mp_preload_flush.py
291
B
-rw-r--r--
2026-03-26 17:22
mp_preload_main.py
287
B
-rw-r--r--
2026-03-26 17:22
mp_preload_sysargv.py
567
B
-rw-r--r--
2026-03-26 17:22
multibytecodec_support.py
14.91
KB
-rw-r--r--
2026-03-26 17:22
picklecommon.py
9.4
KB
-rw-r--r--
2026-03-26 17:22
pickletester.py
179.95
KB
-rw-r--r--
2026-03-26 17:22
profilee.py
2.97
KB
-rw-r--r--
2026-03-26 17:22
pstats.pck
65.05
KB
-rw-r--r--
2026-02-03 21:23
pyclbr_input.py
1.63
KB
-rw-r--r--
2026-03-26 17:22
pythoninfo.py
28.65
KB
-rw-r--r--
2026-03-26 17:22
randv2_32.pck
7.34
KB
-rw-r--r--
2026-02-03 21:23
randv2_64.pck
7.19
KB
-rw-r--r--
2026-02-03 21:23
randv3.pck
7.82
KB
-rw-r--r--
2026-02-03 21:23
re_tests.py
25.94
KB
-rwxr-xr-x
2026-03-26 17:22
regrtest.py
1.28
KB
-rwxr-xr-x
2026-03-26 17:22
relimport.py
27
B
-rw-r--r--
2026-03-26 17:22
reperf.py
538
B
-rw-r--r--
2026-03-26 17:22
seq_tests.py
15
KB
-rw-r--r--
2026-03-26 17:22
signalinterproctester.py
3.12
KB
-rw-r--r--
2026-03-26 17:22
ssl_servers.py
7.12
KB
-rw-r--r--
2026-03-26 17:22
ssltests.py
1.01
KB
-rw-r--r--
2026-03-26 17:22
string_tests.py
72.29
KB
-rw-r--r--
2026-03-26 17:22
test___all__.py
4.89
KB
-rw-r--r--
2026-03-26 17:22
test__colorize.py
6.51
KB
-rw-r--r--
2026-03-26 17:22
test__interpchannels.py
61.88
KB
-rw-r--r--
2026-03-26 17:22
test__interpreters.py
36.56
KB
-rw-r--r--
2026-03-26 17:22
test__locale.py
12.52
KB
-rw-r--r--
2026-03-26 17:22
test__opcode.py
5.82
KB
-rw-r--r--
2026-03-26 17:22
test__osx_support.py
13.53
KB
-rw-r--r--
2026-03-26 17:22
test_abc.py
23.81
KB
-rw-r--r--
2026-03-26 17:22
test_abstract_numbers.py
5.81
KB
-rw-r--r--
2026-03-26 17:22
test_android.py
18.37
KB
-rw-r--r--
2026-03-26 17:22
test_apple.py
4.66
KB
-rw-r--r--
2026-03-26 17:22
test_argparse.py
233.02
KB
-rw-r--r--
2026-03-26 17:22
test_array.py
58.79
KB
-rw-r--r--
2026-03-26 17:22
test_asdl_parser.py
4.44
KB
-rw-r--r--
2026-03-26 17:22
test_asyncgen.py
58.05
KB
-rw-r--r--
2026-03-26 17:22
test_atexit.py
3.2
KB
-rw-r--r--
2026-03-26 17:22
test_audit.py
10.24
KB
-rw-r--r--
2026-03-26 17:22
test_augassign.py
7.68
KB
-rw-r--r--
2026-03-26 17:22
test_base64.py
40.3
KB
-rw-r--r--
2026-03-26 17:22
test_baseexception.py
7.92
KB
-rw-r--r--
2026-03-26 17:22
test_bdb.py
44.6
KB
-rw-r--r--
2026-03-26 17:22
test_bigaddrspace.py
2.83
KB
-rw-r--r--
2026-03-26 17:22
test_bigmem.py
45.01
KB
-rw-r--r--
2026-03-26 17:22
test_binascii.py
22.16
KB
-rw-r--r--
2026-03-26 17:22
test_binop.py
14.14
KB
-rw-r--r--
2026-03-26 17:22
test_bisect.py
16.63
KB
-rw-r--r--
2026-03-26 17:22
test_bool.py
14.25
KB
-rw-r--r--
2026-03-26 17:22
test_buffer.py
173.64
KB
-rw-r--r--
2026-03-26 17:22
test_bufio.py
2.56
KB
-rw-r--r--
2026-03-26 17:22
test_builtin.py
101.42
KB
-rw-r--r--
2026-03-26 17:22
test_bytes.py
88.54
KB
-rw-r--r--
2026-03-26 17:22
test_bz2.py
45.25
KB
-rw-r--r--
2026-03-26 17:22
test_c_locale_coercion.py
20.96
KB
-rw-r--r--
2026-03-26 17:22
test_calendar.py
57.9
KB
-rw-r--r--
2026-03-26 17:22
test_call.py
39.54
KB
-rw-r--r--
2026-03-26 17:22
test_charmapcodec.py
1.77
KB
-rw-r--r--
2026-03-26 17:22
test_class.py
28.72
KB
-rw-r--r--
2026-03-26 17:22
test_clinic.py
146.98
KB
-rw-r--r--
2026-03-26 17:22
test_cmath.py
22.47
KB
-rw-r--r--
2026-03-26 17:22
test_cmd.py
7.82
KB
-rw-r--r--
2026-03-26 17:22
test_cmd_line.py
43.25
KB
-rw-r--r--
2026-03-26 17:22
test_cmd_line_script.py
36.12
KB
-rw-r--r--
2026-03-26 17:22
test_code.py
28.35
KB
-rw-r--r--
2026-03-26 17:22
test_code_module.py
13.46
KB
-rw-r--r--
2026-03-26 17:22
test_codeccallbacks.py
49.23
KB
-rw-r--r--
2026-03-26 17:22
test_codecencodings_cn.py
3.86
KB
-rw-r--r--
2026-03-26 17:22
test_codecencodings_hk.py
701
B
-rw-r--r--
2026-03-26 17:22
test_codecencodings_iso2022.py
3.65
KB
-rw-r--r--
2026-03-26 17:22
test_codecencodings_jp.py
4.79
KB
-rw-r--r--
2026-03-26 17:22
test_codecencodings_kr.py
2.96
KB
-rw-r--r--
2026-03-26 17:22
test_codecencodings_tw.py
681
B
-rw-r--r--
2026-03-26 17:22
test_codecmaps_cn.py
746
B
-rw-r--r--
2026-03-26 17:22
test_codecmaps_hk.py
386
B
-rw-r--r--
2026-03-26 17:22
test_codecmaps_jp.py
1.7
KB
-rw-r--r--
2026-03-26 17:22
test_codecmaps_kr.py
1.16
KB
-rw-r--r--
2026-03-26 17:22
test_codecmaps_tw.py
705
B
-rw-r--r--
2026-03-26 17:22
test_codecs.py
147.37
KB
-rw-r--r--
2026-03-26 17:22
test_codeop.py
8.68
KB
-rw-r--r--
2026-03-26 17:22
test_collections.py
93.72
KB
-rw-r--r--
2026-03-26 17:22
test_colorsys.py
4.27
KB
-rw-r--r--
2026-03-26 17:22
test_compare.py
17.46
KB
-rw-r--r--
2026-03-26 17:22
test_compile.py
95.51
KB
-rw-r--r--
2026-03-26 17:22
test_compileall.py
49.53
KB
-rw-r--r--
2026-03-26 17:22
test_compiler_assemble.py
4.79
KB
-rw-r--r--
2026-03-26 17:22
test_compiler_codegen.py
5.24
KB
-rw-r--r--
2026-03-26 17:22
test_complex.py
35.29
KB
-rw-r--r--
2026-03-26 17:22
test_configparser.py
87.21
KB
-rw-r--r--
2026-03-26 17:22
test_contains.py
3.35
KB
-rw-r--r--
2026-03-26 17:22
test_context.py
32.03
KB
-rw-r--r--
2026-03-26 17:22
test_contextlib.py
43.35
KB
-rw-r--r--
2026-03-26 17:22
test_contextlib_async.py
23.45
KB
-rw-r--r--
2026-03-26 17:22
test_copy.py
29.45
KB
-rw-r--r--
2026-03-26 17:22
test_copyreg.py
4.36
KB
-rw-r--r--
2026-03-26 17:22
test_coroutines.py
67.59
KB
-rw-r--r--
2026-03-26 17:22
test_cprofile.py
9.74
KB
-rw-r--r--
2026-03-26 17:22
test_csv.py
64.09
KB
-rw-r--r--
2026-03-26 17:22
test_curses.py
50.42
KB
-rw-r--r--
2026-03-26 17:22
test_datetime.py
2.67
KB
-rw-r--r--
2026-03-26 17:22
test_dbm.py
8.26
KB
-rw-r--r--
2026-03-26 17:22
test_dbm_dumb.py
14.29
KB
-rw-r--r--
2026-03-26 17:22
test_dbm_gnu.py
8.03
KB
-rw-r--r--
2026-03-26 17:22
test_dbm_ndbm.py
6.3
KB
-rw-r--r--
2026-03-26 17:22
test_dbm_sqlite3.py
10.84
KB
-rw-r--r--
2026-03-26 17:22
test_decimal.py
216.81
KB
-rw-r--r--
2026-03-26 17:22
test_decorators.py
10.73
KB
-rw-r--r--
2026-03-26 17:22
test_defaultdict.py
6.56
KB
-rw-r--r--
2026-03-26 17:22
test_deque.py
33.16
KB
-rw-r--r--
2026-03-26 17:22
test_descr.py
197.53
KB
-rw-r--r--
2026-03-26 17:22
test_descrtut.py
11.02
KB
-rw-r--r--
2026-03-26 17:22
test_devpoll.py
4.44
KB
-rw-r--r--
2026-03-26 17:22
test_dict.py
55.39
KB
-rw-r--r--
2026-03-26 17:22
test_dict_version.py
6.1
KB
-rw-r--r--
2026-03-26 17:22
test_dictcomps.py
6.55
KB
-rw-r--r--
2026-03-26 17:22
test_dictviews.py
14.82
KB
-rw-r--r--
2026-03-26 17:22
test_difflib.py
22.99
KB
-rw-r--r--
2026-03-26 17:22
test_difflib_expect.html
100.9
KB
-rw-r--r--
2026-02-03 21:23
test_dis.py
105.91
KB
-rw-r--r--
2026-03-26 17:22
test_docxmlrpc.py
9.1
KB
-rw-r--r--
2026-03-26 17:22
test_dtrace.py
7.98
KB
-rw-r--r--
2026-03-26 17:22
test_dynamic.py
6.08
KB
-rw-r--r--
2026-03-26 17:22
test_dynamicclassattribute.py
9.57
KB
-rw-r--r--
2026-03-26 17:22
test_eintr.py
627
B
-rw-r--r--
2026-03-26 17:22
test_embed.py
74.88
KB
-rw-r--r--
2026-03-26 17:22
test_ensurepip.py
11.49
KB
-rw-r--r--
2026-03-26 17:22
test_enum.py
196.11
KB
-rw-r--r--
2026-03-26 17:22
test_enumerate.py
9.14
KB
-rw-r--r--
2026-03-26 17:22
test_eof.py
8.01
KB
-rw-r--r--
2026-03-26 17:22
test_epoll.py
9.4
KB
-rw-r--r--
2026-03-26 17:22
test_errno.py
1.04
KB
-rw-r--r--
2026-03-26 17:22
test_except_star.py
39.7
KB
-rw-r--r--
2026-03-26 17:22
test_exception_group.py
37.06
KB
-rw-r--r--
2026-03-26 17:22
test_exception_hierarchy.py
7.54
KB
-rw-r--r--
2026-03-26 17:22
test_exception_variations.py
13.74
KB
-rw-r--r--
2026-03-26 17:22
test_exceptions.py
87.23
KB
-rw-r--r--
2026-03-26 17:22
test_extcall.py
14.69
KB
-rw-r--r--
2026-03-26 17:22
test_external_inspection.py
2.96
KB
-rw-r--r--
2026-03-26 17:22
test_faulthandler.py
31.47
KB
-rw-r--r--
2026-03-26 17:22
test_fcntl.py
8.59
KB
-rw-r--r--
2026-03-26 17:22
test_file.py
11.78
KB
-rw-r--r--
2026-03-26 17:22
test_file_eintr.py
10.74
KB
-rw-r--r--
2026-03-26 17:22
test_filecmp.py
15.66
KB
-rw-r--r--
2026-03-26 17:22
test_fileinput.py
37.86
KB
-rw-r--r--
2026-03-26 17:22
test_fileio.py
20.72
KB
-rw-r--r--
2026-03-26 17:22
test_fileutils.py
951
B
-rw-r--r--
2026-03-26 17:22
test_finalization.py
14.67
KB
-rw-r--r--
2026-03-26 17:22
test_float.py
65.38
KB
-rw-r--r--
2026-03-26 17:22
test_flufl.py
2.81
KB
-rw-r--r--
2026-03-26 17:22
test_fnmatch.py
10.68
KB
-rw-r--r--
2026-03-26 17:22
test_fork1.py
3.3
KB
-rw-r--r--
2026-03-26 17:22
test_format.py
28.54
KB
-rw-r--r--
2026-03-26 17:22
test_fractions.py
67.44
KB
-rw-r--r--
2026-03-26 17:22
test_frame.py
23.28
KB
-rw-r--r--
2026-03-26 17:22
test_frozen.py
2.2
KB
-rw-r--r--
2026-03-26 17:22
test_fstring.py
70.92
KB
-rw-r--r--
2026-03-26 17:22
test_ftplib.py
42.06
KB
-rw-r--r--
2026-03-26 17:22
test_funcattrs.py
17.29
KB
-rw-r--r--
2026-03-26 17:22
test_functools.py
118.41
KB
-rw-r--r--
2026-03-26 17:22
test_gc.py
52.16
KB
-rw-r--r--
2026-03-26 17:22
test_generated_cases.py
27.14
KB
-rw-r--r--
2026-03-26 17:22
test_generator_stop.py
943
B
-rw-r--r--
2026-03-26 17:22
test_generators.py
74.12
KB
-rw-r--r--
2026-03-26 17:22
test_genericalias.py
20
KB
-rw-r--r--
2026-03-26 17:22
test_genericclass.py
9.57
KB
-rw-r--r--
2026-03-26 17:22
test_genericpath.py
23.37
KB
-rw-r--r--
2026-03-26 17:22
test_genexps.py
7.3
KB
-rw-r--r--
2026-03-26 17:22
test_getopt.py
6.94
KB
-rw-r--r--
2026-03-26 17:22
test_getpass.py
6.36
KB
-rw-r--r--
2026-03-26 17:22
test_getpath.py
44.84
KB
-rw-r--r--
2026-03-26 17:22
test_gettext.py
45.41
KB
-rw-r--r--
2026-03-26 17:22
test_glob.py
24.29
KB
-rw-r--r--
2026-03-26 17:22
test_global.py
1.2
KB
-rw-r--r--
2026-03-26 17:22
test_grammar.py
64.9
KB
-rw-r--r--
2026-03-26 17:22
test_graphlib.py
8.31
KB
-rw-r--r--
2026-03-26 17:22
test_grp.py
3.67
KB
-rw-r--r--
2026-03-26 17:22
test_gzip.py
41.97
KB
-rw-r--r--
2026-03-26 17:22
test_hash.py
12.11
KB
-rw-r--r--
2026-03-26 17:22
test_hashlib.py
50.69
KB
-rw-r--r--
2026-03-26 17:22
test_heapq.py
16.42
KB
-rw-r--r--
2026-03-26 17:22
test_hmac.py
26.49
KB
-rw-r--r--
2026-03-26 17:22
test_html.py
4.23
KB
-rw-r--r--
2026-03-26 17:22
test_htmlparser.py
51.09
KB
-rw-r--r--
2026-03-26 17:22
test_http_cookiejar.py
81.84
KB
-rw-r--r--
2026-03-26 17:22
test_http_cookies.py
23.65
KB
-rw-r--r--
2026-03-26 17:22
test_httplib.py
99.32
KB
-rw-r--r--
2026-03-26 17:22
test_httpservers.py
62.86
KB
-rw-r--r--
2026-03-26 17:22
test_idle.py
862
B
-rw-r--r--
2026-03-26 17:22
test_imaplib.py
39.24
KB
-rw-r--r--
2026-03-26 17:22
test_index.py
8.37
KB
-rw-r--r--
2026-03-26 17:22
test_int.py
34.84
KB
-rw-r--r--
2026-03-26 17:22
test_int_literal.py
6.89
KB
-rw-r--r--
2026-03-26 17:22
test_io.py
186.35
KB
-rw-r--r--
2026-03-26 17:22
test_ioctl.py
8.42
KB
-rw-r--r--
2026-03-26 17:22
test_ipaddress.py
128.86
KB
-rw-r--r--
2026-03-26 17:22
test_isinstance.py
12.96
KB
-rw-r--r--
2026-03-26 17:22
test_iter.py
38.22
KB
-rw-r--r--
2026-03-26 17:22
test_iterlen.py
7.1
KB
-rw-r--r--
2026-03-26 17:22
test_itertools.py
114.87
KB
-rw-r--r--
2026-03-26 17:22
test_keyword.py
2
KB
-rw-r--r--
2026-03-26 17:22
test_keywordonlyarg.py
6.89
KB
-rw-r--r--
2026-03-26 17:22
test_kqueue.py
9.36
KB
-rw-r--r--
2026-03-26 17:22
test_largefile.py
11.02
KB
-rw-r--r--
2026-03-26 17:22
test_launcher.py
30.61
KB
-rw-r--r--
2026-03-26 17:22
test_linecache.py
14.18
KB
-rw-r--r--
2026-03-26 17:22
test_list.py
10.94
KB
-rw-r--r--
2026-03-26 17:22
test_listcomps.py
23.53
KB
-rw-r--r--
2026-03-26 17:22
test_lltrace.py
3.71
KB
-rw-r--r--
2026-03-26 17:22
test_locale.py
26.37
KB
-rw-r--r--
2026-03-26 17:22
test_logging.py
261.1
KB
-rw-r--r--
2026-03-26 17:22
test_long.py
64.49
KB
-rw-r--r--
2026-03-26 17:22
test_longexp.py
233
B
-rw-r--r--
2026-03-26 17:22
test_lzma.py
95.71
KB
-rw-r--r--
2026-03-26 17:22
test_mailbox.py
99.06
KB
-rw-r--r--
2026-03-26 17:22
test_marshal.py
25.23
KB
-rw-r--r--
2026-03-26 17:22
test_math.py
120.29
KB
-rw-r--r--
2026-03-26 17:22
test_math_property.py
1.15
KB
-rw-r--r--
2026-03-26 17:22
test_memoryio.py
35.23
KB
-rw-r--r--
2026-03-26 17:22
test_memoryview.py
24.12
KB
-rw-r--r--
2026-03-26 17:22
test_metaclass.py
6.97
KB
-rw-r--r--
2026-03-26 17:22
test_mimetypes.py
17.91
KB
-rw-r--r--
2026-03-26 17:22
test_minidom.py
72.31
KB
-rw-r--r--
2026-03-26 17:22
test_mmap.py
45.55
KB
-rw-r--r--
2026-03-26 17:22
test_modulefinder.py
12.21
KB
-rw-r--r--
2026-03-26 17:22
test_monitoring.py
69.38
KB
-rw-r--r--
2026-03-26 17:22
test_msvcrt.py
3.39
KB
-rw-r--r--
2026-03-26 17:22
test_multibytecodec.py
15.87
KB
-rw-r--r--
2026-03-26 17:22
test_multiprocessing_main_handling.py
11.48
KB
-rw-r--r--
2026-03-26 17:22
test_named_expressions.py
29.62
KB
-rw-r--r--
2026-03-26 17:22
test_netrc.py
12.09
KB
-rw-r--r--
2026-03-26 17:22
test_ntpath.py
75.9
KB
-rw-r--r--
2026-03-26 17:22
test_numeric_tower.py
8
KB
-rw-r--r--
2026-03-26 17:22
test_opcache.py
29.7
KB
-rw-r--r--
2026-03-26 17:22
test_opcodes.py
3.62
KB
-rw-r--r--
2026-03-26 17:22
test_openpty.py
600
B
-rw-r--r--
2026-03-26 17:22
test_operator.py
27.71
KB
-rw-r--r--
2026-03-26 17:22
test_optimizer.py
2.61
KB
-rw-r--r--
2026-03-26 17:22
test_optparse.py
61.39
KB
-rw-r--r--
2026-03-26 17:22
test_ordered_dict.py
40.01
KB
-rw-r--r--
2026-03-26 17:22
test_os.py
202.8
KB
-rw-r--r--
2026-03-26 17:22
test_osx_env.py
1.31
KB
-rw-r--r--
2026-03-26 17:22
test_patma.py
94.55
KB
-rw-r--r--
2026-03-26 17:22
test_pdb.py
133.58
KB
-rw-r--r--
2026-03-26 17:22
test_peepholer.py
43.65
KB
-rw-r--r--
2026-03-26 17:22
test_pep646_syntax.py
7.79
KB
-rw-r--r--
2026-03-26 17:22
test_perf_profiler.py
18.83
KB
-rw-r--r--
2026-03-26 17:22
test_perfmaps.py
773
B
-rw-r--r--
2026-03-26 17:22
test_pickle.py
27.61
KB
-rw-r--r--
2026-03-26 17:22
test_picklebuffer.py
4.99
KB
-rw-r--r--
2026-03-26 17:22
test_pickletools.py
16.78
KB
-rw-r--r--
2026-03-26 17:22
test_pkg.py
9.58
KB
-rw-r--r--
2026-03-26 17:22
test_pkgutil.py
26.22
KB
-rw-r--r--
2026-03-26 17:22
test_platform.py
26.69
KB
-rw-r--r--
2026-03-26 17:22
test_plistlib.py
48.91
KB
-rw-r--r--
2026-03-26 17:22
test_poll.py
7.49
KB
-rw-r--r--
2026-03-26 17:22
test_popen.py
2.36
KB
-rw-r--r--
2026-03-26 17:22
test_poplib.py
18.07
KB
-rw-r--r--
2026-03-26 17:22
test_positional_only_arg.py
18.44
KB
-rw-r--r--
2026-03-26 17:22
test_posix.py
99.63
KB
-rw-r--r--
2026-03-26 17:22
test_posixpath.py
51.64
KB
-rw-r--r--
2026-03-26 17:22
test_pow.py
6.38
KB
-rw-r--r--
2026-03-26 17:22
test_pprint.py
39.08
KB
-rw-r--r--
2026-03-26 17:22
test_print.py
8.05
KB
-rw-r--r--
2026-03-26 17:22
test_profile.py
8.95
KB
-rw-r--r--
2026-03-26 17:22
test_property.py
20.12
KB
-rw-r--r--
2026-03-26 17:22
test_pstats.py
5.38
KB
-rw-r--r--
2026-03-26 17:22
test_pty.py
16.18
KB
-rw-r--r--
2026-03-26 17:22
test_pulldom.py
12.33
KB
-rw-r--r--
2026-03-26 17:22
test_pwd.py
4.32
KB
-rw-r--r--
2026-03-26 17:22
test_py_compile.py
11.93
KB
-rw-r--r--
2026-03-26 17:22
test_pyclbr.py
10.11
KB
-rw-r--r--
2026-03-26 17:22
test_pyexpat.py
39.85
KB
-rw-r--r--
2026-03-26 17:22
test_queue.py
33.55
KB
-rw-r--r--
2026-03-26 17:22
test_quopri.py
7.87
KB
-rw-r--r--
2026-03-26 17:22
test_raise.py
13.85
KB
-rw-r--r--
2026-03-26 17:22
test_random.py
60.35
KB
-rw-r--r--
2026-03-26 17:22
test_range.py
26.99
KB
-rw-r--r--
2026-03-26 17:22
test_re.py
147.27
KB
-rw-r--r--
2026-03-26 17:22
test_readline.py
16.15
KB
-rw-r--r--
2026-03-26 17:22
test_regrtest.py
98.35
KB
-rw-r--r--
2026-03-26 17:22
test_repl.py
14.52
KB
-rw-r--r--
2026-03-26 17:22
test_reprlib.py
30.43
KB
-rw-r--r--
2026-03-26 17:22
test_resource.py
11.12
KB
-rw-r--r--
2026-03-26 17:22
test_richcmp.py
11.95
KB
-rw-r--r--
2026-03-26 17:22
test_rlcompleter.py
7.96
KB
-rw-r--r--
2026-03-26 17:22
test_robotparser.py
15.6
KB
-rw-r--r--
2026-03-26 17:22
test_runpy.py
34.12
KB
-rw-r--r--
2026-03-26 17:22
test_sax.py
54.63
KB
-rw-r--r--
2026-03-26 17:22
test_sched.py
7.38
KB
-rw-r--r--
2026-03-26 17:22
test_scope.py
21.94
KB
-rw-r--r--
2026-03-26 17:22
test_script_helper.py
5.82
KB
-rw-r--r--
2026-03-26 17:22
test_secrets.py
4.28
KB
-rw-r--r--
2026-03-26 17:22
test_select.py
3.42
KB
-rw-r--r--
2026-03-26 17:22
test_selectors.py
19.87
KB
-rw-r--r--
2026-03-26 17:22
test_set.py
73.22
KB
-rw-r--r--
2026-03-26 17:22
test_setcomps.py
5.24
KB
-rw-r--r--
2026-03-26 17:22
test_shelve.py
6.42
KB
-rw-r--r--
2026-03-26 17:22
test_shlex.py
13.39
KB
-rw-r--r--
2026-03-26 17:22
test_shutil.py
140.16
KB
-rw-r--r--
2026-03-26 17:22
test_signal.py
52.78
KB
-rw-r--r--
2026-03-26 17:22
test_site.py
36.06
KB
-rw-r--r--
2026-03-26 17:22
test_slice.py
9.46
KB
-rw-r--r--
2026-03-26 17:22
test_smtplib.py
62.33
KB
-rw-r--r--
2026-03-26 17:22
test_smtpnet.py
3.11
KB
-rw-r--r--
2026-03-26 17:22
test_socket.py
271.05
KB
-rw-r--r--
2026-03-26 17:22
test_socketserver.py
17.39
KB
-rw-r--r--
2026-03-26 17:22
test_sort.py
14.48
KB
-rw-r--r--
2026-03-26 17:22
test_source_encoding.py
15.94
KB
-rw-r--r--
2026-03-26 17:22
test_ssl.py
229.41
KB
-rw-r--r--
2026-03-26 17:22
test_stable_abi_ctypes.py
26.22
KB
-rw-r--r--
2026-03-26 17:22
test_startfile.py
1.7
KB
-rw-r--r--
2026-03-26 17:22
test_stat.py
13.25
KB
-rw-r--r--
2026-03-26 17:22
test_statistics.py
131.36
KB
-rw-r--r--
2026-03-26 17:22
test_str.py
126.35
KB
-rw-r--r--
2026-03-26 17:22
test_strftime.py
7.87
KB
-rw-r--r--
2026-03-26 17:22
test_string.py
21.86
KB
-rw-r--r--
2026-03-26 17:22
test_string_literals.py
14.1
KB
-rw-r--r--
2026-03-26 17:22
test_stringprep.py
3.04
KB
-rw-r--r--
2026-03-26 17:22
test_strptime.py
42.79
KB
-rw-r--r--
2026-03-26 17:22
test_strtod.py
20.06
KB
-rw-r--r--
2026-03-26 17:22
test_struct.py
40.47
KB
-rw-r--r--
2026-03-26 17:22
test_structseq.py
14.19
KB
-rw-r--r--
2026-03-26 17:22
test_subclassinit.py
8.04
KB
-rw-r--r--
2026-03-26 17:22
test_subprocess.py
172.83
KB
-rw-r--r--
2026-03-26 17:22
test_sundry.py
1.02
KB
-rw-r--r--
2026-03-26 17:22
test_super.py
15.68
KB
-rw-r--r--
2026-03-26 17:22
test_support.py
27.95
KB
-rw-r--r--
2026-03-26 17:22
test_symtable.py
20.65
KB
-rw-r--r--
2026-03-26 17:22
test_syntax.py
80.5
KB
-rw-r--r--
2026-03-26 17:22
test_sys.py
74.12
KB
-rw-r--r--
2026-03-26 17:22
test_sys_setprofile.py
14.77
KB
-rw-r--r--
2026-03-26 17:22
test_sys_settrace.py
86.74
KB
-rw-r--r--
2026-03-26 17:22
test_sysconfig.py
29.64
KB
-rw-r--r--
2026-03-26 17:22
test_syslog.py
4.7
KB
-rw-r--r--
2026-03-26 17:22
test_tabnanny.py
13.9
KB
-rw-r--r--
2026-03-26 17:22
test_tarfile.py
187.4
KB
-rw-r--r--
2026-03-26 17:22
test_tcl.py
31.6
KB
-rw-r--r--
2026-03-26 17:22
test_tempfile.py
75.5
KB
-rw-r--r--
2026-03-26 17:22
test_termios.py
13.02
KB
-rw-r--r--
2026-03-26 17:22
test_textwrap.py
41.88
KB
-rw-r--r--
2026-03-26 17:22
test_thread.py
14.35
KB
-rw-r--r--
2026-03-26 17:22
test_threadedtempfile.py
2.03
KB
-rw-r--r--
2026-03-26 17:22
test_threading.py
79.5
KB
-rw-r--r--
2026-03-26 17:22
test_threading_local.py
7.01
KB
-rw-r--r--
2026-03-26 17:22
test_threadsignals.py
9.87
KB
-rw-r--r--
2026-03-26 17:22
test_time.py
45.07
KB
-rw-r--r--
2026-03-26 17:22
test_timeit.py
15.2
KB
-rw-r--r--
2026-03-26 17:22
test_timeout.py
10.65
KB
-rw-r--r--
2026-03-26 17:22
test_tokenize.py
121.49
KB
-rw-r--r--
2026-03-26 17:22
test_trace.py
21.72
KB
-rw-r--r--
2026-03-26 17:22
test_traceback.py
187.96
KB
-rw-r--r--
2026-03-26 17:22
test_tracemalloc.py
41.3
KB
-rw-r--r--
2026-03-26 17:22
test_ttk_textonly.py
17.45
KB
-rw-r--r--
2026-03-26 17:22
test_tty.py
3.63
KB
-rw-r--r--
2026-03-26 17:22
test_tuple.py
19.8
KB
-rw-r--r--
2026-03-26 17:22
test_turtle.py
16.42
KB
-rw-r--r--
2026-03-26 17:22
test_type_aliases.py
12.74
KB
-rw-r--r--
2026-03-26 17:22
test_type_annotations.py
6.23
KB
-rw-r--r--
2026-03-26 17:22
test_type_cache.py
7.58
KB
-rw-r--r--
2026-03-26 17:22
test_type_comments.py
11.01
KB
-rw-r--r--
2026-03-26 17:22
test_type_params.py
44.45
KB
-rw-r--r--
2026-03-26 17:22
test_typechecks.py
2.55
KB
-rw-r--r--
2026-03-26 17:22
test_types.py
87.9
KB
-rw-r--r--
2026-03-26 17:22
test_typing.py
373.85
KB
-rw-r--r--
2026-03-26 17:22
test_ucn.py
9.71
KB
-rw-r--r--
2026-03-26 17:22
test_unary.py
1.52
KB
-rw-r--r--
2026-03-26 17:22
test_unicode_file.py
5.72
KB
-rw-r--r--
2026-03-26 17:22
test_unicode_file_functions.py
6.91
KB
-rw-r--r--
2026-03-26 17:22
test_unicode_identifiers.py
997
B
-rw-r--r--
2026-03-26 17:22
test_unicodedata.py
32.39
KB
-rw-r--r--
2026-03-26 17:22
test_univnewlines.py
3.85
KB
-rw-r--r--
2026-03-26 17:22
test_unpack.py
3.51
KB
-rw-r--r--
2026-03-26 17:22
test_unpack_ex.py
9.98
KB
-rw-r--r--
2026-03-26 17:22
test_unparse.py
30
KB
-rw-r--r--
2026-03-26 17:22
test_urllib.py
72.58
KB
-rw-r--r--
2026-03-26 17:22
test_urllib2.py
81.34
KB
-rw-r--r--
2026-03-26 17:22
test_urllib2_localnet.py
24.22
KB
-rw-r--r--
2026-03-26 17:22
test_urllib2net.py
15.6
KB
-rw-r--r--
2026-03-26 17:22
test_urllib_response.py
2.04
KB
-rw-r--r--
2026-03-26 17:22
test_urllibnet.py
9.37
KB
-rw-r--r--
2026-03-26 17:22
test_urlparse.py
79.89
KB
-rw-r--r--
2026-03-26 17:22
test_userdict.py
7.82
KB
-rw-r--r--
2026-03-26 17:22
test_userlist.py
2.23
KB
-rw-r--r--
2026-03-26 17:22
test_userstring.py
2.48
KB
-rw-r--r--
2026-03-26 17:22
test_utf8_mode.py
10.31
KB
-rw-r--r--
2026-03-26 17:22
test_utf8source.py
1.08
KB
-rw-r--r--
2026-03-26 17:22
test_uuid.py
45.51
KB
-rw-r--r--
2026-03-26 17:22
test_venv.py
43.4
KB
-rw-r--r--
2026-03-26 17:22
test_wait3.py
1.74
KB
-rw-r--r--
2026-03-26 17:22
test_wait4.py
1.14
KB
-rw-r--r--
2026-03-26 17:22
test_wave.py
9.12
KB
-rw-r--r--
2026-03-26 17:22
test_weakref.py
78.11
KB
-rw-r--r--
2026-03-26 17:22
test_weakset.py
16.28
KB
-rw-r--r--
2026-03-26 17:22
test_webbrowser.py
17.31
KB
-rw-r--r--
2026-03-26 17:22
test_winapi.py
5.8
KB
-rw-r--r--
2026-03-26 17:22
test_winconsoleio.py
8.06
KB
-rw-r--r--
2026-03-26 17:22
test_winreg.py
23.71
KB
-rw-r--r--
2026-03-26 17:22
test_winsound.py
5.17
KB
-rw-r--r--
2026-03-26 17:22
test_with.py
27.77
KB
-rw-r--r--
2026-03-26 17:22
test_wmi.py
3.31
KB
-rw-r--r--
2026-03-26 17:22
test_wsgiref.py
29.89
KB
-rw-r--r--
2026-03-26 17:22
test_xml_dom_minicompat.py
4.18
KB
-rw-r--r--
2026-03-26 17:22
test_xml_dom_xmlbuilder.py
3.1
KB
-rw-r--r--
2026-03-26 17:22
test_xml_etree.py
176.35
KB
-rw-r--r--
2026-03-26 17:22
test_xml_etree_c.py
9.34
KB
-rw-r--r--
2026-03-26 17:22
test_xmlrpc.py
58.32
KB
-rw-r--r--
2026-03-26 17:22
test_xpickle.py
9.9
KB
-rw-r--r--
2026-03-26 17:22
test_xxlimited.py
2.46
KB
-rw-r--r--
2026-03-26 17:22
test_xxtestfuzz.py
690
B
-rw-r--r--
2026-03-26 17:22
test_yield_from.py
50.36
KB
-rw-r--r--
2026-03-26 17:22
test_zipapp.py
17.58
KB
-rw-r--r--
2026-03-26 17:22
test_zipfile64.py
5.78
KB
-rw-r--r--
2026-03-26 17:22
test_zipimport.py
39.21
KB
-rw-r--r--
2026-03-26 17:22
test_zipimport_support.py
10.61
KB
-rw-r--r--
2026-03-26 17:22
test_zlib.py
41.65
KB
-rw-r--r--
2026-03-26 17:22
testcodec.py
1.02
KB
-rw-r--r--
2026-03-26 17:22
tf_inherit_check.py
714
B
-rw-r--r--
2026-03-26 17:22
win_console_handler.py
1.38
KB
-rw-r--r--
2026-03-26 17:22
xmltests.py
435
B
-rw-r--r--
2026-03-26 17:22
xpickle_worker.py
2.06
KB
-rw-r--r--
2026-03-26 17:22
Save
Rename
from collections import namedtuple import contextlib import sys from textwrap import dedent import threading import time import unittest from test.support import import_helper _channels = import_helper.import_module('_interpchannels') from test.support.interpreters import _crossinterp from test.test__interpreters import ( _interpreters, _run_output, clean_up_interpreters, ) REPLACE = _crossinterp._UNBOUND_CONSTANT_TO_FLAG[_crossinterp.UNBOUND] # Additional tests are found in Lib/test/test_interpreters/test_channels.py. # New tests should be added there. # XXX The tests here should be moved there. See the note under LowLevelTests. ################################## # helpers def recv_wait(cid): while True: try: obj, unboundop = _channels.recv(cid) except _channels.ChannelEmptyError: time.sleep(0.1) else: assert unboundop is None, repr(unboundop) return obj def recv_nowait(cid, *args, unbound=False): obj, unboundop = _channels.recv(cid, *args) assert (unboundop is None) != unbound, repr(unboundop) return obj #@contextmanager #def run_threaded(id, source, **shared): # def run(): # run_interp(id, source, **shared) # t = threading.Thread(target=run) # t.start() # yield # t.join() def run_interp(id, source, **shared): _run_interp(id, source, shared) def _run_interp(id, source, shared, _mainns={}): source = dedent(source) main, *_ = _interpreters.get_main() if main == id: cur, *_ = _interpreters.get_current() if cur != main: raise RuntimeError # XXX Run a func? exec(source, _mainns) else: _interpreters.run_string(id, source, shared) class Interpreter(namedtuple('Interpreter', 'name id')): @classmethod def from_raw(cls, raw): if isinstance(raw, cls): return raw elif isinstance(raw, str): return cls(raw) else: raise NotImplementedError def __new__(cls, name=None, id=None): main, *_ = _interpreters.get_main() if id == main: if not name: name = 'main' elif name != 'main': raise ValueError( 'name mismatch (expected "main", got "{}")'.format(name)) id = main elif id is not None: if not name: name = 'interp' elif name == 'main': raise ValueError('name mismatch (unexpected "main")') assert isinstance(id, int), repr(id) elif not name or name == 'main': name = 'main' id = main else: id = _interpreters.create() self = super().__new__(cls, name, id) return self # XXX expect_channel_closed() is unnecessary once we improve exc propagation. @contextlib.contextmanager def expect_channel_closed(): try: yield except _channels.ChannelClosedError: pass else: assert False, 'channel not closed' class ChannelAction(namedtuple('ChannelAction', 'action end interp')): def __new__(cls, action, end=None, interp=None): if not end: end = 'both' if not interp: interp = 'main' self = super().__new__(cls, action, end, interp) return self def __init__(self, *args, **kwargs): if self.action == 'use': if self.end not in ('same', 'opposite', 'send', 'recv'): raise ValueError(self.end) elif self.action in ('close', 'force-close'): if self.end not in ('both', 'same', 'opposite', 'send', 'recv'): raise ValueError(self.end) else: raise ValueError(self.action) if self.interp not in ('main', 'same', 'other', 'extra'): raise ValueError(self.interp) def resolve_end(self, end): if self.end == 'same': return end elif self.end == 'opposite': return 'recv' if end == 'send' else 'send' else: return self.end def resolve_interp(self, interp, other, extra): if self.interp == 'same': return interp elif self.interp == 'other': if other is None: raise RuntimeError return other elif self.interp == 'extra': if extra is None: raise RuntimeError return extra elif self.interp == 'main': if interp.name == 'main': return interp elif other and other.name == 'main': return other else: raise RuntimeError # Per __init__(), there aren't any others. class ChannelState(namedtuple('ChannelState', 'pending closed')): def __new__(cls, pending=0, *, closed=False): self = super().__new__(cls, pending, closed) return self def incr(self): return type(self)(self.pending + 1, closed=self.closed) def decr(self): return type(self)(self.pending - 1, closed=self.closed) def close(self, *, force=True): if self.closed: if not force or self.pending == 0: return self return type(self)(0 if force else self.pending, closed=True) def run_action(cid, action, end, state, *, hideclosed=True): if state.closed: if action == 'use' and end == 'recv' and state.pending: expectfail = False else: expectfail = True else: expectfail = False try: result = _run_action(cid, action, end, state) except _channels.ChannelClosedError: if not hideclosed and not expectfail: raise result = state.close() else: if expectfail: raise ... # XXX return result def _run_action(cid, action, end, state): if action == 'use': if end == 'send': _channels.send(cid, b'spam', blocking=False) return state.incr() elif end == 'recv': if not state.pending: try: _channels.recv(cid) except _channels.ChannelEmptyError: return state else: raise Exception('expected ChannelEmptyError') else: recv_nowait(cid) return state.decr() else: raise ValueError(end) elif action == 'close': kwargs = {} if end in ('recv', 'send'): kwargs[end] = True _channels.close(cid, **kwargs) return state.close() elif action == 'force-close': kwargs = { 'force': True, } if end in ('recv', 'send'): kwargs[end] = True _channels.close(cid, **kwargs) return state.close(force=True) else: raise ValueError(action) def clean_up_channels(): for cid, _ in _channels.list_all(): try: _channels.destroy(cid) except _channels.ChannelNotFoundError: pass # already destroyed class TestBase(unittest.TestCase): def tearDown(self): clean_up_channels() clean_up_interpreters() ################################## # channel tests class ChannelIDTests(TestBase): def test_default_kwargs(self): cid = _channels._channel_id(10, force=True) self.assertEqual(int(cid), 10) self.assertEqual(cid.end, 'both') def test_with_kwargs(self): cid = _channels._channel_id(10, send=True, force=True) self.assertEqual(cid.end, 'send') cid = _channels._channel_id(10, send=True, recv=False, force=True) self.assertEqual(cid.end, 'send') cid = _channels._channel_id(10, recv=True, force=True) self.assertEqual(cid.end, 'recv') cid = _channels._channel_id(10, recv=True, send=False, force=True) self.assertEqual(cid.end, 'recv') cid = _channels._channel_id(10, send=True, recv=True, force=True) self.assertEqual(cid.end, 'both') def test_coerce_id(self): class Int(str): def __index__(self): return 10 cid = _channels._channel_id(Int(), force=True) self.assertEqual(int(cid), 10) def test_bad_id(self): self.assertRaises(TypeError, _channels._channel_id, object()) self.assertRaises(TypeError, _channels._channel_id, 10.0) self.assertRaises(TypeError, _channels._channel_id, '10') self.assertRaises(TypeError, _channels._channel_id, b'10') self.assertRaises(ValueError, _channels._channel_id, -1) self.assertRaises(OverflowError, _channels._channel_id, 2**64) def test_bad_kwargs(self): with self.assertRaises(ValueError): _channels._channel_id(10, send=False, recv=False) def test_does_not_exist(self): cid = _channels.create(REPLACE) with self.assertRaises(_channels.ChannelNotFoundError): _channels._channel_id(int(cid) + 1) # unforced def test_str(self): cid = _channels._channel_id(10, force=True) self.assertEqual(str(cid), '10') def test_repr(self): cid = _channels._channel_id(10, force=True) self.assertEqual(repr(cid), 'ChannelID(10)') cid = _channels._channel_id(10, send=True, force=True) self.assertEqual(repr(cid), 'ChannelID(10, send=True)') cid = _channels._channel_id(10, recv=True, force=True) self.assertEqual(repr(cid), 'ChannelID(10, recv=True)') cid = _channels._channel_id(10, send=True, recv=True, force=True) self.assertEqual(repr(cid), 'ChannelID(10)') def test_equality(self): cid1 = _channels.create(REPLACE) cid2 = _channels._channel_id(int(cid1)) cid3 = _channels.create(REPLACE) self.assertTrue(cid1 == cid1) self.assertTrue(cid1 == cid2) self.assertTrue(cid1 == int(cid1)) self.assertTrue(int(cid1) == cid1) self.assertTrue(cid1 == float(int(cid1))) self.assertTrue(float(int(cid1)) == cid1) self.assertFalse(cid1 == float(int(cid1)) + 0.1) self.assertFalse(cid1 == str(int(cid1))) self.assertFalse(cid1 == 2**1000) self.assertFalse(cid1 == float('inf')) self.assertFalse(cid1 == 'spam') self.assertFalse(cid1 == cid3) self.assertFalse(cid1 != cid1) self.assertFalse(cid1 != cid2) self.assertTrue(cid1 != cid3) def test_shareable(self): chan = _channels.create(REPLACE) obj = _channels.create(REPLACE) _channels.send(chan, obj, blocking=False) got = recv_nowait(chan) self.assertEqual(got, obj) self.assertIs(type(got), type(obj)) # XXX Check the following in the channel tests? #self.assertIsNot(got, obj) class ChannelTests(TestBase): def test_create_cid(self): cid = _channels.create(REPLACE) self.assertIsInstance(cid, _channels.ChannelID) def test_sequential_ids(self): before = [cid for cid, _ in _channels.list_all()] id1 = _channels.create(REPLACE) id2 = _channels.create(REPLACE) id3 = _channels.create(REPLACE) after = [cid for cid, _ in _channels.list_all()] self.assertEqual(id2, int(id1) + 1) self.assertEqual(id3, int(id2) + 1) self.assertEqual(set(after) - set(before), {id1, id2, id3}) def test_ids_global(self): id1 = _interpreters.create() out = _run_output(id1, dedent(""" import _interpchannels as _channels cid = _channels.create(3) print(cid) """)) cid1 = int(out.strip()) id2 = _interpreters.create() out = _run_output(id2, dedent(""" import _interpchannels as _channels cid = _channels.create(3) print(cid) """)) cid2 = int(out.strip()) self.assertEqual(cid2, int(cid1) + 1) def test_channel_list_interpreters_none(self): """Test listing interpreters for a channel with no associations.""" # Test for channel with no associated _interpreters. cid = _channels.create(REPLACE) send_interps = _channels.list_interpreters(cid, send=True) recv_interps = _channels.list_interpreters(cid, send=False) self.assertEqual(send_interps, []) self.assertEqual(recv_interps, []) def test_channel_list_interpreters_basic(self): """Test basic listing channel _interpreters.""" interp0, *_ = _interpreters.get_main() cid = _channels.create(REPLACE) _channels.send(cid, "send", blocking=False) # Test for a channel that has one end associated to an interpreter. send_interps = _channels.list_interpreters(cid, send=True) recv_interps = _channels.list_interpreters(cid, send=False) self.assertEqual(send_interps, [interp0]) self.assertEqual(recv_interps, []) interp1 = _interpreters.create() _run_output(interp1, dedent(f""" import _interpchannels as _channels _channels.recv({cid}) """)) # Test for channel that has both ends associated to an interpreter. send_interps = _channels.list_interpreters(cid, send=True) recv_interps = _channels.list_interpreters(cid, send=False) self.assertEqual(send_interps, [interp0]) self.assertEqual(recv_interps, [interp1]) def test_channel_list_interpreters_multiple(self): """Test listing interpreters for a channel with many associations.""" interp0, *_ = _interpreters.get_main() interp1 = _interpreters.create() interp2 = _interpreters.create() interp3 = _interpreters.create() cid = _channels.create(REPLACE) _channels.send(cid, "send", blocking=False) _run_output(interp1, dedent(f""" import _interpchannels as _channels _channels.send({cid}, "send", blocking=False) """)) _run_output(interp2, dedent(f""" import _interpchannels as _channels _channels.recv({cid}) """)) _run_output(interp3, dedent(f""" import _interpchannels as _channels _channels.recv({cid}) """)) send_interps = _channels.list_interpreters(cid, send=True) recv_interps = _channels.list_interpreters(cid, send=False) self.assertEqual(set(send_interps), {interp0, interp1}) self.assertEqual(set(recv_interps), {interp2, interp3}) def test_channel_list_interpreters_destroyed(self): """Test listing channel interpreters with a destroyed interpreter.""" interp0, *_ = _interpreters.get_main() interp1 = _interpreters.create() cid = _channels.create(REPLACE) _channels.send(cid, "send", blocking=False) _run_output(interp1, dedent(f""" import _interpchannels as _channels _channels.recv({cid}) """)) # Should be one interpreter associated with each end. send_interps = _channels.list_interpreters(cid, send=True) recv_interps = _channels.list_interpreters(cid, send=False) self.assertEqual(send_interps, [interp0]) self.assertEqual(recv_interps, [interp1]) _interpreters.destroy(interp1) # Destroyed interpreter should not be listed. send_interps = _channels.list_interpreters(cid, send=True) recv_interps = _channels.list_interpreters(cid, send=False) self.assertEqual(send_interps, [interp0]) self.assertEqual(recv_interps, []) def test_channel_list_interpreters_released(self): """Test listing channel interpreters with a released channel.""" # Set up one channel with main interpreter on the send end and two # subinterpreters on the receive end. interp0, *_ = _interpreters.get_main() interp1 = _interpreters.create() interp2 = _interpreters.create() cid = _channels.create(REPLACE) _channels.send(cid, "data", blocking=False) _run_output(interp1, dedent(f""" import _interpchannels as _channels _channels.recv({cid}) """)) _channels.send(cid, "data", blocking=False) _run_output(interp2, dedent(f""" import _interpchannels as _channels _channels.recv({cid}) """)) # Check the setup. send_interps = _channels.list_interpreters(cid, send=True) recv_interps = _channels.list_interpreters(cid, send=False) self.assertEqual(len(send_interps), 1) self.assertEqual(len(recv_interps), 2) # Release the main interpreter from the send end. _channels.release(cid, send=True) # Send end should have no associated _interpreters. send_interps = _channels.list_interpreters(cid, send=True) recv_interps = _channels.list_interpreters(cid, send=False) self.assertEqual(len(send_interps), 0) self.assertEqual(len(recv_interps), 2) # Release one of the subinterpreters from the receive end. _run_output(interp2, dedent(f""" import _interpchannels as _channels _channels.release({cid}) """)) # Receive end should have the released interpreter removed. send_interps = _channels.list_interpreters(cid, send=True) recv_interps = _channels.list_interpreters(cid, send=False) self.assertEqual(len(send_interps), 0) self.assertEqual(recv_interps, [interp1]) def test_channel_list_interpreters_closed(self): """Test listing channel interpreters with a closed channel.""" interp0, *_ = _interpreters.get_main() interp1 = _interpreters.create() cid = _channels.create(REPLACE) # Put something in the channel so that it's not empty. _channels.send(cid, "send", blocking=False) # Check initial state. send_interps = _channels.list_interpreters(cid, send=True) recv_interps = _channels.list_interpreters(cid, send=False) self.assertEqual(len(send_interps), 1) self.assertEqual(len(recv_interps), 0) # Force close the channel. _channels.close(cid, force=True) # Both ends should raise an error. with self.assertRaises(_channels.ChannelClosedError): _channels.list_interpreters(cid, send=True) with self.assertRaises(_channels.ChannelClosedError): _channels.list_interpreters(cid, send=False) def test_channel_list_interpreters_closed_send_end(self): """Test listing channel interpreters with a channel's send end closed.""" interp0, *_ = _interpreters.get_main() interp1 = _interpreters.create() cid = _channels.create(REPLACE) # Put something in the channel so that it's not empty. _channels.send(cid, "send", blocking=False) # Check initial state. send_interps = _channels.list_interpreters(cid, send=True) recv_interps = _channels.list_interpreters(cid, send=False) self.assertEqual(len(send_interps), 1) self.assertEqual(len(recv_interps), 0) # Close the send end of the channel. _channels.close(cid, send=True) # Send end should raise an error. with self.assertRaises(_channels.ChannelClosedError): _channels.list_interpreters(cid, send=True) # Receive end should not be closed (since channel is not empty). recv_interps = _channels.list_interpreters(cid, send=False) self.assertEqual(len(recv_interps), 0) # Close the receive end of the channel from a subinterpreter. _run_output(interp1, dedent(f""" import _interpchannels as _channels _channels.close({cid}, force=True) """)) return # Both ends should raise an error. with self.assertRaises(_channels.ChannelClosedError): _channels.list_interpreters(cid, send=True) with self.assertRaises(_channels.ChannelClosedError): _channels.list_interpreters(cid, send=False) def test_allowed_types(self): cid = _channels.create(REPLACE) objects = [ None, 'spam', b'spam', 42, ] for obj in objects: with self.subTest(obj): _channels.send(cid, obj, blocking=False) got = recv_nowait(cid) self.assertEqual(got, obj) self.assertIs(type(got), type(obj)) # XXX Check the following? #self.assertIsNot(got, obj) # XXX What about between interpreters? def test_run_string_arg_unresolved(self): cid = _channels.create(REPLACE) interp = _interpreters.create() _interpreters.set___main___attrs(interp, dict(cid=cid.send)) out = _run_output(interp, dedent(""" import _interpchannels as _channels print(cid.end) _channels.send(cid, b'spam', blocking=False) """)) obj = recv_nowait(cid) self.assertEqual(obj, b'spam') self.assertEqual(out.strip(), 'send') # XXX For now there is no high-level channel into which the # sent channel ID can be converted... # Note: this test caused crashes on some buildbots (bpo-33615). @unittest.skip('disabled until high-level channels exist') def test_run_string_arg_resolved(self): cid = _channels.create(REPLACE) cid = _channels._channel_id(cid, _resolve=True) interp = _interpreters.create() out = _run_output(interp, dedent(""" import _interpchannels as _channels print(chan.id.end) _channels.send(chan.id, b'spam', blocking=False) """), dict(chan=cid.send)) obj = recv_nowait(cid) self.assertEqual(obj, b'spam') self.assertEqual(out.strip(), 'send') #------------------- # send/recv def test_send_recv_main(self): cid = _channels.create(REPLACE) orig = b'spam' _channels.send(cid, orig, blocking=False) obj = recv_nowait(cid) self.assertEqual(obj, orig) self.assertIsNot(obj, orig) def test_send_recv_same_interpreter(self): id1 = _interpreters.create() out = _run_output(id1, dedent(""" import _interpchannels as _channels cid = _channels.create(REPLACE) orig = b'spam' _channels.send(cid, orig, blocking=False) obj, _ = _channels.recv(cid) assert obj is not orig assert obj == orig """)) def test_send_recv_different_interpreters(self): cid = _channels.create(REPLACE) id1 = _interpreters.create() out = _run_output(id1, dedent(f""" import _interpchannels as _channels _channels.send({cid}, b'spam', blocking=False) """)) obj = recv_nowait(cid) self.assertEqual(obj, b'spam') def test_send_recv_different_threads(self): cid = _channels.create(REPLACE) def f(): obj = recv_wait(cid) _channels.send(cid, obj) t = threading.Thread(target=f) t.start() _channels.send(cid, b'spam') obj = recv_wait(cid) t.join() self.assertEqual(obj, b'spam') def test_send_recv_different_interpreters_and_threads(self): cid = _channels.create(REPLACE) id1 = _interpreters.create() out = None def f(): nonlocal out out = _run_output(id1, dedent(f""" import time import _interpchannels as _channels while True: try: obj, _ = _channels.recv({cid}) break except _channels.ChannelEmptyError: time.sleep(0.1) assert(obj == b'spam') _channels.send({cid}, b'eggs') """)) t = threading.Thread(target=f) t.start() _channels.send(cid, b'spam') obj = recv_wait(cid) t.join() self.assertEqual(obj, b'eggs') def test_send_not_found(self): with self.assertRaises(_channels.ChannelNotFoundError): _channels.send(10, b'spam') def test_recv_not_found(self): with self.assertRaises(_channels.ChannelNotFoundError): _channels.recv(10) def test_recv_empty(self): cid = _channels.create(REPLACE) with self.assertRaises(_channels.ChannelEmptyError): _channels.recv(cid) def test_recv_default(self): default = object() cid = _channels.create(REPLACE) obj1 = recv_nowait(cid, default) _channels.send(cid, None, blocking=False) _channels.send(cid, 1, blocking=False) _channels.send(cid, b'spam', blocking=False) _channels.send(cid, b'eggs', blocking=False) obj2 = recv_nowait(cid, default) obj3 = recv_nowait(cid, default) obj4 = recv_nowait(cid) obj5 = recv_nowait(cid, default) obj6 = recv_nowait(cid, default) self.assertIs(obj1, default) self.assertIs(obj2, None) self.assertEqual(obj3, 1) self.assertEqual(obj4, b'spam') self.assertEqual(obj5, b'eggs') self.assertIs(obj6, default) def test_recv_sending_interp_destroyed(self): with self.subTest('closed'): cid1 = _channels.create(REPLACE) interp = _interpreters.create() _interpreters.run_string(interp, dedent(f""" import _interpchannels as _channels _channels.send({cid1}, b'spam', blocking=False) """)) _interpreters.destroy(interp) with self.assertRaisesRegex(RuntimeError, f'channel {cid1} is closed'): _channels.recv(cid1) del cid1 with self.subTest('still open'): cid2 = _channels.create(REPLACE) interp = _interpreters.create() _interpreters.run_string(interp, dedent(f""" import _interpchannels as _channels _channels.send({cid2}, b'spam', blocking=False) """)) _channels.send(cid2, b'eggs', blocking=False) _interpreters.destroy(interp) recv_nowait(cid2, unbound=True) recv_nowait(cid2, unbound=False) with self.assertRaisesRegex(RuntimeError, f'channel {cid2} is empty'): _channels.recv(cid2) del cid2 #------------------- # send_buffer def test_send_buffer(self): buf = bytearray(b'spamspamspam') cid = _channels.create(REPLACE) _channels.send_buffer(cid, buf, blocking=False) obj = recv_nowait(cid) self.assertIsNot(obj, buf) self.assertIsInstance(obj, memoryview) self.assertEqual(obj, buf) buf[4:8] = b'eggs' self.assertEqual(obj, buf) obj[4:8] = b'ham.' self.assertEqual(obj, buf) #------------------- # send with waiting def build_send_waiter(self, obj, *, buffer=False): # We want a long enough sleep that send() actually has to wait. if buffer: send = _channels.send_buffer else: send = _channels.send cid = _channels.create(REPLACE) try: started = time.monotonic() send(cid, obj, blocking=False) stopped = time.monotonic() recv_nowait(cid) finally: _channels.destroy(cid) delay = stopped - started # seconds delay *= 3 def wait(): time.sleep(delay) return wait def test_send_blocking_waiting(self): received = None obj = b'spam' wait = self.build_send_waiter(obj) cid = _channels.create(REPLACE) def f(): nonlocal received wait() received = recv_wait(cid) t = threading.Thread(target=f) t.start() _channels.send(cid, obj, blocking=True) t.join() self.assertEqual(received, obj) def test_send_buffer_blocking_waiting(self): received = None obj = bytearray(b'spam') wait = self.build_send_waiter(obj, buffer=True) cid = _channels.create(REPLACE) def f(): nonlocal received wait() received = recv_wait(cid) t = threading.Thread(target=f) t.start() _channels.send_buffer(cid, obj, blocking=True) t.join() self.assertEqual(received, obj) def test_send_blocking_no_wait(self): received = None obj = b'spam' cid = _channels.create(REPLACE) def f(): nonlocal received received = recv_wait(cid) t = threading.Thread(target=f) t.start() _channels.send(cid, obj, blocking=True) t.join() self.assertEqual(received, obj) def test_send_buffer_blocking_no_wait(self): received = None obj = bytearray(b'spam') cid = _channels.create(REPLACE) def f(): nonlocal received received = recv_wait(cid) t = threading.Thread(target=f) t.start() _channels.send_buffer(cid, obj, blocking=True) t.join() self.assertEqual(received, obj) def test_send_timeout(self): obj = b'spam' with self.subTest('non-blocking with timeout'): cid = _channels.create(REPLACE) with self.assertRaises(ValueError): _channels.send(cid, obj, blocking=False, timeout=0.1) with self.subTest('timeout hit'): cid = _channels.create(REPLACE) with self.assertRaises(TimeoutError): _channels.send(cid, obj, blocking=True, timeout=0.1) with self.assertRaises(_channels.ChannelEmptyError): received = recv_nowait(cid) print(repr(received)) with self.subTest('timeout not hit'): cid = _channels.create(REPLACE) def f(): recv_wait(cid) t = threading.Thread(target=f) t.start() _channels.send(cid, obj, blocking=True, timeout=10) t.join() def test_send_buffer_timeout(self): try: self._has_run_once_timeout except AttributeError: # At the moment, this test leaks a few references. # It looks like the leak originates with the addition # of _channels.send_buffer() (gh-110246), whereas the # tests were added afterward. We want this test even # if the refleak isn't fixed yet, so we skip here. raise unittest.SkipTest('temporarily skipped due to refleaks') else: self._has_run_once_timeout = True obj = bytearray(b'spam') with self.subTest('non-blocking with timeout'): cid = _channels.create(REPLACE) with self.assertRaises(ValueError): _channels.send_buffer(cid, obj, blocking=False, timeout=0.1) with self.subTest('timeout hit'): cid = _channels.create(REPLACE) with self.assertRaises(TimeoutError): _channels.send_buffer(cid, obj, blocking=True, timeout=0.1) with self.assertRaises(_channels.ChannelEmptyError): received = recv_nowait(cid) print(repr(received)) with self.subTest('timeout not hit'): cid = _channels.create(REPLACE) def f(): recv_wait(cid) t = threading.Thread(target=f) t.start() _channels.send_buffer(cid, obj, blocking=True, timeout=10) t.join() def test_send_closed_while_waiting(self): obj = b'spam' wait = self.build_send_waiter(obj) with self.subTest('without timeout'): cid = _channels.create(REPLACE) def f(): wait() _channels.close(cid, force=True) t = threading.Thread(target=f) t.start() with self.assertRaises(_channels.ChannelClosedError): _channels.send(cid, obj, blocking=True) t.join() with self.subTest('with timeout'): cid = _channels.create(REPLACE) def f(): wait() _channels.close(cid, force=True) t = threading.Thread(target=f) t.start() with self.assertRaises(_channels.ChannelClosedError): _channels.send(cid, obj, blocking=True, timeout=30) t.join() def test_send_buffer_closed_while_waiting(self): try: self._has_run_once_closed except AttributeError: # At the moment, this test leaks a few references. # It looks like the leak originates with the addition # of _channels.send_buffer() (gh-110246), whereas the # tests were added afterward. We want this test even # if the refleak isn't fixed yet, so we skip here. raise unittest.SkipTest('temporarily skipped due to refleaks') else: self._has_run_once_closed = True obj = bytearray(b'spam') wait = self.build_send_waiter(obj, buffer=True) with self.subTest('without timeout'): cid = _channels.create(REPLACE) def f(): wait() _channels.close(cid, force=True) t = threading.Thread(target=f) t.start() with self.assertRaises(_channels.ChannelClosedError): _channels.send_buffer(cid, obj, blocking=True) t.join() with self.subTest('with timeout'): cid = _channels.create(REPLACE) def f(): wait() _channels.close(cid, force=True) t = threading.Thread(target=f) t.start() with self.assertRaises(_channels.ChannelClosedError): _channels.send_buffer(cid, obj, blocking=True, timeout=30) t.join() #------------------- # close def test_close_single_user(self): cid = _channels.create(REPLACE) _channels.send(cid, b'spam', blocking=False) recv_nowait(cid) _channels.close(cid) with self.assertRaises(_channels.ChannelClosedError): _channels.send(cid, b'eggs') with self.assertRaises(_channels.ChannelClosedError): _channels.recv(cid) def test_close_multiple_users(self): cid = _channels.create(REPLACE) id1 = _interpreters.create() id2 = _interpreters.create() _interpreters.run_string(id1, dedent(f""" import _interpchannels as _channels _channels.send({cid}, b'spam', blocking=False) """)) _interpreters.run_string(id2, dedent(f""" import _interpchannels as _channels _channels.recv({cid}) """)) _channels.close(cid) excsnap = _interpreters.run_string(id1, dedent(f""" _channels.send({cid}, b'spam') """)) self.assertEqual(excsnap.type.__name__, 'ChannelClosedError') excsnap = _interpreters.run_string(id2, dedent(f""" _channels.send({cid}, b'spam') """)) self.assertEqual(excsnap.type.__name__, 'ChannelClosedError') def test_close_multiple_times(self): cid = _channels.create(REPLACE) _channels.send(cid, b'spam', blocking=False) recv_nowait(cid) _channels.close(cid) with self.assertRaises(_channels.ChannelClosedError): _channels.close(cid) def test_close_empty(self): tests = [ (False, False), (True, False), (False, True), (True, True), ] for send, recv in tests: with self.subTest((send, recv)): cid = _channels.create(REPLACE) _channels.send(cid, b'spam', blocking=False) recv_nowait(cid) _channels.close(cid, send=send, recv=recv) with self.assertRaises(_channels.ChannelClosedError): _channels.send(cid, b'eggs') with self.assertRaises(_channels.ChannelClosedError): _channels.recv(cid) def test_close_defaults_with_unused_items(self): cid = _channels.create(REPLACE) _channels.send(cid, b'spam', blocking=False) _channels.send(cid, b'ham', blocking=False) with self.assertRaises(_channels.ChannelNotEmptyError): _channels.close(cid) recv_nowait(cid) _channels.send(cid, b'eggs', blocking=False) def test_close_recv_with_unused_items_unforced(self): cid = _channels.create(REPLACE) _channels.send(cid, b'spam', blocking=False) _channels.send(cid, b'ham', blocking=False) with self.assertRaises(_channels.ChannelNotEmptyError): _channels.close(cid, recv=True) recv_nowait(cid) _channels.send(cid, b'eggs', blocking=False) recv_nowait(cid) recv_nowait(cid) _channels.close(cid, recv=True) def test_close_send_with_unused_items_unforced(self): cid = _channels.create(REPLACE) _channels.send(cid, b'spam', blocking=False) _channels.send(cid, b'ham', blocking=False) _channels.close(cid, send=True) with self.assertRaises(_channels.ChannelClosedError): _channels.send(cid, b'eggs') recv_nowait(cid) recv_nowait(cid) with self.assertRaises(_channels.ChannelClosedError): _channels.recv(cid) def test_close_both_with_unused_items_unforced(self): cid = _channels.create(REPLACE) _channels.send(cid, b'spam', blocking=False) _channels.send(cid, b'ham', blocking=False) with self.assertRaises(_channels.ChannelNotEmptyError): _channels.close(cid, recv=True, send=True) recv_nowait(cid) _channels.send(cid, b'eggs', blocking=False) recv_nowait(cid) recv_nowait(cid) _channels.close(cid, recv=True) def test_close_recv_with_unused_items_forced(self): cid = _channels.create(REPLACE) _channels.send(cid, b'spam', blocking=False) _channels.send(cid, b'ham', blocking=False) _channels.close(cid, recv=True, force=True) with self.assertRaises(_channels.ChannelClosedError): _channels.send(cid, b'eggs') with self.assertRaises(_channels.ChannelClosedError): _channels.recv(cid) def test_close_send_with_unused_items_forced(self): cid = _channels.create(REPLACE) _channels.send(cid, b'spam', blocking=False) _channels.send(cid, b'ham', blocking=False) _channels.close(cid, send=True, force=True) with self.assertRaises(_channels.ChannelClosedError): _channels.send(cid, b'eggs') with self.assertRaises(_channels.ChannelClosedError): _channels.recv(cid) def test_close_both_with_unused_items_forced(self): cid = _channels.create(REPLACE) _channels.send(cid, b'spam', blocking=False) _channels.send(cid, b'ham', blocking=False) _channels.close(cid, send=True, recv=True, force=True) with self.assertRaises(_channels.ChannelClosedError): _channels.send(cid, b'eggs') with self.assertRaises(_channels.ChannelClosedError): _channels.recv(cid) def test_close_never_used(self): cid = _channels.create(REPLACE) _channels.close(cid) with self.assertRaises(_channels.ChannelClosedError): _channels.send(cid, b'spam') with self.assertRaises(_channels.ChannelClosedError): _channels.recv(cid) def test_close_by_unassociated_interp(self): cid = _channels.create(REPLACE) _channels.send(cid, b'spam', blocking=False) interp = _interpreters.create() _interpreters.run_string(interp, dedent(f""" import _interpchannels as _channels _channels.close({cid}, force=True) """)) with self.assertRaises(_channels.ChannelClosedError): _channels.recv(cid) with self.assertRaises(_channels.ChannelClosedError): _channels.close(cid) def test_close_used_multiple_times_by_single_user(self): cid = _channels.create(REPLACE) _channels.send(cid, b'spam', blocking=False) _channels.send(cid, b'spam', blocking=False) _channels.send(cid, b'spam', blocking=False) recv_nowait(cid) _channels.close(cid, force=True) with self.assertRaises(_channels.ChannelClosedError): _channels.send(cid, b'eggs') with self.assertRaises(_channels.ChannelClosedError): _channels.recv(cid) def test_channel_list_interpreters_invalid_channel(self): cid = _channels.create(REPLACE) # Test for invalid channel ID. with self.assertRaises(_channels.ChannelNotFoundError): _channels.list_interpreters(1000, send=True) _channels.close(cid) # Test for a channel that has been closed. with self.assertRaises(_channels.ChannelClosedError): _channels.list_interpreters(cid, send=True) def test_channel_list_interpreters_invalid_args(self): # Tests for invalid arguments passed to the API. cid = _channels.create(REPLACE) with self.assertRaises(TypeError): _channels.list_interpreters(cid) class ChannelReleaseTests(TestBase): # XXX Add more test coverage a la the tests for close(). """ - main / interp / other - run in: current thread / new thread / other thread / different threads - end / opposite - force / no force - used / not used (associated / not associated) - empty / emptied / never emptied / partly emptied - closed / not closed - released / not released - creator (interp) / other - associated interpreter not running - associated interpreter destroyed """ """ use pre-release release after check """ """ release in: main, interp1 creator: same, other (incl. interp2) use: None,send,recv,send/recv in None,same,other(incl. interp2),same+other(incl. interp2),all pre-release: None,send,recv,both in None,same,other(incl. interp2),same+other(incl. interp2),all pre-release forced: None,send,recv,both in None,same,other(incl. interp2),same+other(incl. interp2),all release: same release forced: same use after: None,send,recv,send/recv in None,same,other(incl. interp2),same+other(incl. interp2),all release after: None,send,recv,send/recv in None,same,other(incl. interp2),same+other(incl. interp2),all check released: send/recv for same/other(incl. interp2) check closed: send/recv for same/other(incl. interp2) """ def test_single_user(self): cid = _channels.create(REPLACE) _channels.send(cid, b'spam', blocking=False) recv_nowait(cid) _channels.release(cid, send=True, recv=True) with self.assertRaises(_channels.ChannelClosedError): _channels.send(cid, b'eggs') with self.assertRaises(_channels.ChannelClosedError): _channels.recv(cid) def test_multiple_users(self): cid = _channels.create(REPLACE) id1 = _interpreters.create() id2 = _interpreters.create() _interpreters.run_string(id1, dedent(f""" import _interpchannels as _channels _channels.send({cid}, b'spam', blocking=False) """)) out = _run_output(id2, dedent(f""" import _interpchannels as _channels obj, _ = _channels.recv({cid}) _channels.release({cid}) print(repr(obj)) """)) _interpreters.run_string(id1, dedent(f""" _channels.release({cid}) """)) self.assertEqual(out.strip(), "b'spam'") def test_no_kwargs(self): cid = _channels.create(REPLACE) _channels.send(cid, b'spam', blocking=False) recv_nowait(cid) _channels.release(cid) with self.assertRaises(_channels.ChannelClosedError): _channels.send(cid, b'eggs') with self.assertRaises(_channels.ChannelClosedError): _channels.recv(cid) def test_multiple_times(self): cid = _channels.create(REPLACE) _channels.send(cid, b'spam', blocking=False) recv_nowait(cid) _channels.release(cid, send=True, recv=True) with self.assertRaises(_channels.ChannelClosedError): _channels.release(cid, send=True, recv=True) def test_with_unused_items(self): cid = _channels.create(REPLACE) _channels.send(cid, b'spam', blocking=False) _channels.send(cid, b'ham', blocking=False) _channels.release(cid, send=True, recv=True) with self.assertRaises(_channels.ChannelClosedError): _channels.recv(cid) def test_never_used(self): cid = _channels.create(REPLACE) _channels.release(cid) with self.assertRaises(_channels.ChannelClosedError): _channels.send(cid, b'spam') with self.assertRaises(_channels.ChannelClosedError): _channels.recv(cid) def test_by_unassociated_interp(self): cid = _channels.create(REPLACE) _channels.send(cid, b'spam', blocking=False) interp = _interpreters.create() _interpreters.run_string(interp, dedent(f""" import _interpchannels as _channels _channels.release({cid}) """)) obj = recv_nowait(cid) _channels.release(cid) with self.assertRaises(_channels.ChannelClosedError): _channels.send(cid, b'eggs') self.assertEqual(obj, b'spam') def test_close_if_unassociated(self): # XXX Something's not right with this test... cid = _channels.create(REPLACE) interp = _interpreters.create() _interpreters.run_string(interp, dedent(f""" import _interpchannels as _channels obj = _channels.send({cid}, b'spam', blocking=False) _channels.release({cid}) """)) with self.assertRaises(_channels.ChannelClosedError): _channels.recv(cid) def test_partially(self): # XXX Is partial close too weird/confusing? cid = _channels.create(REPLACE) _channels.send(cid, None, blocking=False) recv_nowait(cid) _channels.send(cid, b'spam', blocking=False) _channels.release(cid, send=True) obj = recv_nowait(cid) self.assertEqual(obj, b'spam') def test_used_multiple_times_by_single_user(self): cid = _channels.create(REPLACE) _channels.send(cid, b'spam', blocking=False) _channels.send(cid, b'spam', blocking=False) _channels.send(cid, b'spam', blocking=False) recv_nowait(cid) _channels.release(cid, send=True, recv=True) with self.assertRaises(_channels.ChannelClosedError): _channels.send(cid, b'eggs') with self.assertRaises(_channels.ChannelClosedError): _channels.recv(cid) class ChannelCloseFixture(namedtuple('ChannelCloseFixture', 'end interp other extra creator')): # Set this to True to avoid creating interpreters, e.g. when # scanning through test permutations without running them. QUICK = False def __new__(cls, end, interp, other, extra, creator): assert end in ('send', 'recv') if cls.QUICK: known = {} else: interp = Interpreter.from_raw(interp) other = Interpreter.from_raw(other) extra = Interpreter.from_raw(extra) known = { interp.name: interp, other.name: other, extra.name: extra, } if not creator: creator = 'same' self = super().__new__(cls, end, interp, other, extra, creator) self._prepped = set() self._state = ChannelState() self._known = known return self @property def state(self): return self._state @property def cid(self): try: return self._cid except AttributeError: creator = self._get_interpreter(self.creator) self._cid = self._new_channel(creator) return self._cid def get_interpreter(self, interp): interp = self._get_interpreter(interp) self._prep_interpreter(interp) return interp def expect_closed_error(self, end=None): if end is None: end = self.end if end == 'recv' and self.state.closed == 'send': return False return bool(self.state.closed) def prep_interpreter(self, interp): self._prep_interpreter(interp) def record_action(self, action, result): self._state = result def clean_up(self): clean_up_interpreters() clean_up_channels() # internal methods def _new_channel(self, creator): if creator.name == 'main': return _channels.create(REPLACE) else: ch = _channels.create(REPLACE) run_interp(creator.id, f""" import _interpreters cid = _xxsubchannels.create() # We purposefully send back an int to avoid tying the # channel to the other interpreter. _xxsubchannels.send({ch}, int(cid), blocking=False) del _interpreters """) self._cid = recv_nowait(ch) return self._cid def _get_interpreter(self, interp): if interp in ('same', 'interp'): return self.interp elif interp == 'other': return self.other elif interp == 'extra': return self.extra else: name = interp try: interp = self._known[name] except KeyError: interp = self._known[name] = Interpreter(name) return interp def _prep_interpreter(self, interp): if interp.id in self._prepped: return self._prepped.add(interp.id) if interp.name == 'main': return run_interp(interp.id, f""" import _interpchannels as channels import test.test__interpchannels as helpers ChannelState = helpers.ChannelState try: cid except NameError: cid = _channels._channel_id({self.cid}) """) @unittest.skip('these tests take several hours to run') class ExhaustiveChannelTests(TestBase): """ - main / interp / other - run in: current thread / new thread / other thread / different threads - end / opposite - force / no force - used / not used (associated / not associated) - empty / emptied / never emptied / partly emptied - closed / not closed - released / not released - creator (interp) / other - associated interpreter not running - associated interpreter destroyed - close after unbound """ """ use pre-close close after check """ """ close in: main, interp1 creator: same, other, extra use: None,send,recv,send/recv in None,same,other,same+other,all pre-close: None,send,recv in None,same,other,same+other,all pre-close forced: None,send,recv in None,same,other,same+other,all close: same close forced: same use after: None,send,recv,send/recv in None,same,other,extra,same+other,all close after: None,send,recv,send/recv in None,same,other,extra,same+other,all check closed: send/recv for same/other(incl. interp2) """ def iter_action_sets(self): # - used / not used (associated / not associated) # - empty / emptied / never emptied / partly emptied # - closed / not closed # - released / not released # never used yield [] # only pre-closed (and possible used after) for closeactions in self._iter_close_action_sets('same', 'other'): yield closeactions for postactions in self._iter_post_close_action_sets(): yield closeactions + postactions for closeactions in self._iter_close_action_sets('other', 'extra'): yield closeactions for postactions in self._iter_post_close_action_sets(): yield closeactions + postactions # used for useactions in self._iter_use_action_sets('same', 'other'): yield useactions for closeactions in self._iter_close_action_sets('same', 'other'): actions = useactions + closeactions yield actions for postactions in self._iter_post_close_action_sets(): yield actions + postactions for closeactions in self._iter_close_action_sets('other', 'extra'): actions = useactions + closeactions yield actions for postactions in self._iter_post_close_action_sets(): yield actions + postactions for useactions in self._iter_use_action_sets('other', 'extra'): yield useactions for closeactions in self._iter_close_action_sets('same', 'other'): actions = useactions + closeactions yield actions for postactions in self._iter_post_close_action_sets(): yield actions + postactions for closeactions in self._iter_close_action_sets('other', 'extra'): actions = useactions + closeactions yield actions for postactions in self._iter_post_close_action_sets(): yield actions + postactions def _iter_use_action_sets(self, interp1, interp2): interps = (interp1, interp2) # only recv end used yield [ ChannelAction('use', 'recv', interp1), ] yield [ ChannelAction('use', 'recv', interp2), ] yield [ ChannelAction('use', 'recv', interp1), ChannelAction('use', 'recv', interp2), ] # never emptied yield [ ChannelAction('use', 'send', interp1), ] yield [ ChannelAction('use', 'send', interp2), ] yield [ ChannelAction('use', 'send', interp1), ChannelAction('use', 'send', interp2), ] # partially emptied for interp1 in interps: for interp2 in interps: for interp3 in interps: yield [ ChannelAction('use', 'send', interp1), ChannelAction('use', 'send', interp2), ChannelAction('use', 'recv', interp3), ] # fully emptied for interp1 in interps: for interp2 in interps: for interp3 in interps: for interp4 in interps: yield [ ChannelAction('use', 'send', interp1), ChannelAction('use', 'send', interp2), ChannelAction('use', 'recv', interp3), ChannelAction('use', 'recv', interp4), ] def _iter_close_action_sets(self, interp1, interp2): ends = ('recv', 'send') interps = (interp1, interp2) for force in (True, False): op = 'force-close' if force else 'close' for interp in interps: for end in ends: yield [ ChannelAction(op, end, interp), ] for recvop in ('close', 'force-close'): for sendop in ('close', 'force-close'): for recv in interps: for send in interps: yield [ ChannelAction(recvop, 'recv', recv), ChannelAction(sendop, 'send', send), ] def _iter_post_close_action_sets(self): for interp in ('same', 'extra', 'other'): yield [ ChannelAction('use', 'recv', interp), ] yield [ ChannelAction('use', 'send', interp), ] def run_actions(self, fix, actions): for action in actions: self.run_action(fix, action) def run_action(self, fix, action, *, hideclosed=True): end = action.resolve_end(fix.end) interp = action.resolve_interp(fix.interp, fix.other, fix.extra) fix.prep_interpreter(interp) if interp.name == 'main': result = run_action( fix.cid, action.action, end, fix.state, hideclosed=hideclosed, ) fix.record_action(action, result) else: _cid = _channels.create(REPLACE) run_interp(interp.id, f""" result = helpers.run_action( {fix.cid}, {repr(action.action)}, {repr(end)}, {repr(fix.state)}, hideclosed={hideclosed}, ) _channels.send({_cid}, result.pending.to_bytes(1, 'little'), blocking=False) _channels.send({_cid}, b'X' if result.closed else b'', blocking=False) """) result = ChannelState( pending=int.from_bytes(recv_nowait(_cid), 'little'), closed=bool(recv_nowait(_cid)), ) fix.record_action(action, result) def iter_fixtures(self): # XXX threads? interpreters = [ ('main', 'interp', 'extra'), ('interp', 'main', 'extra'), ('interp1', 'interp2', 'extra'), ('interp1', 'interp2', 'main'), ] for interp, other, extra in interpreters: for creator in ('same', 'other', 'creator'): for end in ('send', 'recv'): yield ChannelCloseFixture(end, interp, other, extra, creator) def _close(self, fix, *, force): op = 'force-close' if force else 'close' close = ChannelAction(op, fix.end, 'same') if not fix.expect_closed_error(): self.run_action(fix, close, hideclosed=False) else: with self.assertRaises(_channels.ChannelClosedError): self.run_action(fix, close, hideclosed=False) def _assert_closed_in_interp(self, fix, interp=None): if interp is None or interp.name == 'main': with self.assertRaises(_channels.ChannelClosedError): _channels.recv(fix.cid) with self.assertRaises(_channels.ChannelClosedError): _channels.send(fix.cid, b'spam') with self.assertRaises(_channels.ChannelClosedError): _channels.close(fix.cid) with self.assertRaises(_channels.ChannelClosedError): _channels.close(fix.cid, force=True) else: run_interp(interp.id, """ with helpers.expect_channel_closed(): _channels.recv(cid) """) run_interp(interp.id, """ with helpers.expect_channel_closed(): _channels.send(cid, b'spam', blocking=False) """) run_interp(interp.id, """ with helpers.expect_channel_closed(): _channels.close(cid) """) run_interp(interp.id, """ with helpers.expect_channel_closed(): _channels.close(cid, force=True) """) def _assert_closed(self, fix): self.assertTrue(fix.state.closed) for _ in range(fix.state.pending): recv_nowait(fix.cid) self._assert_closed_in_interp(fix) for interp in ('same', 'other'): interp = fix.get_interpreter(interp) if interp.name == 'main': continue self._assert_closed_in_interp(fix, interp) interp = fix.get_interpreter('fresh') self._assert_closed_in_interp(fix, interp) def _iter_close_tests(self, verbose=False): i = 0 for actions in self.iter_action_sets(): print() for fix in self.iter_fixtures(): i += 1 if i > 1000: return if verbose: if (i - 1) % 6 == 0: print() print(i, fix, '({} actions)'.format(len(actions))) else: if (i - 1) % 6 == 0: print(' ', end='') print('.', end=''); sys.stdout.flush() yield i, fix, actions if verbose: print('---') print() # This is useful for scanning through the possible tests. def _skim_close_tests(self): ChannelCloseFixture.QUICK = True for i, fix, actions in self._iter_close_tests(): pass def test_close(self): for i, fix, actions in self._iter_close_tests(): with self.subTest('{} {} {}'.format(i, fix, actions)): fix.prep_interpreter(fix.interp) self.run_actions(fix, actions) self._close(fix, force=False) self._assert_closed(fix) # XXX Things slow down if we have too many interpreters. fix.clean_up() def test_force_close(self): for i, fix, actions in self._iter_close_tests(): with self.subTest('{} {} {}'.format(i, fix, actions)): fix.prep_interpreter(fix.interp) self.run_actions(fix, actions) self._close(fix, force=True) self._assert_closed(fix) # XXX Things slow down if we have too many interpreters. fix.clean_up() if __name__ == '__main__': unittest.main()