mirror of
https://github.com/RustPython/RustPython.git
synced 2026-10-28 10:44:36 +08:00
Remove one-off partial update_lib migration modes (#9394)
Keep whole-file and package migration with existing marker, doctest and explicit customization preservation. Retire copy-test and selected assignment/function/export import machinery rather than maintaining extra partial-file update paths. Remove the separate historical-file deletion command after its migration use. This does not delete any Lib files or make whole-directory migration prune destination-only files; future obsolete-file removal still needs explicit review. Preserve dependency handling and contributor content-comparison fixes. All 103 retained synthetic migration, patch and copy-library tooling tests pass under an isolated bounded host Python harness. No RustPython runtime or migrated Lib source was executed. Assisted-by: OpenAI Codex:model version unavailable Co-authored-by: youknowdot <youknowdot@users.noreply.github.com>
This commit is contained in:
youknowdotandyouknowdot authored and GitHub committed
1 parent
60bd8451c2
commit
b811715799
6 files changed
-1960
No files matched your search
@@ -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
|
||||
|
||||
|
||||
@@ -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
|
||||
@@ -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, "<canonical function>", "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
|
||||
):
|
||||
|
||||
@@ -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"),
|
||||
)
|
||||
@@ -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()
|
||||
@@ -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()
|
||||
Reference in new issue
Block a user