diff --git a/scripts/update_lib/__main__.py b/scripts/update_lib/__main__.py index 7eb91970f..49399db6f 100644 --- a/scripts/update_lib/__main__.py +++ b/scripts/update_lib/__main__.py @@ -8,7 +8,6 @@ Usage: 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 - python scripts/update_lib copy-test SOURCE DESTINATION --test ClassName.method_name --source-sha256 HASH --destination-sha256 HASH """ import argparse @@ -50,11 +49,6 @@ def main(argv: list[str] | None = None) -> int: help="Copy library file/directory from CPython (delete existing first)", add_help=False, ) - subparsers.add_parser( - "copy-test", - help="Copy one complete canonical test from hash-pinned inputs", - add_help=False, - ) subparsers.add_parser( "deps", help="Show dependency information for a module", @@ -78,11 +72,6 @@ def main(argv: list[str] | None = None) -> int: return copy_lib_main(remaining) - if args.command == "copy-test": - from update_lib.cmd_copy_test import main as copy_test_main - - return copy_test_main(remaining) - if args.command == "migrate": from update_lib.cmd_migrate import main as migrate_main diff --git a/scripts/update_lib/cmd_copy_test.py b/scripts/update_lib/cmd_copy_test.py deleted file mode 100644 index 5e14959bb..000000000 --- a/scripts/update_lib/cmd_copy_test.py +++ /dev/null @@ -1,134 +0,0 @@ -"""Copy one complete canonical test definition without refreshing its module.""" - -import argparse -import ast -import hashlib -import json -from pathlib import Path - -from update_lib.patch_spec import _iter_class_methods, _iter_module_classes - - -def _definition(contents: str, qualified_name: str): - parts = qualified_name.split(".") - if len(parts) != 2 or not all(part.isidentifier() for part in parts): - raise ValueError("test name must be ClassName.method_name") - tree = ast.parse(contents) - classes = [node for node in _iter_module_classes(tree) if node.name == parts[0]] - if len(classes) != 1: - raise ValueError(f"Expected one class named {parts[0]}, found {len(classes)}") - methods = [ - node for node in _iter_class_methods(classes[0]) if node.name == parts[1] - ] - if len(methods) != 1: - raise ValueError( - f"Expected one method named {qualified_name}, found {len(methods)}" - ) - method = methods[0] - parents = { - child: parent - for parent in ast.walk(tree) - for child in ast.iter_child_nodes(parent) - } - guards = [] - child = method - while child in parents: - parent = parents[child] - if isinstance(parent, ast.If): - branch = "body" if child in parent.body else "orelse" - guards.append((ast.dump(parent.test), branch)) - elif not isinstance(parent, (ast.Module, ast.ClassDef)): - raise ValueError("Only module/class scopes and if guards are supported") - child = parent - start = min([method.lineno, *(node.lineno for node in method.decorator_list)]) - 1 - return method, guards, start, method.end_lineno - - -def copy_test(source: str, destination: str, qualified_name: str) -> str: - """Replace an unambiguous whole definition, preserving all surrounding bytes. - - Guard, signature, decorator and indentation differences require a broader - migration rather than silently changing the selected test's execution context. - """ - src, src_guards, src_start, src_end = _definition(source, qualified_name) - dst, dst_guards, dst_start, dst_end = _definition(destination, qualified_name) - if src_guards != dst_guards: - raise ValueError("Source and destination test guards differ") - - def signature(node): - return ( - ast.dump(node.args), - ast.dump(node.returns) if node.returns is not None else None, - [ast.dump(param) for param in getattr(node, "type_params", [])], - ) - - if type(src) is not type(dst) or signature(src) != signature(dst): - raise ValueError("Source and destination test signatures differ") - if [ast.dump(node) for node in src.decorator_list] != [ - ast.dump(node) for node in dst.decorator_list - ]: - raise ValueError( - "Source and destination decorators differ; preserve markers explicitly" - ) - if src.col_offset != dst.col_offset: - raise ValueError("Source and destination indentation differs") - source_lines = source.splitlines(keepends=True) - destination_lines = destination.splitlines(keepends=True) - # AST equality omits comments, including project marker rationales. Require - # the complete decorator block to match before replacing the definition. - if ( - source_lines[src_start : src.lineno - 1] - != destination_lines[dst_start : dst.lineno - 1] - ): - raise ValueError( - "Source and destination decorator text differs; preserve markers explicitly" - ) - definition = source_lines[src_start:src_end] - result = "".join( - destination_lines[:dst_start] + definition + destination_lines[dst_end:] - ) - _, _, result_start, result_end = _definition(result, qualified_name) - if result.splitlines(keepends=True)[result_start:result_end] != definition: - raise ValueError("Copied definition is not byte-identical to the source") - return result - - -def main(argv: list[str] | None = None) -> int: - parser = argparse.ArgumentParser(description=__doc__) - parser.add_argument("source", type=Path) - parser.add_argument("destination", type=Path) - parser.add_argument("--test", required=True, help="Exact ClassName.method_name") - parser.add_argument("--source-sha256", required=True) - parser.add_argument("--destination-sha256", required=True) - args = parser.parse_args(argv) - try: - source = args.source.read_bytes() - destination = args.destination.read_bytes() - for label, contents, expected in ( - ("source", source, args.source_sha256), - ("destination", destination, args.destination_sha256), - ): - if hashlib.sha256(contents).hexdigest() != expected: - raise ValueError(f"{label} SHA256 does not match the pinned input") - result = copy_test( - source.decode("utf-8"), destination.decode("utf-8"), args.test - ).encode("utf-8") - # Recheck the destination immediately before writing so stale invocations - # fail instead of overwriting an intervening edit. - if args.destination.read_bytes() != destination: - raise ValueError("Destination changed during test preparation") - args.destination.write_bytes(result) - except (OSError, UnicodeError, SyntaxError, ValueError) as exc: - parser.exit(1, f"Error: {exc}\n") - print( - json.dumps( - { - "test": args.test, - "source_sha256": args.source_sha256, - "destination_before_sha256": args.destination_sha256, - "destination_after_sha256": hashlib.sha256(result).hexdigest(), - }, - sort_keys=True, - ) - ) - return 0 diff --git a/scripts/update_lib/cmd_migrate.py b/scripts/update_lib/cmd_migrate.py index efbce2d03..58008cd35 100644 --- a/scripts/update_lib/cmd_migrate.py +++ b/scripts/update_lib/cmd_migrate.py @@ -5,7 +5,6 @@ Migrate Python Lib files from CPython, preserving explicit local adaptations. 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 - python scripts/update_lib migrate cpython/Lib/test/test_foo --canonical-removals removals.json --dry-run Remaining metadata names the exact repository-relative target and its existing decorator patches. An optional "doctests" object selects existing guarded inputs @@ -19,19 +18,6 @@ Non-marker adaptations require reviewed blocks on each migration, for example Use a host Python that parses the source version (Python 3.15 for 3.15 syntax), as required by the existing marker extraction and application steps. -For additive updates, --assignments or --functions selects contiguous canonical -top-level definitions before an unchanged anchor. Both input SHA256 pins are -required. --exports copies only complete selected __all__ lines. Function -dependencies are checked without importing or executing either module; unchanged -imports and simple assignments are supported, not arbitrary existing helpers. - -Canonical removals are a separate, explicit operation. JSON source metadata -pins path (Lib/test/test_*), commit and tree; removals maps explicit relative -filenames to sha256 and upstream_commit. The caller must first verify and -review historical provenance, including release-branch evidence. This command -checks pinned canonical absence and exact local bytes; it does not authenticate -historical evidence. --dry-run performs all validation without deleting files. - This will: 1. Extract patches from Lib/test/test_foo.py (if exists) 2. Apply them to cpython/Lib/test/test_foo.py @@ -40,16 +26,10 @@ This will: import argparse import ast -import builtins import collections -import hashlib import json -import os import pathlib -import re import shutil -import subprocess -import symtable import sys sys.path.insert(0, str(pathlib.Path(__file__).parent.parent)) @@ -64,553 +44,6 @@ def _preserve_block(old, new, canonical, local): return new.replace(canonical, local, 1) -def _single_assignment(tree, name): - matches = [ - node - for node in ast.walk(tree) - if isinstance(node, ast.Name) - and isinstance(node.ctx, ast.Store) - and node.id == name - ] - nodes = [ - node - for node in tree.body - if isinstance(node, ast.Assign) - and len(node.targets) == 1 - and isinstance(node.targets[0], ast.Name) - and node.targets[0].id == name - ] - rebound = any( - isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)) - and node.name == name - or isinstance(node, ast.alias) - and (node.asname or node.name.split(".")[0]) == name - for node in ast.walk(tree) - ) - if len(nodes) != 1 or matches != [nodes[0].targets[0]] or rebound: - raise ValueError(f"Expected one top-level assignment to {name}") - return nodes[0] - - -def _selected_export_addition(src, dst, source_lines, lines, exports): - src_all, dst_all = ( - _single_assignment(src, "__all__"), - _single_assignment(dst, "__all__"), - ) - if not isinstance(src_all.value, ast.List) or not isinstance( - dst_all.value, ast.List - ): - raise ValueError("__all__ must be a literal list") - if any( - not isinstance(node, ast.Constant) or not isinstance(node.value, str) - for node in src_all.value.elts + dst_all.value.elts - ): - raise ValueError("__all__ entries must be literal strings") - src_values, dst_values = ( - ast.literal_eval(src_all.value), - ast.literal_eval(dst_all.value), - ) - if any(src_values.count(name) != 1 or name in dst_values for name in exports): - raise ValueError( - "Exports must occur once in canonical __all__ and be absent locally" - ) - export_nodes = [node for node in src_all.value.elts if node.value in exports] - export_lines = sorted({node.lineno - 1 for node in export_nodes}) - fragment = "".join(source_lines[line] for line in export_lines) - if ast.literal_eval("[\n" + fragment + "]") != exports: - raise ValueError( - "Canonical export lines include unselected entries or differ in order" - ) - if lines[dst_all.lineno - 1].strip() != "__all__ = [": - raise ValueError("Expected a multiline destination __all__ list") - return dst_all.lineno, [fragment] - - -def copy_selected_assignments(source, destination, names, exports): - """Copy contiguous canonical assignments and export lines, failing closed. - - Only additions are supported. Dependencies must already be imported with the - same bindings, or be builtins/earlier selected assignments. The next unchanged - top-level statement is the insertion anchor; unrelated bytes stay untouched. - """ - if ( - not names - or len(set(names)) != len(names) - or not all(name.isidentifier() for name in names) - ): - raise ValueError("Assignment names must be unique identifiers") - if len(set(exports)) != len(exports) or not set(exports) <= set(names): - raise ValueError("Exports must be unique selected assignment names") - src, dst = ast.parse(source), ast.parse(destination) - - selected = [_single_assignment(src, name) for name in names] - start = src.body.index(selected[0]) - if src.body[start : start + len(names)] != selected or start + len(names) == len( - src.body - ): - raise ValueError("Assignments must be contiguous and followed by an anchor") - if (start and src.body[start - 1].end_lineno >= selected[0].lineno) or ( - selected[-1].end_lineno >= src.body[start + len(names)].lineno - ): - raise ValueError("Selected assignments must occupy complete source lines") - for node in ast.walk(dst): - if ( - isinstance(node, ast.Name) - and isinstance(node.ctx, ast.Store) - and node.id in names - or isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)) - and node.name in names - or isinstance(node, ast.alias) - and (node.asname or node.name.split(".")[0]) in names - ): - raise ValueError("A selected assignment already exists in the destination") - - anchor = src.body[start + len(names)] - matches = [node for node in dst.body if ast.dump(node) == ast.dump(anchor)] - if len(matches) != 1: - raise ValueError("Expected one unchanged destination anchor") - for tree, node in ((src, anchor), (dst, matches[0])): - index = tree.body.index(node) - if ( - getattr(node, "decorator_list", ()) - or node.col_offset - or index - and tree.body[index - 1].end_lineno >= node.lineno - or index + 1 < len(tree.body) - and tree.body[index + 1].lineno <= node.end_lineno - ): - raise ValueError( - "Anchor must be undecorated and occupy complete source lines" - ) - available = set(vars(builtins)) - imports = (set(), set()) - for node in src.body: - matching = [ - other - for other in dst.body - if ast.dump(node) == ast.dump(other) and other.lineno < matches[0].lineno - ] - if ( - isinstance(node, (ast.Import, ast.ImportFrom)) - and node.lineno < selected[0].lineno - and len(matching) == 1 - ): - imports[0].update(node.names) - imports[1].update(matching[0].names) - available.update( - alias.asname or alias.name.split(".")[0] for alias in node.names - ) - dependencies = set() - for name, node in zip(names, selected): - used = { - child.id - for child in ast.walk(node.value) - if isinstance(child, ast.Name) and isinstance(child.ctx, ast.Load) - } - missing = used - available - if missing: - raise ValueError(f"Missing dependencies for {name}: {sorted(missing)}") - dependencies.update(used - set(names)) - available.add(name) - for tree, boundary, allowed in ( - (src, selected[0].lineno, imports[0]), - (dst, matches[0].lineno, imports[1]), - ): - if dependencies and any( - isinstance(node, ast.ImportFrom) - and node.lineno < boundary - and any(alias.name == "*" for alias in node.names) - for node in ast.walk(tree) - ): - raise ValueError("Wildcard import makes dependency bindings ambiguous") - for dependency in dependencies: - bindings = [] - for node in ast.walk(tree): - if getattr(node, "lineno", boundary) >= boundary: - continue - if isinstance(node, ast.Name) and isinstance( - node.ctx, (ast.Store, ast.Del) - ): - name = node.id - elif isinstance(node, ast.alias): - name = node.asname or node.name.split(".")[0] - elif isinstance( - node, - ( - ast.FunctionDef, - ast.AsyncFunctionDef, - ast.ClassDef, - ast.ExceptHandler, - ast.MatchAs, - ast.MatchStar, - ), - ): - name = node.name - elif isinstance(node, ast.arg): - name = node.arg - elif isinstance(node, ast.MatchMapping): - name = node.rest - else: - continue - if name == dependency: - bindings.append(node) - if len(bindings) > 1 or any(node not in allowed for node in bindings): - raise ValueError(f"Ambiguous dependency binding: {dependency}") - - source_lines, lines = source.splitlines(True), destination.splitlines(True) - additions = [ - ( - matches[0].lineno - 1, - source_lines[selected[0].lineno - 1 : anchor.lineno - 1], - ) - ] - if exports: - additions.append( - _selected_export_addition(src, dst, source_lines, lines, exports) - ) - for offset, fragment in sorted(additions, reverse=True): - lines[offset:offset] = fragment - result = "".join(lines) - ast.parse(result) - return result - - -def _module_bindings(tree): - """Collect possible module bindings without mistaking function locals for them.""" - bindings = collections.defaultdict(list) - - def visit(node): - children = ast.iter_child_nodes(node) - names = [] - if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)): - names = [node.name] - # A declaration in a nested scope can rebind a module dependency. - for child in ast.walk(node): - if isinstance(child, ast.Global): - for name in child.names: - bindings[name].append(child) - children = list(node.decorator_list) - if isinstance(node, ast.ClassDef): - children += node.bases + node.keywords - else: - children += node.args.defaults + [ - value for value in node.args.kw_defaults if value is not None - ] - elif isinstance(node, ast.Lambda): - children = node.args.defaults + [ - value for value in node.args.kw_defaults if value is not None - ] - elif isinstance(node, ast.Name) and isinstance(node.ctx, (ast.Store, ast.Del)): - names = [node.id] - elif isinstance(node, ast.alias): - names = [node.asname or node.name.split(".")[0]] - elif isinstance(node, (ast.ExceptHandler, ast.MatchAs, ast.MatchStar)): - names = [node.name] if node.name is not None else [] - elif isinstance(node, ast.MatchMapping) and node.rest is not None: - names = [node.rest] - for name in names: - bindings[name].append(node) - for child in children: - visit(child) - - visit(tree) - return bindings - - -def _function_dependencies(fragment): - """Find definition-time and deferred global reads without executing source.""" - table = symtable.symtable(fragment, "", "exec") - if any( - symbol.is_assigned() and not symbol.is_namespace() - for symbol in table.get_symbols() - ): - raise ValueError("Function headers cannot bind additional module names") - immediate = { - symbol.get_name() for symbol in table.get_symbols() if symbol.is_referenced() - } - deferred = set() - - def visit(scope): - for symbol in scope.get_symbols(): - if symbol.is_global(): - if symbol.is_assigned() or symbol.is_imported(): - raise ValueError("Selected functions cannot rebind module globals") - if symbol.is_referenced(): - deferred.add(symbol.get_name()) - for child in scope.get_children(): - visit(child) - - for child in table.get_children(): - visit(child) - if (immediate | deferred) & {"eval", "exec", "globals", "locals", "vars"}: - raise ValueError("Dynamic namespace dependencies require a full migration") - return immediate, deferred - - -def copy_selected_functions(source, destination, names, exports): - """Add contiguous complete functions, preserving unrelated destination bytes. - - Global dependencies must be builtins, selected functions, or unambiguously - bound identical imports/simple assignments (checked transitively). Existing - helper functions and dynamic namespace dependencies require a full migration. - Scoped expressions in definition-time headers and classes anywhere inside - selected functions also require a full migration. Symbol-table analysis - distinguishes body locals, closures and comprehensions from module - dependencies; it never imports or calls the selected functions. - """ - if ( - not names - or len(set(names)) != len(names) - or not all(name.isidentifier() for name in names) - ): - raise ValueError("Function names must be unique identifiers") - if len(set(exports)) != len(exports) or not set(exports) <= set(names): - raise ValueError("Exports must be unique selected function names") - src, dst = ast.parse(source), ast.parse(destination) - src_bindings, dst_bindings = _module_bindings(src), _module_bindings(dst) - selected = [] - for name in names: - matches = src_bindings.get(name, []) - if ( - len(matches) != 1 - or not isinstance(matches[0], (ast.FunctionDef, ast.AsyncFunctionDef)) - or matches[0] not in src.body - or getattr(matches[0], "type_params", ()) - ): - raise ValueError(f"Expected one top-level non-generic function: {name}") - if dst_bindings.get(name): - raise ValueError( - f"A selected function already exists in the destination: {name}" - ) - node = matches[0] - headers = ( - node.decorator_list - + node.args.defaults - + [value for value in node.args.kw_defaults if value is not None] - ) - arguments = node.args.posonlyargs + node.args.args + node.args.kwonlyargs - arguments += [arg for arg in (node.args.vararg, node.args.kwarg) if arg] - headers += [arg.annotation for arg in arguments if arg.annotation is not None] - if node.returns is not None: - headers.append(node.returns) - if any( - isinstance( - child, - ( - ast.Lambda, - ast.ListComp, - ast.SetComp, - ast.DictComp, - ast.GeneratorExp, - ), - ) - for header in headers - for child in ast.walk(header) - ): - raise ValueError( - "Scoped expressions in function headers require a full migration" - ) - if any(isinstance(child, ast.ClassDef) for child in ast.walk(node)): - raise ValueError("Classes in selected functions require a full migration") - selected.append(node) - start = src.body.index(selected[0]) - if src.body[start : start + len(names)] != selected or start + len(names) == len( - src.body - ): - raise ValueError("Functions must be contiguous and followed by an anchor") - first_line = min( - [selected[0].lineno, *(node.lineno for node in selected[0].decorator_list)] - ) - anchor = src.body[start + len(names)] - matches = [node for node in dst.body if ast.dump(node) == ast.dump(anchor)] - if len(matches) != 1: - raise ValueError("Expected one unchanged destination anchor") - for tree, node in ((src, anchor), (dst, matches[0])): - index = tree.body.index(node) - if ( - getattr(node, "decorator_list", ()) - or node.col_offset - or index - and tree.body[index - 1].end_lineno >= node.lineno - or index + 1 < len(tree.body) - and tree.body[index + 1].lineno <= node.end_lineno - ): - raise ValueError( - "Anchor must be undecorated and occupy complete source lines" - ) - if start and src.body[start - 1].end_lineno >= first_line: - raise ValueError("Selected functions must occupy complete source lines") - - source_lines, lines = source.splitlines(True), destination.splitlines(True) - futures = [ - [ - node - for node in tree.body - if isinstance(node, ast.ImportFrom) and node.module == "__future__" - ] - for tree in (src, dst) - ] - if [ast.dump(node) for node in futures[0]] != [ - ast.dump(node) for node in futures[1] - ]: - raise ValueError("Source and destination future imports differ") - prefix = "".join( - "".join(source_lines[node.lineno - 1 : node.end_lineno]) for node in futures[0] - ) - if "*" in src_bindings or "*" in dst_bindings: - raise ValueError("Wildcard import makes dependency bindings ambiguous") - - def dependency(name, source_boundary, destination_boundary): - if name in {"eval", "exec", "globals", "locals", "vars"}: - raise ValueError("Dynamic namespace dependencies require a full migration") - declarations = [] - for tree, bindings, boundary in ( - (src, src_bindings, source_boundary), - (dst, dst_bindings, destination_boundary), - ): - bound = bindings.get(name, []) - if not bound and name in vars(builtins): - declarations.append(None) - continue - if len(bound) != 1: - raise ValueError(f"Missing or ambiguous dependency binding: {name}") - candidates = [ - node - for node in tree.body - if isinstance(node, (ast.Import, ast.ImportFrom)) - and bound[0] in node.names - or isinstance(node, ast.Assign) - and len(node.targets) == 1 - and node.targets[0] is bound[0] - ] - if len(candidates) != 1 or candidates[0].lineno >= boundary: - raise ValueError( - f"Expected an earlier import or simple assignment: {name}" - ) - declarations.append(candidates[0]) - left, right = declarations - if left is None and right is None: - return - if left is None or right is None or ast.dump(left) != ast.dump(right): - raise ValueError(f"Changed dependency binding: {name}") - if isinstance(left, ast.Assign): - for child in ast.walk(left.value): - if isinstance( - child, - ( - ast.Lambda, - ast.NamedExpr, - ast.ListComp, - ast.SetComp, - ast.DictComp, - ast.GeneratorExp, - ), - ): - raise ValueError(f"Unsupported dependency expression: {name}") - if isinstance(child, ast.Name) and isinstance(child.ctx, ast.Load): - dependency(child.id, left.lineno, right.lineno) - - earlier = set() - for node in selected: - line = min([node.lineno, *(item.lineno for item in node.decorator_list)]) - immediate, deferred = _function_dependencies( - prefix + "".join(source_lines[line - 1 : node.end_lineno]) - ) - if (immediate & set(names)) - earlier: - raise ValueError("Definition-time dependency is not yet bound") - for name in (immediate | deferred) - set(names): - dependency(name, first_line, matches[0].lineno) - earlier.add(node.name) - - # Keep a following anchor comment attached to its original statement. Only - # canonical function bytes and their trailing whitespace/comments are added. - end = anchor.lineno - 1 - comment = next( - ( - i - for i in range(selected[-1].end_lineno, end) - if source_lines[i].startswith("#") - ), - end, - ) - anchor_comments = source_lines[comment:end] - offset = matches[0].lineno - 1 - len(anchor_comments) - if offset < 0 or lines[offset : matches[0].lineno - 1] != anchor_comments: - raise ValueError("Anchor's leading comments differ") - additions = [(offset, source_lines[first_line - 1 : comment])] - if exports: - for tree, bindings in ((src, src_bindings), (dst, dst_bindings)): - if bindings.get("__all__") != [ - _single_assignment(tree, "__all__").targets[0] - ]: - raise ValueError("Ambiguous __all__ binding") - additions.append( - _selected_export_addition(src, dst, source_lines, lines, exports) - ) - for offset, fragment in sorted(additions, reverse=True): - lines[offset:offset] = fragment - result = "".join(lines) - ast.parse(result) - return result - - -def _migrate_selected( - path, names, exports, source_hash, destination_hash, kind, copier -): - target = parse_lib_path(path) - if target.parts[:1] != ("Lib",) or ".." in target.parts or target.suffix != ".py": - raise ValueError("Selective imports require an explicit Lib Python file") - source, destination = path.read_bytes(), target.read_bytes() - for label, contents, expected in ( - ("source", source, source_hash), - ("destination", destination, destination_hash), - ): - if hashlib.sha256(contents).hexdigest() != expected: - raise ValueError(f"{label} SHA256 does not match the pinned input") - result = copier( - source.decode("utf-8"), destination.decode("utf-8"), names, exports - ).encode("utf-8") - if target.read_bytes() != destination: - raise ValueError("Destination changed during selective import preparation") - target.write_bytes(result) - print( - json.dumps( - { - kind: names, - "exports": exports, - "source_sha256": source_hash, - "destination_before_sha256": destination_hash, - "destination_after_sha256": hashlib.sha256(result).hexdigest(), - }, - sort_keys=True, - ) - ) - - -def migrate_selected_assignments(path, names, exports, source_hash, destination_hash): - return _migrate_selected( - path, - names, - exports, - source_hash, - destination_hash, - "assignments", - copy_selected_assignments, - ) - - -def migrate_selected_functions(path, names, exports, source_hash, destination_hash): - return _migrate_selected( - path, - names, - exports, - source_hash, - destination_hash, - "functions", - copy_selected_functions, - ) - - def patch_single_content( src_path: pathlib.Path, lib_path: pathlib.Path, @@ -913,159 +346,6 @@ def migrate_package( legacy.unlink() -def remove_canonical_files(src_dir, manifest_path, *, dry_run=False): - """Remove only explicitly reviewed files, validating the whole batch first.""" - - def unique_object(pairs): - result = {} - for key, value in pairs: - if key in result: - raise ValueError(f"Duplicate removal metadata key: {key}") - result[key] = value - return result - - def relative(name): - path = pathlib.PurePosixPath(name) if isinstance(name, str) else None - if ( - not path - or not path.parts - or path.is_absolute() - or ".." in path.parts - or "\\" in name - or path.as_posix() != name - or "\0" in name - ): - raise ValueError("Expected an exact relative POSIX filename") - return pathlib.Path(path) - - def no_symlinks(path): - if any(p.is_symlink() or p.is_junction() for p in (path, *path.parents)): - raise ValueError(f"Symlink or junction in removal path: {path}") - - def git(*args): - result = subprocess.run( - ["git", "-C", str(src_dir), *args], - env={ - **os.environ, - "GIT_NO_LAZY_FETCH": "1", - "GIT_NO_REPLACE_OBJECTS": "1", - "GIT_TERMINAL_PROMPT": "0", - }, - capture_output=True, - check=False, - ) - if result.returncode: - raise ValueError("Cannot verify pinned canonical source") - return result.stdout.decode().strip() - - document = json.loads( - manifest_path.read_text(encoding="utf-8"), object_pairs_hook=unique_object - ) - if not isinstance(document, dict) or set(document) != {"source", "removals"}: - raise ValueError("Expected source and removals metadata") - source, removals = document["source"], document["removals"] - if not isinstance(source, dict) or set(source) != {"path", "commit", "tree"}: - raise ValueError("Expected source path, commit and tree") - package = relative(source["path"]) - if ( - len(package.parts) != 3 - or package.parts[:2] != ("Lib", "test") - or not package.name.startswith("test_") - or not package.name.isidentifier() - ): - raise ValueError("Removals require an exact Lib/test/test_* package") - if any( - not isinstance(source[k], str) or not re.fullmatch(r"[0-9a-f]{40}", source[k]) - for k in ("commit", "tree") - ): - raise ValueError("Source commit and tree must be full Git hashes") - if not isinstance(removals, dict) or not removals: - raise ValueError("Removals must be a nonempty filename map") - src_dir = src_dir.absolute() - destination = pathlib.Path.cwd() / package - for path in (src_dir, destination): - no_symlinks(path) - if not path.is_dir(): - raise ValueError("Source and destination packages must exist") - if ( - src_dir == destination - or src_dir in destination.parents - or destination in src_dir.parents - ): - raise ValueError("Source and destination overlap") - if ( - src_dir != pathlib.Path(git("rev-parse", "--show-toplevel")) / package - or git("rev-parse", "HEAD") != source["commit"] - or git("rev-parse", f"HEAD:{package.as_posix()}") != source["tree"] - ): - raise ValueError("Canonical source identity differs from the pinned input") - canonical = set( - git( - "ls-tree", - "--full-tree", - "-r", - "--name-only", - "-z", - "HEAD", - "--", - package.as_posix(), - ).split("\0") - ) - snapshots, receipt = {}, {"source": source, "dry_run": dry_run, "removals": {}} - for name, entry in sorted(removals.items()): - path = relative(name) - if ( - not isinstance(entry, dict) - or set(entry) != {"sha256", "upstream_commit"} - or any( - not isinstance(entry[k], str) - or not re.fullmatch(f"[0-9a-f]{{{length}}}", entry[k]) - for k, length in (("sha256", 64), ("upstream_commit", 40)) - ) - ): - raise ValueError(f"Missing or invalid hash/provenance pins: {name}") - target = destination / path - no_symlinks(target) - no_symlinks(src_dir / path) - if (package / path).as_posix() in canonical or (src_dir / path).exists(): - raise ValueError(f"Selected file is still canonical: {name}") - if target.exists() and not target.is_file(): - raise ValueError(f"Removal target is not a regular file: {name}") - data = target.read_bytes() if target.exists() else None - if data is not None and ( - hashlib.sha256(data).hexdigest() != entry["sha256"] - or b"rustpython" in data.lower() - ): - raise ValueError(f"Changed bytes or RustPython customization: {name}") - snapshots[target] = data - receipt["removals"][name] = { - **entry, - "status": "already_absent" - if data is None - else "would_remove" - if dry_run - else "removed", - } - if git("rev-parse", "HEAD") != source["commit"]: - raise ValueError("Canonical source changed during validation") - for target, data in snapshots.items(): - no_symlinks(target) - canonical_path = src_dir / target.relative_to(destination) - no_symlinks(canonical_path) - if canonical_path.exists(): - raise ValueError("Canonical source changed during validation") - if (target.exists() and not target.is_file()) or ( - target.read_bytes() if target.exists() else None - ) != data: - raise ValueError(f"Removal target changed during validation: {target}") - if not dry_run: - for target, data in snapshots.items(): - if data is not None: - target.unlink() - print(json.dumps(receipt, sort_keys=True)) - return receipt - - def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser( description=__doc__, @@ -1082,88 +362,14 @@ def main(argv: list[str] | None = None) -> int: type=pathlib.Path, help="Explicit legacy-file/class to package-file JSON mapping", ) - parser.add_argument( - "--canonical-removals", - type=pathlib.Path, - help="Remove only selected unchanged historical files absent from a pinned canonical tree", - ) - parser.add_argument( - "--dry-run", - action="store_true", - help="Validate canonical removals without deleting", - ) - parser.add_argument( "--patches", type=pathlib.Path, help="Single-file target, remaining decorator patches, and optional doctests JSON", ) - parser.add_argument( - "--assignments", - nargs="+", - help="Copy selected canonical assignments", - ) - parser.add_argument( - "--functions", - nargs="+", - help="Copy complete selected canonical functions without executing them", - ) - parser.add_argument("--exports", nargs="+", default=[]) - parser.add_argument("--source-sha256") - parser.add_argument("--destination-sha256") args = parser.parse_args(argv) try: - if args.canonical_removals is not None: - if ( - args.legacy_map is not None - or args.patches is not None - or args.assignments - or args.functions - or args.exports - or args.source_sha256 - or args.destination_sha256 - ): - raise ValueError( - "Canonical removals cannot combine with other migration modes" - ) - try: - remove_canonical_files( - args.path, args.canonical_removals, dry_run=args.dry_run - ) - except (OSError, UnicodeError) as exc: - raise ValueError(str(exc)) from exc - return 0 - if args.dry_run: - raise ValueError("--dry-run requires --canonical-removals") - if args.assignments or args.functions: - if ( - args.legacy_map is not None - or args.patches is not None - or args.assignments - and args.functions - ): - raise ValueError( - "Selective imports cannot combine with other migration modes" - ) - try: - migrate = ( - migrate_selected_functions - if args.functions - else migrate_selected_assignments - ) - migrate( - args.path, - args.functions or args.assignments, - args.exports, - args.source_sha256, - args.destination_sha256, - ) - except (SyntaxError, UnicodeError) as exc: - raise ValueError(str(exc)) from exc - return 0 - if args.exports or args.source_sha256 or args.destination_sha256: - raise ValueError("Selective options require --assignments or --functions") if args.patches is not None and ( args.path.is_dir() or args.legacy_map is not None ): diff --git a/scripts/update_lib/tests/test_copy_test.py b/scripts/update_lib/tests/test_copy_test.py deleted file mode 100644 index 5dc61452c..000000000 --- a/scripts/update_lib/tests/test_copy_test.py +++ /dev/null @@ -1,206 +0,0 @@ -"""Tests for bounded, hash-pinned canonical test copying.""" - -import ast -import contextlib -import hashlib -import io -import tempfile -import unittest -from pathlib import Path - -from update_lib.cmd_copy_test import copy_test, main - - -class TestCopyTest(unittest.TestCase): - destination = '''# Destination-only header -class Finder: - if int.__doc__: - def example(): - r"""Count is below 800. π\\n""" - - def untouched(self): - return "destination" -''' - source = '''# Canonical header must not be copied -class Finder: - if int.__doc__: - def example(): - r"""Count is below 850. π\\n""" - - def untouched(self): - return "canonical" -''' - - def test_complete_definition_and_surroundings(self): - result = copy_test(self.source, self.destination, "Finder.example") - self.assertEqual(result, self.destination.replace("below 800", "below 850")) - self.assertIn('return "destination"', result) - source_method = ast.parse(self.source).body[0].body[0].body[0] - result_method = ast.parse(result).body[0].body[0].body[0] - self.assertEqual( - ast.get_source_segment(self.source, source_method), - ast.get_source_segment(result, result_method), - ) - - def test_guard_mismatch_is_rejected(self): - with self.assertRaisesRegex(ValueError, "guards differ"): - copy_test( - self.source.replace("if int.__doc__", "if enabled"), - self.destination, - "Finder.example", - ) - - def test_missing_or_ambiguous_definition_is_rejected(self): - for source in ("class Other: pass\n", self.source + self.source): - with self.subTest(source=source): - with self.assertRaisesRegex(ValueError, "Expected one class"): - copy_test(source, self.destination, "Finder.example") - duplicate = self.source.replace( - " def untouched(self):", - " def example():\n pass\n\n def untouched(self):", - ) - with self.assertRaisesRegex(ValueError, "Expected one method"): - copy_test(duplicate, self.destination, "Finder.example") - - def test_decorator_and_signature_changes_are_rejected(self): - decorated = self.source.replace( - " def example():", " @decorator\n def example():" - ) - with self.assertRaisesRegex(ValueError, "decorators differ"): - copy_test(decorated, self.destination, "Finder.example") - with self.assertRaisesRegex(ValueError, "signatures differ"): - copy_test( - self.source.replace("def example():", "def example(arg):"), - self.destination, - "Finder.example", - ) - - def test_nested_local_scope_is_not_selected(self): - local = "def factory():\n" + "".join( - " " + line for line in self.source.splitlines(True) - ) - with self.assertRaisesRegex(ValueError, "Expected one class"): - copy_test(local, self.destination, "Finder.example") - - def test_decorator_comments_must_match_exactly(self): - for source_marker, destination_marker in ( - ( - "@unittest.expectedFailure", - "@unittest.expectedFailure # TODO: RUSTPYTHON; reason", - ), - ( - "@unittest.expectedFailure # original reason", - "@unittest.expectedFailure # changed reason", - ), - ( - "@unittest.expectedFailure\n @other", - "@unittest.expectedFailure\n # marker rationale\n @other", - ), - ( - "@unittest.expectedFailureIf(\n condition,\n 'reason',\n )", - "@unittest.expectedFailureIf(\n condition, # project limitation\n 'reason',\n )", - ), - ): - with self.subTest(destination_marker=destination_marker): - source = self.source.replace( - " def example():", - f" {source_marker}\n def example():", - ) - destination = self.destination.replace( - " def example():", - f" {destination_marker}\n def example():", - ) - with self.assertRaisesRegex(ValueError, "decorator text differs"): - copy_test(source, destination, "Finder.example") - - def test_identical_decorator_comments_are_preserved(self): - prefix = ( - " @unittest.expectedFailure # TODO: RUSTPYTHON; reason\n" - " # Keep this rationale with the marker.\n" - " def example():" - ) - for declaration in ("def example():", "async def example():"): - with self.subTest(declaration=declaration): - source = self.source.replace(" def example():", prefix) - destination = self.destination.replace(" def example():", prefix) - source = source.replace("def example():", declaration) - destination = destination.replace("def example():", declaration) - self.assertEqual( - copy_test(source, destination, "Finder.example"), - destination.replace("below 800", "below 850"), - ) - - def test_cli_does_not_write_on_decorator_comment_mismatch(self): - with tempfile.TemporaryDirectory() as temp: - source = Path(temp) / "source.py" - destination = Path(temp) / "destination.py" - source.write_text( - self.source.replace( - " def example():", - " @unittest.expectedFailure\n def example():", - ) - ) - destination.write_text( - self.destination.replace( - " def example():", - " @unittest.expectedFailure # TODO: RUSTPYTHON; reason\n" - " def example():", - ) - ) - before = destination.read_bytes() - with ( - contextlib.redirect_stderr(io.StringIO()) as errors, - self.assertRaises(SystemExit), - ): - main( - [ - str(source), - str(destination), - "--test", - "Finder.example", - "--source-sha256", - hashlib.sha256(source.read_bytes()).hexdigest(), - "--destination-sha256", - hashlib.sha256(before).hexdigest(), - ] - ) - self.assertIn("decorator text differs", errors.getvalue()) - self.assertEqual(destination.read_bytes(), before) - - def test_return_annotation_and_type_parameters_are_rejected(self): - for declaration in ("def example() -> int:", "def example[T]():"): - with self.subTest(declaration=declaration): - source = self.source.replace("def example():", declaration) - with self.assertRaisesRegex(ValueError, "signatures differ"): - copy_test(source, self.destination, "Finder.example") - - def test_cli_hashes_and_no_write_on_stale_input(self): - with tempfile.TemporaryDirectory() as temp: - source = Path(temp) / "source.py" - destination = Path(temp) / "destination.py" - source.write_bytes(self.source.encode()) - destination.write_bytes(self.destination.encode()) - args = [ - str(source), - str(destination), - "--test", - "Finder.example", - "--source-sha256", - hashlib.sha256(source.read_bytes()).hexdigest(), - "--destination-sha256", - "0" * 64, - ] - with ( - contextlib.redirect_stderr(io.StringIO()), - self.assertRaises(SystemExit), - ): - main(args) - self.assertEqual(destination.read_bytes(), self.destination.encode()) - args[-1] = hashlib.sha256(destination.read_bytes()).hexdigest() - with contextlib.redirect_stdout(io.StringIO()) as output: - self.assertEqual(main(args), 0) - self.assertIn('"source_sha256"', output.getvalue()) - self.assertEqual( - destination.read_text(), - self.destination.replace("below 800", "below 850"), - ) diff --git a/scripts/update_lib/tests/test_migrate.py b/scripts/update_lib/tests/test_migrate.py index 1cf5359b5..97d959571 100644 --- a/scripts/update_lib/tests/test_migrate.py +++ b/scripts/update_lib/tests/test_migrate.py @@ -1,12 +1,9 @@ """Tests for migrate.py - file migration operations.""" import contextlib -import hashlib import io import json -import os import pathlib -import subprocess import tempfile import unittest from pathlib import Path @@ -14,13 +11,11 @@ from unittest.mock import patch from update_lib.cmd_migrate import ( _preserve_block, - copy_selected_assignments, main, migrate_package, patch_directory, patch_file, patch_single_content, - remove_canonical_files, ) from update_lib.patch_spec import ( CANONICAL_LOAD_TESTS, @@ -34,233 +29,6 @@ from update_lib.patch_spec import ( ) -class TestSelectedAssignments(unittest.TestCase): - names = ["_available", "requires_available", "requires_unavailable"] - exports = names[1:] - imports = "import os\nimport unittest\n\n" - export_line = ' "requires_available", "requires_unavailable",\n' - block = ( - "_available = hasattr(os, 'geteuid') and os.geteuid() == 0\n" - 'requires_available = unittest.skipUnless(_available, "needs privilege")\n' - 'requires_unavailable = unittest.skipIf(_available, "needs no privilege")\n\n\n' - ) - source = ( - imports - + '__all__ = [\n "existing",\n' - + export_line - + "]\n\n\n" - + block - + "STATUS = 1\n" - ) - destination = ( - imports - + '__all__ = [\n "existing",\n]\n\n\ndef legacy():\n return "keep π"\n\n\nSTATUS = 1\n' - ) - - def test_canonical_bytes_and_unrelated_helpers_are_preserved(self): - result = copy_selected_assignments( - self.source, self.destination, self.names, self.exports - ) - self.assertEqual( - result, - self.destination.replace( - "__all__ = [\n", "__all__ = [\n" + self.export_line - ).replace("STATUS = 1\n", self.block + "STATUS = 1\n"), - ) - - def test_missing_ambiguous_and_noncontiguous_assignments_are_rejected(self): - for source in ( - self.source.replace("_available =", "_other ="), - self.source + "_available = False\n", - self.source + "def requires_available(): pass\n", - self.source + "import other as requires_unavailable\n", - self.source.replace( - "requires_available =", "unselected = 0\nrequires_available =" - ), - self.source.replace("_available =", "unselected = 0; _available =", 1), - self.source.replace("STATUS = 1", ""), - ): - with self.subTest(source=source), self.assertRaises(ValueError): - copy_selected_assignments( - source, self.destination, self.names, self.exports - ) - - def test_existing_bindings_dependencies_and_anchor_fail_closed(self): - for destination in ( - self.destination + "_available = False\n", - self.destination + "def requires_available(): pass\n", - self.destination + "import other as requires_unavailable\n", - self.destination.replace("import os\n", ""), - self.destination.replace("import os\n", "import os as other\n"), - self.destination.replace("import os\n", "") + "import os\n", - self.destination.replace("STATUS = 1", "STATUS = 2"), - self.destination + "STATUS = 1\n", - ): - with self.subTest(destination=destination), self.assertRaises(ValueError): - copy_selected_assignments( - self.source, destination, self.names, self.exports - ) - - def test_invalid_exports_are_rejected(self): - for source, destination, exports in ( - (self.source, self.destination, self.exports + self.exports), - (self.source, self.destination, ["unselected"]), - (self.source, self.destination, list(reversed(self.exports))), - (self.source, self.destination, self.exports[:1]), - ( - self.source, - self.destination.replace('"existing"', '"requires_available"'), - self.exports, - ), - (self.source.replace(self.export_line, ""), self.destination, self.exports), - ( - self.source, - self.destination.replace('[\n "existing",\n]', '("existing",)'), - self.exports, - ), - (self.source.replace('"existing"', "3"), self.destination, self.exports), - (self.source, self.destination.replace('"existing"', "3"), self.exports), - ): - with self.subTest(exports=exports), self.assertRaises(ValueError): - copy_selected_assignments(source, destination, self.names, exports) - - def test_dependency_rebinding_deletion_and_builtin_shadowing_are_rejected(self): - for binding in ( - "os = replacement", - "del os", - "import other as os", - "hasattr = replacement", - "del hasattr", - "if condition:\n unittest = replacement", - ): - for source_side in (False, True): - source, destination = self.source, self.destination - if source_side: - source = source.replace("__all__ = [", binding + "\n__all__ = [") - else: - destination = destination.replace( - "__all__ = [", binding + "\n__all__ = [" - ) - with ( - self.subTest(binding=binding, source_side=source_side), - self.assertRaisesRegex(ValueError, "Ambiguous dependency binding"), - ): - copy_selected_assignments( - source, destination, self.names, self.exports - ) - - def test_decorated_and_partial_line_anchors_are_rejected(self): - for replacement in ( - "@decorator\ndef anchor(): pass", - "STATUS = 1; other = 2", - "other = 2; STATUS = 1", - ): - with self.subTest(replacement=replacement), self.assertRaises(ValueError): - copy_selected_assignments( - self.source.replace("STATUS = 1", replacement), - self.destination.replace("STATUS = 1", replacement), - self.names, - self.exports, - ) - - def test_wildcard_import_dependencies_are_rejected(self): - for source, destination in ( - ("from other import *\n" + self.source, self.destination), - (self.source, "from other import *\n" + self.destination), - ): - with self.assertRaisesRegex(ValueError, "Wildcard import"): - copy_selected_assignments(source, destination, self.names, self.exports) - - def test_cli_hash_pins_scope_and_receipt(self): - with tempfile.TemporaryDirectory() as temp, contextlib.chdir(temp): - source = pathlib.Path("canonical/Lib/example.py") - target = pathlib.Path("Lib/example.py") - source.parent.mkdir(parents=True) - target.parent.mkdir(parents=True) - source.write_text(self.source) - target.write_text(self.destination) - args = [ - str(source), - "--assignments", - *self.names, - "--exports", - *self.exports, - "--source-sha256", - hashlib.sha256(source.read_bytes()).hexdigest(), - "--destination-sha256", - "0" * 64, - ] - with contextlib.redirect_stderr(io.StringIO()): - self.assertEqual(main(args), 1) - self.assertEqual(main([str(source), "--assignments", *self.names]), 1) - self.assertEqual(target.read_text(), self.destination) - args[-1] = hashlib.sha256(target.read_bytes()).hexdigest() - with contextlib.redirect_stdout(io.StringIO()) as output: - self.assertEqual(main(args), 0) - self.assertIn('"destination_after_sha256"', output.getvalue()) - self.assertIn(self.block, target.read_text()) - for path in ( - "canonical/Lib/example.txt", - "canonical/Lib/../outside.py", - ): - with contextlib.redirect_stderr(io.StringIO()): - self.assertEqual(main([path, *args[1:]]), 1) - - def test_selective_mode_rejects_existing_modes_before_reading_metadata(self): - for mode in ("--patches", "--legacy-map"): - with contextlib.redirect_stderr(io.StringIO()) as errors: - self.assertEqual( - main( - [ - "missing/Lib/example.py", - "--assignments", - *self.names, - mode, - "missing-metadata.json", - ] - ), - 1, - ) - self.assertIn("cannot combine", errors.getvalue()) - with contextlib.redirect_stderr(io.StringIO()) as errors: - self.assertEqual( - main( - [ - "missing/Lib/example.py", - "--exports", - *self.exports, - mode, - "missing-metadata.json", - ] - ), - 1, - ) - self.assertIn("require --assignments", errors.getvalue()) - - def test_default_syntax_error_behavior_is_unchanged(self): - with patch( - "update_lib.cmd_migrate.patch_file", side_effect=SyntaxError("invalid") - ): - with self.assertRaises(SyntaxError): - main(["missing/Lib/example.py"]) - with patch( - "update_lib.cmd_migrate.migrate_selected_assignments", - side_effect=SyntaxError("invalid"), - ): - with contextlib.redirect_stderr(io.StringIO()) as errors: - self.assertEqual( - main( - [ - "missing/Lib/example.py", - "--assignments", - *self.names, - ] - ), - 1, - ) - self.assertIn("invalid", errors.getvalue()) - - class TestPatchSingleContent(unittest.TestCase): """Tests for patch_single_content function.""" @@ -1057,268 +825,5 @@ class TestMigratePackage(unittest.TestCase): self.assert_rejected() -class TestCanonicalRemovals(unittest.TestCase): - def setUp(self): - temporary = tempfile.TemporaryDirectory() - self.addCleanup(temporary.cleanup) - self.root = Path(temporary.name) - self.source = self.root / "canonical/Lib/test/test_example" - self.target = self.root / "local/Lib/test/test_example" - self.source.mkdir(parents=True) - self.target.mkdir(parents=True) - (self.source / "__init__.py").write_bytes(b"") - self.git("init", "-q") - self.git("add", ".") - self.git("commit", "-qm", "canonical test fixture") - self.document = { - "source": { - "path": "Lib/test/test_example", - "commit": self.git("rev-parse", "HEAD"), - "tree": self.git("rev-parse", "HEAD:Lib/test/test_example"), - }, - "removals": {}, - } - for name, data in (("a.py", b"old\n"), ("data/z.bin", b"\x00\xff")): - path = self.target / name - path.parent.mkdir(exist_ok=True) - path.write_bytes(data) - self.document["removals"][name] = { - "sha256": hashlib.sha256(data).hexdigest(), - "upstream_commit": "1" * 40, - } - self.manifest = self.root / "removals.json" - - def git(self, *args): - return subprocess.check_output( - ["git", "-C", str(self.source.parents[2]), *args], - env={ - **os.environ, - "GIT_AUTHOR_NAME": "Fixture", - "GIT_AUTHOR_EMAIL": "fixture@example.invalid", - "GIT_COMMITTER_NAME": "Fixture", - "GIT_COMMITTER_EMAIL": "fixture@example.invalid", - }, - text=True, - ).strip() - - def run_removal(self, dry_run=False): - self.manifest.write_text(json.dumps(self.document)) - with ( - contextlib.chdir(self.target.parents[2]), - contextlib.redirect_stdout(io.StringIO()), - ): - return remove_canonical_files(self.source, self.manifest, dry_run=dry_run) - - def test_dry_run_explicit_scope_receipt_and_idempotence(self): - (self.target / "unselected").write_text("keep") - for dry_run, expected in ( - (True, "would_remove"), - (False, "removed"), - (False, "already_absent"), - ): - receipt = self.run_removal(dry_run) - self.assertEqual(receipt["source"], self.document["source"]) - self.assertEqual( - {v["status"] for v in receipt["removals"].values()}, {expected} - ) - self.assertEqual((self.target / "a.py").exists(), dry_run) - self.assertEqual((self.target / "unselected").read_text(), "keep") - - def test_junction_components_reject_the_whole_batch(self): - before = { - name: (self.target / name).read_bytes() - for name in self.document["removals"] - } - for junction in ( - self.source, - self.source.parent, - self.source / "data", - self.target, - self.target.parent, - self.target / "data", - ): - for dry_run in (True, False): - with self.subTest(junction=junction, dry_run=dry_run): - self.assertFalse(junction.is_symlink()) - with ( - patch.object( - Path, - "is_junction", - autospec=True, - side_effect=lambda path: path == junction, - ), - self.assertRaisesRegex(ValueError, "[Jj]unction"), - ): - self.run_removal(dry_run) - self.assertEqual( - {name: (self.target / name).read_bytes() for name in before}, - before, - ) - - def test_junction_revalidation_prevents_partial_removal(self): - for junction in (self.target / "data", self.source / "data"): - checks = 0 - - def becomes_junction(path): - nonlocal checks - if path == junction: - checks += 1 - return checks > 1 - return False - - with self.subTest(junction=junction): - with ( - patch.object( - Path, "is_junction", autospec=True, side_effect=becomes_junction - ), - self.assertRaisesRegex(ValueError, "[Jj]unction"), - ): - self.run_removal() - self.assertEqual(checks, 2) - self.assertEqual((self.target / "a.py").read_bytes(), b"old\n") - self.assertEqual((self.target / "data/z.bin").read_bytes(), b"\x00\xff") - - def test_utf8_manifest_ignores_non_utf8_default_encoding(self): - name = "caf\u00e9.py" - target = self.target / name - target.write_bytes(b"old\n") - self.document["removals"][name] = dict(self.document["removals"]["a.py"]) - self.manifest.write_bytes( - json.dumps(self.document, ensure_ascii=False).encode("utf-8") - ) - with ( - patch( - "pathlib.io.text_encoding", - side_effect=lambda encoding, *args: ( - "latin-1" if encoding is None else encoding - ), - ), - contextlib.chdir(self.target.parents[2]), - contextlib.redirect_stdout(io.StringIO()), - ): - self.assertNotEqual( - self.manifest.read_text(), self.manifest.read_text(encoding="utf-8") - ) - for dry_run, expected in ( - (True, "would_remove"), - (False, "removed"), - (False, "already_absent"), - ): - receipt = remove_canonical_files( - self.source, self.manifest, dry_run=dry_run - ) - self.assertEqual( - set(receipt["removals"]), set(self.document["removals"]) - ) - self.assertEqual(receipt["removals"][name]["status"], expected) - self.assertEqual(target.exists(), dry_run) - - def test_cli_dry_run_never_migrates_or_changes_destination_bytes(self): - def snapshot(): - return { - p.relative_to(self.target): p.read_bytes() - for p in self.target.rglob("*") - if p.is_file() - } - - before = snapshot() - self.manifest.write_text(json.dumps(self.document)) - with contextlib.ExitStack() as stack: - stack.enter_context(contextlib.chdir(self.target.parents[2])) - stack.enter_context(contextlib.redirect_stdout(io.StringIO())) - for name in ( - "patch_file", - "patch_directory", - "migrate_package", - "migrate_selected_assignments", - ): - stack.enter_context( - patch( - f"update_lib.cmd_migrate.{name}", - side_effect=AssertionError("Dry-run must not migrate"), - ) - ) - self.assertEqual( - main( - [ - str(self.source), - "--canonical-removals", - str(self.manifest), - "--dry-run", - ] - ), - 0, - ) - self.assertEqual(snapshot(), before) - - def test_hash_customization_and_all_batch_preflight(self): - path = self.target / "data/z.bin" - for data, repin in ((b"changed", False), (b"# TODO: RUSTPYTHON", True)): - path.write_bytes(data) - if repin: - self.document["removals"]["data/z.bin"]["sha256"] = hashlib.sha256( - data - ).hexdigest() - with self.assertRaises(ValueError): - self.run_removal() - self.assertEqual((self.target / "a.py").read_bytes(), b"old\n") - - def test_canonical_source_identity_and_tracked_absence(self): - for key, value in ( - ("commit", "0" * 40), - ("tree", "0" * 40), - ("path", "Lib/test/test_other"), - ): - old = self.document["source"][key] - self.document["source"][key] = value - with self.subTest(key=key), self.assertRaises(ValueError): - self.run_removal() - self.document["source"][key] = old - (self.source / "__init__.py").unlink() - self.document["removals"]["__init__.py"] = self.document["removals"]["a.py"] - with self.assertRaisesRegex(ValueError, "still canonical"): - self.run_removal() - self.assertTrue((self.target / "a.py").exists()) - - def test_invalid_paths_symlinks_directories_and_empty_selection(self): - original = self.document["removals"] - for name in ("../a.py", "/a.py", "./a.py", "data//z.bin", "data\\z.bin"): - self.document["removals"] = {name: original["a.py"]} - with self.subTest(name=name), self.assertRaises(ValueError): - self.run_removal() - self.document["removals"] = {} - with self.assertRaises(ValueError): - self.run_removal() - self.document["removals"] = original - target = self.target / "a.py" - target.unlink() - target.symlink_to(self.target / "data/z.bin") - with self.assertRaisesRegex(ValueError, "Symlink"): - self.run_removal() - target.unlink() - target.mkdir() - with self.assertRaisesRegex(ValueError, "regular file"): - self.run_removal() - - def test_cli_refuses_combined_modes_and_duplicate_metadata(self): - for option in ("--patches", "--legacy-map", "--assignments"): - with contextlib.redirect_stderr(io.StringIO()): - self.assertEqual( - main( - [ - str(self.source), - "--canonical-removals", - "unused", - option, - "unused", - ] - ), - 1, - ) - self.manifest.write_text('{"source": {}, "source": {}}') - with self.assertRaisesRegex(ValueError, "Duplicate"): - remove_canonical_files(self.source, self.manifest) - - if __name__ == "__main__": unittest.main() diff --git a/scripts/update_lib/tests/test_selected_functions.py b/scripts/update_lib/tests/test_selected_functions.py deleted file mode 100644 index 0de48a63f..000000000 --- a/scripts/update_lib/tests/test_selected_functions.py +++ /dev/null @@ -1,320 +0,0 @@ -"""Source-only selective function migration, without importing copied helpers.""" - -import contextlib -import hashlib -import io -import json -import pathlib -import tempfile -import unittest -from unittest.mock import patch - -from update_lib.cmd_migrate import copy_selected_functions, main - - -class TestSelectedFunctions(unittest.TestCase): - names = ["available", "requires_available"] - exports = names - imports = "import functools\nimport os\nimport sys\nimport unittest\n\n" - export_line = ' "available", "requires_available",\n' - dependencies = 'IS_MOBILE = sys.platform == "mobile"\nAVAILABLE = not IS_MOBILE\n\n' - block = ( - "@functools.cache\n" - "def available(limit=1):\n" - " import errno\n" - " try:\n" - ' return AVAILABLE and bool(errno.EINVAL) and os.name != "" and limit > 0\n' - " except ValueError as error:\n" - " return bool(str(error))\n\n\n" - "def requires_available():\n" - ' return unittest.skipUnless(available(), "needs availability")\n\n\n' - ) - anchor = "# Keep this comment with STATUS.\nSTATUS = 1\n" - source = ( - imports - + '__all__ = [\n "existing",\n' - + export_line - + "]\n\n" - + dependencies - + block - + anchor - ) - destination = ( - imports - + '__all__ = [\n "existing",\n]\n\n' - + dependencies - + 'def existing(sys):\n return "keep π"\n\n' - + anchor - ) - - def copy(self, source=None, destination=None, names=None, exports=None): - return copy_selected_functions( - self.source if source is None else source, - self.destination if destination is None else destination, - self.names if names is None else names, - self.exports if exports is None else exports, - ) - - def test_exact_functions_exports_and_unrelated_bytes(self): - self.assertEqual( - self.copy(), - self.destination.replace( - "__all__ = [\n", "__all__ = [\n" + self.export_line - ).replace(self.anchor, self.block + self.anchor), - ) - self.assertEqual(self.copy().count(self.anchor), 1) - - def test_destination_line_endings_and_local_adaptations_survive(self): - source = self.source.replace(self.anchor, "STATUS = 1\n") - destination = self.destination.replace(self.anchor, "STATUS = 1\n").replace( - "\n", "\r\n" - ) - self.assertEqual( - self.copy(source, destination), - destination.replace( - "__all__ = [\r\n", "__all__ = [\r\n" + self.export_line - ).replace("STATUS = 1\r\n", self.block + "STATUS = 1\r\n"), - ) - - def test_local_imports_closures_and_comprehensions_are_not_global_dependencies( - self, - ): - block = ( - "@functools.cache\n" - "def available(items=()):\n" - " def inner(value):\n" - " return value in items\n" - " return [inner(value) for value in items]\n\n\n" - "async def requires_available():\n" - " import errno as local\n" - " return local.EINVAL\n\n\n" - ) - self.assertIn(block, self.copy(self.source.replace(self.block, block))) - with self.assertRaises(ValueError): - self.copy( - self.source.replace( - self.block, block.replace("value in items", "MISSING", 1) - ) - ) - - def test_multiple_decorators_defaults_and_body_are_copied_without_execution(self): - block = self.block.replace( - "@functools.cache\ndef available(limit=1):", - "@functools.cache\n@functools.lru_cache(\n maxsize=None\n)\n" - "def available(limit=os.name):", - ).replace( - " import errno\n", - " raise RuntimeError('must not run')\n import errno\n", - ) - self.assertIn(block, self.copy(self.source.replace(self.block, block))) - - def test_selection_and_existing_bindings_fail_closed(self): - for source, destination, names in ( - (self.source, self.destination, []), - (self.source, self.destination, self.names * 2), - (self.source, self.destination, list(reversed(self.names))), - (self.source, self.destination, ["missing"]), - (self.source + "available = None\n", self.destination, self.names), - (self.source, self.destination + "del available\n", self.names), - (self.source, self.destination + "import other as available\n", self.names), - ( - self.source, - self.destination + "if condition:\n available = None\n", - self.names, - ), - ( - self.source.replace( - "def requires_available", "UNSELECTED = 1\ndef requires_available" - ), - self.destination, - self.names, - ), - (self.source.replace(self.anchor, ""), self.destination, self.names), - ): - with self.subTest(names=names), self.assertRaises(ValueError): - self.copy(source, destination, names) - - def test_missing_changed_late_and_rebound_dependencies_fail_closed(self): - for destination in ( - self.destination.replace("import functools\n", ""), - self.destination.replace( - "import functools\n", "import other as functools\n" - ), - self.destination.replace("import os\n", "") + "import os\n", - self.destination.replace( - 'sys.platform == "mobile"', 'sys.platform == "other"' - ), - self.destination + "AVAILABLE = True\n", - self.destination + "del sys\n", - self.destination + "bool = None\n", - self.destination + "def rebind():\n global sys\n sys = None\n", - self.destination + "from unknown import *\n", - ): - with self.subTest(destination=destination), self.assertRaises(ValueError): - self.copy(destination=destination) - for source in ( - self.source + "del sys\n", - self.source + "from unknown import *\n", - self.source.replace("AVAILABLE = not IS_MOBILE", "AVAILABLE = MISSING"), - self.source.replace("limit=1", "limit=requires_available()"), - ): - with self.subTest(source=source), self.assertRaises(ValueError): - self.copy(source=source) - - def test_unsupported_namespace_changes_and_future_mismatch_fail_closed(self): - for statement in ( - "global AVAILABLE\n AVAILABLE = True", - "return globals()", - "return eval('AVAILABLE')", - ): - with self.subTest(statement=statement), self.assertRaises(ValueError): - self.copy( - self.source.replace( - " import errno", " " + statement + "\n import errno" - ) - ) - with self.assertRaises(ValueError): - self.copy("from __future__ import annotations\n" + self.source) - with self.assertRaises(ValueError): - self.copy(self.source.replace("limit=1", "limit=(EXTRA := 1)")) - - def test_anchor_ambiguity_partial_lines_and_changed_comments_fail_closed(self): - for destination in ( - self.destination.replace("STATUS = 1", "STATUS = 2"), - self.destination + "STATUS = 1\n", - self.destination.replace("STATUS = 1", "STATUS = 1; other = 2"), - self.destination.replace("STATUS = 1", "other = 2; STATUS = 1"), - self.destination.replace( - "# Keep this comment with STATUS.", "# Changed comment." - ), - ): - with self.subTest(destination=destination), self.assertRaises(ValueError): - self.copy(destination=destination) - replacement = "@functools.cache\ndef anchor(): pass\n" - with self.assertRaises(ValueError): - self.copy( - self.source.replace("STATUS = 1\n", replacement), - self.destination.replace("STATUS = 1\n", replacement), - ) - - def test_exports_require_complete_canonical_lines(self): - for exports in ( - self.exports[:1], - list(reversed(self.exports)), - self.exports * 2, - ["other"], - ): - with self.subTest(exports=exports), self.assertRaises(ValueError): - self.copy(exports=exports) - with self.assertRaises(ValueError): - self.copy(destination=self.destination + "del __all__\n") - self.assertNotIn(self.export_line, self.copy(exports=[])) - - def test_cli_hash_pins_receipt_and_concurrent_destination_guard(self): - with tempfile.TemporaryDirectory() as temp, contextlib.chdir(temp): - source = pathlib.Path("canonical/Lib/example.py") - destination = pathlib.Path("Lib/example.py") - source.parent.mkdir(parents=True) - destination.parent.mkdir(parents=True) - source.write_text(self.source) - destination.write_text(self.destination) - args = [ - str(source), - "--functions", - *self.names, - "--exports", - *self.exports, - "--source-sha256", - hashlib.sha256(source.read_bytes()).hexdigest(), - "--destination-sha256", - "0" * 64, - ] - with contextlib.redirect_stderr(io.StringIO()): - self.assertEqual(main(args), 1) - self.assertEqual(destination.read_text(), self.destination) - args[-1] = hashlib.sha256(destination.read_bytes()).hexdigest() - with contextlib.redirect_stdout(io.StringIO()) as output: - self.assertEqual(main(args), 0) - receipt = json.loads(output.getvalue()) - self.assertEqual(receipt["functions"], self.names) - self.assertEqual(destination.read_text(), self.copy()) - destination.write_text(self.destination) - - def concurrent_change(*unused): - destination.write_text("# changed meanwhile\n") - return self.copy() - - with patch( - "update_lib.cmd_migrate.copy_selected_functions", concurrent_change - ): - with contextlib.redirect_stderr(io.StringIO()) as errors: - self.assertEqual(main(args), 1) - self.assertIn("Destination changed", errors.getvalue()) - self.assertEqual(destination.read_text(), "# changed meanwhile\n") - - def test_cli_modes_are_exclusive_before_reading_inputs(self): - for options in ( - ["--assignments", "value"], - ["--patches", "missing.json"], - ["--legacy-map", "missing.json"], - ["--canonical-removals", "missing.json"], - ): - with ( - self.subTest(options=options), - contextlib.redirect_stderr(io.StringIO()) as errors, - ): - self.assertEqual( - main( - ["missing/Lib/example.py", "--functions", *self.names, *options] - ), - 1, - ) - self.assertIn("cannot combine", errors.getvalue()) - - def test_scoped_header_expressions_fail_closed(self): - for header in ( - "def available(limit=(lambda: requires_available())()):", - "def available(limit=[requires_available() for _ in (0,)]):", - "def available(limit={requires_available() for _ in (0,)}):", - "def available(limit={0: requires_available() for _ in (0,)}):", - "def available(limit=(requires_available() for _ in (0,))):", - "def available(*, limit=(lambda: requires_available())()):", - "def available(limit: (lambda: requires_available())() = 1):", - ): - with ( - self.subTest(header=header), - self.assertRaisesRegex( - ValueError, "Scoped expressions in function headers" - ), - ): - self.copy(self.source.replace("def available(limit=1):", header)) - for decorator in ( - "@(lambda: requires_available())()", - "@[requires_available for _ in (0,)][0]", - ): - with ( - self.subTest(decorator=decorator), - self.assertRaisesRegex( - ValueError, "Scoped expressions in function headers" - ), - ): - self.copy(self.source.replace("@functools.cache", decorator)) - - def test_nested_classes_fail_closed(self): - for body in ( - " class Inner:\n os = os\n", - " def nested():\n class Inner:\n os = os\n", - ): - with ( - self.subTest(body=body), - self.assertRaisesRegex(ValueError, "Classes in selected functions"), - ): - self.copy( - self.source.replace( - " import errno\n", body + " import errno\n" - ) - ) - - -if __name__ == "__main__": - unittest.main()