634 lines
33 KiB
Python
634 lines
33 KiB
Python
"""Call discovery preserves one-hop ownership, provenance, and bounded traversal."""
|
|
|
|
import pytest
|
|
|
|
from band_aider.judgement import source_discovery_strategy
|
|
from band_aider.judgement.source_discovery_strategy import CallTraceNode, CallImport
|
|
from band_aider.strategies.python.python_source_discovery import PythonSourceDiscovery
|
|
from tests.support.memory_repo_exec import MemoryRepoExec
|
|
|
|
|
|
@pytest.mark.parametrize("previous,current,call,name", [
|
|
("class Service: pass\n", "pass\n", "Service()", "Service"),
|
|
("class Service:\n def run(self): return 1\n", "class Service: pass\n", "Service().run()", "Service.run"),
|
|
("class Service:\n @property\n def value(self): return 1\n", "class Service: pass\n", "Service().value", "Service.value"),
|
|
])
|
|
def test_pristine_resolves_missing_class_method_and_property(previous: str, current: str, call: str, name: str) -> None:
|
|
"""The surviving caller identifies a removed declaration without manufacturing execution."""
|
|
|
|
test = f"from service import Service\ndef test_entry():\n {call}\n"
|
|
discovery = _discovery({"src/service.py": current, "tests/test_entry.py": test})
|
|
pristine = {"src/service.py": previous, "tests/test_entry.py": test}
|
|
|
|
root = discovery.get_call_tree("tests/test_entry.py", "test_entry", pristine=pristine)
|
|
|
|
removed = tuple(filter(lambda node: node.is_removed, root.calls))
|
|
assert tuple(map(lambda node: node.name, removed)) == (name,)
|
|
assert removed[0].line is not None
|
|
assert removed[0].end_line is not None
|
|
assert removed[0].call_line == 3
|
|
assert "[REMOVED target; pristine declaration]" in root.stringify()
|
|
|
|
|
|
def test_abstract_construction_finds_removed_implementation_through_wildcard_reexport() -> None:
|
|
"""A concrete class that loses its abstract override remains attributable to its own file."""
|
|
|
|
files = {
|
|
"tests/__init__.py": "",
|
|
"src/base.py": "from abc import ABC, abstractmethod\nclass ProtoLoop(ABC):\n @abstractmethod\n def _prepare_work(self): pass\n",
|
|
"src/alpha.py": "from base import ProtoLoop\nclass AlphaLoop(ProtoLoop):\n pass\n",
|
|
"tests/helper.py": "from alpha import AlphaLoop\n",
|
|
"tests/test_entry.py": "from tests.helper import *\ndef test_entry():\n AlphaLoop()\n",
|
|
}
|
|
original = "from base import ProtoLoop\nclass AlphaLoop(ProtoLoop):\n def _prepare_work(self):\n return None\n"
|
|
pristine = dict(files, **{"src/alpha.py": original})
|
|
discovery = _discovery(files)
|
|
|
|
root = discovery.get_call_tree("tests/test_entry.py", "test_entry", pristine=pristine)
|
|
|
|
assert [(node.location, node.name) for node in root.calls if node.is_removed] == [
|
|
("src/alpha.py", "AlphaLoop._prepare_work"),
|
|
]
|
|
restored = _discovery(pristine).get_call_tree("tests/test_entry.py", "test_entry", pristine=pristine)
|
|
assert not any(node.is_removed for node in restored.calls)
|
|
|
|
|
|
def test_changed_method_body_is_not_a_removed_symbol() -> None:
|
|
"""Historical text disappearance cannot falsely mark a surviving declaration removed."""
|
|
|
|
test = "from service import Service\ndef test_entry():\n Service().run()\n"
|
|
discovery = _discovery({"src/service.py": "class Service:\n def run(self): return 2\n", "tests/test_entry.py": test})
|
|
root = discovery.get_call_tree("tests/test_entry.py", "test_entry", pristine={
|
|
"src/service.py": "class Service:\n def run(self): return 1\n", "tests/test_entry.py": test,
|
|
})
|
|
assert root.calls
|
|
assert not any(node.is_removed for node in root.calls)
|
|
|
|
|
|
def test_pristine_target_cannot_escape_candidate_predicate() -> None:
|
|
"""Historical discovery cannot expand the caller's permitted production pool."""
|
|
|
|
test = "from service import Service\ndef test_entry():\n Service()\n"
|
|
discovery = _discovery({"src/service.py": "pass\n", "tests/test_entry.py": test})
|
|
root = discovery.get_call_tree("tests/test_entry.py", "test_entry", pristine={
|
|
"src/service.py": "class Service: pass\n", "tests/test_entry.py": test,
|
|
}, predicate=lambda node: node.location != "src/service.py")
|
|
assert root.calls == []
|
|
|
|
|
|
def test_inherited_missing_implementation_points_to_the_removed_override_owner() -> None:
|
|
"""Construction attributes a lost inherited implementation to its defining parent file."""
|
|
|
|
files = {
|
|
"src/base.py": "from abc import ABC, abstractmethod\nclass Base(ABC):\n @abstractmethod\n def run(self): pass\n",
|
|
"src/parent.py": "from base import Base\nclass Parent(Base):\n pass\n",
|
|
"src/child.py": "from parent import Parent\nclass Child(Parent):\n pass\n",
|
|
"tests/test_entry.py": "from child import Child\ndef test_entry():\n Child()\n",
|
|
}
|
|
pristine = dict(files, **{"src/parent.py": "from base import Base\nclass Parent(Base):\n def run(self): return 1\n"})
|
|
root = _discovery(files).get_call_tree("tests/test_entry.py", "test_entry", pristine=pristine)
|
|
|
|
assert [(node.location, node.name) for node in root.calls if node.is_removed] == [("src/parent.py", "Parent.run")]
|
|
|
|
|
|
def test_ambiguous_reference_does_not_hide_a_later_deleted_target() -> None:
|
|
"""Ambiguity remains unresolved while independent, uniquely missing calls stay discoverable."""
|
|
|
|
files = {
|
|
"src/a.py": "class Service: pass\n",
|
|
"src/b.py": "class Service: pass\n",
|
|
"src/removed.py": "pass\n",
|
|
"tests/test_entry.py": "from a import *\nfrom b import *\nfrom removed import Gone\ndef test_entry():\n Service()\n Gone()\n",
|
|
}
|
|
pristine = dict(files, **{"src/removed.py": "class Gone: pass\n"})
|
|
root = _discovery(files).get_call_tree("tests/test_entry.py", "test_entry", pristine=pristine)
|
|
|
|
assert [(node.location, node.name, node.is_removed) for node in root.calls] == [("src/removed.py", "Gone", True)]
|
|
|
|
|
|
@pytest.mark.parametrize("current", ["not valid python !", "Service = choose_dynamically()\n"])
|
|
def test_unreadable_or_unknown_current_target_does_not_prove_deletion(current: str) -> None:
|
|
"""Syntax errors and unknown dynamic bindings never authorize pristine deletion evidence."""
|
|
|
|
test = "from service import Service\ndef test_entry():\n Service()\n"
|
|
root = _discovery({"src/service.py": current, "tests/test_entry.py": test}).get_call_tree(
|
|
"tests/test_entry.py", "test_entry", pristine={"src/service.py": "class Service: pass\n", "tests/test_entry.py": test},
|
|
)
|
|
assert root.calls == []
|
|
|
|
|
|
def _discovery(files: dict[str, str]) -> PythonSourceDiscovery:
|
|
"""Construct an isolated source repository without executing its modules."""
|
|
repository = MemoryRepoExec()
|
|
# Each scenario owns all of its source and import evidence.
|
|
for location, source in files.items():
|
|
repository.write_text(repository.root / location, source)
|
|
return PythonSourceDiscovery(
|
|
repository.root / "src", repository.root / "tests",
|
|
repo_exec=repository, repo_root=repository.root,
|
|
)
|
|
|
|
|
|
def test_visitor_populates_only_target_calls_and_can_resume_each_child() -> None:
|
|
"""The caller, not the visitor, chooses which subsequent node to expand."""
|
|
discovery = _discovery({
|
|
"tests/test_entry.py": "from service import run\ndef test_entry():\n run()\n",
|
|
"src/service.py": "from worker import finish\ndef run():\n finish()\n",
|
|
"src/worker.py": "def finish():\n return True\n",
|
|
})
|
|
root = discovery.get_call_tree("tests/test_entry.py", "test_entry", depth=0)
|
|
seen: set[str] = set()
|
|
|
|
result = discovery.visit_call_sites(root, seen)
|
|
|
|
assert result is None
|
|
assert root.name == "test_entry"
|
|
assert root.import_reference is None
|
|
assert [(node.location, node.name) for node in root.calls] == [("src/service.py", "run")]
|
|
child = root.calls[0]
|
|
assert child.calls == []
|
|
assert seen
|
|
assert child.caller_location == "tests/test_entry.py"
|
|
assert (child.call_line, child.call_column) == (3, 4)
|
|
assert child.import_reference is not None
|
|
assert child.import_reference.text == "from service import run"
|
|
assert child.import_reference.line == 1
|
|
assert child.import_reference.imported_name == "service.run"
|
|
|
|
discovery.visit_call_sites(child, seen)
|
|
discovery.visit_call_sites(root, seen)
|
|
|
|
assert len(root.calls) == 1
|
|
assert [(node.location, node.name) for node in child.calls] == [("src/worker.py", "finish")]
|
|
|
|
|
|
@pytest.mark.parametrize("depth", [0, 1, 2, 3])
|
|
def test_tree_returns_topmost_root_and_obeys_depth(depth: int) -> None:
|
|
"""A depth limit applies to edges from the returned origin, not the last node."""
|
|
discovery = _discovery({
|
|
"tests/test_entry.py": "from service import run\ndef test_entry():\n run()\n",
|
|
"src/service.py": "from worker import finish\ndef run():\n finish()\n",
|
|
"src/worker.py": "def finish():\n return True\n",
|
|
})
|
|
|
|
root = discovery.get_call_tree("tests/test_entry.py", line=3, depth=depth)
|
|
|
|
assert (root.location, root.name, root.line) == ("tests/test_entry.py", "test_entry", 2)
|
|
assert len(root.calls) == int(depth > 0)
|
|
# Only the second permitted expansion can reach the worker.
|
|
if root.calls:
|
|
assert len(root.calls[0].calls) == int(depth > 1)
|
|
# The terminal callable has no children regardless of remaining depth.
|
|
if root.calls[0].calls:
|
|
assert root.calls[0].calls[0].calls == []
|
|
|
|
|
|
def test_tuple_helper_annotation_resolves_swallowed_editor_call() -> None:
|
|
"""The recorded helper binding identifies production without a production frame."""
|
|
discovery = _discovery({
|
|
"tests/test_editor.py": (
|
|
"from editor import DirectCodeEditor\n"
|
|
"def _runtime_editor() -> tuple[DirectCodeEditor, str]:\n"
|
|
" return DirectCodeEditor(), 'file'\n"
|
|
"def test_edit():\n"
|
|
" editor, source = _runtime_editor()\n"
|
|
" changed, summary = editor.edit(source)\n"
|
|
" assert changed\n"
|
|
),
|
|
"src/editor.py": "class DirectCodeEditor:\n def edit(self, source):\n return [], 'caught error'\n",
|
|
})
|
|
|
|
root = discovery.get_call_tree("tests/test_editor.py", "test_edit")
|
|
|
|
assert [(node.location, node.name, node.line) for node in root.calls] == [
|
|
("src/editor.py", "DirectCodeEditor.edit", None),
|
|
]
|
|
assert root.calls[0].call_line == 6
|
|
assert root.calls[0].import_reference is not None
|
|
assert root.calls[0].import_reference.imported_name == "editor.DirectCodeEditor"
|
|
|
|
|
|
@pytest.mark.parametrize("setup,call", [
|
|
("from service import run as invoke", "invoke()"),
|
|
("import service as api", "api.run()"),
|
|
("from service import Service\nclient = Service()", "client.run()"),
|
|
("from service import Service\nclient: Service", "client.run()"),
|
|
])
|
|
def test_import_and_receiver_bindings_retain_provenance(setup: str, call: str) -> None:
|
|
"""Functions, modules, aliases, and explicit receiver types resolve mechanically."""
|
|
discovery = _discovery({
|
|
"tests/test_entry.py": f"{setup}\ndef test_entry():\n {call}\n",
|
|
"src/service.py": "def run():\n pass\nclass Service:\n def run(self):\n pass\n",
|
|
})
|
|
|
|
root = discovery.get_call_tree("tests/test_entry.py", "test_entry")
|
|
|
|
assert len(root.calls) == 1
|
|
assert root.calls[0].location == "src/service.py"
|
|
assert root.calls[0].name in {"run", "Service.run"}
|
|
assert root.calls[0].import_reference is not None
|
|
|
|
|
|
def test_relative_import_and_constructor_resolve_without_execution() -> None:
|
|
"""Relative imports and explicit constructors are calls, not module execution."""
|
|
discovery = _discovery({
|
|
"src/pkg/__init__.py": "",
|
|
"src/pkg/entry.py": "from .worker import Worker\ndef run():\n Worker()\n",
|
|
"src/pkg/worker.py": "raise RuntimeError('must not execute')\nclass Worker:\n def __init__(self):\n pass\n",
|
|
})
|
|
|
|
root = discovery.get_call_tree("src/pkg/entry.py", "run")
|
|
|
|
assert [(node.location, node.name) for node in root.calls] == [
|
|
("src/pkg/worker.py", "Worker"),
|
|
]
|
|
|
|
|
|
def test_source_order_cycles_duplicates_and_distinct_methods_share_one_seen_set() -> None:
|
|
"""File deduplication must not erase a second callable's distinct descendants."""
|
|
discovery = _discovery({
|
|
"src/entry.py": "from service import second, first\ndef run():\n first(); second(); first()\n",
|
|
"src/service.py": "from left import leaf as a\nfrom right import leaf as b\ndef first():\n a()\ndef second():\n b()\n",
|
|
"src/left.py": "from entry import run\ndef leaf():\n run()\n",
|
|
"src/right.py": "def leaf():\n pass\n",
|
|
})
|
|
seen: set[str] = set()
|
|
|
|
root = discovery.get_call_tree("src/entry.py", "run", depth=5, already_seen=seen)
|
|
|
|
assert [node.name for node in root.calls] == ["first", "second"]
|
|
assert [node.calls[0].location for node in root.calls] == ["src/left.py", "src/right.py"]
|
|
assert root.calls[0].calls[0].calls == []
|
|
assert len(seen) == 5
|
|
|
|
|
|
def test_rejected_call_is_not_seen_and_can_be_admitted_later() -> None:
|
|
"""The predicate limits admission without poisoning subsequent discovery."""
|
|
discovery = _discovery({
|
|
"src/entry.py": "from worker import run\ndef start():\n run()\n",
|
|
"src/worker.py": "def run():\n pass\n",
|
|
})
|
|
root = discovery.get_call_tree("src/entry.py", "start", depth=0)
|
|
seen: set[str] = set()
|
|
|
|
discovery.visit_call_sites(root, seen, lambda node: False)
|
|
|
|
assert root.calls == []
|
|
assert len(seen) == 1
|
|
discovery.visit_call_sites(root, seen)
|
|
assert len(root.calls) == 1
|
|
assert len(seen) == 2
|
|
|
|
|
|
@pytest.mark.parametrize("source", [
|
|
"def run():\n pass\ndef test_entry():\n run()\n",
|
|
"from service import run\ndef test_entry(run):\n run()\n",
|
|
"from service import run\ndef test_entry():\n run = lambda: None\n run()\n",
|
|
"from service import *\ndef test_entry():\n run()\n",
|
|
"from external import run\ndef test_entry():\n run()\n",
|
|
"from service import Service\ndef test_entry(flag):\n if flag:\n client = Service()\n client.run()\n",
|
|
"from service import run\ndef test_entry():\n def inner():\n run()\n assert True\n",
|
|
])
|
|
def test_unresolved_or_unexecuted_nested_calls_do_not_nominate_files(source: str) -> None:
|
|
"""Unrelated names and uncertain receivers are not static evidence of a call."""
|
|
discovery = _discovery({
|
|
"tests/test_entry.py": source,
|
|
"src/service.py": "def run():\n pass\nclass Service:\n def run(self):\n pass\n",
|
|
})
|
|
|
|
root = discovery.get_call_tree("tests/test_entry.py", "test_entry")
|
|
|
|
# A constructor syntactically inside a branch is still a direct imported call,
|
|
# but its receiver binding must not establish a later method call.
|
|
assert all(node.name != "Service.run" for node in root.calls)
|
|
assert all(node.name == "Service" for node in root.calls)
|
|
|
|
|
|
def test_invalid_depth_and_ambiguous_origin_are_explicit_failures() -> None:
|
|
"""Malformed requests and unresolved origins cannot fabricate a call tree."""
|
|
discovery = _discovery({"src/entry.py": "def one():\n pass\ndef two():\n pass\n"})
|
|
|
|
with pytest.raises(ValueError):
|
|
discovery.get_call_tree("src/entry.py", "one", depth=-1)
|
|
with pytest.raises(LookupError):
|
|
discovery.get_call_tree("src/entry.py")
|
|
|
|
|
|
def test_caller_can_supply_a_resolved_root_directly() -> None:
|
|
"""The visitor requires neither a wrapper invocation nor an output list."""
|
|
discovery = _discovery({
|
|
"src/entry.py": "from worker import run\ndef start():\n run()\n",
|
|
"src/worker.py": "def run():\n pass\n",
|
|
})
|
|
root = CallTraceNode(location="src/entry.py", name="start", line=2)
|
|
|
|
discovery.visit_call_sites(root)
|
|
|
|
assert root.calls[0].location == "src/worker.py"
|
|
|
|
|
|
def test_invalid_admitted_destination_does_not_erase_a_valid_sibling() -> None:
|
|
"""A broken child owns its discovery failure, not the whole origin's evidence."""
|
|
discovery = _discovery({
|
|
"tests/test_entry.py": "from valid import run\nfrom broken import fail\ndef test_entry():\n run()\n fail()\n",
|
|
"src/valid.py": "from leaf import finish\ndef run():\n finish()\n",
|
|
"src/leaf.py": "def finish(): pass\n",
|
|
"src/broken.py": "def fail(:\n",
|
|
})
|
|
|
|
root = discovery.get_call_tree("tests/test_entry.py", "test_entry", depth=3)
|
|
|
|
assert [node.location for node in root.calls] == ["src/valid.py", "src/broken.py"]
|
|
assert root.calls[0].calls[0].location == "src/leaf.py"
|
|
assert root.calls[1].calls == []
|
|
|
|
|
|
@pytest.mark.parametrize("depth", [1, 2, 3])
|
|
def test_argument_bindings_follow_exactly_the_requested_production_depth(depth: int) -> None:
|
|
"""Concrete caller values survive two parameter bindings without widening the limit."""
|
|
discovery = _discovery({
|
|
"tests/test_builder.py": (
|
|
"from builder import Builder\nfrom dto import SomeDto\nfrom leaf import Leaf\n"
|
|
"from contracts import Protocol\n"
|
|
"def test_builder():\n"
|
|
" dto: Protocol = SomeDto()\n leaf = Leaf()\n builder = Builder()\n"
|
|
" builder.method1(dto, leaf=leaf)\n builder.method2(dto)\n"
|
|
),
|
|
"src/builder.py": (
|
|
"class Builder:\n"
|
|
" def method1(self, value, *, leaf):\n"
|
|
" try:\n value.resolve(leaf)\n except ValueError:\n pass\n"
|
|
" def method2(self, value):\n return value\n"
|
|
),
|
|
"src/dto.py": "class SomeDto:\n def resolve(self, destination):\n destination.finish()\n",
|
|
"src/leaf.py": "from end import stop\nclass Leaf:\n def finish(self):\n stop()\n",
|
|
"src/end.py": "def stop():\n pass\n",
|
|
"src/contracts.py": "class Protocol:\n def resolve(self, value):\n pass\n",
|
|
})
|
|
|
|
root = discovery.get_call_tree("tests/test_builder.py", "test_builder", depth=depth)
|
|
method = next(node for node in root.calls if node.name == "Builder.method1")
|
|
|
|
assert method.arguments[0].expression == "dto"
|
|
assert method.arguments[0].value is not None
|
|
assert (method.arguments[0].value.location, method.arguments[0].value.name) == ("src/dto.py", "SomeDto")
|
|
assert method.arguments[1].keyword == "leaf"
|
|
assert method.arguments[1].value is not None
|
|
assert method.arguments[1].value.name == "Leaf"
|
|
assert method.caller_name == "test_builder"
|
|
assert method.boundary is not None
|
|
assert method.boundary.location == "src/builder.py"
|
|
assert [node.name for node in root.calls][-2:] == ["Builder.method1", "Builder.method2"]
|
|
# Each explicit depth authorizes exactly one more production-body inspection.
|
|
if depth == 1:
|
|
assert method.calls == []
|
|
else:
|
|
resolve = method.calls[0]
|
|
assert (resolve.location, resolve.name) == ("src/dto.py", "SomeDto.resolve")
|
|
assert resolve.import_reference is not None
|
|
assert resolve.import_reference.location == "tests/test_builder.py"
|
|
assert resolve.arguments[0].value is method.arguments[1].value
|
|
# The third level reaches the argument's implementation, not its annotation.
|
|
if depth == 2:
|
|
assert resolve.calls == []
|
|
else:
|
|
assert [(node.location, node.name) for node in resolve.calls] == [("src/leaf.py", "Leaf.finish")]
|
|
assert resolve.calls[0].calls == []
|
|
|
|
|
|
def test_excluded_destination_is_never_read_and_valid_sibling_survives() -> None:
|
|
"""A rejected import cannot erase admitted evidence through a parse failure."""
|
|
discovery = _discovery({
|
|
"tests/test_entry.py": "from allowed import run\nfrom excluded import broken\ndef test_entry():\n run()\n broken()\n",
|
|
"src/allowed.py": "def run():\n pass\n",
|
|
"src/excluded.py": "this is not valid Python !\n",
|
|
})
|
|
from unittest.mock import patch
|
|
|
|
with patch.object(discovery._repo_exec, "read_text", wraps=discovery._repo_exec.read_text) as reads:
|
|
root = discovery.get_call_tree("tests/test_entry.py", "test_entry", depth=3,
|
|
predicate=lambda call: call.location == "src/allowed.py")
|
|
|
|
assert [(node.location, node.name) for node in root.calls] == [("src/allowed.py", "run")]
|
|
assert all(str(call.args[0]).endswith("excluded.py") is False for call in reads.call_args_list)
|
|
|
|
|
|
def test_distinct_argument_types_expand_the_same_method_independently() -> None:
|
|
"""Callable deduplication cannot lose a different imported argument implementation."""
|
|
discovery = _discovery({
|
|
"tests/test_entry.py": "from service import run\nfrom a import A\nfrom b import B\ndef test_entry():\n run(A())\n run(B())\n",
|
|
"src/service.py": "def run(value):\n value.resolve()\n",
|
|
"src/a.py": "class A:\n def resolve(self):\n pass\n",
|
|
"src/b.py": "class B:\n def resolve(self):\n pass\n",
|
|
})
|
|
|
|
root = discovery.get_call_tree("tests/test_entry.py", "test_entry", depth=3)
|
|
methods = tuple(filter(lambda node: node.name == "run", root.calls))
|
|
|
|
assert len(methods) == 2
|
|
assert methods[0].calls[0].location == "src/a.py"
|
|
assert methods[1].calls[0].location == "src/b.py"
|
|
|
|
|
|
def test_visible_setup_receiver_uses_concrete_initializer() -> None:
|
|
"""A visible test setup assignment establishes an imported receiver without execution."""
|
|
discovery = _discovery({
|
|
"tests/test_entry.py": (
|
|
"from service import Service\nclass TestEntry:\n"
|
|
" def setUp(self):\n self.service = Service()\n"
|
|
" def test_entry(self):\n self.service.run()\n"
|
|
),
|
|
"src/service.py": "class Service:\n def run(self):\n pass\n",
|
|
})
|
|
|
|
root = discovery.get_call_tree("tests/test_entry.py", "TestEntry.test_entry", depth=1)
|
|
|
|
assert [(node.location, node.name) for node in root.calls] == [("src/service.py", "Service.run")]
|
|
|
|
|
|
@pytest.mark.parametrize("depth", [1, 2, 3])
|
|
def test_constructor_arguments_survive_instance_field_lookup(depth: int) -> None:
|
|
"""Constructor-held arguments retain nested construction through later methods."""
|
|
discovery = _discovery({
|
|
"tests/test_entry.py": "from builder import Builder\nfrom dto import SomeDto\nfrom dependency import Dependency\ndef test_entry():\n dto = SomeDto(Dependency())\n builder = Builder(value=dto)\n builder.method1()\n builder.method2()\n",
|
|
"src/builder.py": "class Builder:\n def __init__(self, value):\n self.dto = value\n def method1(self):\n self.dto.resolve()\n def method2(self):\n pass\n",
|
|
"src/dto.py": "class SomeDto:\n def __init__(self, dependency):\n self.dependency = dependency\n def resolve(self):\n self.dependency.load()\n",
|
|
"src/dependency.py": "class Dependency:\n def load(self):\n pass\n",
|
|
})
|
|
|
|
root = discovery.get_call_tree("tests/test_entry.py", "test_entry", depth=depth)
|
|
method = next(node for node in root.calls if node.name == "Builder.method1")
|
|
|
|
assert method.receiver is not None
|
|
dto = method.receiver.arguments[0].value
|
|
assert dto is not None and dto.name == "SomeDto"
|
|
assert dto.arguments[0].value is not None
|
|
assert dto.arguments[0].value.name == "Dependency"
|
|
assert tuple(map(lambda node: node.name, method.calls)) == (() if depth == 1 else ("SomeDto.resolve",))
|
|
# Each extra depth authorizes one additional body, not the whole dependency tree.
|
|
if depth > 1:
|
|
assert tuple(map(lambda node: node.name, method.calls[0].calls)) == (() if depth == 2 else ("Dependency.load",))
|
|
|
|
|
|
@pytest.mark.parametrize("initialization", [
|
|
" service = Service()\n",
|
|
" def __init__(self):\n self.service = Service()\n",
|
|
" def __init__(self):\n from service import Service as Actual\n self.service = Actual()\n",
|
|
])
|
|
def test_class_and_initializer_scopes_resolve_receiver_attributes(initialization: str) -> None:
|
|
"""Visible class and instance construction use their own lexical import bindings."""
|
|
discovery = _discovery({
|
|
"tests/test_entry.py": "from service import Service\nclass TestEntry:\n" + initialization + " def test_entry(self):\n self.service.run()\n",
|
|
"src/service.py": "class Service:\n def run(self):\n pass\n",
|
|
})
|
|
|
|
root = discovery.get_call_tree("tests/test_entry.py", "TestEntry.test_entry", depth=2)
|
|
|
|
assert [(node.location, node.name) for node in root.calls] == [("src/service.py", "Service.run")]
|
|
|
|
|
|
def test_explicit_argument_nodes_override_annotations_without_control_name_collisions() -> None:
|
|
"""Argument names such as target and predicate belong to the callee, not visitor controls."""
|
|
discovery = _discovery({
|
|
"src/service.py": "def run(target, predicate: Protocol):\n target.resolve()\n predicate.finish()\n",
|
|
"src/dto.py": "class SomeDto:\n def resolve(self):\n pass\n def finish(self):\n pass\n",
|
|
})
|
|
value = CallTraceNode("src/dto.py", "SomeDto", import_reference=CallImport("tests/test_entry.py", 1, "from dto import SomeDto", "dto.SomeDto"))
|
|
root = CallTraceNode("src/service.py", "run")
|
|
|
|
discovery.visit_call_sites(root, target_args={"target": value, "predicate": value})
|
|
|
|
assert tuple(map(lambda node: node.name, root.calls)) == ("SomeDto.resolve", "SomeDto.finish")
|
|
assert all(node.receiver is value for node in root.calls)
|
|
|
|
|
|
@pytest.mark.parametrize("parameters,body,expected", [
|
|
("value", " value.resolve()\n", ()),
|
|
("", " value.resolve()\n value = Other()\n", ("Other",)),
|
|
("", " value = Other()\n value.resolve()\n", ("Other", "Other.resolve")),
|
|
("", " from other import Other as value\n value.resolve()\n", ("Other.resolve",)),
|
|
("", " value.resolve()\n from other import Other as value\n", ()),
|
|
("", " global value\n value.resolve()\n", ("SomeDto.resolve",)),
|
|
("", " global value\n value = Other()\n value.resolve()\n", ("Other", "Other.resolve")),
|
|
])
|
|
def test_local_and_global_binding_precedence(parameters: str, body: str, expected: tuple[str, ...]) -> None:
|
|
"""An unbound local never falls through to a same-named module import."""
|
|
discovery = _discovery({
|
|
"tests/test_entry.py": "from dto import SomeDto as value\nfrom other import Other\ndef test_entry(" + parameters + "):\n" + body,
|
|
"src/dto.py": "class SomeDto:\n def resolve(self):\n pass\n",
|
|
"src/other.py": "class Other:\n def resolve(self):\n pass\n",
|
|
})
|
|
|
|
root = discovery.get_call_tree("tests/test_entry.py", "test_entry", depth=1)
|
|
|
|
assert tuple(map(lambda node: node.name, root.calls)) == expected
|
|
|
|
|
|
def test_constructor_bindings_keep_distinct_implementations_for_the_same_method() -> None:
|
|
"""Changing only the injected implementation changes the resolved descendant."""
|
|
discovery = _discovery({
|
|
"tests/test_entry.py": "from builder import Builder\nfrom a import A\nfrom b import B\ndef test_entry():\n first = Builder(A())\n second = Builder(B())\n first.run()\n second.run()\n",
|
|
"src/builder.py": "class Builder:\n def __init__(self, value):\n self.value = value\n def run(self):\n self.value.resolve()\n",
|
|
"src/a.py": "class A:\n def resolve(self):\n pass\n",
|
|
"src/b.py": "class B:\n def resolve(self):\n pass\n",
|
|
})
|
|
|
|
root = discovery.get_call_tree("tests/test_entry.py", "test_entry", depth=3)
|
|
methods = tuple(filter(lambda node: node.name == "Builder.run", root.calls))
|
|
|
|
assert len(methods) == 2
|
|
assert methods[0].calls[0].name == "A.resolve"
|
|
assert methods[1].calls[0].name == "B.resolve"
|
|
|
|
|
|
@pytest.mark.parametrize("invocation,method", [
|
|
("service.instance(dto)", "Service.instance"),
|
|
("Service.instance(service, dto)", "Service.instance"),
|
|
("service.class_bound(dto)", "Service.class_bound"),
|
|
("Service.class_bound(dto)", "Service.class_bound"),
|
|
("service.static(dto)", "Service.static"),
|
|
("Service.static(dto)", "Service.static"),
|
|
("run(dto)", "run"),
|
|
])
|
|
def test_callable_kind_binds_only_its_actual_implicit_receiver(invocation: str, method: str) -> None:
|
|
"""Functions and static methods keep their first explicit argument, even named self."""
|
|
discovery = _discovery({
|
|
"tests/test_entry.py": "from service import Service, run\nfrom dto import SomeDto\ndef test_entry():\n service = Service()\n dto = SomeDto()\n " + invocation + "\n",
|
|
"src/service.py": "class Service:\n def instance(receiver, value):\n value.resolve()\n @classmethod\n def class_bound(klass, value):\n value.resolve()\n @staticmethod\n def static(self):\n self.resolve()\ndef run(self):\n self.resolve()\n",
|
|
"src/dto.py": "class SomeDto:\n def resolve(self):\n pass\n",
|
|
})
|
|
|
|
root = discovery.get_call_tree("tests/test_entry.py", "test_entry", depth=2)
|
|
call = next(node for node in root.calls if node.name == method)
|
|
|
|
assert [(node.location, node.name) for node in call.calls] == [("src/dto.py", "SomeDto.resolve")]
|
|
|
|
|
|
def test_class_method_uses_class_state_not_instance_initialization() -> None:
|
|
"""A class receiver cannot acquire instance-only fields from __init__."""
|
|
discovery = _discovery({
|
|
"tests/test_entry.py": "from service import Service\ndef test_entry():\n Service.run()\n",
|
|
"src/service.py": "from dto import SomeDto\nfrom other import Other\nclass Service:\n value = SomeDto()\n def __init__(self):\n self.value = Other()\n @classmethod\n def run(cls):\n cls.value.resolve()\n",
|
|
"src/dto.py": "class SomeDto:\n def resolve(self):\n pass\n",
|
|
"src/other.py": "class Other:\n def resolve(self):\n pass\n",
|
|
})
|
|
|
|
root = discovery.get_call_tree("tests/test_entry.py", "test_entry", depth=2)
|
|
|
|
assert root.calls[0].calls[0].name == "SomeDto.resolve"
|
|
|
|
|
|
@pytest.mark.parametrize("expression,expected", [("VALUE.resolve()", "SomeDto.resolve"), ("self.VALUE.resolve()", "Other.resolve")])
|
|
def test_class_constants_are_not_method_lexical_globals(expression: str, expected: str) -> None:
|
|
"""Class attributes require a receiver; a bare constant uses the module binding."""
|
|
discovery = _discovery({
|
|
"tests/test_entry.py": "from service import Service\ndef test_entry():\n service = Service()\n service.run()\n",
|
|
"src/service.py": "from dto import SomeDto\nfrom other import Other\nVALUE = SomeDto()\nclass Service:\n VALUE = Other()\n def run(self):\n " + expression + "\n",
|
|
"src/dto.py": "class SomeDto:\n def resolve(self):\n pass\n",
|
|
"src/other.py": "class Other:\n def resolve(self):\n pass\n",
|
|
})
|
|
|
|
root = discovery.get_call_tree("tests/test_entry.py", "test_entry", depth=2)
|
|
run = next(node for node in root.calls if node.name == "Service.run")
|
|
|
|
assert tuple(map(lambda node: node.name, run.calls)) == (expected,)
|
|
|
|
|
|
def test_self_argument_node_supplies_constructor_state_for_a_renamed_receiver() -> None:
|
|
"""The special self binding identifies an instance, not the first parameter's spelling."""
|
|
discovery = _discovery({
|
|
"tests/test_entry.py": "from builder import Builder\nfrom dto import SomeDto\ndef test_entry():\n Builder(SomeDto())\n",
|
|
"src/builder.py": "class Builder:\n def __init__(receiver, value):\n receiver.dto = value\n def run(receiver):\n receiver.dto.resolve()\n",
|
|
"src/dto.py": "class SomeDto:\n def resolve(self):\n pass\n",
|
|
})
|
|
root = discovery.get_call_tree("tests/test_entry.py", "test_entry", depth=1)
|
|
instance = next(node for node in root.calls if node.name == "Builder")
|
|
target = CallTraceNode("src/builder.py", "Builder.run")
|
|
|
|
discovery.visit_call_sites(target, target_args={"self": instance})
|
|
|
|
assert tuple(map(lambda node: node.name, target.calls)) == ("SomeDto.resolve",)
|
|
|
|
|
|
@pytest.mark.parametrize("depth", [2, 3])
|
|
def test_stringify_preserves_indented_depth_first_order(depth: int, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""The formatter prints existing children without changing traversal or resolving source."""
|
|
discovery = _discovery({
|
|
"tests/test_entry.py": "from builder import first, second\ndef test_entry():\n first()\n second()\n",
|
|
"src/builder.py": "from dto import resolve\ndef first():\n resolve()\ndef second():\n pass\n",
|
|
"src/dto.py": "from dependency import load\ndef resolve():\n load()\n",
|
|
"src/dependency.py": "def load():\n pass\n",
|
|
})
|
|
root = discovery.get_call_tree("tests/test_entry.py", "test_entry", depth=depth)
|
|
|
|
rendered = root.stringify()
|
|
frames = tuple(filter(lambda line: 'File "' in line, rendered.splitlines()))
|
|
|
|
names = tuple(map(lambda line: line.rsplit(", in ", 1)[1], frames))
|
|
assert names == (("test_entry", "first", "resolve", "second") if depth == 2 else ("test_entry", "first", "resolve", "load", "second"))
|
|
indentation = tuple(map(lambda line: len(line) - len(line.lstrip()), frames))
|
|
assert indentation == ((2, 4, 6, 4) if depth == 2 else (2, 4, 6, 8, 4))
|
|
assert 'File "src/builder.py", in first' in rendered
|
|
assert "not observed execution" in rendered
|
|
monkeypatch.setattr(source_discovery_strategy, "CALL_TRACE_FRAME_TEMPLATE", 'File "{location}"{line_clause}, in {name}')
|
|
assert tuple(filter(lambda line: 'File "' in line, root.stringify().splitlines())) == tuple(map(str.lstrip, frames))
|