diff --git a/scripts/update_lib/cmd_migrate.py b/scripts/update_lib/cmd_migrate.py index 97cdf7b141b..5e4c2656a9c 100644 --- a/scripts/update_lib/cmd_migrate.py +++ b/scripts/update_lib/cmd_migrate.py @@ -4,6 +4,13 @@ Usage: python scripts/update_lib migrate cpython/Lib/test/test_foo.py + python scripts/update_lib migrate cpython/Lib/test/test_foo.py --patches remaining.json + +Remaining metadata names the exact repository-relative target and its existing +decorator patches. An optional "doctests" object selects existing guarded inputs +by exact source text, mapping each to a single-line reason. Guarded doctests need +this explicit selection; an empty object retires all their guards. Canonical +expected output is never copied from the old file. This will: 1. Extract patches from Lib/test/test_foo.py (if exists) @@ -12,6 +19,9 @@ """ import argparse +import ast +import collections +import json import pathlib import shutil import sys @@ -20,10 +30,166 @@ from update_lib.file_utils import parse_lib_path +SYSCONFIG_UPSTREAM = "def _get_implementation():\n return 'Python'\n" +SYSCONFIG_LOCAL = ( + "def _get_implementation():\n" + " return 'RustPython' # XXX: RUSTPYTHON; For site-packages\n" +) +JIT_ERROR = """# NOTE: RUSTPYTHON exceptions +try: + JitError +except NameError: + pass +else: + PYTHON2_EXCEPTIONS += ("JitError",) + +""" +EXCEPTION_LOOP = """for excname in PYTHON2_EXCEPTIONS: + NAME_MAPPING[("exceptions", excname)] = ("builtins", excname) +""" + + +def _unique_anchor(contents, root, predicate, *, direct=False, exact=None): + """Select one real AST anchor, optionally in the immediate owner body.""" + matches = [node for node in ast.walk(root) if predicate(node)] + if ( + len(matches) != 1 + or (direct and matches[0] not in root.body) + or (exact is not None and ast.get_source_segment(contents, matches[0]) != exact) + ): + raise ValueError("Missing, changed, or ambiguous migration anchor") + return matches[0] + + +def _unique_function(contents: str, expected: str) -> bool: + tree = ast.parse(contents) + target = ast.parse(expected).body[0] + node = _unique_anchor( + contents, + tree, + lambda node: ( + isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) + and node.name == target.name + ), + direct=True, + ) + return ast.dump(node) == ast.dump(target) and not any( + ( + isinstance(node, ast.Name) + and node.id == target.name + and isinstance(node.ctx, (ast.Store, ast.Del)) + ) + or (isinstance(node, ast.ClassDef) and node.name == target.name) + or ( + isinstance(node, ast.alias) + and (node.asname or node.name.split(".")[0]) == target.name + ) + for node in ast.walk(tree) + ) + + +def _exception_loop(contents: str) -> bool: + _unique_anchor( + contents, + ast.parse(contents), + lambda node: ( + isinstance(node, ast.For) + and isinstance(node.iter, ast.Name) + and node.iter.id == "PYTHON2_EXCEPTIONS" + ), + direct=True, + exact=EXCEPTION_LOOP.rstrip(), + ) + return True + + +def _jit_block(contents: str) -> bool: + tree = ast.parse(contents) + expected = ast.dump(ast.parse(JIT_ERROR).body[0]) + _unique_anchor(contents, tree, lambda node: ast.dump(node) == expected, direct=True) + return ( + sum( + isinstance(node, ast.Name) and node.id == "JitError" + for node in ast.walk(tree) + ) + == 1 + ) + + +def preserve_migration_customizations(path: pathlib.Path, old: str, new: str) -> str: + """Fail closed if either known local patch or its canonical anchor changes.""" + parts = path.parts + if path.is_absolute(): + raise ValueError("Runtime patch target must be repository-relative") + if parts == ("Lib", "sysconfig", "__init__.py"): + if old.count(SYSCONFIG_LOCAL) != 1 or not _unique_function( + old, SYSCONFIG_LOCAL + ): + raise ValueError("Unrecognized sysconfig implementation identity") + if ( + new.count(SYSCONFIG_UPSTREAM) != 1 + or not _unique_function(new, SYSCONFIG_UPSTREAM) + or "RustPython" in new + ): + raise ValueError("Unrecognized canonical sysconfig implementation anchor") + return new.replace(SYSCONFIG_UPSTREAM, SYSCONFIG_LOCAL, 1) + if parts == ("Lib", "_compat_pickle.py"): + if ( + old.count(JIT_ERROR + EXCEPTION_LOOP) != 1 + or not _jit_block(old) + or not _exception_loop(old) + ): + raise ValueError("Unrecognized local JitError compatibility mapping") + if ( + new.count(EXCEPTION_LOOP) != 1 + or not _exception_loop(new) + or "JitError" in new + ): + raise ValueError("Unrecognized canonical exception mapping anchor") + return new.replace(EXCEPTION_LOOP, JIT_ERROR + EXCEPTION_LOOP, 1) + if parts == ("Lib", "test", "test_cmd_line.py"): + _verify_call(old, LOCAL_CALL) + _verify_call(new, CANONICAL_CALL) + return new.replace(CANONICAL_CALL, LOCAL_CALL, 1) + return new + + +CANONICAL_CALL = """assert_python_ok('-c', code, PYTHONPATH="")""" +LOCAL_CALL = """assert_python_ok('-c', code, PYTHONPATH="", RUSTPYTHONPATH="")""" + + +def _verify_call(contents: str, expected: str) -> None: + tree = ast.parse(contents) + owner = _unique_anchor( + contents, + tree, + lambda node: isinstance(node, ast.ClassDef) and node.name == "CmdLineTest", + direct=True, + ) + method = _unique_anchor( + contents, + owner, + lambda node: ( + isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) + and node.name == "test_empty_PYTHONPATH_issue16309" + ), + direct=True, + ) + target = ast.dump(ast.parse(expected, mode="eval").body) + _unique_anchor( + contents, + method, + lambda node: isinstance(node, ast.Call) and ast.dump(node) == target, + exact=expected, + ) + if contents.count(expected) != 1: + raise ValueError("Ambiguous environment-reset call text") + def patch_single_content( src_path: pathlib.Path, lib_path: pathlib.Path, + patches_document: dict | None = None, ) -> str: """ Patch content without writing to disk. @@ -36,15 +202,40 @@ def patch_single_content( The patched content. """ from update_lib import apply_patches, extract_patches + from update_lib.patch_spec import apply_doctest_metadata, extract_doctest_guards # Extract patches from existing file (if exists) if lib_path.exists(): - patches = extract_patches(lib_path.read_text(encoding="utf-8")) + old_content = lib_path.read_text(encoding="utf-8") + patches = extract_patches(old_content) else: patches = {} + repository = pathlib.Path(__file__).resolve().parents[2] + try: + relative_path = lib_path.resolve().relative_to(repository) + except ValueError: + # Explicit output paths outside this repository are not its Lib. + relative_path = pathlib.Path("__external_output__") + if patches_document is not None: + from update_lib.patch_spec import select_remaining_patches + + patches = select_remaining_patches(patches, patches_document, relative_path) + # Apply patches to source content src_content = src_path.read_text(encoding="utf-8") + if lib_path.exists(): + src_content = preserve_migration_customizations( + relative_path, old_content, src_content + ) + guards = extract_doctest_guards(old_content) + selection = patches_document.get("doctests") if patches_document else None + if guards and selection is None: + raise ValueError( + "Existing doctest guards require explicit doctests metadata" + ) + if selection is not None: + src_content = apply_doctest_metadata(old_content, src_content, selection) return apply_patches(src_content, patches) @@ -52,6 +243,7 @@ def patch_file( src_path: pathlib.Path, lib_path: pathlib.Path | None = None, verbose: bool = True, + patches_document: dict | None = None, ) -> None: """ Patch a single file from source to lib. @@ -67,10 +259,12 @@ def patch_file( if lib_path.exists(): if verbose: print(f"Patching: {src_path} -> {lib_path}") - content = patch_single_content(src_path, lib_path) + content = patch_single_content(src_path, lib_path, patches_document) else: if verbose: print(f"Copying: {src_path} -> {lib_path}") + if patches_document is not None: + raise ValueError("Marker override requires an existing target") content = src_path.read_text(encoding="utf-8") lib_path.parent.mkdir(parents=True, exist_ok=True) @@ -93,7 +287,11 @@ def patch_directory( if lib_dir is None: lib_dir = parse_lib_path(src_dir) - src_files = sorted(f for f in src_dir.glob("**/*") if f.is_file()) + src_files = sorted( + f + for f in src_dir.glob("**/*") + if f.is_file() and "__pycache__" not in f.relative_to(src_dir).parts[:-1] + ) for src_file in src_files: rel_path = src_file.relative_to(src_dir) @@ -118,6 +316,153 @@ def patch_directory( shutil.copy2(src_file, lib_file) +def migrate_package( + src_dir: pathlib.Path, lib_dir: pathlib.Path, mapping_path: pathlib.Path +) -> None: + """Validate the complete migration before writing or removing legacy files. + + The JSON map names legacy files beside lib_dir, mapping every marked class + to its canonical file relative to src_dir. Empty maps retire unmarked files. + No test modules are imported or executed. + """ + from update_lib.patch_spec import ( + _iter_class_methods, + _iter_module_classes, + apply_patches, + build_patch_dict, + extract_doctest_guards, + iter_patches, + ) + + mapping = json.loads(mapping_path.read_text(encoding="utf-8")) + if not isinstance(mapping, dict) or not mapping: + raise ValueError("Legacy mapping must be a nonempty object") + if lib_dir.exists(): + raise ValueError(f"Destination already exists: {lib_dir}") + if ( + src_dir.resolve() == lib_dir.resolve() + or src_dir.resolve() in lib_dir.resolve().parents + ): + raise ValueError("Source and destination overlap") + source_files = sorted( + p + for p in src_dir.rglob("*") + if p.is_file() and "__pycache__" not in p.relative_to(src_dir).parts[:-1] + ) + if not (src_dir / "__init__.py").is_file(): + raise ValueError("Source must be a Python package") + if any( + p.is_symlink() or src_dir.resolve() not in p.resolve().parents + for p in source_files + ): + raise ValueError("Source files must remain inside the canonical package") + contents = {p.relative_to(src_dir): p.read_bytes() for p in source_files} + entries = collections.defaultdict(list) + owners = {} + legacy_files = [] + for name, class_map in mapping.items(): + if pathlib.Path(name).name != name or not name.endswith(".py"): + raise ValueError(f"Legacy file must be a sibling .py file: {name}") + if not isinstance(class_map, dict): + raise ValueError(f"Class map must be an object: {name}") + legacy = lib_dir.parent / name + if legacy.is_symlink() or legacy.resolve().parent != lib_dir.parent.resolve(): + raise ValueError(f"Legacy file escapes destination parent: {name}") + if ( + legacy.resolve() == lib_dir.resolve() + or src_dir.resolve() in legacy.resolve().parents + ): + raise ValueError(f"Legacy file overlaps canonical package: {name}") + legacy_files.append(legacy) + legacy_content = legacy.read_text(encoding="utf-8") + if extract_doctest_guards(legacy_content): + raise ValueError("Legacy package migration cannot select doctest guards") + patches = list(iter_patches(legacy_content)) + legacy_classes = collections.defaultdict(list) + for cls in _iter_module_classes(ast.parse(legacy_content)): + legacy_classes[cls.name].append(cls) + if set(class_map) != {p.parent_class for p in patches}: + raise ValueError(f"Missing or unused class mapping: {name}") + for entry in patches: + source_classes = legacy_classes[entry.parent_class] + if len(source_classes) != 1: + raise ValueError(f"Ambiguous legacy class: {entry.parent_class}") + if ( + entry.test_name != "__self__" + and sum( + method.name == entry.test_name + for method in _iter_class_methods(source_classes[0]) + ) + != 1 + ): + raise ValueError( + f"Ambiguous legacy method: {entry.parent_class}.{entry.test_name}" + ) + target = pathlib.Path(class_map[entry.parent_class]) + if target.is_absolute() or ".." in target.parts or target not in contents: + raise ValueError(f"Invalid canonical destination: {target}") + if target.suffix != ".py": + raise ValueError(f"Marker destination is not Python: {target}") + tree = ast.parse(contents[target].decode("utf-8")) + classes = collections.defaultdict(list) + for cls in _iter_module_classes(tree): + classes[cls.name].append(cls) + + def has_method(class_name, seen): + if class_name in seen: + return False + seen = seen | {class_name} + nodes = classes.get(class_name, []) + if len(nodes) != 1: + raise ValueError(f"Missing or ambiguous class: {class_name}") + cls = nodes[0] + methods = [ + m for m in _iter_class_methods(cls) if m.name == entry.test_name + ] + if len(methods) > 1: + raise ValueError( + f"Ambiguous method: {class_name}.{entry.test_name}" + ) + if methods or entry.test_name == "__self__": + return True + return any( + has_method(base.id, seen) + for base in cls.bases + if isinstance(base, ast.Name) and base.id in classes + ) + + if not has_method(entry.parent_class, set()): + raise ValueError( + f"Missing method: {entry.parent_class}.{entry.test_name}" + ) + owner_key = (target, entry.parent_class, entry.test_name) + if owner_key in owners and owners[owner_key] != legacy: + raise ValueError( + f"Duplicate mapped marker: {entry.parent_class}.{entry.test_name}" + ) + owners[owner_key] = legacy + entries[target].append(entry) + # Render and validate everything before changing the destination. + for target, patches in entries.items(): + contents[target] = apply_patches( + contents[target].decode("utf-8"), build_patch_dict(iter(patches)) + ).encode("utf-8") + for target, data in contents.items(): + if target.suffix == ".py": + text = data.decode("utf-8") + ast.parse(text) + if collections.Counter(iter_patches(text)) != collections.Counter( + entries[target] + ): + raise ValueError(f"Marker roundtrip mismatch: {target}") + for target, data in contents.items(): + destination = lib_dir / target + destination.parent.mkdir(parents=True, exist_ok=True) + destination.write_bytes(data) + for legacy in legacy_files: + legacy.unlink() + + def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser( description=__doc__, @@ -129,13 +474,33 @@ def main(argv: list[str] | None = None) -> int: help="Source path containing /Lib/ (file or directory)", ) + parser.add_argument( + "--legacy-map", + type=pathlib.Path, + help="Explicit legacy-file/class to package-file JSON mapping", + ) + + parser.add_argument( + "--patches", + type=pathlib.Path, + help="Single-file target, remaining decorator patches, and optional doctests JSON", + ) args = parser.parse_args(argv) try: - if args.path.is_dir(): + if args.patches is not None and ( + args.path.is_dir() or args.legacy_map is not None + ): + raise ValueError("--patches requires a single file without --legacy-map") + document = ( + json.loads(args.patches.read_text()) if args.patches is not None else None + ) + if args.legacy_map is not None: + migrate_package(args.path, parse_lib_path(args.path), args.legacy_map) + elif args.path.is_dir(): patch_directory(args.path) else: - patch_file(args.path) + patch_file(args.path, patches_document=document) return 0 except ValueError as e: print(f"Error: {e}", file=sys.stderr) diff --git a/scripts/update_lib/patch_spec.py b/scripts/update_lib/patch_spec.py index 1483c56fc56..4dee842a8f7 100644 --- a/scripts/update_lib/patch_spec.py +++ b/scripts/update_lib/patch_spec.py @@ -11,10 +11,12 @@ import ast import collections import contextlib +import doctest import enum import re import textwrap import typing +from pathlib import Path if typing.TYPE_CHECKING: from collections.abc import Iterator @@ -121,7 +123,9 @@ def try_from_ast_node( # If our ut_method has args then, # we need to search for a constant that contains our `COMMENT`. # Otherwise we need to search it in the raw source code :/ - if ut_method.has_args(): + if ut_method.has_args() and not ( + ut_method is UtMethod.Skip and isinstance(node, ast.Attribute) + ): reason = next( ( inner_node.value @@ -303,8 +307,23 @@ def _iter_patch_lines( class_bases: dict[str, list[str]] = {} all_classes = set() all_class_nodes = list(_iter_module_classes(tree)) + # Only names used to place a patch (including inherited methods) need a + # unique definition. Conditional, unpatched helpers are common upstream. + relevant_classes = set(patches) + while True: + ancestors = { + base.id + for node in all_class_nodes + if node.name in relevant_classes + for base in node.bases + if isinstance(base, ast.Name) + } + expanded = relevant_classes | ancestors + if expanded == relevant_classes: + break + relevant_classes = expanded for node in all_class_nodes: - if node.name in all_classes: + if node.name in all_classes and node.name in relevant_classes: raise ValueError(f"Ambiguous module-scope class name: {node.name}") all_classes.add(node.name) method_counts = collections.Counter( @@ -530,3 +549,140 @@ def temp_attr(obj: object, attr: str, value: object): yield obj finally: setattr(obj, attr, old) + + +def select_remaining_patches(current, document, relative_path: Path): + if not isinstance(document, dict) or ( + set(document) not in ({"target", "patches"}, {"target", "patches", "doctests"}) + ): + raise ValueError("Expected target and patches metadata") + if ( + relative_path.parts[:2] != ("Lib", "test") + or relative_path.suffix != ".py" + or document["target"] != relative_path.as_posix() + ): + raise ValueError("Marker override target does not match the test file") + if "doctests" in document and not isinstance(document["doctests"], dict): + raise ValueError("Doctests metadata must be an object") + try: + remaining = patches_from_json(document["patches"]) + except (TypeError, KeyError, AttributeError, ValueError) as error: + raise ValueError("Invalid remaining marker metadata") from error + for cls, methods in remaining.items(): + if cls not in current or any(method not in current[cls] for method in methods): + raise ValueError("Marker override cannot add targets") + for cls, methods in current.items(): + for method, existing in methods.items(): + pending = list(existing) + omitted = [] + for marker in remaining.get(cls, {}).get(method, []): + if marker not in pending: + raise ValueError( + "Marker override cannot add, change, or reorder markers" + ) + index = pending.index(marker) + omitted.extend(pending[:index]) + pending = pending[index + 1 :] + omitted.extend(pending) + if any(marker.ut_method != UtMethod.ExpectedFailure for marker in omitted): + raise ValueError( + "Only unconditional expectedFailure markers may be removed" + ) + return remaining + + +EXPECTED_FAILURE = doctest.register_optionflag("EXPECTED_FAILURE") +CANONICAL_LOAD_TESTS = """def load_tests(loader, tests, pattern): + tests.addTest(doctest.DocTestSuite()) + return tests""" +PATCHED_LOAD_TESTS = """def load_tests(loader, tests, pattern): + from test.support.rustpython import DocTestChecker # TODO: RUSTPYTHON + tests.addTest(doctest.DocTestSuite(checker=DocTestChecker())) # TODO: RUSTPYTHON + return tests""" + + +def _examples(contents): + tree = ast.parse(contents) + if not tree.body or not isinstance(tree.body[0], ast.Expr): + raise ValueError("Missing module doctest") + doc = tree.body[0].value + if not isinstance(doc, ast.Constant) or not isinstance(doc.value, str): + raise ValueError("Missing module doctest") + # Raw docstring offsets map directly to source lines, unlike cleaned docs. + return tree, doc, doctest.DocTestParser().get_examples(doc.value) + + +def apply_doctest_metadata(old, canonical, metadata): + """Preserve only listed guards, without carrying old expected output.""" + if not isinstance(metadata, dict) or any( + not isinstance(source, str) + or not isinstance(reason, str) + or "\n" in reason + or "\r" in reason + for source, reason in metadata.items() + ): + raise ValueError("Doctest metadata must map exact input to single-line reasons") + _, _, old_examples = _examples(old) + tree, doc, examples = _examples(canonical) + lines = canonical.splitlines(keepends=True) + for source in metadata: + old_matches = [ + example + for example in old_examples + if re.sub(r"\s+# TODO: RUSTPYTHON[^\n]*", "", example.source) == source + ] + if len(old_matches) != 1 or not old_matches[0].options.get(EXPECTED_FAILURE): + raise ValueError("Missing or ambiguous existing doctest guard") + matches = [example for example in examples if example.source == source] + if len(matches) != 1 or matches[0].options: + raise ValueError("Missing, ambiguous, or already marked canonical doctest") + index = doc.lineno - 1 + matches[0].lineno + line = lines[index] + if ( + ">>>" not in line + or line.split(">>>", 1)[1].strip() != source.splitlines()[0] + ): + raise ValueError("Unrecognized canonical doctest prompt") + lines[index] = ( + line.rstrip("\n") + + " # TODO: RUSTPYTHON; " + + metadata[source] + + " # doctest: +EXPECTED_FAILURE\n" + ) + + if not metadata: + return canonical + + loaders = [ + node + for node in ast.walk(tree) + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) + and node.name == "load_tests" + ] + if len(loaders) != 1 or loaders[0] not in tree.body: + raise ValueError("Missing or ambiguous doctest loader") + if ast.get_source_segment(canonical, loaders[0]) != CANONICAL_LOAD_TESTS: + raise ValueError("Changed canonical doctest loader") + result = "".join(lines) + if result.count(CANONICAL_LOAD_TESTS) != 1: + raise ValueError("Ambiguous canonical doctest loader text") + return result.replace(CANONICAL_LOAD_TESTS, PATCHED_LOAD_TESTS, 1) + + +def extract_doctest_guards(contents: str) -> dict[str, str]: + """Extract guarded module-docstring inputs without executing examples.""" + tree = ast.parse(contents) + if not tree.body or not isinstance(tree.body[0], ast.Expr): + return {} + doc = tree.body[0].value + if not isinstance(doc, ast.Constant) or not isinstance(doc.value, str): + return {} + guards = {} + for example in doctest.DocTestParser().get_examples(doc.value): + if not example.options.get(EXPECTED_FAILURE): + continue + source = re.sub(r"\s+# TODO: RUSTPYTHON[^\n]*", "", example.source) + if source in guards: + raise ValueError("Ambiguous existing doctest guard") + guards[source] = example.source + return guards diff --git a/scripts/update_lib/tests/test_migrate.py b/scripts/update_lib/tests/test_migrate.py index 0cc247ba841..2c18a9e1a3c 100644 --- a/scripts/update_lib/tests/test_migrate.py +++ b/scripts/update_lib/tests/test_migrate.py @@ -1,20 +1,127 @@ """Tests for migrate.py - file migration operations.""" +import contextlib +import io +import json import pathlib import tempfile import unittest +from pathlib import Path +from unittest.mock import patch from update_lib.cmd_migrate import ( + CANONICAL_CALL, + EXCEPTION_LOOP, + JIT_ERROR, + LOCAL_CALL, + SYSCONFIG_LOCAL, + SYSCONFIG_UPSTREAM, + main, + migrate_package, patch_directory, patch_file, patch_single_content, + preserve_migration_customizations, +) +from update_lib.patch_spec import ( + CANONICAL_LOAD_TESTS, + COMMENT, + PatchSpec, + UtMethod, + extract_patches, + iter_patches, + patches_to_json, + select_remaining_patches, ) -from update_lib.patch_spec import COMMENT class TestPatchSingleContent(unittest.TestCase): """Tests for patch_single_content function.""" + def migrate_doctest(self, selection=None, explicit=False): + old = '''""" +>>> raise X, Y # TODO: RUSTPYTHON; old mismatch # doctest: +EXPECTED_FAILURE +SyntaxError: old result +>>> raise Z # TODO: RUSTPYTHON; obsolete mismatch # doctest: +EXPECTED_FAILURE +SyntaxError: obsolete result +""" +''' + canonical = ( + '''""" +>>> raise X, Y +SyntaxError: canonical result +>>> raise Z +SyntaxError: canonical second result +""" +''' + + CANONICAL_LOAD_TESTS + + "\n" + ) + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) / "repository" + target = root / "Lib/test/test_example.py" + target.parent.mkdir(parents=True) + target.write_text(old) + source = Path(temporary) / "canonical/Lib/test/test_example.py" + source.parent.mkdir(parents=True) + source.write_text(canonical) + args = [str(source)] + if explicit: + metadata = Path(temporary) / "selection.json" + metadata.write_text( + json.dumps( + { + "target": "Lib/test/test_example.py", + "patches": {}, + "doctests": selection, + } + ) + ) + args += ["--patches", str(metadata)] + with ( + contextlib.chdir(root), + patch( + "update_lib.cmd_migrate.__file__", + str(root / "scripts/update_lib/cmd_migrate.py"), + ), + contextlib.redirect_stdout(io.StringIO()), + contextlib.redirect_stderr(io.StringIO()), + ): + result = main(args) + return result, target.read_text(), old, canonical + + def test_doctest_default_requires_explicit_selection(self): + status, result, old, _ = self.migrate_doctest() + self.assertEqual(status, 1) + self.assertEqual(result, old) + + def test_doctest_cli_selection_keeps_only_reviewed_guard(self): + status, result, _, _ = self.migrate_doctest( + {"raise X, Y\n": "known mismatch"}, True + ) + self.assertEqual(status, 0) + self.assertEqual(result.count("+EXPECTED_FAILURE"), 1) + self.assertIn("SyntaxError: canonical result", result) + self.assertIn("SyntaxError: canonical second result", result) + self.assertNotIn("obsolete mismatch", result) + + def test_doctest_empty_selection_keeps_canonical_unchanged(self): + status, result, _, canonical = self.migrate_doctest({}, True) + self.assertEqual(status, 0) + self.assertEqual(result, canonical) + + def test_doctest_invalid_selection_writes_nothing(self): + for selection in ( + None, + [], + {"raise unknown\n": "unrecognized"}, + {"raise X, Y\n": "line\nbreak"}, + ): + with self.subTest(selection=selection): + status, result, old, _ = self.migrate_doctest(selection, True) + self.assertEqual(status, 1) + self.assertEqual(result, old) + def test_patch_with_no_existing_file(self): """Test patching when lib file doesn't exist.""" with tempfile.TemporaryDirectory() as tmpdir: @@ -136,6 +243,37 @@ def test_one(self): class TestPatchDirectory(unittest.TestCase): """Tests for patch_directory function.""" + def test_excludes_cache_directories_but_preserves_canonical_assets(self): + with tempfile.TemporaryDirectory() as temporary: + root = pathlib.Path(temporary) + source = root / "source" + destination = root / "destination" + for relative in ("__pycache__", "nested/__pycache__", "assets"): + (source / relative).mkdir(parents=True) + files = { + "module.py": b"# canonical source\n", + "__pycache__/module.cpython-315.pyc": b"stale cache", + "nested/__pycache__/module.cpython-315.pyc": b"nested cache", + "assets/__pycache__": b"ordinary canonical file", + "assets/fixture.pyc": b"canonical bytecode fixture", + } + for relative, content in files.items(): + (source / relative).write_bytes(content) + patch_directory(source, destination, verbose=False) + expected = { + relative: content + for relative, content in files.items() + if "__pycache__" not in pathlib.Path(relative).parts[:-1] + } + actual = { + path.relative_to(destination).as_posix(): path.read_bytes() + for path in destination.rglob("*") + if path.is_file() + } + self.assertEqual(actual, expected) + for relative, content in files.items(): + self.assertEqual((source / relative).read_bytes(), content) + def test_patch_directory_all_files(self): """Test that patch_directory processes all .py files.""" with tempfile.TemporaryDirectory() as tmpdir: @@ -192,5 +330,489 @@ def test_one(self): self.assertIn("@unittest.expectedFailure", content) +class RuntimePatchTests(unittest.TestCase): + def test_sysconfig_identity_and_new_content(self): + path = Path("Lib/sysconfig/__init__.py") + source = SYSCONFIG_UPSTREAM + "new_final_behavior = True\n" + result = preserve_migration_customizations(path, SYSCONFIG_LOCAL, source) + self.assertEqual(result, SYSCONFIG_LOCAL + "new_final_behavior = True\n") + self.assertEqual( + preserve_migration_customizations(path, result, source), result + ) + + def test_pickle_jit_mapping_and_new_content(self): + path = Path("Lib/_compat_pickle.py") + source = "final_mapping = 'ImportCycleError'\n" + EXCEPTION_LOOP + result = preserve_migration_customizations( + path, JIT_ERROR + EXCEPTION_LOOP, source + ) + self.assertEqual( + result, "final_mapping = 'ImportCycleError'\n" + JIT_ERROR + EXCEPTION_LOOP + ) + self.assertEqual( + preserve_migration_customizations(path, result, source), result + ) + + def test_changed_or_ambiguous_anchors_fail(self): + cases = [ + (Path("Lib/sysconfig/__init__.py"), SYSCONFIG_LOCAL, SYSCONFIG_UPSTREAM), + (Path("Lib/_compat_pickle.py"), JIT_ERROR + EXCEPTION_LOOP, EXCEPTION_LOOP), + ] + for path, old, source in cases: + for bad_old in ( + "", + old + old, + old.replace("RustPython", "Other").replace("JitError", "Other"), + ): + with ( + self.subTest(path=path, bad_old=bad_old), + self.assertRaises(ValueError), + ): + preserve_migration_customizations(path, bad_old, source) + for bad_source in ( + "", + source + source, + source.replace("Python", "Other").replace("excname", "other"), + ): + with ( + self.subTest(path=path, bad_source=bad_source), + self.assertRaises(ValueError), + ): + preserve_migration_customizations(path, old, bad_source) + + def test_unrelated_file_unchanged(self): + self.assertEqual( + preserve_migration_customizations(Path("Lib/other.py"), JIT_ERROR, "new\n"), + "new\n", + ) + + def test_decoy_and_shadowed_sysconfig_anchors_fail(self): + path = Path("Lib/sysconfig/__init__.py") + for source in ( + 'doc = """' + SYSCONFIG_UPSTREAM + '"""\n', + SYSCONFIG_UPSTREAM + "def _get_implementation():\n return 'Other'\n", + SYSCONFIG_UPSTREAM + "_get_implementation = lambda: 'Other'\n", + SYSCONFIG_UPSTREAM + "from other import _get_implementation\n", + SYSCONFIG_UPSTREAM + "import other as _get_implementation\n", + SYSCONFIG_UPSTREAM + "class _get_implementation: pass\n", + SYSCONFIG_UPSTREAM + "del _get_implementation\n", + ): + with self.subTest(source=source), self.assertRaises(ValueError): + preserve_migration_customizations(path, SYSCONFIG_LOCAL, source) + + def test_decoy_and_reordered_pickle_anchors_fail(self): + path = Path("Lib/_compat_pickle.py") + old = JIT_ERROR + EXCEPTION_LOOP + with self.assertRaises(ValueError): + preserve_migration_customizations( + path, old, 'doc = """' + EXCEPTION_LOOP + '"""\n' + ) + with self.assertRaises(ValueError): + preserve_migration_customizations( + path, EXCEPTION_LOOP + JIT_ERROR, EXCEPTION_LOOP + ) + + def test_similarly_named_nested_files_are_untouched(self): + for path in ( + "Lib/other/sysconfig/__init__.py", + "Lib/other/_compat_pickle.py", + "other/sysconfig/__init__.py", + ): + self.assertEqual( + preserve_migration_customizations(Path(path), "old", "new"), "new" + ) + + def test_absolute_target_requires_explicit_repository_resolution(self): + with self.assertRaises(ValueError): + preserve_migration_customizations( + Path("/parent/Lib/repository/Lib/sysconfig/__init__.py"), + SYSCONFIG_LOCAL, + SYSCONFIG_UPSTREAM, + ) + + def test_migrate_resolves_real_repository_under_lib_parent(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) / "Lib" / "repository" + output = root / "Lib" / "sysconfig" / "__init__.py" + output.parent.mkdir(parents=True) + output.write_text(SYSCONFIG_LOCAL, encoding="utf-8") + source = Path(temporary) / "canonical.py" + source.write_text(SYSCONFIG_UPSTREAM, encoding="utf-8") + module_file = root / "scripts" / "update_lib" / "cmd_migrate.py" + with patch("update_lib.cmd_migrate.__file__", str(module_file)): + self.assertEqual(patch_single_content(source, output), SYSCONFIG_LOCAL) + outside = ( + Path(temporary) / "outside" / "Lib" / "sysconfig" / "__init__.py" + ) + outside.parent.mkdir(parents=True) + outside.write_text(SYSCONFIG_LOCAL, encoding="utf-8") + self.assertEqual( + patch_single_content(source, outside), SYSCONFIG_UPSTREAM + ) + + +CALL_PATH = Path("Lib/test/test_cmd_line.py") +CALL_TEMPLATE = """class CmdLineTest: + def test_empty_PYTHONPATH_issue16309(self): + result = CALL + self.assertEqual(result, 'new final assertion') +""" +CALL_OLD = CALL_TEMPLATE.replace("CALL", LOCAL_CALL).replace( + "new final assertion", "old assertion" +) +CALL_CANONICAL = CALL_TEMPLATE.replace("CALL", CANONICAL_CALL) + + +class CallPatchTests(unittest.TestCase): + def test_final_body_and_roundtrip(self): + result = preserve_migration_customizations(CALL_PATH, CALL_OLD, CALL_CANONICAL) + self.assertEqual(result, CALL_TEMPLATE.replace("CALL", LOCAL_CALL)) + self.assertIn("new final assertion", result) + self.assertEqual( + preserve_migration_customizations(CALL_PATH, result, CALL_CANONICAL), result + ) + + def test_missing_duplicate_owners_and_calls_fail(self): + for source in ( + CALL_CANONICAL.replace("CmdLineTest", "Other"), + CALL_CANONICAL + CALL_CANONICAL, + CALL_CANONICAL.replace( + " result = ", " " + CANONICAL_CALL + "\n result = " + ), + CALL_CANONICAL.replace("test_empty_PYTHONPATH_issue16309", "other"), + ): + with self.subTest(source=source), self.assertRaises(ValueError): + preserve_migration_customizations(CALL_PATH, CALL_OLD, source) + + def test_changed_keywords_and_decoy_fail(self): + for old in ( + CALL_OLD.replace('RUSTPYTHONPATH=""', 'RUSTPYTHONPATH="changed"'), + CALL_CANONICAL, + 'doc = """' + CALL_OLD + '"""\n', + ): + with self.subTest(old=old), self.assertRaises(ValueError): + preserve_migration_customizations(CALL_PATH, old, CALL_CANONICAL) + with self.assertRaises(ValueError): + preserve_migration_customizations( + CALL_PATH, + CALL_OLD, + CALL_CANONICAL.replace('PYTHONPATH=""', 'PYTHONPATH="changed"'), + ) + + def test_unrelated_file_unchanged(self): + self.assertEqual( + preserve_migration_customizations(Path("Lib/test/other.py"), "old", "new"), + "new", + ) + + +OVERRIDE_CLEAN = """import unittest +class Example(unittest.TestCase): + def test_known(self): + self.assertEqual(1, 1) + def test_guarded(self): + self.assertEqual(2, 2) +""" +OVERRIDE_OLD = OVERRIDE_CLEAN.replace( + " def test_known", + " @unittest.expectedFailure # TODO: RUSTPYTHON; old failure\n def test_known", +).replace( + " def test_guarded", + ' @unittest.skip("TODO: RUSTPYTHON; retained guard")\n def test_guarded', +) + + +class MarkerOverrideTests(unittest.TestCase): + def invoke(self, mutation=None, directory=False, override=True, legacy=False): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) / "repository" + output = root / "Lib" / "test" / "test_example.py" + output.parent.mkdir(parents=True) + output.write_text(OVERRIDE_OLD) + source = Path(temporary) / "cpython" / "Lib" / "test" / "test_example.py" + source.parent.mkdir(parents=True) + source.write_text(OVERRIDE_CLEAN) + markers = patches_to_json(extract_patches(OVERRIDE_OLD)) + del markers["Example"]["test_known"] + document = {"target": "Lib/test/test_example.py", "patches": markers} + if mutation: + mutation(document) + metadata = Path(temporary) / "remaining.json" + metadata.write_text(json.dumps(document)) + args = [str(source.parent if directory else source)] + if override: + args += ["--patches", str(metadata)] + if legacy: + args += ["--legacy-map", str(metadata)] + with ( + contextlib.chdir(root), + patch( + "update_lib.cmd_migrate.__file__", + str(root / "scripts/update_lib/cmd_migrate.py"), + ), + contextlib.redirect_stdout(io.StringIO()), + contextlib.redirect_stderr(io.StringIO()), + ): + status = main(args) + return status, output.read_text() + + def test_cli_override_removes_only_requested_expected_failure(self): + status, result = self.invoke() + self.assertEqual(status, 0) + self.assertNotIn("expectedFailure", result) + self.assertIn('unittest.skip("TODO: RUSTPYTHON; retained guard")', result) + self.assertIn("self.assertEqual(1, 1)", result) + self.assertIn("self.assertEqual(2, 2)", result) + + def test_default_migration_preserves_all_markers(self): + status, result = self.invoke(override=False) + self.assertEqual(status, 0) + self.assertIn("expectedFailure", result) + self.assertIn("retained guard", result) + + def test_invalid_target_does_not_write(self): + status, result = self.invoke( + lambda document: document.update(target="Lib/test/other.py") + ) + self.assertEqual(status, 1) + self.assertEqual(result, OVERRIDE_OLD) + + def test_removing_skip_does_not_write(self): + status, result = self.invoke( + lambda document: document["patches"]["Example"].clear() + ) + self.assertEqual(status, 1) + self.assertEqual(result, OVERRIDE_OLD) + + def test_directory_override_does_not_write(self): + status, result = self.invoke(directory=True) + self.assertEqual(status, 1) + self.assertEqual(result, OVERRIDE_OLD) + + def test_legacy_combination_does_not_write(self): + status, result = self.invoke(legacy=True) + self.assertEqual(status, 1) + self.assertEqual(result, OVERRIDE_OLD) + + def test_added_changed_or_reordered_markers_fail(self): + failure = PatchSpec(UtMethod.ExpectedFailure, reason="known failure") + success = PatchSpec(UtMethod.ExpectedSuccess, reason="known success") + current = {"Example": {"test_known": [failure, success]}} + for markers in ( + [failure, success, failure], + [PatchSpec(UtMethod.ExpectedFailure, reason="changed"), success], + [success, failure], + ): + document = { + "target": "Lib/test/test_example.py", + "patches": patches_to_json({"Example": {"test_known": markers}}), + } + with self.subTest(markers=markers), self.assertRaises(ValueError): + select_remaining_patches(current, document, Path(document["target"])) + + +PACKAGE_MARKED = """import unittest +class Example: + @unittest.expectedFailure # TODO: RUSTPYTHON; retained reason + def test_one(self): + pass +""" +PACKAGE_CANONICAL = """import unittest +class Example: + def test_one(self): + pass +""" + + +class TestMigratePackage(unittest.TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory() + self.addCleanup(self.temp.cleanup) + self.root = Path(self.temp.name) + self.src = self.root / "canonical" / "package" + self.src.mkdir(parents=True) + (self.src / "__init__.py").write_text("# canonical package\n") + (self.src / "test_part.py").write_text(PACKAGE_CANONICAL) + self.dest = self.root / "Lib" / "test" / "package" + self.dest.parent.mkdir(parents=True) + self.old = self.dest.parent / "old.py" + self.old.write_text(PACKAGE_MARKED) + self.manifest = self.root / "map.json" + self.mapping = {"old.py": {"Example": "test_part.py"}} + + def run_migration(self): + self.manifest.write_text(json.dumps(self.mapping)) + migrate_package(self.src, self.dest, self.manifest) + + def assert_rejected(self): + before = self.old.read_bytes() + with self.assertRaises((ValueError, FileNotFoundError, SyntaxError)): + self.run_migration() + self.assertFalse(self.dest.exists()) + self.assertEqual(self.old.read_bytes(), before) + + def test_preserves_marker_and_canonical_asset(self): + (self.src / "asset.dat").write_bytes(b"\x00canonical") + self.run_migration() + self.assertFalse(self.old.exists()) + self.assertEqual((self.dest / "asset.dat").read_bytes(), b"\x00canonical") + self.assertEqual( + list(iter_patches((self.dest / "test_part.py").read_text())), + list(iter_patches(PACKAGE_MARKED)), + ) + + def test_preserves_inherited_override(self): + (self.src / "test_part.py").write_text( + PACKAGE_CANONICAL.replace("class Example:", "class Base:") + + "\nclass Example(Base):\n pass\n" + ) + self.run_migration() + text = (self.dest / "test_part.py").read_text() + self.assertIn("return super().test_one()", text) + self.assertEqual(list(iter_patches(text)), list(iter_patches(PACKAGE_MARKED))) + + def test_excludes_cache_directories_but_preserves_canonical_assets(self): + for relative in ("__pycache__", "nested/__pycache__", "assets"): + (self.src / relative).mkdir(parents=True) + files = { + "__pycache__/module.cpython-315.pyc": b"stale cache", + "nested/__pycache__/module.cpython-315.pyc": b"nested cache", + "assets/__pycache__": b"ordinary canonical file", + "assets/fixture.pyc": b"canonical bytecode fixture", + } + for relative, content in files.items(): + (self.src / relative).write_bytes(content) + self.run_migration() + self.assertFalse((self.dest / "__pycache__").exists()) + self.assertFalse((self.dest / "nested" / "__pycache__").exists()) + for relative in ("assets/__pycache__", "assets/fixture.pyc"): + self.assertEqual((self.dest / relative).read_bytes(), files[relative]) + for relative, content in files.items(): + self.assertEqual((self.src / relative).read_bytes(), content) + + def test_missing_method(self): + (self.src / "test_part.py").write_text( + PACKAGE_CANONICAL.replace("test_one", "test_other") + ) + self.assert_rejected() + + def test_missing_class(self): + (self.src / "test_part.py").write_text( + PACKAGE_CANONICAL.replace("Example", "Other") + ) + self.assert_rejected() + + def test_ambiguous_class(self): + (self.src / "test_part.py").write_text(PACKAGE_CANONICAL + PACKAGE_CANONICAL) + self.assert_rejected() + + def test_ambiguous_method(self): + (self.src / "test_part.py").write_text( + PACKAGE_CANONICAL + " def test_one(self):\n pass\n" + ) + self.assert_rejected() + + def test_missing_destination(self): + self.mapping["old.py"]["Example"] = "missing.py" + self.assert_rejected() + + def test_unused_mapping(self): + self.mapping["old.py"]["Unused"] = "test_part.py" + self.assert_rejected() + + def test_unmapped_marker(self): + self.mapping["old.py"] = {} + self.assert_rejected() + + def test_target_traversal(self): + self.mapping["old.py"]["Example"] = "../outside.py" + self.assert_rejected() + + def test_legacy_traversal(self): + self.mapping["../old.py"] = self.mapping.pop("old.py") + self.assert_rejected() + + def test_duplicate_marker(self): + (self.old.parent / "second.py").write_text(PACKAGE_MARKED) + self.mapping["second.py"] = self.mapping["old.py"] + self.assert_rejected() + self.assertEqual((self.old.parent / "second.py").read_text(), PACKAGE_MARKED) + + def test_preserves_ordered_markers_from_one_legacy_method(self): + marked = PACKAGE_MARKED.replace( + " def test_one", + ' @unittest.skip("TODO: RUSTPYTHON; retained skip")\n def test_one', + ) + self.old.write_text(marked) + self.run_migration() + result = (self.dest / "test_part.py").read_text() + self.assertEqual(list(iter_patches(result)), list(iter_patches(marked))) + self.assertLess(result.index("expectedFailure"), result.index("unittest.skip")) + + def test_duplicate_legacy_owner_is_rejected(self): + for source in ( + PACKAGE_MARKED + PACKAGE_MARKED, + PACKAGE_MARKED + " def test_one(self):\n pass\n", + ): + with self.subTest(source=source): + self.old.write_text(source) + self.assert_rejected() + + def test_unmarked_legacy_retired(self): + self.old.write_text(PACKAGE_CANONICAL) + self.mapping["old.py"] = {} + self.run_migration() + self.assertFalse(self.old.exists()) + self.assertEqual((self.dest / "test_part.py").read_text(), PACKAGE_CANONICAL) + + def test_legacy_doctest_guards_prevent_all_writes(self): + self.old.write_text( + '"""\n' + ">>> raise ValueError # TODO: RUSTPYTHON; mismatch # doctest: +EXPECTED_FAILURE\n" + "ValueError\n" + '"""\n' + PACKAGE_MARKED + ) + self.assert_rejected() + + def test_validation_failure_after_valid_entry_writes_nothing(self): + self.mapping["missing.py"] = {} + self.assert_rejected() + + def test_existing_destination_is_not_overwritten(self): + self.dest.mkdir() + with self.assertRaises(ValueError): + self.run_migration() + self.assertTrue(self.old.exists()) + self.assertEqual(list(self.dest.iterdir()), []) + + def test_source_escape_rejected(self): + outside = self.root / "outside.py" + outside.write_text(PACKAGE_CANONICAL) + (self.src / "test_part.py").unlink() + (self.src / "test_part.py").symlink_to(outside) + self.assert_rejected() + + def test_legacy_symlink_rejected(self): + outside = self.root / "outside.py" + outside.write_text(PACKAGE_MARKED) + self.old.unlink() + self.old.symlink_to(outside) + self.assert_rejected() + + def test_inheritance_cycle_rejected(self): + (self.src / "test_part.py").write_text( + "class Example(Base):\n pass\nclass Base(Example):\n pass\n" + ) + self.assert_rejected() + + def test_canonical_marker_collision_rejected(self): + (self.src / "test_part.py").write_text(PACKAGE_MARKED) + self.assert_rejected() + + def test_unmarked_syntax_error_prevents_all_writes(self): + (self.src / "broken.py").write_text("class :\n") + self.assert_rejected() + + 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 index 754adaca46a..1a68ca10d26 100644 --- a/scripts/update_lib/tests/test_patch_spec.py +++ b/scripts/update_lib/tests/test_patch_spec.py @@ -4,11 +4,15 @@ import unittest from update_lib.patch_spec import ( + CANONICAL_LOAD_TESTS, COMMENT, + PATCHED_LOAD_TESTS, PatchSpec, UtMethod, _find_import_insert_line, + apply_doctest_metadata, apply_patches, + extract_doctest_guards, extract_patches, iter_tests, ) @@ -99,6 +103,41 @@ def test_value(self): """ self.assertEqual(extract_patches(code), {}) + def test_bare_marked_skip_roundtrip(self): + clean = """class Owner: + def test_value(self): + return 42 +""" + marked = clean.replace( + " def test_value", + " @unittest.skip # TODO: RUSTPYTHON; original guard\n def test_value", + ) + patches = extract_patches(marked) + self.assertEqual( + patches["Owner"]["test_value"], + [PatchSpec(UtMethod.Skip, reason="original guard")], + ) + result = apply_patches(clean, patches) + self.assertIn('@unittest.skip("TODO: RUSTPYTHON; original guard")', result) + self.assertEqual(apply_patches(clean, extract_patches(result)), result) + method = list(iter_tests(ast.parse(result)))[0][1] + self.assertEqual(ast.dump(method.body[0]), "Return(value=Constant(value=42))") + + def test_unmarked_bare_skip_and_unrelated_decorators_are_not_patches(self): + code = """class Owner: + @unittest.skip + def test_unmarked(self): + pass + @other.skip # TODO: RUSTPYTHON; unrelated + def test_other(self): + pass + @unittest.other # TODO: RUSTPYTHON; unrelated + def test_unknown(self): + pass +""" + self.assertEqual(extract_patches(code), {}) + self.assertEqual(apply_patches(code, {}), code) + def test_extract_expected_failure(self): """Test extracting @unittest.expectedFailure decorator.""" code = f""" @@ -266,6 +305,43 @@ class Duplicate: {"Duplicate": {"test_value": [PatchSpec(UtMethod.ExpectedFailure)]}}, ) + def test_unpatched_conditional_helpers_are_allowed(self): + code = """if enabled: + class Helper: + pass +else: + class Helper: + pass +class Target: + helper = Helper + def test_value(self): + return 1 +""" + result = apply_patches( + code, {"Target": {"test_value": [PatchSpec(UtMethod.ExpectedFailure)]}} + ) + self.assertEqual(result.count("class Helper:"), 2) + self.assertEqual(result.count("@unittest.expectedFailure"), 1) + + def test_ambiguous_transitive_ancestor_is_rejected(self): + code = """if enabled: + class Ancestor: + async def test_value(self): + pass +else: + class Ancestor: + def test_value(self): + pass +class Middle(Ancestor): + pass +class Target(Middle): + pass +""" + with self.assertRaisesRegex(ValueError, "Ambiguous.*Ancestor"): + apply_patches( + code, {"Target": {"test_value": [PatchSpec(UtMethod.ExpectedFailure)]}} + ) + def test_local_class_names_do_not_shadow_module_class(self): code = """import unittest class Target: @@ -573,5 +649,90 @@ def test_no_imports_no_docstring(self): self.assertEqual(line, 0) +DOCTEST_SOURCE = "raise X, Y\n" +DOCTEST_METADATA = {DOCTEST_SOURCE: "known mismatch"} +DOCTEST_OLD = ( + '''""" +>>> raise X, Y # TODO: RUSTPYTHON; old reason # doctest: +EXPECTED_FAILURE +Traceback (most recent call last): +SyntaxError: old message +""" +''' + + PATCHED_LOAD_TESTS + + "\n" +) +DOCTEST_CANONICAL = ( + '''""" +>>> raise X, Y +Traceback (most recent call last): +SyntaxError: final message +""" +''' + + CANONICAL_LOAD_TESTS + + "\n" +) + + +class ExactDoctestTests(unittest.TestCase): + def test_extracts_only_existing_guards(self): + self.assertEqual(set(extract_doctest_guards(DOCTEST_OLD)), {DOCTEST_SOURCE}) + self.assertEqual(extract_doctest_guards(DOCTEST_CANONICAL), {}) + self.assertEqual(extract_doctest_guards("value = 1\n"), {}) + + def test_duplicate_existing_guards_are_rejected(self): + duplicate = DOCTEST_OLD.replace( + ">>> raise X, Y # TODO:", + ">>> raise X, Y # TODO: RUSTPYTHON; duplicate # doctest: +EXPECTED_FAILURE\n" + ">>> raise X, Y # TODO:", + ) + with self.assertRaises(ValueError): + extract_doctest_guards(duplicate) + + def test_preserves_final_output_and_roundtrips(self): + result = apply_doctest_metadata( + DOCTEST_OLD, DOCTEST_CANONICAL, DOCTEST_METADATA + ) + self.assertIn("SyntaxError: final message", result) + self.assertNotIn("old message", result) + self.assertEqual(result.count("doctest: +EXPECTED_FAILURE"), 1) + self.assertIn(PATCHED_LOAD_TESTS, result) + self.assertEqual( + apply_doctest_metadata(result, DOCTEST_CANONICAL, DOCTEST_METADATA), result + ) + + def test_duplicate_input_is_rejected(self): + duplicate = DOCTEST_CANONICAL.replace( + ">>> raise X, Y\n", ">>> raise X, Y\n>>> raise X, Y\n" + ) + with self.assertRaises(ValueError): + apply_doctest_metadata(DOCTEST_OLD, duplicate, DOCTEST_METADATA) + + def test_missing_canonical_or_old_case_is_rejected(self): + with self.assertRaises(ValueError): + apply_doctest_metadata( + DOCTEST_OLD, + DOCTEST_CANONICAL.replace("raise X, Y", "raise X"), + DOCTEST_METADATA, + ) + with self.assertRaises(ValueError): + apply_doctest_metadata( + DOCTEST_OLD.replace("raise X, Y", "raise X"), + DOCTEST_CANONICAL, + DOCTEST_METADATA, + ) + + def test_changed_loader_and_unmarked_old_case_are_rejected(self): + with self.assertRaises(ValueError): + apply_doctest_metadata( + DOCTEST_OLD, + DOCTEST_CANONICAL.replace("DocTestSuite()", "DocTestSuite(None)"), + DOCTEST_METADATA, + ) + with self.assertRaises(ValueError): + apply_doctest_metadata( + DOCTEST_CANONICAL, DOCTEST_CANONICAL, DOCTEST_METADATA + ) + + if __name__ == "__main__": unittest.main()