mirrors/RustPython
0
0 Fork 0
mirror of https://github.com/RustPython/RustPython.git synced 2026-10-28 10:44:36 +08:00
Code Activity

Remake update_lib

Browse Source
This commit is contained in:
Jeong, YunWon authored and Jeong, YunWon committed 2026-01-20 11:38:26 +09:00
1 parent 6f41a9491b
commit d119c47d2e
17 files changed
+3230 -581
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

.claude/commands/upgrade-pylib.md
+35 -14
View File
@@ -5,6 +5,22 @@ Upgrade a Python standard library module from CPython to RustPython.
## Arguments
- `$ARGUMENTS`: Library name to upgrade (e.g., `inspect`, `asyncio`, `json`)
## Important: Report Tool Issues First
If during the upgrade process you encounter any of the following issues with `scripts/update_lib`:
- A feature that should be automated but isn't supported
- A bug or unexpected behavior in the tool
- Missing functionality that would make the upgrade easier
**STOP the upgrade and report the issue first.** Describe:
1. What you were trying to do
- Library name
- The full command executed (e.g. python scripts/update_lib quick cpython/Lib/$ARGUMENTS.py)
2. What went wrong or what's missing
3. Expected vs actual behavior
This helps improve the tooling for future upgrades.
## Steps
1. **Delete existing library in Lib/**
@@ -15,18 +31,14 @@ Upgrade a Python standard library module from CPython to RustPython.
- If `cpython/Lib/$ARGUMENTS.py` exists, copy it to `Lib/$ARGUMENTS.py`
- If `cpython/Lib/$ARGUMENTS/` directory exists, copy it to `Lib/$ARGUMENTS/`
3. **Upgrade tests (quick upgrade with lib_updater)**
- If `cpython/Lib/test/test_$ARGUMENTS.py` is a single file:
- Run: `python3 scripts/lib_updater.py --quick-upgrade cpython/Lib/test/test_$ARGUMENTS.py`
- If `cpython/Lib/test/test_$ARGUMENTS/` is a directory:
- Run the script for each `.py` file in the directory:
```bash
for f in cpython/Lib/test/test_$ARGUMENTS/*.py; do
python3 scripts/lib_updater.py --quick-upgrade "$f"
done
```
- This will update the test files with basic RustPython markers (`@unittest.expectedFailure`, `@unittest.skip`, etc.)
- **Handle lib_updater warnings**: If you see warnings like `WARNING: TestCFoo does not exist in remote file`, it means the class structure changed between versions and markers couldn't be transferred automatically. These need to be manually restored in step 4 or added in step 5.
3. **Upgrade tests (quick upgrade with update_lib)**
- Run: `python3 scripts/update_lib quick cpython/Lib/test/test_$ARGUMENTS.py` (single file)
- Or: `python3 scripts/update_lib quick cpython/Lib/test/test_$ARGUMENTS/` (directory)
- This will:
- Patch test files preserving existing RustPython markers
- Run tests and auto-mark new test failures (not regressions)
- Remove `@unittest.expectedFailure` from tests that now pass
- **Handle warnings**: If you see warnings like `WARNING: TestCFoo does not exist in remote file`, it means the class structure changed and markers couldn't be transferred automatically. These need to be manually restored in step 4 or added in step 5.
4. **Review git diff and restore RUSTPYTHON-specific changes**
- Run `git diff Lib/test/test_$ARGUMENTS` to review all changes
@@ -96,10 +108,19 @@ cjson = import_helper.import_fresh_module('json') #, fresh=['_json'])
## Notes
- The cpython/ directory should contain the CPython source that we're syncing from
- `scripts/lib_updater.py` handles basic patching:
- `scripts/update_lib` package handles patching and auto-marking:
- `quick` - Combined patch + auto-mark (recommended)
- `migrate` - Only migrate (patch), no test running
- `auto-mark` - Only run tests and mark failures
- `copy-lib` - Copy library files (not tests)
- The patching:
- Transfers `@unittest.expectedFailure` and `@unittest.skip` decorators with `TODO: RUSTPYTHON` markers
- Adds `import unittest # XXX: RUSTPYTHON` if needed for the decorators
- **Limitation**: If a class was restructured (e.g., method overrides removed), lib_updater will warn and skip those markers
- **Limitation**: If a class was restructured (e.g., method overrides removed), update_lib will warn and skip those markers
- The smart auto-mark:
- Marks NEW test failures automatically (tests that didn't exist before)
- Does NOT mark regressions (existing tests that now fail) - these are warnings
- Removes `@unittest.expectedFailure` from tests that now pass
- The script does NOT preserve all RustPython-specific changes - you must review `git diff` and restore them
- Common RustPython markers to look for:
- `# XXX: RUSTPYTHON` or `# XXX RUSTPYTHON` - Inline comments for code modifications
scripts/auto_mark_test.py
-411
View File
@@ -1,411 +0,0 @@
"""
An automated script to mark failures in python test suite.
It adds @unittest.expectedFailure to the test functions that are failing in RustPython, but not in CPython.
As well as marking the test with a TODO comment.
Quick Import (recommended):
python ./scripts/fix_test.py --quick-import cpython/Lib/test/test_foo.py
This will:
1. Copy cpython/Lib/test/test_foo.py to Lib/test/test_foo.py (if not exists)
2. Run the test with RustPython
3. Mark failing tests with @unittest.expectedFailure
Manual workflow:
1. Copy a specific test from the CPython repository to the RustPython repository.
2. Remove all unexpected failures from the test and skip the tests that hang.
3. Build RustPython: cargo build --release
4. Run from the project root:
- For single-file tests: python ./scripts/fix_test.py --path ./Lib/test/test_venv.py
- For package tests: python ./scripts/fix_test.py --path ./Lib/test/test_inspect/test_inspect.py
5. Verify: cargo run --release -- -m test test_venv (should pass with expected failures)
6. Actually fix the tests marked with # TODO: RUSTPYTHON
"""
import argparse
import ast
import re
import shutil
import sys
from pathlib import Path
from lib_updater import (
COMMENT,
PatchSpec,
UtMethod,
apply_patches,
)
def parse_args():
parser = argparse.ArgumentParser(description="Fix test.")
group = parser.add_mutually_exclusive_group(required=True)
group.add_argument("--path", type=Path, help="Path to test file")
group.add_argument(
"--quick-import",
type=Path,
metavar="PATH",
help="Import from path containing /Lib/ (e.g., cpython/Lib/test/foo.py)",
)
parser.add_argument("--force", action="store_true", help="Force modification")
parser.add_argument(
"--platform", action="store_true", help="Platform specific failure"
)
args = parser.parse_args()
return args
class Test:
name: str = ""
path: str = ""
result: str = ""
def __str__(self):
return f"Test(name={self.name}, path={self.path}, result={self.result})"
class TestResult:
tests_result: str = ""
tests = []
unexpected_successes = [] # Tests that passed but were marked as expectedFailure
stdout = ""
def __str__(self):
return f"TestResult(tests_result={self.tests_result},tests={len(self.tests)},unexpected_successes={len(self.unexpected_successes)})"
def parse_results(result):
lines = result.stdout.splitlines()
test_results = TestResult()
test_results.tests = []
test_results.unexpected_successes = []
test_results.stdout = result.stdout
in_test_results = False
for line in lines:
if re.search(r"Run \d+ tests? sequentially", line):
in_test_results = True
elif line.startswith("-----------"):
in_test_results = False
if in_test_results and " ... " in line:
line = line.strip()
# Skip lines that don't look like test results
if line.startswith("tests") or line.startswith("["):
continue
# Parse: "test_name (path) [subtest] ... RESULT"
parts = line.split(" ... ")
if len(parts) >= 2:
test_info = parts[0]
result_str = parts[-1].lower()
# Only process FAIL or ERROR
if result_str not in ("fail", "error"):
continue
# Extract test name (first word)
first_space = test_info.find(" ")
if first_space > 0:
test = Test()
test.name = test_info[:first_space]
# Extract path from (path)
rest = test_info[first_space:].strip()
if rest.startswith("("):
end_paren = rest.find(")")
if end_paren > 0:
test.path = rest[1:end_paren]
test.result = result_str
test_results.tests.append(test)
elif "== Tests result: " in line:
res = line.split("== Tests result: ")[1]
res = res.split(" ")[0]
test_results.tests_result = res
# Parse: "UNEXPECTED SUCCESS: test_name (path)"
elif line.startswith("UNEXPECTED SUCCESS: "):
rest = line[len("UNEXPECTED SUCCESS: ") :]
# Format: "test_name (path)"
first_space = rest.find(" ")
if first_space > 0:
test = Test()
test.name = rest[:first_space]
path_part = rest[first_space:].strip()
if path_part.startswith("(") and path_part.endswith(")"):
test.path = path_part[1:-1]
test.result = "unexpected_success"
test_results.unexpected_successes.append(test)
return test_results
def path_to_test(path) -> list[str]:
# path format: test.module_name[.submodule].ClassName.test_method
# We need [ClassName, test_method] - always the last 2 elements
parts = path.split(".")
return parts[-2:] # Get class name and method name
def is_super_call_only(func_node: ast.FunctionDef | ast.AsyncFunctionDef) -> bool:
"""Check if the method body is just 'return super().method_name()'."""
if len(func_node.body) != 1:
return False
stmt = func_node.body[0]
if not isinstance(stmt, ast.Return) or stmt.value is None:
return False
# Check for super().method_name() pattern
call = stmt.value
if not isinstance(call, ast.Call):
return False
if not isinstance(call.func, ast.Attribute):
return False
super_call = call.func.value
if not isinstance(super_call, ast.Call):
return False
if not isinstance(super_call.func, ast.Name) or super_call.func.id != "super":
return False
return True
def build_inheritance_info(tree: ast.Module) -> tuple[dict, dict]:
"""
Build inheritance information from AST.
Returns:
class_bases: dict[str, list[str]] - parent classes for each class (only those defined in the file)
class_methods: dict[str, set[str]] - methods directly defined in each class
"""
all_classes = {
node.name for node in ast.walk(tree) if isinstance(node, ast.ClassDef)
}
class_bases = {}
class_methods = {}
for node in ast.walk(tree):
if isinstance(node, ast.ClassDef):
# Collect only parent classes defined in this file
bases = [
base.id
for base in node.bases
if isinstance(base, ast.Name) and base.id in all_classes
]
class_bases[node.name] = bases
# Collect directly defined methods
methods = {
item.name
for item in node.body
if isinstance(item, (ast.FunctionDef, ast.AsyncFunctionDef))
}
class_methods[node.name] = methods
return class_bases, class_methods
def find_method_definition(
class_name: str, method_name: str, class_bases: dict, class_methods: dict
) -> str | None:
"""Find the class where a method is actually defined. Traverses inheritance chain (BFS)."""
# Check current class first
if method_name in class_methods.get(class_name, set()):
return class_name
# Search parent classes
visited = set()
queue = list(class_bases.get(class_name, []))
while queue:
current = queue.pop(0)
if current in visited:
continue
visited.add(current)
if method_name in class_methods.get(current, set()):
return current
queue.extend(class_bases.get(current, []))
return None
def remove_expected_failures(
contents: str, tests_to_remove: set[tuple[str, str]]
) -> str:
"""Remove @unittest.expectedFailure decorators from tests that now pass."""
if not tests_to_remove:
return contents
tree = ast.parse(contents)
lines = contents.splitlines()
lines_to_remove = set()
# Build inheritance information
class_bases, class_methods = build_inheritance_info(tree)
# Resolve to actual defining classes
resolved_tests = set()
for class_name, method_name in tests_to_remove:
defining_class = find_method_definition(
class_name, method_name, class_bases, class_methods
)
if defining_class:
resolved_tests.add((defining_class, method_name))
for node in ast.walk(tree):
if not isinstance(node, ast.ClassDef):
continue
class_name = node.name
for item in node.body:
if not isinstance(item, (ast.FunctionDef, ast.AsyncFunctionDef)):
continue
method_name = item.name
if (class_name, method_name) not in resolved_tests:
continue
# Check if we should remove the entire method (super() call only)
remove_entire_method = is_super_call_only(item)
if remove_entire_method:
# Remove entire method including decorators and any preceding comment
first_line = item.lineno - 1 # 0-indexed, def line
if item.decorator_list:
first_line = item.decorator_list[0].lineno - 1
# Check for TODO comment before first decorator/def
if first_line > 0:
prev_line = lines[first_line - 1].strip()
if prev_line.startswith("#") and COMMENT in prev_line:
first_line -= 1
# Remove from first_line to end_lineno (inclusive)
for i in range(first_line, item.end_lineno):
lines_to_remove.add(i)
else:
# Only remove the expectedFailure decorator
for dec in item.decorator_list:
dec_line = dec.lineno - 1 # 0-indexed
line_content = lines[dec_line]
# Check if it's @unittest.expectedFailure
if "expectedFailure" not in line_content:
continue
# Check if TODO: RUSTPYTHON is on the same line or the line before
has_comment_on_line = COMMENT in line_content
has_comment_before = (
dec_line > 0
and lines[dec_line - 1].strip().startswith("#")
and COMMENT in lines[dec_line - 1]
)
if has_comment_on_line or has_comment_before:
lines_to_remove.add(dec_line)
if has_comment_before:
lines_to_remove.add(dec_line - 1)
# Remove lines in reverse order to maintain line numbers
for line_idx in sorted(lines_to_remove, reverse=True):
del lines[line_idx]
return "\n".join(lines) + "\n" if lines else ""
def build_patches(test_parts_set: set[tuple[str, str]]) -> dict:
"""Convert failing tests to lib_updater patch format."""
patches = {}
for class_name, method_name in test_parts_set:
if class_name not in patches:
patches[class_name] = {}
patches[class_name][method_name] = [
PatchSpec(UtMethod.ExpectedFailure, None, "")
]
return patches
def run_test(test_name):
print(f"Running test: {test_name}")
rustpython_location = "./target/release/rustpython"
if sys.platform == "win32":
rustpython_location += ".exe"
import subprocess
result = subprocess.run(
[rustpython_location, "-m", "test", "-v", "-u", "all", "--slowest", test_name],
capture_output=True,
text=True,
)
return parse_results(result)
if __name__ == "__main__":
args = parse_args()
# Handle --quick-import: extract Lib/... path and copy if needed
if args.quick_import is not None:
# Normalize path separators to forward slashes for cross-platform support
src_str = str(args.quick_import).replace("\\", "/")
lib_marker = "/Lib/"
if lib_marker not in src_str:
print(
f"Error: --quick-import path must contain '/Lib/' or '\\Lib\\' (got: {args.quick_import})"
)
sys.exit(1)
idx = src_str.index(lib_marker)
lib_path = Path(src_str[idx + 1 :]) # Lib/test/foo.py
src_path = args.quick_import
if not src_path.exists():
print(f"Error: Source file not found: {src_path}")
sys.exit(1)
if not lib_path.exists():
print(f"Copying: {src_path} -> {lib_path}")
lib_path.parent.mkdir(parents=True, exist_ok=True)
shutil.copy(src_path, lib_path)
else:
print(f"File already exists: {lib_path}")
args.path = lib_path
test_path = args.path.resolve()
if not test_path.exists():
print(f"Error: File not found: {test_path}")
sys.exit(1)
# Detect package tests (e.g., test_ctypes/test_random_things.py)
if test_path.parent.name.startswith("test_"):
test_name = f"{test_path.parent.name}.{test_path.stem}"
else:
test_name = test_path.stem
tests = run_test(test_name)
f = test_path.read_text(encoding="utf-8")
# Collect failing tests (with deduplication for subtests)
failing_tests = set() # Track (class_name, method_name) to avoid duplicates
for test in tests.tests:
if test.result == "fail" or test.result == "error":
test_parts = path_to_test(test.path)
if len(test_parts) == 2:
test_key = tuple(test_parts)
if test_key not in failing_tests:
failing_tests.add(test_key)
print(f"Marking as failing: {test_parts[0]}.{test_parts[1]}")
# Collect unexpected successes (tests that now pass but have expectedFailure)
unexpected_successes = set()
for test in tests.unexpected_successes:
test_parts = path_to_test(test.path)
if len(test_parts) == 2:
test_key = tuple(test_parts)
if test_key not in unexpected_successes:
unexpected_successes.add(test_key)
print(f"Removing expectedFailure: {test_parts[0]}.{test_parts[1]}")
# Remove expectedFailure from tests that now pass
if unexpected_successes:
f = remove_expected_failures(f, unexpected_successes)
# Apply patches for failing tests
if failing_tests:
patches = build_patches(failing_tests)
f = apply_patches(f, patches)
# Write changes if any modifications were made
if failing_tests or unexpected_successes:
test_path.write_text(f, encoding="utf-8")
print(f"Added expectedFailure to {len(failing_tests)} tests")
print(f"Removed expectedFailure from {len(unexpected_successes)} tests")
scripts/update_lib/__init__.py
+37
View File
@@ -0,0 +1,37 @@
"""
Library for updating Python test files with RustPython-specific patches.
"""
from .patch_spec import (
COMMENT,
DEFAULT_INDENT,
UT,
PatchEntry,
Patches,
PatchSpec,
UtMethod,
apply_patches,
build_patch_dict,
extract_patches,
iter_patches,
iter_tests,
patches_from_json,
patches_to_json,
)
__all__ = [
"COMMENT",
"DEFAULT_INDENT",
"UT",
"Patches",
"PatchEntry",
"PatchSpec",
"UtMethod",
"apply_patches",
"build_patch_dict",
"extract_patches",
"iter_patches",
"iter_tests",
"patches_from_json",
"patches_to_json",
]
scripts/update_lib/__main__.py
+84
View File
@@ -0,0 +1,84 @@
#!/usr/bin/env python
"""
Update library tools for RustPython.
Usage:
python scripts/update_lib quick cpython/Lib/test/test_foo.py
python scripts/update_lib copy-lib cpython/Lib/dataclasses.py
python scripts/update_lib migrate cpython/Lib/test/test_foo.py
python scripts/update_lib patches --from Lib/test/foo.py --to cpython/Lib/test/foo.py
python scripts/update_lib auto-mark Lib/test/test_foo.py
"""
import argparse
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent.parent))
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(
description="Update library tools for RustPython",
)
subparsers = parser.add_subparsers(dest="command", required=True)
subparsers.add_parser(
"quick",
help="Quick update: patch + auto-mark (recommended)",
add_help=False,
)
subparsers.add_parser(
"migrate",
help="Migrate test file(s) from CPython, preserving RustPython markers",
add_help=False,
)
subparsers.add_parser(
"patches",
help="Patch management (extract/apply patches between files)",
add_help=False,
)
subparsers.add_parser(
"auto-mark",
help="Run tests and auto-mark failures with @expectedFailure",
add_help=False,
)
subparsers.add_parser(
"copy-lib",
help="Copy library file/directory from CPython (delete existing first)",
add_help=False,
)
args, remaining = parser.parse_known_args(argv)
if args.command == "quick":
from update_lib.quick import main as quick_main
return quick_main(remaining)
if args.command == "copy-lib":
from update_lib.copy_lib import main as copy_lib_main
return copy_lib_main(remaining)
if args.command == "migrate":
from update_lib.migrate import main as migrate_main
return migrate_main(remaining)
if args.command == "patches":
from update_lib.patches import main as patches_main
return patches_main(remaining)
if args.command == "auto-mark":
from update_lib.auto_mark import main as auto_mark_main
return auto_mark_main(remaining)
return 0
if __name__ == "__main__":
sys.exit(main())
scripts/update_lib/auto_mark.py
+704
View File
@@ -0,0 +1,704 @@
#!/usr/bin/env python
"""
Auto-mark test failures in Python test suite.
This module provides functions to:
- Run tests with RustPython and parse results
- Extract test names from test file paths
- Mark failing tests with @unittest.expectedFailure
- Remove expectedFailure from tests that now pass
"""
import ast
import pathlib
import re
import subprocess
import sys
from dataclasses import dataclass, field
sys.path.insert(0, str(pathlib.Path(__file__).parent.parent))
from update_lib import COMMENT, PatchSpec, UtMethod, apply_patches
from update_lib.path import test_name_from_path
@dataclass
class Test:
name: str = ""
path: str = ""
result: str = ""
error_message: str = ""
@dataclass
class TestResult:
tests_result: str = ""
tests: list[Test] = field(default_factory=list)
unexpected_successes: list[Test] = field(default_factory=list)
stdout: str = ""
def run_test(test_name: str, skip_build: bool = False) -> TestResult:
"""
Run a test with RustPython and return parsed results.
Args:
test_name: Test module name (e.g., "test_foo" or "test_ctypes.test_bar")
skip_build: If True, use pre-built binary instead of cargo run
Returns:
TestResult with parsed test results
"""
if skip_build:
cmd = ["./target/release/rustpython"]
if sys.platform == "win32":
cmd = ["./target/release/rustpython.exe"]
else:
cmd = ["cargo", "run", "--release", "--"]
result = subprocess.run(
cmd + ["-m", "test", "-v", "-u", "all", "--slowest", test_name],
capture_output=True,
text=True,
)
return parse_results(result)
def parse_results(result: subprocess.CompletedProcess) -> TestResult:
"""Parse subprocess result into TestResult."""
lines = result.stdout.splitlines()
test_results = TestResult()
test_results.stdout = result.stdout
in_test_results = False
for line in lines:
if re.search(r"Run \d+ tests? sequentially", line):
in_test_results = True
elif line.startswith("-----------"):
in_test_results = False
if in_test_results and " ... " in line:
line = line.strip()
# Skip lines that don't look like test results
if line.startswith("tests") or line.startswith("["):
continue
# Parse: "test_name (path) [subtest] ... RESULT"
parts = line.split(" ... ")
if len(parts) >= 2:
test_info = parts[0]
result_str = parts[-1].lower()
# Only process FAIL or ERROR
if result_str not in ("fail", "error"):
continue
# Extract test name (first word)
first_space = test_info.find(" ")
if first_space > 0:
test = Test()
test.name = test_info[:first_space]
# Extract path from (path)
rest = test_info[first_space:].strip()
if rest.startswith("("):
end_paren = rest.find(")")
if end_paren > 0:
test.path = rest[1:end_paren]
test.result = result_str
test_results.tests.append(test)
elif "== Tests result: " in line:
res = line.split("== Tests result: ")[1]
res = res.split(" ")[0]
test_results.tests_result = res
# Parse: "UNEXPECTED SUCCESS: test_name (path)"
elif line.startswith("UNEXPECTED SUCCESS: "):
rest = line[len("UNEXPECTED SUCCESS: ") :]
# Format: "test_name (path)"
first_space = rest.find(" ")
if first_space > 0:
test = Test()
test.name = rest[:first_space]
path_part = rest[first_space:].strip()
if path_part.startswith("(") and path_part.endswith(")"):
test.path = path_part[1:-1]
test.result = "unexpected_success"
test_results.unexpected_successes.append(test)
# Parse error details to extract error messages
_parse_error_details(test_results, lines)
return test_results
def _parse_error_details(test_results: TestResult, lines: list[str]) -> None:
"""Parse error details section to extract error messages for each test."""
# Build a lookup dict for tests by (name, path)
test_lookup: dict[tuple[str, str], Test] = {}
for test in test_results.tests:
test_lookup[(test.name, test.path)] = test
# Parse error detail blocks
# Format:
# ======================================================================
# FAIL: test_name (path)
# ----------------------------------------------------------------------
# Traceback (most recent call last):
# ...
# AssertionError: message
#
# ======================================================================
i = 0
while i < len(lines):
line = lines[i]
# Look for FAIL: or ERROR: header
if line.startswith(("FAIL: ", "ERROR: ")):
# Parse: "FAIL: test_name (path)" or "ERROR: test_name (path)"
header = line.split(": ", 1)[1] if ": " in line else ""
first_space = header.find(" ")
if first_space > 0:
test_name = header[:first_space]
path_part = header[first_space:].strip()
if path_part.startswith("(") and path_part.endswith(")"):
test_path = path_part[1:-1]
# Find the last non-empty line before the next separator or end
error_lines = []
i += 1
# Skip the separator line
if i < len(lines) and lines[i].startswith("-----"):
i += 1
# Collect lines until the next separator or end
while i < len(lines):
current = lines[i]
if current.startswith("====="):
break
error_lines.append(current)
i += 1
# Find the last non-empty line (the error message)
error_message = ""
for err_line in reversed(error_lines):
stripped = err_line.strip()
if stripped:
error_message = stripped
break
# Update the test with the error message
if (test_name, test_path) in test_lookup:
test_lookup[
(test_name, test_path)
].error_message = error_message
continue
i += 1
def path_to_test_parts(path: str) -> list[str]:
"""
Extract [ClassName, method_name] from test path.
Args:
path: Test path like "test.module_name.ClassName.test_method"
Returns:
[ClassName, method_name] - last 2 elements
"""
parts = path.split(".")
return parts[-2:]
def build_patches(
test_parts_set: set[tuple[str, str]],
error_messages: dict[tuple[str, str], str] | None = None,
) -> dict:
"""Convert failing tests to patch format."""
patches = {}
error_messages = error_messages or {}
for class_name, method_name in test_parts_set:
if class_name not in patches:
patches[class_name] = {}
reason = error_messages.get((class_name, method_name), "")
patches[class_name][method_name] = [
PatchSpec(UtMethod.ExpectedFailure, None, reason)
]
return patches
def _is_super_call_only(func_node: ast.FunctionDef | ast.AsyncFunctionDef) -> bool:
"""Check if the method body is just 'return super().method_name()'."""
if len(func_node.body) != 1:
return False
stmt = func_node.body[0]
if not isinstance(stmt, ast.Return) or stmt.value is None:
return False
call = stmt.value
if not isinstance(call, ast.Call):
return False
if not isinstance(call.func, ast.Attribute):
return False
super_call = call.func.value
if not isinstance(super_call, ast.Call):
return False
if not isinstance(super_call.func, ast.Name) or super_call.func.id != "super":
return False
return True
def _build_inheritance_info(tree: ast.Module) -> tuple[dict, dict]:
"""
Build inheritance information from AST.
Returns:
class_bases: dict[str, list[str]] - parent classes for each class
class_methods: dict[str, set[str]] - methods directly defined in each class
"""
all_classes = {
node.name for node in ast.walk(tree) if isinstance(node, ast.ClassDef)
}
class_bases = {}
class_methods = {}
for node in ast.walk(tree):
if isinstance(node, ast.ClassDef):
bases = [
base.id
for base in node.bases
if isinstance(base, ast.Name) and base.id in all_classes
]
class_bases[node.name] = bases
methods = {
item.name
for item in node.body
if isinstance(item, (ast.FunctionDef, ast.AsyncFunctionDef))
}
class_methods[node.name] = methods
return class_bases, class_methods
def _find_method_definition(
class_name: str, method_name: str, class_bases: dict, class_methods: dict
) -> str | None:
"""Find the class where a method is actually defined (BFS)."""
if method_name in class_methods.get(class_name, set()):
return class_name
visited = set()
queue = list(class_bases.get(class_name, []))
while queue:
current = queue.pop(0)
if current in visited:
continue
visited.add(current)
if method_name in class_methods.get(current, set()):
return current
queue.extend(class_bases.get(current, []))
return None
def remove_expected_failures(
contents: str, tests_to_remove: set[tuple[str, str]]
) -> str:
"""Remove @unittest.expectedFailure decorators from tests that now pass."""
if not tests_to_remove:
return contents
tree = ast.parse(contents)
lines = contents.splitlines()
lines_to_remove = set()
class_bases, class_methods = _build_inheritance_info(tree)
resolved_tests = set()
for class_name, method_name in tests_to_remove:
defining_class = _find_method_definition(
class_name, method_name, class_bases, class_methods
)
if defining_class:
resolved_tests.add((defining_class, method_name))
for node in ast.walk(tree):
if not isinstance(node, ast.ClassDef):
continue
class_name = node.name
for item in node.body:
if not isinstance(item, (ast.FunctionDef, ast.AsyncFunctionDef)):
continue
method_name = item.name
if (class_name, method_name) not in resolved_tests:
continue
remove_entire_method = _is_super_call_only(item)
if remove_entire_method:
first_line = item.lineno - 1
if item.decorator_list:
first_line = item.decorator_list[0].lineno - 1
if first_line > 0:
prev_line = lines[first_line - 1].strip()
if prev_line.startswith("#") and COMMENT in prev_line:
first_line -= 1
for i in range(first_line, item.end_lineno):
lines_to_remove.add(i)
else:
for dec in item.decorator_list:
dec_line = dec.lineno - 1
line_content = lines[dec_line]
if "expectedFailure" not in line_content:
continue
has_comment_on_line = COMMENT in line_content
has_comment_before = (
dec_line > 0
and lines[dec_line - 1].strip().startswith("#")
and COMMENT in lines[dec_line - 1]
)
if has_comment_on_line or has_comment_before:
lines_to_remove.add(dec_line)
if has_comment_before:
lines_to_remove.add(dec_line - 1)
for line_idx in sorted(lines_to_remove, reverse=True):
del lines[line_idx]
return "\n".join(lines) + "\n" if lines else ""
def collect_test_changes(
results: TestResult,
module_prefix: str | None = None,
) -> tuple[set[tuple[str, str]], set[tuple[str, str]], dict[tuple[str, str], str]]:
"""
Collect failing tests and unexpected successes from test results.
Args:
results: TestResult from run_test()
module_prefix: If set, only collect tests whose path starts with this prefix
Returns:
(failing_tests, unexpected_successes, error_messages)
- failing_tests: set of (class_name, method_name) tuples
- unexpected_successes: set of (class_name, method_name) tuples
- error_messages: dict mapping (class_name, method_name) to error message
"""
failing_tests = set()
error_messages: dict[tuple[str, str], str] = {}
for test in results.tests:
if test.result in ("fail", "error"):
if module_prefix and not test.path.startswith(module_prefix):
continue
test_parts = path_to_test_parts(test.path)
if len(test_parts) == 2:
key = tuple(test_parts)
failing_tests.add(key)
if test.error_message:
error_messages[key] = test.error_message
unexpected_successes = set()
for test in results.unexpected_successes:
if module_prefix and not test.path.startswith(module_prefix):
continue
test_parts = path_to_test_parts(test.path)
if len(test_parts) == 2:
unexpected_successes.add(tuple(test_parts))
return failing_tests, unexpected_successes, error_messages
def apply_test_changes(
contents: str,
failing_tests: set[tuple[str, str]],
unexpected_successes: set[tuple[str, str]],
error_messages: dict[tuple[str, str], str] | None = None,
) -> str:
"""
Apply test changes to content.
Args:
contents: File content
failing_tests: Set of (class_name, method_name) to mark as expectedFailure
unexpected_successes: Set of (class_name, method_name) to remove expectedFailure
error_messages: Dict mapping (class_name, method_name) to error message
Returns:
Modified content
"""
if unexpected_successes:
contents = remove_expected_failures(contents, unexpected_successes)
if failing_tests:
patches = build_patches(failing_tests, error_messages)
contents = apply_patches(contents, patches)
return contents
def extract_test_methods(contents: str) -> set[tuple[str, str]]:
"""
Extract all test method names from file contents.
Returns:
Set of (class_name, method_name) tuples
"""
import ast
try:
tree = ast.parse(contents)
except SyntaxError:
return set()
from update_lib.patch_spec import iter_tests
return {(cls_node.name, fn_node.name) for cls_node, fn_node in iter_tests(tree)}
def auto_mark_file(
test_path: pathlib.Path,
mark_failure: bool = False,
verbose: bool = True,
original_methods: set[tuple[str, str]] | None = None,
skip_build: bool = False,
) -> tuple[int, int, int]:
"""
Run tests and auto-mark failures in a test file.
Args:
test_path: Path to the test file
mark_failure: If True, add @expectedFailure to ALL failing tests
verbose: Print progress messages
original_methods: If provided, only auto-mark failures for NEW methods
(methods not in original_methods) even without mark_failure.
Failures in existing methods are treated as regressions.
Returns:
(num_failures_added, num_successes_removed, num_regressions)
"""
test_path = pathlib.Path(test_path).resolve()
if not test_path.exists():
raise FileNotFoundError(f"File not found: {test_path}")
test_name = test_name_from_path(test_path)
if verbose:
print(f"Running test: {test_name}")
results = run_test(test_name, skip_build=skip_build)
contents = test_path.read_text(encoding="utf-8")
all_failing_tests, unexpected_successes, error_messages = collect_test_changes(
results
)
# Determine which failures to mark
if mark_failure:
failing_tests = all_failing_tests
elif original_methods is not None:
# Smart mode: only mark NEW test failures (not regressions)
current_methods = extract_test_methods(contents)
new_methods = current_methods - original_methods
failing_tests = {t for t in all_failing_tests if t in new_methods}
else:
failing_tests = set()
regressions = all_failing_tests - failing_tests
if verbose:
for class_name, method_name in failing_tests:
label = "(new test)" if original_methods is not None else ""
err_msg = error_messages.get((class_name, method_name), "")
err_hint = f" - {err_msg}" if err_msg else ""
print(
f"Marking as failing {label}: {class_name}.{method_name}{err_hint}".replace(
" ", " "
)
)
for class_name, method_name in unexpected_successes:
print(f"Removing expectedFailure: {class_name}.{method_name}")
contents = apply_test_changes(
contents, failing_tests, unexpected_successes, error_messages
)
if failing_tests or unexpected_successes:
test_path.write_text(contents, encoding="utf-8")
# Show hints about unmarked failures
if verbose:
unmarked_failures = all_failing_tests - failing_tests
if unmarked_failures:
print(
f"Hint: {len(unmarked_failures)} failing tests can be marked with --mark-failure; "
"but review first and do not blindly mark them all"
)
for class_name, method_name in sorted(unmarked_failures):
err_msg = error_messages.get((class_name, method_name), "")
err_hint = f" - {err_msg}" if err_msg else ""
print(f" {class_name}.{method_name}{err_hint}")
return len(failing_tests), len(unexpected_successes), len(regressions)
def auto_mark_directory(
test_dir: pathlib.Path,
mark_failure: bool = False,
verbose: bool = True,
original_methods_per_file: dict[pathlib.Path, set[tuple[str, str]]] | None = None,
skip_build: bool = False,
) -> tuple[int, int, int]:
"""
Run tests and auto-mark failures in a test directory.
Runs the test once for the whole directory, then applies results to each file.
Args:
test_dir: Path to the test directory
mark_failure: If True, add @expectedFailure to ALL failing tests
verbose: Print progress messages
original_methods_per_file: If provided, only auto-mark failures for NEW methods
even without mark_failure. Dict maps file path to
set of (class_name, method_name) tuples.
Returns:
(num_failures_added, num_successes_removed, num_regressions)
"""
test_dir = pathlib.Path(test_dir).resolve()
if not test_dir.exists():
raise FileNotFoundError(f"Directory not found: {test_dir}")
if not test_dir.is_dir():
raise ValueError(f"Not a directory: {test_dir}")
test_name = test_name_from_path(test_dir)
if verbose:
print(f"Running test: {test_name}")
results = run_test(test_name, skip_build=skip_build)
total_added = 0
total_removed = 0
total_regressions = 0
all_regressions: list[tuple[str, str, str, str]] = []
# Get all .py files in directory
test_files = sorted(test_dir.glob("**/*.py"))
for test_file in test_files:
# Get module prefix for this file (e.g., "test_inspect.test_inspect")
module_prefix = test_name_from_path(test_file)
# For __init__.py, the test path doesn't include "__init__"
if module_prefix.endswith(".__init__"):
module_prefix = module_prefix[:-9] # Remove ".__init__"
all_failing_tests, unexpected_successes, error_messages = collect_test_changes(
results, module_prefix="test." + module_prefix + "."
)
# Determine which failures to mark
if mark_failure:
failing_tests = all_failing_tests
elif original_methods_per_file is not None:
# Smart mode: only mark NEW test failures
contents = test_file.read_text(encoding="utf-8")
current_methods = extract_test_methods(contents)
original_methods = original_methods_per_file.get(test_file, set())
new_methods = current_methods - original_methods
failing_tests = {t for t in all_failing_tests if t in new_methods}
else:
failing_tests = set()
regressions = all_failing_tests - failing_tests
if failing_tests or unexpected_successes:
if verbose:
for class_name, method_name in failing_tests:
label = (
"(new test)" if original_methods_per_file is not None else ""
)
err_msg = error_messages.get((class_name, method_name), "")
err_hint = f" - {err_msg}" if err_msg else ""
print(
f" {test_file.name}: Marking as failing {label}: {class_name}.{method_name}{err_hint}".replace(
" :", ":"
)
)
for class_name, method_name in unexpected_successes:
print(
f" {test_file.name}: Removing expectedFailure: {class_name}.{method_name}"
)
contents = test_file.read_text(encoding="utf-8")
contents = apply_test_changes(
contents, failing_tests, unexpected_successes, error_messages
)
test_file.write_text(contents, encoding="utf-8")
# Collect regressions with error messages for later reporting
for class_name, method_name in regressions:
err_msg = error_messages.get((class_name, method_name), "")
all_regressions.append((test_file.name, class_name, method_name, err_msg))
total_added += len(failing_tests)
total_removed += len(unexpected_successes)
total_regressions += len(regressions)
# Show hints about unmarked failures
if verbose and total_regressions > 0:
print(
f"Hint: {total_regressions} failing tests can be marked with --mark-failure; "
"but review first and do not blindly mark them all"
)
for file_name, class_name, method_name, err_msg in sorted(all_regressions):
err_hint = f" - {err_msg}" if err_msg else ""
print(f" {file_name}: {class_name}.{method_name}{err_hint}")
return total_added, total_removed, total_regressions
def main(argv: list[str] | None = None) -> int:
import argparse
parser = argparse.ArgumentParser(
description=__doc__,
formatter_class=argparse.RawDescriptionHelpFormatter,
)
parser.add_argument(
"path",
type=pathlib.Path,
help="Path to test file or directory (e.g., Lib/test/test_foo.py or Lib/test/test_foo/)",
)
parser.add_argument(
"--mark-failure",
action="store_true",
help="Also add @expectedFailure to failing tests (default: only remove unexpected successes)",
)
parser.add_argument(
"--skip-build",
action="store_true",
help="Skip cargo build, use pre-built binary",
)
args = parser.parse_args(argv)
try:
if args.path.is_dir():
num_added, num_removed, _ = auto_mark_directory(
args.path, mark_failure=args.mark_failure, skip_build=args.skip_build
)
else:
num_added, num_removed, _ = auto_mark_file(
args.path, mark_failure=args.mark_failure, skip_build=args.skip_build
)
if args.mark_failure:
print(f"Added expectedFailure to {num_added} tests")
print(f"Removed expectedFailure from {num_removed} tests")
return 0
except (FileNotFoundError, ValueError) as e:
print(f"Error: {e}", file=sys.stderr)
return 1
if __name__ == "__main__":
sys.exit(main())
scripts/update_lib/copy_lib.py
+82
View File
@@ -0,0 +1,82 @@
#!/usr/bin/env python
"""
Copy library files from CPython.
Usage:
# Single file
python scripts/update_lib copy-lib cpython/Lib/dataclasses.py
# Directory
python scripts/update_lib copy-lib cpython/Lib/json
"""
import argparse
import pathlib
import shutil
import sys
def copy_lib(
src_path: pathlib.Path,
verbose: bool = True,
) -> None:
"""
Copy library file or directory from CPython.
Args:
src_path: Source path (e.g., cpython/Lib/dataclasses.py or cpython/Lib/json)
verbose: Print progress messages
"""
from update_lib.path import parse_lib_path
lib_path = parse_lib_path(src_path)
# Remove existing file/directory
if lib_path.exists():
if lib_path.is_dir():
if verbose:
print(f"Removing directory: {lib_path}")
shutil.rmtree(lib_path)
else:
if verbose:
print(f"Removing file: {lib_path}")
lib_path.unlink()
# Copy
if src_path.is_dir():
if verbose:
print(f"Copying directory: {src_path} -> {lib_path}")
shutil.copytree(src_path, lib_path)
else:
if verbose:
print(f"Copying file: {src_path} -> {lib_path}")
lib_path.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(src_path, lib_path)
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(
description=__doc__,
formatter_class=argparse.RawDescriptionHelpFormatter,
)
parser.add_argument(
"path",
type=pathlib.Path,
help="Source path containing /Lib/ (e.g., cpython/Lib/dataclasses.py)",
)
args = parser.parse_args(argv)
try:
copy_lib(args.path)
return 0
except ValueError as e:
print(f"Error: {e}", file=sys.stderr)
return 1
except FileNotFoundError as e:
print(f"Error: {e}", file=sys.stderr)
return 1
if __name__ == "__main__":
sys.exit(main())
scripts/update_lib/migrate.py
+142
View File
@@ -0,0 +1,142 @@
#!/usr/bin/env python
"""
Migrate test file(s) from CPython, preserving RustPython markers.
Usage:
python scripts/update_lib migrate cpython/Lib/test/test_foo.py
This will:
1. Extract patches from Lib/test/test_foo.py (if exists)
2. Apply them to cpython/Lib/test/test_foo.py
3. Write result to Lib/test/test_foo.py
"""
import argparse
import pathlib
import sys
sys.path.insert(0, str(pathlib.Path(__file__).parent.parent))
from update_lib.path import parse_lib_path
def patch_single_content(
src_path: pathlib.Path,
lib_path: pathlib.Path,
) -> str:
"""
Patch content without writing to disk.
Args:
src_path: Source file path (e.g., cpython/Lib/test/foo.py)
lib_path: Lib path to extract patches from (e.g., Lib/test/foo.py)
Returns:
The patched content.
"""
from update_lib import apply_patches, extract_patches
# Extract patches from existing file (if exists)
if lib_path.exists():
patches = extract_patches(lib_path.read_text())
else:
patches = {}
# Apply patches to source content
src_content = src_path.read_text()
return apply_patches(src_content, patches)
def patch_file(
src_path: pathlib.Path,
lib_path: pathlib.Path | None = None,
verbose: bool = True,
) -> None:
"""
Patch a single file from source to lib.
Args:
src_path: Source file path (e.g., cpython/Lib/test/foo.py)
lib_path: Target lib path. If None, derived from src_path.
verbose: Print progress messages
"""
if lib_path is None:
lib_path = parse_lib_path(src_path)
if lib_path.exists():
if verbose:
print(f"Patching: {src_path} -> {lib_path}")
content = patch_single_content(src_path, lib_path)
else:
if verbose:
print(f"Copying: {src_path} -> {lib_path}")
content = src_path.read_text()
lib_path.parent.mkdir(parents=True, exist_ok=True)
lib_path.write_text(content)
def patch_directory(
src_dir: pathlib.Path,
lib_dir: pathlib.Path | None = None,
verbose: bool = True,
) -> None:
"""
Patch all files in a directory from source to lib.
Args:
src_dir: Source directory path (e.g., cpython/Lib/test/test_foo/)
lib_dir: Target lib directory. If None, derived from src_dir.
verbose: Print progress messages
"""
if lib_dir is None:
lib_dir = parse_lib_path(src_dir)
src_files = sorted(src_dir.glob("**/*.py"))
for src_file in src_files:
rel_path = src_file.relative_to(src_dir)
lib_file = lib_dir / rel_path
if lib_file.exists():
if verbose:
print(f"Patching: {src_file} -> {lib_file}")
content = patch_single_content(src_file, lib_file)
else:
if verbose:
print(f"Copying: {src_file} -> {lib_file}")
content = src_file.read_text()
lib_file.parent.mkdir(parents=True, exist_ok=True)
lib_file.write_text(content)
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(
description=__doc__,
formatter_class=argparse.RawDescriptionHelpFormatter,
)
parser.add_argument(
"path",
type=pathlib.Path,
help="Source path containing /Lib/ (file or directory)",
)
args = parser.parse_args(argv)
try:
if args.path.is_dir():
patch_directory(args.path)
else:
patch_file(args.path)
return 0
except ValueError as e:
print(f"Error: {e}", file=sys.stderr)
return 1
except FileNotFoundError as e:
print(f"Error: {e}", file=sys.stderr)
return 1
if __name__ == "__main__":
sys.exit(main())
scripts/lib_updater.py → scripts/update_lib/patch_spec.py
Executable → Regular
+81 -156
View File
@@ -1,47 +1,14 @@
#!/usr/bin/env python
__doc__ = """
This tool helps with updating test files from CPython.
"""
Low-level module for converting between test files and JSON patches.
Quick Upgrade
-------------
./{fname} --quick-upgrade cpython/Lib/test/test_threading.py
./{fname} --quick-upgrade ../somewhere/Lib/threading.py
This module handles:
- Extracting patches from test files (file -> JSON)
- Applying patches to test files (JSON -> file)
"""
Any path containing `/Lib/` will auto-detect the target:
-> Extracts patches from Lib/... (auto-detected from path)
-> Applies them to the source file
-> Writes result to Lib/...
Examples
--------
To move the patches found in `Lib/test/foo.py` to `~/cpython/Lib/test/foo.py` then write the contents back to `Lib/test/foo.py`
>>> ./{fname} --from Lib/test/foo.py --to ~/cpython/Lib/test/foo.py -o Lib/test/foo.py
You can run the same command without `-o` to override the `--from` path:
>>> ./{fname} --from Lib/test/foo.py --to ~/cpython/Lib/test/foo.py
To get a baseline of patches, you can alter the patches file with your favorite tool/script/etc and then reapply it with:
>>> ./{fname} --from Lib/test/foo.py --show-patches -o my_patches.json
(By default the output is set to print to stdout).
When you want to apply your own patches:
>>> ./{fname} -p my_patches.json --to Lib/test/foo.py
""".format(fname=__import__("os").path.basename(__file__))
import argparse
import ast
import collections
import enum
import json
import pathlib
import re
import sys
import textwrap
import typing
@@ -115,13 +82,43 @@ class PatchSpec(typing.NamedTuple):
def as_decorator(self) -> str:
unparsed = ast.unparse(self.as_ast_node())
# ast.unparse uses single quotes; convert to double quotes for ruff compatibility
unparsed = _single_to_double_quotes(unparsed)
if not self.ut_method.has_args():
unparsed = f"{unparsed} # {self._reason}"
unparsed = f"{unparsed} # {self._reason}"
return f"@{unparsed}"
def _single_to_double_quotes(s: str) -> str:
"""Convert single-quoted strings to double-quoted strings.
Falls back to original if conversion breaks the AST equivalence.
"""
import re
def replace_string(match: re.Match) -> str:
content = match.group(1)
# Unescape single quotes and escape double quotes
content = content.replace("\\'", "'").replace('"', '\\"')
return f'"{content}"'
# Match single-quoted strings (handles escaped single quotes inside)
converted = re.sub(r"'((?:[^'\\]|\\.)*)'", replace_string, s)
# Verify: parse converted and unparse should equal original
try:
converted_ast = ast.parse(converted, mode="eval")
if ast.unparse(converted_ast) == s:
return converted
except SyntaxError:
pass
# Fall back to original if conversion failed
return s
class PatchEntry(typing.NamedTuple):
"""
Stores patch metadata.
@@ -141,9 +138,12 @@ class PatchEntry(typing.NamedTuple):
spec: PatchSpec
@classmethod
def iter_patch_entires(
def iter_patch_entries(
cls, tree: ast.Module, lines: list[str]
) -> "Iterator[typing.Self]":
import re
import sys
for cls_node, fn_node in iter_tests(tree):
parent_class = cls_node.name
for dec_node in fn_node.decorator_list:
@@ -224,7 +224,7 @@ def iter_tests(
def iter_patches(contents: str) -> "Iterator[PatchEntry]":
lines = contents.splitlines()
tree = ast.parse(contents)
yield from PatchEntry.iter_patch_entires(tree, lines)
yield from PatchEntry.iter_patch_entries(tree, lines)
def build_patch_dict(it: "Iterator[PatchEntry]") -> Patches:
@@ -235,7 +235,16 @@ def build_patch_dict(it: "Iterator[PatchEntry]") -> Patches:
return {k: dict(v) for k, v in patches.items()}
def iter_patch_lines(tree: ast.Module, patches: Patches) -> "Iterator[tuple[int, str]]":
def extract_patches(contents: str) -> Patches:
"""Extract patches from file contents and return as dict."""
return build_patch_dict(iter_patches(contents))
def _iter_patch_lines(
tree: ast.Module, patches: Patches
) -> "Iterator[tuple[int, str]]":
import sys
# Build cache of all classes (for Phase 2 to find classes without methods)
cache = {}
for node in tree.body:
@@ -256,7 +265,7 @@ def iter_patch_lines(tree: ast.Module, patches: Patches) -> "Iterator[tuple[int,
patch_lines = "\n".join(spec.as_decorator() for spec in specs)
yield (lineno - 1, textwrap.indent(patch_lines, indent))
# Phase 2: Iterate and mark inhereted tests
# Phase 2: Iterate and mark inherited tests
for cls_name, tests in patches.items():
lineno = cache.get(cls_name)
if not lineno:
@@ -273,7 +282,7 @@ def {test_name}(self):
yield (lineno, textwrap.indent(patch_lines, DEFAULT_INDENT))
def has_unittest_import(tree: ast.Module) -> bool:
def _has_unittest_import(tree: ast.Module) -> bool:
"""Check if 'import unittest' is already present in the file."""
for node in tree.body:
if isinstance(node, ast.Import):
@@ -283,7 +292,7 @@ def has_unittest_import(tree: ast.Module) -> bool:
return False
def find_import_insert_line(tree: ast.Module) -> int:
def _find_import_insert_line(tree: ast.Module) -> int:
"""Find the line number after the last import statement."""
last_import_line = None
for node in tree.body:
@@ -294,22 +303,23 @@ def find_import_insert_line(tree: ast.Module) -> int:
def apply_patches(contents: str, patches: Patches) -> str:
"""Apply patches to file contents and return modified contents."""
tree = ast.parse(contents)
lines = contents.splitlines()
modifications = list(iter_patch_lines(tree, patches))
modifications = list(_iter_patch_lines(tree, patches))
# If we have modifications and unittest is not imported, add it
if modifications and not has_unittest_import(tree):
import_line = find_import_insert_line(tree)
if modifications and not _has_unittest_import(tree):
import_line = _find_import_insert_line(tree)
modifications.append(
(
import_line,
"\nimport unittest # XXX: RUSTPYTHON; importing to be able to skip tests",
"\nimport unittest # XXX: RUSTPYTHON; importing to be able to skip tests",
)
)
# Going in reverse to not distrupt the line offset
# Going in reverse to not disrupt the line offset
for lineno, patch in sorted(modifications, reverse=True):
lines.insert(lineno, patch)
@@ -317,111 +327,26 @@ def apply_patches(contents: str, patches: Patches) -> str:
return f"{joined}\n"
def write_output(data: str, dest: str) -> None:
if dest == "-":
print(data, end="")
return
with open(dest, "w") as fd:
fd.write(data)
def build_argparse() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter
)
patches_group = parser.add_mutually_exclusive_group(required=True)
patches_group.add_argument(
"--quick-upgrade",
help="Quick upgrade: path containing /Lib/ (e.g., cpython/Lib/test/foo.py)",
type=pathlib.Path,
metavar="PATH",
)
patches_group.add_argument(
"-p",
"--patches",
help="File path to file containing patches in a JSON format",
type=pathlib.Path,
)
patches_group.add_argument(
"--from",
help="File to gather patches from",
dest="gather_from",
type=pathlib.Path,
)
group = parser.add_mutually_exclusive_group(required=False)
group.add_argument(
"--to",
help="File to apply patches to",
type=pathlib.Path,
)
group.add_argument(
"--show-patches", action="store_true", help="Show the patches and exit"
)
parser.add_argument(
"-o", "--output", default="-", help="Output file. Set to '-' for stdout"
)
return parser
if __name__ == "__main__":
parser = build_argparse()
args = parser.parse_args()
# Quick upgrade: auto-fill --from, --to, -o from path
if args.quick_upgrade is not None:
# Normalize path separators to forward slashes for cross-platform support
path_str = str(args.quick_upgrade).replace("\\", "/")
lib_marker = "/Lib/"
if lib_marker not in path_str:
parser.error(
f"--quick-upgrade path must contain '/Lib/' or '\\Lib\\' (got: {args.quick_upgrade})"
)
idx = path_str.index(lib_marker)
lib_path = pathlib.Path(path_str[idx + 1 :])
args.gather_from = lib_path
args.to = args.quick_upgrade
if args.output == "-":
args.output = str(lib_path)
# Validate required arguments
if args.patches is None and args.gather_from is None:
parser.error("--from or --patches is required (or use --quick-upgrade)")
if args.to is None and not args.show_patches:
parser.error("--to or --show-patches is required")
if args.patches:
patches = {
cls_name: {
test_name: [
PatchSpec(**spec)._replace(ut_method=UtMethod(spec["ut_method"]))
for spec in specs
]
for test_name, specs in tests.items()
}
for cls_name, tests in json.loads(args.patches.read_text()).items()
def patches_to_json(patches: Patches) -> dict:
"""Convert patches to JSON-serializable dict."""
return {
cls_name: {
test_name: [spec._asdict() for spec in specs]
for test_name, specs in tests.items()
}
else:
patches = build_patch_dict(iter_patches(args.gather_from.read_text()))
for cls_name, tests in patches.items()
}
if args.show_patches:
patches = {
cls_name: {
test_name: [spec._asdict() for spec in specs]
for test_name, specs in tests.items()
}
for cls_name, tests in patches.items()
def patches_from_json(data: dict) -> Patches:
"""Convert JSON dict back to Patches."""
return {
cls_name: {
test_name: [
PatchSpec(**spec)._replace(ut_method=UtMethod(spec["ut_method"]))
for spec in specs
]
for test_name, specs in tests.items()
}
output = json.dumps(patches, indent=4) + "\n"
write_output(output, args.output)
sys.exit(0)
patched = apply_patches(args.to.read_text(), patches)
write_output(patched, args.output)
for cls_name, tests in data.items()
}
scripts/update_lib/patches.py
+106
View File
@@ -0,0 +1,106 @@
#!/usr/bin/env python
"""
Patch management for test files.
Usage:
# Extract patches from one file and apply to another
python scripts/update_lib patches --from Lib/test/foo.py --to cpython/Lib/test/foo.py
# Show patches as JSON
python scripts/update_lib patches --from Lib/test/foo.py --show-patches
# Apply patches from JSON file
python scripts/update_lib patches -p patches.json --to Lib/test/foo.py
"""
import argparse
import json
import pathlib
import sys
sys.path.insert(0, str(pathlib.Path(__file__).parent.parent))
def write_output(data: str, dest: str) -> None:
if dest == "-":
print(data, end="")
return
with open(dest, "w") as fd:
fd.write(data)
def main(argv: list[str] | None = None) -> int:
from update_lib import (
apply_patches,
extract_patches,
patches_from_json,
patches_to_json,
)
parser = argparse.ArgumentParser(
description=__doc__,
formatter_class=argparse.RawDescriptionHelpFormatter,
)
patches_group = parser.add_mutually_exclusive_group(required=True)
patches_group.add_argument(
"-p",
"--patches",
type=pathlib.Path,
help="File path to file containing patches in a JSON format",
)
patches_group.add_argument(
"--from",
dest="gather_from",
type=pathlib.Path,
help="File to gather patches from",
)
group = parser.add_mutually_exclusive_group(required=False)
group.add_argument(
"--to",
type=pathlib.Path,
help="File to apply patches to",
)
group.add_argument(
"--show-patches",
action="store_true",
help="Show the patches and exit",
)
parser.add_argument(
"-o",
"--output",
default="-",
help="Output file. Set to '-' for stdout",
)
args = parser.parse_args(argv)
# Validate required arguments
if args.to is None and not args.show_patches:
parser.error("--to or --show-patches is required")
try:
if args.patches:
patches = patches_from_json(json.loads(args.patches.read_text()))
else:
patches = extract_patches(args.gather_from.read_text())
if args.show_patches:
output = json.dumps(patches_to_json(patches), indent=4) + "\n"
write_output(output, args.output)
return 0
patched = apply_patches(args.to.read_text(), patches)
write_output(patched, args.output)
return 0
except FileNotFoundError as e:
print(f"Error: {e}", file=sys.stderr)
return 1
if __name__ == "__main__":
sys.exit(main())
scripts/update_lib/path.py
+99
View File
@@ -0,0 +1,99 @@
"""
Path utilities for update_lib.
This module provides functions for:
- Parsing and converting library paths
- Detecting test paths vs library paths
- Extracting test module names from paths
"""
import pathlib
def parse_lib_path(path: pathlib.Path | str) -> pathlib.Path:
"""
Extract the Lib/... portion from a path containing /Lib/.
Example:
parse_lib_path("cpython/Lib/test/foo.py") -> Path("Lib/test/foo.py")
"""
path_str = str(path).replace("\\", "/")
lib_marker = "/Lib/"
if lib_marker not in path_str:
raise ValueError(f"Path must contain '/Lib/' or '\\Lib\\' (got: {path})")
idx = path_str.index(lib_marker)
return pathlib.Path(path_str[idx + 1 :])
def is_lib_path(path: pathlib.Path) -> bool:
"""Check if path starts with Lib/"""
path_str = str(path).replace("\\", "/")
return path_str.startswith("Lib/") or path_str.startswith("./Lib/")
def is_test_path(path: pathlib.Path) -> bool:
"""Check if path is a test path (contains /Lib/test/ or starts with Lib/test/)"""
path_str = str(path).replace("\\", "/")
return "/Lib/test/" in path_str or path_str.startswith("Lib/test/")
def lib_to_test_path(src_path: pathlib.Path) -> pathlib.Path:
"""
Convert library path to test path.
Examples:
cpython/Lib/dataclasses.py -> cpython/Lib/test/test_dataclasses/ (if dir exists)
cpython/Lib/typing.py -> cpython/Lib/test/test_typing.py (if file exists)
cpython/Lib/json/ -> cpython/Lib/test/test_json/
Lib/dataclasses.py -> Lib/test/test_dataclasses/
"""
path_str = str(src_path).replace("\\", "/")
lib_marker = "/Lib/"
if lib_marker in path_str:
lib_path = parse_lib_path(src_path)
lib_name = lib_path.stem if lib_path.suffix == ".py" else lib_path.name
prefix = path_str[: path_str.index(lib_marker)]
# Try directory first, then file
dir_path = pathlib.Path(f"{prefix}/Lib/test/test_{lib_name}/")
if dir_path.exists():
return dir_path
file_path = pathlib.Path(f"{prefix}/Lib/test/test_{lib_name}.py")
if file_path.exists():
return file_path
# Default to directory (caller will handle non-existence)
return dir_path
else:
# Path starts with Lib/ - extract name directly
lib_name = src_path.stem if src_path.suffix == ".py" else src_path.name
# Try directory first, then file
dir_path = pathlib.Path(f"Lib/test/test_{lib_name}/")
if dir_path.exists():
return dir_path
file_path = pathlib.Path(f"Lib/test/test_{lib_name}.py")
if file_path.exists():
return file_path
return dir_path
def get_test_files(path: pathlib.Path) -> list[pathlib.Path]:
"""Get all .py test files in a path (file or directory)."""
if path.is_file():
return [path]
return sorted(path.glob("**/*.py"))
def test_name_from_path(test_path: pathlib.Path) -> str:
"""
Extract test module name from a test file path.
Examples:
Lib/test/test_foo.py -> test_foo
Lib/test/test_ctypes/test_bar.py -> test_ctypes.test_bar
"""
test_path = pathlib.Path(test_path)
if test_path.parent.name.startswith("test_"):
return f"{test_path.parent.name}.{test_path.stem}"
return test_path.stem
scripts/update_lib/quick.py
+365
View File
@@ -0,0 +1,365 @@
#!/usr/bin/env python
"""
Quick update for test files from CPython.
Usage:
# Library + test: copy lib, then patch + auto-mark test + commit
python scripts/update_lib quick cpython/Lib/dataclasses.py
# Shortcut: just the module name
python scripts/update_lib quick dataclasses
# Test file: patch + auto-mark
python scripts/update_lib quick cpython/Lib/test/test_foo.py
# Test file: migrate only
python scripts/update_lib quick cpython/Lib/test/test_foo.py --no-auto-mark
# Test file: auto-mark only (Lib/ path implies --no-migrate)
python scripts/update_lib quick Lib/test/test_foo.py
# Directory: patch all + auto-mark all
python scripts/update_lib quick cpython/Lib/test/test_dataclasses/
# Skip git commit
python scripts/update_lib quick dataclasses --no-commit
"""
import argparse
import pathlib
import sys
sys.path.insert(0, str(pathlib.Path(__file__).parent.parent))
from update_lib.path import (
get_test_files,
is_lib_path,
is_test_path,
lib_to_test_path,
parse_lib_path,
)
def collect_original_methods(
lib_path: pathlib.Path,
) -> set[tuple[str, str]] | dict[pathlib.Path, set[tuple[str, str]]] | None:
"""
Collect original test methods from lib path before patching.
Returns:
- For file: set of (class_name, method_name) or None if file doesn't exist
- For directory: dict mapping file path to set of methods, or None if dir doesn't exist
"""
from update_lib.auto_mark import extract_test_methods
if not lib_path.exists():
return None
if lib_path.is_file():
return extract_test_methods(lib_path.read_text())
else:
result = {}
for lib_file in get_test_files(lib_path):
result[lib_file.resolve()] = extract_test_methods(lib_file.read_text())
return result
def quick(
src_path: pathlib.Path,
no_migrate: bool = False,
no_auto_mark: bool = False,
mark_failure: bool = False,
verbose: bool = True,
skip_build: bool = False,
) -> None:
"""
Process a file or directory: migrate + auto-mark.
Args:
src_path: Source path (file or directory)
no_migrate: Skip migration step
no_auto_mark: Skip auto-mark step
mark_failure: Add @expectedFailure to ALL failing tests
verbose: Print progress messages
skip_build: Skip cargo build, use pre-built binary
"""
from update_lib.auto_mark import auto_mark_directory, auto_mark_file
from update_lib.migrate import patch_directory, patch_file
# Determine lib_path and whether to migrate
if is_lib_path(src_path):
no_migrate = True
lib_path = src_path
else:
lib_path = parse_lib_path(src_path)
is_dir = src_path.is_dir()
# Capture original test methods before migration (for smart auto-mark)
original_methods = collect_original_methods(lib_path)
# Step 1: Migrate
if not no_migrate:
if is_dir:
patch_directory(src_path, lib_path, verbose=verbose)
else:
patch_file(src_path, lib_path, verbose=verbose)
# Step 2: Auto-mark
if not no_auto_mark:
if not lib_path.exists():
raise FileNotFoundError(f"Path not found: {lib_path}")
if is_dir:
num_added, num_removed, _ = auto_mark_directory(
lib_path,
mark_failure=mark_failure,
verbose=verbose,
original_methods_per_file=original_methods,
skip_build=skip_build,
)
else:
num_added, num_removed, _ = auto_mark_file(
lib_path,
mark_failure=mark_failure,
verbose=verbose,
original_methods=original_methods,
skip_build=skip_build,
)
if verbose:
if num_added:
print(f"Added expectedFailure to {num_added} tests")
print(f"Removed expectedFailure from {num_removed} tests")
def get_cpython_dir(src_path: pathlib.Path) -> pathlib.Path:
"""Extract cpython directory from source path.
Example:
cpython/Lib/dataclasses.py -> cpython
/some/path/cpython/Lib/foo.py -> /some/path/cpython
"""
path_str = str(src_path).replace("\\", "/")
lib_marker = "/Lib/"
if lib_marker in path_str:
idx = path_str.index(lib_marker)
return pathlib.Path(path_str[:idx])
# Shortcut case: assume "cpython"
return pathlib.Path("cpython")
def get_cpython_version(cpython_dir: pathlib.Path) -> str:
"""Get CPython version from git tag."""
import subprocess
result = subprocess.run(
["git", "describe", "--tags"],
cwd=cpython_dir,
capture_output=True,
text=True,
check=True,
)
return result.stdout.strip()
def git_commit(
name: str,
lib_path: pathlib.Path | None,
test_path: pathlib.Path | None,
cpython_dir: pathlib.Path,
verbose: bool = True,
) -> bool:
"""Commit changes with CPython author.
Args:
name: Module name (e.g., "dataclasses")
lib_path: Path to library file/directory (or None)
test_path: Path to test file/directory (or None)
cpython_dir: Path to cpython directory
verbose: Print progress messages
Returns:
True if commit was created, False otherwise
"""
import subprocess
# Stage changes
paths_to_add = []
if lib_path and lib_path.exists():
paths_to_add.append(str(lib_path))
if test_path and test_path.exists():
paths_to_add.append(str(test_path))
if not paths_to_add:
return False
version = get_cpython_version(cpython_dir)
subprocess.run(["git", "add"] + paths_to_add, check=True)
# Check if there are staged changes
result = subprocess.run(
["git", "diff", "--cached", "--quiet"],
capture_output=True,
)
if result.returncode == 0:
if verbose:
print("No changes to commit")
return False
# Commit with CPython author
message = f"Update {name} from {version}"
subprocess.run(
[
"git",
"commit",
"--author",
"CPython Developers <>",
"-m",
message,
],
check=True,
)
if verbose:
print(f"Committed: {message}")
return True
def _expand_shortcut(path: pathlib.Path) -> pathlib.Path:
"""Expand simple name to cpython/Lib path if it exists.
Examples:
dataclasses -> cpython/Lib/dataclasses.py (if exists)
json -> cpython/Lib/json/ (if exists)
test_types -> cpython/Lib/test/test_types.py (if exists)
"""
# Only expand if it's a simple name (no path separators) and doesn't exist
if "/" in str(path) or path.exists():
return path
name = str(path)
# Test shortcut: test_foo -> cpython/Lib/test/test_foo
if name.startswith("test_"):
dir_path = pathlib.Path(f"cpython/Lib/test/{name}")
if dir_path.exists():
return dir_path
file_path = pathlib.Path(f"cpython/Lib/test/{name}.py")
if file_path.exists():
return file_path
# Library shortcut: foo -> cpython/Lib/foo
file_path = pathlib.Path(f"cpython/Lib/{name}.py")
if file_path.exists():
return file_path
dir_path = pathlib.Path(f"cpython/Lib/{name}")
if dir_path.exists():
return dir_path
# Return original (will likely fail later with a clear error)
return path
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(
description=__doc__,
formatter_class=argparse.RawDescriptionHelpFormatter,
)
parser.add_argument(
"path",
type=pathlib.Path,
help="Source path (file or directory)",
)
parser.add_argument(
"--no-copy",
action="store_true",
help="Skip copy-lib step (implied if path is a test path)",
)
parser.add_argument(
"--no-migrate",
action="store_true",
help="Skip migration step (implied if path starts with Lib/)",
)
parser.add_argument(
"--no-auto-mark",
action="store_true",
help="Skip auto-mark step",
)
parser.add_argument(
"--mark-failure",
action="store_true",
help="Add @expectedFailure to failing tests",
)
parser.add_argument(
"--no-commit",
action="store_true",
help="Skip git commit step",
)
parser.add_argument(
"--skip-build",
action="store_true",
help="Skip cargo build, use pre-built binary",
)
args = parser.parse_args(argv)
try:
src_path = args.path
no_copy = args.no_copy
# Shortcut: expand simple name to cpython/Lib path
src_path = _expand_shortcut(src_path)
original_src = src_path # Keep for commit
# Track library path for commit
lib_file_path = None
test_path = None
# If it's a library path (not test path), do copy_lib first
if not is_test_path(src_path):
# Get library destination path for commit
lib_file_path = parse_lib_path(src_path)
if not no_copy:
from update_lib.copy_lib import copy_lib
copy_lib(src_path)
# Convert to test path
src_path = lib_to_test_path(original_src)
if not src_path.exists():
print(f"Test path does not exist: {src_path}")
return 1
test_path = parse_lib_path(src_path) if not is_lib_path(src_path) else src_path
# Process the test path
quick(
src_path,
no_migrate=args.no_migrate,
no_auto_mark=args.no_auto_mark,
mark_failure=args.mark_failure,
skip_build=args.skip_build,
)
# Step 3: Git commit
if not args.no_commit:
# Extract module name from path
name = original_src.stem
if name == "__init__":
name = original_src.parent.name
cpython_dir = get_cpython_dir(original_src)
git_commit(name, lib_file_path, test_path, cpython_dir)
return 0
except ValueError as e:
print(f"Error: {e}", file=sys.stderr)
return 1
except FileNotFoundError as e:
print(f"Error: {e}", file=sys.stderr)
return 1
if __name__ == "__main__":
sys.exit(main())
scripts/update_lib/tests/__init__.py
View File
Whitespace-only changes.
scripts/update_lib/tests/test_auto_mark.py
+593
View File
@@ -0,0 +1,593 @@
"""Tests for auto_mark.py - test result parsing and auto-marking."""
import subprocess
import unittest
from update_lib.auto_mark import (
Test,
TestResult,
apply_test_changes,
collect_test_changes,
extract_test_methods,
parse_results,
path_to_test_parts,
remove_expected_failures,
)
from update_lib.patch_spec import COMMENT
class TestParseResults(unittest.TestCase):
"""Tests for parse_results function."""
def _make_result(self, stdout: str) -> subprocess.CompletedProcess:
"""Create a mock CompletedProcess."""
return subprocess.CompletedProcess(
args=["test"],
returncode=0,
stdout=stdout,
stderr="",
)
def test_parse_failing_test(self):
"""Test parsing a failing test."""
stdout = """
Run 1 tests sequentially
test_foo (test.test_example.TestClass) ... FAIL
-----------
"""
result = parse_results(self._make_result(stdout))
self.assertEqual(len(result.tests), 1)
self.assertEqual(result.tests[0].name, "test_foo")
self.assertEqual(result.tests[0].path, "test.test_example.TestClass")
self.assertEqual(result.tests[0].result, "fail")
def test_parse_error_test(self):
"""Test parsing an error test."""
stdout = """
Run 1 tests sequentially
test_bar (test.test_example.TestClass) ... ERROR
-----------
"""
result = parse_results(self._make_result(stdout))
self.assertEqual(len(result.tests), 1)
self.assertEqual(result.tests[0].result, "error")
def test_parse_ok_test_ignored(self):
"""Test that passing tests are ignored."""
stdout = """
Run 1 tests sequentially
test_foo (test.test_example.TestClass) ... ok
-----------
"""
result = parse_results(self._make_result(stdout))
self.assertEqual(len(result.tests), 0)
def test_parse_unexpected_success(self):
"""Test parsing unexpected success."""
stdout = """
Run 1 tests sequentially
test_foo (test.test_example.TestClass) ... unexpected success
-----------
UNEXPECTED SUCCESS: test_foo (test.test_example.TestClass)
"""
result = parse_results(self._make_result(stdout))
self.assertEqual(len(result.unexpected_successes), 1)
self.assertEqual(result.unexpected_successes[0].name, "test_foo")
self.assertEqual(
result.unexpected_successes[0].path, "test.test_example.TestClass"
)
def test_parse_tests_result(self):
"""Test parsing tests result line."""
stdout = """
== Tests result: FAILURE ==
"""
result = parse_results(self._make_result(stdout))
self.assertEqual(result.tests_result, "FAILURE")
def test_parse_multiple_tests(self):
"""Test parsing multiple test results."""
stdout = """
Run 3 tests sequentially
test_one (test.test_example.TestA) ... FAIL
test_two (test.test_example.TestA) ... ok
test_three (test.test_example.TestB) ... ERROR
-----------
"""
result = parse_results(self._make_result(stdout))
self.assertEqual(len(result.tests), 2) # Only FAIL and ERROR
def test_parse_error_message(self):
"""Test parsing error message from traceback."""
stdout = """
Run 1 tests sequentially
test_foo (test.test_example.TestClass) ... FAIL
-----------
======================================================================
FAIL: test_foo (test.test_example.TestClass)
----------------------------------------------------------------------
Traceback (most recent call last):
File "test.py", line 10, in test_foo
self.assertEqual(1, 2)
AssertionError: 1 != 2
======================================================================
"""
result = parse_results(self._make_result(stdout))
self.assertEqual(len(result.tests), 1)
self.assertEqual(result.tests[0].error_message, "AssertionError: 1 != 2")
def test_parse_multiple_error_messages(self):
"""Test parsing multiple error messages."""
stdout = """
Run 2 tests sequentially
test_foo (test.test_example.TestClass) ... FAIL
test_bar (test.test_example.TestClass) ... ERROR
-----------
======================================================================
FAIL: test_foo (test.test_example.TestClass)
----------------------------------------------------------------------
Traceback (most recent call last):
File "test.py", line 10, in test_foo
self.assertEqual(1, 2)
AssertionError: 1 != 2
======================================================================
ERROR: test_bar (test.test_example.TestClass)
----------------------------------------------------------------------
Traceback (most recent call last):
File "test.py", line 20, in test_bar
raise ValueError("oops")
ValueError: oops
======================================================================
"""
result = parse_results(self._make_result(stdout))
self.assertEqual(len(result.tests), 2)
# Find tests by name
test_foo = next(t for t in result.tests if t.name == "test_foo")
test_bar = next(t for t in result.tests if t.name == "test_bar")
self.assertEqual(test_foo.error_message, "AssertionError: 1 != 2")
self.assertEqual(test_bar.error_message, "ValueError: oops")
class TestPathToTestParts(unittest.TestCase):
"""Tests for path_to_test_parts function."""
def test_simple_path(self):
"""Test extracting parts from simple path."""
parts = path_to_test_parts("test.test_foo.TestClass.test_method")
self.assertEqual(parts, ["TestClass", "test_method"])
def test_nested_path(self):
"""Test extracting parts from nested path."""
parts = path_to_test_parts("test.test_foo.test_bar.TestClass.test_method")
self.assertEqual(parts, ["TestClass", "test_method"])
class TestCollectTestChanges(unittest.TestCase):
"""Tests for collect_test_changes function."""
def test_collect_failing_tests(self):
"""Test collecting failing tests."""
results = TestResult()
results.tests = [
Test(
name="test_foo",
path="test.test_example.TestClass.test_foo",
result="fail",
),
Test(
name="test_bar",
path="test.test_example.TestClass.test_bar",
result="error",
),
]
failing, successes, error_messages = collect_test_changes(results)
self.assertEqual(len(failing), 2)
self.assertIn(("TestClass", "test_foo"), failing)
self.assertIn(("TestClass", "test_bar"), failing)
self.assertEqual(len(successes), 0)
def test_collect_unexpected_successes(self):
"""Test collecting unexpected successes."""
results = TestResult()
results.unexpected_successes = [
Test(
name="test_foo",
path="test.test_example.TestClass.test_foo",
result="unexpected_success",
),
]
failing, successes, error_messages = collect_test_changes(results)
self.assertEqual(len(failing), 0)
self.assertEqual(len(successes), 1)
self.assertIn(("TestClass", "test_foo"), successes)
def test_collect_with_module_prefix(self):
"""Test collecting with module prefix filter."""
results = TestResult()
results.tests = [
Test(name="test_foo", path="test_a.TestClass.test_foo", result="fail"),
Test(name="test_bar", path="test_b.TestClass.test_bar", result="fail"),
]
failing, _, _ = collect_test_changes(results, module_prefix="test_a.")
self.assertEqual(len(failing), 1)
self.assertIn(("TestClass", "test_foo"), failing)
def test_collect_error_messages(self):
"""Test collecting error messages."""
results = TestResult()
results.tests = [
Test(
name="test_foo",
path="test.test_example.TestClass.test_foo",
result="fail",
error_message="AssertionError: 1 != 2",
),
Test(
name="test_bar",
path="test.test_example.TestClass.test_bar",
result="error",
error_message="",
),
]
failing, successes, error_messages = collect_test_changes(results)
self.assertEqual(len(error_messages), 1)
self.assertEqual(
error_messages[("TestClass", "test_foo")], "AssertionError: 1 != 2"
)
def test_collect_with_test_prefix_in_path(self):
"""Test collecting with 'test.' prefix in path (like real test output)."""
results = TestResult()
results.tests = [
Test(
name="test_foo",
path="test.test_dataclasses.TestCase.test_foo",
result="fail",
),
Test(
name="test_bar",
path="test.test_other.TestOther.test_bar",
result="fail",
),
]
# Filter with prefix that matches real test module path format
failing, _, _ = collect_test_changes(
results, module_prefix="test.test_dataclasses."
)
self.assertEqual(len(failing), 1)
self.assertIn(("TestCase", "test_foo"), failing)
def test_collect_init_module_matching(self):
"""Test that __init__.py tests match without __init__ in path.
When test results come from a package's __init__.py, the path is like:
'test.test_dataclasses.TestCase.test_foo' (no __init__)
But module_prefix from test_name_from_path would be:
'test_dataclasses.__init__'
So we need to strip '.__init__' and add 'test.' prefix.
"""
results = TestResult()
results.tests = [
Test(
name="test_field_repr",
path="test.test_dataclasses.TestCase.test_field_repr",
result="fail",
),
]
# Simulate the corrected prefix (after stripping .__init__ and adding test.)
module_prefix = "test_dataclasses.__init__"
if module_prefix.endswith(".__init__"):
module_prefix = module_prefix[:-9]
module_prefix = "test." + module_prefix + "."
failing, _, _ = collect_test_changes(results, module_prefix=module_prefix)
self.assertEqual(len(failing), 1)
self.assertIn(("TestCase", "test_field_repr"), failing)
class TestExtractTestMethods(unittest.TestCase):
"""Tests for extract_test_methods function."""
def test_extract_simple(self):
"""Test extracting test methods from simple class."""
code = """
class TestFoo(unittest.TestCase):
def test_one(self):
pass
def test_two(self):
pass
"""
methods = extract_test_methods(code)
self.assertEqual(len(methods), 2)
self.assertIn(("TestFoo", "test_one"), methods)
self.assertIn(("TestFoo", "test_two"), methods)
def test_extract_multiple_classes(self):
"""Test extracting from multiple classes."""
code = """
class TestA(unittest.TestCase):
def test_a(self):
pass
class TestB(unittest.TestCase):
def test_b(self):
pass
"""
methods = extract_test_methods(code)
self.assertEqual(len(methods), 2)
self.assertIn(("TestA", "test_a"), methods)
self.assertIn(("TestB", "test_b"), methods)
def test_extract_syntax_error_returns_empty(self):
"""Test that syntax error returns empty set."""
code = "this is not valid python {"
methods = extract_test_methods(code)
self.assertEqual(methods, set())
class TestRemoveExpectedFailures(unittest.TestCase):
"""Tests for remove_expected_failures function."""
def test_remove_simple(self):
"""Test removing simple expectedFailure decorator."""
code = f"""import unittest
class TestFoo(unittest.TestCase):
# {COMMENT}
@unittest.expectedFailure
def test_one(self):
pass
"""
result = remove_expected_failures(code, {("TestFoo", "test_one")})
self.assertNotIn("@unittest.expectedFailure", result)
self.assertIn("def test_one(self):", result)
def test_remove_with_inline_comment(self):
"""Test removing expectedFailure with inline comment."""
code = f"""import unittest
class TestFoo(unittest.TestCase):
@unittest.expectedFailure # {COMMENT}
def test_one(self):
pass
"""
result = remove_expected_failures(code, {("TestFoo", "test_one")})
self.assertNotIn("@unittest.expectedFailure", result)
def test_remove_super_call_method(self):
"""Test removing method that just calls super()."""
code = f"""import unittest
class TestFoo(unittest.TestCase):
# {COMMENT}
@unittest.expectedFailure
def test_one(self):
return super().test_one()
"""
result = remove_expected_failures(code, {("TestFoo", "test_one")})
self.assertNotIn("def test_one", result)
def test_no_removal_without_comment(self):
"""Test that decorators without COMMENT are not removed."""
code = """import unittest
class TestFoo(unittest.TestCase):
@unittest.expectedFailure
def test_one(self):
pass
"""
result = remove_expected_failures(code, {("TestFoo", "test_one")})
# Should still have the decorator
self.assertIn("@unittest.expectedFailure", result)
class TestApplyTestChanges(unittest.TestCase):
"""Tests for apply_test_changes function."""
def test_apply_failing_tests(self):
"""Test applying expectedFailure to failing tests."""
code = """import unittest
class TestFoo(unittest.TestCase):
def test_one(self):
pass
"""
failing = {("TestFoo", "test_one")}
result = apply_test_changes(code, failing, set())
self.assertIn("@unittest.expectedFailure", result)
self.assertIn(COMMENT, result)
def test_apply_removes_unexpected_success(self):
"""Test removing expectedFailure from unexpected success."""
code = f"""import unittest
class TestFoo(unittest.TestCase):
# {COMMENT}
@unittest.expectedFailure
def test_one(self):
pass
"""
successes = {("TestFoo", "test_one")}
result = apply_test_changes(code, set(), successes)
self.assertNotIn("@unittest.expectedFailure", result)
self.assertIn("def test_one(self):", result)
def test_apply_both_changes(self):
"""Test applying both failing tests and removing unexpected successes."""
code = f"""import unittest
class TestFoo(unittest.TestCase):
def test_one(self):
pass
# {COMMENT}
@unittest.expectedFailure
def test_two(self):
pass
"""
failing = {("TestFoo", "test_one")}
successes = {("TestFoo", "test_two")}
result = apply_test_changes(code, failing, successes)
# test_one should now have expectedFailure
self.assertIn("def test_one(self):", result)
# Only one expectedFailure decorator should remain (on test_one)
self.assertEqual(result.count("@unittest.expectedFailure"), 1)
def test_apply_with_error_message(self):
"""Test applying expectedFailure with error message."""
code = """import unittest
class TestFoo(unittest.TestCase):
def test_one(self):
pass
"""
failing = {("TestFoo", "test_one")}
error_messages = {("TestFoo", "test_one"): "AssertionError: 1 != 2"}
result = apply_test_changes(code, failing, set(), error_messages)
self.assertIn("@unittest.expectedFailure", result)
self.assertIn("AssertionError: 1 != 2", result)
self.assertIn(COMMENT, result)
class TestSmartAutoMarkFiltering(unittest.TestCase):
"""Tests for smart auto-mark filtering logic (regression exclusion).
The smart auto-mark feature:
- Marks NEW test failures (tests that didn't exist before)
- Does NOT mark regressions (existing tests that now fail)
"""
def _filter_failures(
self,
all_failing_tests: set[tuple[str, str]],
original_methods: set[tuple[str, str]],
current_methods: set[tuple[str, str]],
) -> tuple[set[tuple[str, str]], set[tuple[str, str]]]:
"""Simulate the filtering logic from auto_mark_file().
Returns:
(failing_tests_to_mark, regressions)
"""
new_methods = current_methods - original_methods
failing_tests = {t for t in all_failing_tests if t in new_methods}
regressions = all_failing_tests - failing_tests
return failing_tests, regressions
def test_new_tests_get_marked(self):
"""Test that new failing tests are marked."""
original_methods = {("TestFoo", "test_existing")}
current_methods = {("TestFoo", "test_existing"), ("TestFoo", "test_new")}
all_failing = {("TestFoo", "test_new")}
to_mark, regressions = self._filter_failures(
all_failing, original_methods, current_methods
)
self.assertEqual(to_mark, {("TestFoo", "test_new")})
self.assertEqual(regressions, set())
def test_regressions_not_marked(self):
"""Test that existing failing tests (regressions) are NOT marked."""
original_methods = {("TestFoo", "test_existing")}
current_methods = {("TestFoo", "test_existing")}
all_failing = {("TestFoo", "test_existing")}
to_mark, regressions = self._filter_failures(
all_failing, original_methods, current_methods
)
self.assertEqual(to_mark, set())
self.assertEqual(regressions, {("TestFoo", "test_existing")})
def test_mixed_new_and_regression(self):
"""Test with both new failures and regressions."""
original_methods = {("TestFoo", "test_old1"), ("TestFoo", "test_old2")}
current_methods = {
("TestFoo", "test_old1"),
("TestFoo", "test_old2"),
("TestFoo", "test_new1"),
("TestFoo", "test_new2"),
}
# test_old1 regressed, test_new1 is a new failure
all_failing = {("TestFoo", "test_old1"), ("TestFoo", "test_new1")}
to_mark, regressions = self._filter_failures(
all_failing, original_methods, current_methods
)
self.assertEqual(to_mark, {("TestFoo", "test_new1")})
self.assertEqual(regressions, {("TestFoo", "test_old1")})
def test_multiple_classes(self):
"""Test filtering across multiple classes."""
original_methods = {("TestA", "test_a"), ("TestB", "test_b")}
current_methods = {
("TestA", "test_a"),
("TestA", "test_new_a"),
("TestB", "test_b"),
("TestC", "test_c"), # entirely new class
}
all_failing = {
("TestA", "test_a"), # regression
("TestA", "test_new_a"), # new
("TestC", "test_c"), # new (new class)
}
to_mark, regressions = self._filter_failures(
all_failing, original_methods, current_methods
)
self.assertEqual(to_mark, {("TestA", "test_new_a"), ("TestC", "test_c")})
self.assertEqual(regressions, {("TestA", "test_a")})
def test_all_new_tests(self):
"""Test when all failing tests are new (no regressions)."""
original_methods = set() # file was new
current_methods = {("TestFoo", "test_one"), ("TestFoo", "test_two")}
all_failing = {("TestFoo", "test_one"), ("TestFoo", "test_two")}
to_mark, regressions = self._filter_failures(
all_failing, original_methods, current_methods
)
self.assertEqual(to_mark, all_failing)
self.assertEqual(regressions, set())
def test_all_regressions(self):
"""Test when all failing tests are regressions (no new tests)."""
original_methods = {("TestFoo", "test_one"), ("TestFoo", "test_two")}
current_methods = original_methods.copy()
all_failing = {("TestFoo", "test_one")}
to_mark, regressions = self._filter_failures(
all_failing, original_methods, current_methods
)
self.assertEqual(to_mark, set())
self.assertEqual(regressions, {("TestFoo", "test_one")})
if __name__ == "__main__":
unittest.main()
scripts/update_lib/tests/test_migrate.py
+196
View File
@@ -0,0 +1,196 @@
"""Tests for migrate.py - file migration operations."""
import pathlib
import tempfile
import unittest
from update_lib.migrate import (
patch_directory,
patch_file,
patch_single_content,
)
from update_lib.patch_spec import COMMENT
class TestPatchSingleContent(unittest.TestCase):
"""Tests for patch_single_content function."""
def test_patch_with_no_existing_file(self):
"""Test patching when lib file doesn't exist."""
with tempfile.TemporaryDirectory() as tmpdir:
tmpdir = pathlib.Path(tmpdir)
# Create source file
src_path = tmpdir / "src.py"
src_path.write_text("""import unittest
class TestFoo(unittest.TestCase):
def test_one(self):
pass
""")
# Non-existent lib path
lib_path = tmpdir / "lib.py"
result = patch_single_content(src_path, lib_path)
# Should return source content unchanged
self.assertIn("def test_one(self):", result)
self.assertNotIn(COMMENT, result)
def test_patch_with_existing_patches(self):
"""Test patching preserves existing patches."""
with tempfile.TemporaryDirectory() as tmpdir:
tmpdir = pathlib.Path(tmpdir)
# Create source file (new version)
src_path = tmpdir / "src.py"
src_path.write_text("""import unittest
class TestFoo(unittest.TestCase):
def test_one(self):
pass
def test_two(self):
pass
""")
# Create lib file with existing patch
lib_path = tmpdir / "lib.py"
lib_path.write_text(f"""import unittest
class TestFoo(unittest.TestCase):
# {COMMENT}
@unittest.expectedFailure
def test_one(self):
pass
""")
result = patch_single_content(src_path, lib_path)
# Should have patch on test_one
self.assertIn("@unittest.expectedFailure", result)
self.assertIn(COMMENT, result)
# Should have test_two from source
self.assertIn("def test_two(self):", result)
class TestPatchFile(unittest.TestCase):
"""Tests for patch_file function."""
def test_patch_file_creates_output(self):
"""Test that patch_file writes output file."""
with tempfile.TemporaryDirectory() as tmpdir:
tmpdir = pathlib.Path(tmpdir)
# Create source file
src_path = tmpdir / "src.py"
src_path.write_text("""import unittest
class TestFoo(unittest.TestCase):
def test_one(self):
pass
""")
# Output path
lib_path = tmpdir / "Lib" / "test.py"
patch_file(src_path, lib_path, verbose=False)
# File should exist
self.assertTrue(lib_path.exists())
content = lib_path.read_text()
self.assertIn("def test_one(self):", content)
def test_patch_file_preserves_patches(self):
"""Test that patch_file preserves existing patches."""
with tempfile.TemporaryDirectory() as tmpdir:
tmpdir = pathlib.Path(tmpdir)
# Create source file
src_path = tmpdir / "src.py"
src_path.write_text("""import unittest
class TestFoo(unittest.TestCase):
def test_one(self):
pass
""")
# Create existing lib file with patch
lib_path = tmpdir / "lib.py"
lib_path.write_text(f"""import unittest
class TestFoo(unittest.TestCase):
# {COMMENT}
@unittest.expectedFailure
def test_one(self):
pass
""")
patch_file(src_path, lib_path, verbose=False)
content = lib_path.read_text()
self.assertIn("@unittest.expectedFailure", content)
class TestPatchDirectory(unittest.TestCase):
"""Tests for patch_directory function."""
def test_patch_directory_all_files(self):
"""Test that patch_directory processes all .py files."""
with tempfile.TemporaryDirectory() as tmpdir:
tmpdir = pathlib.Path(tmpdir)
# Create source directory with files
src_dir = tmpdir / "src"
src_dir.mkdir()
(src_dir / "test_a.py").write_text("# test_a")
(src_dir / "test_b.py").write_text("# test_b")
(src_dir / "subdir").mkdir()
(src_dir / "subdir" / "test_c.py").write_text("# test_c")
# Output directory
lib_dir = tmpdir / "lib"
patch_directory(src_dir, lib_dir, verbose=False)
# All files should exist
self.assertTrue((lib_dir / "test_a.py").exists())
self.assertTrue((lib_dir / "test_b.py").exists())
self.assertTrue((lib_dir / "subdir" / "test_c.py").exists())
def test_patch_directory_preserves_patches(self):
"""Test that patch_directory preserves patches in existing files."""
with tempfile.TemporaryDirectory() as tmpdir:
tmpdir = pathlib.Path(tmpdir)
# Create source directory
src_dir = tmpdir / "src"
src_dir.mkdir()
(src_dir / "test_a.py").write_text("""import unittest
class TestA(unittest.TestCase):
def test_one(self):
pass
""")
# Create lib directory with patched file
lib_dir = tmpdir / "lib"
lib_dir.mkdir()
(lib_dir / "test_a.py").write_text(f"""import unittest
class TestA(unittest.TestCase):
# {COMMENT}
@unittest.expectedFailure
def test_one(self):
pass
""")
patch_directory(src_dir, lib_dir, verbose=False)
content = (lib_dir / "test_a.py").read_text()
self.assertIn("@unittest.expectedFailure", content)
if __name__ == "__main__":
unittest.main()
scripts/update_lib/tests/test_patch_spec.py
+325
View File
@@ -0,0 +1,325 @@
"""Tests for patch_spec.py - core patch extraction and application."""
import ast
import unittest
from update_lib.patch_spec import (
COMMENT,
PatchSpec,
UtMethod,
apply_patches,
extract_patches,
iter_tests,
)
class TestIterTests(unittest.TestCase):
"""Tests for iter_tests function."""
def test_iter_tests_simple(self):
"""Test iterating over test methods in a class."""
code = """
class TestFoo(unittest.TestCase):
def test_one(self):
pass
def test_two(self):
pass
"""
tree = ast.parse(code)
results = list(iter_tests(tree))
self.assertEqual(len(results), 2)
self.assertEqual(results[0][0].name, "TestFoo")
self.assertEqual(results[0][1].name, "test_one")
self.assertEqual(results[1][1].name, "test_two")
def test_iter_tests_multiple_classes(self):
"""Test iterating over multiple test classes."""
code = """
class TestFoo(unittest.TestCase):
def test_foo(self):
pass
class TestBar(unittest.TestCase):
def test_bar(self):
pass
"""
tree = ast.parse(code)
results = list(iter_tests(tree))
self.assertEqual(len(results), 2)
self.assertEqual(results[0][0].name, "TestFoo")
self.assertEqual(results[1][0].name, "TestBar")
def test_iter_tests_async(self):
"""Test iterating over async test methods."""
code = """
class TestAsync(unittest.TestCase):
async def test_async(self):
pass
"""
tree = ast.parse(code)
results = list(iter_tests(tree))
self.assertEqual(len(results), 1)
self.assertEqual(results[0][1].name, "test_async")
class TestExtractPatches(unittest.TestCase):
"""Tests for extract_patches function."""
def test_extract_expected_failure(self):
"""Test extracting @unittest.expectedFailure decorator."""
code = f"""
class TestFoo(unittest.TestCase):
# {COMMENT}
@unittest.expectedFailure
def test_one(self):
pass
"""
patches = extract_patches(code)
self.assertIn("TestFoo", patches)
self.assertIn("test_one", patches["TestFoo"])
specs = patches["TestFoo"]["test_one"]
self.assertEqual(len(specs), 1)
self.assertEqual(specs[0].ut_method, UtMethod.ExpectedFailure)
def test_extract_expected_failure_inline_comment(self):
"""Test extracting expectedFailure with inline comment."""
code = f"""
class TestFoo(unittest.TestCase):
@unittest.expectedFailure # {COMMENT}
def test_one(self):
pass
"""
patches = extract_patches(code)
self.assertIn("TestFoo", patches)
self.assertIn("test_one", patches["TestFoo"])
def test_extract_skip_with_reason(self):
"""Test extracting @unittest.skip with reason."""
code = f'''
class TestFoo(unittest.TestCase):
@unittest.skip("{COMMENT}; not implemented")
def test_one(self):
pass
'''
patches = extract_patches(code)
self.assertIn("TestFoo", patches)
specs = patches["TestFoo"]["test_one"]
self.assertEqual(specs[0].ut_method, UtMethod.Skip)
self.assertIn("not implemented", specs[0].reason)
def test_extract_skip_if(self):
"""Test extracting @unittest.skipIf decorator."""
code = f'''
class TestFoo(unittest.TestCase):
@unittest.skipIf(sys.platform == "win32", "{COMMENT}; windows issue")
def test_one(self):
pass
'''
patches = extract_patches(code)
specs = patches["TestFoo"]["test_one"]
self.assertEqual(specs[0].ut_method, UtMethod.SkipIf)
# ast.unparse normalizes quotes to single quotes
self.assertIn("sys.platform", specs[0].cond)
self.assertIn("win32", specs[0].cond)
def test_no_patches_without_comment(self):
"""Test that decorators without COMMENT are not extracted."""
code = """
class TestFoo(unittest.TestCase):
@unittest.expectedFailure
def test_one(self):
pass
"""
patches = extract_patches(code)
self.assertEqual(patches, {})
def test_multiple_patches_same_method(self):
"""Test extracting multiple decorators on same method."""
code = f'''
class TestFoo(unittest.TestCase):
# {COMMENT}
@unittest.expectedFailure
@unittest.skip("{COMMENT}; reason")
def test_one(self):
pass
'''
patches = extract_patches(code)
specs = patches["TestFoo"]["test_one"]
self.assertEqual(len(specs), 2)
class TestApplyPatches(unittest.TestCase):
"""Tests for apply_patches function."""
def test_apply_expected_failure(self):
"""Test applying @unittest.expectedFailure."""
code = """import unittest
class TestFoo(unittest.TestCase):
def test_one(self):
pass
"""
patches = {
"TestFoo": {"test_one": [PatchSpec(UtMethod.ExpectedFailure, None, "")]}
}
result = apply_patches(code, patches)
self.assertIn("@unittest.expectedFailure", result)
self.assertIn(COMMENT, result)
def test_apply_skip_with_reason(self):
"""Test applying @unittest.skip with reason."""
code = """import unittest
class TestFoo(unittest.TestCase):
def test_one(self):
pass
"""
patches = {
"TestFoo": {"test_one": [PatchSpec(UtMethod.Skip, None, "not ready")]}
}
result = apply_patches(code, patches)
self.assertIn("@unittest.skip", result)
self.assertIn("not ready", result)
def test_apply_skip_if(self):
"""Test applying @unittest.skipIf."""
code = """import unittest
class TestFoo(unittest.TestCase):
def test_one(self):
pass
"""
patches = {
"TestFoo": {
"test_one": [
PatchSpec(UtMethod.SkipIf, "sys.platform == 'win32'", "windows")
]
}
}
result = apply_patches(code, patches)
self.assertIn("@unittest.skipIf", result)
self.assertIn('sys.platform == "win32"', result)
def test_apply_preserves_existing_decorators(self):
"""Test that existing decorators are preserved."""
code = """import unittest
class TestFoo(unittest.TestCase):
@some_decorator
def test_one(self):
pass
"""
patches = {
"TestFoo": {"test_one": [PatchSpec(UtMethod.ExpectedFailure, None, "")]}
}
result = apply_patches(code, patches)
self.assertIn("@some_decorator", result)
self.assertIn("@unittest.expectedFailure", result)
def test_apply_inherited_method(self):
"""Test applying patch to inherited method (creates override)."""
code = """import unittest
class TestFoo(unittest.TestCase):
pass
"""
patches = {
"TestFoo": {
"test_inherited": [PatchSpec(UtMethod.ExpectedFailure, None, "")]
}
}
result = apply_patches(code, patches)
self.assertIn("def test_inherited(self):", result)
self.assertIn("return super().test_inherited()", result)
def test_apply_adds_unittest_import(self):
"""Test that unittest import is added if missing."""
code = """import sys
class TestFoo:
def test_one(self):
pass
"""
patches = {
"TestFoo": {"test_one": [PatchSpec(UtMethod.ExpectedFailure, None, "")]}
}
result = apply_patches(code, patches)
# Should add unittest import after existing imports
self.assertIn("import unittest", result)
def test_apply_no_duplicate_import(self):
"""Test that unittest import is not duplicated."""
code = """import unittest
class TestFoo(unittest.TestCase):
def test_one(self):
pass
"""
patches = {
"TestFoo": {"test_one": [PatchSpec(UtMethod.ExpectedFailure, None, "")]}
}
result = apply_patches(code, patches)
# Count occurrences of 'import unittest'
count = result.count("import unittest")
self.assertEqual(count, 1)
class TestPatchSpec(unittest.TestCase):
"""Tests for PatchSpec class."""
def test_as_decorator_expected_failure(self):
"""Test generating expectedFailure decorator string."""
spec = PatchSpec(UtMethod.ExpectedFailure, None, "reason")
decorator = spec.as_decorator()
self.assertIn("@unittest.expectedFailure", decorator)
self.assertIn(COMMENT, decorator)
self.assertIn("reason", decorator)
def test_as_decorator_skip(self):
"""Test generating skip decorator string."""
spec = PatchSpec(UtMethod.Skip, None, "not ready")
decorator = spec.as_decorator()
self.assertIn("@unittest.skip", decorator)
self.assertIn("not ready", decorator)
def test_as_decorator_skip_if(self):
"""Test generating skipIf decorator string."""
spec = PatchSpec(UtMethod.SkipIf, "condition", "reason")
decorator = spec.as_decorator()
self.assertIn("@unittest.skipIf", decorator)
self.assertIn("condition", decorator)
class TestRoundTrip(unittest.TestCase):
"""Tests for extract -> apply round trip."""
def test_round_trip_expected_failure(self):
"""Test that extracted patches can be re-applied."""
original = f"""import unittest
class TestFoo(unittest.TestCase):
# {COMMENT}
@unittest.expectedFailure
def test_one(self):
pass
"""
# Extract patches
patches = extract_patches(original)
# Apply to clean code
clean = """import unittest
class TestFoo(unittest.TestCase):
def test_one(self):
pass
"""
result = apply_patches(clean, patches)
# Should have the decorator
self.assertIn("@unittest.expectedFailure", result)
self.assertIn(COMMENT, result)
if __name__ == "__main__":
unittest.main()
scripts/update_lib/tests/test_path.py
+201
View File
@@ -0,0 +1,201 @@
"""Tests for path.py - path utilities."""
import pathlib
import tempfile
import unittest
from update_lib.path import (
get_test_files,
is_lib_path,
is_test_path,
lib_to_test_path,
parse_lib_path,
test_name_from_path,
)
class TestParseLibPath(unittest.TestCase):
"""Tests for parse_lib_path function."""
def test_parse_cpython_path(self):
"""Test parsing cpython/Lib/... path."""
result = parse_lib_path("cpython/Lib/test/test_foo.py")
self.assertEqual(result, pathlib.Path("Lib/test/test_foo.py"))
def test_parse_nested_path(self):
"""Test parsing deeply nested path."""
result = parse_lib_path("/home/user/cpython/Lib/test/test_foo/test_bar.py")
self.assertEqual(result, pathlib.Path("Lib/test/test_foo/test_bar.py"))
def test_parse_windows_path(self):
"""Test parsing Windows-style path."""
result = parse_lib_path("C:\\cpython\\Lib\\test\\test_foo.py")
self.assertEqual(result, pathlib.Path("Lib/test/test_foo.py"))
def test_parse_directory(self):
"""Test parsing directory path."""
result = parse_lib_path("cpython/Lib/test/test_json/")
self.assertEqual(result, pathlib.Path("Lib/test/test_json/"))
def test_parse_no_lib_raises(self):
"""Test that path without /Lib/ raises ValueError."""
with self.assertRaises(ValueError) as ctx:
parse_lib_path("some/random/path.py")
self.assertIn("/Lib/", str(ctx.exception))
class TestIsLibPath(unittest.TestCase):
"""Tests for is_lib_path function."""
def test_lib_path(self):
"""Test detecting Lib/ path."""
self.assertTrue(is_lib_path(pathlib.Path("Lib/test/test_foo.py")))
self.assertTrue(is_lib_path(pathlib.Path("./Lib/test/test_foo.py")))
def test_cpython_path_not_lib(self):
"""Test that cpython/Lib/ is not detected as lib path."""
self.assertFalse(is_lib_path(pathlib.Path("cpython/Lib/test/test_foo.py")))
def test_random_path_not_lib(self):
"""Test that random path is not lib path."""
self.assertFalse(is_lib_path(pathlib.Path("some/other/path.py")))
class TestIsTestPath(unittest.TestCase):
"""Tests for is_test_path function."""
def test_cpython_test_path(self):
"""Test detecting cpython test path."""
self.assertTrue(is_test_path(pathlib.Path("cpython/Lib/test/test_foo.py")))
def test_lib_test_path(self):
"""Test detecting Lib/test path."""
self.assertTrue(is_test_path(pathlib.Path("Lib/test/test_foo.py")))
def test_library_path_not_test(self):
"""Test that library path (not test) is not test path."""
self.assertFalse(is_test_path(pathlib.Path("cpython/Lib/dataclasses.py")))
self.assertFalse(is_test_path(pathlib.Path("Lib/dataclasses.py")))
class TestLibToTestPath(unittest.TestCase):
"""Tests for lib_to_test_path function."""
def test_prefers_directory_over_file(self):
"""Test that directory is preferred when both exist."""
with tempfile.TemporaryDirectory() as tmpdir:
tmpdir = pathlib.Path(tmpdir)
# Create structure: tmpdir/Lib/foo.py, tmpdir/Lib/test/test_foo/, tmpdir/Lib/test/test_foo.py
lib_dir = tmpdir / "Lib"
lib_dir.mkdir()
(lib_dir / "foo.py").write_text("# lib")
test_dir = lib_dir / "test"
test_dir.mkdir()
(test_dir / "test_foo").mkdir()
(test_dir / "test_foo.py").write_text("# test file")
result = lib_to_test_path(tmpdir / "Lib" / "foo.py")
# Should prefer directory
self.assertEqual(result, tmpdir / "Lib" / "test" / "test_foo/")
def test_falls_back_to_file(self):
"""Test that file is used when directory doesn't exist."""
with tempfile.TemporaryDirectory() as tmpdir:
tmpdir = pathlib.Path(tmpdir)
# Create structure: tmpdir/Lib/foo.py, tmpdir/Lib/test/test_foo.py (no directory)
lib_dir = tmpdir / "Lib"
lib_dir.mkdir()
(lib_dir / "foo.py").write_text("# lib")
test_dir = lib_dir / "test"
test_dir.mkdir()
(test_dir / "test_foo.py").write_text("# test file")
result = lib_to_test_path(tmpdir / "Lib" / "foo.py")
# Should fall back to file
self.assertEqual(result, tmpdir / "Lib" / "test" / "test_foo.py")
def test_defaults_to_directory_when_neither_exists(self):
"""Test that directory path is returned when neither exists."""
with tempfile.TemporaryDirectory() as tmpdir:
tmpdir = pathlib.Path(tmpdir)
lib_dir = tmpdir / "Lib"
lib_dir.mkdir()
(lib_dir / "foo.py").write_text("# lib")
test_dir = lib_dir / "test"
test_dir.mkdir()
# Neither test_foo/ nor test_foo.py exists
result = lib_to_test_path(tmpdir / "Lib" / "foo.py")
# Should default to directory
self.assertEqual(result, tmpdir / "Lib" / "test" / "test_foo/")
def test_lib_path_prefers_directory(self):
"""Test Lib/ path prefers directory when it exists."""
# This test uses actual Lib/ paths, checking current behavior
# When neither exists, defaults to directory
result = lib_to_test_path(pathlib.Path("Lib/nonexistent_module.py"))
self.assertEqual(result, pathlib.Path("Lib/test/test_nonexistent_module/"))
class TestGetTestFiles(unittest.TestCase):
"""Tests for get_test_files function."""
def test_single_file(self):
"""Test getting single file."""
with tempfile.TemporaryDirectory() as tmpdir:
tmpdir = pathlib.Path(tmpdir)
test_file = tmpdir / "test.py"
test_file.write_text("# test")
files = get_test_files(test_file)
self.assertEqual(len(files), 1)
self.assertEqual(files[0], test_file)
def test_directory(self):
"""Test getting all .py files from directory."""
with tempfile.TemporaryDirectory() as tmpdir:
tmpdir = pathlib.Path(tmpdir)
(tmpdir / "test_a.py").write_text("# a")
(tmpdir / "test_b.py").write_text("# b")
(tmpdir / "not_python.txt").write_text("# not python")
files = get_test_files(tmpdir)
self.assertEqual(len(files), 2)
names = [f.name for f in files]
self.assertIn("test_a.py", names)
self.assertIn("test_b.py", names)
def test_nested_directory(self):
"""Test getting .py files from nested directory."""
with tempfile.TemporaryDirectory() as tmpdir:
tmpdir = pathlib.Path(tmpdir)
(tmpdir / "test_a.py").write_text("# a")
subdir = tmpdir / "subdir"
subdir.mkdir()
(subdir / "test_b.py").write_text("# b")
files = get_test_files(tmpdir)
self.assertEqual(len(files), 2)
class TestTestNameFromPath(unittest.TestCase):
"""Tests for test_name_from_path function."""
def test_simple_test_file(self):
"""Test extracting name from simple test file."""
path = pathlib.Path("Lib/test/test_foo.py")
self.assertEqual(test_name_from_path(path), "test_foo")
def test_nested_test_file(self):
"""Test extracting name from nested test directory."""
path = pathlib.Path("Lib/test/test_ctypes/test_bar.py")
self.assertEqual(test_name_from_path(path), "test_ctypes.test_bar")
def test_test_directory(self):
"""Test extracting name from test directory."""
path = pathlib.Path("Lib/test/test_json")
self.assertEqual(test_name_from_path(path), "test_json")
if __name__ == "__main__":
unittest.main()
scripts/update_lib/tests/test_quick.py
+180
View File
@@ -0,0 +1,180 @@
"""Tests for quick.py - quick update functionality."""
import pathlib
import tempfile
import unittest
from unittest.mock import patch
from update_lib.path import lib_to_test_path
from update_lib.quick import _expand_shortcut, collect_original_methods, get_cpython_dir, git_commit
class TestGetCpythonDir(unittest.TestCase):
"""Tests for get_cpython_dir function."""
def test_extract_from_full_path(self):
"""Test extracting cpython dir from full path."""
path = pathlib.Path("cpython/Lib/dataclasses.py")
result = get_cpython_dir(path)
self.assertEqual(result, pathlib.Path("cpython"))
def test_extract_from_absolute_path(self):
"""Test extracting cpython dir from absolute path."""
path = pathlib.Path("/some/path/cpython/Lib/test/test_foo.py")
result = get_cpython_dir(path)
self.assertEqual(result, pathlib.Path("/some/path/cpython"))
def test_shortcut_defaults_to_cpython(self):
"""Test that shortcut (no /Lib/) defaults to 'cpython'."""
path = pathlib.Path("dataclasses")
result = get_cpython_dir(path)
self.assertEqual(result, pathlib.Path("cpython"))
class TestExpandShortcut(unittest.TestCase):
"""Tests for _expand_shortcut function."""
def test_expand_shortcut_to_test_path_integration(self):
"""Test that expanded shortcut works with lib_to_test_path.
This tests the fix for the bug where args.path was used instead of
the expanded src_path when calling lib_to_test_path.
"""
# Simulate the flow in main():
# 1. User provides "dataclasses"
# 2. _expand_shortcut converts to "cpython/Lib/dataclasses.py"
# 3. lib_to_test_path should receive the expanded path, not original
original_path = pathlib.Path("dataclasses")
expanded_path = _expand_shortcut(original_path)
# If cpython/Lib/dataclasses.py exists, it should be expanded
if expanded_path != original_path:
# The expanded path should work with lib_to_test_path
test_path = lib_to_test_path(expanded_path)
# Should return a valid test path, not raise an error
self.assertTrue(str(test_path).startswith("cpython/Lib/test/"))
# The original unexpanded path would fail or give wrong result
# This is what the bug was - using args.path instead of src_path
def test_expand_shortcut_file(self):
"""Test expanding a simple name to file path."""
# This test checks the shortcut works when file exists
path = pathlib.Path("dataclasses")
result = _expand_shortcut(path)
expected_file = pathlib.Path("cpython/Lib/dataclasses.py")
expected_dir = pathlib.Path("cpython/Lib/dataclasses")
if expected_file.exists():
self.assertEqual(result, expected_file)
elif expected_dir.exists():
self.assertEqual(result, expected_dir)
else:
# If neither exists, should return original
self.assertEqual(result, path)
def test_expand_shortcut_already_full_path(self):
"""Test that full paths are not modified."""
path = pathlib.Path("cpython/Lib/dataclasses.py")
result = _expand_shortcut(path)
self.assertEqual(result, path)
def test_expand_shortcut_nonexistent(self):
"""Test that nonexistent names are returned as-is."""
path = pathlib.Path("nonexistent_module_xyz")
result = _expand_shortcut(path)
self.assertEqual(result, path)
class TestCollectOriginalMethods(unittest.TestCase):
"""Tests for collect_original_methods function."""
def test_collect_from_file(self):
"""Test collecting methods from single file."""
with tempfile.TemporaryDirectory() as tmpdir:
tmpdir = pathlib.Path(tmpdir)
test_file = tmpdir / "test.py"
test_file.write_text("""
class TestFoo:
def test_one(self):
pass
def test_two(self):
pass
""")
methods = collect_original_methods(test_file)
self.assertIsInstance(methods, set)
self.assertEqual(len(methods), 2)
self.assertIn(("TestFoo", "test_one"), methods)
self.assertIn(("TestFoo", "test_two"), methods)
def test_collect_from_directory(self):
"""Test collecting methods from directory."""
with tempfile.TemporaryDirectory() as tmpdir:
tmpdir = pathlib.Path(tmpdir)
(tmpdir / "test_a.py").write_text("""
class TestA:
def test_a(self):
pass
""")
(tmpdir / "test_b.py").write_text("""
class TestB:
def test_b(self):
pass
""")
methods = collect_original_methods(tmpdir)
self.assertIsInstance(methods, dict)
self.assertEqual(len(methods), 2)
class TestGitCommit(unittest.TestCase):
"""Tests for git_commit function."""
@patch("subprocess.run")
@patch("update_lib.quick.get_cpython_version")
def test_none_lib_path_not_added(self, mock_version, mock_run):
"""Test that None lib_path doesn't add '.' to git."""
mock_version.return_value = "v3.14.0"
mock_run.return_value.returncode = 1 # Has changes
with tempfile.TemporaryDirectory() as tmpdir:
test_file = pathlib.Path(tmpdir) / "test.py"
test_file.write_text("# test")
git_commit("test", None, test_file, pathlib.Path("cpython"), verbose=False)
# Check git add was called with only test_file, not "."
add_call = mock_run.call_args_list[0]
self.assertIn(str(test_file), add_call[0][0])
self.assertNotIn(".", add_call[0][0][2:]) # Skip "git" and "add"
@patch("subprocess.run")
@patch("update_lib.quick.get_cpython_version")
def test_none_test_path_not_added(self, mock_version, mock_run):
"""Test that None test_path doesn't add '.' to git."""
mock_version.return_value = "v3.14.0"
mock_run.return_value.returncode = 1
with tempfile.TemporaryDirectory() as tmpdir:
lib_file = pathlib.Path(tmpdir) / "lib.py"
lib_file.write_text("# lib")
git_commit("lib", lib_file, None, pathlib.Path("cpython"), verbose=False)
add_call = mock_run.call_args_list[0]
self.assertIn(str(lib_file), add_call[0][0])
self.assertNotIn(".", add_call[0][0][2:])
def test_both_none_returns_false(self):
"""Test that both paths None returns False without git operations."""
# No mocking needed - should return early before any subprocess calls
result = git_commit("test", None, None, pathlib.Path("cpython"), verbose=False)
self.assertFalse(result)
if __name__ == "__main__":
unittest.main()