From d119c47d2ec3444c81137fa493f0e5b0e59d315e Mon Sep 17 00:00:00 2001 From: "Jeong, YunWon" Date: Mon, 19 Jan 2026 10:19:47 +0900 Subject: [PATCH] Remake update_lib --- .claude/commands/upgrade-pylib.md | 49 +- scripts/auto_mark_test.py | 411 ---------- scripts/update_lib/__init__.py | 37 + scripts/update_lib/__main__.py | 84 +++ scripts/update_lib/auto_mark.py | 704 ++++++++++++++++++ scripts/update_lib/copy_lib.py | 82 ++ scripts/update_lib/migrate.py | 142 ++++ .../patch_spec.py} | 237 ++---- scripts/update_lib/patches.py | 106 +++ scripts/update_lib/path.py | 99 +++ scripts/update_lib/quick.py | 365 +++++++++ scripts/update_lib/tests/__init__.py | 0 scripts/update_lib/tests/test_auto_mark.py | 593 +++++++++++++++ scripts/update_lib/tests/test_migrate.py | 196 +++++ scripts/update_lib/tests/test_patch_spec.py | 325 ++++++++ scripts/update_lib/tests/test_path.py | 201 +++++ scripts/update_lib/tests/test_quick.py | 180 +++++ 17 files changed, 3230 insertions(+), 581 deletions(-) delete mode 100644 scripts/auto_mark_test.py create mode 100644 scripts/update_lib/__init__.py create mode 100644 scripts/update_lib/__main__.py create mode 100644 scripts/update_lib/auto_mark.py create mode 100644 scripts/update_lib/copy_lib.py create mode 100644 scripts/update_lib/migrate.py rename scripts/{lib_updater.py => update_lib/patch_spec.py} (60%) mode change 100755 => 100644 create mode 100644 scripts/update_lib/patches.py create mode 100644 scripts/update_lib/path.py create mode 100644 scripts/update_lib/quick.py create mode 100644 scripts/update_lib/tests/__init__.py create mode 100644 scripts/update_lib/tests/test_auto_mark.py create mode 100644 scripts/update_lib/tests/test_migrate.py create mode 100644 scripts/update_lib/tests/test_patch_spec.py create mode 100644 scripts/update_lib/tests/test_path.py create mode 100644 scripts/update_lib/tests/test_quick.py diff --git a/.claude/commands/upgrade-pylib.md b/.claude/commands/upgrade-pylib.md index 02ec7c0d3..67603ce2d 100644 --- a/.claude/commands/upgrade-pylib.md +++ b/.claude/commands/upgrade-pylib.md @@ -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 diff --git a/scripts/auto_mark_test.py b/scripts/auto_mark_test.py deleted file mode 100644 index 59c7d8a05..000000000 --- a/scripts/auto_mark_test.py +++ /dev/null @@ -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") diff --git a/scripts/update_lib/__init__.py b/scripts/update_lib/__init__.py new file mode 100644 index 000000000..ccb2628d6 --- /dev/null +++ b/scripts/update_lib/__init__.py @@ -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", +] diff --git a/scripts/update_lib/__main__.py b/scripts/update_lib/__main__.py new file mode 100644 index 000000000..f20f41c84 --- /dev/null +++ b/scripts/update_lib/__main__.py @@ -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()) diff --git a/scripts/update_lib/auto_mark.py b/scripts/update_lib/auto_mark.py new file mode 100644 index 000000000..a716acea5 --- /dev/null +++ b/scripts/update_lib/auto_mark.py @@ -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()) diff --git a/scripts/update_lib/copy_lib.py b/scripts/update_lib/copy_lib.py new file mode 100644 index 000000000..5af0cab85 --- /dev/null +++ b/scripts/update_lib/copy_lib.py @@ -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()) diff --git a/scripts/update_lib/migrate.py b/scripts/update_lib/migrate.py new file mode 100644 index 000000000..d0f049ec1 --- /dev/null +++ b/scripts/update_lib/migrate.py @@ -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()) diff --git a/scripts/lib_updater.py b/scripts/update_lib/patch_spec.py old mode 100755 new mode 100644 similarity index 60% rename from scripts/lib_updater.py rename to scripts/update_lib/patch_spec.py index 712ea7bc8..1f88c06c1 --- a/scripts/lib_updater.py +++ b/scripts/update_lib/patch_spec.py @@ -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() + } diff --git a/scripts/update_lib/patches.py b/scripts/update_lib/patches.py new file mode 100644 index 000000000..67ebf1822 --- /dev/null +++ b/scripts/update_lib/patches.py @@ -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()) diff --git a/scripts/update_lib/path.py b/scripts/update_lib/path.py new file mode 100644 index 000000000..7a3535fd2 --- /dev/null +++ b/scripts/update_lib/path.py @@ -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 diff --git a/scripts/update_lib/quick.py b/scripts/update_lib/quick.py new file mode 100644 index 000000000..6221fcbc6 --- /dev/null +++ b/scripts/update_lib/quick.py @@ -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()) diff --git a/scripts/update_lib/tests/__init__.py b/scripts/update_lib/tests/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/scripts/update_lib/tests/test_auto_mark.py b/scripts/update_lib/tests/test_auto_mark.py new file mode 100644 index 000000000..ac357f5ea --- /dev/null +++ b/scripts/update_lib/tests/test_auto_mark.py @@ -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() diff --git a/scripts/update_lib/tests/test_migrate.py b/scripts/update_lib/tests/test_migrate.py new file mode 100644 index 000000000..ff93052f6 --- /dev/null +++ b/scripts/update_lib/tests/test_migrate.py @@ -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() diff --git a/scripts/update_lib/tests/test_patch_spec.py b/scripts/update_lib/tests/test_patch_spec.py new file mode 100644 index 000000000..ec52aa6d2 --- /dev/null +++ b/scripts/update_lib/tests/test_patch_spec.py @@ -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() diff --git a/scripts/update_lib/tests/test_path.py b/scripts/update_lib/tests/test_path.py new file mode 100644 index 000000000..06d10e2f1 --- /dev/null +++ b/scripts/update_lib/tests/test_path.py @@ -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() diff --git a/scripts/update_lib/tests/test_quick.py b/scripts/update_lib/tests/test_quick.py new file mode 100644 index 000000000..43f829a96 --- /dev/null +++ b/scripts/update_lib/tests/test_quick.py @@ -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()