From 00542647859ee414e4cf3088d49a1560c37c62bd Mon Sep 17 00:00:00 2001 From: Yang Fan Date: Mon, 12 Oct 2026 03:39:43 +0800 Subject: [PATCH] Bind receivers for ordinary functions stored on classes Retain the receiver in direct and saved class-slot access values without changing the original function's lexical parent or source bindings. Preserve inherited and imported functions and direct instance overrides. Record canonical calls during analysis and persist bound callable values. Add execution and source-identity regressions for these consumers. --- CHANGELOG.md | 1 + docs/library.rst | 14 + rope/base/classfunctions.py | 134 +++++++ rope/base/evaluate.py | 7 +- rope/base/oi/soa.py | 5 +- rope/base/oi/transform.py | 15 +- .../refactor/class_function_access_test.py | 171 +++++++++ .../class_stored_function_regression_test.py | 236 ++++++++++++ .../refactor/class_stored_function_test.py | 356 ++++++++++++++++++ 9 files changed, 935 insertions(+), 4 deletions(-) create mode 100644 rope/base/classfunctions.py create mode 100644 ropetest/refactor/class_function_access_test.py create mode 100644 ropetest/refactor/class_stored_function_regression_test.py create mode 100644 ropetest/refactor/class_stored_function_test.py diff --git a/CHANGELOG.md b/CHANGELOG.md index b3e7311ec..a9ea7036c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,6 @@ # **Upcoming release** +- Preserve receivers and source identities for ordinary functions stored on classes (@yangfan-yf-yf) - #895 patchedast Starred and keyword now consumes their syntactically expected and ** (@lieryan) - #896 patchedast cleanup and refactoring (@lieryan) diff --git a/docs/library.rst b/docs/library.rst index 35d497337..09985d986 100644 --- a/docs/library.rst +++ b/docs/library.rst @@ -456,6 +456,20 @@ Also ``Pycore.analyze_module()`` collects object information for a module. The collected information can be used to enhance rope's static object inference. +For an undecorated module function stored by a single class-body name +assignment, direct and saved instance accesses retain the receiver. Inherited +slots and imported functions keep the original function, class slot, and +parameter definitions, so field and keyword renames use those definitions. +Call analysis and saved object records also preserve the bound callable and +receiver across closing and reopening a project. + +This inference uses the existing ordinary argument protocol and requires a +positional parameter for the receiver, without positional-only or keyword-only +parameters. Recognized function writes to an instance, including aliases, +remain unbound. Override checks cover the storage, receiver, and access +modules; arbitrary writes in other modules, decorated storage classes, +metaclasses, and multiple storage assignments are outside this inference. + `rope.base.taskhandle.TaskHandle` ================================= diff --git a/rope/base/classfunctions.py b/rope/base/classfunctions.py new file mode 100644 index 000000000..f7a343759 --- /dev/null +++ b/rope/base/classfunctions.py @@ -0,0 +1,134 @@ +"""Bound access values for ordinary functions stored in a class body.""" + +from rope.base import arguments, ast, pynames, pyobjects + + +class BoundFunction(pyobjects.AbstractFunction): + def __init__(self, function, receiver): + super().__init__() + self.function = function + self.receiver = receiver + + def get_name(self): + return self.function.get_name() + + def get_doc(self): + return self.function.get_doc() + + def get_param_names(self, special_args=True): + return self.function.get_param_names(special_args) + + def bind_arguments(self, args): + return arguments.MixedArguments(self.receiver, args, None) + + def get_returned_object(self, args): + return self.function.get_returned_object(self.bind_arguments(args)) + + +class BoundAccessName(pynames.PyName): + def __init__(self, canonical, function, receiver): + self.canonical = canonical + self.value = BoundFunction(function, receiver) + + def get_object(self): + return self.value + + def get_definition_location(self): + return self.canonical.get_definition_location() + + def __eq__(self, other): + if isinstance(other, BoundAccessName): + other = other.canonical + return self.canonical == other + + def __hash__(self): + return hash(self.canonical) + + +def canonical_function(value): + if isinstance(value, BoundFunction): + return value.function + return value + + +def instance_access(primary, attribute, pyname, access_scope): + if not isinstance(pyname, pynames.AssignedName): + return pyname + instance = primary.get_object() + if isinstance(instance, pyobjects.AbstractClass): + return pyname + owner = instance.get_type() + if not isinstance(owner, pyobjects.PyClass): + return pyname + function = pyname.get_object() + if not isinstance(function, pyobjects.PyFunction): + return pyname + if not isinstance(function.parent, pyobjects.PyModule): + return pyname + node = function.get_ast() + if ( + not node.args.args + or node.args.posonlyargs + or node.args.kwonlyargs + or node.decorator_list + or getattr(node, "type_params", ()) + ): + return pyname + assignments = pyname.assignments + if len(assignments) != 1 or not isinstance(assignments[0].ast_node, ast.Name): + return pyname + assignment = assignments[0] + if assignment.levels or assignment.evaluation or assignment.assign_type: + return pyname + module, _ = pyname.get_definition_location() + if module is None: + return pyname + storage_scope = module.get_scope().get_inner_scope_for_line(assignment.get_lineno()) + if storage_scope.get_kind() != "Class": + return pyname + storage = storage_scope.pyobject + storage_node = storage.get_ast() + if storage_node.decorator_list or storage_node.keywords: + return pyname + if not any( + isinstance(statement, ast.Assign) + and len(statement.targets) == 1 + and isinstance(statement.targets[0], ast.Name) + and statement.targets[0].id == attribute + and statement.value is assignment.ast_node + for statement in storage_node.body + ): + return pyname + modules = (module, owner.get_module(), access_scope.pyobject.get_module()) + for current in dict.fromkeys(modules): + if _has_override(current, primary, owner, storage, attribute): + return pyname + return BoundAccessName(pyname, function, primary) + + +def _has_override(module, primary, owner, storage, attribute): + # Instance dictionary functions remain ordinary unbound function values. + from rope.base import evaluate + + receiver = primary.get_object() + for target in ast.walk(module.get_ast()): + if ( + isinstance(target, ast.Attribute) + and target.attr == attribute + and isinstance(target.ctx, (ast.Store, ast.Del)) + ): + if not isinstance(target.value, ast.Name): + return True + scope = module.get_scope().get_inner_scope_for_line(target.lineno) + target_name = evaluate.eval_node(scope, target.value) + if target_name == primary: + return True + if target_name is not None: + target_object = target_name.get_object() + if ( + target_object is receiver + or target_object is owner + or target_object is storage + ): + return True + return False diff --git a/rope/base/evaluate.py b/rope/base/evaluate.py index 4c7580cce..24c48dd4c 100644 --- a/rope/base/evaluate.py +++ b/rope/base/evaluate.py @@ -7,6 +7,7 @@ from rope.base import ( arguments, ast, + classfunctions, exceptions, nameanalyze, pyobjects, @@ -137,7 +138,7 @@ def get_enclosing_function(self, offset): except BadIdentifierError: function_pyname = None if function_pyname is not None: - pyobject = function_pyname.get_object() + pyobject = classfunctions.canonical_function(function_pyname.get_object()) if isinstance(pyobject, pyobjects.AbstractFunction): return pyobject elif ( @@ -173,7 +174,9 @@ def _Attribute(self, node): self.old_result = pyname if pyname.get_object() != rope.base.pyobjects.get_unknown(): try: - self.result = pyname.get_object()[node.attr] + self.result = classfunctions.instance_access( + pyname, node.attr, pyname.get_object()[node.attr], self.scope + ) except exceptions.AttributeNotFoundError: self.result = None diff --git a/rope/base/oi/soa.py b/rope/base/oi/soa.py index df60c6fbb..d3bad87df 100644 --- a/rope/base/oi/soa.py +++ b/rope/base/oi/soa.py @@ -1,7 +1,7 @@ import rope.base.ast import rope.base.oi.soi import rope.base.pynames -from rope.base import arguments, evaluate, nameanalyze, pyobjects +from rope.base import arguments, classfunctions, evaluate, nameanalyze, pyobjects def analyze_module(pycore, pymodule, should_analyze, search_subscopes, followed_calls): @@ -75,6 +75,9 @@ def _args_with_self(self, primary, self_pyname, pyfunction, node): return arguments.MixedArguments(self_pyname, base_args, self.scope) def _call(self, pyfunction, args): + if isinstance(pyfunction, classfunctions.BoundFunction): + args = pyfunction.bind_arguments(args) + pyfunction = pyfunction.function if isinstance(pyfunction, pyobjects.PyFunction): if self.follow is not None: before = self._parameter_objects(pyfunction) diff --git a/rope/base/oi/transform.py b/rope/base/oi/transform.py index ace4261f3..83681fff2 100644 --- a/rope/base/oi/transform.py +++ b/rope/base/oi/transform.py @@ -4,7 +4,7 @@ import re import rope.base.builtins # Use full qualification for clarity. -from rope.base import exceptions +from rope.base import classfunctions, exceptions, pynames, pyobjects class PyObjectToTextual: @@ -43,6 +43,13 @@ def PyObject_to_textual(self, pyobject): def PyFunction_to_textual(self, pyobject): return self._defined_to_textual(pyobject) + def BoundFunction_to_textual(self, pyobject): + return ( + "bound_function", + self.transform(pyobject.function), + self.transform(pyobject.receiver.get_object()), + ) + def PyClass_to_textual(self, pyobject): return self._defined_to_textual(pyobject) @@ -206,6 +213,12 @@ def instance_to_pyobject(self, textual): if type is not None: return rope.base.pyobjects.PyObject(type) + def bound_function_to_pyobject(self, textual): + function = self.transform(textual[1]) + receiver = self.transform(textual[2]) + if isinstance(function, pyobjects.PyFunction) and receiver is not None: + return classfunctions.BoundFunction(function, pynames.UnboundName(receiver)) + def _get_pymodule(self, path): resource = self.path_to_resource(path) if resource is not None: diff --git a/ropetest/refactor/class_function_access_test.py b/ropetest/refactor/class_function_access_test.py new file mode 100644 index 000000000..ff2898591 --- /dev/null +++ b/ropetest/refactor/class_function_access_test.py @@ -0,0 +1,171 @@ +import os +import subprocess +import sys +from textwrap import dedent + +import pytest + +from rope.base import arguments, evaluate, libutils, pyobjects +from rope.base.project import Project +from rope.refactor.rename import Rename + +CLASSES = dedent("""\ + class Owner: + def __init__(self): + self.field = 1 + callback = module_callback + class Child(Owner): + def __init__(self): + self.field = 3 + class Other: + def __init__(self): + self.field = 2 +""") +PREFS = dict( + save_history=False, + perform_doa=False, + automatic_soa=False, +) + + +def _write_inputs(directory, helper, subject): + directory.mkdir() + (directory / "helper.py").write_text(helper, encoding="utf-8") + (directory / "subject.py").write_text(subject, encoding="utf-8") + (directory / "sentinel.txt").write_bytes(b"untouched\n") + return _input_bytes(directory) + + +def _input_bytes(directory): + return { + name: (directory / name).read_bytes() + for name in ("helper.py", "subject.py", "sentinel.txt") + } + + +def _run(directory): + env = os.environ.copy() + env.pop("PYTHONPATH", None) + result = subprocess.run( + [sys.executable, "-B", "subject.py"], + cwd=directory, + env=env, + capture_output=True, + text=True, + timeout=45, + ) + return result.returncode, result.stdout, result.stderr + + +@pytest.mark.parametrize("access", ["direct", "saved"]) +def test_imported_inherited_function_parameter_rename(tmp_path, access): + helper_code = dedent("""\ + def module_callback(self, item): + return item + """) + call = ( + "result = Child().callback(item=Other())\n" + if access == "direct" + else "saved = Child().callback\nresult = saved(item=Other())\n" + ) + subject_code = ( + "from helper import module_callback\n" + + CLASSES + + call + + "print(result.field)\n" + ) + directory = tmp_path / "project" + original = _write_inputs(directory, helper_code, subject_code) + before = _run(directory) + assert before == (0, "2\n", "") + project = Project(directory, ropefolder=None, save_objectdb=False, **PREFS) + try: + helper = project.get_file("helper.py") + resource = project.get_file("subject.py") + offset = helper.read().index("item") + 1 + changes = Rename(project, helper, offset).get_changes("renamed") + assert _input_bytes(directory) == original + project.do(changes) + after_helper, after_subject = helper.read(), resource.read() + assert "def module_callback(self, renamed):" in after_helper + assert "return renamed" in after_helper + assert "renamed=Other()" in after_subject + assert _input_bytes(directory) != original + assert _run(directory) == before + assert (directory / "sentinel.txt").read_bytes() == original["sentinel.txt"] + + helper_module = project.get_pymodule(helper) + subject_module = project.get_pymodule(resource) + function = helper_module.get_scope().lookup("module_callback").get_object() + keyword = evaluate.eval_location( + subject_module, after_subject.index("renamed=Other") + 1 + ) + assert keyword is function.get_parameters()["renamed"] + assert function.parent is helper_module + assert function.get_module() is helper_module + owner = subject_module.get_scope().lookup("Owner").get_object() + slot = evaluate.eval_location( + subject_module, after_subject.index("callback = module_callback") + 1 + ) + assert slot is owner["callback"] + assert slot.get_definition_location()[0] is subject_module + finally: + project.close() + + +def test_imported_inherited_callback_survives_analysis_and_reopen(tmp_path): + helper_code = dedent("""\ + def module_callback(self, item): + return self + """) + use_code = dedent("""\ + def use(callback, item): + return callback(item) + """) + calls = dedent("""\ + direct = Child().callback(Other()) + saved = Child().callback + result = use(saved, Other()) + print(result.field) + """) + subject_code = "from helper import module_callback\n" + use_code + CLASSES + calls + directory = tmp_path / "project" + original = _write_inputs(directory, helper_code, subject_code) + before = _run(directory) + assert before == (0, "3\n", "") + + for analyze in (True, False): + project = Project( + directory, save_objectdb=True, validate_objectdb=True, **PREFS + ) + try: + resource = project.get_file("subject.py") + helper = project.get_file("helper.py") + if analyze: + libutils.analyze_module(project, resource) + subject_module = project.get_pymodule(resource) + helper_module = project.get_pymodule(helper) + scope = subject_module.get_scope() + child = scope.lookup("Child").get_object() + function = helper_module.get_scope().lookup("module_callback").get_object() + use = scope.lookup("use").get_object() + callback_values = project.pycore.object_info.get_passed_objects(use, 0) + assert callback_values + for callback in callback_values: + assert isinstance(callback, pyobjects.AbstractFunction) + assert callback.get_name() == function.get_name() + returned = callback.get_returned_object( + arguments.ObjectArguments([scope.lookup("Other")]) + ) + assert returned.get_type() is child + receivers = project.pycore.object_info.get_passed_objects(function, 0) + assert any(receiver.get_type() is child for receiver in receivers) + assert scope.lookup("result").get_object().get_type() is child + assert function.parent is helper_module + assert function.get_module() is helper_module + assert _input_bytes(directory) == original + assert _run(directory) == before + finally: + project.close() + assert any((directory / ".ropeproject").iterdir()) + assert _input_bytes(directory) == original diff --git a/ropetest/refactor/class_stored_function_regression_test.py b/ropetest/refactor/class_stored_function_regression_test.py new file mode 100644 index 000000000..ca3c5efa8 --- /dev/null +++ b/ropetest/refactor/class_stored_function_regression_test.py @@ -0,0 +1,236 @@ +import json + +import pytest + +from rope.base import pyobjects +from rope.refactor.rename import Rename +from ropetest.refactor.class_stored_function_test import ( + CLASS_ALIAS, + MODULE_FUNCTION, + OTHER, + OWNER, + _project, + _run, +) + +PREFIX = MODULE_FUNCTION + OWNER + CLASS_ALIAS + OTHER +INHERITED = ( + PREFIX + + "class Child(Owner):\n def __init__(self):\n self.field = 3\n" + + "result = Child().callback(Other())\nprint(result.field)\n" +) +CASES = { + "alias_override": ( + PREFIX + + "instance = Owner()\nalias = instance\nalias.callback = module_callback\n" + + "result = instance.callback(Other(), Owner())\nprint(result.field)\n", + "Other", + "2\n", + ), + "inherited": (INHERITED, "Child", "3\n"), + "object_base": ( + PREFIX.replace("class Owner:", "class Owner(object):") + + "result = Owner().callback(Other())\nprint(result.field)\n", + "Owner", + "1\n", + ), + "different_instance_override": ( + PREFIX + + "first = Owner()\nsecond = Owner()\nsecond.callback = module_callback\n" + + "result = first.callback(Other())\nprint(result.field)\n", + "Owner", + "1\n", + ), +} + + +def _query(directory, code, expected_class): + project, resource = _project(directory, code) + try: + scope = project.get_pymodule(resource).get_scope() + result = scope.lookup("result").get_object() + return { + "unknown": result is pyobjects.get_unknown(), + "type_expected": result.get_type() + is scope.lookup(expected_class).get_object(), + } + finally: + project.close() + + +def _exercise(tmp_path, fixture, code, expected_class, expected_stdout, selected): + query = _query(tmp_path / "query", code, expected_class) + directory = tmp_path / "cold" + project, resource = _project(directory, code) + original = {p.name: p.read_bytes() for p in directory.iterdir() if p.is_file()} + if selected == "parent_slot": + offset = code.index(" callback =") + len(" ") + 1 + new_name = "renamed_callback" + else: + value = {"Owner": "1", "Other": "2", "Child": "3"}[selected] + offset = code.index("self.field = " + value) + len("self.") + 1 + new_name = "changed" + evidence = { + "fixture": fixture, + "selected": selected, + "query": query, + "before_source": code, + "offset": offset, + "new_name": new_name, + "query_project_separate_from_cold": True, + "original_bytes_hex": {name: data.hex() for name, data in original.items()}, + } + try: + before = _run(code, directory) + evidence["before"] = before + assert before == (0, expected_stdout, "") + changes = Rename(project, resource, offset).get_changes(new_name) + assert original == {name: (directory / name).read_bytes() for name in original} + evidence["changeset"] = changes.get_description() + project.do(changes) + after_code = resource.read() + after = _run(after_code, directory) + evidence.update( + after_source=after_code, + after=after, + actual_source_change=after_code != code, + runtime_preserved=after == before, + ) + assert (directory / "sentinel.txt").read_bytes() == original["sentinel.txt"] + assert after_code != code + assert after == before, evidence + if selected == "parent_slot": + assert " renamed_callback = module_callback" in after_code + assert "Child().renamed_callback(Other())" in after_code + else: + assert "self.changed = " + value in after_code + expected_read = ( + "print(result.changed)" + if selected == expected_class + else "print(result.field)" + ) + assert expected_read in after_code + finally: + project.close() + (tmp_path / "evidence.json").write_text( + json.dumps(evidence, indent=2) + "\n", encoding="utf-8" + ) + + +@pytest.mark.parametrize("fixture", list(CASES)) +@pytest.mark.parametrize("field_role", ["primary", "Other"]) +def test_receiver_storage_regressions_preserve_execution(tmp_path, fixture, field_role): + code, expected_class, stdout = CASES[fixture] + primary = "Child" if fixture == "inherited" else "Owner" + selected = primary if field_role == "primary" else "Other" + _exercise(tmp_path, fixture, code, expected_class, stdout, selected) + + +def test_inherited_callback_slot_rename_preserves_source_identity(tmp_path): + _exercise( + tmp_path, "inherited_parent_slot", INHERITED, "Child", "3\n", "parent_slot" + ) + + +IMPORTED = ( + "from helper import module_callback\n" + + OWNER + + CLASS_ALIAS + + OTHER + + "result = Owner().callback(Other())\nprint(result.field)\n" +) + + +def _imported_project(directory): + project, resource = _project(directory, IMPORTED) + helper = project.root.create_file("helper.py") + helper.write(MODULE_FUNCTION) + return project, resource, helper + + +@pytest.mark.parametrize("selected", ["Owner", "Other", "class_slot", "function"]) +def test_imported_function_class_storage_preserves_execution(tmp_path, selected): + query_project, query_resource, query_helper = _imported_project(tmp_path / "query") + try: + scope = query_project.get_pymodule(query_resource).get_scope() + result = scope.lookup("result").get_object() + function = scope.lookup("module_callback").get_object() + helper_module = query_project.get_pymodule(query_helper) + query = { + "unknown": result is pyobjects.get_unknown(), + "type_expected": result.get_type() is scope.lookup("Owner").get_object(), + "function_is_canonical": function + is helper_module.get_scope().lookup("module_callback").get_object(), + "function_module_is_helper": function.get_module() is helper_module, + } + finally: + query_project.close() + directory = tmp_path / "cold" + project, resource, helper = _imported_project(directory) + original = {p.name: p.read_bytes() for p in directory.iterdir() if p.is_file()} + target, text = ( + (helper, MODULE_FUNCTION) if selected == "function" else (resource, IMPORTED) + ) + if selected == "function": + offset = text.index("module_callback") + 1 + new_name = "renamed_function" + elif selected == "class_slot": + offset = text.index(" callback =") + len(" ") + 1 + new_name = "renamed_callback" + else: + value = "1" if selected == "Owner" else "2" + offset = text.index("self.field = " + value) + len("self.") + 1 + new_name = "changed" + evidence = { + "fixture": "imported_function", + "selected": selected, + "query": query, + "offset": offset, + "target": target.path, + "new_name": new_name, + "query_project_separate_from_cold": True, + "before_source": {"subject.py": resource.read(), "helper.py": helper.read()}, + "original_bytes_hex": {name: data.hex() for name, data in original.items()}, + } + try: + before = _run(IMPORTED, directory) + evidence["before"] = before + assert before == (0, "1\n", "") + changes = Rename(project, target, offset).get_changes(new_name) + assert original == {name: (directory / name).read_bytes() for name in original} + evidence["changeset"] = changes.get_description() + project.do(changes) + after_source = {"subject.py": resource.read(), "helper.py": helper.read()} + after_bytes = {name: (directory / name).read_bytes() for name in original} + after = _run(resource.read(), directory) + evidence.update( + after_source=after_source, + after_bytes_hex={name: data.hex() for name, data in after_bytes.items()}, + after=after, + actual_source_change=after_bytes != original, + runtime_preserved=after == before, + ) + assert after_bytes["sentinel.txt"] == original["sentinel.txt"] + assert after_bytes != original + assert after == before, evidence + assert query["function_is_canonical"] and query["function_module_is_helper"] + subject = after_source["subject.py"] + if selected == "function": + assert "def renamed_function(" in after_source["helper.py"] + assert "from helper import renamed_function" in subject + assert "callback = renamed_function" in subject + elif selected == "class_slot": + assert "renamed_callback = module_callback" in subject + assert "Owner().renamed_callback(Other())" in subject + else: + assert "self.changed = " + value in subject + assert ( + "print(result.changed)" + if selected == "Owner" + else "print(result.field)" + ) in subject + finally: + project.close() + (tmp_path / "evidence.json").write_text( + json.dumps(evidence, indent=2) + "\n", encoding="utf-8" + ) diff --git a/ropetest/refactor/class_stored_function_test.py b/ropetest/refactor/class_stored_function_test.py new file mode 100644 index 000000000..04b56a4c9 --- /dev/null +++ b/ropetest/refactor/class_stored_function_test.py @@ -0,0 +1,356 @@ +import json +import os +import subprocess +import sys +from textwrap import dedent + +import pytest + +from rope.base import evaluate, pyobjects +from rope.base.project import Project +from rope.refactor.rename import Rename + +OWNER = dedent("""\ + class Owner: + def __init__(self): + self.field = 1 +""") +OTHER = dedent("""\ + class Other: + def __init__(self): + self.field = 2 +""") +MODULE_FUNCTION = dedent("""\ + def module_callback(self, item): + return self +""") +CLASS_ALIAS = " callback = module_callback\n" +PRINT_RESULT = "print(result.field)\n" + + +def _fixtures(): + stored = MODULE_FUNCTION + OWNER + CLASS_ALIAS + OTHER + return { + "direct_instance": ( + stored + "result = Owner().callback(Other())\n" + PRINT_RESULT, + "Owner", + "1\n", + ), + "saved_bound": ( + stored + + "saved = Owner().callback\nresult = saved(Other())\n" + + PRINT_RESULT, + "Owner", + "1\n", + ), + "class_unbound": ( + stored + "result = Owner.callback(Owner(), Other())\n" + PRINT_RESULT, + "Owner", + "1\n", + ), + "instance_own_function": ( + stored + + "instance = Owner()\ninstance.callback = module_callback\n" + + "result = instance.callback(Other(), Owner())\n" + + PRINT_RESULT, + "Other", + "2\n", + ), + "ordinary_method": ( + OWNER + + " def callback(self, item):\n return self\n" + + OTHER + + "result = Owner().callback(Other())\n" + + PRINT_RESULT, + "Owner", + "1\n", + ), + "lexical_global": ( + OTHER + + "GLOBAL = Other()\n" + + "def module_callback(self, item):\n return GLOBAL\n" + + OWNER + + CLASS_ALIAS + + "result = Owner().callback(Other())\n" + + PRINT_RESULT, + "Other", + "2\n", + ), + } + + +def _run(code, directory): + compile((directory / "subject.py").read_bytes(), "subject.py", "exec") + env = os.environ.copy() + env.pop("PYTHONPATH", None) + result = subprocess.run( + [sys.executable, "-B", "subject.py"], + cwd=directory, + env=env, + capture_output=True, + text=True, + timeout=45, + ) + return result.returncode, result.stdout, result.stderr + + +def _project(directory, code): + directory.mkdir() + (directory / "subject.py").write_text(code, encoding="utf-8") + (directory / "sentinel.txt").write_text("untouched\n", encoding="utf-8") + project = Project( + directory, + ropefolder=None, + save_history=False, + save_objectdb=False, + perform_doa=False, + automatic_soa=False, + ) + return project, project.get_file("subject.py") + + +def _query(directory, code, expected_class): + project, resource = _project(directory, code) + try: + module = project.get_pymodule(resource) + scope = module.get_scope() + expected_type = scope.lookup(expected_class).get_object() + other_type = scope.lookup( + "Other" if expected_class == "Owner" else "Owner" + ).get_object() + result = scope.lookup("result").get_object() + result_type = result.get_type() + finder = evaluate.ScopeNameFinder(module) + read = finder.get_pyname_at( + code.index("print(result.field)") + len("print(result.") + 1 + ) + expected_field = expected_type.get_attribute("field") + return { + "expected_class": expected_class, + "result_unknown": result is pyobjects.get_unknown(), + "result_type_expected": result_type is expected_type, + "result_type_other": result_type is other_type, + "field_is_expected_binding": read is expected_field, + "field_unknown": read is None + or read.get_object() is pyobjects.get_unknown(), + } + finally: + project.close() + + +@pytest.mark.parametrize("fixture_name", list(_fixtures())) +@pytest.mark.parametrize("selected", ["Owner", "Other"]) +def test_class_stored_function_field_rename_preserves_execution( + tmp_path, fixture_name, selected +): + code, expected_class, expected_stdout = _fixtures()[fixture_name] + query = _query(tmp_path / "separate-query", code, expected_class) + project, resource = _project(tmp_path / "cold-rename", code) + directory = tmp_path / "cold-rename" + original = { + path.name: path.read_bytes() for path in directory.iterdir() if path.is_file() + } + offset = ( + code.index("self.field = " + ("1" if selected == "Owner" else "2")) + + len("self.") + + 1 + ) + evidence = { + "fixture": fixture_name, + "selected": selected, + "query": query, + "before_source": code, + "offset": offset, + "new_name": "changed", + "cold_project_has_no_query_before_rename": True, + "original_file_bytes_hex": { + name: data.hex() for name, data in original.items() + }, + } + try: + before = _run(code, directory) + evidence["before"] = before + assert before == (0, expected_stdout, "") + changes = Rename(project, resource, offset).get_changes("changed") + assert original == {name: (directory / name).read_bytes() for name in original} + evidence["changeset_description"] = changes.get_description() + project.do(changes) + after_code = resource.read() + after = _run(after_code, directory) + evidence.update( + after_source=after_code, + after=after, + actual_source_change=after_code != code, + ) + assert (directory / "sentinel.txt").read_bytes() == original["sentinel.txt"] + assert after_code != code + changed_definition = "self.changed = " + ("1" if selected == "Owner" else "2") + assert changed_definition in after_code + evidence["runtime_preserved"] = after == before + assert after == before, evidence + expected_read = ( + "print(result.changed)" + if selected == expected_class + else "print(result.field)" + ) + assert expected_read in after_code + # A known wrong public value is an error; unknown alone is not a red case. + assert not query["result_type_other"], query + finally: + project.close() + (tmp_path / "evidence.json").write_text( + json.dumps(evidence, indent=2) + "\n", encoding="utf-8" + ) + + +@pytest.mark.parametrize("selected", ["class_slot", "module_function"]) +def test_class_stored_function_name_rename_preserves_identity(tmp_path, selected): + code, _, expected_stdout = _fixtures()["direct_instance"] + directory = tmp_path / "cold-rename" + project, resource = _project(directory, code) + original = { + path.name: path.read_bytes() for path in directory.iterdir() if path.is_file() + } + offset = ( + code.index(" callback =") + len(" ") + 1 + if selected == "class_slot" + else code.index("def module_callback") + len("def ") + 1 + ) + new_name = "renamed_callback" + evidence = { + "fixture": "canonical_name_rename", + "selected": selected, + "before_source": code, + "offset": offset, + "new_name": new_name, + "cold_project_has_no_query_before_rename": True, + "original_file_bytes_hex": { + name: data.hex() for name, data in original.items() + }, + } + try: + before = _run(code, directory) + evidence["before"] = before + assert before == (0, expected_stdout, "") + changes = Rename(project, resource, offset).get_changes(new_name) + assert original == {name: (directory / name).read_bytes() for name in original} + evidence["changeset_description"] = changes.get_description() + project.do(changes) + after_code = resource.read() + after = _run(after_code, directory) + evidence.update( + after_source=after_code, + after=after, + actual_source_change=after_code != code, + ) + assert (directory / "sentinel.txt").read_bytes() == original["sentinel.txt"] + assert after_code != code + assert after == before, evidence + if selected == "class_slot": + assert " renamed_callback = module_callback\n" in after_code + assert "Owner().renamed_callback(Other())" in after_code + assert "def module_callback(self, item):" in after_code + else: + assert "def renamed_callback(self, item):" in after_code + assert " callback = renamed_callback\n" in after_code + assert "Owner().callback(Other())" in after_code + evidence["runtime_preserved"] = True + finally: + project.close() + (tmp_path / "evidence.json").write_text( + json.dumps(evidence, indent=2) + "\n", encoding="utf-8" + ) + + +@pytest.mark.parametrize("query_order", [("first", "second"), ("second", "first")]) +@pytest.mark.parametrize("selected", ["Owner", "SecondOwner"]) +def test_shared_class_function_keeps_receivers_separate( + tmp_path, query_order, selected +): + code = ( + MODULE_FUNCTION + + OWNER + + CLASS_ALIAS + + "class SecondOwner:\n def __init__(self):\n self.field = 3\n" + + CLASS_ALIAS + + OTHER + + "first = Owner().callback(Other())\n" + + "second = SecondOwner().callback(Other())\n" + + "print(first.field, second.field)\n" + ) + query_project, query_resource = _project(tmp_path / "separate-query", code) + try: + scope = query_project.get_pymodule(query_resource).get_scope() + expected_types = { + "first": scope.lookup("Owner").get_object(), + "second": scope.lookup("SecondOwner").get_object(), + } + other_type = scope.lookup("Other").get_object() + queries = [] + for name in (*query_order, *query_order): + value = scope.lookup(name).get_object() + value_type = value.get_type() + queries.append( + { + "name": name, + "unknown": value is pyobjects.get_unknown(), + "type_expected": value_type is expected_types[name], + "type_other": value_type is other_type, + } + ) + finally: + query_project.close() + directory = tmp_path / "cold-rename" + project, resource = _project(directory, code) + original = { + path.name: path.read_bytes() for path in directory.iterdir() if path.is_file() + } + offset = ( + code.index("self.field = " + ("1" if selected == "Owner" else "3")) + + len("self.") + + 1 + ) + evidence = { + "fixture": "shared_two_classes-" + "-".join(query_order), + "selected": selected, + "queries": queries, + "before_source": code, + "offset": offset, + "new_name": "changed", + "cold_project_has_no_query_before_rename": True, + "original_file_bytes_hex": { + name: data.hex() for name, data in original.items() + }, + } + try: + before = _run(code, directory) + evidence["before"] = before + assert before == (0, "1 3\n", "") + changes = Rename(project, resource, offset).get_changes("changed") + assert original == {name: (directory / name).read_bytes() for name in original} + evidence["changeset_description"] = changes.get_description() + project.do(changes) + after_code = resource.read() + after = _run(after_code, directory) + evidence.update( + after_source=after_code, + after=after, + actual_source_change=after_code != code, + runtime_preserved=after == before, + ) + assert (directory / "sentinel.txt").read_bytes() == original["sentinel.txt"] + assert after_code != code + assert "self.changed = " + ("1" if selected == "Owner" else "3") in after_code + assert after == before, evidence + expected_read = ( + "print(first.changed, second.field)" + if selected == "Owner" + else "print(first.field, second.changed)" + ) + assert expected_read in after_code + assert not any(item["type_other"] for item in queries), queries + finally: + project.close() + (tmp_path / "evidence.json").write_text( + json.dumps(evidence, indent=2) + "\n", encoding="utf-8" + )