Repository navigation
Expand file tree
/
Copy pathcmd_copy_test.py
More file actions
134 lines (123 loc) · 5.58 KB
/
Copy pathcmd_copy_test.py
File metadata and controls
134 lines (123 loc) · 5.58 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
"""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