diff --git a/CHANGELOG.md b/CHANGELOG.md index b3e7311e..afc18354 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,6 @@ # **Upcoming release** +- Prevent unsafe initializer and annotation removal when inlining variables (@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 35d49733..df8c0ee3 100644 --- a/docs/library.rst +++ b/docs/library.rst @@ -569,6 +569,13 @@ folder. * ``rope.refactor.inline``: Inline occurrences of a method/variable/parameter. + Removing a variable definition can raise ``RefactoringError`` when it would + discard an unused initializer that is not known to be safe, remove a module + or class annotation, or leave reads outside the selected occurrences. + ``remove=False`` keeps the definition while replacing selected reads. + Selecting resources that exclude the defining module also keeps its + definition. Function-local annotations, safe literals, and proven bound + names remain eligible for removal. * ``rope.refactor.usefunction``: Try to use a function wherever possible. diff --git a/rope/refactor/inline.py b/rope/refactor/inline.py index 768d30b7..5f42dc97 100644 --- a/rope/refactor/inline.py +++ b/rope/refactor/inline.py @@ -21,6 +21,7 @@ import rope.base.builtins # Use fully qualified names for clarity. from rope.base import ( + ast, codeanalyze, evaluate, exceptions, @@ -267,6 +268,8 @@ def get_changes( resources = [self.original] if remove and self.original != self.resource: resources.append(self.resource) + if remove and self.resource in resources: + self._check_initializer_removal(resources, only_current) changes = ChangeSet("Inline variable <%s>" % self.name) jobset = task_handle.create_jobset("Calculating changes", len(resources)) @@ -283,6 +286,78 @@ def get_changes( jobset.finished_job() return changes + def _check_initializer_removal(self, resources, only_current): + initializer = self.pyname.assignments[0].ast_node + statement = next( + ( + node + for node in ast.walk(self.pymodule.get_ast()) + if isinstance(node, (ast.Assign, ast.AnnAssign)) + and node.value is initializer + ), + None, + ) + statement_region = ( + codeanalyze.ASTLinesAdapter(self.pymodule.source_code)[statement] + if statement is not None + else None + ) + local_annotation = isinstance(statement, ast.AnnAssign) and ( + isinstance(statement.target, ast.Name) + and self.pymodule.get_scope() + .get_inner_scope_for_line(statement.lineno) + .get_kind() + == "Function" + ) + if isinstance(statement, ast.AnnAssign) and not local_annotation: + raise exceptions.RefactoringError( + "Cannot discard a variable annotation while inlining." + ) + bound_name = isinstance(initializer, ast.Name) and _is_bound_initializer_name( + self.pymodule, statement, initializer + ) + if isinstance(initializer, ast.Name) and not bound_name: + raise exceptions.RefactoringError( + "Cannot discard an initializer name that may be unbound." + ) + finder = occurrences.create_finder( + self.project, self.name, self.pyname, imports=False + ) + has_reads = False + for resource in self.project.get_python_files(): + for occurrence in finder.find_occurrences(resource=resource): + start, _ = occurrence.get_primary_range() + if ( + resource == self.resource + and statement_region is not None + and (statement_region[0] <= start < statement_region[1]) + ): + continue + if occurrence.is_written() or occurrence.is_a_fixed_primary(): + continue + has_reads = True + if resource not in resources or ( + only_current + and resource == self.original + and not self.region[0] <= start < self.region[1] + ): + raise exceptions.RefactoringError( + "Cannot remove a variable referenced outside the selected occurrences." + ) + if has_reads: + return + plain_assignment = ( + isinstance(statement, ast.Assign) + and len(statement.targets) == 1 + and isinstance(statement.targets[0], ast.Name) + ) + if not (plain_assignment or local_annotation) or not ( + bound_name or _is_pure_unused_initializer(initializer) + ): + raise exceptions.RefactoringError( + "Cannot discard an unused variable initializer while inlining." + ) + def _change_main_module(self, remove, only_current, docs): region = None if only_current and self.original == self.resource: @@ -646,6 +721,118 @@ def _inline_variable( return source +def _is_bound_initializer_name(pymodule, statement, initializer): + if statement is None: + return False + scope = pymodule.get_scope().get_inner_scope_for_line(statement.lineno) + if scope.get_kind() == "Class": + return False + pyname = evaluate.eval_node(scope, initializer) + if any( + isinstance(node, ast.Name) + and node.id == initializer.id + and isinstance(node.ctx, ast.Del) + and evaluate.eval_node( + pymodule.get_scope().get_inner_scope_for_line(node.lineno), node + ) + is pyname + for node in ast.walk(pymodule.get_ast()) + ): + return False + if isinstance(pyname, rope.base.builtins.BuiltinName): + return True + if isinstance(pyname, pynames.ParameterName): + return True + parents = { + child: node + for node in ast.walk(pymodule.get_ast()) + for child in ast.iter_child_nodes(node) + } + anchor = statement + while anchor in parents: + owner = parents[anchor] + for _, block in ast.iter_fields(owner): + if not isinstance(block, list) or anchor not in block: + continue + for previous in reversed(block[: block.index(anchor)]): + binding_scope = pymodule.get_scope().get_inner_scope_for_line( + previous.lineno + ) + if ( + isinstance(previous, ast.Assign) + and len(previous.targets) == 1 + and isinstance(previous.targets[0], ast.Name) + ): + if previous.targets[0].id == initializer.id: + if ( + evaluate.eval_node(binding_scope, previous.targets[0]) + is pyname + ): + return True + break + if _is_pure_unused_initializer(previous.value): + continue + if isinstance( + previous, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef) + ): + if ( + previous.name == initializer.id + and pyname is not None + and pyname.get_definition_location() + == (pymodule, previous.lineno) + ): + return True + if isinstance(previous, (ast.FunctionDef, ast.AsyncFunctionDef)): + arguments = previous.args + parameters = ( + arguments.posonlyargs + + arguments.args + + arguments.kwonlyargs + + [arguments.vararg, arguments.kwarg] + ) + if not ( + previous.decorator_list + or arguments.defaults + or any(arguments.kw_defaults) + or previous.returns + or any( + parameter.annotation + for parameter in parameters + if parameter is not None + ) + or getattr(previous, "type_params", ()) + ): + continue + break + if isinstance(previous, (ast.Import, ast.ImportFrom)): + matching = any( + (alias.asname or alias.name.split(".")[0]) == initializer.id + for alias in previous.names + ) + if ( + matching + and evaluate.eval_node(binding_scope, initializer) is pyname + ): + return True + break + if isinstance(previous, ast.Pass): + continue + break + anchor = owner + return False + + +def _is_pure_unused_initializer(node): + # literal_eval accepts set(); even that name may be rebound to user code. + if any(isinstance(child, ast.Call) for child in ast.walk(node)): + return False + try: + ast.literal_eval(node) + except (TypeError, ValueError): + return False + return True + + def _getvardef(pymodule, pyname): assignment = pyname.assignments[0] lines = pymodule.lines diff --git a/ropetest/refactor/inline_unused_initializer_test.py b/ropetest/refactor/inline_unused_initializer_test.py new file mode 100644 index 00000000..ce23dcf2 --- /dev/null +++ b/ropetest/refactor/inline_unused_initializer_test.py @@ -0,0 +1,422 @@ +import os +import subprocess +import sys +from textwrap import dedent + +import pytest + +from rope.base.exceptions import RefactoringError +from rope.base.project import Project +from rope.refactor.inline import create_inline + +NAME_SOURCE = dedent("""\ + def run(): + try: + target = missing + print("body") + except (NameError, TypeError, ValueError): + print("caught") + run() + print("end") +""") + + +def _run(root): + env = os.environ.copy() + env.pop("PYTHONPATH", None) + result = subprocess.run( + [sys.executable, "-B", "-m", "main"], + cwd=root, + env=env, + capture_output=True, + text=True, + timeout=15, + ) + return result.returncode, result.stdout, result.stderr + + +def _exercise(tmp_path, sources, expected, refuse=False, selected=None, **options): + for name, code in sources.items(): + (tmp_path / name).write_text(code, encoding="utf-8") + (tmp_path / "sentinel.txt").write_text("untouched\n", encoding="utf-8") + before = {p.name: p.read_bytes() for p in tmp_path.iterdir() if p.is_file()} + assert _run(tmp_path) == (0, expected, "") + project = Project( + tmp_path, + save_history=False, + save_objectdb=False, + validate_objectdb=False, + automatic_soa=False, + import_dynload_stdmods=False, + ) + try: + resource = project.get_file("subject.py") + code = resource.read() + offset = code.index("target") + 1 + if selected == "last_read": + offset = code.rindex("target") + 1 + if "resources" in options: + options["resources"] = [ + project.get_file(name) for name in options["resources"] + ] + try: + changes = create_inline(project, resource, offset).get_changes(**options) + except RefactoringError: + if not refuse: + raise + assert before == {name: (tmp_path / name).read_bytes() for name in before} + assert _run(tmp_path) == (0, expected, "") + return + assert before == {name: (tmp_path / name).read_bytes() for name in before} + project.do(changes) + after = _run(tmp_path) + changed = { + name: (tmp_path / name).read_text(encoding="utf-8") for name in sources + } + assert after == (0, expected, ""), {"after": after, "sources": changed} + assert not refuse, {"unsafe_removal_accepted": changed, "after": after} + assert (tmp_path / "sentinel.txt").read_text() == "untouched\n" + return changed + finally: + project.close() + + +def _single(tmp_path, code, expected, **options): + return _exercise( + tmp_path, + {"subject.py": code, "main.py": "import subject\n"}, + expected, + **options, + ) + + +def test_unused_unknown_name_removal_is_atomic(tmp_path): + _single(tmp_path, NAME_SOURCE, "caught\nend\n", refuse=True) + + +def test_unused_unknown_name_remove_false_keeps_exception(tmp_path): + changed = _single(tmp_path, NAME_SOURCE, "caught\nend\n", remove=False) + assert changed["subject.py"] == NAME_SOURCE + + +@pytest.mark.parametrize("initializer", ["42", "[1, (-2, {'key': {1, 2}})]"]) +def test_unused_literal_removal_preserves_execution(tmp_path, initializer): + changed = _single( + tmp_path, NAME_SOURCE.replace("missing", initializer), "body\nend\n" + ) + assert "target =" not in changed["subject.py"] + + +@pytest.mark.parametrize("initializer", ["[missing]", "{[]: 42}"]) +def test_unused_container_removal_keeps_lookup_or_hash_failure(tmp_path, initializer): + _single( + tmp_path, + NAME_SOURCE.replace("missing", initializer), + "caught\nend\n", + refuse=True, + ) + + +def test_unused_shadowed_empty_set_call_is_not_a_literal(tmp_path): + code = dedent("""\ + def set(): + print('called') + return [] + target = set() + print('end') + """) + _single(tmp_path, code, "called\nend\n", refuse=True) + + +def test_used_name_initializer_is_still_inlined(tmp_path): + code = dedent("""\ + value = 42 + target = value + print(target) + """) + changed = _single(tmp_path, code, "42\n") + assert "target =" not in changed["subject.py"] + assert "print(value)" in changed["subject.py"] + + +def test_only_current_remove_false_keeps_other_read_and_definition(tmp_path): + code = "target = 42\nprint(target)\nprint(target)\n" + changed = _single( + tmp_path, + code, + "42\n42\n", + selected="last_read", + only_current=True, + remove=False, + ) + assert changed["subject.py"] == "target = 42\nprint(target)\nprint(42)\n" + + +def test_only_current_removal_with_another_read_is_atomic(tmp_path): + code = "target = 42\nprint(target)\nprint(target)\n" + _single( + tmp_path, code, "42\n42\n", selected="last_read", only_current=True, refuse=True + ) + + +def test_only_current_removal_with_one_read_is_still_supported(tmp_path): + changed = _single( + tmp_path, + "target = 42\nprint(target)\n", + "42\n", + selected="last_read", + only_current=True, + ) + assert changed["subject.py"] == "print(42)\n" + + +def _imported_sources(): + return { + "subject.py": "value = 42\ntarget = value\n", + "main.py": "from subject import target\nprint(target)\n", + } + + +def test_actual_importer_read_allows_name_initializer_removal(tmp_path): + changed = _exercise(tmp_path, _imported_sources(), "42\n") + assert "target =" not in changed["subject.py"] + + +def test_resources_excluding_a_read_cannot_remove_definition(tmp_path): + _exercise( + tmp_path, _imported_sources(), "42\n", resources=["subject.py"], refuse=True + ) + + +def test_resources_excluding_definition_leave_initializer_in_place(tmp_path): + sources = _imported_sources() + changed = _exercise(tmp_path, sources, "42\n", resources=["main.py"]) + assert changed["subject.py"] == sources["subject.py"] + + +@pytest.mark.parametrize("annotation", ["missing", "record()"]) +def test_module_annotation_cannot_be_discarded_with_literal_initializer( + tmp_path, annotation +): + code = dedent("""\ + def record(): + print('annotation') + return int + try: + target: ANNOTATION = 42 + print('body') + except NameError: + print('caught') + """).replace("ANNOTATION", annotation) + main = dedent("""\ + import subject + try: + print(subject.__annotations__['target']) + except (NameError, KeyError): + print('annotation caught') + """) + # Read the actual annotation dictionary, including lazy annotations on 3.14. + if annotation == "missing": + expected = ( + "caught\nannotation caught\n" + if sys.version_info < (3, 14) + else "body\nannotation caught\n" + ) + else: + expected = ( + "annotation\nbody\n\n" + if sys.version_info < (3, 14) + else "body\nannotation\n\n" + ) + _exercise(tmp_path, {"subject.py": code, "main.py": main}, expected, refuse=True) + + +def test_function_local_annotation_does_not_block_pure_unused_initializer(tmp_path): + code = NAME_SOURCE.replace("target = missing", "target: missing = 42") + changed = _single(tmp_path, code, "body\nend\n") + assert "target:" not in changed["subject.py"] + + +def test_docs_occurrence_is_not_a_runtime_read(tmp_path): + code = NAME_SOURCE.replace('print("body")', 'print("body") # target') + _single(tmp_path, code, "caught\nend\n", docs=True, refuse=True) + + +@pytest.mark.parametrize("remove", [True, False]) +def test_dormant_load_does_not_erase_initializer_evaluation(tmp_path, remove): + code = dedent("""\ + def run(): + try: + target = missing + if False: + print(target) + print("body") + except (NameError, TypeError, ValueError): + print("caught") + run() + print("end") + """) + _single(tmp_path, code, "caught\nend\n", remove=remove, refuse=remove) + + +def test_used_unknown_name_keeps_exception_before_intervening_print(tmp_path): + code = NAME_SOURCE.replace('print("body")', 'print("body")\n print(target)') + _single(tmp_path, code, "caught\nend\n", refuse=True) + + +def test_used_module_annotation_keeps_observed_effect(tmp_path): + code = dedent("""\ + import sys + import typing + def record(): + print('annotation') + return int + target: record() = 42 + print(target) + typing.get_type_hints(sys.modules[__name__]) + print('end') + """) + expected = ( + "annotation\n42\nend\n" + if sys.version_info < (3, 14) + else "42\nannotation\nend\n" + ) + _single(tmp_path, code, expected, refuse=True) + + +def test_used_name_bound_in_outer_module_is_still_inlined(tmp_path): + code = dedent("""\ + value = 42 + def run(): + target = value + print(target) + run() + """) + changed = _single(tmp_path, code, "42\n") + assert "target =" not in changed["subject.py"] + assert "print(value)" in changed["subject.py"] + + +def test_unused_bound_name_can_be_removed_without_discarding_an_effect(tmp_path): + code = "value = 42\ntarget = value\nprint('end')\n" + changed = _single(tmp_path, code, "end\n") + assert changed["subject.py"] == "value = 42\nprint('end')\n" + + +def test_deleted_outer_name_cannot_be_discarded_with_unused_initializer(tmp_path): + code = dedent("""\ + value = 42 + def run(): + try: + target = value + print('body') + except NameError: + print('caught') + del value + run() + print('end') + """) + _single(tmp_path, code, "caught\nend\n", refuse=True) + + +def test_unrelated_local_deletion_does_not_block_outer_name_initializer(tmp_path): + code = dedent("""\ + value = 42 + def unrelated(): + value = 0 + del value + def run(): + target = value + print(target) + unrelated() + run() + """) + changed = _single(tmp_path, code, "42\n") + assert "target =" not in changed["subject.py"] + assert "print(value)" in changed["subject.py"] + + +def test_builtin_name_alias_is_still_inlined(tmp_path): + code = "target = len\nprint(target([1, 2]))\n" + changed = _single(tmp_path, code, "2\n") + assert changed["subject.py"] == "print(len([1, 2]))\n" + assert changed["main.py"] == "import subject\n" + + +def test_function_parameter_alias_is_still_inlined(tmp_path): + code = dedent("""\ + def run(value): + target = value + print(target) + run(42) + """) + changed = _single(tmp_path, code, "42\n") + assert changed["subject.py"] == dedent("""\ + def run(value): + print(value) + run(42) + """) + assert changed["main.py"] == "import subject\n" + + +def test_same_module_function_alias_is_still_inlined(tmp_path): + code = dedent("""\ + def value(): + return 42 + target = value + print(target()) + """) + changed = _single(tmp_path, code, "42\n") + assert changed["subject.py"] == dedent("""\ + def value(): + return 42 + print(value()) + """) + assert changed["main.py"] == "import subject\n" + + +def test_imported_module_alias_is_still_inlined(tmp_path): + code = "import math as value\ntarget = value\nprint(target.sqrt(4))\n" + changed = _single(tmp_path, code, "2.0\n") + assert changed["subject.py"] == "import math as value\nprint(value.sqrt(4))\n" + assert changed["main.py"] == "import subject\n" + + +def test_imported_value_alias_is_still_inlined(tmp_path): + code = "from math import pi as value\ntarget = value\nprint(target)\n" + changed = _single(tmp_path, code, "3.141592653589793\n") + assert changed["subject.py"] == "from math import pi as value\nprint(value)\n" + assert changed["main.py"] == "import subject\n" + + +def test_outer_name_remains_bound_across_unrelated_literal_and_pass(tmp_path): + code = dedent("""\ + value = 42 + unrelated = 0 + pass + def run(): + target = value + print(target) + run() + """) + changed = _single(tmp_path, code, "42\n") + assert changed["subject.py"] == dedent("""\ + value = 42 + unrelated = 0 + pass + def run(): + print(value) + run() + """) + assert changed["main.py"] == "import subject\n" + + +def test_class_name_alias_removal_is_conservatively_atomic(tmp_path): + code = dedent("""\ + class Box: + value = 42 + target = value + print(target) + print(Box.value) + """) + _single(tmp_path, code, "42\n42\n", refuse=True)