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" + )