mirrors/RustPython
0
0 Fork 0
mirror of https://github.com/RustPython/RustPython.git synced 2026-10-28 18:59:04 +08:00
Code Activity
Files
1ffba21557cfca4fdb07b40516c489b0ee91cd0c
RustPython/scripts/update_lib/deps.py
T
youknowdot 7d15b50cd4 Align weak-reference collection with Python 3.15 finalization (#9168) 
* Align weak-reference collection with Python 3.15 finalization

Trace and retain owned callback edges independently of weak referent links.
Clear callback-bearing weakrefs before finalizers, suppress callbacks from
unreachable weakrefs, and close remaining weakref ingress after resurrection
checks while the world is stopped. Publish and unlink intrusive nodes safely.

Refresh weakref and canonical weakref/GC/finalization tests through update_lib
from pinned CPython 3.15.0. Retire four obsolete guards only after fresh body
passes, retaining applicable RustPython limitations.

Assisted-by: OpenAI Codex:model version unavailable

* Refresh canonical IO tests for weakref finalization

Migrate the complete CPython 3.15.0 test_io package at d220409 with the
fail-closed legacy-map updater, preserving all ten existing guards. Retire
the seven consolidated standalone suites and update io/file dependency
paths with mapping regression tests.

The canonical GC tests ignore ResourceWarning without retaining its
source, so callback-free weakrefs correctly reflect surviving owners.
This closes the stale test dependency exposed by the lifecycle change.

Assisted-by: OpenAI Codex:model version unavailable

* Protect weakref callbacks with the original referent stripe

Follow CPython's free-threading weakref locking model instead of keeping
an independent callback RwLock. Retain the original referent stripe key
so callback access stays synchronized after GC clears the referent.
Unlink and steal callbacks under that stripe, then release ownership
outside the lock. Preserve synchronization for running-world GC paths
and gc.get_referents while keeping ordinary weakref upgrades unchanged.

Assisted-by: OpenAI Codex:model version unavailable

* Defer weakref GC tracking until referent publication

Keep freshly allocated weakrefs out of GC's object lists until they are
linked and their referent is published. Release the referent stripe before
tracking, since tracking can collect and invoke finalizers. Generic reuse
losers remain untracked.

Exercise the allocator boundary directly for reference, proxy, callable
proxy and subclass payloads, and ensure completed constructors track their
objects. This prevents observers from retaining a half-built dead weakref
that later becomes live.

Assisted-by: OpenAI Codex:model version unavailable

* Retain the original weakref stripe lock like CPython

Store a safe handle to the selected static weakref stripe instead of
retaining and rehashing the referent address. Keep the same lock after
referent clearing, matching CPython free-threaded weakrefs_lock. The
nonthreaded handle is zero-sized and introduces no locking.

Preserve lock selection, acquire/release ordering, fork reset, deferred
GC publication, and callback-owner release outside the stripe. No new
locks or unsafe code are needed.

Assisted-by: OpenAI Codex:model version unavailable
2026-10-17 23:53:07 +09:00

1728 lines
50 KiB
Python
Raw Blame History

"""
Dependency resolution for library updates.
Handles:
- Irregular library paths (e.g., libregrtest at Lib/test/libregrtest/)
- Library dependencies (e.g., datetime requires _pydatetime)
- Test dependencies (auto-detected from 'from test import ...')
"""
import ast
import difflib
import functools
import pathlib
import shelve
import subprocess
from update_lib.file_utils import (
_dircmp_is_same,
compare_dir_contents,
compare_file_contents,
compare_paths,
construct_lib_path,
cpython_to_local_path,
read_python_files,
resolve_module_path,
resolve_test_path,
safe_parse_ast,
safe_read_text,
)
# === Import parsing utilities ===
class ImportVisitor(ast.NodeVisitor):
def __init__(self) -> None:
self.__imports = set()
@property
def test_imports(self) -> set[str]:
imports = set()
for module in self.__imports:
if not module.startswith("test."):
continue
name = module.removeprefix("test.")
if name == "support" or name.startswith("support."):
continue
imports.add(name)
return imports
@property
def lib_imports(self) -> set[str]:
return {module for module in self.__imports if not module.startswith("test.")}
def visit_Import(self, node):
for alias in node.names:
self.__imports.add(alias.name)
def visit_ImportFrom(self, node):
try:
module = node.module
except AttributeError:
# Ignore `from . import my_internal_module`
return
if module is None: # Ignore `from . import my_internal_module`
return
for alias in node.names:
# We only care about what we import if it was from the "test" module
if module == "test":
name = f"{module}.{alias.name}"
else:
name = module
self.__imports.add(name)
def visit_Call(self, node) -> None:
"""
In test files, there's sometimes use of:
```python
import test.support
from test.support import script_helper
script = support.findfile("_test_atexit.py")
script_helper.run_test_script(script)
```
This imports "_test_atexit.py" but does not show as an import node.
"""
func = node.func
if not isinstance(func, ast.Attribute):
return
value = func.value
if not isinstance(value, ast.Name):
return
if (value.id != "support") or (func.attr != "findfile"):
return
arg = node.args[0]
if not isinstance(arg, ast.Constant):
return
target = arg.value
if not target.endswith(".py"):
return
target = target.removesuffix(".py")
self.__imports.add(f"test.{target}")
def parse_test_imports(content: str) -> frozenset[str]:
"""Parse test file content and extract test package dependencies."""
if not (tree := safe_parse_ast(content)):
return set()
visitor = ImportVisitor()
visitor.visit(tree)
return visitor.test_imports
def parse_lib_imports(content: str) -> frozenset[str]:
"""Parse library file and extract all imported module names."""
if not (tree := safe_parse_ast(content)):
return set()
visitor = ImportVisitor()
visitor.visit(tree)
return visitor.lib_imports
# === TODO marker utilities ===
TODO_MARKER = "TODO: RUSTPYTHON"
def filter_rustpython_todo(content: str) -> str:
"""Remove lines containing RustPython TODO markers."""
lines = content.splitlines(keepends=True)
return "".join(line for line in lines if TODO_MARKER not in line)
def count_rustpython_todo(content: str) -> int:
"""Count lines containing RustPython TODO markers."""
return content.count(TODO_MARKER)
def count_todo_in_path(path: pathlib.Path) -> int:
"""Count RustPython TODO markers in a file or directory of .py files."""
if path.is_file():
content = safe_read_text(path)
return count_rustpython_todo(content) if content else 0
return sum(count_rustpython_todo(content) for _, content in read_python_files(path))
# === Test utilities ===
def _get_cpython_test_path(test_name: str, cpython_prefix: str) -> pathlib.Path | None:
"""Return the CPython test path for a test name, or None if missing."""
cpython_path = resolve_test_path(test_name, cpython_prefix, prefer="dir")
return cpython_path if cpython_path.exists() else None
def _get_local_test_path(
cpython_test_path: pathlib.Path, lib_prefix: str
) -> pathlib.Path:
"""Return the local Lib/test path matching a CPython test path."""
return pathlib.Path(lib_prefix) / "test" / cpython_test_path.name
def is_test_tracked(test_name: str, cpython_prefix: str, lib_prefix: str) -> bool:
"""Check if a test exists in the local Lib/test."""
cpython_path = _get_cpython_test_path(test_name, cpython_prefix)
if cpython_path is None:
return True
local_path = _get_local_test_path(cpython_path, lib_prefix)
return local_path.exists()
def is_test_up_to_date(test_name: str, cpython_prefix: str, lib_prefix: str) -> bool:
"""Check if a test is up-to-date, ignoring RustPython TODO markers."""
cpython_path = _get_cpython_test_path(test_name, cpython_prefix)
if cpython_path is None:
return True
local_path = _get_local_test_path(cpython_path, lib_prefix)
if not local_path.exists():
return False
if cpython_path.is_file():
return compare_file_contents(
cpython_path, local_path, local_filter=filter_rustpython_todo
)
return compare_dir_contents(
cpython_path, local_path, local_filter=filter_rustpython_todo
)
def count_test_todos(test_name: str, lib_prefix: str) -> int:
"""Count RustPython TODO markers in a test file/directory."""
local_dir = pathlib.Path(lib_prefix) / "test" / test_name
local_file = pathlib.Path(lib_prefix) / "test" / f"{test_name}.py"
if local_dir.exists():
return count_todo_in_path(local_dir)
if local_file.exists():
return count_todo_in_path(local_file)
return 0
# === Cross-process cache using shelve ===
def _get_cpython_version(cpython_prefix: str) -> str:
"""Get CPython version from git tag for cache namespace."""
try:
result = subprocess.run(
["git", "describe", "--tags", "--abbrev=0"],
cwd=cpython_prefix,
capture_output=True,
text=True,
)
if result.returncode == 0:
return result.stdout.strip()
except Exception:
pass
return "unknown"
def _get_cache_path() -> str:
"""Get cache file path (without extension - shelve adds its own)."""
cache_dir = pathlib.Path(__file__).parent / ".cache"
cache_dir.mkdir(parents=True, exist_ok=True)
return str(cache_dir / "import_graph_cache")
def clear_import_graph_caches() -> None:
"""Clear in-process import graph caches (for testing)."""
if "_test_import_graph_cache" in globals():
globals()["_test_import_graph_cache"].clear()
if "_lib_import_graph_cache" in globals():
globals()["_lib_import_graph_cache"].clear()
# Manual dependency table for irregular cases
# Format: "name" -> {"lib": [...], "test": [...], "data": [...], "hard_deps": [...]}
# - lib: override default path (default: name.py or name/)
# - hard_deps: additional files to copy alongside the main module
DEPENDENCIES = {
# regrtest is in Lib/test/libregrtest/, not Lib/libregrtest/
"regrtest": {
"lib": ["test/libregrtest"],
"test": ["test_regrtest"],
"data": ["test/regrtestdata"],
},
# Rust-implemented modules (no lib file, only test)
"int": {
"lib": [],
"hard_deps": ["_pylong.py"],
"test": [
"test_int.py",
"test_long.py",
"test_int_literal.py",
],
},
"exception": {
"lib": [],
"test": [
"test_exceptions.py",
"test_baseexception.py",
"test_except_star.py",
"test_exception_group.py",
"test_exception_hierarchy.py",
"test_exception_variations.py",
],
},
"dict": {
"lib": [],
"test": [
"test_dict.py",
"test_dictcomps.py",
"test_dictviews.py",
"test_userdict.py",
"mapping_tests.py",
],
},
"list": {
"lib": [],
"test": [
"test_list.py",
"test_listcomps.py",
"test_userlist.py",
],
},
"__future__": {
"test": [
"test___future__.py",
"test_future_stmt.py",
],
},
"site": {
"hard_deps": ["_sitebuiltins.py"],
},
"opcode": {
"hard_deps": ["_opcode_metadata.py"],
"test": [
"test_opcode.py",
"test__opcode.py",
"test_opcodes.py",
],
},
"pickle": {
"hard_deps": ["_compat_pickle.py"],
"test": [
"picklecommon.py",
"test_pickle.py",
"test_picklebuffer.py",
"test_pickletools.py",
"test_xpickle.py",
"xpickle_worker.py",
],
},
"re": {
"hard_deps": ["sre_compile.py", "sre_constants.py", "sre_parse.py"],
"test": [
"test_re.py",
"re_tests.py",
],
},
"weakref": {
"hard_deps": ["_weakrefset.py"],
"test": [
"test_weakref.py",
"test_weakset.py",
],
},
"codecs": {
"test": [
"test_charmapcodec.py",
"test_codeccallbacks.py",
"test_codecencodings_cn.py",
"test_codecencodings_hk.py",
"test_codecencodings_iso2022.py",
"test_codecencodings_jp.py",
"test_codecencodings_kr.py",
"test_codecencodings_tw.py",
"test_codecmaps_cn.py",
"test_codecmaps_hk.py",
"test_codecmaps_jp.py",
"test_codecmaps_kr.py",
"test_codecmaps_tw.py",
"test_codecs.py",
"test_multibytecodec.py",
"testcodec.py",
],
},
# Non-pattern hard_deps (can't be auto-detected)
"ast": {
"hard_deps": ["_ast_unparse.py"],
"test": [
"test_ast.py",
"test_unparse.py",
"test_type_comments.py",
],
},
# Data directories
"pydoc": {
"hard_deps": ["pydoc_data"],
},
"turtle": {
"hard_deps": ["turtledemo"],
},
"sysconfig": {
"hard_deps": ["_aix_support.py", "_osx_support.py"],
"test": [
"test_sysconfig.py",
"test__osx_support.py",
],
},
"tkinter": {
"test": [
"test_tkinter",
"test_ttk",
"test_ttk_textonly.py",
"test_tcl.py",
"test_idle",
],
},
# Test support library (like regrtest)
"support": {
"lib": ["test/support"],
"data": ["test/wheeldata"],
"test": [
"test_support.py",
"test_script_helper.py",
],
},
# test_htmlparser tests html.parser
"html": {
"hard_deps": ["_markupbase.py"],
"test": ["test_html.py", "test_htmlparser.py"],
},
"xml": {
"test": [
"test_xml_etree.py",
"test_xml_etree_c.py",
"test_minidom.py",
"test_pulldom.py",
"test_pyexpat.py",
"test_sax.py",
"test_xml_dom_minicompat.py",
"test_xml_dom_xmlbuilder.py",
],
},
"multiprocessing": {
"test": [
"test_multiprocessing_fork",
"test_multiprocessing_forkserver",
"test_multiprocessing_spawn",
"test_multiprocessing_main_handling.py",
"_test_multiprocessing.py",
],
},
"urllib": {
"test": [
"test_urllib.py",
"test_urllib2.py",
"test_urllib2_localnet.py",
"test_urllib2net.py",
"test_urllibnet.py",
"test_urlparse.py",
"test_urllib_response.py",
"test_robotparser.py",
],
},
"collections": {
"hard_deps": ["_collections_abc.py"],
"test": [
"test_collections.py",
"test_deque.py",
"test_defaultdict.py",
"test_ordered_dict.py",
],
},
"http": {
"test": [
"test_httplib.py",
"test_http_cookiejar.py",
"test_http_cookies.py",
"test_httpservers.py",
],
},
"unicode": {
"lib": [],
"test": [
"test_unicodedata.py",
"test_unicode_file.py",
"test_unicode_file_functions.py",
"test_unicode_identifiers.py",
"test_ucn.py",
],
},
"typing": {
"test": [
"test_typing.py",
"test_type_aliases.py",
"test_type_annotations.py",
"test_type_params.py",
"test_genericalias.py",
],
},
"unpack": {
"lib": [],
"test": [
"test_unpack.py",
"test_unpack_ex.py",
],
},
"zipimport": {
"test": [
"test_zipimport.py",
"test_zipimport_support.py",
],
},
"time": {
"lib": [],
"test": [
"test_time.py",
"test_strftime.py",
],
},
"sys": {
"lib": [],
"test": [
"test_sys.py",
"test_syslog.py",
"test_sys_setprofile.py",
"test_sys_settrace.py",
"test_audit.py",
"audit-tests.py",
],
},
"str": {
"lib": [],
"test": [
"test_str.py",
"test_fstring.py",
"test_string_literals.py",
],
},
"thread": {
"lib": [],
"test": [
"test_thread.py",
"test_thread_local_bytecode.py",
"test_threadsignals.py",
],
},
"threading": {
"hard_deps": ["_threading_local.py"],
"test": [
"test_threading.py",
"test_threadedtempfile.py",
"test_threading_local.py",
],
},
"class": {
"lib": [],
"test": [
"test_class.py",
"test_genericclass.py",
"test_subclassinit.py",
],
},
"generator": {
"lib": [],
"test": [
"test_generators.py",
"test_genexps.py",
"test_generator_stop.py",
"test_yield_from.py",
],
},
"descr": {
"lib": [],
"test": [
"test_descr.py",
"test_descrtut.py",
],
},
"code": {
"test": [
"test_code_module.py",
],
},
"contextlib": {
"test": [
"test_contextlib.py",
"test_contextlib_async.py",
],
},
"io": {
"hard_deps": ["_pyio.py"],
"test": [
"test_io",
],
},
"dbm": {
"test": [
"test_dbm.py",
"test_dbm_dumb.py",
"test_dbm_gnu.py",
"test_dbm_ndbm.py",
"test_dbm_sqlite3.py",
],
},
"datetime": {
"hard_deps": ["_strptime.py"],
"test": [
"test_datetime.py",
"test_strptime.py",
],
},
"locale": {
"test": [
"test_locale.py",
"test__locale.py",
],
},
"numbers": {
"test": [
"test_numbers.py",
"test_abstract_numbers.py",
],
},
"file": {
"lib": [],
"test": [
"test_io",
],
},
"fcntl": {
"lib": [],
"test": [
"test_fcntl.py",
"test_ioctl.py",
],
},
"select": {
"lib": [],
"test": [
"test_select.py",
"test_poll.py",
],
},
"xmlrpc": {
"test": [
"test_xmlrpc.py",
"test_docxmlrpc.py",
],
},
"ctypes": {
"test": [
"test_ctypes",
"test_stable_abi_ctypes.py",
],
},
# Grouped tests for modules without custom lib paths
"compile": {
"lib": [],
"test": [
"test_compile.py",
"test_compiler_assemble.py",
"test_compiler_codegen.py",
"test_peepholer.py",
],
},
"math": {
"lib": [],
"test": [
"test_math.py",
"test_math_integer.py",
"test_math_property.py",
],
},
"float": {
"lib": [],
"test": [
"test_float.py",
"test_strtod.py",
],
},
"zipfile": {
"test": [
"test_zipfile.py",
"test_zipfile64.py",
],
},
"smtplib": {
"test": [
"test_smtplib.py",
"test_smtpnet.py",
],
},
"profile": {
"test": [
"test_profile.py",
"test_cprofile.py",
],
},
"string": {
"test": [
"test_string.py",
"test_userstring.py",
],
},
"os": {
"test": [
"test_os.py",
"test_popen.py",
],
},
"pyrepl": {
"test": [
"test_pyrepl",
"test_repl.py",
],
},
"concurrent": {
"test": [
"test_concurrent_futures",
"test_interpreters",
"test__interpreters.py",
"test__interpchannels.py",
"test_crossinterp.py",
],
},
"atexit": {
"test": [
"test_atexit.py",
"_test_atexit.py",
],
},
"eintr": {
"test": [
"test_eintr.py",
"_test_eintr.py",
]
},
"curses": {
"test": [
"test_curses.py",
"curses_tests.py",
],
},
"signal": {
"test": [
"signalinterproctester.py",
"test_signal.py",
]
},
}
def resolve_hard_dep_parent(name: str, cpython_prefix: str) -> str | None:
"""Resolve a hard_dep name to its parent module.
Only returns a parent if the file is actually tracked:
- Explicitly listed in DEPENDENCIES as a hard_dep
- Or auto-detected _py{module}.py pattern where the parent module exists
Args:
name: Module or file name (with or without .py extension)
cpython_prefix: CPython directory prefix
Returns:
Parent module name if found and tracked, None otherwise
"""
# Normalize: remove .py extension if present
if name.endswith(".py"):
name = name[:-3]
# Check DEPENDENCIES table first (explicit hard_deps)
for module_name, dep_info in DEPENDENCIES.items():
hard_deps = dep_info.get("hard_deps", [])
for dep in hard_deps:
# Normalize dep: remove .py extension
dep_normalized = dep[:-3] if dep.endswith(".py") else dep
if dep_normalized == name:
return module_name
# Auto-detect _py{module} or _py_{module} patterns
# Only if the parent module actually exists
if name.startswith("_py"):
# _py_abc -> abc
# _pydatetime -> datetime
parent = name.removeprefix("_py_").removeprefix("_py")
# Verify the parent module exists
lib_dir = pathlib.Path(cpython_prefix) / "Lib"
parent_file = lib_dir / f"{parent}.py"
parent_dir = lib_dir / parent
if parent_file.exists() or (
parent_dir.exists() and (parent_dir / "__init__.py").exists()
):
return parent
return None
def resolve_test_to_lib(test_name: str) -> str | None:
"""Resolve a test name to its library group from DEPENDENCIES.
Args:
test_name: Test name with or without test_ prefix (e.g., "test_urllib2" or "urllib2")
Returns:
Library name if test belongs to a group, None otherwise
"""
# Normalize: add test_ prefix if not present
if not test_name.startswith("test_"):
test_name = f"test_{test_name}"
for lib_name, dep_info in DEPENDENCIES.items():
tests = dep_info.get("test", [])
for test_path in tests:
# test_path is like "test_urllib2.py" or "test_multiprocessing_fork"
path_stem = test_path.removesuffix(".py")
if path_stem == test_name:
return lib_name
return None
# Test-specific dependencies (only when auto-detection isn't enough)
# - hard_deps: files to migrate (tightly coupled, must be migrated together)
# - data: directories to copy without migration
TEST_DEPENDENCIES = {
# Audio tests
"test_winsound": {
"data": ["audiodata"],
},
"test_wave": {
"data": ["audiodata"],
},
"audiotests": {
"data": ["audiodata"],
},
# Archive tests
"test_tarfile": {
"data": ["archivetestdata"],
},
"test_zipfile": {
"data": ["archivetestdata"],
},
# Config tests
"test_configparser": {
"data": ["configdata"],
},
"test_config": {
"data": ["configdata"],
},
# Other data directories
"test_decimal": {
"data": ["decimaltestdata"],
},
"test_dtrace": {
"data": ["dtracedata"],
},
"test_math": {
"data": ["mathdata"],
},
"test_ssl": {
"data": ["certdata"],
},
"test_subprocess": {
"data": ["subprocessdata"],
},
"test_tkinter": {
"data": ["tkinterdata"],
},
"test_tokenize": {
"data": ["tokenizedata"],
},
"test_type_annotations": {
"data": ["typinganndata"],
},
"test_zipimport": {
"data": ["zipimport_data"],
},
# XML tests share xmltestdata
"test_xml_etree": {
"data": ["xmltestdata"],
},
"test_pulldom": {
"data": ["xmltestdata"],
},
"test_sax": {
"data": ["xmltestdata"],
},
"test_minidom": {
"data": ["xmltestdata"],
},
# Multibytecodec support needs cjkencodings
"multibytecodec_support": {
"data": ["cjkencodings"],
},
# i18n
"i18n_helper": {
"data": ["translationdata"],
},
# wheeldata is used by test_makefile and support
"test_makefile": {
"data": ["wheeldata"],
},
# profilee is used by test_monitoring
"test_monitoring": {
"hard_deps": ["profilee"],
},
}
@functools.cache
def get_lib_paths(name: str, cpython_prefix: str) -> tuple[pathlib.Path, ...]:
"""Get all library paths for a module.
Args:
name: Module name (e.g., "datetime", "libregrtest")
cpython_prefix: CPython directory prefix
Returns:
Tuple of paths to copy
"""
dep_info = DEPENDENCIES.get(name, {})
# Get main lib path (override or default)
if "lib" in dep_info:
paths = [construct_lib_path(cpython_prefix, p) for p in dep_info["lib"]]
else:
# Default: try file first, then directory
paths = [resolve_module_path(name, cpython_prefix, prefer="file")]
# Add hard_deps from DEPENDENCIES
for dep in dep_info.get("hard_deps", []):
paths.append(construct_lib_path(cpython_prefix, dep))
# Auto-detect _py{module}.py or _py_{module}.py patterns
for pattern in [f"_py{name}.py", f"_py_{name}.py"]:
auto_path = construct_lib_path(cpython_prefix, pattern)
if auto_path.exists() and auto_path not in paths:
paths.append(auto_path)
return tuple(paths)
def get_all_hard_deps(name: str, cpython_prefix: str) -> list[str]:
"""Get all hard_deps for a module (explicit + auto-detected).
Args:
name: Module name (e.g., "decimal", "datetime")
cpython_prefix: CPython directory prefix
Returns:
List of hard_dep names (without .py extension)
"""
dep_info = DEPENDENCIES.get(name, {})
hard_deps = set()
# Explicit hard_deps from DEPENDENCIES
for hd in dep_info.get("hard_deps", []):
# Remove .py extension if present
hard_deps.add(hd[:-3] if hd.endswith(".py") else hd)
# Auto-detect _py{module}.py or _py_{module}.py patterns
for pattern in [f"_py{name}.py", f"_py_{name}.py"]:
auto_path = construct_lib_path(cpython_prefix, pattern)
if auto_path.exists():
hard_deps.add(auto_path.stem)
return sorted(hard_deps)
@functools.cache
def get_test_paths(name: str, cpython_prefix: str) -> tuple[pathlib.Path, ...]:
"""Get all test paths for a module.
Args:
name: Module name (e.g., "datetime", "libregrtest")
cpython_prefix: CPython directory prefix
Returns:
Tuple of test paths
"""
if name in DEPENDENCIES and "test" in DEPENDENCIES[name]:
return tuple(
construct_lib_path(cpython_prefix, f"test/{p}")
for p in DEPENDENCIES[name]["test"]
)
# Default: try directory first, then file
return (resolve_module_path(f"test/test_{name}", cpython_prefix, prefer="dir"),)
@functools.cache
def get_all_imports(name: str, cpython_prefix: str) -> frozenset[str]:
"""Get all imports from a library file.
Args:
name: Module name
cpython_prefix: CPython directory prefix
Returns:
Frozenset of all imported module names
"""
all_imports = set()
for lib_path in get_lib_paths(name, cpython_prefix):
if lib_path.exists():
for _, content in read_python_files(lib_path):
all_imports.update(parse_lib_imports(content))
# Remove self
all_imports.discard(name)
return frozenset(all_imports)
@functools.cache
def get_soft_deps(name: str, cpython_prefix: str) -> frozenset[str]:
"""Get soft dependencies by parsing imports from library file.
Args:
name: Module name
cpython_prefix: CPython directory prefix
Returns:
Frozenset of imported stdlib module names (those that exist in cpython/Lib/)
"""
all_imports = get_all_imports(name, cpython_prefix)
# Filter: only include modules that exist in cpython/Lib/
stdlib_deps = set()
for imp in all_imports:
module_path = resolve_module_path(imp, cpython_prefix)
if module_path.exists():
stdlib_deps.add(imp)
return frozenset(stdlib_deps)
@functools.cache
def get_rust_deps(name: str, cpython_prefix: str) -> frozenset[str]:
"""Get Rust/C dependencies (imports that don't exist in cpython/Lib/).
Args:
name: Module name
cpython_prefix: CPython directory prefix
Returns:
Frozenset of imported module names that are built-in or C extensions
"""
all_imports = get_all_imports(name, cpython_prefix)
soft_deps = get_soft_deps(name, cpython_prefix)
return frozenset(all_imports - soft_deps)
def is_path_synced(
cpython_path: pathlib.Path,
cpython_prefix: str,
lib_prefix: str,
) -> bool:
"""Check if a CPython path is synced with local.
Args:
cpython_path: Path in CPython directory
cpython_prefix: CPython directory prefix
lib_prefix: Local Lib directory prefix
Returns:
True if synced, False otherwise
"""
local_path = cpython_to_local_path(cpython_path, cpython_prefix, lib_prefix)
if local_path is None:
return False
return compare_paths(cpython_path, local_path)
@functools.cache
def is_up_to_date(name: str, cpython_prefix: str, lib_prefix: str) -> bool:
"""Check if a module is up-to-date by comparing files.
Args:
name: Module name
cpython_prefix: CPython directory prefix
lib_prefix: Local Lib directory prefix
Returns:
True if all files match, False otherwise
"""
lib_paths = get_lib_paths(name, cpython_prefix)
found_any = False
for cpython_path in lib_paths:
if not cpython_path.exists():
continue
found_any = True
# Convert cpython path to local path
# cpython/Lib/foo.py -> Lib/foo.py
rel_path = cpython_path.relative_to(cpython_prefix)
local_path = pathlib.Path(lib_prefix) / rel_path.relative_to("Lib")
if not compare_paths(cpython_path, local_path):
return False
if not found_any:
dep_info = DEPENDENCIES.get(name, {})
if dep_info.get("lib") == []:
return True
return found_any
def _count_file_diff(file_a: pathlib.Path, file_b: pathlib.Path) -> int:
"""Count changed lines between two text files using difflib."""
a_content = safe_read_text(file_a)
b_content = safe_read_text(file_b)
if a_content is None or b_content is None:
return 0
if a_content == b_content:
return 0
a_lines = a_content.splitlines()
b_lines = b_content.splitlines()
count = 0
for line in difflib.unified_diff(a_lines, b_lines, lineterm=""):
if (line.startswith("+") and not line.startswith("+++")) or (
line.startswith("-") and not line.startswith("---")
):
count += 1
return count
def _count_path_diff(path_a: pathlib.Path, path_b: pathlib.Path) -> int:
"""Count changed lines between two paths (file or directory, *.py only)."""
if path_a.is_file() and path_b.is_file():
return _count_file_diff(path_a, path_b)
if path_a.is_dir() and path_b.is_dir():
total = 0
a_files = {f.relative_to(path_a) for f in path_a.rglob("*.py")}
b_files = {f.relative_to(path_b) for f in path_b.rglob("*.py")}
for rel in a_files & b_files:
total += _count_file_diff(path_a / rel, path_b / rel)
for rel in a_files - b_files:
content = safe_read_text(path_a / rel)
if content:
total += len(content.splitlines())
for rel in b_files - a_files:
content = safe_read_text(path_b / rel)
if content:
total += len(content.splitlines())
return total
return 0
@functools.cache
def _bulk_last_updated() -> dict[str, str]:
"""Get last git commit dates for all paths under Lib/ in one git call.
Keys are Lib/-relative paths (e.g. "re/__init__.py", "test/test_os.py",
"os.py"), plus directory rollups (e.g. "re", "test/test_zoneinfo").
Returns:
Dict mapping Lib/-relative path to date string.
"""
file_map: dict[str, str] = {}
try:
result = subprocess.run(
["git", "log", "--format=%cd", "--date=short", "--name-only", "--", "Lib/"],
capture_output=True,
text=True,
timeout=30,
)
if result.returncode != 0:
return file_map
except Exception:
return file_map
current_date = None
for line in result.stdout.splitlines():
line = line.strip()
if not line:
continue
# Date lines are YYYY-MM-DD format
if len(line) == 10 and line[4] == "-" and line[7] == "-":
current_date = line
elif current_date and line.startswith("Lib/"):
# Strip "Lib/" prefix to get Lib-relative key
rel = line[4:]
if rel and rel not in file_map:
file_map[rel] = current_date
# Pre-compute directory rollups
dir_map: dict[str, str] = {}
for filepath, date in file_map.items():
parts = filepath.split("/")
for i in range(1, len(parts)):
dirpath = "/".join(parts[:i])
if dirpath not in dir_map or date > dir_map[dirpath]:
dir_map[dirpath] = date
dir_map.update(file_map)
return dir_map
@functools.cache
def _lib_prefix_stripped(lib_prefix: str) -> str:
"""Get the normalized prefix to strip from paths, with trailing /."""
# e.g. "Lib" -> "Lib/", "./Lib" -> "Lib/", "../Lib" -> "../Lib/"
return pathlib.Path(lib_prefix).as_posix().rstrip("/") + "/"
def _lookup_last_updated(paths: list[str], lib_prefix: str) -> str | None:
"""Look up the most recent date among paths from the bulk cache."""
cache = _bulk_last_updated()
prefix = _lib_prefix_stripped(lib_prefix)
latest = None
for p in paths:
p_norm = pathlib.Path(p).as_posix()
# Strip lib_prefix to get Lib-relative key
# e.g. "Lib/test/test_os.py" -> "test/test_os.py"
# "../Lib/re" -> "re"
if p_norm.startswith(prefix):
key = p_norm[len(prefix) :]
else:
key = p_norm
date = cache.get(key)
if date and (latest is None or date > latest):
latest = date
return latest
def get_module_last_updated(
name: str, cpython_prefix: str, lib_prefix: str
) -> str | None:
"""Get the last git commit date for a module's Lib files."""
local_paths = []
for cpython_path in get_lib_paths(name, cpython_prefix):
if not cpython_path.exists():
continue
try:
rel_path = cpython_path.relative_to(cpython_prefix)
local_path = pathlib.Path(lib_prefix) / rel_path.relative_to("Lib")
if local_path.exists():
local_paths.append(str(local_path))
except ValueError:
continue
if not local_paths:
return None
return _lookup_last_updated(local_paths, lib_prefix)
def get_module_diff_stat(name: str, cpython_prefix: str, lib_prefix: str) -> int:
"""Count differing lines between cpython and local Lib for a module."""
total = 0
for cpython_path in get_lib_paths(name, cpython_prefix):
if not cpython_path.exists():
continue
try:
rel_path = cpython_path.relative_to(cpython_prefix)
local_path = pathlib.Path(lib_prefix) / rel_path.relative_to("Lib")
except ValueError:
continue
if not local_path.exists():
continue
total += _count_path_diff(cpython_path, local_path)
return total
def get_test_last_updated(
test_name: str, cpython_prefix: str, lib_prefix: str
) -> str | None:
"""Get the last git commit date for a test's files."""
cpython_path = _get_cpython_test_path(test_name, cpython_prefix)
if cpython_path is None:
return None
local_path = _get_local_test_path(cpython_path, lib_prefix)
if not local_path.exists():
return None
return _lookup_last_updated([str(local_path)], lib_prefix)
def get_test_dependencies(
test_path: pathlib.Path,
) -> dict[str, list[pathlib.Path]]:
"""Get test dependencies by parsing imports.
Args:
test_path: Path to test file or directory
Returns:
Dict with "hard_deps" (files to migrate) and "data" (dirs to copy)
"""
result = {"hard_deps": [], "data": []}
if not test_path.exists():
return result
# Parse all files for imports (auto-detect deps)
all_imports = set()
for _, content in read_python_files(test_path):
all_imports.update(parse_test_imports(content))
# Also add manual dependencies from TEST_DEPENDENCIES
test_name = test_path.stem if test_path.is_file() else test_path.name
manual_deps = TEST_DEPENDENCIES.get(test_name, {})
if "hard_deps" in manual_deps:
all_imports.update(manual_deps["hard_deps"])
# Convert imports to paths (deps)
for imp in all_imports:
# Skip other test modules (test_*) - they are independently managed
# via their own update_lib entry. Only support/helper modules
# (e.g., string_tests, mapping_tests) should be treated as hard deps.
if imp.startswith("test_"):
continue
dep_path = test_path.parent / f"{imp}.py"
if not dep_path.exists():
dep_path = test_path.parent / imp
if dep_path.exists() and dep_path not in result["hard_deps"]:
result["hard_deps"].append(dep_path)
# Add data paths from manual table (for the test file itself)
if "data" in manual_deps:
for data_name in manual_deps["data"]:
data_path = test_path.parent / data_name
if data_path.exists() and data_path not in result["data"]:
result["data"].append(data_path)
# Also add data from auto-detected deps' TEST_DEPENDENCIES
# e.g., test_codecencodings_kr -> multibytecodec_support -> cjkencodings
for imp in all_imports:
dep_info = TEST_DEPENDENCIES.get(imp, {})
if "data" in dep_info:
for data_name in dep_info["data"]:
data_path = test_path.parent / data_name
if data_path.exists() and data_path not in result["data"]:
result["data"].append(data_path)
return result
def _parse_test_submodule_imports(content: str) -> dict[str, set[str]]:
"""Parse 'from test.X import Y' to get submodule imports.
Args:
content: Python file content
Returns:
Dict mapping submodule (e.g., "test_bar") -> set of imported names (e.g., {"helper"})
"""
tree = safe_parse_ast(content)
if tree is None:
return {}
result: dict[str, set[str]] = {}
for node in ast.walk(tree):
if isinstance(node, ast.ImportFrom):
if node.module and node.module.startswith("test."):
# from test.test_bar import helper -> test_bar: {helper}
parts = node.module.split(".")
if len(parts) >= 2:
submodule = parts[1]
if submodule not in ("support", "__init__"):
if submodule not in result:
result[submodule] = set()
for alias in node.names:
result[submodule].add(alias.name)
return result
_test_import_graph_cache: dict[
str, tuple[dict[str, set[str]], dict[str, set[str]]]
] = {}
def _is_standard_lib_path(path: str) -> bool:
"""Check if path is the standard Lib directory (not a temp dir)."""
if "/tmp" in path.lower() or "/var/folders" in path.lower():
return False
return (
path == "Lib/test"
or path.endswith("/Lib/test")
or path == "Lib"
or path.endswith("/Lib")
)
def _build_test_import_graph(
test_dir: pathlib.Path,
) -> tuple[dict[str, set[str]], dict[str, set[str]]]:
"""Build import graphs for files within test directory (recursive).
Uses cross-process shelve cache based on CPython version.
Args:
test_dir: Path to Lib/test/ directory
Returns:
Tuple of:
- Dict mapping relative path (without .py) -> set of test modules it imports
- Dict mapping relative path (without .py) -> set of all lib imports
"""
# In-process cache
cache_key = str(test_dir)
if cache_key in _test_import_graph_cache:
return _test_import_graph_cache[cache_key]
# Cross-process cache (only for standard Lib/test directory)
use_file_cache = _is_standard_lib_path(cache_key)
if use_file_cache:
version = _get_cpython_version("cpython")
shelve_key = f"test_import_graph:{version}"
try:
with shelve.open(_get_cache_path()) as db:
if shelve_key in db:
import_graph, lib_imports_graph = db[shelve_key]
_test_import_graph_cache[cache_key] = (
import_graph,
lib_imports_graph,
)
return import_graph, lib_imports_graph
except Exception:
pass
# Build from scratch
import_graph: dict[str, set[str]] = {}
lib_imports_graph: dict[str, set[str]] = {}
for py_file in test_dir.glob("**/*.py"):
content = safe_read_text(py_file)
if content is None:
continue
imports = set()
imports.update(parse_test_imports(content))
all_imports = parse_lib_imports(content)
for imp in all_imports:
if (py_file.parent / f"{imp}.py").exists():
imports.add(imp)
if (test_dir / f"{imp}.py").exists():
imports.add(imp)
submodule_imports = _parse_test_submodule_imports(content)
for submodule, imported_names in submodule_imports.items():
submodule_dir = test_dir / submodule
if submodule_dir.is_dir():
for name in imported_names:
if (submodule_dir / f"{name}.py").exists():
imports.add(name)
rel_path = py_file.relative_to(test_dir)
key = str(rel_path.with_suffix(""))
import_graph[key] = imports
lib_imports_graph[key] = all_imports
# Save to cross-process cache
if use_file_cache:
try:
with shelve.open(_get_cache_path()) as db:
db[shelve_key] = (import_graph, lib_imports_graph)
except Exception:
pass
_test_import_graph_cache[cache_key] = (import_graph, lib_imports_graph)
return import_graph, lib_imports_graph
_lib_import_graph_cache: dict[str, dict[str, set[str]]] = {}
def _build_lib_import_graph(lib_prefix: str) -> dict[str, set[str]]:
"""Build import graph for Lib modules (full module paths like urllib.request).
Uses cross-process shelve cache based on CPython version.
Args:
lib_prefix: RustPython Lib directory
Returns:
Dict mapping full_module_path -> set of modules it imports
"""
# In-process cache
if lib_prefix in _lib_import_graph_cache:
return _lib_import_graph_cache[lib_prefix]
# Cross-process cache (only for standard Lib directory)
use_file_cache = _is_standard_lib_path(lib_prefix)
if use_file_cache:
version = _get_cpython_version("cpython")
shelve_key = f"lib_import_graph:{version}"
try:
with shelve.open(_get_cache_path()) as db:
if shelve_key in db:
import_graph = db[shelve_key]
_lib_import_graph_cache[lib_prefix] = import_graph
return import_graph
except Exception:
pass
# Build from scratch
lib_dir = pathlib.Path(lib_prefix)
if not lib_dir.exists():
return {}
import_graph: dict[str, set[str]] = {}
for entry in lib_dir.iterdir():
if entry.name.startswith(("_", ".")):
continue
if entry.name == "test":
continue
if entry.is_file() and entry.suffix == ".py":
content = safe_read_text(entry)
if content:
imports = parse_lib_imports(content)
imports.discard(entry.stem)
import_graph[entry.stem] = imports
elif entry.is_dir() and (entry / "__init__.py").exists():
for py_file in entry.glob("**/*.py"):
content = safe_read_text(py_file)
if content:
imports = parse_lib_imports(content)
rel_path = py_file.relative_to(lib_dir)
if rel_path.name == "__init__.py":
full_name = str(rel_path.parent).replace("/", ".")
else:
full_name = str(rel_path.with_suffix("")).replace("/", ".")
imports.discard(full_name.split(".")[0])
import_graph[full_name] = imports
# Save to cross-process cache
if use_file_cache:
try:
with shelve.open(_get_cache_path()) as db:
db[shelve_key] = import_graph
except Exception:
pass
_lib_import_graph_cache[lib_prefix] = import_graph
return import_graph
def _get_lib_modules_importing(
module_name: str, lib_import_graph: dict[str, set[str]]
) -> set[str]:
"""Find Lib modules (full paths) that import module_name or any of its submodules."""
importers: set[str] = set()
target_top = module_name.split(".")[0]
for full_path, imports in lib_import_graph.items():
if full_path.split(".")[0] == target_top:
continue # Skip same package
# Match if module imports target OR any submodule of target
# e.g., for "xml": match imports of "xml", "xml.parsers", "xml.etree.ElementTree"
matches = any(
imp == module_name or imp.startswith(module_name + ".") for imp in imports
)
if matches:
importers.add(full_path)
return importers
def _consolidate_submodules(
modules: set[str], threshold: int = 3
) -> dict[str, set[str]]:
"""Consolidate submodules if count exceeds threshold.
Args:
modules: Set of full module paths (e.g., {"urllib.request", "urllib.parse", "xml.dom", "xml.sax"})
threshold: If submodules > threshold, consolidate to parent
Returns:
Dict mapping display_name -> set of original module paths
e.g., {"urllib.request": {"urllib.request"}, "xml": {"xml.dom", "xml.sax", "xml.etree", "xml.parsers"}}
"""
# Group by top-level package
by_package: dict[str, set[str]] = {}
for mod in modules:
parts = mod.split(".")
top = parts[0]
if top not in by_package:
by_package[top] = set()
by_package[top].add(mod)
result: dict[str, set[str]] = {}
for top, submods in by_package.items():
if len(submods) > threshold:
# Consolidate to top-level
result[top] = submods
else:
# Keep individual
for mod in submods:
result[mod] = {mod}
return result
# Modules that are used everywhere - show but don't expand their dependents
_BLOCKLIST_MODULES = frozenset(
{
"unittest",
"test.support",
"support",
"doctest",
"typing",
"abc",
"collections.abc",
"functools",
"itertools",
"operator",
"contextlib",
"warnings",
"types",
"enum",
"re",
"io",
"os",
"sys",
}
)
def find_dependent_tests_tree(
module_name: str,
lib_prefix: str,
max_depth: int = 1,
_depth: int = 0,
_visited_tests: set[str] | None = None,
_visited_modules: set[str] | None = None,
) -> dict:
"""Find dependent tests in a tree structure.
Args:
module_name: Module to search for (e.g., "ftplib")
lib_prefix: RustPython Lib directory
max_depth: Maximum depth to recurse (default 1 = show direct + 1 level of Lib deps)
Returns:
Dict with structure:
{
"module": "ftplib",
"tests": ["test_ftplib", "test_urllib2"], # Direct importers
"children": [
{"module": "urllib.request", "tests": [...], "children": []},
...
]
}
"""
lib_dir = pathlib.Path(lib_prefix)
test_dir = lib_dir / "test"
if _visited_tests is None:
_visited_tests = set()
if _visited_modules is None:
_visited_modules = set()
# Build graphs
test_import_graph, test_lib_imports = _build_test_import_graph(test_dir)
lib_import_graph = _build_lib_import_graph(lib_prefix)
# Find tests that directly import this module
target_top = module_name.split(".")[0]
direct_tests: set[str] = set()
for file_key, imports in test_lib_imports.items():
if file_key in _visited_tests:
continue
# Match exact module OR any child submodule
# e.g., "xml" matches imports of "xml", "xml.parsers", "xml.etree.ElementTree"
# but "collections._defaultdict" only matches "collections._defaultdict" (no children)
matches = any(
imp == module_name or imp.startswith(module_name + ".") for imp in imports
)
if matches:
# Check if it's a test file
if pathlib.Path(file_key).name.startswith("test_"):
direct_tests.add(file_key)
_visited_tests.add(file_key)
# Consolidate test names (test_sqlite3/test_dbapi -> test_sqlite3)
consolidated_tests = {_consolidate_file_key(t) for t in direct_tests}
# Mark this module as visited (cycle detection)
_visited_modules.add(module_name)
_visited_modules.add(target_top)
children = []
# Check blocklist and depth limit
should_expand = (
_depth < max_depth
and module_name not in _BLOCKLIST_MODULES
and target_top not in _BLOCKLIST_MODULES
)
if should_expand:
# Find Lib modules that import this module
lib_importers = _get_lib_modules_importing(module_name, lib_import_graph)
# Skip already visited modules (cycle detection) and blocklisted modules
lib_importers = {
m
for m in lib_importers
if m not in _visited_modules
and m.split(".")[0] not in _visited_modules
and m not in _BLOCKLIST_MODULES
and m.split(".")[0] not in _BLOCKLIST_MODULES
}
# Consolidate submodules (xml.dom, xml.sax, xml.etree -> xml if > 3)
consolidated_libs = _consolidate_submodules(lib_importers, threshold=3)
# Build children
for display_name, original_mods in sorted(consolidated_libs.items()):
child = find_dependent_tests_tree(
display_name,
lib_prefix,
max_depth,
_depth + 1,
_visited_tests,
_visited_modules,
)
if child["tests"] or child["children"]:
children.append(child)
return {
"module": module_name,
"tests": sorted(consolidated_tests),
"children": children,
}
def _consolidate_file_key(file_key: str) -> str:
"""Consolidate file_key to test name.
Args:
file_key: Relative path without .py (e.g., "test_foo", "test_bar/test_sub")
Returns:
Consolidated test name:
- "test_foo" for "test_foo"
- "test_sqlite3" for "test_sqlite3/test_dbapi"
"""
parts = pathlib.Path(file_key).parts
if len(parts) == 1:
return parts[0]
return parts[0]
View Git Blame Copy Permalink