diff --git a/CHANGELOG.md b/CHANGELOG.md index b3e7311ec..ebec2f10b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,12 @@ # **Upcoming release** +- Preserve source bindings in function annotations, lambdas and comprehensions + during renaming, and keep call-time lambda values separate from source names. +- Keep initialized instance lambda values separate from same-named class values + when resolving attribute calls and renaming fields. +- Refuse edits to ordinary positional-only and keyword-only parameter bindings + that the current function argument protocol cannot represent safely. + - #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..64bb3a2f1 100644 --- a/docs/library.rst +++ b/docs/library.rst @@ -875,6 +875,53 @@ It provides two new search methods: - search() - simpler wrapper around search_full with a basic sorting algorithm +Source binding environments +--------------------------- + +The structural scope of an expression and the environment in which its names +are resolved can differ. Function annotations use their declaration or +annotation environment; lambda parameters and comprehension targets have their +own bindings. The existing ``GlobalScope.get_inner_scope_for_offset(offset)`` +continues to return the structural scope. Its explicit +``get_structural_scope_for_offset(offset)`` alias has the same behavior. + +A module's ``get_source_binding_index()`` provides source identity queries: + +.. code-block:: python + + pymodule = project.get_pymodule(resource) + index = pymodule.get_source_binding_index() + context = index.context_at(offset) + environment = context.environment + structural_scope = context.structural_scope + binding = index.binding_for_token(offset) + +``context_for_node(node)`` resolves a node from that module's AST. Source +queries in ``rope.base.evaluate`` use these environments while keeping source +identities separate from call-time values. A lambda's private argument matcher +handles its own positional, keyword and creation-time default values; it does +not change the ordinary ``Arguments`` or ``PyFunction`` parameter protocols. +Unknown argument expansion or conflicting values remain unknown. + +When a directly assigned class attribute and a directly assigned instance +attribute in ``__init__`` hold different lambdas, attribute value views select +the class or initialized instance value independently. This selection does not +replace their source names or change the ordinary class attribute protocol. + +Ordinary positional-only and keyword-only parameters are represented by +distinct readable source names, so their bodies do not fall back to same-named +outer variables. Their call inference and signature rewriting are outside this +model's supported range. Rename refuses edits to these parameter bindings, and +introduce-parameter refuses to rewrite such a signature. Queries still return +definition locations and an unknown value. An unresolved generic annotation +binding likewise refuses a potentially affected edit before any source changes. + +The model does not provide arbitrary dynamic class namespace inference. For +example, assigning a lambda to ``Owner.callback`` after the class definition +can leave attribute inference unknown; source binding identity is not a promise +of complete dynamic value inference. Inline safety for deferred annotations and +initializer removal is outside the source model's new consumer scope. + Cross-Project Refactorings -------------------------- diff --git a/rope/base/evaluate.py b/rope/base/evaluate.py index 4c7580cce..e3bb59661 100644 --- a/rope/base/evaluate.py +++ b/rope/base/evaluate.py @@ -7,16 +7,38 @@ from rope.base import ( arguments, ast, + codeanalyze, exceptions, nameanalyze, pyobjects, pyobjectsdef, + pyscopes, worder, ) BadIdentifierError = exceptions.BadIdentifierError +class SourceValueScope(pyscopes.TemporaryScope): + """A call's value overlay, without replacing canonical source binders.""" + + def __init__(self, canonical_scope, capture_scope, names): + super().__init__(canonical_scope.pycore, canonical_scope, names) + self.pyobject = canonical_scope.pyobject + self.canonical_scope = canonical_scope + self.capture_scope = capture_scope + + def lookup(self, name): + if name in self.names: + return self.names[name] + if name in self.canonical_scope.get_names(): + return self.canonical_scope.get_names()[name] + return self.capture_scope.lookup(name) + + def _propagated_lookup(self, name): + return self.lookup(name) + + def eval_location(pymodule, offset): """Find the pyname at the offset""" return eval_location2(pymodule, offset)[1] @@ -37,6 +59,10 @@ def eval_node(scope, node): def eval_node2(scope, node): + module = scope.pyobject.get_module() + context = module.get_source_binding_index().context_for_node(node) + if context is not None and scope.get_kind() != "Temporary": + scope = context.environment evaluator = StatementEvaluator(scope) evaluator.visit(node) return evaluator.old_result, evaluator.result @@ -92,31 +118,44 @@ def get_primary_and_pyname_at( offset: int, ) -> Tuple[Optional[rope.base.pynames.PyName], Optional[rope.base.pynames.PyName]]: lineno = self.lines.get_line_number(offset) - holding_scope = self.module_scope.get_inner_scope_for_offset(offset) + index = self.module_scope.pyobject.get_source_binding_index() + context = index.context_at(offset) + holding_scope = context.environment + structural_scope = context.structural_scope + parameter = index.binding_for_token(offset) + if parameter is not None: + return (None, parameter) # function keyword parameter if self.worder.is_function_keyword_parameter(offset): keyword_name = self.worder.get_word_at(offset) pyobject = self.get_enclosing_function(offset) - if isinstance(pyobject, pyobjectsdef.PyFunction): + from rope.base.sourcebindings import SourceLambdaFunction + + if isinstance(pyobject, SourceLambdaFunction): + parameter_name = None + if keyword_name in pyobject.get_keyword_param_names(): + parameter_name = pyobject.get_parameters().get(keyword_name) + return (None, parameter_name) + elif isinstance(pyobject, pyobjectsdef.PyFunction): parameter_name = pyobject.get_parameters().get(keyword_name, None) return (None, parameter_name) elif isinstance(pyobject, pyobjects.AbstractFunction): parameter_name = rope.base.pynames.ParameterName() return (None, parameter_name) # class body - if self._is_defined_in_class_body(holding_scope, offset, lineno): - class_scope = holding_scope - if lineno == holding_scope.get_start(): - class_scope = holding_scope.parent + if self._is_defined_in_class_body(structural_scope, offset, lineno): + class_scope = structural_scope + if lineno == structural_scope.get_start(): + class_scope = structural_scope.parent name = self.worder.get_primary_at(offset).strip() try: return (None, class_scope.pyobject[name]) except rope.base.exceptions.AttributeNotFoundError: return (None, None) # function header - if self._is_function_name_in_function_header(holding_scope, offset, lineno): + if self._is_function_name_in_function_header(structural_scope, offset, lineno): name = self.worder.get_primary_at(offset).strip() - return (None, holding_scope.parent[name]) + return (None, structural_scope.parent[name]) # module in a from statement or an imported name that is aliased if self.worder.is_from_statement_module( offset @@ -128,14 +167,36 @@ def get_primary_and_pyname_at( name = self.worder.get_from_aliased(offset) else: name = self.worder.get_primary_at(offset) - return eval_str2(holding_scope, name) + primary, pyname = eval_str2(holding_scope, name) + from rope.base.sourcebindings import SourceValueName + + if isinstance(pyname, SourceValueName): + pyname = pyname.source_pyname + return primary, pyname def get_enclosing_function(self, offset): + module = self.module_scope.pyobject + lines = codeanalyze.ASTLinesAdapter(module.source_code) + for node in ast.walk(module.get_ast()): + if not isinstance(node, ast.Call): + continue + for keyword in node.keywords: + if keyword.arg is None: + continue + start, _ = lines[keyword] + if start <= offset < start + len(keyword.arg): + return self._callable_object( + eval_node(self.module_scope, node.func) + ) function_parens = self.worder.find_parens_start_from_inside(offset) try: function_pyname = self.get_pyname_at(function_parens - 1) except BadIdentifierError: function_pyname = None + return self._callable_object(function_pyname) + + @staticmethod + def _callable_object(function_pyname): if function_pyname is not None: pyobject = function_pyname.get_object() if isinstance(pyobject, pyobjects.AbstractFunction): @@ -174,6 +235,22 @@ def _Attribute(self, node): if pyname.get_object() != rope.base.pyobjects.get_unknown(): try: self.result = pyname.get_object()[node.attr] + from rope.base.sourcebindings import ( + SourceLambdaFunction, + SourceValueName, + source_lambda_attribute_value, + ) + + selected = source_lambda_attribute_value(pyname, self.result) + if selected is not None: + self.result = selected + function = self.result.get_object() + if isinstance(function, SourceLambdaFunction): + receiver = function.descriptor_receiver(pyname, self.result) + if receiver is not None: + self.result = SourceValueName( + self.result, function.bind_receiver(receiver) + ) except exceptions.AttributeNotFoundError: self.result = None @@ -187,6 +264,21 @@ def _get_returned(pyobject): return pyobject.get_returned_object(args) if isinstance(pyobject, rope.base.pyobjects.AbstractClass): + from rope.base.sourcebindings import SourceLambdaFunction + + if ( + isinstance(pyobject, rope.base.builtins.BuiltinClass) + and pyobject.builtin in (staticmethod, classmethod) + and node.args + ): + function = eval_node(self.scope, node.args[0]) + if function is not None and isinstance( + function.get_object(), SourceLambdaFunction + ): + self.result = rope.base.pynames.UnboundName( + function.get_object().as_descriptor(pyobject.builtin.__name__) + ) + return result = None if "__new__" in pyobject: new_function = pyobject["__new__"].get_object() @@ -201,6 +293,12 @@ def _get_returned(pyobject): pyfunction = pyobject elif "__call__" in pyobject: pyfunction = pyobject["__call__"].get_object() + from rope.base.sourcebindings import SourceLambdaFunction + + if isinstance(pyfunction, SourceLambdaFunction): + pyfunction = pyfunction.bind_receiver( + rope.base.pynames.UnboundName(pyobject) + ) if pyfunction is not None: self.result = rope.base.pynames.UnboundName( pyobject=_get_returned(pyfunction) @@ -284,12 +382,76 @@ def _GeneratorExp(self, node): pyobject=rope.base.builtins.get_iterator(pyobject) ) + def _SetComp(self, node): + pyobject = self._what_does_comprehension_hold(node) + self.result = rope.base.pynames.UnboundName( + pyobject=rope.base.builtins.get_set(pyobject) + ) + + def _DictComp(self, node): + scope = self._make_comprehension_scope(node) + key = eval_node(scope, node.key) + value = eval_node(scope, node.value) + self.result = rope.base.pynames.UnboundName( + pyobject=rope.base.builtins.get_dict( + key.get_object() if key is not None else None, + value.get_object() if value is not None else None, + ) + ) + def _what_does_comprehension_hold(self, node): scope = self._make_comprehension_scope(node) pyname = eval_node(scope, node.elt) return pyname.get_object() if pyname is not None else None def _make_comprehension_scope(self, node): + module = self.scope.pyobject.get_module() + binding_scope = module.get_source_binding_index().comprehension_scope(node) + if binding_scope is not None: + if self.scope.get_kind() == "Temporary": + values = { + name: rope.base.pynames.UnboundName() + for name in binding_scope.get_names() + } + value_scope = SourceValueScope(binding_scope, self.scope, values) + for number, generator in enumerate(node.generators): + entry = self.scope if number == 0 else value_scope + iterator = eval_node(entry, generator.iter) + if iterator is None: + continue + iterator_scope = pyscopes.TemporaryScope( + entry.pycore, entry, {"__rope_iterable": iterator} + ) + element = eval_str( + iterator_scope, "__rope_iterable.__iter__().next()" + ) + if element is None: + continue + for name, levels in nameanalyze.get_name_levels(generator.target): + value = element + for level in levels: + element_type = value.get_object().get_type() + if isinstance(element_type, rope.base.builtins.Tuple): + holdings = element_type.get_holding_objects() + value = ( + rope.base.pynames.UnboundName(holdings[level]) + if level < len(holdings) + and holdings[level] is not None + else None + ) + else: + item_scope = pyscopes.TemporaryScope( + entry.pycore, entry, {"__rope_element": value} + ) + value = eval_str( + item_scope, "__rope_element[%s]" % level + ) + if value is None: + break + if value is not None: + values[name] = value + return value_scope + return binding_scope scope = self.scope module = scope.pyobject.get_module() names = {} @@ -356,9 +518,13 @@ def _call_function(self, node, function_name, other_args=None): ) def _Lambda(self, node): - self.result = rope.base.pynames.UnboundName( - pyobject=rope.base.builtins.Lambda(node, self.scope) - ) + module = self.scope.pyobject.get_module() + function = module.get_source_binding_index().callable_for_node(node) + if function is None: + function = rope.base.builtins.Lambda(node, self.scope) + elif self.scope.get_kind() == "Temporary": + function = function.bind_creation_scope(self.scope) + self.result = rope.base.pynames.UnboundName(pyobject=function) def _get_evaluated_names(targets, assigned, module, evaluation, lineno): diff --git a/rope/base/pyobjectsdef.py b/rope/base/pyobjectsdef.py index 95bd15691..4bddd563b 100644 --- a/rope/base/pyobjectsdef.py +++ b/rope/base/pyobjectsdef.py @@ -214,6 +214,22 @@ def _create_concluded_attributes(self): def _create_scope(self): return rope.base.pyscopes.GlobalScope(self.pycore, self) + @utils.saveit + def get_source_binding_index(self): + from rope.base.sourcebindings import SourceBindingIndex + + return SourceBindingIndex(self) + + def get_attributes(self): + if ":=" in self.source_code: + self.get_source_binding_index().prepare_header_bindings() + return super().get_attributes() + + def get_attribute(self, name): + if ":=" in self.source_code: + self.get_source_binding_index().prepare_header_bindings() + return super().get_attribute(name) + @property @utils.saveit def lines(self): diff --git a/rope/base/pyscopes.py b/rope/base/pyscopes.py index 565541903..006bdd473 100644 --- a/rope/base/pyscopes.py +++ b/rope/base/pyscopes.py @@ -46,6 +46,10 @@ def lookup(self, name): return self.parent._propagated_lookup(name) return None + def get_binding_scopes(self): + module = self._get_global_scope().pyobject + return module.get_source_binding_index().children_of(self) + def get_propagated_names(self): """Return the visible names of this scope @@ -155,6 +159,9 @@ def get_inner_scope_for_line(self, lineno, indents=None): def get_inner_scope_for_offset(self, offset): return self._scope_finder.get_holding_scope_for_offset(self, offset) + def get_structural_scope_for_offset(self, offset): + return self._scope_finder.get_holding_scope_for_offset(self, offset) + @property @utils.saveit def _scope_finder(self): @@ -211,12 +218,16 @@ def __init__(self, pycore, pyobject, visitor): def _get_names(self): if self.names is None: self._visit_function() + module = self.pyobject.get_module() + module.get_source_binding_index().install_unsupported_formals(self) + if ":=" in module.source_code: + module.get_source_binding_index().prepare_header_bindings(self) return self.names def _visit_function(self): if self.names is None: new_visitor = self.visitor(self.pycore, self.pyobject) - for n in ast.iter_child_nodes(self.pyobject.get_ast()): + for n in self.pyobject.get_ast().body: new_visitor.visit(n) self.names = new_visitor.names self.names.update(self.pyobject.get_parameters()) diff --git a/rope/base/sourcebindings.py b/rope/base/sourcebindings.py new file mode 100644 index 000000000..018263065 --- /dev/null +++ b/rope/base/sourcebindings.py @@ -0,0 +1,1642 @@ +"""Canonical source identities for expression scopes and function headers. + +The index is built solely from the module AST. Structural scopes are resolved +only after construction, so looking up a decorator cannot recursively build a +partially published index. Source identities never contain call-time values. +""" + +import sys +from dataclasses import dataclass + +from rope.base import ( + ast, + codeanalyze, + exceptions, + nameanalyze, + pynames, + pyobjects, + pyscopes, +) + + +def _resolved(environment): + if isinstance(environment, _StructuralReference): + return environment.resolve() + return environment + + +class _StructuralReference: + """An AST owner, without forcing Rope's structural visitor during build.""" + + def __init__(self, index, node, parent=None, entry=None): + self.index = index + self.node = node + self.parent = parent + self.entry = parent if entry is None else entry + self._scope = None + self.header_slots = {} + self.redirected_headers = {} + self._installed_slots = False + self._installing_slots = False + self.unsupported_formals = {} + + def resolve(self): + if self._scope is None: + if self.parent is None: + self._scope = self.index.pymodule.get_scope() + else: + parent_scope = self.parent.resolve() + for scope in parent_scope.get_scopes(): + if scope.pyobject.get_ast() is self.node: + self._scope = scope + break + if self._scope is None: + raise RuntimeError("source owner has no structural scope") + if ( + (self.header_slots or self.redirected_headers) + and not self._installed_slots + and not self._installing_slots + ): + self._installing_slots = True + try: + self._install_header_slots() + self._installed_slots = True + finally: + self._installing_slots = False + return self._scope + + def _install_header_slots(self): + from rope.base import pynamesdef + + names = self._scope.get_names() + for name, assignments in self.header_slots.items(): + binding = names.get(name) + if not isinstance(binding, (pynames.AssignedName, pynames.ParameterName)): + binding = pynamesdef.AssignedName( + assignments[0].get_lineno(), self.index.pymodule + ) + names[name] = binding + if self.get_kind() in ("Module", "Class"): + self._scope.pyobject._get_structural_attributes()[name] = binding + attributes = self._scope.pyobject.attributes.get() + if attributes is not None: + attributes[name] = binding + binding.binding_scope = self._scope + if isinstance(binding, pynames.AssignedName): + present = { + id(assignment.ast_node) for assignment in binding.assignments + } + binding.assignments.extend( + assignment + for assignment in assignments + if id(assignment.ast_node) not in present + ) + binding.invalidate() + for name, owner in self.redirected_headers.items(): + names[name] = owner.resolve().get_names()[name] + + def lookup(self, name): + return self.resolve().lookup(name) + + def _propagated_lookup(self, name): + return self.resolve()._propagated_lookup(name) + + def get_names(self): + return self.resolve().get_names() + + def get_defined_names(self): + return self.resolve().get_defined_names() + + def get_propagated_names(self): + return self.resolve().get_propagated_names() + + def get_module(self): + return self.index.pymodule + + def get_region(self): + return self.resolve().get_region() + + def __getattr__(self, name): + return getattr(self.resolve(), name) + + def get_kind(self): + if isinstance(self.node, ast.ClassDef): + return "Class" + if isinstance(self.node, (ast.FunctionDef, ast.AsyncFunctionDef)): + return "Function" + return "Module" + + @property + def pyobject(self): + return self.resolve().pyobject + + +class _PropagationView: + """Use an enclosing environment without inheriting ordinary class names.""" + + def __init__(self, environment): + self.environment = environment + self.parent = environment.parent + + def lookup(self, name): + return self.environment._propagated_lookup(name) + + def _propagated_lookup(self, name): + return self.lookup(name) + + def get_names(self): + return self.environment.get_propagated_names() + + def get_propagated_names(self): + return self.get_names() + + def __getattr__(self, name): + return getattr(self.environment, name) + + +@dataclass(frozen=True) +class AnnotationPolicy: + mode: str + runtime_profile: tuple + resolver_contract: str + + +@dataclass(frozen=True) +class SourceContext: + _environment: object + _structural_scope: object + source_node: object + role: str + annotation_owner: object = None + annotation_policy: object = None + + @property + def environment(self): + return _resolved(self._environment) + + @property + def structural_scope(self): + return _resolved(self._structural_scope) + + +@dataclass(frozen=True) +class SourceFragment: + start: int + end: int + depth: int + context: SourceContext + + +class SourceLocalName(pynames.AssignedName): + def __init__(self, environment, name, node): + self.environment = environment + self.binding_scope = environment + self.name = name + self.node = node + self.assignments = [] + self._value_name = None + + def get_object(self): + if self.assignments: + if self._value_name is None: + from rope.base import pynamesdef + + self._value_name = pynamesdef.AssignedName( + self.node.lineno, self.environment.get_module() + ) + self._value_name.assignments.extend(self.assignments) + return self._value_name.get_object() + return pyobjects.get_unknown() + + def get_definition_location(self): + return self.environment.get_module(), self.node.lineno + + def invalidate(self): + if self._value_name is not None: + self._value_name.invalidate() + + +class LambdaParameterName(pynames.ParameterName): + def __init__(self, pyfunction, name, index, node): + self.pyfunction = pyfunction + self.binding_scope = pyfunction.get_scope() + self.name = name + self.index = index + self.node = node + + def get_object(self): + return pyobjects.get_unknown() + + def get_objects(self): + return [] + + def get_definition_location(self): + return self.pyfunction.get_module(), self.node.lineno + + +class TypeParameterName(SourceLocalName): + pass + + +class SourceValueName(pynames.PyName): + """An attribute's call value, retaining its canonical source name.""" + + def __init__(self, source_pyname, pyobject): + self.source_pyname = source_pyname + self.pyobject = pyobject + + def get_object(self): + return self.pyobject + + def get_definition_location(self): + return self.source_pyname.get_definition_location() + + def __getattr__(self, name): + return getattr(self.source_pyname, name) + + +def source_lambda_attribute_value(primary, attribute): + """Separate an initializer's lambda value from the class namespace value.""" + from rope.base import evaluate + + receiver = primary.get_object() + class_access = isinstance(receiver, pyobjects.AbstractClass) + receiver_class = receiver if class_access else receiver.get_type() + if not isinstance(receiver_class, pyobjects.PyClass): + return None + module, _ = attribute.get_definition_location() + if module is None: + return None + index = module.get_source_binding_index() + storage = {"class": [], "instance": []} + for assignment in getattr(attribute, "assignments", ()): + context = index.context_for_node(assignment.ast_node) + if context is None or assignment.levels: + continue + owner = context.structural_scope + if owner.get_kind() == "Class" and owner.pyobject is receiver_class: + kind = "class" + elif ( + owner.get_kind() == "Function" + and owner.pyobject.parent is receiver_class + and owner.pyobject.get_name() == "__init__" + and owner.pyobject.get_kind() == "method" + ): + kind = "instance" + else: + continue + if any( + isinstance(statement, (ast.Assign, ast.AnnAssign)) + and statement.value is assignment.ast_node + for statement in owner.pyobject.get_ast().body + ): + storage[kind].append((assignment, context)) + if not storage["class"] or not storage["instance"]: + return None + assignment, context = storage["class" if class_access else "instance"][-1] + value = evaluate.eval_node(context.environment, assignment.ast_node) + function = value.get_object() if value is not None else None + if isinstance(function, SourceLambdaFunction): + return SourceValueName(attribute, function) + return None + + +class OpaqueSourceName(pynames.PyName): + is_refactoring_identity_safe = False + + def __init__(self, environment, name, node, candidates): + self.environment = environment + self.name = name + self.node = node + self.reason = "generic annotation resolver identity is not established" + self.candidate_environments = tuple(candidates) + + def get_object(self): + return pyobjects.get_unknown() + + def get_definition_location(self): + return self.environment.get_module(), self.node.lineno + + @property + def candidate_binders(self): + result = [] + for environment in self.candidate_environments: + candidate = environment.lookup(self.name) + if candidate is not None and candidate is not self: + if ( + not isinstance(candidate, OpaqueSourceName) + and candidate not in result + ): + result.append(candidate) + return tuple(result) + + def may_reference(self, pyname): + from rope.refactor.occurrences import same_pyname + + return any( + same_pyname(candidate, pyname) for candidate in self.candidate_binders + ) + + +class UnsupportedFormalName(OpaqueSourceName): + """A readable lexical binder outside the ordinary function's old protocol.""" + + def __init__(self, environment, name, node): + super().__init__(environment, name, node, ()) + self.binding_scope = environment + self.reason = "ordinary function parameter category is not supported" + + +@dataclass(frozen=True) +class OpaqueReference: + node: object + context: SourceContext + pyname: OpaqueSourceName + + def may_reference(self, pyname): + return self.pyname.may_reference(pyname) + + +class _SourceObject(pyobjects.PyObject): + def __init__(self, environment): + super().__init__(pyobjects.get_base_type("Unknown")) + self.environment = environment + + def get_module(self): + return self.environment.get_module() + + def get_scope(self): + return self.environment + + def get_ast(self): + return self.environment.node + + def get_attributes(self): + return self.environment.get_names() + + def get_kind(self): + return self.environment.get_kind() + + +class BindingEnvironment(pyscopes.Scope): + def __init__(self, index, node, parent, structural_scope, kind): + self.index = index + self.node = node + self.structural_reference = structural_scope + self.kind = kind + self.names = {} + self.children = [] + self.capture_parent = parent + super().__init__(index.pymodule.pycore, _SourceObject(self), parent) + + @property + def structural_scope(self): + return _resolved(self.structural_reference) + + def get_module(self): + return self.index.pymodule + + def get_names(self): + return self.names + + def get_defined_names(self): + return self.names + + def _create_scopes(self): + return [] + + def get_binding_scopes(self): + return tuple(self.children) + + def get_kind(self): + return self.kind + + def get_start(self): + return self.node.lineno + + def get_end(self): + return getattr(self.node, "end_lineno", self.node.lineno) + + def get_body_start(self): + return self.get_start() + + def get_logical_end(self): + return self.get_end() + + def get_region(self): + return self.index._span(self.node) + + def _get_global_scope(self): + return self.index.pymodule.get_scope() + + start = property(get_start) + end = property(get_end) + logical_end = property(get_logical_end) + + +class AnnotationEnvironment(BindingEnvironment): + def __init__( + self, index, node, entry, owner, policy, generic_names, candidates=None + ): + super().__init__(index, node, entry, owner, "Annotation") + self.policy = policy + self.capture_parent = index._capture(entry) + self.generic_names = generic_names + self.candidates = candidates + self.opaque_names = {} + + def lookup(self, name): + if name in self.generic_names: + if name not in self.opaque_names: + candidates = self.candidates or self.index._candidate_environments( + self.parent + ) + self.opaque_names[name] = OpaqueSourceName( + self, name, self.node, candidates + ) + return self.opaque_names[name] + class_reference = self.parent + if isinstance(class_reference, _GenericOwnerEnvironment): + class_reference = class_reference.structural_reference + if ( + self.policy.mode in ("eager", "lazy") + and isinstance(class_reference, _StructuralReference) + and class_reference.get_kind() == "Class" + ): + cutoff = self.node if self.policy.mode == "eager" else None + status = self.index._class_binding_before( + class_reference.node, name, cutoff + ) + if status == "absent": + return self.index._class_fallback(class_reference, name).lookup(name) + if status == "ambiguous": + if name not in self.opaque_names: + fallback = self.index._class_fallback(class_reference, name) + self.opaque_names[name] = OpaqueSourceName( + self, name, self.node, (class_reference, fallback) + ) + return self.opaque_names[name] + return self.parent.lookup(name) + + def _propagated_lookup(self, name): + return self.lookup(name) + + def get_names(self): + names = set(self.generic_names) + environment = self.parent + direct = True + while environment is not None: + visible = ( + environment.get_names() + if direct + else environment.get_propagated_names() + ) + names.update(visible) + if not isinstance(environment, TypeParameterEnvironment): + direct = False + environment = environment.parent + result = {} + for name in names: + binding = self.lookup(name) + if binding is not None: + result[name] = binding + return result + + def get_propagated_names(self): + return self.get_names() + + +class TypeAliasEnvironment(AnnotationEnvironment): + """A lazy alias resolver with its own canonical type parameters.""" + + def __init__(self, index, node, entry, generic, owner): + super().__init__( + index, + node, + generic, + owner, + index.alias_policy, + index._generic_names(generic), + ) + self.entry_environment = entry + + def _class_reference(self): + entry = self.entry_environment + if isinstance(entry, _GenericOwnerEnvironment) and entry.kind == "Class": + entry = entry.structural_reference + if isinstance(entry, _StructuralReference) and entry.get_kind() == "Class": + return entry + return None + + def _parameter_environment(self, name): + environment = self.parent + while environment is not None: + if ( + isinstance(environment, TypeParameterEnvironment) + and name in environment.names + ): + return environment + environment = environment.parent + return None + + def opaque_binding(self, name): + class_reference = self._class_reference() + if class_reference is None: + return None + status = self.index._class_binding_before(class_reference.node, name, None) + parameter = self._parameter_environment(name) + if parameter is not None and status != "absent": + candidates = (class_reference, parameter) + elif parameter is None and status == "ambiguous": + candidates = ( + class_reference, + self.index._class_fallback(class_reference, name), + ) + else: + return None + if name not in self.opaque_names: + self.opaque_names[name] = OpaqueSourceName( + self, name, self.node, candidates + ) + return self.opaque_names[name] + + def lookup(self, name): + opaque = self.opaque_binding(name) + if opaque is not None: + return opaque + class_reference = self._class_reference() + if class_reference is not None and name not in self.generic_names: + status = self.index._class_binding_before(class_reference.node, name, None) + if status == "absent": + return self.index._class_fallback(class_reference, name).lookup(name) + return self.parent.lookup(name) + + +class ComprehensionBindingScope(BindingEnvironment): + pass + + +class LambdaBindingScope(BindingEnvironment): + pass + + +class TypeParameterEnvironment(BindingEnvironment): + def lookup(self, name): + return self.names[name] if name in self.names else self.parent.lookup(name) + + +class _LambdaArgumentValues: + """Match values locally without changing the ordinary Arguments protocol.""" + + def __init__(self, function, receiver): + self.function = function + self.receiver_name = ( + function.positional[0] + if receiver is not None and function.positional + else None + ) + + def match(self, argument_nodes, argument_scope): + from rope.base import evaluate + + values = {} + positional = [ + name for name in self.function.positional if name != self.receiver_name + ] + keyword_names = set(self.function.keywords) - {self.receiver_name} + next_position = 0 + expanded_position = False + expanded_keyword = False + keyword_supplied = set() + conflicts = set() + for node in argument_nodes: + if isinstance(node, ast.Starred): + expanded_position = True + elif isinstance(node, ast.keyword): + if node.arg is None: + expanded_keyword = True + elif node.arg in keyword_names: + if node.arg in values: + conflicts.add(node.arg) + keyword_supplied.add(node.arg) + values[node.arg] = evaluate.eval_node(argument_scope, node.value) + elif not expanded_position: + if next_position < len(positional): + values[positional[next_position]] = evaluate.eval_node( + argument_scope, node + ) + next_position += 1 + for name in tuple(values): + if ( + name in conflicts + or (expanded_keyword and name in keyword_names) + or ( + expanded_position + and name in positional + and name in keyword_supplied + ) + ): + values[name] = None + for name, expression in self.function.defaults.items(): + if name in values or name == self.receiver_name: + continue + if expanded_position and name in positional: + continue + if expanded_keyword and name in keyword_names: + continue + values[name] = evaluate.eval_node( + _resolved(self.function.creation_scope), expression + ) + return values + + +class SourceLambdaFunction(pyobjects.AbstractFunction): + def __init__(self, index, node, creation_scope, scope): + super().__init__() + self.index = index + self.node = node + self.creation_scope = creation_scope + self.source_function = self + self.bound_creation_scope = None + self.bound_receiver = None + self.descriptor_kind = "function" + self.scope = scope + args = node.args + self.posonly = [arg.arg for arg in args.posonlyargs] + self.positional = self.posonly + [arg.arg for arg in args.args] + self.keywords = [arg.arg for arg in args.args + args.kwonlyargs] + self.param_names = self.positional + [arg.arg for arg in args.kwonlyargs] + if args.vararg is not None: + self.param_names.append(args.vararg.arg) + if args.kwarg is not None: + self.param_names.append(args.kwarg.arg) + self.defaults = dict( + zip( + self.positional[len(self.positional) - len(args.defaults) :], + args.defaults, + ) + ) + self.defaults.update( + (arg.arg, default) + for arg, default in zip(args.kwonlyargs, args.kw_defaults) + if default is not None + ) + + def get_parameters(self): + return {name: self.scope.names[name] for name in self.param_names} + + def bind_creation_scope(self, scope): + bound = object.__new__(SourceLambdaFunction) + bound.__dict__ = dict(self.__dict__) + bound.source_function = self.source_function + bound.bound_creation_scope = scope + bound.creation_scope = scope + return bound + + def as_descriptor(self, kind): + descriptor = object.__new__(SourceLambdaFunction) + descriptor.__dict__ = dict(self.__dict__) + descriptor.descriptor_kind = kind + return descriptor + + def bind_receiver(self, receiver): + bound = object.__new__(SourceLambdaFunction) + bound.__dict__ = dict(self.__dict__) + bound.bound_receiver = receiver + return bound + + def descriptor_receiver(self, primary, attribute): + from rope.base import evaluate + + if self.bound_receiver is not None: + return None + receiver = primary.get_object() + receiver_class = ( + receiver + if isinstance(receiver, pyobjects.AbstractClass) + else receiver.get_type() + ) + if not isinstance(receiver_class, pyobjects.PyClass): + return None + module, _ = attribute.get_definition_location() + if module is None: + return None + class_storage = False + for assignment in getattr(attribute, "assignments", ()): + context = module.get_source_binding_index().context_for_node( + assignment.ast_node + ) + if context is None or context.structural_scope.get_kind() != "Class": + continue + value = evaluate.eval_node(context.environment, assignment.ast_node) + assigned = value.get_object() if value is not None else None + if ( + isinstance(assigned, SourceLambdaFunction) + and assigned.source_function is self.source_function + ): + class_storage = True + break + if not class_storage or self.descriptor_kind == "staticmethod": + return None + if self.descriptor_kind == "classmethod": + return pynames.UnboundName(receiver_class) + if not isinstance(receiver, pyobjects.AbstractClass): + return primary + return None + + @property + def parent(self): + return self.scope.structural_scope.pyobject + + def get_param_names(self, special_args=True): + if not special_args: + return self.positional + [arg.arg for arg in self.node.args.kwonlyargs] + return list(self.param_names) + + def get_positional_param_names(self): + return list(self.positional) + + def get_keyword_param_names(self): + return list(self.keywords) + + def get_parameter_defaults(self): + return dict(self.defaults) + + def get_scope(self): + return self.scope + + def get_module(self): + return self.index.pymodule + + def get_ast(self): + return self.node + + def get_name(self): + return "lambda" + + def get_kind(self): + return "lambda" + + def get_attributes(self): + return {} + + def get_returned_object(self, arguments): + from rope.base import arguments as argument_matching + from rope.base import evaluate + + values = {} + receiver = self.bound_receiver + if isinstance(arguments, argument_matching.MixedArguments): + if receiver is None: + receiver = arguments.pyname + arguments = arguments.args + if arguments is not None and hasattr(arguments, "args"): + values.update( + _LambdaArgumentValues(self, receiver).match( + arguments.args, arguments.scope + ) + ) + elif arguments is not None: + values.update( + zip(self.param_names, arguments.get_pynames(self.param_names)) + ) + else: + values.update( + _LambdaArgumentValues(self, receiver).match( + [], _resolved(self.creation_scope) + ) + ) + if receiver is not None and self.positional: + values[self.positional[0]] = receiver + for name in self.param_names: + if values.get(name) is None: + values[name] = pynames.UnboundName() + capture = self.bound_creation_scope + if capture is None: + capture = _resolved(self.scope.parent) + overlay = evaluate.SourceValueScope(self.scope, capture, values) + result = evaluate.eval_node(overlay, self.node.body) + return result.get_object() if result is not None else pyobjects.get_unknown() + + +class SourceBindingIndex: + def __init__(self, pymodule): + self.pymodule = pymodule + self.source = pymodule.source_code + self.fragments = [] + self._fragment_positions = {} + self.node_contexts = {} + self.token_bindings = [] + self.source_callables = {} + self.comprehensions = {} + self.environments = [] + self._opaque_references = [] + self.lines = codeanalyze.ASTLinesAdapter(self.source) + tree = pymodule.get_ast() + self.global_reference = _StructuralReference(self, tree) + self.owner_references = [self.global_reference] + self._preparing_headers = False + self._class_symbols = {} + self.future_annotations = any( + isinstance(node, ast.ImportFrom) + and node.module == "__future__" + and any(alias.name == "annotations" for alias in node.names) + for node in tree.body + ) + mode = ( + "future-string" + if self.future_annotations + else "lazy" if sys.version_info >= (3, 14) else "eager" + ) + self.policy = AnnotationPolicy( + mode, + sys.version_info[:2], + "default-function-hints" if self.future_annotations else "source-value", + ) + self.alias_policy = AnnotationPolicy( + "lazy", sys.version_info[:2], "source-value" + ) + self.state = "BUILDING" + self._visit(tree, self.global_reference, self.global_reference, "module", 0) + self._owners_by_node = { + id(owner.node): owner for owner in self.owner_references + } + self.fragments.sort(key=lambda item: (item.start, item.end, item.depth)) + self.state = "FROZEN" + + def _span(self, node): + if not hasattr(node, "lineno"): + return 0, len(self.source) + return self.lines[node] + + def _record(self, node, environment, owner, role, depth, annotation=None): + if not hasattr(node, "lineno") and not isinstance(node, ast.Module): + return None + context = SourceContext( + environment, + owner, + node, + role, + annotation, + ( + ( + self.alias_policy + if isinstance(annotation, getattr(ast, "TypeAlias", ())) + else self.policy + ) + if annotation is not None + else None + ), + ) + self.node_contexts[id(node)] = context + if hasattr(node, "lineno"): + start, end = self._span(node) + fragment = SourceFragment(start, end, depth, context) + position = self._fragment_positions.get(id(node)) + if position is None: + self._fragment_positions[id(node)] = len(self.fragments) + self.fragments.append(fragment) + else: + self.fragments[position] = fragment + return context + + def context_at(self, offset): + matched = [item for item in self.fragments if item.start <= offset < item.end] + if matched: + return max( + matched, key=lambda item: (item.depth, -item.end + item.start) + ).context + scope = self.pymodule.get_scope().get_structural_scope_for_offset(offset) + return SourceContext(scope, scope, self.pymodule.get_ast(), "structure") + + def context_for_node(self, node): + return self.node_contexts.get(id(node)) + + def binding_for_token(self, offset): + for start, end, binding in self.token_bindings: + if start <= offset < end: + if isinstance(binding, tuple): + owner, name = binding + return owner.resolve().get_names().get(name) + return binding + return None + + def callable_for_node(self, node): + return self.source_callables.get(id(node)) + + def comprehension_scope(self, node): + return self.comprehensions.get(id(node)) + + def prepare_header_bindings(self, scope=None): + if self.state == "FROZEN" and not self._preparing_headers: + self._preparing_headers = True + try: + if scope is not None: + owner = self._owners_by_node.get(id(scope.pyobject.get_ast())) + if owner is not None and ( + owner.header_slots or owner.redirected_headers + ): + owner._scope = scope + owner.resolve() + else: + for owner in self.owner_references: + if ( + owner.header_slots or owner.redirected_headers + ) and owner.get_kind() in ("Module", "Class"): + owner.resolve() + finally: + self._preparing_headers = False + + def install_unsupported_formals(self, scope): + owner = self._owners_by_node.get(id(scope.pyobject.get_ast())) + if owner is not None: + scope.names.update(owner.unsupported_formals) + + def opaque_references(self): + known = {id(reference.node) for reference in self._opaque_references} + for context in tuple(self.node_contexts.values()): + node = context.source_node + if ( + context.annotation_owner is not None + and isinstance(node, ast.Name) + and isinstance(node.ctx, ast.Load) + and id(node) not in known + ): + binding = context.environment.lookup(node.id) + if isinstance(binding, OpaqueSourceName): + self._opaque_references.append( + OpaqueReference(node, context, binding) + ) + known.add(id(node)) + return tuple(self._opaque_references) + + def children_of(self, environment): + if isinstance(environment, BindingEnvironment): + return tuple(environment.children) + return tuple( + candidate + for candidate in self.environments + if _resolved(candidate.parent) is environment + ) + + def _capture(self, environment): + if isinstance(environment, TypeAliasEnvironment): + return environment.capture_parent + if isinstance(environment, AnnotationEnvironment): + if environment.generic_names: + capture = AnnotationEnvironment( + self, + environment.node, + environment.capture_parent, + environment.structural_reference, + environment.policy, + environment.generic_names, + environment.candidates, + ) + capture.opaque_names = environment.opaque_names + return self._environment(capture) + return environment.capture_parent + if ( + isinstance(environment, _GenericOwnerEnvironment) + and environment.kind == "Class" + ): + return _PropagationView(environment.parent) + if isinstance(environment, TypeParameterEnvironment): + return _PropagationView(environment) + if ( + isinstance(environment, _StructuralReference) + and environment.get_kind() == "Class" + ): + return self._capture(environment.entry) + return environment + + def _candidate_environments(self, environment): + result = [] + while environment is not None: + if environment not in result: + result.append(environment) + environment = environment.parent + if self.global_reference not in result: + result.append(self.global_reference) + return result + + def _environment(self, environment): + self.environments.append(environment) + if isinstance(environment.parent, BindingEnvironment): + environment.parent.children.append(environment) + return environment + + def _token(self, node, name, binding): + start = self._span(node)[0] + while self.source[start : start + 1] == "*": + start += 1 + self.token_bindings.append((start, start + len(name), binding)) + + def _generic_names(self, environment): + names = set() + while environment is not None: + if isinstance(environment, TypeParameterEnvironment): + names.update(environment.names) + environment = environment.parent + return names + + def _class_fallback(self, reference, name): + names, globals_, nonlocals = self._class_lexical_symbols(reference.node) + if name in globals_ or (name in names and name not in nonlocals): + return self.global_reference + return _PropagationView(reference.entry) + + def _class_lexical_symbols(self, node): + cached = self._class_symbols.get(id(node)) + if cached is not None: + return cached + names, globals_, nonlocals = set(), set(), set() + + def visit(expression): + if isinstance(expression, (ast.FunctionDef, ast.AsyncFunctionDef)): + names.add(expression.name) + headers = expression.decorator_list + expression.args.defaults + headers += [ + value for value in expression.args.kw_defaults if value is not None + ] + if self.policy.mode == "eager": + arguments = ( + expression.args.posonlyargs + + expression.args.args + + expression.args.kwonlyargs + ) + arguments += [ + arg + for arg in (expression.args.vararg, expression.args.kwarg) + if arg is not None + ] + headers += [ + arg.annotation + for arg in arguments + if arg.annotation is not None + ] + if expression.returns is not None: + headers.append(expression.returns) + for header in headers: + visit(header) + return + if isinstance(expression, ast.ClassDef): + names.add(expression.name) + for header in expression.decorator_list + expression.bases: + visit(header) + for keyword in expression.keywords: + visit(keyword.value) + return + if isinstance(expression, ast.Lambda): + for default in expression.args.defaults + [ + value for value in expression.args.kw_defaults if value is not None + ]: + visit(default) + return + if isinstance( + expression, (ast.ListComp, ast.SetComp, ast.DictComp, ast.GeneratorExp) + ): + visit(expression.generators[0].iter) + return + if isinstance(expression, ast.Name) and isinstance( + expression.ctx, (ast.Store, ast.Del) + ): + names.add(expression.id) + elif isinstance(expression, ast.Global): + globals_.update(expression.names) + elif isinstance(expression, ast.Nonlocal): + nonlocals.update(expression.names) + elif isinstance(expression, (ast.Import, ast.ImportFrom)): + names.update( + alias.asname or alias.name.split(".")[0] + for alias in expression.names + ) + elif isinstance(expression, ast.ExceptHandler) and expression.name: + names.add(expression.name) + elif ( + isinstance(expression, (ast.MatchAs, ast.MatchStar)) and expression.name + ): + names.add(expression.name) + elif isinstance(expression, ast.MatchMapping) and expression.rest: + names.add(expression.rest) + for child in ast.iter_child_nodes(expression): + visit(child) + + for statement in node.body: + visit(statement) + result = names, globals_, nonlocals + self._class_symbols[id(node)] = result + return result + + def _class_binding_before(self, node, name, expression): + status = "absent" + for statement in node.body: + if isinstance(statement, (ast.FunctionDef, ast.AsyncFunctionDef)): + headers = list(statement.decorator_list) + list(statement.args.defaults) + headers.extend( + value for value in statement.args.kw_defaults if value is not None + ) + if self.policy.mode == "eager": + arguments = statement.args.posonlyargs + statement.args.args + arguments += ( + [statement.args.vararg] + if statement.args.vararg is not None + else [] + ) + arguments += statement.args.kwonlyargs + arguments += ( + [statement.args.kwarg] + if statement.args.kwarg is not None + else [] + ) + headers.extend( + arg.annotation + for arg in arguments + if arg.annotation is not None + ) + if statement.returns is not None: + headers.append(statement.returns) + for header in headers: + if expression is header: + if self._header_binding_status(header, name) != "absent": + return "ambiguous" + return status + if expression is not None and statement.lineno > expression.lineno: + break + header_status = self._header_binding_status(header, name) + if header_status != "absent": + status = header_status + if expression is not None and statement.lineno >= expression.lineno: + break + if statement.name == name: + status = "bound" + continue + if expression is not None and statement.lineno >= expression.lineno: + break + if isinstance(statement, ast.ClassDef): + if statement.name == name: + status = "bound" + continue + if isinstance(statement, (ast.Assign, ast.AnnAssign, ast.AugAssign)): + if isinstance(statement, ast.AnnAssign) and statement.value is None: + continue + targets = ( + statement.targets + if isinstance(statement, ast.Assign) + else [statement.target] + ) + if any( + isinstance(child, ast.Name) + and child.id == name + and isinstance(child.ctx, ast.Store) + for target in targets + for child in ast.walk(target) + ): + status = "bound" + continue + if isinstance(statement, (ast.Import, ast.ImportFrom)): + if any( + (alias.asname or alias.name.split(".")[0]) == name + for alias in statement.names + ): + status = "bound" + continue + if isinstance(statement, ast.Delete): + if any( + isinstance(child, ast.Name) and child.id == name + for child in ast.walk(statement) + ): + status = "absent" + continue + if any( + ( + isinstance(child, ast.Name) + and child.id == name + and isinstance(child.ctx, (ast.Store, ast.Del)) + ) + or ( + isinstance(child, (ast.ExceptHandler, ast.MatchAs, ast.MatchStar)) + and child.name == name + ) + or (isinstance(child, ast.MatchMapping) and child.rest == name) + for child in ast.walk(statement) + ): + status = "ambiguous" + return status + + def _header_binding_status(self, expression, name): + if isinstance(expression, ast.NamedExpr) and expression.target.id == name: + return "bound" + if isinstance(expression, ast.Lambda): + expressions = expression.args.defaults + [ + value for value in expression.args.kw_defaults if value is not None + ] + statuses = [ + self._header_binding_status(value, name) for value in expressions + ] + return next( + (status for status in reversed(statuses) if status != "absent"), + "absent", + ) + if any( + isinstance(child, ast.NamedExpr) and child.target.id == name + for child in ast.walk(expression) + ): + return "ambiguous" + return "absent" + + def _header_binding_owner(self, owner, name): + if owner.parent is None: + return owner + _, globals_, nonlocals = self._class_lexical_symbols(owner.node) + if name in globals_: + return self.global_reference + if name not in nonlocals: + return owner + outer = owner.parent + while outer is not None: + if outer.get_kind() == "Function": + names, outer_globals, outer_nonlocals = self._class_lexical_symbols( + outer.node + ) + args = outer.node.args + parameters = args.posonlyargs + args.args + args.kwonlyargs + parameters += [ + arg for arg in (args.vararg, args.kwarg) if arg is not None + ] + locals_ = names | {arg.arg for arg in parameters} + if ( + name in locals_ + and name not in outer_globals + and name not in outer_nonlocals + ): + return outer + outer = outer.parent + raise exceptions.RefactoringError( + "Cannot resolve the declared nonlocal source binding." + ) + + def _register_header_binding(self, owner, target, value): + declaring = self._header_binding_owner(owner, target.id) + declaring.header_slots.setdefault(target.id, []).append( + pynames.AssignmentValue(value) + ) + if declaring is not owner: + owner.redirected_headers[target.id] = declaring + self._token(target, target.id, (declaring, target.id)) + return declaring + + def _typeparams(self, node, entry, owner, depth, visit_expressions=True): + parameters = getattr(node, "type_params", ()) + if not parameters: + return entry + environment = self._environment( + TypeParameterEnvironment(self, node, entry, owner, "TypeParameters") + ) + for parameter in parameters: + name = parameter.name + binding = TypeParameterName(environment, name, parameter) + environment.names[name] = binding + self._record(parameter, environment, owner, "type-parameter-token", depth) + self._token(parameter, name, binding) + if visit_expressions: + for parameter in parameters: + for field in ("bound", "default_value"): + expression = getattr(parameter, field, None) + if expression is not None: + self._visit( + expression, + environment, + owner, + "type-parameter-expression", + depth + 1, + ) + return environment + + def _visit(self, node, environment, owner, role, depth, annotation=None): + if node is None: + return + context = self._record(node, environment, owner, role, depth, annotation) + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): + self._function(node, environment, owner, depth) + return + if isinstance(node, getattr(ast, "TypeAlias", ())): + generic = self._typeparams( + node, environment, owner, depth + 1, visit_expressions=False + ) + alias_environment = self._environment( + TypeAliasEnvironment(self, node, environment, generic, owner) + ) + self._visit(node.name, environment, owner, "type-alias-name", depth + 1) + for parameter in node.type_params: + for field in ("bound", "default_value"): + expression = getattr(parameter, field, None) + if expression is not None: + self._visit( + expression, + alias_environment, + owner, + "type-parameter-expression", + depth + 2, + node, + ) + self._visit( + node.value, + alias_environment, + owner, + "type-alias-value", + depth + 1, + node, + ) + return + if isinstance(node, ast.ClassDef): + class_owner = _StructuralReference(self, node, owner, environment) + self.owner_references.append(class_owner) + generic = self._typeparams(node, environment, class_owner, depth + 1) + for expression in node.decorator_list: + self._visit(expression, environment, owner, "decorator", depth + 1) + for expression in node.bases: + self._visit(expression, generic, class_owner, "class-base", depth + 1) + for keyword in node.keywords: + self._visit( + keyword.value, generic, class_owner, "class-keyword", depth + 1 + ) + body_env = class_owner + if self._generic_names(generic): + body_env = self._environment( + _GenericOwnerEnvironment(self, node, generic, class_owner, "Class") + ) + self._record(node, body_env, class_owner, "class-body", depth) + for statement in node.body: + self._visit(statement, body_env, class_owner, "class-body", depth + 1) + return + if isinstance(node, ast.Lambda): + self._lambda(node, environment, owner, depth, annotation) + return + if isinstance( + node, (ast.ListComp, ast.SetComp, ast.DictComp, ast.GeneratorExp) + ): + self._comprehension(node, environment, owner, depth, annotation) + return + if ( + isinstance(node, ast.Name) + and isinstance(node.ctx, ast.Load) + and annotation is not None + ): + candidate = self._opaque_binding(environment, node.id) + if candidate is not None: + self._opaque_references.append( + OpaqueReference(node, context, candidate) + ) + if isinstance(node, ast.NamedExpr): + target_env = environment + while isinstance(target_env, ComprehensionBindingScope): + target_env = target_env.parent + if isinstance(target_env, BindingEnvironment) and not isinstance( + target_env, AnnotationEnvironment + ): + if node.target.id not in target_env.names: + target_env.names[node.target.id] = SourceLocalName( + target_env, node.target.id, node.target + ) + self._token( + node.target, node.target.id, target_env.names[node.target.id] + ) + elif isinstance(target_env, _StructuralReference): + target_env = self._register_header_binding( + target_env, node.target, node.value + ) + elif ( + isinstance(target_env, AnnotationEnvironment) + and target_env.policy.mode == "eager" + ): + declaring = target_env.parent + while isinstance(declaring, BindingEnvironment): + declaring = declaring.parent + if isinstance(declaring, _StructuralReference): + target_env = self._register_header_binding( + declaring, node.target, node.value + ) + self._visit( + node.target, + target_env, + owner, + "named-expression-target", + depth + 1, + annotation, + ) + self._visit(node.value, environment, owner, role, depth + 1, annotation) + return + for child in ast.iter_child_nodes(node): + self._visit(child, environment, owner, role, depth + 1, annotation) + + def _opaque_binding(self, environment, name): + while isinstance(environment, BindingEnvironment): + if name in environment.names: + return None + if isinstance(environment, TypeAliasEnvironment): + return environment.opaque_binding(name) + if isinstance(environment, AnnotationEnvironment): + if name in environment.generic_names: + return environment.lookup(name) + class_reference = environment.parent + if isinstance(class_reference, _GenericOwnerEnvironment): + class_reference = class_reference.structural_reference + if ( + environment.policy.mode in ("eager", "lazy") + and isinstance(class_reference, _StructuralReference) + and class_reference.get_kind() == "Class" + ): + cutoff = ( + environment.node if environment.policy.mode == "eager" else None + ) + if ( + self._class_binding_before(class_reference.node, name, cutoff) + == "ambiguous" + ): + if name not in environment.opaque_names: + fallback = self._class_fallback(class_reference, name) + environment.opaque_names[name] = OpaqueSourceName( + environment, + name, + environment.node, + (class_reference, fallback), + ) + return environment.opaque_names[name] + return None + environment = environment.parent + return None + + def _function(self, node, entry, parent_owner, depth): + owner = _StructuralReference(self, node, parent_owner, entry) + self.owner_references.append(owner) + generic = self._typeparams(node, entry, owner, depth + 1) + args = node.args + parameters = args.posonlyargs + args.args + args.kwonlyargs + parameters += [arg for arg in (args.vararg, args.kwarg) if arg is not None] + for parameter in parameters: + self._record(parameter, owner, owner, "formal-token", depth + 1) + if parameter in args.posonlyargs + args.kwonlyargs: + binding = UnsupportedFormalName(owner, parameter.arg, parameter) + owner.unsupported_formals[parameter.arg] = binding + self._token(parameter, parameter.arg, binding) + else: + self._token(parameter, parameter.arg, (owner, parameter.arg)) + for expression in node.decorator_list: + self._visit(expression, entry, parent_owner, "decorator", depth + 1) + for expression in args.defaults + [ + value for value in args.kw_defaults if value is not None + ]: + self._visit(expression, entry, parent_owner, "function-default", depth + 1) + annotated_parameters = args.posonlyargs + args.args + annotated_parameters += [args.vararg] if args.vararg is not None else [] + annotated_parameters += args.kwonlyargs + annotated_parameters += [args.kwarg] if args.kwarg is not None else [] + annotations = [ + (arg.annotation, "parameter-annotation") for arg in annotated_parameters + ] + annotations.append((node.returns, "return-annotation")) + for expression, role in annotations: + if expression is None: + continue + resolver = self.global_reference if self.future_annotations else generic + names = self._generic_names(generic) + candidates = None + if self.future_annotations: + own_parameters = getattr(node, "type_params", ()) + names = set(generic.names) if own_parameters else set() + candidates = ( + (generic, self.global_reference) + if own_parameters + else (self.global_reference,) + ) + annotation_env = self._environment( + AnnotationEnvironment( + self, expression, resolver, owner, self.policy, names, candidates + ) + ) + self._visit(expression, annotation_env, owner, role, depth + 2, node) + body_env = owner + if self._generic_names(generic): + body_env = self._environment( + _GenericOwnerEnvironment(self, node, generic, owner, "Function") + ) + self._record(node, body_env, owner, "function-body", depth) + for statement in node.body: + self._visit(statement, body_env, owner, "function-body", depth + 1) + + def _lambda(self, node, entry, owner, depth, annotation): + environment = self._environment( + LambdaBindingScope(self, node, self._capture(entry), owner, "Lambda") + ) + callable_ = SourceLambdaFunction(self, node, entry, environment) + environment.pyobject = callable_ + self._record(node, environment, owner, "lambda-body", depth, annotation) + self.source_callables[id(node)] = callable_ + parameters = node.args.posonlyargs + node.args.args + node.args.kwonlyargs + parameters += [ + arg for arg in (node.args.vararg, node.args.kwarg) if arg is not None + ] + for number, parameter in enumerate(parameters): + binding = LambdaParameterName(callable_, parameter.arg, number, parameter) + environment.names[parameter.arg] = binding + self._record( + parameter, environment, owner, "lambda-formal", depth + 1, annotation + ) + self._token(parameter, parameter.arg, binding) + self._predeclare_walrus(node.body, environment) + for expression in node.args.defaults + [ + value for value in node.args.kw_defaults if value is not None + ]: + self._visit( + expression, entry, owner, "lambda-default", depth + 1, annotation + ) + self._visit(node.body, environment, owner, "lambda-body", depth + 1, annotation) + + def _predeclare_walrus(self, node, environment): + if isinstance(node, ast.Lambda): + return + if isinstance(node, ast.NamedExpr): + if node.target.id not in environment.names: + environment.names[node.target.id] = SourceLocalName( + environment, node.target.id, node.target + ) + binding = environment.names[node.target.id] + if isinstance(binding, SourceLocalName): + binding.assignments.append(pynames.AssignmentValue(node.value)) + for child in ast.iter_child_nodes(node): + self._predeclare_walrus(child, environment) + + def _comprehension(self, node, entry, owner, depth, annotation): + environment = self._environment( + ComprehensionBindingScope( + self, node, self._capture(entry), owner, "Comprehension" + ) + ) + self.comprehensions[id(node)] = environment + self._record(node, environment, owner, "comprehension-inner", depth, annotation) + for generator in node.generators: + for child in ast.walk(generator.target): + if isinstance(child, ast.Name) and isinstance(child.ctx, ast.Store): + if child.id not in environment.names: + environment.names[child.id] = SourceLocalName( + environment, child.id, child + ) + self._token(child, child.id, environment.names[child.id]) + for name, levels in nameanalyze.get_name_levels(generator.target): + environment.names[name].assignments.append( + pynames.AssignmentValue( + generator.iter, levels, ".__iter__().next()" + ) + ) + for number, generator in enumerate(node.generators): + self._visit( + generator.iter, + entry if number == 0 else environment, + owner, + "comprehension-first-iter" if number == 0 else "comprehension-inner", + depth + 1, + annotation, + ) + self._visit( + generator.target, + environment, + owner, + "comprehension-store", + depth + 1, + annotation, + ) + for condition in generator.ifs: + self._visit( + condition, + environment, + owner, + "comprehension-inner", + depth + 1, + annotation, + ) + for field in ("elt", "key", "value"): + expression = getattr(node, field, None) + if expression is not None: + self._visit( + expression, + environment, + owner, + "comprehension-inner", + depth + 1, + annotation, + ) + + +class _GenericOwnerEnvironment(BindingEnvironment): + def lookup(self, name): + own = self.structural_scope.get_names().get(name) + return own if own is not None else self.parent._propagated_lookup(name) + + def _propagated_lookup(self, name): + if self.kind == "Class": + return self.parent._propagated_lookup(name) + return self.lookup(name) + + def get_names(self): + return self.structural_scope.get_names() diff --git a/rope/contrib/codeassist.py b/rope/contrib/codeassist.py index 32547891e..4e438cfe0 100644 --- a/rope/contrib/codeassist.py +++ b/rope/contrib/codeassist.py @@ -471,6 +471,13 @@ def _find_module(self, pymodule, module_name): return pyname.get_object() def _is_defined_after(self, scope, pyname, lineno): + binding_scope = getattr(pyname, "binding_scope", None) + if binding_scope is not None and binding_scope.get_kind() in { + "Lambda", + "Comprehension", + "TypeParameters", + }: + return False location = pyname.get_definition_location() if location is not None and location[1] is not None: if ( @@ -492,6 +499,25 @@ def _code_completions(self): start = fixsyntax._logical_start(lines, lineno) indents = fixsyntax._get_line_indents(lines[start - 1]) inner_scope = module_scope.get_inner_scope_for_line(start, indents) + transferred = fixer.commenter.transferred_offset(self.offset) + if code.startswith(self.code[: self.offset]): + context = pymodule.get_source_binding_index().context_at( + max(0, transferred if self.starting else transferred - 1) + ) + if context.role in { + "parameter-annotation", + "return-annotation", + "function-default", + "lambda-formal", + "lambda-default", + "lambda-body", + "comprehension-first-iter", + "comprehension-inner", + "comprehension-store", + "type-parameter-token", + "type-parameter-expression", + }: + inner_scope = context.environment if self.word_finder.is_a_name_after_from_import(self.offset): return self._from_import_completions(pymodule) if self.expression.strip() != "": diff --git a/rope/contrib/finderrors.py b/rope/contrib/finderrors.py index 32ee31f98..ac5dd6d82 100644 --- a/rope/contrib/finderrors.py +++ b/rope/contrib/finderrors.py @@ -47,7 +47,9 @@ def __init__(self, pymodule): def _Name(self, node): if isinstance(node.ctx, (ast.Store, ast.Param)): return - scope = self.scope.get_inner_scope_for_line(node.lineno) + scope = ( + self.pymodule.get_source_binding_index().context_for_node(node).environment + ) pyname = scope.lookup(node.id) if pyname is None: self._add_error(node, "Unresolved variable") @@ -56,7 +58,11 @@ def _Name(self, node): def _Attribute(self, node): if not isinstance(node.ctx, ast.Store): - scope = self.scope.get_inner_scope_for_line(node.lineno) + scope = ( + self.pymodule.get_source_binding_index() + .context_for_node(node) + .environment + ) pyname = evaluate.eval_node(scope, node.value) if pyname is not None and pyname.get_object() != pyobjects.get_unknown(): if node.attr not in pyname.get_object(): @@ -73,6 +79,15 @@ def _add_error(self, node, msg): self.errors.append(error) def _is_defined_after(self, scope, pyname, lineno): + binding_scope = getattr(pyname, "binding_scope", None) + if not getattr(pyname, "is_refactoring_identity_safe", True): + return False + if binding_scope is not None and binding_scope.get_kind() in { + "Lambda", + "Comprehension", + "TypeParameters", + }: + return False location = pyname.get_definition_location() if location is not None and location[1] is not None: if ( diff --git a/rope/refactor/introduce_parameter.py b/rope/refactor/introduce_parameter.py index 93b30d0be..f2f4784a1 100644 --- a/rope/refactor/introduce_parameter.py +++ b/rope/refactor/introduce_parameter.py @@ -1,5 +1,5 @@ import rope.base.change # Use fully qualified names for clarity. -from rope.base import codeanalyze, evaluate, exceptions, worder +from rope.base import codeanalyze, evaluate, exceptions, pyscopes, worder from rope.refactor import functionutils, occurrences, sourceutils @@ -39,13 +39,25 @@ def __init__(self, project, resource, offset): self.resource = resource self.offset = offset self.pymodule = self.project.get_pymodule(self.resource) - scope = self.pymodule.get_scope().get_inner_scope_for_offset(offset) - if scope.get_kind() != "Function": + context = self.pymodule.get_source_binding_index().context_at(offset) + scope = context.environment + self.name, self.pyname = self._get_name_and_pyname() + binding_scope = getattr(self.pyname, "binding_scope", None) + if ( + scope.get_kind() in {"Lambda", "Comprehension"} + and context.annotation_owner is None + and ( + binding_scope is None + or binding_scope.get_kind() + not in {"Lambda", "Comprehension", "TypeParameters"} + ) + ): + scope = context.structural_scope + if not isinstance(scope, pyscopes.FunctionScope): raise exceptions.RefactoringError( "Introduce parameter should be performed inside functions" ) self.pyfunction = scope.pyobject - self.name, self.pyname = self._get_name_and_pyname() if self.pyname is None: raise exceptions.RefactoringError( "Cannot find the definition of <%s>" % self.name @@ -62,6 +74,15 @@ def _get_name_and_pyname(self): ) def get_changes(self, new_parameter): + args = self.pyfunction.get_ast().args + if args.posonlyargs or args.kwonlyargs: + raise exceptions.RefactoringError( + "Introduce parameter cannot rewrite unsupported parameter categories." + ) + if not getattr(self.pyname, "is_refactoring_identity_safe", True): + raise exceptions.RefactoringError( + "Cannot refactor a name with an unsupported source binding." + ) definition_info = functionutils.DefinitionInfo.read(self.pyfunction) definition_info.args_with_defaults.append((new_parameter, self._get_primary())) collector = codeanalyze.ChangeCollector(self.resource.read()) diff --git a/rope/refactor/occurrences.py b/rope/refactor/occurrences.py index b136cc5c9..b3afa48df 100644 --- a/rope/refactor/occurrences.py +++ b/rope/refactor/occurrences.py @@ -194,6 +194,8 @@ def same_pyname(expected, pyname): """Check whether `expected` and `pyname` are the same""" if expected is None or pyname is None: return False + expected = getattr(expected, "source_pyname", expected) + pyname = getattr(pyname, "source_pyname", pyname) if expected == pyname: return True if not isinstance( @@ -214,12 +216,35 @@ def unsure_pyname(pyname, unbound=True): """Return `True` if we don't know what this name references""" if pyname is None: return True + if not getattr(pyname, "is_refactoring_identity_safe", True): + return True if unbound and not isinstance(pyname, pynames.UnboundName): return False if pyname.get_object() == pyobjects.get_unknown(): return True +def check_opaque_references(project, pyname, resources, region=None): + if not getattr(pyname, "is_refactoring_identity_safe", True): + raise exceptions.RefactoringError( + "Cannot refactor a name with an unresolved source binding." + ) + for resource in resources: + module = project.get_pymodule(resource) + index = module.get_source_binding_index() + for reference in index.opaque_references(): + if region is not None: + start, _ = codeanalyze.ASTLinesAdapter(module.source_code)[ + reference.node + ] + if not region[0] <= start < region[1]: + continue + if reference.may_reference(pyname): + raise exceptions.RefactoringError( + "Cannot refactor a name possibly referenced in an unresolved annotation binding." + ) + + class PyNameFilter: """For finding occurrences of a name.""" diff --git a/rope/refactor/rename.py b/rope/refactor/rename.py index fcdd17964..5e424bf0c 100644 --- a/rope/refactor/rename.py +++ b/rope/refactor/rename.py @@ -106,7 +106,9 @@ def unsure_func(value=unsure): resources = [self.resource] if resources is None: resources = self.project.get_python_files() + resources = list(resources) self.validate_changes(new_name) + occurrences.check_opaque_references(self.project, self.old_pyname, resources) changes = ChangeSet(f"Renaming <{self.old_name}> to <{new_name}>") finder = occurrences.create_finder( self.project, @@ -197,12 +199,23 @@ def get_old_name(self): return word_finder.get_primary_at(self.offset) def _get_scope_offset(self): - scope = self.pymodule.get_scope().get_inner_scope_for_offset(self.offset) + scope = getattr(self.old_pyname, "binding_scope", None) + if scope is None or scope.get_kind() not in { + "Lambda", + "Comprehension", + "TypeParameters", + }: + scope = self.pymodule.get_scope().get_structural_scope_for_offset( + self.offset + ) return scope.get_region() def get_changes(self, new_name, only_calls=False, reads=True, writes=True): changes = ChangeSet(f"Changing <{self.old_name}> occurrences to <{new_name}>") scope_start, scope_end = self._get_scope_offset() + occurrences.check_opaque_references( + self.project, self.old_pyname, [self.resource], (scope_start, scope_end) + ) finder = occurrences.create_finder( self.project, self.old_name, @@ -257,6 +270,16 @@ def rename_in_module( def _is_local(pyname): + binding_scope = getattr(pyname, "binding_scope", None) + if binding_scope is not None: + return not isinstance( + pyname, pynames.ParameterName + ) and binding_scope.get_kind() in ( + "Function", + "Lambda", + "Comprehension", + "TypeParameters", + ) module, lineno = pyname.get_definition_location() if lineno is None: return False diff --git a/ropetest/source_model_master_test.py b/ropetest/source_model_master_test.py new file mode 100644 index 000000000..82a82adeb --- /dev/null +++ b/ropetest/source_model_master_test.py @@ -0,0 +1,1424 @@ +"""Source identities on the unmodified ordinary-function argument protocol.""" + +import sys +import typing +from textwrap import indent + +import pytest + +from rope.base import ast, evaluate, exceptions, pyobjects +from rope.base.project import Project +from rope.contrib import codeassist, finderrors +from rope.refactor import introduce_parameter, rename + + +@pytest.fixture +def model_project(tmp_path): + project = Project( + str(tmp_path), save_objectdb=False, save_history=False, automatic_soa=False + ) + yield project + project.close() + + +def module(project, source): + resource = project.root.create_file("subject.py") + resource.write(source) + return resource + + +def prefix(mode): + if mode == "eager" and sys.version_info >= (3, 14): + pytest.skip("eager function annotations require Python before 3.14") + if mode == "lazy" and sys.version_info < (3, 14): + pytest.skip("lazy function annotations require Python 3.14") + return "from __future__ import annotations\n" if mode == "future" else "" + + +def observe(source): + namespace = {"__name__": "fixture"} + exec(compile(source, "fixture.py", "exec"), namespace) + return namespace["drive"](), tuple( + typing.get_type_hints(namespace["subject"]).values() + ) + + +@pytest.mark.parametrize("mode", ["eager", "future", "lazy"]) +@pytest.mark.parametrize("owner", ["module", "class", "nested"]) +@pytest.mark.parametrize("operation", ["outer", "formal"]) +def test_existing_rename_api_keeps_runtime_annotation_contract( + model_project, mode, owner, operation +): + source = prefix(mode) + "target = int\n" + definition = "def func(target: target) -> target:\n return target\n" + if owner == "module": + source += definition + "subject = func\n" + outer = source.index("target = int") + elif owner == "class": + source += "class Owner:\n" + indent("target = bytes\n" + definition, " ") + source += "subject = Owner.func\n" + outer = source.index("target = bytes") + else: + source += "def owner():\n" + indent("target = bytes\n" + definition, " ") + source += " return func\nsubject = owner()\n" + outer = source.index("target = bytes") + source += "def drive():\n return subject(str)\n" + before = observe(source) + resource = module(model_project, source) + offset = ( + outer + if operation == "outer" + else source.index("target", source.index("def func(")) + ) + changes = rename.Rename(model_project, resource, offset).get_changes("renamed") + model_project.do(changes) + assert observe(resource.read()) == before + + +@pytest.mark.parametrize( + "signature,call", + [("target", "str"), ("target, /", "str"), ("*, target", "target=str")], +) +def test_lambda_formal_rename_preserves_keyword_driver(model_project, signature, call): + source = f"target = int\ncallback = lambda {signature}: target\nresult = callback({call})\n" + namespace = {} + exec(source, namespace) + before = namespace["result"] + resource = module(model_project, source) + offset = source.index("target", source.index("lambda ")) + changes = rename.Rename(model_project, resource, offset).get_changes("renamed") + model_project.do(changes) + namespace = {} + exec(resource.read(), namespace) + assert namespace["result"] is before is str + assert "target = int" in resource.read() + + +@pytest.mark.parametrize( + "expression", + [ + "[target for target in [str]]", + "{target for target in [str]}", + "{target: target for target in [str]}", + "tuple(target for target in [str])", + ], +) +def test_outer_rename_does_not_capture_comprehension_local(model_project, expression): + source = f"target = int\nvalue = {expression}\nresult = (target, value)\n" + before = {} + exec(source, before) + resource = module(model_project, source) + changes = rename.Rename(model_project, resource, 0).get_changes("renamed") + model_project.do(changes) + after = {} + exec(resource.read(), after) + assert after["result"] == before["result"] + assert "for target in" in resource.read() + + +@pytest.mark.parametrize( + "signature,call", [("target, /", "str"), ("*, target", "target=str")] +) +def test_unsupported_ordinary_formals_have_readable_distinct_bindings( + model_project, signature, call +): + source = f"target = int\ndef func({signature}):\n return target\nresult = (target, func({call}))\n" + resource = module(model_project, source) + pymodule = model_project.get_pymodule(resource) + function = pymodule["func"].get_object() + formal_offset = source.index("target", source.index("def func(")) + binder = evaluate.eval_location(pymodule, formal_offset) + assert binder is not None and binder is not pymodule["target"] + assert not binder.is_refactoring_identity_safe + assert binder.get_object() is pyobjects.get_unknown() + assert ( + evaluate.eval_location( + pymodule, source.index("target", source.index("return ")) + ) + is binder + ) + assert function.get_parameters() == {} + assert function.get_param_names() == [] + assert codeassist.get_definition_location( + model_project, source, formal_offset, resource + ) == (resource, 2) + assert finderrors.find_errors(model_project, resource) == [] + for offset in (formal_offset, source.index("target", source.index("return "))): + with pytest.raises( + exceptions.RefactoringError, match="unresolved source binding" + ): + rename.Rename(model_project, resource, offset).get_changes("renamed") + assert resource.read() == source + changes = rename.Rename(model_project, resource, 0).get_changes("outer") + model_project.do(changes) + namespace = {} + exec(resource.read(), namespace) + assert namespace["result"] == (int, str) + assert "return target" in resource.read() + + +@pytest.mark.parametrize( + "expression,expected", + [ + ("(lambda target=str: target)()", "str"), + ("(lambda *, target=str: target)()", "str"), + ("(lambda target=int, /, **kw: target)(target=str)", "int"), + ("(lambda target: target)(str)", "str"), + ("(lambda *, target: target)(target=str)", "str"), + ("(lambda target=int: target)(*[str])", None), + ("(lambda target=int: target)(**{'target': str})", None), + ("(lambda target: target)(str, target=int)", None), + ("(lambda target: target)(str, **{'target': int})", None), + ("(lambda target: target)(*[str], target=int)", None), + ("(lambda target=int: target)(missing)", None), + ], +) +def test_lambda_private_matching_does_not_use_ordinary_function_api( + model_project, expression, expected +): + source = "result = " + expression + "\n" + pymodule = model_project.get_pymodule(module(model_project, source)) + result = pymodule["result"].get_object() + if expected is None: + assert result is pyobjects.get_unknown() + else: + from rope.base import builtins + + assert result is builtins.builtins[expected].get_object() + node = next(n for n in ast.walk(pymodule.get_ast()) if isinstance(n, ast.Lambda)) + function = pymodule.get_source_binding_index().callable_for_node(node) + canonical = function.get_parameters()["target"] + assert canonical.get_object() is pyobjects.get_unknown() + model_project.pycore.module_cache.forget_all_data() + assert function.get_parameters()["target"] is canonical + + +@pytest.mark.parametrize( + "expression", + [ + "Owner().callback(item=Other()).field", + "Owner.callback(Other(), item=Owner()).field", + "Owner().callback(Other()).field", + ], +) +def test_lambda_descriptor_field_rename_keeps_receiver(model_project, expression): + source = ( + "class Other:\n field = 2\nclass Owner:\n field = 1\n callback = lambda self, item: self\nresult = " + + expression + + "\n" + ) + before = {} + exec(source, before) + resource = module(model_project, source) + changes = rename.Rename( + model_project, resource, source.index("field = 1") + ).get_changes("renamed") + model_project.do(changes) + after = {} + exec(resource.read(), after) + assert after["result"] == before["result"] + assert "field = 2" in resource.read() + + +@pytest.mark.parametrize( + "source", + [ + "target = str\ndef func(target=target):\n return target\n", + "target = str\ndef func(target: target):\n return target\n", + ], +) +def test_first_function_names_do_not_visit_header_into_local_scope( + model_project, source +): + pymodule = model_project.get_pymodule(module(model_project, source)) + scope = pymodule["func"].get_object().get_scope() + formal = scope.pyobject.get_parameters()["target"] + assert scope.get_names()["target"] is formal + assert scope.lookup("target") is formal + assert ( + evaluate.eval_location( + pymodule, source.index("target", source.index("return ")) + ) + is formal + ) + assert ( + evaluate.eval_location( + pymodule, + ( + source.index("target", source.index("target=") + 7) + if "target=" in source + else source.index("target", source.index(": target")) + ), + ) + is pymodule["target"] + ) + + +def test_introduce_parameter_keeps_structural_body_and_refuses_lambda(model_project): + source = ( + "value = 1\ndef func():\n callback = lambda item: value\n return value\n" + ) + resource = module(model_project, source) + with pytest.raises(exceptions.RefactoringError, match="inside functions"): + introduce_parameter.IntroduceParameter( + model_project, resource, source.index("item", source.index("lambda")) + ) + capture = introduce_parameter.IntroduceParameter( + model_project, resource, source.index("value", source.index("lambda")) + ) + model_project.do(capture.get_changes("captured")) + assert "def func(captured=value)" in resource.read() + assert "lambda item: captured" in resource.read() + resource.write(source) + operation = introduce_parameter.IntroduceParameter( + model_project, resource, source.index("value", source.index("return")) + ) + model_project.do(operation.get_changes("provided")) + assert "def func(provided=value)" in resource.read() + assert "lambda item: provided" in resource.read() + + +@pytest.mark.parametrize( + "signature", ["target", "target, /", "*, target", "*target", "**target"] +) +def test_source_lambda_index_keeps_all_formal_identities_across_cache_reset( + model_project, signature +): + source = f"target = int\ncallback = lambda {signature}: target\n" + pymodule = model_project.get_pymodule(module(model_project, source)) + index = pymodule.get_source_binding_index() + node = next(n for n in ast.walk(pymodule.get_ast()) if isinstance(n, ast.Lambda)) + callable_ = index.callable_for_node(node) + binder = callable_.get_parameters()["target"] + offset = source.index("target", source.index("lambda")) + assert index.binding_for_token(offset) is binder + assert evaluate.eval_location(pymodule, offset) is binder + body_offset = source.rindex("target") + assert evaluate.eval_location(pymodule, body_offset) is binder + assert binder is not pymodule["target"] + assert index.context_at(body_offset).structural_scope is pymodule.get_scope() + model_project.pycore.module_cache.forget_all_data() + assert pymodule.get_source_binding_index() is index + assert callable_.get_parameters()["target"] is binder + assert evaluate.eval_location(pymodule, body_offset) is binder + + +@pytest.mark.parametrize( + "source,offset_text", + [ + ("target = int\ndef func(target: target):\n return target\n", ": target"), + ("target = int\ndef func(target=target):\n return target\n", "=target"), + ("target = int\ncallback = lambda target=target: target\n", "=target"), + ], +) +def test_change_occurrences_uses_structure_or_canonical_expression_scope( + model_project, source, offset_text +): + resource = module(model_project, source) + offset = source.index(offset_text) + 1 + if source[offset] == " ": + offset += 1 + operation = rename.ChangeOccurrences(model_project, resource, offset) + model_project.do(operation.get_changes("changed")) + if "def func" in source: + assert "target = int" in resource.read() + assert ( + "target: changed" in resource.read() or "target=changed" in resource.read() + ) + else: + assert "changed = int" in resource.read() + assert "return target" in resource.read() or ": target" in resource.read() + + +@pytest.mark.parametrize("kind", ["global", "nonlocal"]) +def test_header_named_expression_uses_declared_owner_before_first_lookup( + model_project, kind +): + if kind == "global": + source = "target = int\ndef owner():\n global target\n def inner(value=(target := str)):\n return target\n return inner\nresult = owner()()\n" + owner_offset = source.index("target = int") + else: + source = "def enclosing():\n target = int\n def owner():\n nonlocal target\n def inner(value=(target := str)):\n return target\n return inner\n return owner()\nresult = enclosing()()\n" + owner_offset = source.index("target = int") + resource = module(model_project, source) + pymodule = model_project.get_pymodule(resource) + binder = evaluate.eval_location(pymodule, owner_offset) + walrus = source.index("target :=") + assert evaluate.eval_location(pymodule, walrus) is binder + model_project.do( + rename.Rename(model_project, resource, owner_offset).get_changes("renamed") + ) + namespace = {} + exec(resource.read(), namespace) + assert namespace["result"] is str + assert "target" not in resource.read() + + +@pytest.mark.skipif( + sys.version_info < (3, 12), reason="type aliases require Python 3.12" +) +def test_alias_own_parameter_is_distinct_and_rename_keeps_runtime_value(model_project): + source = "T = int\ntype Alias[T] = tuple[T, T]\nresult = Alias.__value__\n" + resource = module(model_project, source) + pymodule = model_project.get_pymodule(resource) + own_offset = source.index("[T]") + 1 + binder = evaluate.eval_location(pymodule, own_offset) + assert binder is not pymodule["T"] + assert evaluate.eval_location(pymodule, source.index("[T, T]") + 1) is binder + model_project.do( + rename.Rename(model_project, resource, own_offset).get_changes("V") + ) + namespace = {} + exec(resource.read(), namespace) + alias = namespace["Alias"] + assert namespace["T"] is int + assert alias.__value__.__args__ == ( + alias.__type_params__[0], + alias.__type_params__[0], + ) + + +@pytest.mark.skipif( + sys.version_info < (3, 12), reason="type parameters require Python 3.12" +) +def test_generic_class_collision_uses_opaque_source_and_refuses_atomically( + model_project, +): + source = "T = int\nclass Owner[T]:\n T = str\n def method(self, value: T):\n return value\n" + resource = module(model_project, source) + for offset in (source.index("[T]") + 1, source.index("T = str")): + with pytest.raises(exceptions.RefactoringError, match="unresolved"): + rename.Rename(model_project, resource, offset).get_changes("V") + assert resource.read() == source + + +@pytest.mark.parametrize("signature", ["target, /", "*, target"]) +def test_introduce_parameter_refuses_unsupported_ordinary_signature_atomically( + model_project, signature +): + source = f"value = int\ndef func({signature}):\n return value\n" + resource = module(model_project, source) + operation = introduce_parameter.IntroduceParameter( + model_project, resource, source.index("value", source.index("return")) + ) + with pytest.raises( + exceptions.RefactoringError, match="unsupported parameter categories" + ): + operation.get_changes("provided") + assert resource.read() == source + + +@pytest.mark.parametrize( + "source", + [ + "target = int\ncallback = lambda target: target\n", + "target = int\nresult = [target for target in [str]]\n", + ], +) +def test_completion_follows_canonical_expression_environment(model_project, source): + resource = module(model_project, source) + offset = ( + source.index("target", source.index(": target") + 2) + 3 + if "lambda" in source + else source.index("target", source.index("result")) + 3 + ) + proposals = codeassist.code_assist( + model_project, source, offset, resource=resource, later_locals=False + ) + assert "target" in {proposal.name for proposal in proposals} + assert finderrors.find_errors(model_project, resource) == [] + + +def test_old_offset_scope_remains_structural_while_index_resolves_comp_identity( + model_project, +): + source = "target = int\nresult = [target for target in [str]]\n" + pymodule = model_project.get_pymodule(module(model_project, source)) + scope = pymodule.get_scope() + offset = source.index("target", source.index("result")) + assert scope.get_inner_scope_for_offset(offset) is scope.get_scopes()[0] + assert scope.get_structural_scope_for_offset(offset) is scope.get_scopes()[0] + canonical = pymodule.get_source_binding_index().context_at(offset).environment + assert canonical is not scope.get_scopes()[0] + assert evaluate.eval_location(pymodule, offset) is canonical.lookup("target") + assert evaluate.eval_location(pymodule, offset) is not pymodule["target"] + + +@pytest.mark.parametrize("class_first", [False, True]) +@pytest.mark.parametrize("class_return", ["self", "item", "static"]) +@pytest.mark.parametrize("access", ["direct", "saved", "class", "class-saved"]) +@pytest.mark.parametrize("renamed_owner", ["Owner", "Other"]) +def test_lambda_storage_view_selects_instance_and_class_values_independently( + model_project, class_first, class_return, access, renamed_owner +): + initializer = ( + " def __init__(self):\n" + " self.field = 1\n" + " self.callback = lambda item: item\n" + ) + class_value = ( + " callback = staticmethod(lambda item: item)\n" + if class_return == "static" + else f" callback = lambda self, item: {class_return}\n" + ) + source = "class Owner:\n" + ( + class_value + initializer if class_first else initializer + class_value + ) + source += "class Other:\n def __init__(self):\n self.field = 2\n" + if access == "direct": + source += "result = Owner().callback(Other())\n" + elif access == "saved": + source += "saved = Owner().callback\nresult = saved(Other())\n" + elif access == "class": + source += ( + "result = Owner.callback(Other())\n" + if class_return == "static" + else "result = Owner.callback(Owner(), Other())\n" + ) + else: + source += "saved = Owner.callback\n" + source += ( + "result = saved(Other())\n" + if class_return == "static" + else "result = saved(Owner(), Other())\n" + ) + source += "observed = result.field\n" + before = {} + exec(compile(source, "storage_fixture.py", "exec"), before) + expected_owner = ( + "Owner" + if access in {"class", "class-saved"} and class_return == "self" + else "Other" + ) + assert before["observed"] == (1 if expected_owner == "Owner" else 2) + resource = module(model_project, source) + pymodule = model_project.get_pymodule(resource) + # Reading either namespace first must not change the other value view. + first = "Owner.callback" if class_first else "Owner().callback" + second = "Owner().callback" if class_first else "Owner.callback" + first_view = evaluate.eval_str(pymodule.get_scope(), first) + first_function = first_view.get_object() + first_parameters = first_function.get_parameters() + second_view = evaluate.eval_str(pymodule.get_scope(), second) + second_function = second_view.get_object() + assert first_function.source_function is not second_function.source_function + assert first_view.get_object() is first_function + assert first_function.get_parameters() == first_parameters + assert ( + evaluate.eval_str(pymodule.get_scope(), first).get_object().source_function + is first_function.source_function + ) + # Repeat in the opposite order on the same module and keep both views alive. + reversed_second = evaluate.eval_str(pymodule.get_scope(), second) + reversed_first = evaluate.eval_str(pymodule.get_scope(), first) + assert ( + reversed_second.get_object().source_function is second_function.source_function + ) + assert reversed_first.get_object().source_function is first_function.source_function + assert second_view.get_object() is second_function + field_value = "1" if renamed_owner == "Owner" else "2" + offset = source.index("self.field = " + field_value) + len("self.") + model_project.do( + rename.Rename(model_project, resource, offset).get_changes("changed") + ) + after = {} + exec(compile(resource.read(), "storage_fixture.py", "exec"), after) + assert after["observed"] == before["observed"] + assert ("observed = result.changed" in resource.read()) == ( + renamed_owner == expected_owner + ) + + +def runtime_observed(source): + namespace = {} + exec(compile(source, "source_contract.py", "exec", dont_inherit=True), namespace) + return namespace["observed"] + + +@pytest.mark.parametrize( + "factory", + [ + "lambda supplied: (lambda *, option=None: supplied)", + "lambda supplied: (lambda option=supplied: option)", + ], +) +@pytest.mark.parametrize("renamed", ["first_field", "second_field"]) +def test_lambda_creation_values_are_isolated_between_calls( + model_project, factory, renamed +): + source = ( + "class First:\n first_field = 11\n" + "class Second:\n second_field = 22\n" + f"factory = {factory}\n" + "first = factory(First())\nsecond = factory(Second())\n" + "observed = (first().first_field, second().second_field)\n" + ) + resource = module(model_project, source) + pymodule = model_project.get_pymodule(resource) + first = evaluate.eval_str(pymodule.get_scope(), "first") + second = evaluate.eval_str(pymodule.get_scope(), "second") + assert first.get_object().source_function is second.get_object().source_function + assert first.get_object() is not second.get_object() + for expression, expected in [("second()", "Second"), ("first()", "First")]: + assert evaluate.eval_str( + pymodule.get_scope(), expression + ).get_object().get_type() is (pymodule[expected].get_object()) + inner = first.get_object().source_function + assert inner.get_parameters()["option"].get_object() is pyobjects.get_unknown() + model_project.pycore.module_cache.forget_all_data() + assert inner.get_parameters()["option"].get_object() is pyobjects.get_unknown() + before = runtime_observed(source) + model_project.do( + rename.Rename( + model_project, resource, source.index(renamed + " =") + ).get_changes("changed") + ) + assert runtime_observed(resource.read()) == before == (11, 22) + other = "second_field" if renamed == "first_field" else "first_field" + assert f".{other}" in resource.read() + + +@pytest.mark.parametrize( + "expression", + [ + "[value for value in (item,)][0]", + "[second for first, second in ((None, item),)][0]", + "[second for first in (item,) for second in (first,)][0]", + "{key: value for key, value in ((0, item),)}[0]", + ], +) +@pytest.mark.parametrize("renamed_owner", ["Payload", "Other"]) +def test_comprehension_call_overlay_drives_field_rename( + model_project, expression, renamed_owner +): + source = ( + "class Payload:\n field = 42\n" + "class Other:\n field = 99\n" + f"callback = lambda item: {expression}\n" + "result = callback(Payload())\nobserved = result.field\n" + ) + resource = module(model_project, source) + pymodule = model_project.get_pymodule(resource) + callback = pymodule["callback"].get_object() + parameter = callback.get_parameters()["item"] + assert ( + pymodule["result"].get_object().get_type() is pymodule["Payload"].get_object() + ) + assert parameter.get_object() is pyobjects.get_unknown() + model_project.pycore.module_cache.forget_all_data() + assert callback.get_parameters()["item"] is parameter + assert parameter.get_object() is pyobjects.get_unknown() + before = runtime_observed(source) + declaration = "field = 42" if renamed_owner == "Payload" else "field = 99" + model_project.do( + rename.Rename(model_project, resource, source.index(declaration)).get_changes( + "changed" + ) + ) + assert runtime_observed(resource.read()) == before == 42 + assert ("observed = result.changed" in resource.read()) == ( + renamed_owner == "Payload" + ) + + +@pytest.mark.parametrize( + "descriptor,body", + [ + ("lambda self, item: self", "Owner"), + ("staticmethod(lambda item: item)", "Other"), + ("classmethod(lambda cls, item: item)", "Other"), + ], +) +def test_saved_lambda_keyword_rename_keeps_descriptor_and_source_identity( + model_project, descriptor, body +): + source = ( + "class Owner:\n field = 1\n" + f" callback = {descriptor}\n" + "class Other:\n field = 2\n" + "saved = Owner().callback\nresult = saved(item=Other())\n" + "observed = result.field\n" + ) + resource = module(model_project, source) + pymodule = model_project.get_pymodule(resource) + offset = source.index("item=Other()") + function = pymodule["saved"].get_object() + parameter = function.get_parameters()["item"] + assert evaluate.eval_location(pymodule, offset) is parameter + assert codeassist.get_definition_location( + model_project, source, offset, resource + ) == (resource, 3) + assert pymodule["result"].get_object().get_type() is pymodule[body].get_object() + assert parameter.get_object() is pyobjects.get_unknown() + before = runtime_observed(source) + model_project.do( + rename.Rename(model_project, resource, offset).get_changes("changed") + ) + assert "saved(changed=Other())" in resource.read() + assert runtime_observed(resource.read()) == before + + +@pytest.mark.parametrize("kind", ["closure", "saved-bound"]) +@pytest.mark.parametrize("renamed_owner", ["Owner", "Other"]) +def test_saved_bound_lambda_is_not_bound_again(model_project, kind, renamed_owner): + source = ( + "class Owner:\n field = 1\n" + " callback = lambda self, item: " + + ("(lambda: self)\n" if kind == "closure" else "self\n") + + "class Other:\n field = 2\n" + ) + source += ( + "saved = Owner().callback\nresult = saved(Other())()\n" + if kind == "closure" + else "class Saved:\n callback = Owner().callback\nresult = Saved().callback(Other())\n" + ) + source += "observed = result.field\n" + resource = module(model_project, source) + pymodule = model_project.get_pymodule(resource) + assert pymodule["result"].get_object().get_type() is pymodule["Owner"].get_object() + before = runtime_observed(source) + offset = source.index("field = " + ("1" if renamed_owner == "Owner" else "2")) + model_project.do( + rename.Rename(model_project, resource, offset).get_changes("changed") + ) + assert runtime_observed(resource.read()) == before == 1 + assert ("observed = result.changed" in resource.read()) == ( + renamed_owner == "Owner" + ) + + +@pytest.mark.parametrize("owner", ["module", "class", "global", "nonlocal-parameter"]) +def test_header_created_name_is_queryable_before_and_after_cache_reset( + model_project, owner +): + if owner == "module": + source = "def inner(value=(target := str)):\n return target\nobserved = inner() is str\n" + elif owner == "class": + source = "class Owner:\n def inner(value=(target := str)):\n return value\nobserved = Owner.target is str and Owner.inner() is str\n" + elif owner == "global": + source = "def owner():\n global target\n def inner(value=(target := str)):\n return target\n return inner\nobserved = owner()() is str and target is str\n" + else: + source = "def enclosing(target):\n def owner():\n nonlocal target\n def inner(value=(target := str)):\n return target\n return inner\n return owner()\nobserved = enclosing(int)() is str\n" + resource = module(model_project, source) + pymodule = model_project.get_pymodule(resource) + offset = source.index("target :=") + binder = evaluate.eval_location(pymodule, offset) + assert binder is not None + location = codeassist.get_definition_location( + model_project, source, offset, resource + ) + assert location[0] == resource + model_project.pycore.module_cache.forget_all_data() + assert evaluate.eval_location(pymodule, offset) is binder + assert ( + codeassist.get_definition_location(model_project, source, offset, resource) + == location + ) + before = runtime_observed(source) + model_project.do( + rename.Rename(model_project, resource, offset).get_changes("changed") + ) + assert runtime_observed(resource.read()) == before is True + assert "target" not in resource.read() + + +@pytest.mark.parametrize("selection", ["module", "lambda", "comprehension"]) +def test_nested_lambda_comprehension_shadowing_rename_preserves_each_binding( + model_project, selection +): + source = ( + "target = int\n" + "result = (lambda *, target: [target for target in target])(target=(str, bytes))\n" + "observed = (target is int, result == [str, bytes])\n" + ) + resource = module(model_project, source) + markers = { + "module": "target = int", + "lambda": "target:", + "comprehension": "for target", + } + offset = source.index(markers[selection]) + ( + 4 if selection == "comprehension" else 0 + ) + before = runtime_observed(source) + model_project.do( + rename.Rename(model_project, resource, offset).get_changes("changed") + ) + assert runtime_observed(resource.read()) == before == (True, True) + if selection == "lambda": + assert "[target for target in changed]" in resource.read() + assert ")(changed=" in resource.read() + elif selection == "comprehension": + assert "[changed for changed in target]" in resource.read() + else: + assert "lambda *, target:" in resource.read() + + +@pytest.mark.parametrize( + "expression", + ["lambda: (target := str)", "lambda: [(target := value) for value in [str]][0]"], +) +@pytest.mark.parametrize("operation", ["local", "outer", "occurrences"]) +def test_lambda_walrus_source_binding_does_not_escape_to_module( + model_project, expression, operation +): + source = f"target = int\ncallback = {expression}\nobserved = (target is int, callback() is str)\n" + resource = module(model_project, source) + pymodule = model_project.get_pymodule(resource) + local_offset = source.index("target :=") + local = evaluate.eval_location(pymodule, local_offset) + outer = pymodule["target"] + assert local is not outer + assert local.get_definition_location() == (pymodule, 2) + before = runtime_observed(source) + if operation == "occurrences": + changes = rename.ChangeOccurrences( + model_project, resource, local_offset + ).get_changes("changed") + else: + offset = local_offset if operation == "local" else 0 + changes = rename.Rename(model_project, resource, offset).get_changes("changed") + model_project.do(changes) + assert runtime_observed(resource.read()) == before == (True, True) + assert ("target = int" in resource.read()) == (operation != "outer") + assert ("target :=" in resource.read()) == (operation == "outer") + + +@pytest.mark.parametrize("mode", ["eager", "future", "lazy"]) +@pytest.mark.parametrize( + "class_body,selected", + [ + ("target = bytes\n", "class"), + ("target: object = bytes\n", "class"), + ("target: object\n", "module"), + ("target = bytes\ndel target\n", "module"), + ("pass\n", "module"), + ], +) +def test_annotation_query_completion_and_rename_follow_class_namespace( + model_project, mode, class_body, selected +): + source = prefix(mode) + "target = int\nclass Owner:\n" + source += indent( + class_body + "def func(value: target) -> target:\n return value\n", " " + ) + source += "subject = Owner.func\ndef drive():\n return subject(str)\n" + resource = module(model_project, source) + pymodule = model_project.get_pymodule(resource) + offset = source.index("target", source.index("value: ")) + # Future-string get_type_hints uses the module namespace by its default contract. + expected = pymodule["target"] + if mode != "future" and selected == "class": + expected = pymodule["Owner"].get_object()["target"] + context = pymodule.get_source_binding_index().context_at(offset) + assert evaluate.eval_location(pymodule, offset) is expected + assert context.structural_scope.get_kind() == "Function" + assert context.environment.get_kind() == "Annotation" + assert ( + context.annotation_policy.mode + == {"eager": "eager", "future": "future-string", "lazy": "lazy"}[mode] + ) + assert codeassist.get_definition_location( + model_project, source, offset, resource + ) == (resource, expected.get_definition_location()[1]) + proposals = codeassist.code_assist( + model_project, source, offset + 3, resource=resource, later_locals=False + ) + assert "target" in {proposal.name for proposal in proposals} + model_project.pycore.module_cache.forget_all_data() + assert evaluate.eval_location(pymodule, offset) is expected + before = observe(source) + model_project.do( + rename.Rename(model_project, resource, offset).get_changes("changed") + ) + assert observe(resource.read()) == before + + +@pytest.mark.skipif( + sys.version_info < (3, 12), reason="type aliases require Python 3.12" +) +@pytest.mark.parametrize("owner", ["module", "class", "class-fallback", "nested"]) +def test_type_alias_outer_reference_rename_preserves_lazy_value(model_project, owner): + if owner == "module": + source = "target = int\ntype Alias = tuple[target, target]\nsubject = Alias\n" + elif owner == "class": + source = "target = int\nclass Owner:\n target = bytes\n type Alias = tuple[target, target]\nsubject = Owner.Alias\n" + elif owner == "class-fallback": + source = "target = int\nclass Owner:\n type Alias = tuple[target, target]\nsubject = Owner.Alias\n" + else: + source = "target = int\ndef factory():\n target = bytes\n type Alias = tuple[target, target]\n return Alias\nsubject = factory()\n" + source += "observed = tuple(item.__name__ for item in subject.__value__.__args__)\n" + resource = module(model_project, source) + pymodule = model_project.get_pymodule(resource) + offset = source.index("target", source.index("tuple[")) + binding = evaluate.eval_location(pymodule, offset) + assert binding is not None and getattr( + binding, "is_refactoring_identity_safe", True + ) + assert ( + codeassist.get_definition_location(model_project, source, offset, resource)[0] + == resource + ) + before = runtime_observed(source) + model_project.do( + rename.Rename(model_project, resource, offset).get_changes("changed") + ) + assert runtime_observed(resource.read()) == before + + +@pytest.mark.skipif( + sys.version_info < (3, 12), reason="generic classes require Python 3.12" +) +@pytest.mark.parametrize("query_first", [False, True]) +def test_generic_alias_queries_remain_readable_and_refusal_is_project_atomic( + model_project, query_first +): + source = "target = int\nclass Risk[target]:\n target = str\n type Alias = tuple[target, target]\n" + resource = module(model_project, source) + client = model_project.root.create_file("client.py") + client.write("from subject import target\nobserved = target is int\n") + pymodule = model_project.get_pymodule(resource) + offset = source.index("target", source.index("tuple[")) + before = {file: file.read() for file in (resource, client)} + if query_first: + assert codeassist.get_definition_location( + model_project, source, offset, resource + ) == (resource, 4) + assert finderrors.find_errors(model_project, resource) == [] + proposals = codeassist.code_assist( + model_project, source, offset + 3, resource=resource + ) + assert "target" in {proposal.name for proposal in proposals} + opaque = evaluate.eval_location(pymodule, offset) + assert opaque.get_object() is pyobjects.get_unknown() + assert not opaque.is_refactoring_identity_safe + for selected in [ + offset, + source.index("[target]") + 1, + source.index("target = str"), + ]: + with pytest.raises(exceptions.RefactoringError, match="unresolved"): + rename.Rename(model_project, resource, selected).get_changes("changed") + assert {file: file.read() for file in before} == before + # The same module's unrelated exported source name remains independently editable. + model_project.do(rename.Rename(model_project, resource, 0).get_changes("changed")) + assert "from subject import changed" in client.read() + assert "type Alias = tuple[target, target]" in resource.read() + + +def test_change_occurrences_checks_only_the_selected_structural_region(model_project): + source = ( + "target = int\nenabled = True\nclass Risk:\n if enabled:\n target = str\n" + " def method(value: target):\n return value\n" + "def allowed():\n return target\nobserved = allowed() is int\n" + ) + resource = module(model_project, source) + before = runtime_observed(source) + offset = source.index("target", source.index("return ")) + model_project.do( + rename.ChangeOccurrences(model_project, resource, offset).get_changes("int") + ) + assert runtime_observed(resource.read()) == before is True + assert "return int" in resource.read() + assert "value: target" in resource.read() + resource.write(source) + with pytest.raises(exceptions.RefactoringError, match="unresolved"): + rename.ChangeOccurrences(model_project, resource, 0).get_changes("changed") + assert resource.read() == source + + +@pytest.mark.parametrize( + "source,entry_marker,body_marker", + [ + ( + "lex_outer = int\ncallback = lambda lex_item=lex_outer: lex_item\n", + "=lex_outer", + ": lex_item", + ), + ( + "lex_outer = [int]\nvalues = [lex_item for lex_item in lex_outer]\n", + "in lex_outer", + "[lex_item", + ), + ], +) +def test_completion_separates_expression_entry_from_local_body( + model_project, source, entry_marker, body_marker +): + resource = module(model_project, source) + for marker, body in [(entry_marker, False), (body_marker, True)]: + start = source.index("lex_", source.index(marker)) + proposals = { + item.name: item.scope + for item in codeassist.code_assist( + model_project, source, start + 4, resource=resource, later_locals=False + ) + } + assert proposals["lex_outer"] == "global" + if body: + assert proposals["lex_item"] == "local" + else: + assert "lex_item" not in proposals + + +@pytest.mark.parametrize( + "source,expected", + [ + ("callback = lambda item: item\n", []), + ("values = [item for item in [int]]\n", []), + ( + "callback = lambda item: (item, absent)\n", + [(1, "Unresolved variable absent")], + ), + ( + "values = [item for item in [int] if absent]\n", + [(1, "Unresolved variable absent")], + ), + ("result = later\nlater = int\n", [(1, "Defined later later")]), + ], +) +def test_find_errors_preserves_real_diagnostics_without_expression_false_positives( + model_project, source, expected +): + resource = module(model_project, source) + assert [ + (error.lineno, error.error) + for error in finderrors.find_errors(model_project, resource) + ] == expected + + +def test_introduce_parameter_rewrites_outer_capture_in_comprehension(model_project): + source = "value = int\ndef subject():\n return [value for item in [str]]\nobserved = subject() == [int]\n" + resource = module(model_project, source) + before = runtime_observed(source) + operation = introduce_parameter.IntroduceParameter( + model_project, resource, source.index("value", source.index("return ")) + ) + model_project.do(operation.get_changes("provided")) + assert runtime_observed(resource.read()) == before is True + assert "def subject(provided=value)" in resource.read() + assert "[provided for item" in resource.read() + assert ( + runtime_observed(resource.read() + "observed = subject(str) == [str]\n") is True + ) + resource.write(source) + with pytest.raises(exceptions.RefactoringError): + introduce_parameter.IntroduceParameter( + model_project, resource, source.index("item in") + ) + assert resource.read() == source + + +@pytest.mark.parametrize("future", [False, True]) +def test_introduce_parameter_refuses_annotation_environment_atomically( + model_project, future +): + source = "from __future__ import annotations\n" if future else "" + source += "value = int\ndef subject(item: value):\n return item\n" + resource = module(model_project, source) + offset = source.index("value", source.index("item:")) + with pytest.raises(exceptions.RefactoringError, match="inside functions"): + introduce_parameter.IntroduceParameter(model_project, resource, offset) + assert resource.read() == source + + +@pytest.mark.parametrize("renamed_owner", ["Owner", "Other"]) +def test_callable_instance_lambda_drives_field_rename(model_project, renamed_owner): + source = "class Owner:\n field = 1\n __call__ = lambda self, item: self\nclass Other:\n field = 2\nresult = Owner()(Other())\nobserved = result.field\n" + resource = module(model_project, source) + before = runtime_observed(source) + offset = source.index("field = " + ("1" if renamed_owner == "Owner" else "2")) + model_project.do( + rename.Rename(model_project, resource, offset).get_changes("changed") + ) + assert runtime_observed(resource.read()) == before == 1 + assert ("observed = result.changed" in resource.read()) == ( + renamed_owner == "Owner" + ) + + +def test_callable_instance_lambda_keyword_has_canonical_formal_identity(model_project): + source = "class Owner:\n __call__ = lambda self, item: item\nresult = Owner()(item=str)\nobserved = result is str\n" + resource = module(model_project, source) + offset = source.index("item=str") + pymodule = model_project.get_pymodule(resource) + binder = ( + pymodule["Owner"].get_object()["__call__"].get_object().get_parameters()["item"] + ) + assert evaluate.eval_location(pymodule, offset) is binder + model_project.do( + rename.Rename(model_project, resource, offset).get_changes("changed") + ) + assert runtime_observed(resource.read()) is True + assert "Owner()(changed=str)" in resource.read() + + +@pytest.mark.skipif( + sys.version_info < (3, 12), reason="type aliases require Python 3.12" +) +def test_class_type_alias_captures_generic_owner_without_capturing_class_values( + model_project, +): + source = "T = int\nclass Owner[T]:\n unrelated = bytes\n type Alias = tuple[T, T]\nobserved = Owner.Alias.__value__.__args__ == (Owner.__type_params__[0],) * 2\n" + resource = module(model_project, source) + pymodule = model_project.get_pymodule(resource) + offset = source.index("T", source.index("type Alias")) + binder = evaluate.eval_location(pymodule, offset) + own = evaluate.eval_location(pymodule, source.index("[T]") + 1) + assert binder is own and binder is not pymodule["T"] + proposals = codeassist.code_assist( + model_project, source, offset + 1, resource=resource + ) + assert "T" in {proposal.name for proposal in proposals} + before = runtime_observed(source) + model_project.do( + rename.Rename(model_project, resource, source.index("[T]") + 1).get_changes("V") + ) + assert runtime_observed(resource.read()) == before is True + assert "T = int" in resource.read() + + +@pytest.mark.skipif( + sys.version_info < (3, 12), reason="generic functions require Python 3.12" +) +@pytest.mark.parametrize("expression", ["T", "(lambda: T)()"]) +def test_generic_annotation_query_remains_opaque_through_nested_lambda( + model_project, expression +): + source = f"T = int\ndef subject[T](value: {expression}):\n return value\nobserved = subject(str) is str\n" + resource = module(model_project, source) + pymodule = model_project.get_pymodule(resource) + offset = source.index("T", source.index("value: ")) + binder = evaluate.eval_location(pymodule, offset) + assert not binder.is_refactoring_identity_safe + assert binder.get_object() is pyobjects.get_unknown() + assert ( + codeassist.get_definition_location(model_project, source, offset, resource)[0] + == resource + ) + proposals = codeassist.code_assist( + model_project, source, offset + 1, resource=resource + ) + assert "T" in {proposal.name for proposal in proposals} + assert runtime_observed(source) is True + for selected in [offset, source.index("[T]") + 1]: + with pytest.raises(exceptions.RefactoringError, match="unresolved"): + rename.Rename(model_project, resource, selected).get_changes("V") + assert resource.read() == source + + +@pytest.mark.skipif( + sys.version_info >= (3, 14), reason="annotation walrus requires eager annotations" +) +@pytest.mark.parametrize("owner", ["module", "class"]) +def test_eager_annotation_walrus_installs_its_declaration_binding(model_project, owner): + if owner == "module": + source = "def subject(value: (target := str)):\n return value\nobserved = subject(int) is int and target is str\n" + else: + source = "target = int\nclass Owner:\n def subject(value: (target := str)):\n return value\nobserved = Owner.subject(int) is int and Owner.target is str and target is int\n" + resource = module(model_project, source) + pymodule = model_project.get_pymodule(resource) + offset = source.index("target :=") + binding = evaluate.eval_location(pymodule, offset) + if owner == "class": + assert binding is pymodule["Owner"].get_object()["target"] + assert binding is not pymodule["target"] + else: + assert binding is pymodule["target"] + assert binding.get_definition_location()[0] is pymodule + before = runtime_observed(source) + model_project.do( + rename.Rename(model_project, resource, offset).get_changes("changed") + ) + assert runtime_observed(resource.read()) == before is True + + +@pytest.mark.parametrize("selection", ["outer", "class"]) +def test_class_lambda_default_walrus_does_not_change_outer_source_binding( + model_project, selection +): + source = "target = int\nclass Owner:\n callback = staticmethod(lambda value=(target := str): value)\nobserved = Owner.callback() is str and Owner.target is str and target is int\n" + resource = module(model_project, source) + pymodule = model_project.get_pymodule(resource) + offset = source.index("target :=") if selection == "class" else 0 + selected = evaluate.eval_location(pymodule, offset) + class_binding = evaluate.eval_location(pymodule, source.index("target :=")) + assert class_binding is not pymodule["target"] + assert selected is (class_binding if selection == "class" else pymodule["target"]) + before = runtime_observed(source) + model_project.do( + rename.Rename(model_project, resource, offset).get_changes("changed") + ) + assert runtime_observed(resource.read()) == before is True + + +@pytest.mark.parametrize("mode", ["eager", "future", "lazy"]) +@pytest.mark.parametrize("declaration", ["import", "function", "class", "late"]) +def test_class_annotation_namespace_obeys_declaration_order_and_name_shadowing( + model_project, mode, declaration +): + source = prefix(mode) + "import typing\ntarget = int\nclass Owner:\n" + if declaration == "import": + body = "from builtins import str as target\n" + expression = "typing.get_type_hints(Owner.subject)['value'] is str" + elif declaration == "function": + body = "def target():\n return bytes\n" + expression = "typing.get_type_hints(Owner.subject)['value']() is bytes" + elif declaration == "class": + body = "class target:\n marker = 7\n" + expression = "typing.get_type_hints(Owner.subject)['value'].marker == 7" + else: + body = "" + expression = ( + "typing.get_type_hints(Owner.subject)['value'] is bytes" + if mode == "lazy" + else "typing.get_type_hints(Owner.subject)['value'] is int" + ) + body += "def subject(value: target):\n return value\n" + if declaration == "late": + body += "target = bytes\n" + if mode == "future": + expression = "typing.get_type_hints(Owner.subject)['value'] is int" + source += indent(body, " ") + source += f"observed = (Owner.subject(str) is str, {expression}, target is int)\n" + resource = module(model_project, source) + pymodule = model_project.get_pymodule(resource) + offset = source.index("target", source.index("value: ")) + expected = pymodule["target"] + class_resolver = mode != "future" and (declaration != "late" or mode == "lazy") + if class_resolver: + expected = pymodule["Owner"].get_object()["target"] + assert evaluate.eval_location(pymodule, offset) is expected + proposals = codeassist.code_assist( + model_project, source, offset + 3, resource=resource, later_locals=False + ) + assert "target" in {proposal.name for proposal in proposals} + before = runtime_observed(source) + # Renaming the outer source must leave independently declared class names intact. + model_project.do( + rename.Rename( + model_project, resource, source.index("target = int") + ).get_changes("changed") + ) + assert runtime_observed(resource.read()) == before == (True, True, True) + assert ("value: target" in resource.read()) == class_resolver + + +@pytest.mark.parametrize( + "expression", ["lambda item: item", "[item for item in [str]]"] +) +def test_change_occurrences_limits_edits_to_expression_binding( + model_project, expression +): + if "lambda" in expression: + source = f"item = int\ncallback = {expression}\nobserved = (callback(str) is str, item is int)\n" + offset = source.index("item:") + else: + source = f"item = int\nresult = {expression}\nobserved = (result == [str], item is int)\n" + offset = source.index("item", source.index("for ")) + resource = module(model_project, source) + before = runtime_observed(source) + model_project.do( + rename.ChangeOccurrences(model_project, resource, offset).get_changes("changed") + ) + assert runtime_observed(resource.read()) == before == (True, True) + assert "item = int" in resource.read() + assert "item is int" in resource.read() + + +@pytest.mark.parametrize( + "expression", + [ + "[item for item in owners][0]", + "{0: item for item in owners}[0]", + "{item for item in owners}.pop()", + "tuple(item for item in owners)[0]", + "[right for left, right in ((other, owner),)][0]", + "(lambda ignored: owner)(other)", + ], +) +def test_independently_parsed_expression_queries_keep_legacy_value_scope( + model_project, expression +): + source = ( + "class Owner:\n field = 1\nclass Other:\n field = 2\n" + "owner = Owner()\nother = Other()\nitem = other\nowners = [owner]\n" + ) + resource = module(model_project, source) + pymodule = model_project.get_pymodule(resource) + namespace = {} + exec(compile(source, "query_source.py", "exec", dont_inherit=True), namespace) + result = eval(expression, namespace) + assert type(result) is namespace["Owner"] and result.field == 1 + item_offset = source.index("item = other") + source_binding = evaluate.eval_location(pymodule, item_offset) + assert source_binding.get_object().get_type() is pymodule["Other"].get_object() + for cold in (False, True): + if cold: + model_project.pycore.module_cache.forget_all_data() + inferred = evaluate.eval_str(pymodule.get_scope(), expression) + _, inferred_pair = evaluate.eval_str2(pymodule.get_scope(), expression) + assert inferred.get_object().get_type() is pymodule["Owner"].get_object() + assert inferred_pair.get_object().get_type() is pymodule["Owner"].get_object() + assert evaluate.eval_location(pymodule, item_offset) is source_binding + assert source_binding.get_object().get_type() is pymodule["Other"].get_object() + assert resource.read() == source + assert namespace["item"] is namespace["other"] + + +def test_public_binding_tree_locates_local_region_and_preserves_rename_driver( + model_project, +): + source = ( + "item = int\nouter_only = bytes\ncallback = (\n" + " lambda item: [\n item\n for item in item\n ]\n)\n" + "observed = (callback([str, bytes]) == [str, bytes], item is int)\n" + ) + resource = module(model_project, source) + pymodule = model_project.get_pymodule(resource) + + def locate(module_, code, name): + # A client traverses binding environments separately from structural scopes. + (lambda_scope,) = module_.get_scope().get_binding_scopes() + (comp_scope,) = lambda_scope.get_binding_scopes() + assert lambda_scope.get_kind() == "Lambda" + assert comp_scope.get_kind() == "Comprehension" + assert "outer_only" not in lambda_scope.get_names() + assert "outer_only" not in comp_scope.get_names() + formal = lambda_scope["item"] + local = comp_scope[name] + assert local is not formal and formal is not module_["item"] + lambda_start, lambda_end = lambda_scope.get_region() + comp_start, comp_end = comp_scope.get_region() + assert lambda_start < comp_start < comp_end <= lambda_end + assert code[lambda_start:lambda_end].startswith("lambda item:") + assert code[comp_start:comp_end].startswith("[") + declaration = code.index(name, code.index("for ")) + body = code.index(name, comp_start) + iterable = code.index("item", code.index("in item")) + assert evaluate.eval_location(module_, declaration) is local + assert evaluate.eval_location(module_, body) is local + assert evaluate.eval_location(module_, iterable) is formal + assert local.get_definition_location() == (module_, 6) + assert codeassist.get_definition_location( + model_project, code, body, resource + ) == (resource, 6) + return lambda_scope, comp_scope, local, declaration + + lambda_scope, comp_scope, binding, declaration = locate(pymodule, source, "item") + model_project.pycore.module_cache.forget_all_data() + cold_lambda, cold_comp, cold_binding, _ = locate(pymodule, source, "item") + assert cold_lambda is lambda_scope and cold_comp is comp_scope + assert cold_binding is binding + before = runtime_observed(source) + model_project.do( + rename.Rename(model_project, resource, declaration).get_changes("changed") + ) + after = resource.read() + assert runtime_observed(after) == before == (True, True) + assert "for changed in item" in after + assert "lambda item:" in after and "item = int" in after + updated = model_project.get_pymodule(resource) + _, _, renamed_binding, _ = locate(updated, after, "changed") + assert renamed_binding is not updated["item"] + + +def test_calltip_reads_simple_source_lambda_without_changing_its_binding(model_project): + source = "callback = lambda item: item\nobserved = callback(str) is str\n" + resource = module(model_project, source) + offset = source.index("callback(str)") + 1 + assert ( + codeassist.get_calltip(model_project, source, offset, resource=resource) + == "lambda(item)" + ) + assert runtime_observed(source) is True + assert resource.read() == source + + +@pytest.mark.parametrize("renamed_owner", ["Owner", "Other"]) +def test_lambda_walrus_local_value_keeps_outer_value_and_field_rename_independent( + model_project, renamed_owner +): + source = ( + "class Owner:\n field = 1\nclass Other:\n field = 2\n" + "local = Other()\nresult = (lambda: ((local := Owner()), local)[1])()\n" + "observed = (result.field, local.field)\n" + ) + resource = module(model_project, source) + pymodule = model_project.get_pymodule(resource) + local = evaluate.eval_location(pymodule, source.index("local :=")) + assert local is not pymodule["local"] + assert local.get_object().get_type() is pymodule["Owner"].get_object() + assert pymodule["local"].get_object().get_type() is pymodule["Other"].get_object() + assert pymodule["result"].get_object().get_type() is pymodule["Owner"].get_object() + before = runtime_observed(source) + offset = source.index("field = " + ("1" if renamed_owner == "Owner" else "2")) + model_project.do( + rename.Rename(model_project, resource, offset).get_changes("changed") + ) + assert runtime_observed(resource.read()) == before == (1, 2) + assert ("result.changed" in resource.read()) == (renamed_owner == "Owner") + assert ("local.changed" in resource.read()) == (renamed_owner == "Other") + + +@pytest.mark.skipif( + sys.version_info < (3, 12), reason="type aliases require Python 3.12" +) +@pytest.mark.parametrize("enabled", [False, True]) +def test_conditional_class_alias_queries_are_readable_and_candidate_edits_are_atomic( + model_project, enabled +): + source = ( + f"target = int\nenabled = {enabled}\nclass Owner:\n" + " if enabled:\n target = bytes\n" + " type Alias = tuple[target, target]\n" + "unrelated = 42\nobserved = unrelated\n" + ) + resource = module(model_project, source) + pymodule = model_project.get_pymodule(resource) + offset = source.index("target", source.index("tuple[")) + assert ( + evaluate.eval_location(pymodule, offset).get_object() is pyobjects.get_unknown() + ) + assert codeassist.get_definition_location( + model_project, source, offset, resource + ) == (resource, 6) + assert finderrors.find_errors(model_project, resource) == [] + before = {} + exec(compile(source, "conditional_alias.py", "exec", dont_inherit=True), before) + expected = bytes if enabled else int + assert before["Owner"].Alias.__value__.__args__ == (expected, expected) + for selection in (offset, 0, source.index("target = bytes")): + with pytest.raises(exceptions.RefactoringError, match="unresolved"): + rename.Rename(model_project, resource, selection).get_changes("changed") + assert resource.read() == source + model_project.do( + rename.Rename(model_project, resource, source.index("unrelated =")).get_changes( + "safe" + ) + ) + after = {} + exec( + compile(resource.read(), "conditional_alias.py", "exec", dont_inherit=True), + after, + ) + assert after["Owner"].Alias.__value__.__args__ == (expected, expected) + assert after["observed"] == before["observed"] == 42 + + +@pytest.mark.skipif( + sys.version_info < (3, 12), reason="type parameters require Python 3.12" +) +def test_generic_class_bound_query_and_outer_rename_keep_runtime_bound(model_project): + source = "target = int\nclass Owner[T: target]:\n pass\nobserved = Owner.__type_params__[0].__bound__ is int\n" + resource = module(model_project, source) + pymodule = model_project.get_pymodule(resource) + offset = source.index("target", source.index("[T:")) + assert evaluate.eval_location(pymodule, offset) is pymodule["target"] + assert codeassist.get_definition_location( + model_project, source, offset, resource + ) == (resource, 1) + before = runtime_observed(source) + model_project.do(rename.Rename(model_project, resource, 0).get_changes("changed")) + assert runtime_observed(resource.read()) == before is True + assert "class Owner[T: changed]" in resource.read()