tests/test_go_interface_dispatch.py
Rama v8 · réplica de lectura del repositorio original.
"""Go member-level interface dispatch.
`d.provider.GetWithDataSource()` on a field typed with a local interface
resolves to the interface's own `.GetWithDataSource()` requirement, which is
what the call site names. Go types satisfy an interface implicitly, so there
is no `implements` edge for the shared dispatch mechanism to read, and the
concrete method used to sit unreachable from every call through the interface.
`resolve_go_interface_dispatch` infers the implementers from method sets (every
method name the interface requires) and links each requirement to the matching
method. Confidence falls as the implementer count grows (0.85 / 0.75 / 0.65).
"""
from __future__ import annotations
from pathlib import Path
from graphify.extract import extract
_GO_MOD = "module example.com/app\n\ngo 1.22\n"
_PROVIDER = (
"package plugincontext\n\n"
"type Provider struct{}\n\n"
"func (p *Provider) GetWithDataSource(uid string) error { return p.lookup(uid) }\n\n"
"func (p *Provider) lookup(uid string) error { return nil }\n"
)
_NODES = (
"package expr\n\n"
"type pluginContextProvider interface {\n"
"\tGetWithDataSource(uid string) error\n"
"}\n\n"
"type DSNode struct {\n"
"\tprovider pluginContextProvider\n"
"}\n\n"
"func (d *DSNode) Execute() error {\n"
"\treturn d.provider.GetWithDataSource(\"x\")\n"
"}\n"
)
def _extract(tmp_path: Path, files: dict[str, str]):
(tmp_path / "go.mod").write_text(_GO_MOD)
for name, body in files.items():
p = tmp_path / name
p.parent.mkdir(parents=True, exist_ok=True)
p.write_text(body)
r = extract(
sorted(tmp_path / n for n in files),
cache_root=tmp_path / "cache",
root=tmp_path,
parallel=False,
)
dispatch = {(e["source"], e["target"]) for e in r["edges"]
if e["relation"] == "dispatches_to"}
return dispatch, r
def _find(r, label, id_contains):
return next(n["id"] for n in r["nodes"]
if n["label"] == label and id_contains in n["id"])
def _reachable(r, start: str) -> set[str]:
adjacency: dict[str, list[str]] = {}
for e in r["edges"]:
adjacency.setdefault(e["source"], []).append(e["target"])
seen = {start}
queue = [start]
while queue:
for nxt in adjacency.get(queue.pop(0), []):
if nxt not in seen:
seen.add(nxt)
queue.append(nxt)
return seen
def test_local_interface_dispatches_to_an_implementer_in_another_package(tmp_path):
dispatch, r = _extract(tmp_path, {
"plugincontext/provider.go": _PROVIDER,
"expr/nodes.go": _NODES,
})
requirement = _find(r, ".GetWithDataSource()", "plugincontextprovider")
concrete = _find(r, ".GetWithDataSource()", "plugincontext_provider")
assert dispatch == {(requirement, concrete)}
edge = next(e for e in r["edges"] if e["relation"] == "dispatches_to")
assert edge["confidence"] == "INFERRED"
assert edge["confidence_score"] == 0.85
assert edge["context"] == "call"
assert edge["source_file"].endswith("provider.go")
# Additive: the call to the interface method stays, and the chain now
# continues into the concrete method's own callees.
calls = {(e["source"], e["target"]) for e in r["edges"] if e["relation"] == "calls"}
assert (_find(r, ".Execute()", "dsnode"), requirement) in calls
assert _find(r, ".lookup()", "provider") in _reachable(r, _find(r, ".Execute()", "dsnode"))
# The inferred pairs feed the dispatch pass; they are not graph edges.
assert not any(e["relation"] == "implements" for e in r["edges"])
def test_value_and_pointer_receivers_both_count(tmp_path):
dispatch, r = _extract(tmp_path, {"shape/shape.go": (
"package shape\n\n"
"type Shape interface {\n\tArea() float64\n\tName() string\n}\n\n"
"type Square struct{ side float64 }\n\n"
"func (s Square) Area() float64 { return s.side * s.side }\n\n"
"func (s *Square) Name() string { return \"square\" }\n"
)})
assert dispatch == {
(_find(r, ".Area()", "shape_shape"), _find(r, ".Area()", "square")),
(_find(r, ".Name()", "shape_shape"), _find(r, ".Name()", "square")),
}
def test_a_type_missing_one_method_does_not_implement(tmp_path):
dispatch, _ = _extract(tmp_path, {"shape/shape.go": (
"package shape\n\n"
"type Shape interface {\n\tArea() float64\n\tName() string\n}\n\n"
"type Square struct{}\n\n"
"func (s Square) Area() float64 { return 1 }\n"
)})
assert not dispatch
def test_two_implementers_both_get_an_edge_at_lower_confidence(tmp_path):
dispatch, r = _extract(tmp_path, {"shape/shape.go": (
"package shape\n\n"
"type Shape interface {\n\tArea() float64\n}\n\n"
"type Square struct{}\n\n"
"func (s Square) Area() float64 { return 1 }\n\n"
"type Circle struct{}\n\n"
"func (c *Circle) Area() float64 { return 2 }\n"
)})
requirement = _find(r, ".Area()", "shape_shape")
assert dispatch == {
(requirement, _find(r, ".Area()", "square")),
(requirement, _find(r, ".Area()", "circle")),
}
assert {e["confidence_score"] for e in r["edges"]
if e["relation"] == "dispatches_to"} == {0.75}
def test_empty_interface_is_ignored(tmp_path):
dispatch, _ = _extract(tmp_path, {"box/box.go": (
"package box\n\n"
"type Anything interface{}\n\n"
"type Box struct{ v Anything }\n\n"
"func (b Box) Get() Anything { return b.v }\n"
)})
assert not dispatch
def test_embedded_interface_methods_are_required(tmp_path):
# ReadCloser requires Read (via Reader) and Close; File has both, Pipe only Close.
dispatch, r = _extract(tmp_path, {"io2/io.go": (
"package io2\n\n"
"type Reader interface {\n\tRead() int\n}\n\n"
"type ReadCloser interface {\n\tReader\n\tClose() error\n}\n\n"
"type File struct{}\n\n"
"func (f *File) Read() int { return 0 }\n\n"
"func (f *File) Close() error { return nil }\n\n"
"type Pipe struct{}\n\n"
"func (p *Pipe) Close() error { return nil }\n"
)})
assert (_find(r, ".Close()", "readcloser"), _find(r, ".Close()", "file")) in dispatch
assert (_find(r, ".Read()", "reader"), _find(r, ".Read()", "file")) in dispatch
assert not any("pipe" in target for _, target in dispatch)
def test_interface_embedding_an_external_interface_is_skipped(tmp_path):
# io.Reader's methods are not in the corpus, so the method set is unknown.
dispatch, _ = _extract(tmp_path, {"rc/rc.go": (
"package rc\n\n"
"import \"io\"\n\n"
"type ReadCloser interface {\n\tio.Reader\n\tClose() error\n}\n\n"
"type Pipe struct{}\n\n"
"func (p *Pipe) Close() error { return nil }\n"
)})
assert not dispatch
def test_unexported_method_is_only_implemented_in_the_same_package(tmp_path):
dispatch, r = _extract(tmp_path, {
"a/a.go": (
"package a\n\n"
"type runner interface {\n\trun()\n}\n\n"
"type local struct{}\n\n"
"func (l local) run() {}\n"
),
"b/b.go": "package b\n\ntype remote struct{}\n\nfunc (r remote) run() {}\n",
})
assert dispatch == {(_find(r, ".run()", "a_runner"), _find(r, ".run()", "a_local"))}
def test_interface_with_too_many_implementers_is_skipped(tmp_path):
from graphify.go_dispatch import _MAX_IMPLICIT_IMPLEMENTERS
types = "".join(
f"type T{i} struct{{}}\n\nfunc (t T{i}) Close() error {{ return nil }}\n\n"
for i in range(_MAX_IMPLICIT_IMPLEMENTERS + 1)
)
dispatch, _ = _extract(tmp_path, {"c/c.go": (
"package c\n\ntype closer interface {\n\tClose() error\n}\n\n" + types
)})
assert not dispatch
def test_a_non_go_interface_is_untouched(tmp_path):
from graphify.go_dispatch import resolve_go_interface_dispatch
nodes = [
{"id": "iface", "label": "IR", "source_file": "a.cs",
"metadata": {"language": "go", "kind": "interface"}},
{"id": "impl", "label": "A", "source_file": "a.go"},
{"id": "iface_m", "label": ".M()", "source_file": "a.cs"},
{"id": "impl_m", "label": ".M()", "source_file": "a.go"},
]
edges = [
{"source": "iface", "target": "iface_m", "relation": "method"},
{"source": "impl", "target": "impl_m", "relation": "method"},
]
resolve_go_interface_dispatch([], nodes, edges)
assert not any(e["relation"] == "dispatches_to" for e in edges)