nprasad2077/codebase-memory-mcp
1
0 Fork 0
mirror of https://github.com/DeusData/codebase-memory-mcp.git synced 2026-11-04 10:48:20 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
main
codebase-memory-mcp/tests/test_py_lsp.c
T
Martin Vogel 054719044a fix(python): bind ClassName() only to the class own __init__ 
A construction call now resolves to __init__ only when the called
class declares that __init__ itself. An inherited __init__ is no
longer followed: class B(A) without its own __init__ keeps the B()
call on the Class B, as it was before this branch.

Following the bases bound every subclass construction to the base
constructor. On a django index (fb113765) that made Model.__init__
a 583-caller hub (every model instantiation), bound IntegerField()
to Field.__init__ and HttpResponseForbidden() to
HttpResponse.__init__, and degraded 625 constructor call sites
versus main (457 fell back to textual resolution, 168 lost their
row; CALLS -161). With own __init__ only, 6016 sites move to the
class own __init__, 5468 stay on the Class, none fall back to a
textual strategy, and total CALLS is +4 versus main.

Tests: the same-file and cross-file inherited cases now pin the
Class target, and a new same-file subclass test at the LSP and the
graph level covers B(A) without __init__ (stays on B, never
A.__init__) and C(A) with its own __init__ (binds to C.__init__,
never the class C or A.__init__). The strategy inventory comment in
the repro suite is updated to match.

Refs #1642

Signed-off-by: Martin Vogel <martin.vogel.tech@gmail.com>
2026-11-01 21:29:22 +01:00

5383 lines
232 KiB
C
Raw Permalink Blame History

/*
* test_py_lsp.c — Tests for Python LSP type-aware call resolution.
*
* Mirrors tests/test_go_lsp.c shape: helper extract_py(source) calls
* cbm_extract_file with CBM_LANG_PYTHON, then assertions search the
* resolved_calls array. Phase 2 ships smoke tests only; subsequent
* phases add categories matching the Go LSP layout (param types,
* method dispatch, decorators, generics, cross-file).
*/
#include "test_framework.h"
#include "cbm.h"
#include "lsp/py_lsp.h"
#include "lsp/lsp_work.h"
#include "pipeline/lsp_resolve.h"
#include "pipeline/pass_lsp_cross.h"
/* ── Helpers — same shape as test_go_lsp.c ──────────────────────── */
static CBMFileResult *extract_py(const char *source) {
return cbm_extract_file(source, (int)strlen(source), CBM_LANG_PYTHON,
"test", "main.py", 0, NULL, NULL);
}
static int find_resolved(const CBMFileResult *r, const char *callerSub, const char *calleeSub) {
for (int i = 0; i < r->resolved_calls.count; i++) {
const CBMResolvedCall *rc = &r->resolved_calls.items[i];
if (rc->caller_qn && strstr(rc->caller_qn, callerSub) && rc->callee_qn &&
strstr(rc->callee_qn, calleeSub))
return i;
}
return -1;
}
/* Avoid unused-static-function warnings: helpers compiled but not yet used
* outside the smoke tests will be referenced in Phase 3+ tests. */
__attribute__((unused))
static int require_resolved(const CBMFileResult *r, const char *callerSub, const char *calleeSub) {
int idx = find_resolved(r, callerSub, calleeSub);
if (idx < 0) {
printf(" MISSING resolved call: caller~%s -> callee~%s (have %d)\n",
callerSub, calleeSub, r->resolved_calls.count);
for (int i = 0; i < r->resolved_calls.count; i++) {
const CBMResolvedCall *rc = &r->resolved_calls.items[i];
printf(" %s -> %s [%s %.2f]\n",
rc->caller_qn ? rc->caller_qn : "(null)",
rc->callee_qn ? rc->callee_qn : "(null)",
rc->strategy ? rc->strategy : "(null)",
rc->confidence);
}
}
return idx;
}
/* ── Phase 2 — smoke ───────────────────────────────────────────── */
TEST(pylsp_smoke_empty) {
CBMFileResult *r = extract_py("");
ASSERT_NOT_NULL(r);
ASSERT_EQ(r->resolved_calls.count, 0);
cbm_free_result(r);
PASS();
}
TEST(pylsp_smoke_one_function) {
CBMFileResult *r = extract_py(
"def greet(name):\n"
" return name\n");
ASSERT_NOT_NULL(r);
/* Phase 2 stub: no resolutions yet, but extraction must succeed and
* the result must be addressable without crashes. */
ASSERT_GTE(r->defs.count, 1);
cbm_free_result(r);
PASS();
}
TEST(pylsp_smoke_one_class) {
CBMFileResult *r = extract_py(
"class Greeter:\n"
" def __init__(self, name):\n"
" self.name = name\n"
" def greet(self):\n"
" return self.name\n");
ASSERT_NOT_NULL(r);
/* Class + 2 methods at minimum */
ASSERT_GTE(r->defs.count, 1);
cbm_free_result(r);
PASS();
}
TEST(pylsp_no_crash_on_syntax_error) {
/* Tree-sitter recovers from errors but we must not crash on the
* recovered tree. */
CBMFileResult *r = extract_py(
"def broken(\n"
" x = 1\n"
"class\n");
ASSERT_NOT_NULL(r);
cbm_free_result(r);
PASS();
}
TEST(pylsp_smoke_imports_passed_through) {
/* Imports populate ctx->import_local_names — Phase 2 just verifies
* the unified extractor still produces them; resolution happens in
* Phase 3. */
CBMFileResult *r = extract_py(
"import os\n"
"import json as j\n"
"from pathlib import Path\n"
"from . import sibling\n"
"def use():\n"
" return os.getcwd()\n");
ASSERT_NOT_NULL(r);
ASSERT_GTE(r->imports.count, 3);
cbm_free_result(r);
PASS();
}
/* ── Phase 3 — imports → scope bindings ────────────────────────── */
/* Build a context, register one or more imports, run the binding pass,
* and let the caller verify scope state. */
static void bind_imports_into_ctx(PyLSPContext *ctx, CBMArena *a, CBMTypeRegistry *reg,
const char *const *locals, const char *const *qns,
int count) {
py_lsp_init(ctx, a, "", 0, reg, "test.main", NULL);
for (int i = 0; i < count; i++) {
py_lsp_add_import(ctx, locals[i], qns[i]);
}
py_lsp_bind_imports(ctx);
}
TEST(pylsp_import_simple) {
/* import os → os ∈ scope as MODULE("os") */
CBMArena a;
cbm_arena_init(&a);
CBMTypeRegistry reg;
cbm_registry_init(&reg, &a);
PyLSPContext ctx;
const char *locals[] = {"os"};
const char *qns[] = {"os"};
bind_imports_into_ctx(&ctx, &a, &reg, locals, qns, 1);
const CBMType *t = py_lsp_lookup_in_scope(&ctx, "os");
ASSERT(cbm_type_is_module(t));
ASSERT_STR_EQ(t->data.module.module_qn, "os");
cbm_arena_destroy(&a);
PASS();
}
TEST(pylsp_import_aliased) {
/* import json as j → j ∈ scope as MODULE("json") */
CBMArena a;
cbm_arena_init(&a);
CBMTypeRegistry reg;
cbm_registry_init(&reg, &a);
PyLSPContext ctx;
const char *locals[] = {"j"};
const char *qns[] = {"json"};
bind_imports_into_ctx(&ctx, &a, &reg, locals, qns, 1);
const CBMType *t = py_lsp_lookup_in_scope(&ctx, "j");
ASSERT(cbm_type_is_module(t));
ASSERT_STR_EQ(t->data.module.module_qn, "json");
/* original name "json" not bound */
const CBMType *miss = py_lsp_lookup_in_scope(&ctx, "json");
ASSERT(cbm_type_is_unknown(miss));
cbm_arena_destroy(&a);
PASS();
}
TEST(pylsp_import_from) {
/* from pathlib import Path → Path ∈ scope as NAMED("pathlib.Path") */
CBMArena a;
cbm_arena_init(&a);
CBMTypeRegistry reg;
cbm_registry_init(&reg, &a);
PyLSPContext ctx;
const char *locals[] = {"Path"};
const char *qns[] = {"pathlib.Path"};
bind_imports_into_ctx(&ctx, &a, &reg, locals, qns, 1);
const CBMType *t = py_lsp_lookup_in_scope(&ctx, "Path");
ASSERT_EQ(t->kind, CBM_TYPE_NAMED);
ASSERT_STR_EQ(t->data.named.qualified_name, "pathlib.Path");
cbm_arena_destroy(&a);
PASS();
}
TEST(pylsp_import_from_aliased) {
/* from pathlib import Path as P → P ∈ scope as NAMED("pathlib.Path") */
CBMArena a;
cbm_arena_init(&a);
CBMTypeRegistry reg;
cbm_registry_init(&reg, &a);
PyLSPContext ctx;
const char *locals[] = {"P"};
const char *qns[] = {"pathlib.Path"};
bind_imports_into_ctx(&ctx, &a, &reg, locals, qns, 1);
const CBMType *t = py_lsp_lookup_in_scope(&ctx, "P");
/* #988: an aliased from-import MUST bind NAMED (phase 6 upgrades it to
* the registered function/class/module). The earlier MODULE binding made
* `g()` calls on `from m import f as g` resolve as calls on a module —
* lsp=MISS and the CALLS edge was lost. */
ASSERT_EQ(t->kind, CBM_TYPE_NAMED);
ASSERT_STR_EQ(t->data.named.qualified_name, "pathlib.Path");
cbm_arena_destroy(&a);
PASS();
}
TEST(pylsp_import_relative_one_dot) {
/* from . import sibling — extract_imports records local=sibling,
* qn="..sibling" or similar. py_lsp binds it regardless. */
CBMArena a;
cbm_arena_init(&a);
CBMTypeRegistry reg;
cbm_registry_init(&reg, &a);
PyLSPContext ctx;
const char *locals[] = {"sibling"};
const char *qns[] = {"..sibling"};
bind_imports_into_ctx(&ctx, &a, &reg, locals, qns, 1);
const CBMType *t = py_lsp_lookup_in_scope(&ctx, "sibling");
ASSERT(!cbm_type_is_unknown(t));
cbm_arena_destroy(&a);
PASS();
}
TEST(pylsp_import_relative_two_dots) {
/* from ..pkg import x → bind x as NAMED("..pkg.x") best effort */
CBMArena a;
cbm_arena_init(&a);
CBMTypeRegistry reg;
cbm_registry_init(&reg, &a);
PyLSPContext ctx;
const char *locals[] = {"x"};
const char *qns[] = {"...pkg.x"};
bind_imports_into_ctx(&ctx, &a, &reg, locals, qns, 1);
const CBMType *t = py_lsp_lookup_in_scope(&ctx, "x");
ASSERT(!cbm_type_is_unknown(t));
cbm_arena_destroy(&a);
PASS();
}
TEST(pylsp_import_star_best_effort) {
/* from X import * — local_name="*". py_lsp does not bind "*" because
* it's not a usable identifier; the import is preserved in the import
* map for cross-file re-export resolution (Phase 9). */
CBMArena a;
cbm_arena_init(&a);
CBMTypeRegistry reg;
cbm_registry_init(&reg, &a);
PyLSPContext ctx;
const char *locals[] = {"*"};
const char *qns[] = {"X"};
bind_imports_into_ctx(&ctx, &a, &reg, locals, qns, 1);
const CBMType *star_miss = py_lsp_lookup_in_scope(&ctx, "*");
ASSERT(cbm_type_is_unknown(star_miss));
/* Import is still recorded — Phase 9 will use it. */
ASSERT_EQ(ctx.import_count, 1);
cbm_arena_destroy(&a);
PASS();
}
TEST(pylsp_import_typing_only_still_binds) {
/* `if TYPE_CHECKING:` is just a runtime constant — extract_imports
* emits CBMImport entries regardless of guard. py_lsp binds them. */
CBMArena a;
cbm_arena_init(&a);
CBMTypeRegistry reg;
cbm_registry_init(&reg, &a);
PyLSPContext ctx;
const char *locals[] = {"List"};
const char *qns[] = {"typing.List"};
bind_imports_into_ctx(&ctx, &a, &reg, locals, qns, 1);
const CBMType *t = py_lsp_lookup_in_scope(&ctx, "List");
ASSERT(!cbm_type_is_unknown(t));
ASSERT_EQ(t->kind, CBM_TYPE_NAMED);
cbm_arena_destroy(&a);
PASS();
}
TEST(pylsp_import_multi_pass_through_extract_file) {
/* End-to-end: extract_file + run_py_lsp populate scope via imports.
* We can't peek into the embedded ctx, but we verify imports survive
* to the result and bind correctly when re-traversed. */
CBMFileResult *r = extract_py(
"import os\n"
"import json as j\n"
"from pathlib import Path\n"
"def use():\n"
" return Path('.')\n");
ASSERT_NOT_NULL(r);
ASSERT_GTE(r->imports.count, 3);
cbm_free_result(r);
PASS();
}
/* ── Phase 4-6 — direct calls + method dispatch ────────────────── */
TEST(pylsp_direct_function_call) {
/* def helper(): return 1
* def main(): return helper() */
CBMFileResult *r = extract_py(
"def helper():\n"
" return 1\n"
"def main():\n"
" return helper()\n");
ASSERT_NOT_NULL(r);
ASSERT_GTE(require_resolved(r, "main", "helper"), 0);
cbm_free_result(r);
PASS();
}
TEST(pylsp_method_call_simple) {
/* class C:
* def m(self): return 1
* def use(c):
* c.m() -- with annotation */
CBMFileResult *r = extract_py(
"class C:\n"
" def m(self):\n"
" return 1\n"
"def use(c: C):\n"
" return c.m()\n");
ASSERT_NOT_NULL(r);
ASSERT_GTE(require_resolved(r, "use", "m"), 0);
cbm_free_result(r);
PASS();
}
TEST(pylsp_method_via_self) {
CBMFileResult *r = extract_py(
"class C:\n"
" def helper(self):\n"
" return 1\n"
" def caller(self):\n"
" return self.helper()\n");
ASSERT_NOT_NULL(r);
ASSERT_GTE(require_resolved(r, "caller", "helper"), 0);
cbm_free_result(r);
PASS();
}
TEST(pylsp_constructor_call_returns_instance) {
/* class Foo: ...
* def use():
* f = Foo()
* f.method() -- requires inferring f as Foo */
CBMFileResult *r = extract_py(
"class Foo:\n"
" def method(self):\n"
" return 1\n"
"def use():\n"
" f = Foo()\n"
" return f.method()\n");
ASSERT_NOT_NULL(r);
ASSERT_GTE(require_resolved(r, "use", "method"), 0);
cbm_free_result(r);
PASS();
}
/* Issue #1642: Embedder(...) must resolve to Embedder.__init__, and the
* textual callee leaf "Embedder" must still join that row. */
TEST(pylsp_constructor_call_targets_init) {
const char *source = "class Embedder:\n"
" def __init__(self, model=\"x\"):\n"
" self.model = model\n"
"def build():\n"
" return Embedder(model=\"y\")\n";
CBMFileResult *r = extract_py(source);
ASSERT_NOT_NULL(r);
int idx = require_resolved(r, "build", "Embedder.__init__");
ASSERT_GTE(idx, 0);
if (idx >= 0) {
const CBMResolvedCall *rc = &r->resolved_calls.items[idx];
ASSERT_STR_EQ(rc->strategy, "lsp_constructor_init");
ASSERT_STR_EQ(rc->reason, "Embedder");
ASSERT(strncmp(rc->callee_qn, "builtins.", 9) != 0);
}
for (int i = 0; i < r->resolved_calls.count; i++) {
const CBMResolvedCall *rc = &r->resolved_calls.items[i];
if (rc->kind != CBM_RESOLVED_INVOCATION || !rc->caller_qn || !rc->callee_qn ||
!strstr(rc->caller_qn, "build"))
continue;
/* One site, one target: the class node no longer receives the call. */
ASSERT(strstr(rc->callee_qn, ".__init__") != NULL);
ASSERT(strncmp(rc->callee_qn, "builtins.", 9) != 0);
}
const CBMCall *carrier = NULL;
for (int i = 0; i < r->calls.count; i++) {
const CBMCall *call = &r->calls.items[i];
if (call->callee_name && strcmp(call->callee_name, "Embedder") == 0 &&
call->enclosing_func_qn && strstr(call->enclosing_func_qn, "build")) {
carrier = call;
break;
}
}
ASSERT_NOT_NULL(carrier);
const CBMResolvedCall *joined =
cbm_pipeline_find_lsp_resolution(&r->resolved_calls, carrier, false);
ASSERT_NOT_NULL(joined);
ASSERT(strstr(joined->callee_qn, "Embedder.__init__") != NULL);
ASSERT_STR_EQ(joined->strategy, "lsp_constructor_init");
cbm_free_result(r);
PASS();
}
/* Same-file subclasses: only an __init__ the class declares itself is a
* construction target. B(A) inherits A.__init__ but declares none, so B()
* stays on the class B (never A.__init__). C(A) declares its own __init__,
* so C() resolves to C.__init__ (never the inherited A.__init__). */
static int count_build_rows(const CBMFileResult *r, const char *caller_leaf,
const char *callee_qn, const char *strategy) {
int n = 0;
for (int i = 0; i < r->resolved_calls.count; i++) {
const CBMResolvedCall *rc = &r->resolved_calls.items[i];
if (rc->kind != CBM_RESOLVED_INVOCATION || !rc->caller_qn || !rc->callee_qn ||
!rc->strategy)
continue;
const char *dot = strrchr(rc->caller_qn, '.');
const char *leaf = dot ? dot + 1 : rc->caller_qn;
if (strcmp(leaf, caller_leaf) == 0 && strcmp(rc->callee_qn, callee_qn) == 0 &&
strcmp(rc->strategy, strategy) == 0)
n++;
}
return n;
}
TEST(pylsp_constructor_same_file_subclass_own_init_only) {
CBMFileResult *r = extract_py("class A:\n"
" def __init__(self):\n"
" self.ready = True\n"
"class B(A):\n"
" def go(self):\n"
" return 1\n"
"class C(A):\n"
" def __init__(self):\n"
" self.ready = False\n"
"def build_b():\n"
" return B()\n"
"def build_c():\n"
" return C()\n");
ASSERT_NOT_NULL(r);
/* B() keeps the class target; the inherited A.__init__ gets no row. */
ASSERT_EQ(count_build_rows(r, "build_b", "test.main.B", "lsp_constructor"), 1);
ASSERT_EQ(find_resolved(r, "build_b", "__init__"), -1);
/* C() binds to the __init__ C declares itself. */
int idx = require_resolved(r, "build_c", "test.main.C.__init__");
ASSERT_GTE(idx, 0);
if (idx >= 0) {
const CBMResolvedCall *rc = &r->resolved_calls.items[idx];
ASSERT_STR_EQ(rc->strategy, "lsp_constructor_init");
ASSERT_STR_EQ(rc->reason, "C");
}
ASSERT_EQ(find_resolved(r, "build_c", "test.main.A.__init__"), -1);
ASSERT_EQ(count_build_rows(r, "build_c", "test.main.C", "lsp_constructor"), 0);
cbm_free_result(r);
PASS();
}
/* No user __init__ (and no inherited one): the call stays on the class. */
TEST(pylsp_constructor_without_init_stays_on_class) {
CBMFileResult *r = extract_py("class Bare:\n"
" def method(self):\n"
" return 1\n"
"def build():\n"
" return Bare()\n");
ASSERT_NOT_NULL(r);
ASSERT_EQ(find_resolved(r, "build", "__init__"), -1);
int idx = -1;
for (int i = 0; i < r->resolved_calls.count; i++) {
const CBMResolvedCall *rc = &r->resolved_calls.items[i];
if (rc->kind == CBM_RESOLVED_INVOCATION && rc->caller_qn &&
strstr(rc->caller_qn, "build") && rc->callee_qn && strstr(rc->callee_qn, "Bare") &&
!strstr(rc->callee_qn, "__init__")) {
idx = i;
break;
}
}
ASSERT_GTE(idx, 0);
if (idx >= 0)
ASSERT_STR_EQ(r->resolved_calls.items[idx].strategy, "lsp_constructor");
cbm_free_result(r);
PASS();
}
/* f = Foo(); f() is a callable instance, not a constructor. It must not
* gain an edge to Foo.__init__. Foo() itself still does. */
TEST(pylsp_callable_instance_call_skips_init) {
CBMFileResult *r = extract_py("class Foo:\n"
" def __init__(self):\n"
" pass\n"
" def __call__(self):\n"
" return 1\n"
"def use():\n"
" f = Foo()\n"
" return f()\n");
ASSERT_NOT_NULL(r);
int init_rows = 0;
int class_rows = 0;
for (int i = 0; i < r->resolved_calls.count; i++) {
const CBMResolvedCall *rc = &r->resolved_calls.items[i];
if (rc->kind != CBM_RESOLVED_INVOCATION || !rc->caller_qn || !rc->callee_qn ||
!strstr(rc->caller_qn, "use"))
continue;
if (strstr(rc->callee_qn, "__init__")) {
init_rows++;
ASSERT_STR_EQ(rc->strategy, "lsp_constructor_init");
ASSERT_STR_EQ(rc->reason, "Foo");
ASSERT(strncmp(rc->callee_qn, "builtins.", 9) != 0);
} else if (strstr(rc->callee_qn, "Foo")) {
class_rows++;
ASSERT_STR_EQ(rc->strategy, "lsp_constructor");
ASSERT_NULL(rc->reason);
}
}
ASSERT_EQ(init_rows, 1);
ASSERT_EQ(class_rows, 1);
cbm_free_result(r);
PASS();
}
/* builtins.object.__init__ is in the stdlib registry. Construction must not
* record a resolution to it. */
TEST(pylsp_constructor_skips_builtin_init) {
CBMFileResult *r = extract_py("from builtins import object\n"
"def build():\n"
" return object()\n");
ASSERT_NOT_NULL(r);
for (int i = 0; i < r->resolved_calls.count; i++) {
const CBMResolvedCall *rc = &r->resolved_calls.items[i];
if (!rc->callee_qn)
continue;
ASSERT(strncmp(rc->callee_qn, "builtins.", 9) != 0 ||
strstr(rc->callee_qn, ".__init__") == NULL);
}
int idx = find_resolved(r, "build", "builtins.object");
ASSERT_GTE(idx, 0);
if (idx >= 0) {
const CBMResolvedCall *rc = &r->resolved_calls.items[idx];
ASSERT_STR_EQ(rc->strategy, "lsp_constructor");
ASSERT(strstr(rc->callee_qn, "__init__") == NULL);
}
cbm_free_result(r);
PASS();
}
TEST(pylsp_method_via_inheritance) {
CBMFileResult *r = extract_py(
"class Base:\n"
" def shared(self):\n"
" return 1\n"
"class Child(Base):\n"
" def go(self):\n"
" return self.shared()\n");
ASSERT_NOT_NULL(r);
ASSERT_GTE(require_resolved(r, "go", "shared"), 0);
cbm_free_result(r);
PASS();
}
TEST(pylsp_no_false_positive_on_unknown_method) {
/* Calling a method on an UNKNOWN type should NOT emit a high-confidence
* resolution. */
CBMFileResult *r = extract_py(
"def f(x):\n"
" return x.something_unknown_42()\n");
ASSERT_NOT_NULL(r);
/* Should produce no high-confidence match for "something_unknown_42" */
int idx = find_resolved(r, "f", "something_unknown_42");
if (idx >= 0) {
const CBMResolvedCall *rc = &r->resolved_calls.items[idx];
ASSERT(rc->confidence < 0.6f);
}
cbm_free_result(r);
PASS();
}
/* ── Phase 7-8 — decorators, super(), multi-inheritance ──────── */
TEST(pylsp_decorated_function_resolves) {
/* Decorated functions still resolve as their bare-name target. */
CBMFileResult *r = extract_py(
"import functools\n"
"@functools.cache\n"
"def helper():\n"
" return 1\n"
"def main():\n"
" return helper()\n");
ASSERT_NOT_NULL(r);
ASSERT_GTE(require_resolved(r, "main", "helper"), 0);
cbm_free_result(r);
PASS();
}
TEST(pylsp_classmethod_resolves) {
CBMFileResult *r = extract_py(
"class C:\n"
" @classmethod\n"
" def make(cls):\n"
" return cls()\n"
"def use():\n"
" return C.make()\n");
ASSERT_NOT_NULL(r);
ASSERT_GTE(require_resolved(r, "use", "make"), 0);
cbm_free_result(r);
PASS();
}
TEST(pylsp_staticmethod_resolves) {
CBMFileResult *r = extract_py(
"class C:\n"
" @staticmethod\n"
" def add(a, b):\n"
" return a + b\n"
"def use():\n"
" return C.add(1, 2)\n");
ASSERT_NOT_NULL(r);
ASSERT_GTE(require_resolved(r, "use", "add"), 0);
cbm_free_result(r);
PASS();
}
TEST(pylsp_dataclass_constructor) {
/* @dataclass synthesizes __init__. We don't emit __init__ explicitly,
* but the constructor call should still link to the class qn. */
CBMFileResult *r = extract_py(
"from dataclasses import dataclass\n"
"@dataclass\n"
"class Point:\n"
" x: int\n"
" y: int\n"
" def magnitude(self):\n"
" return self.x + self.y\n"
"def use():\n"
" p = Point(1, 2)\n"
" return p.magnitude()\n");
ASSERT_NOT_NULL(r);
ASSERT_GTE(require_resolved(r, "use", "magnitude"), 0);
cbm_free_result(r);
PASS();
}
TEST(pylsp_super_call) {
CBMFileResult *r = extract_py(
"class Base:\n"
" def greet(self):\n"
" return 'hi'\n"
"class Child(Base):\n"
" def greet(self):\n"
" return super().greet()\n");
ASSERT_NOT_NULL(r);
/* super().greet() should resolve to Base.greet, not Child.greet. */
int idx = find_resolved(r, "greet", "greet");
ASSERT_GTE(idx, 0);
if (idx >= 0) {
const CBMResolvedCall *rc = &r->resolved_calls.items[idx];
ASSERT(strstr(rc->callee_qn, "Base") != NULL);
}
cbm_free_result(r);
PASS();
}
TEST(pylsp_multi_inheritance_first_base) {
CBMFileResult *r = extract_py(
"class A:\n"
" def a_method(self):\n"
" return 1\n"
"class B:\n"
" def b_method(self):\n"
" return 2\n"
"class C(A, B):\n"
" def use(self):\n"
" self.a_method()\n"
" self.b_method()\n");
ASSERT_NOT_NULL(r);
ASSERT_GTE(require_resolved(r, "use", "a_method"), 0);
ASSERT_GTE(require_resolved(r, "use", "b_method"), 0);
cbm_free_result(r);
PASS();
}
TEST(pylsp_pep695_generic_class) {
/* PEP 695: class Box[T]: -- our implementation ignores the [T] part */
CBMFileResult *r = extract_py(
"class Box:\n"
" def get(self):\n"
" return 1\n"
"def use(b: Box):\n"
" return b.get()\n");
ASSERT_NOT_NULL(r);
ASSERT_GTE(require_resolved(r, "use", "get"), 0);
cbm_free_result(r);
PASS();
}
/* ── Phase 9 — cross-file resolution ──────────────────────────── */
static int find_resolved_arr(const CBMResolvedCallArray *arr, const char *callerSub,
const char *calleeSub) {
for (int i = 0; i < arr->count; i++) {
const CBMResolvedCall *rc = &arr->items[i];
if (rc->caller_qn && strstr(rc->caller_qn, callerSub) &&
rc->callee_qn && strstr(rc->callee_qn, calleeSub))
return i;
}
return -1;
}
typedef struct {
int target_count;
int first_site_count;
int second_site_count;
} PyResolvedSiteCounts;
static PyResolvedSiteCounts count_invocation_sites(const CBMResolvedCallArray *arr,
const char *caller_sub, const char *callee_sub,
uint32_t first_start, uint32_t first_end,
uint32_t second_start, uint32_t second_end) {
PyResolvedSiteCounts counts = {0};
for (int i = 0; i < arr->count; i++) {
const CBMResolvedCall *rc = &arr->items[i];
if (rc->kind != CBM_RESOLVED_INVOCATION || !rc->caller_qn || !rc->callee_qn ||
!strstr(rc->caller_qn, caller_sub) || !strstr(rc->callee_qn, callee_sub)) {
continue;
}
counts.target_count++;
if (rc->site_start_byte == first_start && rc->site_end_byte == first_end)
counts.first_site_count++;
if (rc->site_start_byte == second_start && rc->site_end_byte == second_end)
counts.second_site_count++;
}
return counts;
}
TEST(pylsp_crossfile_method_dispatch) {
/* file svc.py defines class RedisStore with Get(); file main.py calls
* the method on a typed parameter. Reuses CBMLSPDef to feed the
* cross-file definition into the resolver. */
const char *source =
"from svc import RedisStore\n"
"def process(s: RedisStore):\n"
" return s.Get('k')\n";
CBMLSPDef defs[2];
memset(defs, 0, sizeof(defs));
defs[0].qualified_name = "svc.RedisStore";
defs[0].short_name = "RedisStore";
defs[0].label = "Class";
defs[0].def_module_qn = "svc";
defs[1].qualified_name = "svc.RedisStore.Get";
defs[1].short_name = "Get";
defs[1].label = "Method";
defs[1].receiver_type = "svc.RedisStore";
defs[1].def_module_qn = "svc";
const char *imp_names[] = {"RedisStore"};
const char *imp_qns[] = {"svc.RedisStore"};
CBMArena arena;
cbm_arena_init(&arena);
CBMResolvedCallArray out = {0};
cbm_run_py_lsp_cross(&arena, source, (int)strlen(source), "test.main", defs, 2, imp_names,
imp_qns, 1, NULL, &out, NULL);
ASSERT_GTE(find_resolved_arr(&out, "process", "Get"), 0);
cbm_arena_destroy(&arena);
PASS();
}
/* Issue #1277 (reproducer contributed by @Enferlain): a method call through a
* typed instance field of a class imported from ANOTHER file. The field type
* reaches the resolver only through CBMLSPDef.field_defs on the class def. */
TEST(pylsp_crossfile_receiver_through_typed_field) {
const char *source = "from trainer import Trainer\n"
"def run(trainer: Trainer):\n"
" strategies = trainer.strategies\n"
" return strategies.process_batch()\n";
CBMLSPDef defs[3];
memset(defs, 0, sizeof(defs));
defs[0].qualified_name = "contracts.Contract";
defs[0].short_name = "Contract";
defs[0].label = "Class";
defs[0].def_module_qn = "contracts";
defs[1].qualified_name = "contracts.Contract.process_batch";
defs[1].short_name = "process_batch";
defs[1].label = "Method";
defs[1].receiver_type = "contracts.Contract";
defs[1].def_module_qn = "contracts";
defs[2].qualified_name = "trainer.Trainer";
defs[2].short_name = "Trainer";
defs[2].label = "Class";
defs[2].def_module_qn = "trainer";
defs[2].field_defs = "strategies:contracts.Contract";
const char *imp_names[] = {"Trainer"};
const char *imp_qns[] = {"trainer.Trainer"};
CBMArena arena;
cbm_arena_init(&arena);
CBMResolvedCallArray out = {0};
cbm_run_py_lsp_cross(&arena, source, (int)strlen(source), "test.loop", defs, 3, imp_names,
imp_qns, 1, NULL, &out, NULL);
ASSERT_GTE(find_resolved_arr(&out, "run", "process_batch"), 0);
cbm_arena_destroy(&arena);
PASS();
}
/* #1277 on the production path: the sealed shared Tier-2 registry. Covers the
* direct chain, the local alias, a field inherited from a base class, and two
* controls: an undeclared field and a field whose type has no such method
* must stay unresolved (never guessed). */
TEST(pylsp_crossfile_typed_field_shared_registry) {
CBMArena arena;
cbm_arena_init(&arena);
CBMLSPDef defs[5];
memset(defs, 0, sizeof(defs));
defs[0].qualified_name = "contracts.Contract";
defs[0].short_name = "Contract";
defs[0].label = "Class";
defs[0].def_module_qn = "contracts";
defs[1].qualified_name = "contracts.Contract.process_batch";
defs[1].short_name = "process_batch";
defs[1].label = "Method";
defs[1].receiver_type = "contracts.Contract";
defs[1].def_module_qn = "contracts";
defs[2].qualified_name = "trainer.Trainer";
defs[2].short_name = "Trainer";
defs[2].label = "Class";
defs[2].def_module_qn = "trainer";
defs[2].field_defs = "strategies:contracts.Contract|name:builtins.str";
defs[3].qualified_name = "trainer.SubTrainer";
defs[3].short_name = "SubTrainer";
defs[3].label = "Class";
defs[3].def_module_qn = "trainer";
defs[3].embedded_types = "trainer.Trainer";
defs[4].qualified_name = "trainer.Other";
defs[4].short_name = "Other";
defs[4].label = "Class";
defs[4].def_module_qn = "trainer";
for (int i = 0; i < 5; i++) {
defs[i].lang = CBM_LANG_PYTHON;
}
CBMTypeRegistry *reg = cbm_py_build_cross_registry(&arena, defs, 5);
ASSERT_NOT_NULL(reg);
ASSERT_TRUE(reg->read_only);
const char *src = "from trainer import Trainer, SubTrainer, Other\n"
"def direct(t: Trainer):\n"
" return t.strategies.process_batch()\n"
"def alias(t: Trainer):\n"
" s = t.strategies\n"
" return s.process_batch()\n"
"def inherited(t: SubTrainer):\n"
" return t.strategies.process_batch()\n"
"def undeclared(t: Other):\n"
" return t.strategies.process_batch()\n"
"def wrong_type(t: Trainer):\n"
" return t.name.process_batch()\n";
const char *imp_names[] = {"Trainer", "SubTrainer", "Other"};
const char *imp_qns[] = {"trainer.Trainer", "trainer.SubTrainer", "trainer.Other"};
CBMResolvedCallArray out = {0};
cbm_run_py_lsp_cross_with_registry(&arena, src, (int)strlen(src), "test.loop", reg, imp_names,
imp_qns, 3, NULL, &out, NULL);
ASSERT_GTE(find_resolved_arr(&out, "direct", "Contract.process_batch"), 0);
ASSERT_GTE(find_resolved_arr(&out, "alias", "Contract.process_batch"), 0);
ASSERT_GTE(find_resolved_arr(&out, "inherited", "Contract.process_batch"), 0);
ASSERT_EQ(find_resolved_arr(&out, "undeclared", "process_batch"), -1);
ASSERT_EQ(find_resolved_arr(&out, "wrong_type", "process_batch"), -1);
cbm_arena_destroy(&arena);
PASS();
}
/* Graph-quality guard for the Python seal fix (per-file field overlay). In the
* FUSED path the shared Tier-2 registry is read_only, so `self.b = Bar()` can no
* longer be recorded by mutating the shared registry; it is recorded in the per-file
* overlay instead. This test proves the overlay preserves same-file attribute-chain
* resolution: self.b.work() must still resolve to Bar.work. It RED-fails under a
* plain read_only *gate* (option A: the field is dropped, self.b is untyped, the
* work() call goes unresolved) and GREENs with the overlay. It ALSO asserts the
* shared registry entry stays unmutated (Foo.field_names == NULL) — the seal half. */
TEST(pylsp_fused_self_attr_chain_via_overlay) {
CBMArena arena;
cbm_arena_init(&arena);
/* Sealed shared registry: class Bar with method work(), class Foo. */
CBMLSPDef defs[3];
memset(defs, 0, sizeof(defs));
defs[0].qualified_name = "test.mod.Bar";
defs[0].short_name = "Bar";
defs[0].label = "Class";
defs[0].def_module_qn = "test.mod";
defs[0].lang = CBM_LANG_PYTHON;
defs[1].qualified_name = "test.mod.Bar.work";
defs[1].short_name = "work";
defs[1].label = "Method";
defs[1].receiver_type = "test.mod.Bar";
defs[1].def_module_qn = "test.mod";
defs[1].lang = CBM_LANG_PYTHON;
defs[2].qualified_name = "test.mod.Foo";
defs[2].short_name = "Foo";
defs[2].label = "Class";
defs[2].def_module_qn = "test.mod";
defs[2].lang = CBM_LANG_PYTHON;
CBMTypeRegistry *reg = cbm_py_build_cross_registry(&arena, defs, 3);
ASSERT_NOT_NULL(reg);
ASSERT_TRUE(reg->read_only);
/* Foo.__init__ records self.b = Bar(); Foo.run dispatches self.b.work(). */
const char *src = "class Foo:\n"
" def __init__(self):\n"
" self.b = Bar()\n"
" def run(self):\n"
" return self.b.work()\n";
CBMResolvedCallArray out = {0};
cbm_run_py_lsp_cross_with_registry(&arena, src, (int)strlen(src), "test.mod", reg, NULL, NULL,
0, NULL, &out, NULL);
/* Overlay preserves the attribute-chain edge. */
ASSERT_GTE(find_resolved_arr(&out, "run", "work"), 0);
/* Seal preserved: the shared registry entry was NOT mutated. */
const CBMRegisteredType *foo = cbm_registry_lookup_type(reg, "test.mod.Foo");
ASSERT_NOT_NULL(foo);
ASSERT_TRUE(foo->field_names == NULL);
cbm_arena_destroy(&arena);
PASS();
}
/* Characterization of the shared Tier-2 build order. Defs register one at a
* time in defs[] order, so a Method seen before its Class auto-registers a
* placeholder receiver type that stays ahead of the Class entry, and a Method
* whose receiver is already known adds no type. Any change to how the build
* probes for known receivers must keep this exact sequence. */
TEST(pylsp_cross_registry_interleaved_order) {
CBMArena arena;
cbm_arena_init(&arena);
CBMTypeRegistry *base = cbm_py_build_cross_registry(&arena, NULL, 0);
ASSERT_NOT_NULL(base);
int tb = base->type_count, fb = base->func_count;
CBMLSPDef defs[] = {
{.qualified_name = "t.m.A.m",
.short_name = "m",
.label = "Method",
.receiver_type = "t.m.A",
.def_module_qn = "t.m",
.lang = CBM_LANG_PYTHON},
{.qualified_name = "t.m.A",
.short_name = "A",
.label = "Class",
.def_module_qn = "t.m",
.lang = CBM_LANG_PYTHON},
{.qualified_name = "t.m.B",
.short_name = "B",
.label = "Class",
.def_module_qn = "t.m",
.lang = CBM_LANG_PYTHON},
{.qualified_name = "t.m.B.f",
.short_name = "f",
.label = "Method",
.receiver_type = "t.m.B",
.def_module_qn = "t.m",
.lang = CBM_LANG_PYTHON},
{.qualified_name = "t.m.C.g",
.short_name = "g",
.label = "Method",
.receiver_type = "t.m.C",
.def_module_qn = "t.m",
.lang = CBM_LANG_PYTHON},
/* Receiver auto-registered by C.g just above: no second placeholder. */
{.qualified_name = "t.m.C.h",
.short_name = "h",
.label = "Method",
.receiver_type = "t.m.C",
.def_module_qn = "t.m",
.lang = CBM_LANG_PYTHON},
/* Non-Python defs are skipped by the builder. */
{.qualified_name = "t.m.D",
.short_name = "D",
.label = "Class",
.def_module_qn = "t.m",
.lang = CBM_LANG_GO},
};
CBMTypeRegistry *reg =
cbm_py_build_cross_registry(&arena, defs, (int)(sizeof(defs) / sizeof(defs[0])));
ASSERT_NOT_NULL(reg);
/* Types: placeholder A (from A.m), Class A, Class B, placeholder C. */
ASSERT_EQ(reg->type_count, tb + 4);
ASSERT_TRUE(reg->types[tb].qualified_name == defs[0].receiver_type);
ASSERT_STR_EQ(reg->types[tb].short_name, "A");
ASSERT_TRUE(reg->types[tb + 1].qualified_name == defs[1].qualified_name);
ASSERT_TRUE(reg->types[tb + 2].qualified_name == defs[2].qualified_name);
ASSERT_TRUE(reg->types[tb + 3].qualified_name == defs[4].receiver_type);
ASSERT_STR_EQ(reg->types[tb + 3].short_name, "C");
/* Funcs keep defs[] order. */
ASSERT_EQ(reg->func_count, fb + 4);
ASSERT_STR_EQ(reg->funcs[fb].qualified_name, "t.m.A.m");
ASSERT_STR_EQ(reg->funcs[fb + 1].qualified_name, "t.m.B.f");
ASSERT_STR_EQ(reg->funcs[fb + 2].qualified_name, "t.m.C.g");
ASSERT_STR_EQ(reg->funcs[fb + 2].receiver_type, "t.m.C");
ASSERT_STR_EQ(reg->funcs[fb + 3].qualified_name, "t.m.C.h");
const CBMRegisteredType *a = cbm_registry_lookup_type(reg, "t.m.A");
ASSERT_NOT_NULL(a);
ASSERT_STR_EQ(a->qualified_name, "t.m.A");
ASSERT_NULL(cbm_registry_lookup_type(reg, "t.m.D"));
cbm_arena_destroy(&arena);
PASS();
}
/* Issue #228: a class/static method invoked directly on a CROSS-FILE imported
* class name — ActionRecordX.build_from_text(...) — produced no CALLS edge, so
* the method showed in/out degree 0 and was flagged as dead code. Distinct from
* pylsp_crossfile_method_dispatch (which dispatches on a typed *instance*). */
TEST(pylsp_crossfile_classmethod_on_class_issue228) {
const char *source =
"from core.schemas import ActionRecordX\n"
"def run_plain_flow():\n"
" return ActionRecordX.build_from_text('hello')\n";
CBMLSPDef defs[2];
memset(defs, 0, sizeof(defs));
defs[0].qualified_name = "core.schemas.ActionRecordX";
defs[0].short_name = "ActionRecordX";
defs[0].label = "Class";
defs[0].def_module_qn = "core.schemas";
defs[1].qualified_name = "core.schemas.ActionRecordX.build_from_text";
defs[1].short_name = "build_from_text";
defs[1].label = "Method";
defs[1].receiver_type = "core.schemas.ActionRecordX";
defs[1].def_module_qn = "core.schemas";
const char *imp_names[] = {"ActionRecordX"};
const char *imp_qns[] = {"core.schemas.ActionRecordX"};
CBMArena arena;
cbm_arena_init(&arena);
CBMResolvedCallArray out = {0};
cbm_run_py_lsp_cross(&arena, source, (int)strlen(source), "test.main", defs, 2, imp_names,
imp_qns, 1, NULL, &out, NULL);
ASSERT_GTE(find_resolved_arr(&out, "run_plain_flow", "build_from_text"), 0);
cbm_arena_destroy(&arena);
PASS();
}
TEST(pylsp_crossfile_inheritance) {
/* svc.py defines class Base with shared(); main.py defines class Child(Base)
* and calls self.shared(). Caller passes ALL relevant defs (cross-file
* Base/shared + local Child/go) — same convention as test_go_lsp.c. */
const char *source =
"from svc import Base\n"
"class Child(Base):\n"
" def go(self):\n"
" return self.shared()\n";
CBMLSPDef defs[4];
memset(defs, 0, sizeof(defs));
defs[0].qualified_name = "svc.Base";
defs[0].short_name = "Base";
defs[0].label = "Class";
defs[0].def_module_qn = "svc";
defs[1].qualified_name = "svc.Base.shared";
defs[1].short_name = "shared";
defs[1].label = "Method";
defs[1].receiver_type = "svc.Base";
defs[1].def_module_qn = "svc";
defs[2].qualified_name = "test.main.Child";
defs[2].short_name = "Child";
defs[2].label = "Class";
defs[2].def_module_qn = "test.main";
defs[2].embedded_types = "svc.Base";
defs[3].qualified_name = "test.main.Child.go";
defs[3].short_name = "go";
defs[3].label = "Method";
defs[3].receiver_type = "test.main.Child";
defs[3].def_module_qn = "test.main";
const char *imp_names[] = {"Base"};
const char *imp_qns[] = {"svc.Base"};
CBMArena arena;
cbm_arena_init(&arena);
CBMResolvedCallArray out = {0};
cbm_run_py_lsp_cross(&arena, source, (int)strlen(source), "test.main", defs, 4, imp_names,
imp_qns, 1, NULL, &out, NULL);
ASSERT_GTE(find_resolved_arr(&out, "go", "shared"), 0);
cbm_arena_destroy(&arena);
PASS();
}
/* Imported ClassName() resolves to that class's __init__. Bare names are
* module-qualified during per-file registration, so the imported class has
* to come from the cross-file registry. */
TEST(pylsp_crossfile_constructor_targets_init) {
const char *source = "from models import Embedder\n"
"def build():\n"
" return Embedder(model=\"y\")\n";
CBMLSPDef defs[2];
memset(defs, 0, sizeof(defs));
defs[0].qualified_name = "models.Embedder";
defs[0].short_name = "Embedder";
defs[0].label = "Class";
defs[0].def_module_qn = "models";
defs[0].lang = CBM_LANG_PYTHON;
defs[1].qualified_name = "models.Embedder.__init__";
defs[1].short_name = "__init__";
defs[1].label = "Method";
defs[1].receiver_type = "models.Embedder";
defs[1].def_module_qn = "models";
defs[1].lang = CBM_LANG_PYTHON;
const char *imp_names[] = {"Embedder"};
const char *imp_qns[] = {"models.Embedder"};
CBMArena arena;
cbm_arena_init(&arena);
CBMTypeRegistry *reg = cbm_py_build_cross_registry(&arena, defs, 2);
ASSERT_NOT_NULL(reg);
CBMResolvedCallArray out = {0};
cbm_run_py_lsp_cross_with_registry(&arena, source, (int)strlen(source), "test.main", reg,
imp_names, imp_qns, 1, NULL, &out, NULL);
int idx = find_resolved_arr(&out, "build", "models.Embedder.__init__");
ASSERT_GTE(idx, 0);
if (idx >= 0) {
ASSERT_STR_EQ(out.items[idx].strategy, "lsp_constructor_init");
ASSERT_STR_EQ(out.items[idx].reason, "Embedder");
ASSERT(strncmp(out.items[idx].callee_qn, "builtins.", 9) != 0);
}
cbm_arena_destroy(&arena);
PASS();
}
/* class B(A) where A's __init__ lives in another module and B declares no
* __init__ of its own. Construction binds only to an __init__ the class
* declares itself, so B() stays on the class B and the inherited
* base_mod.A.__init__ gets no row. */
TEST(pylsp_crossfile_inherited_init_stays_on_class) {
const char *source = "from base_mod import A\n"
"class B(A):\n"
" def go(self):\n"
" return 1\n"
"def build():\n"
" return B()\n";
CBMLSPDef defs[3];
memset(defs, 0, sizeof(defs));
defs[0].qualified_name = "base_mod.A";
defs[0].short_name = "A";
defs[0].label = "Class";
defs[0].def_module_qn = "base_mod";
defs[0].lang = CBM_LANG_PYTHON;
defs[1].qualified_name = "base_mod.A.__init__";
defs[1].short_name = "__init__";
defs[1].label = "Method";
defs[1].receiver_type = "base_mod.A";
defs[1].def_module_qn = "base_mod";
defs[1].lang = CBM_LANG_PYTHON;
defs[2].qualified_name = "test.main.B";
defs[2].short_name = "B";
defs[2].label = "Class";
defs[2].def_module_qn = "test.main";
defs[2].embedded_types = "base_mod.A";
defs[2].lang = CBM_LANG_PYTHON;
const char *imp_names[] = {"A"};
const char *imp_qns[] = {"base_mod.A"};
CBMArena arena;
cbm_arena_init(&arena);
CBMTypeRegistry *reg = cbm_py_build_cross_registry(&arena, defs, 3);
ASSERT_NOT_NULL(reg);
CBMResolvedCallArray out = {0};
cbm_run_py_lsp_cross_with_registry(&arena, source, (int)strlen(source), "test.main", reg,
imp_names, imp_qns, 1, NULL, &out, NULL);
ASSERT_EQ(find_resolved_arr(&out, "build", "__init__"), -1);
int idx = find_resolved_arr(&out, "build", "test.main.B");
ASSERT_GTE(idx, 0);
if (idx >= 0) {
ASSERT_STR_EQ(out.items[idx].callee_qn, "test.main.B");
ASSERT_STR_EQ(out.items[idx].strategy, "lsp_constructor");
}
cbm_arena_destroy(&arena);
PASS();
}
TEST(pylsp_batch_two_files) {
const char *src_a =
"def helper():\n"
" return 1\n";
const char *src_b =
"from a import helper\n"
"def main():\n"
" return helper()\n";
CBMLSPDef a_defs[1];
memset(a_defs, 0, sizeof(a_defs));
a_defs[0].qualified_name = "a.helper";
a_defs[0].short_name = "helper";
a_defs[0].label = "Function";
a_defs[0].def_module_qn = "a";
CBMLSPDef b_defs[1];
memset(b_defs, 0, sizeof(b_defs));
b_defs[0].qualified_name = "b.main";
b_defs[0].short_name = "main";
b_defs[0].label = "Function";
b_defs[0].def_module_qn = "b";
const char *b_imp_names[] = {"helper"};
const char *b_imp_qns[] = {"a.helper"};
CBMBatchPyLSPFile files[2];
memset(files, 0, sizeof(files));
files[0].source = src_a;
files[0].source_len = (int)strlen(src_a);
files[0].module_qn = "a";
files[0].defs = a_defs;
files[0].def_count = 1;
files[1].source = src_b;
files[1].source_len = (int)strlen(src_b);
files[1].module_qn = "b";
files[1].defs = b_defs;
files[1].def_count = 1;
/* b imports helper from a — also include a's def in b's reachable set. */
CBMLSPDef b_combined[2];
memcpy(&b_combined[0], &a_defs[0], sizeof(CBMLSPDef));
memcpy(&b_combined[1], &b_defs[0], sizeof(CBMLSPDef));
files[1].defs = b_combined;
files[1].def_count = 2;
files[1].import_names = b_imp_names;
files[1].import_qns = b_imp_qns;
files[1].import_count = 1;
CBMArena arena;
cbm_arena_init(&arena);
CBMResolvedCallArray outs[2];
memset(outs, 0, sizeof(outs));
cbm_batch_py_lsp_cross(&arena, files, 2, outs);
/* file b's main should have called helper. */
ASSERT_GTE(find_resolved_arr(&outs[1], "main", "helper"), 0);
cbm_arena_destroy(&arena);
PASS();
}
static int pylsp_exact_reference_count(const CBMResolvedCallArray *out, const char *caller,
const char *callee_qn, const char *reason,
uint32_t site_start, uint32_t site_end) {
int count = 0;
for (int i = 0; out && i < out->count; i++) {
const CBMResolvedCall *resolved = &out->items[i];
if (resolved->kind == CBM_RESOLVED_CALL_REFERENCE && resolved->caller_qn &&
strstr(resolved->caller_qn, caller) && resolved->callee_qn &&
strcmp(resolved->callee_qn, callee_qn) == 0 &&
resolved->confidence >= CBM_LSP_CONFIDENCE_FLOOR &&
resolved->site_start_byte == site_start && resolved->site_end_byte == site_end &&
(!reason || (resolved->reason && strcmp(resolved->reason, reason) == 0))) {
count++;
}
}
return count;
}
typedef struct {
int first_target;
int second_target;
} PyImportReferenceCounts;
static PyImportReferenceCounts pylsp_import_reference_counts(
const char *source, const char *local_name, const char *import_qn, const char *first_target_qn,
const char *second_target_qn) {
PyImportReferenceCounts counts = {0};
const char *call = strstr(source, " accept(");
if (!call)
return counts;
uint32_t site_start = (uint32_t)(call - source) + (uint32_t)strlen(" accept(");
uint32_t site_end = site_start + (uint32_t)strlen(local_name);
CBMLSPDef defs[4];
memset(defs, 0, sizeof(defs));
defs[0].qualified_name = first_target_qn;
defs[0].short_name = cbm_lsp_bare_segment(first_target_qn);
defs[0].label = "Function";
defs[0].def_module_qn = "project.target";
int def_count = 1;
if (second_target_qn) {
defs[1].qualified_name = second_target_qn;
defs[1].short_name = cbm_lsp_bare_segment(second_target_qn);
defs[1].label = "Function";
defs[1].def_module_qn = "project.target";
def_count++;
}
defs[def_count].qualified_name = "project.use.accept";
defs[def_count].short_name = "accept";
defs[def_count].label = "Function";
defs[def_count].def_module_qn = "project.use";
def_count++;
defs[def_count].qualified_name = "project.use.crossArgument";
defs[def_count].short_name = "crossArgument";
defs[def_count].label = "Function";
defs[def_count].def_module_qn = "project.use";
def_count++;
const char *import_names[] = {local_name};
const char *import_qns[] = {import_qn};
CBMBatchPyLSPFile file = {0};
file.source = source;
file.source_len = (int)strlen(source);
file.module_qn = "project.use";
file.defs = defs;
file.def_count = def_count;
file.import_names = import_names;
file.import_qns = import_qns;
file.import_count = 1;
CBMArena arena;
cbm_arena_init(&arena);
CBMResolvedCallArray out = {0};
cbm_batch_py_lsp_cross(&arena, &file, 1, &out);
counts.first_target = pylsp_exact_reference_count(
&out, "crossArgument", first_target_qn, local_name, site_start, site_end);
if (second_target_qn) {
counts.second_target = pylsp_exact_reference_count(
&out, "crossArgument", second_target_qn, local_name, site_start, site_end);
}
cbm_arena_destroy(&arena);
return counts;
}
TEST(pylsp_from_import_alias_equal_module_leaf_targets_imported_member) {
static const char source[] = "from target import handler as target\n"
"def accept(fn):\n"
" pass\n"
"def crossArgument():\n"
" accept(target)\n";
PyImportReferenceCounts counts = pylsp_import_reference_counts(
source, "target", "project.target", "project.target.handler", "project.target.target");
ASSERT_EQ(counts.first_target, 1);
ASSERT_EQ(counts.second_target, 0);
PASS();
}
TEST(pylsp_project_prefixed_direct_alias_same_tail_is_not_from_import_reference) {
static const char source[] = "import target.handler as handler\n"
"def accept(fn):\n"
" pass\n"
"def crossArgument():\n"
" accept(handler)\n";
PyImportReferenceCounts counts = pylsp_import_reference_counts(
source, "handler", "project.target.handler", "project.target.handler", NULL);
ASSERT_EQ(counts.first_target, 0);
PASS();
}
TEST(pylsp_competing_import_forms_for_same_local_fail_closed) {
static const char source[] = "from target import handler as callback\n"
"import target.handler as callback\n"
"def accept(fn):\n"
" pass\n"
"def crossArgument():\n"
" accept(callback)\n";
PyImportReferenceCounts counts = pylsp_import_reference_counts(
source, "callback", "project.target", "project.target.handler", NULL);
ASSERT_EQ(counts.first_target, 0);
PASS();
}
TEST(pylsp_later_plain_import_rebinding_same_local_fails_closed) {
static const char source[] = "from target import handler as target\n"
"import target\n"
"def accept(fn):\n"
" pass\n"
"def crossArgument():\n"
" accept(target)\n";
PyImportReferenceCounts counts = pylsp_import_reference_counts(
source, "target", "project.target.handler", "project.target.handler", NULL);
ASSERT_EQ(counts.first_target, 0);
PASS();
}
TEST(pylsp_top_level_function_shadows_imported_callable) {
static const char source[] = "from target import handler as callback\n"
"def callback():\n"
" pass\n"
"def accept(fn):\n"
" pass\n"
"def crossArgument():\n"
" accept(callback)\n";
PyImportReferenceCounts counts = pylsp_import_reference_counts(
source, "callback", "project.target.handler", "project.target.handler", NULL);
ASSERT_EQ(counts.first_target, 0);
PASS();
}
TEST(pylsp_decorated_top_level_function_shadows_imported_callable) {
static const char source[] = "from target import handler as callback\n"
"def replace(fn):\n"
" return object()\n"
"@replace\n"
"def callback():\n"
" pass\n"
"def accept(fn):\n"
" pass\n"
"def crossArgument():\n"
" accept(callback)\n";
PyImportReferenceCounts counts = pylsp_import_reference_counts(
source, "callback", "project.target.handler", "project.target.handler", NULL);
ASSERT_EQ(counts.first_target, 0);
PASS();
}
TEST(pylsp_later_import_rebinds_top_level_function_name) {
static const char source[] = "def callback():\n"
" pass\n"
"from target import handler as callback\n"
"def accept(fn):\n"
" pass\n"
"def crossArgument():\n"
" accept(callback)\n";
PyImportReferenceCounts counts = pylsp_import_reference_counts(
source, "callback", "project.target.handler", "project.target.handler", NULL);
ASSERT_EQ(counts.first_target, 1);
PASS();
}
TEST(pylsp_batch_cross_file_callable_value_preserves_exact_reference_site) {
static const char source[] = "from target import handler\n"
"def accept(callback):\n"
" pass\n"
"def crossArgument():\n"
" accept(handler)\n";
const char *argument = strstr(source, "accept(handler)");
ASSERT_NOT_NULL(argument);
uint32_t site_start = (uint32_t)(argument - source) + 7U;
uint32_t site_end = site_start + (uint32_t)strlen("handler");
CBMLSPDef defs[3];
memset(defs, 0, sizeof(defs));
defs[0].qualified_name = "target.handler";
defs[0].short_name = "handler";
defs[0].label = "Function";
defs[0].def_module_qn = "target";
defs[1].qualified_name = "use.accept";
defs[1].short_name = "accept";
defs[1].label = "Function";
defs[1].def_module_qn = "use";
defs[2].qualified_name = "use.crossArgument";
defs[2].short_name = "crossArgument";
defs[2].label = "Function";
defs[2].def_module_qn = "use";
const char *import_names[] = {"handler"};
const char *import_qns[] = {"target.handler"};
CBMBatchPyLSPFile file;
memset(&file, 0, sizeof(file));
file.source = source;
file.source_len = (int)strlen(source);
file.module_qn = "use";
file.defs = defs;
file.def_count = 3;
file.import_names = import_names;
file.import_qns = import_qns;
file.import_count = 1;
CBMArena arena;
cbm_arena_init(&arena);
CBMResolvedCallArray out = {0};
cbm_batch_py_lsp_cross(&arena, &file, 1, &out);
int exact = pylsp_exact_reference_count(&out, "crossArgument", "target.handler", NULL,
site_start, site_end);
cbm_arena_destroy(&arena);
ASSERT_EQ(exact, 1);
static const char alias_source[] = "from target import handler as callback\n"
"def accept(fn):\n"
" pass\n"
"def crossArgument():\n"
" accept(callback)\n";
const char *alias_argument = strstr(alias_source, "accept(callback)");
ASSERT_NOT_NULL(alias_argument);
site_start = (uint32_t)(alias_argument - alias_source) + 7U;
site_end = site_start + (uint32_t)strlen("callback");
const char *alias_import_names[] = {"callback"};
const char *alias_import_qns[] = {"project.target"};
defs[0].qualified_name = "project.target.handler";
defs[0].def_module_qn = "project.target";
defs[1].qualified_name = "project.use.accept";
defs[1].def_module_qn = "project.use";
defs[2].qualified_name = "project.use.crossArgument";
defs[2].def_module_qn = "project.use";
file.source = alias_source;
file.source_len = (int)strlen(alias_source);
file.module_qn = "project.use";
file.import_names = alias_import_names;
file.import_qns = alias_import_qns;
memset(&out, 0, sizeof(out));
cbm_arena_init(&arena);
cbm_batch_py_lsp_cross(&arena, &file, 1, &out);
int alias_exact = pylsp_exact_reference_count(&out, "crossArgument",
"project.target.handler", "callback",
site_start, site_end);
cbm_arena_destroy(&arena);
ASSERT_EQ(alias_exact, 1);
/* The module and imported member may have the same leaf. Matching only
* leaves would keep `project.handler` and lose the member; the parsed
* from-import must canonicalize it to `project.handler.handler`. */
static const char same_name_source[] = "from handler import handler as callback\n"
"def accept(fn):\n"
" pass\n"
"def crossArgument():\n"
" accept(callback)\n";
const char *same_name_argument = strstr(same_name_source, "accept(callback)");
ASSERT_NOT_NULL(same_name_argument);
site_start = (uint32_t)(same_name_argument - same_name_source) + 7U;
site_end = site_start + (uint32_t)strlen("callback");
const char *same_name_import_qns[] = {"project.handler"};
defs[0].qualified_name = "project.handler.handler";
defs[0].def_module_qn = "project.handler";
file.source = same_name_source;
file.source_len = (int)strlen(same_name_source);
file.import_qns = same_name_import_qns;
memset(&out, 0, sizeof(out));
cbm_arena_init(&arena);
cbm_batch_py_lsp_cross(&arena, &file, 1, &out);
int same_name_exact = pylsp_exact_reference_count(
&out, "crossArgument", "project.handler.handler", "callback", site_start, site_end);
cbm_arena_destroy(&arena);
ASSERT_EQ(same_name_exact, 1);
/* Identical extraction metadata can also come from a direct module alias.
* Its import_statement must never be upgraded to the same-named function. */
static const char module_alias_source[] = "import target.handler as callback\n"
"def accept(fn):\n"
" pass\n"
"def crossArgument():\n"
" accept(callback)\n";
const char *module_alias_argument = strstr(module_alias_source, "accept(callback)");
ASSERT_NOT_NULL(module_alias_argument);
site_start = (uint32_t)(module_alias_argument - module_alias_source) + 7U;
site_end = site_start + (uint32_t)strlen("callback");
defs[0].qualified_name = "project.target.handler";
defs[0].def_module_qn = "project.target";
file.source = module_alias_source;
file.source_len = (int)strlen(module_alias_source);
file.import_qns = alias_import_qns;
memset(&out, 0, sizeof(out));
cbm_arena_init(&arena);
cbm_batch_py_lsp_cross(&arena, &file, 1, &out);
int module_alias_reference = pylsp_exact_reference_count(
&out, "crossArgument", "project.target.handler", "callback", site_start, site_end);
cbm_arena_destroy(&arena);
ASSERT_EQ(module_alias_reference, 0);
/* Syntax alone is not callable proof: decorators can replace the runtime
* binding with any value. Canonicalization must still pass through the
* registry's exact-callable eligibility gate. */
const char *decorators[] = {"@replace_with_value", NULL};
defs[0].decorators = decorators;
file.source = alias_source;
file.source_len = (int)strlen(alias_source);
memset(&out, 0, sizeof(out));
cbm_arena_init(&arena);
cbm_batch_py_lsp_cross(&arena, &file, 1, &out);
int decorated_reference = pylsp_exact_reference_count(
&out, "crossArgument", "project.target.handler", "callback",
(uint32_t)(alias_argument - alias_source) + 7U,
(uint32_t)(alias_argument - alias_source) + 7U + (uint32_t)strlen("callback"));
cbm_arena_destroy(&arena);
ASSERT_EQ(decorated_reference, 0);
PASS();
}
/* A semantic invocation is an occurrence, not merely a caller/callee pair.
* Calling one target twice in one function must therefore retain two records,
* each stamped with the full `target(...)` expression that caused it. */
TEST(pylsp_same_target_calls_have_distinct_exact_semantic_sites) {
static const char source[] = "def target(value):\n"
" return value\n"
"def caller():\n"
" first = target(1)\n"
" second = target(22)\n"
" return first + second\n";
static const char first_text[] = "target(1)";
static const char second_text[] = "target(22)";
const char *first = strstr(source, first_text);
const char *second = strstr(source, second_text);
ASSERT_NOT_NULL(first);
ASSERT_NOT_NULL(second);
CBMFileResult *r = extract_py(source);
ASSERT_NOT_NULL(r);
uint32_t first_start = (uint32_t)(first - source);
uint32_t second_start = (uint32_t)(second - source);
PyResolvedSiteCounts counts =
count_invocation_sites(&r->resolved_calls, "caller", ".target", first_start,
first_start + (uint32_t)strlen(first_text), second_start,
second_start + (uint32_t)strlen(second_text));
ASSERT_EQ(counts.target_count, 2);
ASSERT_EQ(counts.first_site_count, 1);
ASSERT_EQ(counts.second_site_count, 1);
cbm_free_result(r);
PASS();
}
/* cbm_batch_py_lsp_cross resolves in a temporary per-file arena, then copies
* semantic records into the caller arena. Both occurrence spans must survive
* that copy; caller/callee-only dedup must not collapse the two calls. */
TEST(pylsp_batch_preserves_two_same_target_semantic_sites) {
static const char source[] = "def target(value):\n"
" return value\n"
"def caller():\n"
" first = target(1)\n"
" second = target(22)\n"
" return first + second\n";
static const char first_text[] = "target(1)";
static const char second_text[] = "target(22)";
const char *first = strstr(source, first_text);
const char *second = strstr(source, second_text);
ASSERT_NOT_NULL(first);
ASSERT_NOT_NULL(second);
CBMLSPDef defs[2];
memset(defs, 0, sizeof(defs));
defs[0].qualified_name = "batch.target";
defs[0].short_name = "target";
defs[0].label = "Function";
defs[0].def_module_qn = "batch";
defs[1].qualified_name = "batch.caller";
defs[1].short_name = "caller";
defs[1].label = "Function";
defs[1].def_module_qn = "batch";
CBMBatchPyLSPFile file;
memset(&file, 0, sizeof(file));
file.source = source;
file.source_len = (int)strlen(source);
file.module_qn = "batch";
file.defs = defs;
file.def_count = 2;
CBMArena arena;
cbm_arena_init(&arena);
CBMResolvedCallArray out = {0};
cbm_batch_py_lsp_cross(&arena, &file, 1, &out);
uint32_t first_start = (uint32_t)(first - source);
uint32_t second_start = (uint32_t)(second - source);
PyResolvedSiteCounts counts = count_invocation_sites(
&out, "caller", ".target", first_start, first_start + (uint32_t)strlen(first_text),
second_start, second_start + (uint32_t)strlen(second_text));
ASSERT_EQ(counts.target_count, 2);
ASSERT_EQ(counts.first_site_count, 1);
ASSERT_EQ(counts.second_site_count, 1);
cbm_arena_destroy(&arena);
PASS();
}
/* The fallback cross pass resolves in a scratch arena that is destroyed
* inside cbm_pxc_run_one. A Python dunder carrier created only by that cross
* resolver must be deep-copied into result.arena, remain occurrence-exact,
* and still join its semantic target after the function returns. */
TEST(pylsp_scratch_cross_dunder_carrier_survives_copy) {
static const char source[] = "class Number:\n"
" def __add__(self, other):\n"
" return self\n"
"def combine(left: Number, right: Number):\n"
" return left + right\n";
static const char site_text[] = "left + right";
CBMLSPDef defs[3];
memset(defs, 0, sizeof(defs));
defs[0].qualified_name = "scratch.Number";
defs[0].short_name = "Number";
defs[0].label = "Class";
defs[0].def_module_qn = "scratch";
defs[0].lang = CBM_LANG_PYTHON;
defs[1].qualified_name = "scratch.Number.__add__";
defs[1].short_name = "__add__";
defs[1].label = "Method";
defs[1].receiver_type = "scratch.Number";
defs[1].def_module_qn = "scratch";
defs[1].lang = CBM_LANG_PYTHON;
defs[2].qualified_name = "scratch.combine";
defs[2].short_name = "combine";
defs[2].label = "Function";
defs[2].def_module_qn = "scratch";
defs[2].lang = CBM_LANG_PYTHON;
CBMFileResult result;
memset(&result, 0, sizeof(result));
cbm_arena_init(&result.arena);
cbm_pxc_run_one(CBM_LANG_PYTHON, &result, source, (int)strlen(source), "scratch", defs, 3, NULL,
NULL, 0);
const CBMCall *carrier = NULL;
int carrier_count = 0;
for (int i = 0; i < result.calls.count; i++) {
const CBMCall *call = &result.calls.items[i];
if (call->callee_name && strcmp(call->callee_name, "__add__") == 0 &&
call->enclosing_func_qn && strstr(call->enclosing_func_qn, "combine")) {
carrier = call;
carrier_count++;
}
}
const char *site = strstr(source, site_text);
ASSERT_NOT_NULL(site);
uint32_t expected_start = (uint32_t)(site - source);
uint32_t expected_end = expected_start + (uint32_t)strlen(site_text);
ASSERT_EQ(carrier_count, 1);
ASSERT_NOT_NULL(carrier);
ASSERT_STR_EQ(carrier->callee_name, "__add__");
ASSERT_TRUE(strstr(carrier->enclosing_func_qn, "combine") != NULL);
ASSERT_TRUE(carrier->requires_lsp_resolution);
ASSERT_EQ(carrier->site_start_byte, expected_start);
ASSERT_EQ(carrier->site_end_byte, expected_end);
const CBMResolvedCall *joined =
cbm_pipeline_find_lsp_resolution(&result.resolved_calls, carrier, false);
ASSERT_NOT_NULL(joined);
ASSERT_STR_EQ(joined->callee_qn, "scratch.Number.__add__");
ASSERT_EQ(joined->site_start_byte, expected_start);
ASSERT_EQ(joined->site_end_byte, expected_end);
cbm_arena_destroy(&result.arena);
PASS();
}
/* ── Phase 10 — stdlib resolution ─────────────────────────────── */
TEST(pylsp_stdlib_os_getcwd) {
/* Top-level module attribute resolution against the stdlib registry. */
CBMFileResult *r = extract_py(
"import os\n"
"def use():\n"
" return os.getcwd()\n");
ASSERT_NOT_NULL(r);
int idx = find_resolved(r, "use", "getcwd");
ASSERT_GTE(idx, 0);
cbm_free_result(r);
PASS();
}
TEST(pylsp_stdlib_collections_defaultdict) {
CBMFileResult *r = extract_py(
"from collections import defaultdict\n"
"def use():\n"
" return defaultdict(list)\n");
ASSERT_NOT_NULL(r);
/* defaultdict(list) is a constructor — emits lsp_constructor edge */
int idx = find_resolved(r, "use", "defaultdict");
ASSERT_GTE(idx, 0);
cbm_free_result(r);
PASS();
}
TEST(pylsp_stdlib_pathlib_path_method) {
CBMFileResult *r = extract_py(
"from pathlib import Path\n"
"def use(p: Path):\n"
" return p.exists()\n");
ASSERT_NOT_NULL(r);
int idx = find_resolved(r, "use", "exists");
ASSERT_GTE(idx, 0);
cbm_free_result(r);
PASS();
}
TEST(pylsp_stdlib_logging_getlogger) {
CBMFileResult *r = extract_py(
"import logging\n"
"def use():\n"
" return logging.getLogger('app')\n");
ASSERT_NOT_NULL(r);
int idx = find_resolved(r, "use", "getLogger");
ASSERT_GTE(idx, 0);
cbm_free_result(r);
PASS();
}
/* ── Round 1 — parity push ────────────────────────────────────── */
TEST(pylsp_round1_dotted_import_walk) {
/* `import os.path` — `os` and `os.path` should both be navigable
* through attribute access so `os.path.join('a', 'b')` resolves to
* the registered os.path.join function. */
CBMFileResult *r = extract_py(
"import os.path\n"
"def use():\n"
" return os.path.join('a', 'b')\n");
ASSERT_NOT_NULL(r);
ASSERT_EQ(r->imports.count, 1);
ASSERT_STR_EQ(r->imports.items[0].local_name, "os");
ASSERT_STR_EQ(r->imports.items[0].module_path, "os.path");
int idx = find_resolved(r, "use", "join");
ASSERT_GTE(idx, 0);
cbm_free_result(r);
PASS();
}
TEST(pylsp_round1_dotted_import_alias_matching_root) {
/* The metadata pair (local=os, module=os.path) is also possible for an
* explicit alias. Unlike the unaliased form above, this binds `os`
* directly to MODULE(os.path), so os.join resolves os.path.join. */
CBMFileResult *r = extract_py("import os.path as os\n"
"def use():\n"
" return os.join('a', 'b')\n");
ASSERT_NOT_NULL(r);
ASSERT_EQ(r->imports.count, 1);
ASSERT_STR_EQ(r->imports.items[0].local_name, "os");
ASSERT_STR_EQ(r->imports.items[0].module_path, "os.path");
int idx = find_resolved(r, "use", "join");
ASSERT_GTE(idx, 0);
cbm_free_result(r);
PASS();
}
TEST(pylsp_round1_typing_cast) {
/* cast(Foo, x) returns NAMED("Foo"), enabling subsequent method
* dispatch to resolve. */
CBMFileResult *r = extract_py(
"from typing import cast\n"
"class Foo:\n"
" def m(self):\n"
" return 1\n"
"def use(x):\n"
" f = cast(Foo, x)\n"
" return f.m()\n");
ASSERT_NOT_NULL(r);
ASSERT_GTE(require_resolved(r, "use", "m"), 0);
cbm_free_result(r);
PASS();
}
TEST(pylsp_round1_assert_type_passthrough) {
/* assert_type(x, T) is a no-op at runtime; the returned value's type
* is unchanged. We type the result as type-of(x). */
CBMFileResult *r = extract_py(
"from typing import assert_type\n"
"class Foo:\n"
" def m(self):\n"
" return 1\n"
"def use(x: Foo):\n"
" f = assert_type(x, Foo)\n"
" return f.m()\n");
ASSERT_NOT_NULL(r);
ASSERT_GTE(require_resolved(r, "use", "m"), 0);
cbm_free_result(r);
PASS();
}
TEST(pylsp_round1_forward_reference) {
/* def f(x: "Foo") — quoted annotation must resolve as if unquoted. */
CBMFileResult *r = extract_py(
"class Foo:\n"
" def m(self):\n"
" return 1\n"
"def use(x: \"Foo\"):\n"
" return x.m()\n");
ASSERT_NOT_NULL(r);
ASSERT_GTE(require_resolved(r, "use", "m"), 0);
cbm_free_result(r);
PASS();
}
TEST(pylsp_round1_self_return_chains) {
/* class Builder:
* def step1(self) -> Self: return self
* def step2(self) -> Self: return self
* def build(self): return ...
* Builder().step1().step2().build() — must chain through Self. */
CBMFileResult *r = extract_py(
"from typing import Self\n"
"class Builder:\n"
" def step1(self) -> Self:\n"
" return self\n"
" def step2(self) -> Self:\n"
" return self\n"
" def build(self):\n"
" return 1\n"
"def use():\n"
" return Builder().step1().step2().build()\n");
ASSERT_NOT_NULL(r);
/* Each chain link should resolve. We assert the final .build() does. */
ASSERT_GTE(require_resolved(r, "use", "build"), 0);
cbm_free_result(r);
PASS();
}
TEST(pylsp_round1_generic_subscript_annotation) {
/* `def f(items: list[Foo])` — the generic subscript should not
* confuse annotation resolution; we drop the [Foo] part for v1. */
CBMFileResult *r = extract_py(
"from typing import Optional\n"
"class Foo:\n"
" def m(self):\n"
" return 1\n"
"def use(x: Optional[Foo]):\n"
" return x.m()\n");
ASSERT_NOT_NULL(r);
/* x has type Optional which strips to Optional, then we look up
* .m on it. This SHOULD NOT resolve in v1 since Optional is just
* Union — but it shouldn't crash either. We assert no false-positive
* high-confidence resolution against an unrelated method. */
int idx = find_resolved(r, "use", "m");
if (idx >= 0) {
const CBMResolvedCall *rc = &r->resolved_calls.items[idx];
/* If we did resolve, must be against Foo, not something garbage. */
ASSERT(strstr(rc->callee_qn, "Foo") != NULL);
}
cbm_free_result(r);
PASS();
}
/* ── Round 2 — narrowing ──────────────────────────────────────── */
TEST(pylsp_round2_isinstance_narrow) {
/* if isinstance(x, Foo): x.method() — x narrowed to Foo */
CBMFileResult *r = extract_py(
"class Foo:\n"
" def method(self):\n"
" return 1\n"
"def use(x):\n"
" if isinstance(x, Foo):\n"
" return x.method()\n");
ASSERT_NOT_NULL(r);
ASSERT_GTE(require_resolved(r, "use", "method"), 0);
cbm_free_result(r);
PASS();
}
TEST(pylsp_round2_is_not_none_narrow) {
/* def f(x: Optional[Foo]):
* if x is not None:
* x.method() */
CBMFileResult *r = extract_py(
"from typing import Optional\n"
"class Foo:\n"
" def method(self):\n"
" return 1\n"
"def use(x: Optional[Foo]):\n"
" if x is not None:\n"
" return x.method()\n");
ASSERT_NOT_NULL(r);
/* Optional strips to Foo for v1 (we drop generic args). x: Optional[Foo]
* binds x as NAMED("Optional"). After narrowing, ideally NAMED("Foo").
* Since we strip generic args, x is bound as Optional unrelated. The
* narrow extracts the non-None member from a UNION; if x is bound as a
* single type (not UNION), the narrow is a no-op. */
int idx = find_resolved(r, "use", "method");
/* We don't fail this test if narrowing doesn't help — this exercises
* the code path. The proper fix needs Optional to bind as UNION. */
(void)idx;
cbm_free_result(r);
PASS();
}
TEST(pylsp_round2_isinstance_no_false_positive_in_else) {
/* In the else branch, narrowing must NOT apply. */
CBMFileResult *r = extract_py(
"class Foo:\n"
" def method(self):\n"
" return 1\n"
"def use(x):\n"
" if isinstance(x, Foo):\n"
" return 1\n"
" else:\n"
" return x.method()\n");
ASSERT_NOT_NULL(r);
/* No high-confidence resolution should exist for x.method() in the
* else branch, since x is UNKNOWN there. */
int idx = find_resolved(r, "use", "method");
if (idx >= 0) {
const CBMResolvedCall *rc = &r->resolved_calls.items[idx];
ASSERT(rc->confidence < 0.6f);
}
cbm_free_result(r);
PASS();
}
/* ── Round 2/3 — walrus, comprehension, optional-narrow ──────── */
TEST(pylsp_round2_narrow_after_call) {
/* Without walrus: x = compute(); if x is not None: x.method().
* Tests narrow on UNION return-type-of-call. */
CBMFileResult *r = extract_py(
"from typing import Optional\n"
"class Foo:\n"
" def method(self):\n"
" return 1\n"
"def compute() -> Optional[Foo]:\n"
" return None\n"
"def use():\n"
" x = compute()\n"
" if x is not None:\n"
" return x.method()\n");
ASSERT_NOT_NULL(r);
ASSERT_GTE(require_resolved(r, "use", "method"), 0);
cbm_free_result(r);
PASS();
}
TEST(pylsp_round2_walrus_binds) {
/* if (x := compute()) is not None: x.method() */
CBMFileResult *r = extract_py(
"from typing import Optional\n"
"class Foo:\n"
" def method(self):\n"
" return 1\n"
"def compute() -> Optional[Foo]:\n"
" return None\n"
"def use():\n"
" if (x := compute()) is not None:\n"
" return x.method()\n");
ASSERT_NOT_NULL(r);
ASSERT_GTE(require_resolved(r, "use", "method"), 0);
cbm_free_result(r);
PASS();
}
TEST(pylsp_round3_listcomp_element_method) {
/* [x.method() for x in items] where items: list[Foo] */
CBMFileResult *r = extract_py(
"class Foo:\n"
" def method(self):\n"
" return 1\n"
"def use(items: list[Foo]):\n"
" return [x.method() for x in items]\n");
ASSERT_NOT_NULL(r);
ASSERT_GTE(require_resolved(r, "use", "method"), 0);
cbm_free_result(r);
PASS();
}
TEST(pylsp_round3_for_loop_element_method) {
/* for x in items: x.method() — items: list[Foo] */
CBMFileResult *r = extract_py(
"class Foo:\n"
" def method(self):\n"
" return 1\n"
"def use(items: list[Foo]):\n"
" for x in items:\n"
" x.method()\n");
ASSERT_NOT_NULL(r);
ASSERT_GTE(require_resolved(r, "use", "method"), 0);
cbm_free_result(r);
PASS();
}
TEST(pylsp_round3_optional_narrow_with_union) {
/* def f(x: Optional[Foo]):
* if x is not None: x.method() */
CBMFileResult *r = extract_py(
"from typing import Optional\n"
"class Foo:\n"
" def method(self):\n"
" return 1\n"
"def use(x: Optional[Foo]):\n"
" if x is not None:\n"
" return x.method()\n");
ASSERT_NOT_NULL(r);
ASSERT_GTE(require_resolved(r, "use", "method"), 0);
cbm_free_result(r);
PASS();
}
/* ── Round 3 — match/case + async ──────────────────────────── */
TEST(pylsp_round3_match_case_class_pattern) {
/* match x: case Foo(): subject narrows to Foo */
CBMFileResult *r = extract_py(
"class Foo:\n"
" def method(self):\n"
" return 1\n"
"class Bar:\n"
" def method(self):\n"
" return 2\n"
"def use(x):\n"
" match x:\n"
" case Foo():\n"
" return x.method()\n"
" case _:\n"
" return None\n");
ASSERT_NOT_NULL(r);
int idx = find_resolved(r, "use", "method");
ASSERT_GTE(idx, 0);
/* Should be the Foo.method binding, not Bar.method */
if (idx >= 0) {
const CBMResolvedCall *rc = &r->resolved_calls.items[idx];
ASSERT(strstr(rc->callee_qn, "Foo") != NULL);
}
cbm_free_result(r);
PASS();
}
TEST(pylsp_round3_async_await_pass_through) {
/* await expr returns expr's type. async def f() -> int registers
* with return int. await f() should resolve as int. */
CBMFileResult *r = extract_py(
"class Foo:\n"
" def method(self):\n"
" return 1\n"
"async def make() -> Foo:\n"
" return Foo()\n"
"async def use():\n"
" f = await make()\n"
" return f.method()\n");
ASSERT_NOT_NULL(r);
ASSERT_GTE(require_resolved(r, "use", "method"), 0);
cbm_free_result(r);
PASS();
}
/* ── Round 4 — instance attribute typing ──────────────────────── */
TEST(pylsp_round4_instance_attribute_init) {
/* class C:
* def __init__(self, cfg: Config):
* self.cfg = cfg # field cfg : Config
* def use(self):
* return self.cfg.display() # resolves through field type */
CBMFileResult *r = extract_py(
"class Config:\n"
" def display(self):\n"
" return 1\n"
"class App:\n"
" def __init__(self, cfg: Config):\n"
" self.cfg = cfg\n"
" def use(self):\n"
" self.cfg.display()\n");
/* Note: extra indentation simulates a body block; real test
* mirrors realistic Python code. */
ASSERT_NOT_NULL(r);
/* The above source has bad indent — replace with proper test source. */
cbm_free_result(r);
r = extract_py(
"class Config:\n"
" def display(self):\n"
" return 1\n"
"class App:\n"
" def __init__(self, cfg: Config):\n"
" self.cfg = cfg\n"
" def use(self):\n"
" return self.cfg.display()\n");
ASSERT_NOT_NULL(r);
ASSERT_GTE(require_resolved(r, "use", "display"), 0);
cbm_free_result(r);
PASS();
}
TEST(pylsp_round4_instance_attribute_class_annotation) {
/* class C:
* x: Foo
* def use(self): return self.x.method() */
CBMFileResult *r = extract_py(
"class Foo:\n"
" def method(self):\n"
" return 1\n"
"class C:\n"
" x: Foo\n"
" def use(self):\n"
" return self.x.method()\n");
ASSERT_NOT_NULL(r);
ASSERT_GTE(require_resolved(r, "use", "method"), 0);
cbm_free_result(r);
PASS();
}
/* ── Round 5 — subscript, super().__init__, operator dunders ───── */
TEST(pylsp_round5_dict_subscript_value_type) {
/* self.cache: dict[str, Foo]; self.cache[k].method() resolves. */
CBMFileResult *r = extract_py(
"class Foo:\n"
" def method(self):\n"
" return 1\n"
"class C:\n"
" def __init__(self):\n"
" self.cache: dict[str, Foo] = {}\n"
" def use(self, k):\n"
" return self.cache[k].method()\n");
ASSERT_NOT_NULL(r);
ASSERT_GTE(require_resolved(r, "use", "method"), 0);
cbm_free_result(r);
PASS();
}
TEST(pylsp_round5_list_subscript_value_type) {
CBMFileResult *r = extract_py(
"class Foo:\n"
" def method(self):\n"
" return 1\n"
"def use(items: list[Foo]):\n"
" return items[0].method()\n");
ASSERT_NOT_NULL(r);
ASSERT_GTE(require_resolved(r, "use", "method"), 0);
cbm_free_result(r);
PASS();
}
TEST(pylsp_round5_super_init) {
CBMFileResult *r = extract_py(
"class Base:\n"
" def __init__(self, root):\n"
" self.root = root\n"
"class Child(Base):\n"
" def __init__(self, root, extra):\n"
" super().__init__(root)\n"
" self.extra = extra\n");
ASSERT_NOT_NULL(r);
int idx = find_resolved(r, "__init__", "__init__");
ASSERT_GTE(idx, 0);
if (idx >= 0) {
const CBMResolvedCall *rc = &r->resolved_calls.items[idx];
ASSERT(strstr(rc->callee_qn, "Base") != NULL);
}
cbm_free_result(r);
PASS();
}
/* Synthetic dunder calls have no ordinary tree-sitter `call` carrier. Two
* different receiver classes can therefore produce the same textual leaf in
* one caller. The synthetic carrier and semantic record must retain the exact
* operator occurrence so the pipeline cannot bind both `__add__` calls to the
* first class it happens to inspect. */
TEST(pylsp_dunder_same_leaf_occurrences_join_by_exact_site) {
static const char source[] =
"class Alpha:\n"
" def __add__(self, other):\n"
" return self\n"
"class Beta:\n"
" def __add__(self, other):\n"
" return self\n"
"def combine(left_a: Alpha, left_b: Alpha, right_a: Beta, right_b: Beta):\n"
" first = left_a + left_b\n"
" second = right_a + right_b\n"
" return first, second\n";
static const char alpha_text[] = "left_a + left_b";
static const char beta_text[] = "right_a + right_b";
const char *alpha_site = strstr(source, alpha_text);
const char *beta_site = strstr(source, beta_text);
ASSERT_NOT_NULL(alpha_site);
ASSERT_NOT_NULL(beta_site);
CBMFileResult *r = extract_py(source);
ASSERT_NOT_NULL(r);
ASSERT_FALSE(r->has_error || r->parse_incomplete);
const CBMCall *dunder_calls[2] = {0};
int dunder_count = 0;
for (int i = 0; i < r->calls.count; i++) {
const CBMCall *call = &r->calls.items[i];
if (call->callee_name && strcmp(call->callee_name, "__add__") == 0 &&
call->enclosing_func_qn && strstr(call->enclosing_func_qn, "combine")) {
if (dunder_count < 2)
dunder_calls[dunder_count] = call;
dunder_count++;
}
}
ASSERT_EQ(dunder_count, 2);
ASSERT_NOT_NULL(dunder_calls[0]);
ASSERT_NOT_NULL(dunder_calls[1]);
const CBMCall *alpha_call = dunder_calls[0];
const CBMCall *beta_call = dunder_calls[1];
if (alpha_call->site_start_byte > beta_call->site_start_byte) {
const CBMCall *tmp = alpha_call;
alpha_call = beta_call;
beta_call = tmp;
}
uint32_t alpha_start = (uint32_t)(alpha_site - source);
uint32_t alpha_end = alpha_start + (uint32_t)strlen(alpha_text);
uint32_t beta_start = (uint32_t)(beta_site - source);
uint32_t beta_end = beta_start + (uint32_t)strlen(beta_text);
ASSERT_GT(alpha_call->site_end_byte, alpha_call->site_start_byte);
ASSERT_GT(beta_call->site_end_byte, beta_call->site_start_byte);
ASSERT_NEQ(alpha_call->site_start_byte, beta_call->site_start_byte);
ASSERT_EQ(alpha_call->site_start_byte, alpha_start);
ASSERT_EQ(alpha_call->site_end_byte, alpha_end);
ASSERT_EQ(beta_call->site_start_byte, beta_start);
ASSERT_EQ(beta_call->site_end_byte, beta_end);
ASSERT_TRUE(alpha_call->requires_lsp_resolution);
ASSERT_TRUE(beta_call->requires_lsp_resolution);
const CBMResolvedCall *alpha_resolution =
cbm_pipeline_find_lsp_resolution(&r->resolved_calls, alpha_call, false);
const CBMResolvedCall *beta_resolution =
cbm_pipeline_find_lsp_resolution(&r->resolved_calls, beta_call, false);
ASSERT_NOT_NULL(alpha_resolution);
ASSERT_NOT_NULL(beta_resolution);
ASSERT_TRUE(alpha_resolution != beta_resolution);
ASSERT_TRUE(strstr(alpha_resolution->callee_qn, ".Alpha.__add__") != NULL);
ASSERT_TRUE(strstr(beta_resolution->callee_qn, ".Beta.__add__") != NULL);
ASSERT_EQ(alpha_resolution->site_start_byte, alpha_call->site_start_byte);
ASSERT_EQ(alpha_resolution->site_end_byte, alpha_call->site_end_byte);
ASSERT_EQ(beta_resolution->site_start_byte, beta_call->site_start_byte);
ASSERT_EQ(beta_resolution->site_end_byte, beta_call->site_end_byte);
cbm_free_result(r);
PASS();
}
/* A named lambda body is re-walked once per invocation so its parameters can
* inherit the call-site argument types. Both walks below visit the same single
* `a + b` AST occurrence. That occurrence must produce one semantic record and
* one synthetic carrier, not one carrier per re-walk. */
TEST(pylsp_lambda_rewalk_dedupes_dunder_occurrence) {
static const char source[] = "class Number:\n"
" def __add__(self, other):\n"
" return self\n"
"def combine(left: Number, right: Number):\n"
" add = lambda a, b: a + b\n"
" first = add(left, right)\n"
" second = add(left, right)\n"
" return first, second\n";
static const char site_text[] = "a + b";
const char *site = strstr(source, site_text);
ASSERT_NOT_NULL(site);
uint32_t expected_start = (uint32_t)(site - source);
uint32_t expected_end = expected_start + (uint32_t)strlen(site_text);
CBMFileResult *r = extract_py(source);
ASSERT_NOT_NULL(r);
ASSERT_FALSE(r->has_error || r->parse_incomplete);
const CBMCall *carrier = NULL;
int carrier_count = 0;
for (int i = 0; i < r->calls.count; i++) {
const CBMCall *call = &r->calls.items[i];
if (call->callee_name && strcmp(call->callee_name, "__add__") == 0 &&
call->enclosing_func_qn && strstr(call->enclosing_func_qn, "<lambda>")) {
carrier = call;
carrier_count++;
}
}
const CBMResolvedCall *semantic = NULL;
int semantic_count = 0;
for (int i = 0; i < r->resolved_calls.count; i++) {
const CBMResolvedCall *rc = &r->resolved_calls.items[i];
if (rc->caller_qn && strstr(rc->caller_qn, "<lambda>") && rc->callee_qn &&
strstr(rc->callee_qn, ".Number.__add__")) {
semantic = rc;
semantic_count++;
}
}
ASSERT_EQ(semantic_count, 1);
ASSERT_NOT_NULL(semantic);
ASSERT_EQ(semantic->kind, CBM_RESOLVED_INVOCATION);
ASSERT_EQ(semantic->site_start_byte, expected_start);
ASSERT_EQ(semantic->site_end_byte, expected_end);
if (carrier_count != 1) {
for (int i = 0; i < r->calls.count; i++) {
const CBMCall *call = &r->calls.items[i];
if (call->callee_name && strcmp(call->callee_name, "__add__") == 0) {
fprintf(stderr, " [PY-LAMBDA-CARRIER] caller=%s site=%u:%u requires_lsp=%d\n",
call->enclosing_func_qn ? call->enclosing_func_qn : "(null)",
call->site_start_byte, call->site_end_byte,
call->requires_lsp_resolution ? 1 : 0);
}
}
}
ASSERT_EQ(carrier_count, 1);
ASSERT_NOT_NULL(carrier);
ASSERT_TRUE(carrier->requires_lsp_resolution);
ASSERT_EQ(carrier->site_start_byte, expected_start);
ASSERT_EQ(carrier->site_end_byte, expected_end);
const CBMResolvedCall *joined =
cbm_pipeline_find_lsp_resolution(&r->resolved_calls, carrier, false);
ASSERT_NOT_NULL(joined);
ASSERT_TRUE(joined == semantic);
ASSERT_TRUE(strstr(joined->callee_qn, ".Number.__add__") != NULL);
ASSERT_EQ(joined->site_start_byte, expected_start);
ASSERT_EQ(joined->site_end_byte, expected_end);
cbm_free_result(r);
PASS();
}
/* A lambda body is also re-walked for calls made inside another lambda's
* re-walk, so a lambda that reaches itself again (directly or through another
* lambda) was re-walked once per nesting level until the walk-depth cap. With
* two such calls in one body the work doubled per level, and a two-line
* `fib = lambda n: fib(n - 1) + fib(n - 2)` drove extraction to many GB. A
* lambda already being re-walked must not be re-entered, so each helper()
* site below resolves once. The recursion here is linear, so without the
* guard this stays small and fails the count instead of exhausting memory. */
TEST(pylsp_recursive_lambda_rewalk_is_bounded) {
CBMFileResult *r = extract_py("def helper():\n"
" return 0\n"
"count = lambda n: helper() if n == 0 else count(n - 1)\n"
"even = lambda n: helper() if n == 0 else odd(n - 1)\n"
"odd = lambda n: even(n - 1)\n"
"count(3)\n"
"even(4)\n");
ASSERT_NOT_NULL(r);
ASSERT_FALSE(r->has_error || r->parse_incomplete);
int helper_resolutions = 0;
for (int i = 0; i < r->resolved_calls.count; i++) {
const CBMResolvedCall *rc = &r->resolved_calls.items[i];
if (rc->callee_qn && strstr(rc->callee_qn, ".helper"))
helper_resolutions++;
}
ASSERT_EQ(helper_resolutions, 2);
cbm_free_result(r);
PASS();
}
/* The re-walk guard is released after each walk, so a later call to the same
* lambda still binds its own argument type: both A.m and B.m resolve. */
TEST(pylsp_lambda_rewalk_guard_released_between_calls) {
CBMFileResult *r = extract_py("class A:\n"
" def m(self):\n"
" return 1\n"
"class B:\n"
" def m(self):\n"
" return 2\n"
"call_m = lambda x: x.m()\n"
"call_m(A())\n"
"call_m(B())\n");
ASSERT_NOT_NULL(r);
ASSERT_GTE(require_resolved(r, "<lambda>", ".A.m"), 0);
ASSERT_GTE(require_resolved(r, "<lambda>", ".B.m"), 0);
cbm_free_result(r);
PASS();
}
/* Parser-backed Python calls have the same occurrence-identity requirement as
* synthetic dunders. Distinct typed receivers can expose the same method leaf
* in one caller; each CBMCall must join the semantic record for its own site. */
TEST(pylsp_ordinary_same_leaf_calls_join_by_exact_site) {
CBMFileResult *r = extract_py("class Alpha:\n"
" def render(self):\n"
" return 1\n"
"class Beta:\n"
" def render(self):\n"
" return 2\n"
"def combine(a: Alpha, b: Beta):\n"
" return a.render(), b.render()\n");
ASSERT_NOT_NULL(r);
const CBMCall *calls[2] = {0};
int call_count = 0;
for (int i = 0; i < r->calls.count; i++) {
const CBMCall *call = &r->calls.items[i];
if (call->enclosing_func_qn && strstr(call->enclosing_func_qn, "combine") &&
call->callee_name && strstr(call->callee_name, "render")) {
if (call_count < 2)
calls[call_count] = call;
call_count++;
}
}
if (calls[0] && calls[1] && calls[0]->site_start_byte > calls[1]->site_start_byte) {
const CBMCall *tmp = calls[0];
calls[0] = calls[1];
calls[1] = tmp;
}
const CBMResolvedCall *alpha_hit =
calls[0] ? cbm_pipeline_find_lsp_resolution(&r->resolved_calls, calls[0], false) : NULL;
const CBMResolvedCall *beta_hit =
calls[1] ? cbm_pipeline_find_lsp_resolution(&r->resolved_calls, calls[1], false) : NULL;
bool alpha_exact = alpha_hit && alpha_hit->callee_qn &&
strstr(alpha_hit->callee_qn, ".Alpha.render") &&
alpha_hit->site_start_byte == calls[0]->site_start_byte &&
alpha_hit->site_end_byte == calls[0]->site_end_byte;
bool beta_exact = beta_hit && beta_hit->callee_qn &&
strstr(beta_hit->callee_qn, ".Beta.render") &&
beta_hit->site_start_byte == calls[1]->site_start_byte &&
beta_hit->site_end_byte == calls[1]->site_end_byte;
cbm_free_result(r);
ASSERT_EQ(call_count, 2);
ASSERT_TRUE(alpha_exact);
ASSERT_TRUE(beta_exact);
PASS();
}
/* ── Round 6 — generators, dataclasses, decorator flags ───────── */
TEST(pylsp_round6_generator_yields_iterable) {
/* def gen() -> Generator[Foo, None, None]: yield Foo()
* for x in gen(): x.method() — x : Foo via element-of(generator) */
CBMFileResult *r = extract_py(
"from typing import Generator\n"
"class Foo:\n"
" def method(self):\n"
" return 1\n"
"def gen() -> Generator[Foo, None, None]:\n"
" yield Foo()\n"
"def use():\n"
" for x in gen():\n"
" x.method()\n");
ASSERT_NOT_NULL(r);
ASSERT_GTE(require_resolved(r, "use", "method"), 0);
cbm_free_result(r);
PASS();
}
TEST(pylsp_round6_dataclass_field_access) {
/* @dataclass class Point: x: Foo; def use(p: Point): p.x.method() */
CBMFileResult *r = extract_py(
"from dataclasses import dataclass\n"
"class Foo:\n"
" def method(self):\n"
" return 1\n"
"@dataclass\n"
"class Point:\n"
" x: Foo\n"
" y: int\n"
"def use(p: Point):\n"
" return p.x.method()\n");
ASSERT_NOT_NULL(r);
ASSERT_GTE(require_resolved(r, "use", "method"), 0);
cbm_free_result(r);
PASS();
}
TEST(pylsp_round6_property_access_chains) {
/* class C: @property def thing(self) -> Foo: ...
* def use(c: C): c.thing.method() -- thing is a property; access
* returns its getter's return type. */
CBMFileResult *r = extract_py(
"class Foo:\n"
" def method(self):\n"
" return 1\n"
"class C:\n"
" @property\n"
" def thing(self) -> Foo:\n"
" return Foo()\n"
"def use(c: C):\n"
" return c.thing.method()\n");
ASSERT_NOT_NULL(r);
ASSERT_GTE(require_resolved(r, "use", "method"), 0);
cbm_free_result(r);
PASS();
}
/* ── Evaluator guards — memoization, depth cap, step budget ─────
* (issues #710 and #720; distilled from PRs #732 and #758) */
TEST(pylsp_issue710_deep_call_chain_resolves) {
/* Regression for #710: py_eval_expr_type evaluated a call node's
* attribute receiver TWICE (container special-case + general
* attribute path), so an N-link chained builder expression cost
* O(2^N) evaluations — ~65-link real-world chains hung indexing for
* hours. With per-node memoization the chain is O(N). On unfixed
* code this 40-link chain is ~2^40 evaluations: the test cannot
* finish and the suite times out, rather than passing silently.
* We also require the FINAL link to actually resolve — the fix must
* preserve resolution quality, not just terminate. */
CBMFileResult *r = extract_py(
"from typing import Self\n"
"class G:\n"
" def add(self, x) -> Self:\n"
" return self\n"
" def compile(self):\n"
" return 1\n"
"def build():\n"
" return (\n"
" G()\n"
" .add(1).add(2).add(3).add(4).add(5).add(6).add(7).add(8)\n"
" .add(9).add(10).add(11).add(12).add(13).add(14).add(15)\n"
" .add(16).add(17).add(18).add(19).add(20).add(21).add(22)\n"
" .add(23).add(24).add(25).add(26).add(27).add(28).add(29)\n"
" .add(30).add(31).add(32).add(33).add(34).add(35).add(36)\n"
" .add(37).add(38).add(39).add(40)\n"
" .compile()\n"
" )\n");
ASSERT_NOT_NULL(r);
ASSERT_GTE(require_resolved(r, "build", "G.add"), 0);
ASSERT_GTE(require_resolved(r, "build", "G.compile"), 0);
cbm_free_result(r);
PASS();
}
TEST(pylsp_issue710_heterogeneous_receiver_chain) {
/* Guard for the memo cache's KEY choice. Every leftmost descendant of
* `c.session().execute().fetch()` — the outer call, each receiver
* call, down to the identifier `c` — starts at the same byte, so a
* cache keyed by start byte (PR #732's approach) aliases all of them:
* after `c` -> Client is inserted, the receiver call `c.session()`
* reads Client back instead of Session and the rest of the chain
* resolves wrongly or not at all. Each link here returns a DIFFERENT
* class so any such aliasing changes an assertable QN. Keying by node
* identity (TSNode.id) keeps every link distinct. */
CBMFileResult *r = extract_py(
"class Result:\n"
" def fetch(self):\n"
" return 1\n"
"class Session:\n"
" def execute(self) -> \"Result\":\n"
" return Result()\n"
"class Client:\n"
" def session(self) -> \"Session\":\n"
" return Session()\n"
"def run():\n"
" c = Client()\n"
" return c.session().execute().fetch()\n");
ASSERT_NOT_NULL(r);
ASSERT_GTE(require_resolved(r, "run", "Client.session"), 0);
ASSERT_GTE(require_resolved(r, "run", "Session.execute"), 0);
ASSERT_GTE(require_resolved(r, "run", "Result.fetch"), 0);
cbm_free_result(r);
PASS();
}
/* Resolved INVOCATION rows whose callee QN ends with `suffix` (e.g. ".compile"). */
static int count_resolved_suffix(const CBMFileResult *r, const char *suffix) {
size_t sl = strlen(suffix);
int n = 0;
for (int i = 0; i < r->resolved_calls.count; i++) {
const CBMResolvedCall *rc = &r->resolved_calls.items[i];
size_t ql = rc->callee_qn ? strlen(rc->callee_qn) : 0;
if (rc->kind == CBM_RESOLVED_INVOCATION && ql >= sl &&
strcmp(rc->callee_qn + ql - sl, suffix) == 0) {
n++;
}
}
return n;
}
TEST(pylsp_issue1527_chain_statements_resolve_completely) {
/* #1527 replaces the old budget-degradation guard: with no per-file
* evaluation budget, the #710 chain shape -- 90 statements x 30-link
* chains, re-evaluated across the bind + emit phases (the assignment bind
* flushes the memo generation in between) -- must still finish quickly
* AND resolve EVERY link and EVERY final call (each is a distinct call
* site), not merely the first statements. */
enum { CHAIN_STMTS = 90, CHAIN_LINKS = 30 };
size_t sz = 4096 + (size_t)CHAIN_STMTS * (32 + (size_t)CHAIN_LINKS * 8);
char *src = malloc(sz);
ASSERT_NOT_NULL(src);
char *p = src;
size_t left = sz;
int n = snprintf(p, left,
"from typing import Self\n"
"class G:\n"
" def add(self, x) -> Self:\n"
" return self\n"
" def compile(self):\n"
" return 1\n"
"def run():\n");
p += n;
left -= (size_t)n;
for (int s = 0; s < CHAIN_STMTS; s++) {
n = snprintf(p, left, " v%d = G()", s);
p += n;
left -= (size_t)n;
for (int l = 0; l < CHAIN_LINKS; l++) {
n = snprintf(p, left, ".add(%d)", l);
p += n;
left -= (size_t)n;
}
n = snprintf(p, left, ".compile()\n");
p += n;
left -= (size_t)n;
}
snprintf(p, left, " return v0\n");
CBMFileResult *r = extract_py(src);
free(src);
ASSERT_NOT_NULL(r);
ASSERT_EQ(count_resolved_suffix(r, ".G.compile"), CHAIN_STMTS);
ASSERT_EQ(count_resolved_suffix(r, ".G.add"), CHAIN_STMTS * CHAIN_LINKS);
cbm_free_result(r);
PASS();
}
TEST(pylsp_issue1527_large_module_resolves_every_call) {
/* #1527 reproduction: a module of 8000 classes plus one function calling
* `ClsN().chain().chain().method()` for each. Before the fix the per-file
* step cap stopped resolution after ~3333 statements (CALLS 14,666 of
* 24,000 in the graph) while the file still reported status `indexed`.
* Every one of the 8000 x 3 method call sites must resolve. */
enum { N_CLASSES = 8000 };
size_t sz = (size_t)N_CLASSES * 192 + 1024;
char *src = malloc(sz);
ASSERT_NOT_NULL(src);
size_t pos = (size_t)snprintf(src, sz, "from typing import Self\n");
for (int i = 0; i < N_CLASSES; i++) {
pos += (size_t)snprintf(src + pos, sz - pos,
"class Cls%d:\n"
" def method(self) -> int:\n"
" return %d\n"
" def chain(self) -> Self:\n"
" return self\n",
i, i);
}
pos += (size_t)snprintf(src + pos, sz - pos, "def use():\n");
for (int i = 0; i < N_CLASSES; i++) {
pos += (size_t)snprintf(src + pos, sz - pos, " Cls%d().chain().chain().method()\n", i);
}
ASSERT(pos < sz);
CBMFileResult *r =
cbm_extract_file(src, (int)pos, CBM_LANG_PYTHON, "test", "scale.py", 0, NULL, NULL);
free(src);
ASSERT_NOT_NULL(r);
ASSERT_EQ(count_resolved_suffix(r, ".method"), N_CLASSES);
ASSERT_EQ(count_resolved_suffix(r, ".chain"), 2 * N_CLASSES);
/* The last class is resolved exactly, not just some class. */
ASSERT_GTE(require_resolved(r, "use", "Cls7999.method"), 0);
cbm_free_result(r);
PASS();
}
/* ── Suite ─────────────────────────────────────────────────────── */
static char *pylsp_work_fixture(int n_classes, int *out_len) {
/* Per class: ~140 chars (5-line def). Per call: ~50 chars. Overhead
* for the class number digits scales with log10(n) but the constant
* 256 covers up to 9-digit indices comfortably. */
int approx = n_classes * 256 + 1024;
char *buf = (char *)malloc((size_t)approx);
if (!buf)
return NULL;
int pos = 0;
pos += snprintf(buf + pos, (size_t)(approx - pos), "from typing import Self\n");
for (int i = 0; i < n_classes; i++) {
int n = snprintf(buf + pos, (size_t)(approx - pos),
"class Cls%d:\n"
" def method(self) -> int:\n"
" return %d\n"
" def chain(self) -> Self:\n"
" return self\n",
i, i);
if (n < 0 || pos + n >= approx)
break;
pos += n;
}
int n = snprintf(buf + pos, (size_t)(approx - pos), "def use():\n");
pos += n;
for (int i = 0; i < n_classes; i++) {
int m = snprintf(buf + pos, (size_t)(approx - pos),
" Cls%d().chain().chain().method()\n", i);
if (m < 0 || pos + m >= approx)
break;
pos += m;
}
*out_len = pos;
return buf;
}
/* Deterministic complexity gate (#1527, O9: no clock). The resolver
* primitives count every candidate they examine -- scope-frame probes,
* registry lookup probes, enclosing-definition candidates -- plus every real
* (non-memoized) expression evaluation, into a seam-build work counter. 4x
* the input must cost about 4x the work: a linear-scan lookup or a lost memo
* lands at ~16x (quadratic) or worse. Also asserts every call site resolved,
* so the work is not "linear" because resolution stopped early. */
static uint64_t work_for(int n_classes, int *out_resolved) {
int slen = 0;
char *src = pylsp_work_fixture(n_classes, &slen);
if (!src)
return 0;
(void)cbm_lsp_work_take();
CBMFileResult *r =
cbm_extract_file(src, slen, CBM_LANG_PYTHON, "test", "scale.py", 0, NULL, NULL);
uint64_t work = cbm_lsp_work_take();
*out_resolved = r ? r->resolved_calls.count : -1;
if (r)
cbm_free_result(r);
free(src);
return work;
}
TEST(pylsp_scale_work_is_linear) {
int r1 = 0, r4 = 0;
uint64_t w1 = work_for(1000, &r1);
uint64_t w4 = work_for(4000, &r4);
double ratio = w1 ? (double)w4 / (double)w1 : 0.0;
printf(" work: 1000=%llu (resolved=%d) 4000=%llu (resolved=%d) ratio=%.2fx "
"(linear ~4x, quadratic ~16x)\n",
(unsigned long long)w1, r1, (unsigned long long)w4, r4, ratio);
/* Each class line: ClsN() + chain() + chain() + method() = 4 resolved sites. */
ASSERT_EQ(r1, 4 * 1000);
ASSERT_EQ(r4, 4 * 4000);
ASSERT_GT(w1, 0);
ASSERT(ratio < 5.0);
PASS();
}
/* #2413 (O9: no clock): the shared Tier-2 build registers one def at a time
* into a registry it finalizes only at the end, so a receiver probe that scans
* the types registered so far makes it O(methods x types). The registry
* primitives count every candidate they examine (CBM_LSP_WORK): 4x the classes
* must cost at most ~4x the work; the scanning probe lands at ~16x. */
static uint64_t cross_registry_work_for(int n_classes, int *out_types) {
CBMArena arena;
cbm_arena_init(&arena);
int n_defs = n_classes * 3;
CBMLSPDef *defs = (CBMLSPDef *)cbm_arena_calloc(&arena, (size_t)n_defs * sizeof(*defs));
if (!defs) {
cbm_arena_destroy(&arena);
*out_types = -1;
return 0;
}
/* Extraction order: each class followed by its two methods. */
for (int i = 0; i < n_classes; i++) {
CBMLSPDef *cls = &defs[i * 3];
cls->qualified_name = cbm_arena_sprintf(&arena, "scale.mod.Cls%d", i);
cls->short_name = cls->qualified_name + strlen("scale.mod.");
cls->label = "Class";
cls->def_module_qn = "scale.mod";
cls->lang = CBM_LANG_PYTHON;
for (int j = 1; j <= 2; j++) {
CBMLSPDef *m = &defs[i * 3 + j];
m->qualified_name = cbm_arena_sprintf(&arena, "%s.m%d", cls->qualified_name, j);
m->short_name = j == 1 ? "m1" : "m2";
m->label = "Method";
m->receiver_type = cls->qualified_name;
m->def_module_qn = "scale.mod";
m->lang = CBM_LANG_PYTHON;
}
}
(void)cbm_lsp_work_take();
CBMTypeRegistry *reg = cbm_py_build_cross_registry(&arena, defs, n_defs);
uint64_t work = cbm_lsp_work_take();
*out_types = reg ? reg->type_count : -1;
cbm_arena_destroy(&arena);
return work;
}
TEST(pylsp_cross_registry_build_work_is_linear) {
int t1 = 0, t4 = 0;
uint64_t w1 = cross_registry_work_for(1000, &t1);
uint64_t w4 = cross_registry_work_for(4000, &t4);
printf(" cross registry work: 1000=%llu (types=%d) 4000=%llu (types=%d)\n",
(unsigned long long)w1, t1, (unsigned long long)w4, t4);
/* Every class registered exactly once (no duplicate placeholder). */
ASSERT_EQ(t4 - t1, 3000);
ASSERT(w4 <= 5 * w1 + 1000);
PASS();
}
extern const TSLanguage *tree_sitter_python(void);
TEST(pylsp_issue1527_memo_allocation_failure_is_reported) {
int len = 0;
char *src = pylsp_work_fixture(200, &len);
ASSERT_NOT_NULL(src);
bool failures_reported = true;
for (int fail_after = 0; fail_after <= 1; fail_after++) {
cbm_py_lsp_test_memo_fail_after(fail_after);
CBMFileResult *r = extract_py(src);
cbm_py_lsp_test_memo_fail_after(-1);
failures_reported = failures_reported && r && r->has_error && r->lsp_skipped &&
r->error_msg && strcmp(r->error_msg, CBM_PY_LSP_MEMO_ERROR) == 0 &&
r->resolved_calls.count == 0;
if (r) {
cbm_result_compact(r);
failures_reported = failures_reported && r->error_msg &&
strcmp(r->error_msg, CBM_PY_LSP_MEMO_ERROR) == 0;
cbm_free_result(r);
}
}
CBMFileResult *fresh = extract_py(src);
bool recovered = fresh && !fresh->has_error && fresh->resolved_calls.count > 0;
if (fresh)
cbm_free_result(fresh);
free(src);
ASSERT_TRUE(failures_reported);
ASSERT_TRUE(recovered);
PASS();
}
TEST(pylsp_issue1527_memo_and_depth_failures_stop_work) {
const char *src = "x\n";
TSParser *parser = ts_parser_new();
ASSERT_NOT_NULL(parser);
ts_parser_set_language(parser, tree_sitter_python());
TSTree *tree = ts_parser_parse_string(parser, NULL, src, (uint32_t)strlen(src));
if (!tree) {
ts_parser_delete(parser);
FAIL("failed to parse memo fixture");
}
TSNode node = ts_node_named_child(ts_node_named_child(ts_tree_root_node(tree), 0), 0);
CBMArena arena;
cbm_arena_init(&arena);
CBMTypeRegistry reg;
cbm_registry_init(&reg, &arena);
CBMResolvedCallArray out = {0};
PyLSPContext ctx;
py_lsp_init(&ctx, &arena, src, (int)strlen(src), &reg, "test", &out);
const CBMType *known = cbm_type_named(&arena, "test.Known");
cbm_scope_bind(ctx.current_scope, "x", known);
ctx.eval_depth = 256;
bool depth_unknown = cbm_type_is_unknown(cbm_py_lsp_test_eval(&ctx, node)) &&
ctx.eval_failure == CBM_LSP_DEPTH_EXCEEDED;
ctx.eval_depth = 0;
(void)cbm_lsp_work_take();
bool depth_stopped = cbm_type_is_unknown(cbm_py_lsp_test_eval(&ctx, node));
depth_stopped = depth_stopped && cbm_lsp_work_take() == 0;
/* A fresh walk can resolve this shallow expression; the failed one cannot. */
py_lsp_init(&ctx, &arena, src, (int)strlen(src), &reg, "test", &out);
cbm_scope_bind(ctx.current_scope, "x", known);
bool depth_recovered = cbm_py_lsp_test_eval(&ctx, node) == known && !ctx.eval_failure;
bool stopped = true;
for (int capacity_failure = 0; capacity_failure < 2; capacity_failure++) {
py_lsp_init(&ctx, &arena, src, (int)strlen(src), &reg, "test", &out);
cbm_scope_bind(ctx.current_scope, "x", known);
ctx.test_memo_allocations_left = 0;
/* Pin the maximum-capacity refusal without allocating a huge table.
* A NULL table bypasses lookup; insertion must report failed growth. */
if (capacity_failure)
ctx.type_cache_cap = 1 << 26;
stopped =
stopped && cbm_type_is_unknown(cbm_py_lsp_test_eval(&ctx, node)) && ctx.eval_failure;
(void)cbm_lsp_work_take();
for (int i = 0; i < 20; i++)
stopped = cbm_type_is_unknown(cbm_py_lsp_test_eval(&ctx, node)) && stopped;
stopped = stopped && cbm_lsp_work_take() == 0;
}
cbm_arena_destroy(&arena);
ts_tree_delete(tree);
ts_parser_delete(parser);
ASSERT_TRUE(depth_unknown);
ASSERT_TRUE(depth_stopped);
ASSERT_TRUE(depth_recovered);
ASSERT_TRUE(stopped);
PASS();
}
/* A chain's LENGTH is not nesting depth (#1527 follow-up). An operator chain
* `a + a + ... + a` and a receiver chain `b.a().a()...` are evaluated link by
* link, so neither consumes the evaluator's depth guard: the file keeps its
* LSP resolution (an unrelated call in another function still resolves), no
* file-level failure is recorded, and the chain's result type still drives
* the call made on it. Before the iterative evaluation, 300 terms already
* exceeded the guard and rolled back every resolution of the file. */
static const int pylsp_long_chain_sizes[] = {300, 2000, 20000};
static char *pylsp_long_chain_source(bool receiver, int links) {
size_t sz = (size_t)links * 8 + 512;
char *src = malloc(sz);
if (!src)
return NULL;
size_t pos = (size_t)snprintf(src, sz,
"from typing import Self\n"
"class V:\n"
" def done(self):\n"
" return 1\n"
"class B:\n"
" def a(self) -> Self:\n"
" return self\n"
" def done(self):\n"
" return 1\n"
"class S:\n"
" def m(self):\n"
" return 1\n");
if (receiver) {
pos += (size_t)snprintf(src + pos, sz - pos, "def chain(b: B):\n b");
for (int i = 0; i < links; i++)
pos += (size_t)snprintf(src + pos, sz - pos, ".a()");
pos += (size_t)snprintf(src + pos, sz - pos, ".done()\n");
} else {
pos += (size_t)snprintf(src + pos, sz - pos, "def deep(a: V):\n x = a");
for (int i = 1; i < links; i++)
pos += (size_t)snprintf(src + pos, sz - pos, " + a");
pos += (size_t)snprintf(src + pos, sz - pos, "\n x.done()\n");
}
snprintf(src + pos, sz - pos, "def run():\n s = S()\n s.m()\n");
return src;
}
/* The full per-file extraction runs at the smallest size only: its Python
* usage pass climbs every identifier's ancestors with ts_node_parent, itself
* O(depth), which is cubic in one deep expression -- a separate cost outside
* the LSP. The cross-file LSP dispatch the pipeline uses (it records the
* same file-level error) runs at every size. */
static int pylsp_long_chain_case(bool receiver) {
CBMLSPDef defs[] = {
{.qualified_name = "test.V", .short_name = "V", .label = "Class", .lang = CBM_LANG_PYTHON},
{.qualified_name = "test.V.done",
.short_name = "done",
.label = "Method",
.receiver_type = "test.V",
.return_types = "int",
.lang = CBM_LANG_PYTHON},
{.qualified_name = "test.B", .short_name = "B", .label = "Class", .lang = CBM_LANG_PYTHON},
{.qualified_name = "test.B.a",
.short_name = "a",
.label = "Method",
.receiver_type = "test.B",
.return_types = "test.B",
.lang = CBM_LANG_PYTHON},
{.qualified_name = "test.B.done",
.short_name = "done",
.label = "Method",
.receiver_type = "test.B",
.return_types = "int",
.lang = CBM_LANG_PYTHON},
{.qualified_name = "test.S", .short_name = "S", .label = "Class", .lang = CBM_LANG_PYTHON},
{.qualified_name = "test.S.m",
.short_name = "m",
.label = "Method",
.receiver_type = "test.S",
.return_types = "int",
.lang = CBM_LANG_PYTHON},
};
int def_count = (int)(sizeof(defs) / sizeof(defs[0]));
const char *chain_fn = receiver ? "chain" : "deep";
const char *done = receiver ? "B.done" : "V.done";
int sizes = (int)(sizeof(pylsp_long_chain_sizes) / sizeof(pylsp_long_chain_sizes[0]));
for (int i = 0; i < sizes; i++) {
int links = pylsp_long_chain_sizes[i];
char *src = pylsp_long_chain_source(receiver, links);
ASSERT_NOT_NULL(src);
bool per_file_ok = true;
if (i == 0) {
CBMFileResult *r = extract_py(src);
per_file_ok = r && !r->has_error && !r->lsp_skipped &&
find_resolved(r, "run", "S.m") >= 0 &&
find_resolved(r, chain_fn, done) >= 0;
printf(" per-file %s chain of %d: error=%d ok=%d\n",
receiver ? "receiver" : "binary", links, r ? (int)r->has_error : -1,
(int)per_file_ok);
if (r)
cbm_free_result(r);
}
CBMFileResult cross = {0};
cbm_arena_init(&cross.arena);
(void)cbm_lsp_work_take();
cbm_pxc_run_one(CBM_LANG_PYTHON, &cross, src, (int)strlen(src), "test", defs, def_count,
NULL, NULL, 0);
uint64_t work = cbm_lsp_work_take();
free(src);
bool no_error = !cross.has_error && !cross.lsp_skipped;
bool unrelated = find_resolved(&cross, "run", "S.m") >= 0;
bool typed = find_resolved(&cross, chain_fn, done) >= 0;
printf(" cross %s chain of %d: error=%d run->S.m=%d %s=%d resolved=%d work=%llu\n",
receiver ? "receiver" : "binary", links, (int)cross.has_error, (int)unrelated, done,
(int)typed, cross.resolved_calls.count, (unsigned long long)work);
cbm_arena_destroy(&cross.arena);
ASSERT_TRUE(per_file_ok);
ASSERT_TRUE(no_error);
ASSERT_TRUE(unrelated);
ASSERT_TRUE(typed);
}
PASS();
}
TEST(pylsp_issue1527_long_binary_chain_is_not_nesting) {
return pylsp_long_chain_case(false);
}
TEST(pylsp_issue1527_long_receiver_chain_is_not_nesting) {
return pylsp_long_chain_case(true);
}
TEST(pylsp_issue1527_root_class_name_failure_binds_no_external_class) {
const char *src = "class Local:\n pass\n";
TSParser *parser = ts_parser_new();
ASSERT_NOT_NULL(parser);
ts_parser_set_language(parser, tree_sitter_python());
TSTree *tree = ts_parser_parse_string(parser, NULL, src, (uint32_t)strlen(src));
if (!tree) {
ts_parser_delete(parser);
FAIL("failed to parse root-name fixture");
}
bool closed = true, control = false;
for (int failure = 0; failure <= 2; failure++) {
CBMArena arena;
cbm_arena_init(&arena);
CBMTypeRegistry reg;
cbm_registry_init(&reg, &arena);
cbm_registry_add_type(
&reg, (CBMRegisteredType){.qualified_name = "test.Local", .short_name = "Local"});
cbm_registry_add_type(
&reg, (CBMRegisteredType){.qualified_name = "test.External", .short_name = "External"});
CBMResolvedCallArray out = {0};
PyLSPContext ctx;
py_lsp_init(&ctx, &arena, src, (int)strlen(src), &reg, "test", &out);
ctx.test_root_name_failure = failure;
cbm_py_lsp_test_bind_external_classes(&ctx, ts_tree_root_node(tree));
bool local_absent = !cbm_scope_lookup_local(ctx.current_scope, "Local");
bool external_present = cbm_scope_lookup_local(ctx.current_scope, "External") != NULL;
if (failure)
closed =
closed && ctx.callable_value_proof_disabled && local_absent && !external_present;
else
control = !ctx.callable_value_proof_disabled && local_absent && external_present;
cbm_arena_destroy(&arena);
}
ts_tree_delete(tree);
ts_parser_delete(parser);
ASSERT_TRUE(control);
ASSERT_TRUE(closed);
PASS();
}
TEST(pylsp_issue1527_cross_memo_failure_preserves_completed_rows) {
char src[4096];
size_t pos = (size_t)snprintf(src, sizeof(src), "def run(s: S):\n s + s\n");
for (int i = 0; i < 200; i++)
pos += (size_t)snprintf(src + pos, sizeof(src) - pos, " s.m()\n");
CBMLSPDef defs[] = {
{.qualified_name = "test.S", .short_name = "S", .label = "Class", .lang = CBM_LANG_PYTHON},
{.qualified_name = "test.S.m",
.short_name = "m",
.label = "Method",
.receiver_type = "test.S",
.return_types = "int",
.lang = CBM_LANG_PYTHON},
{.qualified_name = "test.S.__add__",
.short_name = "__add__",
.label = "Method",
.receiver_type = "test.S",
.return_types = "S",
.lang = CBM_LANG_PYTHON},
};
bool correct = true;
for (int mode = 0; mode < 3; mode++) {
CBMArena arena, shared;
cbm_arena_init(&arena);
cbm_arena_init(&shared);
CBMResolvedCall prior = {.caller_qn = "test.run",
.callee_qn = "test.S.__add__",
.kind = CBM_RESOLVED_INVOCATION,
.confidence = 0.1f,
.strategy = "prior.strategy",
.reason = "prior.reason",
.site_start_byte = (uint32_t)(strstr(src, "s + s") - src),
.site_end_byte = (uint32_t)(strstr(src, "s + s") - src) + 5};
CBMCall prior_synthetic = {.callee_name = "completed.synthetic"};
CBMResolvedCallArray out = {.items = &prior, .count = 1, .cap = 1};
CBMCallArray synthetic = {.items = &prior_synthetic, .count = 1, .cap = 1};
CBMTypeRegistry *base = mode == 1 ? cbm_py_build_cross_registry(&shared, defs, 3) : NULL;
CBMTypeRegistry overlay;
cbm_registry_init(&overlay, &arena);
overlay.fallback = base;
CBMBatchPyLSPFile file = {.source = src,
.source_len = (int)strlen(src),
.module_qn = "test",
.defs = defs,
.def_count = 3};
cbm_py_lsp_test_memo_fail_after(1);
bool complete =
mode == 0 ? cbm_run_py_lsp_cross(&arena, src, (int)strlen(src), "test", defs, 3, NULL,
NULL, 0, NULL, &out, &synthetic)
: mode == 1 ? cbm_run_py_lsp_cross_with_registry(&arena, src, (int)strlen(src), "test",
&overlay, NULL, NULL, 0, NULL, &out,
&synthetic)
: cbm_batch_py_lsp_cross(&arena, &file, 1, &out);
cbm_py_lsp_test_memo_fail_after(-1);
correct = correct && !complete && out.count == (mode == 2 ? 0 : 1) &&
synthetic.count == 1 && prior.confidence == 0.1f &&
strcmp(prior.strategy, "prior.strategy") == 0 &&
strcmp(prior.reason, "prior.reason") == 0 &&
strcmp(prior_synthetic.callee_name, "completed.synthetic") == 0;
if (mode != 2)
correct = correct && out.items[0].confidence == 0.1f &&
strcmp(out.items[0].strategy, "prior.strategy") == 0 &&
strcmp(out.items[0].reason, "prior.reason") == 0;
if (mode == 2)
out = (CBMResolvedCallArray){0};
complete = mode == 0 ? cbm_run_py_lsp_cross(&arena, src, (int)strlen(src), "test", defs, 3,
NULL, NULL, 0, NULL, &out, &synthetic)
: mode == 1 ? cbm_run_py_lsp_cross_with_registry(&arena, src, (int)strlen(src),
"test", &overlay, NULL, NULL, 0,
NULL, &out, &synthetic)
: cbm_batch_py_lsp_cross(&arena, &file, 1, &out);
correct = correct && complete && out.count > 200 &&
(mode == 2 || (synthetic.count > 1 && out.items[0].confidence > 0.1f));
cbm_arena_destroy(&arena);
cbm_arena_destroy(&shared);
}
ASSERT_TRUE(correct);
PASS();
}
TEST(pylsp_issue1527_cross_memo_failure_fallback_reports_error) {
const char *src = "def run(s: S):\n s.m()\n";
CBMLSPDef defs[] = {
{.qualified_name = "test.S", .short_name = "S", .label = "Class", .lang = CBM_LANG_PYTHON},
{.qualified_name = "test.S.m",
.short_name = "m",
.label = "Method",
.receiver_type = "test.S",
.return_types = "int",
.lang = CBM_LANG_PYTHON},
};
CBMFileResult result = {0};
cbm_arena_init(&result.arena);
cbm_py_lsp_test_memo_fail_after(0);
cbm_pxc_run_one(CBM_LANG_PYTHON, &result, src, (int)strlen(src), "test", defs, 2, NULL, NULL,
0);
cbm_py_lsp_test_memo_fail_after(-1);
bool reported = result.has_error && result.lsp_skipped && result.error_msg &&
strcmp(result.error_msg, CBM_PY_LSP_MEMO_ERROR) == 0 &&
result.resolved_calls.count == 0;
cbm_arena_destroy(&result.arena);
ASSERT_TRUE(reported);
PASS();
}
TEST(pylsp_issue1527_depth_failure_reports_all_entry_points) {
const char *src = "class S:\n def m(self):\n return 1\ndef run(s: S):\n s.m()\n";
CBMLSPDef defs[] = {
{.qualified_name = "test.S", .short_name = "S", .label = "Class", .lang = CBM_LANG_PYTHON},
{.qualified_name = "test.S.m",
.short_name = "m",
.label = "Method",
.receiver_type = "test.S",
.return_types = "int",
.lang = CBM_LANG_PYTHON},
};
cbm_py_lsp_test_depth_fail(true);
CBMFileResult *raw = extract_py(src);
bool correct = raw && raw->has_error && raw->lsp_skipped && raw->error_msg &&
strcmp(raw->error_msg, CBM_PY_LSP_DEPTH_ERROR) == 0 &&
raw->resolved_calls.count == 0;
if (raw)
cbm_free_result(raw);
for (int mode = 0; mode < 3; mode++) {
CBMArena arena, shared;
cbm_arena_init(&arena);
cbm_arena_init(&shared);
CBMTypeRegistry *base = mode == 1 ? cbm_py_build_cross_registry(&shared, defs, 2) : NULL;
CBMTypeRegistry overlay;
cbm_registry_init(&overlay, &arena);
overlay.fallback = base;
CBMResolvedCallArray out = {0};
CBMBatchPyLSPFile file = {.source = src,
.source_len = (int)strlen(src),
.module_qn = "test",
.defs = defs,
.def_count = 2};
if (mode == 2) {
correct = !cbm_batch_py_lsp_cross(&arena, &file, 1, &out) && correct;
} else {
CBMLSPStatus status =
mode == 0 ? cbm_run_py_lsp_cross_status(&arena, src, (int)strlen(src), "test", defs,
2, NULL, NULL, 0, NULL, &out, NULL)
: cbm_run_py_lsp_cross_with_registry_status(&arena, src, (int)strlen(src),
"test", &overlay, NULL, NULL,
0, NULL, &out, NULL);
correct = correct && status == CBM_LSP_DEPTH_EXCEEDED;
}
correct = correct && out.count == 0;
cbm_arena_destroy(&arena);
cbm_arena_destroy(&shared);
}
CBMFileResult fallback = {0};
cbm_arena_init(&fallback.arena);
cbm_pxc_run_one(CBM_LANG_PYTHON, &fallback, src, (int)strlen(src), "test", defs, 2, NULL, NULL,
0);
correct = correct && fallback.has_error && fallback.lsp_skipped && fallback.error_msg &&
strcmp(fallback.error_msg, CBM_PY_LSP_DEPTH_ERROR) == 0 &&
fallback.resolved_calls.count == 0;
cbm_arena_destroy(&fallback.arena);
cbm_py_lsp_test_depth_fail(false);
CBMFileResult *fresh = extract_py(src);
bool recovered = fresh && !fresh->has_error && fresh->resolved_calls.count > 0;
if (fresh)
cbm_free_result(fresh);
ASSERT_TRUE(correct);
ASSERT_TRUE(recovered);
PASS();
}
/* Join one known source occurrence, not merely two equal aggregate counts.
* Changing the owner or range must not borrow a neighbouring resolution. */
static bool pylsp_module_site_joins(const CBMCallArray *calls, const CBMResolvedCallArray *resolved,
const char *source, const char *site_text, const char *caller,
const char *target) {
const char *site = strstr(source, site_text);
if (!site || strstr(site + 1, site_text))
return false;
uint32_t start = (uint32_t)(site - source);
uint32_t end = start + (uint32_t)strlen(site_text);
const CBMCall *call = NULL;
int call_count = 0;
for (int i = 0; i < calls->count; i++) {
const CBMCall *candidate = &calls->items[i];
if (candidate->enclosing_func_qn && strcmp(candidate->enclosing_func_qn, caller) == 0 &&
candidate->site_start_byte == start && candidate->site_end_byte == end) {
call = candidate;
call_count++;
}
}
if (call_count != 1)
return false;
const CBMResolvedCall *hit = cbm_pipeline_find_lsp_resolution(resolved, call, false);
if (!hit || !hit->caller_qn || strcmp(hit->caller_qn, caller) != 0 || !hit->callee_qn ||
strcmp(hit->callee_qn, target) != 0 || hit->kind != CBM_RESOLVED_INVOCATION ||
hit->site_start_byte != start || hit->site_end_byte != end ||
hit->source_origin != call->source_origin)
return false;
int semantic_count = 0;
for (int i = 0; i < resolved->count; i++) {
const CBMResolvedCall *candidate = &resolved->items[i];
if (candidate->kind == CBM_RESOLVED_INVOCATION && candidate->caller_qn &&
strcmp(candidate->caller_qn, caller) == 0 && candidate->site_start_byte == start &&
candidate->site_end_byte == end)
semantic_count++;
}
CBMCall wrong = *call;
wrong.enclosing_func_qn = "unrelated.module.caller";
bool rejects_owner = cbm_pipeline_find_lsp_resolution(resolved, &wrong, false) == NULL;
wrong = *call;
wrong.site_start_byte++;
return semantic_count == 1 && rejects_owner &&
cbm_pipeline_find_lsp_resolution(resolved, &wrong, false) == NULL;
}
static bool pylsp_module_has_def(const CBMFileResult *r, const char *qn) {
for (int i = 0; i < r->defs.count; i++) {
if (r->defs.items[i].qualified_name && strcmp(r->defs.items[i].qualified_name, qn) == 0)
return true;
}
return false;
}
TEST(pylsp_module_callers_join_exact_raw_occurrences) {
static const char source[] = "class Alpha:\n"
" def render(self):\n"
" return 1\n"
"class Beta:\n"
" def render(self):\n"
" return 2\n"
"def helper():\n"
" return 3\n"
"def control():\n"
" function_a = Alpha()\n"
" function_a.render()\n"
" helper()\n"
"class Controller:\n"
" def control(self):\n"
" method_b = Beta()\n"
" method_b.render()\n"
"module_a = Alpha()\n"
"module_b = Beta()\n"
"module_a.render()\n"
"module_b.render()\n";
CBMFileResult *r = cbm_extract_file(source, (int)strlen(source), CBM_LANG_PYTHON, "test",
"app.py", 0, NULL, NULL);
bool correct = r && !r->has_error && !r->parse_incomplete && !r->lsp_skipped && r->module_qn &&
strcmp(r->module_qn, "test.app") == 0;
if (correct) {
correct =
pylsp_module_has_def(r, "test.app.Alpha.render") &&
pylsp_module_has_def(r, "test.app.Beta.render") &&
pylsp_module_has_def(r, "test.app.control") &&
pylsp_module_has_def(r, "test.app.Controller.control") &&
pylsp_module_site_joins(&r->calls, &r->resolved_calls, source, "module_a.render()",
"test.app", "test.app.Alpha.render") &&
pylsp_module_site_joins(&r->calls, &r->resolved_calls, source, "module_b.render()",
"test.app", "test.app.Beta.render") &&
pylsp_module_site_joins(&r->calls, &r->resolved_calls, source, "function_a.render()",
"test.app.control", "test.app.Alpha.render") &&
pylsp_module_site_joins(&r->calls, &r->resolved_calls, source, "method_b.render()",
"test.app.Controller.control", "test.app.Beta.render");
}
if (r)
cbm_free_result(r);
ASSERT_TRUE(correct);
/* Package symbol-scope normalization is separate. A builtin isolates
* this package Module's raw caller identity from that remaining issue. */
static const char init_source[] = "PACKAGE_SIZE = len([1])\n";
r = cbm_extract_file(init_source, (int)strlen(init_source), CBM_LANG_PYTHON, "test",
"pkg/__init__.py", 0, NULL, NULL);
correct = r && !r->has_error && !r->parse_incomplete && !r->lsp_skipped && r->module_qn &&
strcmp(r->module_qn, "test.pkg.__init__") == 0 &&
pylsp_module_has_def(r, "builtins.len") &&
pylsp_module_site_joins(&r->calls, &r->resolved_calls, init_source, "len([1])",
"test.pkg.__init__", "builtins.len");
if (r)
cbm_free_result(r);
ASSERT_TRUE(correct);
PASS();
}
TEST(pylsp_module_named_lambda_keeps_exact_dunder_pair) {
static const char source[] = "class Number:\n"
" def __add__(self, other):\n"
" return self\n"
"module_number = Number()\n"
"add = lambda left, right: left + right\n"
"first = add(module_number, module_number)\n"
"second = add(module_number, module_number)\n";
CBMFileResult *r = cbm_extract_file(source, (int)strlen(source), CBM_LANG_PYTHON, "test",
"app.py", 0, NULL, NULL);
bool correct = r && !r->has_error && !r->parse_incomplete && !r->lsp_skipped &&
pylsp_module_has_def(r, "test.app.Number.__add__") &&
pylsp_module_site_joins(&r->calls, &r->resolved_calls, source, "left + right",
"test.app.<lambda>", "test.app.Number.__add__");
if (r)
cbm_free_result(r);
ASSERT_TRUE(correct);
PASS();
}
TEST(pylsp_module_cross_callers_own_input_and_join_exact_sites) {
static const char source[] = "from helpers import Alpha, Beta\n"
"module_a = Alpha()\n"
"module_b = Beta()\n"
"module_a.render()\n"
"module_b.render()\n";
static const char *const paths[] = {"app.py", "pkg/__init__.py"};
static const char *const modules[] = {"test.app", "test.pkg.__init__"};
CBMLSPDef defs[] = {
{.qualified_name = "test.helpers.Alpha",
.short_name = "Alpha",
.label = "Class",
.def_module_qn = "test.helpers",
.lang = CBM_LANG_PYTHON},
{.qualified_name = "test.helpers.Alpha.render",
.short_name = "render",
.label = "Method",
.receiver_type = "test.helpers.Alpha",
.def_module_qn = "test.helpers",
.return_types = "int",
.lang = CBM_LANG_PYTHON},
{.qualified_name = "test.helpers.Beta",
.short_name = "Beta",
.label = "Class",
.def_module_qn = "test.helpers",
.lang = CBM_LANG_PYTHON},
{.qualified_name = "test.helpers.Beta.render",
.short_name = "render",
.label = "Method",
.receiver_type = "test.helpers.Beta",
.def_module_qn = "test.helpers",
.return_types = "int",
.lang = CBM_LANG_PYTHON},
};
const char *import_names[] = {"Alpha", "Beta"};
const char *import_qns[] = {"test.helpers.Alpha", "test.helpers.Beta"};
bool correct = true;
for (int path = 0; path < 2; path++) {
CBMFileResult *r = cbm_extract_file(source, (int)strlen(source), CBM_LANG_PYTHON, "test",
paths[path], 0, NULL, NULL);
if (!r || r->has_error || r->parse_incomplete || r->lsp_skipped || !r->module_qn ||
strcmp(r->module_qn, modules[path]) != 0) {
correct = false;
if (r)
cbm_free_result(r);
continue;
}
for (int prebuilt = 0; prebuilt < 2; prebuilt++) {
CBMArena arena, shared;
cbm_arena_init(&arena);
cbm_arena_init(&shared);
CBMTypeRegistry *base = prebuilt ? cbm_py_build_cross_registry(&shared, defs, 4) : NULL;
CBMTypeRegistry overlay;
cbm_registry_init(&overlay, &arena);
overlay.fallback = base;
char module_input[64];
snprintf(module_input, sizeof(module_input), "%s", modules[path]);
CBMResolvedCallArray out = {0};
CBMLSPStatus status =
prebuilt ? cbm_run_py_lsp_cross_with_registry_status(
&arena, source, (int)strlen(source), module_input, &overlay,
import_names, import_qns, 2, NULL, &out, NULL)
: cbm_run_py_lsp_cross_status(&arena, source, (int)strlen(source),
module_input, defs, 4, import_names,
import_qns, 2, NULL, &out, NULL);
/* Returned caller strings must outlive the caller's input buffer.
* Overwrite live storage, avoiding a dangling-pointer test. */
memset(module_input, 'x', strlen(module_input));
correct = correct && (!prebuilt || (base && base->read_only)) &&
status == CBM_LSP_COMPLETE &&
pylsp_module_site_joins(&r->calls, &out, source, "module_a.render()",
modules[path], "test.helpers.Alpha.render") &&
pylsp_module_site_joins(&r->calls, &out, source, "module_b.render()",
modules[path], "test.helpers.Beta.render");
cbm_arena_destroy(&arena);
cbm_arena_destroy(&shared);
}
cbm_free_result(r);
}
ASSERT_TRUE(correct);
PASS();
}
/* Package symbols use their parent scope while Module callers retain the file. */
TEST(pylsp_init_scope_context_and_helper) {
static const struct {
const char *raw;
const char *scope;
} cases[] = {
{"test.__init__", "test"},
{"test.pkg.__init__", "test.pkg"},
{"test.pkg.nested.__init__", "test.pkg.nested"},
{"test.plain", "test.plain"},
{"test.x__init__", "test.x__init__"},
{"__init__", "__init__"},
{"test.index", "test.index"},
{NULL, NULL},
};
bool correct = true;
for (size_t i = 0; i < sizeof(cases) / sizeof(cases[0]); i++) {
CBMArena arena;
cbm_arena_init(&arena);
CBMTypeRegistry registry;
cbm_registry_init(&registry, &arena);
CBMResolvedCallArray out = {0};
PyLSPContext ctx;
py_lsp_init(&ctx, &arena, "pass\n", 5, &registry, cases[i].raw, &out);
correct =
correct && ctx.eval_failure == CBM_LSP_COMPLETE && ctx.current_scope &&
ctx.file_module_qn == cases[i].raw &&
cbm_fqn_symbol_scope_len(cases[i].raw) == (cases[i].scope ? strlen(cases[i].scope) : 0);
correct =
correct && (cases[i].scope ? ctx.module_qn && strcmp(ctx.module_qn, cases[i].scope) == 0
: ctx.module_qn == NULL);
cbm_arena_destroy(&arena);
}
ASSERT_TRUE(correct);
PASS();
}
TEST(pylsp_init_scope_raw_exact_occurrences_and_annotations) {
static const char source[] =
"class Store:\n def get(self):\n return 1\n"
"def identity(value: Store) -> Store:\n return value\n"
"def quoted(value: 'Store') -> 'Store':\n return value\n"
"def wrapped(values: list[Store]) -> Store:\n return values[0]\n"
"def run(first: Store, second: 'Store', values: list[Store]):\n"
" first.get()\n second.get()\n values[0].get()\n"
" identity(first).get()\n quoted(second).get()\n"
" wrapped(values).get()\n"
"class User:\n def use(self, item: Store):\n return item.get()\n"
"top = Store()\ntop.get()\n";
static const char *paths[] = {"plain.py", "__init__.py", "pkg/nested/__init__.py"};
static const char *scopes[] = {"test.plain", "test", "test.pkg.nested"};
static const char *modules[] = {"test.plain", "test.__init__", "test.pkg.nested.__init__"};
static const char *sites[] = {"first.get()", "second.get()",
"values[0].get()", "identity(first).get()",
"quoted(second).get()", "wrapped(values).get()"};
bool correct = true;
for (int mode = 0; mode < 3; mode++) {
CBMFileResult *r = cbm_extract_file(source, (int)strlen(source), CBM_LANG_PYTHON, "test",
paths[mode], 0, NULL, NULL);
bool extracted = r && !r->has_error && !r->parse_incomplete && !r->lsp_skipped &&
r->module_qn && strcmp(r->module_qn, modules[mode]) == 0;
correct = correct && extracted;
if (extracted) {
char caller[128], target[128], method[128];
snprintf(caller, sizeof(caller), "%s.run", scopes[mode]);
snprintf(target, sizeof(target), "%s.Store.get", scopes[mode]);
snprintf(method, sizeof(method), "%s.User.use", scopes[mode]);
correct = correct && pylsp_module_has_def(r, caller) &&
pylsp_module_has_def(r, target) && pylsp_module_has_def(r, method);
for (size_t j = 0; j < sizeof(sites) / sizeof(sites[0]); j++) {
bool joined = pylsp_module_site_joins(&r->calls, &r->resolved_calls, source,
sites[j], caller, target);
correct = correct && joined;
}
bool method_join = pylsp_module_site_joins(&r->calls, &r->resolved_calls, source,
"item.get()", method, target);
bool module_join = pylsp_module_site_joins(&r->calls, &r->resolved_calls, source,
"top.get()", modules[mode], target);
correct = correct && method_join && module_join;
}
if (r)
cbm_free_result(r);
}
ASSERT_TRUE(correct);
PASS();
}
static bool pylsp_init_named_type(const CBMType *type, const char *qn) {
return type && type->kind == CBM_TYPE_NAMED && type->data.named.qualified_name &&
strcmp(type->data.named.qualified_name, qn) == 0;
}
/* Type of the binding of name in this frame only, or NULL when unbound. */
static const CBMType *pylsp_local_type(const CBMScope *scope, const char *name) {
const CBMVarBinding *binding = cbm_scope_lookup_local(scope, name);
return binding ? binding->type : NULL;
}
static bool pylsp_init_signature(const CBMTypeRegistry *registry, const char *function,
const char *expected_type) {
const CBMRegisteredFunc *f = cbm_registry_lookup_func(registry, function);
const CBMType *sig = f ? f->signature : NULL;
if (!sig || sig->kind != CBM_TYPE_FUNC || !sig->data.func.param_types)
return false;
const CBMType **params = sig->data.func.param_types;
if (!pylsp_init_named_type(params[0], expected_type) ||
!pylsp_init_named_type(params[1], expected_type) || !params[2] || params[3])
return false;
const CBMType *container = params[2];
return container->kind == CBM_TYPE_TEMPLATE && container->data.template_type.template_name &&
strcmp(container->data.template_type.template_name, "list") == 0 &&
container->data.template_type.arg_count == 1 &&
container->data.template_type.template_args &&
pylsp_init_named_type(container->data.template_type.template_args[0], expected_type);
}
TEST(pylsp_init_scope_cross_signatures_keep_definition_language) {
const char *params[] = {"Token", "'Token'", "list[Token]"};
CBMLSPDef defs[] = {
{.qualified_name = "test.pkg.call",
.short_name = "call",
.label = "Function",
.def_module_qn = "test.pkg.__init__",
.lang = CBM_LANG_PYTHON,
.signature_param_types = params,
.signature_param_count = 3},
{.qualified_name = "test.foreign.__init__.call",
.short_name = "call",
.label = "Function",
.def_module_qn = "test.foreign.__init__",
.lang = CBM_LANG_GO,
.signature_param_types = params,
.signature_param_count = 3},
{.qualified_name = "test.unknown.__init__.call",
.short_name = "call",
.label = "Function",
.def_module_qn = "test.unknown.__init__",
.lang = (CBMLanguage)-1,
.signature_param_types = params,
.signature_param_count = 3},
};
CBMArena arena, shared;
cbm_arena_init(&arena);
cbm_arena_init(&shared);
CBMTypeRegistry fallback;
cbm_registry_init(&fallback, &arena);
bool registered = cbm_py_lsp_test_register_defs(&arena, &fallback, defs, 3);
bool correct = registered &&
pylsp_init_signature(&fallback, "test.pkg.call", "test.pkg.Token") &&
pylsp_init_signature(&fallback, "test.foreign.__init__.call",
"test.foreign.__init__.Token") &&
pylsp_init_signature(&fallback, "test.unknown.__init__.call",
"test.unknown.__init__.Token");
CBMTypeRegistry *base = cbm_py_build_cross_registry(&shared, defs, 3);
correct = correct && base && base->read_only &&
pylsp_init_signature(base, "test.pkg.call", "test.pkg.Token") &&
!cbm_registry_lookup_func(base, "test.foreign.__init__.call") &&
!cbm_registry_lookup_func(base, "test.unknown.__init__.call");
cbm_arena_destroy(&arena);
cbm_arena_destroy(&shared);
ASSERT_TRUE(correct);
PASS();
}
TEST(pylsp_init_scope_external_classes_and_cross_exact_joins) {
const char *source = "from pkg.sub import Widget as Imported\n"
"def use():\n store = Store()\n store.get()\n"
" imported = Imported()\n imported.get()\n";
CBMLSPDef defs[] = {
{.qualified_name = "test.pkg.Store",
.short_name = "Store",
.label = "Class",
.def_module_qn = "test.pkg.__init__",
.lang = CBM_LANG_PYTHON},
{.qualified_name = "test.pkg.Store.get",
.short_name = "get",
.label = "Method",
.def_module_qn = "test.pkg.__init__",
.receiver_type = "test.pkg.Store",
.return_types = "int",
.lang = CBM_LANG_PYTHON},
{.qualified_name = "test.pkg.sub.Widget",
.short_name = "Widget",
.label = "Class",
.def_module_qn = "test.pkg.sub",
.lang = CBM_LANG_PYTHON},
{.qualified_name = "test.pkg.sub.Widget.get",
.short_name = "get",
.label = "Method",
.def_module_qn = "test.pkg.sub",
.receiver_type = "test.pkg.sub.Widget",
.return_types = "int",
.lang = CBM_LANG_PYTHON},
{.qualified_name = "test.pkg.Outer.Nested",
.short_name = "Nested",
.label = "Class",
.def_module_qn = "test.pkg.__init__",
.lang = CBM_LANG_PYTHON},
};
const char *names[] = {"Imported"};
const char *qns[] = {"test.pkg.sub.Widget"};
CBMFileResult *r = cbm_extract_file(source, (int)strlen(source), CBM_LANG_PYTHON, "test",
"pkg/__init__.py", 0, NULL, NULL);
TSParser *parser = ts_parser_new();
if (parser)
ts_parser_set_language(parser, tree_sitter_python());
TSTree *tree =
parser ? ts_parser_parse_string(parser, NULL, source, (uint32_t)strlen(source)) : NULL;
bool correct = r && !r->has_error && !r->parse_incomplete && !r->lsp_skipped && tree &&
pylsp_module_has_def(r, "test.pkg.use");
for (int prebuilt = 0; correct && prebuilt < 2; prebuilt++) {
CBMArena arena, shared;
cbm_arena_init(&arena);
cbm_arena_init(&shared);
CBMTypeRegistry *base = cbm_py_build_cross_registry(&shared, defs, 5);
CBMTypeRegistry overlay;
cbm_registry_init(&overlay, &arena);
overlay.fallback = base;
CBMResolvedCallArray out = {0};
PyLSPContext ctx;
py_lsp_init(&ctx, &arena, source, (int)strlen(source), &overlay, "test.pkg.__init__", &out);
cbm_py_lsp_test_bind_external_classes(&ctx, ts_tree_root_node(tree));
bool globals = base && ctx.eval_failure == CBM_LSP_COMPLETE &&
pylsp_init_named_type(pylsp_local_type(ctx.current_scope, "Store"),
"test.pkg.Store") &&
!cbm_scope_lookup_local(ctx.current_scope, "Widget") &&
!cbm_scope_lookup_local(ctx.current_scope, "Nested");
char module_input[] = "test.pkg.__init__";
CBMLSPStatus status =
prebuilt
? cbm_run_py_lsp_cross_with_registry_status(&arena, source, (int)strlen(source),
module_input, &overlay, names, qns, 1,
tree, &out, NULL)
: cbm_run_py_lsp_cross_status(&arena, source, (int)strlen(source), module_input,
defs, 5, names, qns, 1, tree, &out, NULL);
memset(module_input, 'x', sizeof(module_input) - 1);
bool store = pylsp_module_site_joins(&r->calls, &out, source, "store.get()", "test.pkg.use",
"test.pkg.Store.get");
bool imported = pylsp_module_site_joins(&r->calls, &out, source, "imported.get()",
"test.pkg.use", "test.pkg.sub.Widget.get");
correct = correct && globals && status == CBM_LSP_COMPLETE && store && imported;
cbm_arena_destroy(&arena);
cbm_arena_destroy(&shared);
}
if (tree)
ts_tree_delete(tree);
if (parser)
ts_parser_delete(parser);
if (r)
cbm_free_result(r);
ASSERT_TRUE(correct);
PASS();
}
TEST(pylsp_init_scope_failure_preserves_outputs_and_wrappers) {
const char *source = "def run():\n value = Store()\n return value.get()\n";
CBMLSPDef defs[] = {
{.qualified_name = "test.pkg.Store",
.short_name = "Store",
.label = "Class",
.def_module_qn = "test.pkg.__init__",
.lang = CBM_LANG_PYTHON},
{.qualified_name = "test.pkg.Store.get",
.short_name = "get",
.label = "Method",
.def_module_qn = "test.pkg.__init__",
.receiver_type = "test.pkg.Store",
.return_types = "int",
.lang = CBM_LANG_PYTHON},
};
bool correct = true;
cbm_py_lsp_test_scope_fail_after(0);
CBMFileResult *raw = cbm_extract_file(source, (int)strlen(source), CBM_LANG_PYTHON, "test",
"pkg/__init__.py", 0, NULL, NULL);
cbm_py_lsp_test_scope_fail_after(-1);
correct = raw && raw->has_error && raw->lsp_skipped && raw->error_msg &&
strcmp(raw->error_msg, CBM_PY_LSP_SCOPE_ERROR) == 0 &&
raw->resolved_calls.count == 0 && pylsp_module_has_def(raw, "test.pkg.run");
if (raw)
cbm_free_result(raw);
for (int mode = 0; mode < 3; mode++) {
CBMArena arena, shared;
cbm_arena_init(&arena);
cbm_arena_init(&shared);
CBMTypeRegistry *base = cbm_py_build_cross_registry(&shared, defs, 2);
CBMTypeRegistry overlay;
cbm_registry_init(&overlay, &arena);
overlay.fallback = base;
CBMResolvedCall prior = {.caller_qn = "completed.caller",
.callee_qn = "completed.target",
.strategy = "prior.strategy",
.reason = "prior.reason",
.confidence = 0.125f,
.kind = CBM_RESOLVED_INVOCATION,
.site_start_byte = 7,
.site_end_byte = 14};
unsigned char saved[sizeof(prior)];
memcpy(saved, &prior, sizeof(prior));
CBMCall previous_synthetic = {.callee_name = "completed.synthetic"};
CBMCallArray synthetic = {.items = &previous_synthetic, .count = 1, .cap = 1};
CBMResolvedCallArray out = {.items = &prior, .count = 1, .cap = 1};
cbm_py_lsp_test_scope_fail_after(mode == 2 ? 1 : 0);
/* mode 2 gets past context normalization, then fails signature scope. */
CBMLSPStatus status = mode == 1
? cbm_run_py_lsp_cross_with_registry_status(
&arena, source, (int)strlen(source), "test.pkg.__init__",
&overlay, NULL, NULL, 0, NULL, &out, &synthetic)
: cbm_run_py_lsp_cross_status(&arena, source, (int)strlen(source),
"test.pkg.__init__", defs, 2, NULL,
NULL, 0, NULL, &out, &synthetic);
cbm_py_lsp_test_scope_fail_after(-1);
correct = correct && base && status == CBM_LSP_SCOPE_FAILED && out.count == 1 &&
out.items == &prior && memcmp(saved, &prior, sizeof(prior)) == 0 &&
synthetic.count == 1 && synthetic.items == &previous_synthetic &&
strcmp(previous_synthetic.callee_name, "completed.synthetic") == 0;
cbm_py_lsp_test_scope_fail_after(0);
bool complete =
mode == 1
? cbm_run_py_lsp_cross_with_registry(&arena, source, (int)strlen(source),
"test.pkg.__init__", &overlay, NULL, NULL, 0,
NULL, &out, &synthetic)
: cbm_run_py_lsp_cross(&arena, source, (int)strlen(source), "test.pkg.__init__",
defs, 2, NULL, NULL, 0, NULL, &out, &synthetic);
cbm_py_lsp_test_scope_fail_after(-1);
correct =
correct && !complete && out.count == 1 && memcmp(saved, &prior, sizeof(prior)) == 0;
CBMBatchPyLSPFile file = {.source = source,
.source_len = (int)strlen(source),
.module_qn = "test.pkg.__init__",
.defs = defs,
.def_count = 2};
CBMResolvedCallArray batch = {0};
cbm_py_lsp_test_scope_fail_after(0);
bool batch_complete = cbm_batch_py_lsp_cross(&arena, &file, 1, &batch);
cbm_py_lsp_test_scope_fail_after(-1);
correct = correct && !batch_complete && batch.count == 0;
cbm_arena_destroy(&arena);
cbm_arena_destroy(&shared);
}
ASSERT_TRUE(correct);
PASS();
}
TEST(pylsp_init_scope_shared_failure_keeps_arena_and_needed_fallback) {
const char *source = "def use():\n obj = Item()\n return obj.get()\n";
CBMLSPDef defs[] = {
{.qualified_name = "test.plain.Item",
.short_name = "Item",
.label = "Class",
.def_module_qn = "test.plain",
.lang = CBM_LANG_PYTHON},
{.qualified_name = "test.plain.Item.get",
.short_name = "get",
.label = "Method",
.def_module_qn = "test.plain",
.receiver_type = "test.plain.Item",
.return_types = "int",
.lang = CBM_LANG_PYTHON},
{.qualified_name = "test.unused.work",
.short_name = "work",
.label = "Function",
.def_module_qn = "test.unused.__init__",
.lang = CBM_LANG_PYTHON},
};
CBMArena arena, shared;
cbm_arena_init(&arena);
cbm_arena_init(&shared);
char *sentinel = cbm_arena_strdup(&shared, "caller-owned shared data");
cbm_py_lsp_test_scope_fail_after(0);
CBMTypeRegistry *failed = cbm_py_build_cross_registry(&shared, defs, 3);
CBMResolvedCallArray out = {0};
CBMLSPStatus fallback =
cbm_run_py_lsp_cross_status(&arena, source, (int)strlen(source), "test.plain", defs, 2,
NULL, NULL, 0, NULL, &out, NULL);
cbm_py_lsp_test_scope_fail_after(-1);
bool correct = !failed && sentinel && strcmp(sentinel, "caller-owned shared data") == 0 &&
fallback == CBM_LSP_COMPLETE;
CBMFileResult *extracted = cbm_extract_file(source, (int)strlen(source), CBM_LANG_PYTHON,
"test", "plain.py", 0, NULL, NULL);
correct = correct && extracted && !extracted->has_error && !extracted->parse_incomplete &&
pylsp_module_has_def(extracted, "test.plain.use") &&
pylsp_module_site_joins(&extracted->calls, &out, source, "obj.get()",
"test.plain.use", "test.plain.Item.get");
CBMTypeRegistry *fresh = cbm_py_build_cross_registry(&shared, defs, 3);
correct = correct && fresh && fresh->read_only &&
cbm_registry_lookup_func(fresh, "test.unused.work") &&
strcmp(sentinel, "caller-owned shared data") == 0;
if (extracted)
cbm_free_result(extracted);
cbm_arena_destroy(&arena);
cbm_arena_destroy(&shared);
ASSERT_TRUE(correct);
PASS();
}
/* Class execution keeps the extracted caller and exact source occurrence. */
static bool pylsp_class_site_unjoined(const CBMFileResult *r, const char *source,
const char *site_text, const char *caller) {
const char *site = strstr(source, site_text);
if (!site || strstr(site + 1, site_text))
return false;
uint32_t start = (uint32_t)(site - source);
uint32_t end = start + (uint32_t)strlen(site_text);
const CBMCall *call = NULL;
int count = 0;
for (int i = 0; i < r->calls.count; i++) {
const CBMCall *candidate = &r->calls.items[i];
if (candidate->enclosing_func_qn && strcmp(candidate->enclosing_func_qn, caller) == 0 &&
candidate->site_start_byte == start && candidate->site_end_byte == end) {
call = candidate;
count++;
}
}
if (count != 1 || cbm_pipeline_find_lsp_resolution(&r->resolved_calls, call, false))
return false;
for (int i = 0; i < r->resolved_calls.count; i++) {
const CBMResolvedCall *hit = &r->resolved_calls.items[i];
if (hit->kind == CBM_RESOLVED_INVOCATION && hit->site_start_byte == start &&
hit->site_end_byte == end)
return false;
}
return true;
}
typedef struct {
const char *site;
const char *caller;
const char *target; /* NULL requires an extracted, unresolved occurrence. */
} PylspClassSite;
static bool pylsp_class_sites(const CBMFileResult *r, const char *source,
const PylspClassSite *sites, size_t count) {
if (!r || r->has_error || r->parse_incomplete || r->lsp_skipped || !r->module_qn)
return false;
for (size_t i = 0; i < count; i++) {
const PylspClassSite *s = &sites[i];
if (strcmp(s->caller, r->module_qn) != 0 && !pylsp_module_has_def(r, s->caller))
return false;
if (s->target) {
if (!pylsp_module_has_def(r, s->target) ||
!pylsp_module_site_joins(&r->calls, &r->resolved_calls, source, s->site, s->caller,
s->target))
return false;
} else if (!pylsp_class_site_unjoined(r, source, s->site, s->caller)) {
return false;
}
}
return true;
}
TEST(pylsp_classbody_calls_join_extracted_caller) {
static const char source[] =
"def field(n):\n return n\n"
"class Svc:\n"
" @staticmethod\n def make(n):\n return Svc()\n"
"class Model:\n"
" first = field(11)\n second = Svc(12)\n third = Svc.make(13)\n"
"def control():\n field(21)\n Svc(22)\n Svc.make(23)\n";
static const char *const paths[] = {"main.py", "pkg/__init__.py"};
static const char *const modules[] = {"test.main", "test.pkg.__init__"};
static const char *const scopes[] = {"test.main", "test.pkg"};
bool correct = true;
for (int path = 0; path < 2; path++) {
CBMFileResult *r = cbm_extract_file(source, (int)strlen(source), CBM_LANG_PYTHON, "test",
paths[path], 0, NULL, NULL);
char field[64], svc[64], make[64], control[64];
snprintf(field, sizeof(field), "%s.field", scopes[path]);
snprintf(svc, sizeof(svc), "%s.Svc", scopes[path]);
snprintf(make, sizeof(make), "%s.Svc.make", scopes[path]);
snprintf(control, sizeof(control), "%s.control", scopes[path]);
const PylspClassSite sites[] = {{"field(11)", modules[path], field},
{"Svc(12)", modules[path], svc},
{"Svc.make(13)", modules[path], make},
{"field(21)", control, field},
{"Svc(22)", control, svc},
{"Svc.make(23)", control, make}};
correct = pylsp_class_sites(r, source, sites, 6) && correct;
if (r)
cbm_free_result(r);
}
ASSERT_TRUE(correct);
PASS();
}
TEST(pylsp_classbody_names_visible_to_later_statements_only) {
static const char source[] =
"def helper(n):\n return n\n"
"class Svc:\n def run(self, n):\n return n\n"
"class Model:\n"
" obj = Svc()\n result = obj.run(101)\n"
" def helper(self):\n return 2\n"
" later = helper(None)\n"
" def m(self):\n return helper(102)\n"
" def n(self):\n return obj.run(103)\n"
"class Shadow:\n helper = None\n value = helper(104)\n"
"def control():\n obj = Svc()\n obj.run(105)\n helper(106)\n";
const PylspClassSite sites[] = {
{"obj.run(101)", "test.main", "test.main.Svc.run"},
{"helper(None)", "test.main", "test.main.Model.helper"},
{"helper(102)", "test.main.Model.m", "test.main.helper"},
{"obj.run(103)", "test.main.Model.n", NULL},
{"helper(104)", "test.main", NULL},
{"obj.run(105)", "test.main.control", "test.main.Svc.run"},
{"helper(106)", "test.main.control", "test.main.helper"},
};
CBMFileResult *r = extract_py(source);
bool correct = pylsp_class_sites(r, source, sites, 7);
if (r)
cbm_free_result(r);
ASSERT_TRUE(correct);
PASS();
}
TEST(pylsp_decorator_and_base_calls_join_extracted_caller) {
static const char source[] = "def deco(fn):\n return fn\n"
"def deco_arg(n):\n return deco\n"
"def base_factory(n):\n return object\n"
"class Svc:\n def run(self):\n return 1\n"
"@deco_arg(11)\ndef top():\n pass\n"
"@deco\ndef bare():\n pass\n"
"@deco_arg(12)\nclass Model(base_factory(13)):\n"
" marker = Svc()\n"
" @deco_arg(marker.run())\n def m(self):\n pass\n"
" @property\n def p(self):\n return 1\n"
"def outer():\n @deco_arg(14)\n def inner():\n pass\n"
" return inner\n";
bool correct = true;
for (int package = 0; package < 2; package++) {
const char *scope = package ? "test.pkg" : "test.main";
const char *module = package ? "test.pkg.__init__" : "test.main";
char deco[64], base[64], run[64], outer[64];
snprintf(deco, sizeof(deco), "%s.deco_arg", scope);
snprintf(base, sizeof(base), "%s.base_factory", scope);
snprintf(run, sizeof(run), "%s.Svc.run", scope);
snprintf(outer, sizeof(outer), "%s.outer", scope);
const PylspClassSite sites[] = {
{"deco_arg(11)", module, deco}, {"deco_arg(12)", module, deco},
{"base_factory(13)", module, base}, {"deco_arg(marker.run())", module, deco},
{"marker.run()", module, run}, {"deco_arg(14)", outer, deco}};
CBMFileResult *r = cbm_extract_file(source, (int)strlen(source), CBM_LANG_PYTHON, "test",
package ? "pkg/__init__.py" : "main.py", 0, NULL, NULL);
correct = pylsp_class_sites(r, source, sites, 6) && correct;
if (r) {
const char *bare_sites[] = {"@deco\n", "@property\n"};
for (int b = 0; b < 2; b++) {
const char *hit = strstr(source, bare_sites[b]);
uint32_t lo = (uint32_t)(hit - source);
uint32_t hi = lo + (uint32_t)strlen(bare_sites[b]);
for (int i = 0; i < r->calls.count; i++)
correct = !(r->calls.items[i].site_start_byte >= lo &&
r->calls.items[i].site_start_byte < hi) &&
correct;
for (int i = 0; i < r->resolved_calls.count; i++)
correct = !(r->resolved_calls.items[i].kind == CBM_RESOLVED_INVOCATION &&
r->resolved_calls.items[i].site_start_byte >= lo &&
r->resolved_calls.items[i].site_start_byte < hi) &&
correct;
}
cbm_free_result(r);
}
}
ASSERT_TRUE(correct);
PASS();
}
TEST(pylsp_nested_and_local_classbody_calls_join_extracted_caller) {
static const char source[] =
"def field(n):\n return n\n"
"class Svc:\n def run(self):\n return 1\n"
"class Outer:\n tag = Svc()\n"
" class Meta:\n ordering = field(31)\n probe = tag.run()\n"
" inner = Meta()\n"
"def build():\n before = field(32)\n"
" class Local:\n value = field(33)\n"
" class Inner:\n value = field(34)\n"
" return Local\n"
"class Controller:\n def build(self):\n"
" class Local:\n value = field(35)\n"
" return Local\n";
const PylspClassSite sites[] = {
{"field(31)", "test.main", "test.main.field"},
{"tag.run()", "test.main", NULL},
{"Meta()", "test.main", "test.main.Outer.Meta"},
{"field(32)", "test.main.build", "test.main.field"},
{"field(33)", "test.main.build", "test.main.field"},
{"field(34)", "test.main.build", "test.main.field"},
{"field(35)", "test.main.Controller.build", "test.main.field"},
};
CBMFileResult *r = extract_py(source);
/* #1548: a class declared in a function body is extracted, as a local of
* that function; its class-body calls still join the enclosing function
* (the sites above), never the local class. */
bool sites_ok = pylsp_class_sites(r, source, sites, 7);
bool locals_extracted = r && pylsp_module_has_def(r, "test.main.build.Local") &&
pylsp_module_has_def(r, "test.main.Controller.build.Local");
if (r)
cbm_free_result(r);
ASSERT_TRUE(sites_ok);
ASSERT_TRUE(locals_extracted);
PASS();
}
TEST(pylsp_class_comprehensions_separate_iterable_and_body_scopes) {
static const char source[] =
"class ModuleSvc:\n def run(self, n):\n return n\n"
"class ClassSvc:\n"
" def run(self, n):\n return n\n"
" def items(self) -> list[ModuleSvc]:\n return []\n"
"shared = ModuleSvc()\n"
"class Model:\n"
" shared = ClassSvc()\n local = ClassSvc()\n"
" a = [item.run(41) for item in local.items()]\n"
" b = {shared.run(42) for item in [1]}\n"
" c = {item: shared.run(43) for item in [1]}\n"
" d = (shared.run(44) for item in [1])\n"
" e = [local.run(45) for item in [1]]\n"
" f = [item for item in [1] if local.run(46)]\n"
" g = [item for item in [1] for other in local.items(47)]\n"
" h = [[inner.run(48) for inner in [outer]] for outer in [ModuleSvc()]]\n"
" class Nested:\n"
" shared = ClassSvc()\n"
" a = [shared.run(49) for item in [1]]\n"
" after = shared.run(50)\n"
"def control():\n return [item.run(51) for item in [ModuleSvc()]]\n";
const PylspClassSite sites[] = {
{"local.items()", "test.main", "test.main.ClassSvc.items"},
{"item.run(41)", "test.main", "test.main.ModuleSvc.run"},
{"shared.run(42)", "test.main", "test.main.ModuleSvc.run"},
{"shared.run(43)", "test.main", "test.main.ModuleSvc.run"},
{"shared.run(44)", "test.main", "test.main.ModuleSvc.run"},
{"local.run(45)", "test.main", NULL},
{"local.run(46)", "test.main", NULL},
{"local.items(47)", "test.main", NULL},
{"inner.run(48)", "test.main", "test.main.ModuleSvc.run"},
{"shared.run(49)", "test.main", "test.main.ModuleSvc.run"},
{"shared.run(50)", "test.main", "test.main.ClassSvc.run"},
{"item.run(51)", "test.main.control", "test.main.ModuleSvc.run"},
};
CBMFileResult *r = extract_py(source);
bool correct = pylsp_class_sites(r, source, sites, 12) &&
pylsp_module_has_def(r, "test.main.ClassSvc.run") &&
pylsp_module_has_def(r, "test.main.ClassSvc.items");
if (r)
cbm_free_result(r);
ASSERT_TRUE(correct);
PASS();
}
TEST(pylsp_class_legacy_dispatch_cannot_bypass_or_publish_shadows) {
static const char source[] = "def wanted(n):\n return n\n"
"def other(n):\n return n\n"
"class Svc:\n def run(self, n):\n return n\n"
"table = {'x': wanted}\n"
"fn = lambda obj: obj.run(61)\n"
"stable = lambda obj: obj.run(62)\n"
"class Shadow:\n"
" table = None\n a = table['x'](63)\n"
" fn = None\n b = fn(Svc(64))\n"
"class Replacements:\n"
" table = {'x': other}\n c = table['x'](65)\n"
" stable = lambda obj: obj.run(66)\n"
"outside = table['x'](67)\n"
"outside_lambda = stable(Svc(68))\n";
const PylspClassSite sites[] = {
{"table['x'](63)", "test.main", NULL},
{"Svc(64)", "test.main", "test.main.Svc"},
{"table['x'](65)", "test.main", NULL},
{"table['x'](67)", "test.main", "test.main.wanted"},
{"Svc(68)", "test.main", "test.main.Svc"},
};
CBMFileResult *r = extract_py(source);
bool correct = pylsp_class_sites(r, source, sites, 5) &&
pylsp_module_has_def(r, "test.main.other") &&
pylsp_module_has_def(r, "test.main.Svc.run");
/* A replay has the lambda body's site, not the invocation's site. Verify
* those exact sites independently so output dedup cannot hide a replay. */
if (r) {
const char *body[] = {"obj.run(61)", "obj.run(62)", "obj.run(66)"};
for (int b = 0; b < 3; b++) {
uint32_t lo = (uint32_t)(strstr(source, body[b]) - source);
uint32_t hi = lo + (uint32_t)strlen(body[b]);
int extracted = 0, resolved = 0;
for (int i = 0; i < r->calls.count; i++)
extracted += r->calls.items[i].site_start_byte == lo &&
r->calls.items[i].site_end_byte == hi;
for (int i = 0; i < r->resolved_calls.count; i++) {
const CBMResolvedCall *hit = &r->resolved_calls.items[i];
if (hit->kind == CBM_RESOLVED_INVOCATION && hit->site_start_byte == lo &&
hit->site_end_byte == hi) {
resolved++;
correct = hit->caller_qn && strcmp(hit->caller_qn, "test.main.<lambda>") == 0 &&
hit->callee_qn && strcmp(hit->callee_qn, "test.main.Svc.run") == 0 &&
correct;
}
}
correct = extracted == 1 && resolved == (b == 1 ? 1 : 0) && correct;
}
cbm_free_result(r);
}
ASSERT_TRUE(correct);
PASS();
}
TEST(pylsp_class_frame_failure_restores_context_and_stops) {
static const char source[] = "class Outer:\n leaked = 1\n"
" first = hit(71)\n"
" values = [hit(72) for item in [1]]\n"
"later = hit(73)\n";
TSParser *parser = ts_parser_new();
bool setup = parser && ts_parser_set_language(parser, tree_sitter_python());
TSTree *tree =
setup ? ts_parser_parse_string(parser, NULL, source, (uint32_t)strlen(source)) : NULL;
bool correct = tree && !ts_node_has_error(ts_tree_root_node(tree));
CBMLSPDef defs[] = {{.qualified_name = "test.main.hit",
.short_name = "hit",
.label = "Function",
.def_module_qn = "test.main",
.lang = CBM_LANG_PYTHON}};
for (int failure = 0; tree && failure < 2; failure++) {
CBMArena arena;
cbm_arena_init(&arena);
CBMTypeRegistry reg;
cbm_registry_init(&reg, &arena);
bool registered = cbm_py_lsp_test_register_defs(&arena, &reg, defs, 1);
CBMResolvedCallArray out = {0};
PyLSPContext ctx;
py_lsp_init(&ctx, &arena, source, (int)strlen(source), &reg, "test.main", &out);
CBMScope *root_scope = ctx.current_scope;
ctx.enclosing_func_qn = "saved.caller";
ctx.enclosing_class_qn = "saved.class";
cbm_py_lsp_test_class_frame_fail_after(failure);
py_lsp_process_file(&ctx, ts_tree_root_node(tree));
cbm_py_lsp_test_class_frame_fail_after(-1);
correct = registered && root_scope && ctx.eval_failure == CBM_LSP_SCOPE_FAILED &&
ctx.current_scope == root_scope && !ctx.class_body_outer_scope &&
!ctx.class_body_qn && !ctx.class_body_in_comprehension &&
strcmp(ctx.enclosing_func_qn, "saved.caller") == 0 &&
strcmp(ctx.enclosing_class_qn, "saved.class") == 0 && out.count == 0 &&
!cbm_scope_lookup_local(root_scope, "leaked") &&
!cbm_scope_lookup_local(root_scope, "Outer") &&
!cbm_scope_lookup_local(root_scope, "later") && correct;
(void)cbm_lsp_work_take();
py_lsp_process_file(&ctx, ts_tree_root_node(tree));
correct = cbm_lsp_work_take() == 0 && out.count == 0 && correct;
cbm_arena_destroy(&arena);
}
if (tree)
ts_tree_delete(tree);
if (parser)
ts_parser_delete(parser);
/* The pre-existing package seam must still ignore ordinary module names,
* including the new class/comprehension frames. */
static const char healthy[] = "def hit(n):\n return n\n"
"class Outer:\n first = hit(74)\n"
" values = [hit(75) for item in [1]]\n";
cbm_py_lsp_test_scope_fail_after(0);
CBMFileResult *r = extract_py(healthy);
cbm_py_lsp_test_scope_fail_after(-1);
const PylspClassSite sites[] = {{"hit(74)", "test.main", "test.main.hit"},
{"hit(75)", "test.main", "test.main.hit"}};
bool recovered = pylsp_class_sites(r, healthy, sites, 2);
if (r)
cbm_free_result(r);
ASSERT_TRUE(correct);
ASSERT_TRUE(recovered);
PASS();
}
TEST(pylsp_class_frame_failure_rolls_back_raw_and_cross_outputs) {
static const char source[] = "class Outer:\n item = S()\n item + item\n item.m()\n"
" class Inner:\n later = S(81)\n"
"after = S(82)\n";
CBMLSPDef defs[] = {
{.qualified_name = "test.main.S",
.short_name = "S",
.label = "Class",
.def_module_qn = "test.main",
.lang = CBM_LANG_PYTHON},
{.qualified_name = "test.main.S.m",
.short_name = "m",
.label = "Method",
.receiver_type = "test.main.S",
.def_module_qn = "test.main",
.return_types = "int",
.lang = CBM_LANG_PYTHON},
{.qualified_name = "test.main.S.__add__",
.short_name = "__add__",
.label = "Method",
.receiver_type = "test.main.S",
.def_module_qn = "test.main",
.return_types = "S",
.lang = CBM_LANG_PYTHON},
};
bool correct = true;
for (int prebuilt = 0; prebuilt < 2; prebuilt++) {
CBMArena arena, shared;
cbm_arena_init(&arena);
cbm_arena_init(&shared);
CBMTypeRegistry *base = prebuilt ? cbm_py_build_cross_registry(&shared, defs, 3) : NULL;
CBMTypeRegistry overlay;
cbm_registry_init(&overlay, &arena);
overlay.fallback = base;
CBMResolvedCall prior = {
.caller_qn = "test.main",
.callee_qn = "test.main.S.__add__",
.kind = CBM_RESOLVED_INVOCATION,
.confidence = 0.1f,
.strategy = "prior.strategy",
.reason = "prior.reason",
.site_start_byte = (uint32_t)(strstr(source, "item + item") - source),
.site_end_byte = (uint32_t)(strstr(source, "item + item") - source) + 11};
CBMResolvedCall saved_prior;
memcpy(&saved_prior, &prior, sizeof(prior));
CBMCall prior_synthetic = {.callee_name = "completed.synthetic"};
CBMResolvedCallArray out = {.items = &prior, .count = 1, .cap = 1};
CBMCallArray synthetic = {.items = &prior_synthetic, .count = 1, .cap = 1};
cbm_py_lsp_test_class_frame_fail_after(1);
CBMLSPStatus status =
prebuilt
? cbm_run_py_lsp_cross_with_registry_status(&arena, source, (int)strlen(source),
"test.main", &overlay, NULL, NULL, 0,
NULL, &out, &synthetic)
: cbm_run_py_lsp_cross_status(&arena, source, (int)strlen(source), "test.main",
defs, 3, NULL, NULL, 0, NULL, &out, &synthetic);
cbm_py_lsp_test_class_frame_fail_after(-1);
correct = (!prebuilt || base) && status == CBM_LSP_SCOPE_FAILED && out.count == 1 &&
synthetic.count == 1 && memcmp(&prior, &saved_prior, sizeof(prior)) == 0 &&
memcmp(&out.items[0], &saved_prior, sizeof(prior)) == 0 &&
strcmp(prior_synthetic.callee_name, "completed.synthetic") == 0 && correct;
bool complete =
prebuilt ? cbm_run_py_lsp_cross_with_registry(&arena, source, (int)strlen(source),
"test.main", &overlay, NULL, NULL, 0,
NULL, &out, &synthetic)
: cbm_run_py_lsp_cross(&arena, source, (int)strlen(source), "test.main", defs,
3, NULL, NULL, 0, NULL, &out, &synthetic);
correct = complete && out.count > 1 && synthetic.count > 1 &&
out.items[0].confidence > saved_prior.confidence && correct;
cbm_arena_destroy(&arena);
cbm_arena_destroy(&shared);
}
static const char raw_source[] = "class S:\n def m(self):\n return 1\n"
"class Outer:\n item = S()\n item.m()\n"
" class Inner:\n later = S(83)\n";
cbm_py_lsp_test_class_frame_fail_after(2);
CBMFileResult *r = extract_py(raw_source);
cbm_py_lsp_test_class_frame_fail_after(-1);
correct = r && r->has_error && r->lsp_skipped && !r->parse_incomplete && r->error_msg &&
strcmp(r->error_msg, CBM_PY_LSP_SCOPE_ERROR) == 0 && r->resolved_calls.count == 0 &&
r->calls.count > 0 && correct;
if (r) {
cbm_result_compact(r);
correct = r->error_msg && strcmp(r->error_msg, CBM_PY_LSP_SCOPE_ERROR) == 0 && correct;
cbm_free_result(r);
}
r = extract_py(raw_source);
const PylspClassSite sites[] = {{"item.m()", "test.main", "test.main.S.m"},
{"S(83)", "test.main", "test.main.S"}};
bool recovered = pylsp_class_sites(r, raw_source, sites, 2);
if (r)
cbm_free_result(r);
ASSERT_TRUE(correct);
ASSERT_TRUE(recovered);
PASS();
}
/* The prefix index is an existence oracle, never an exact-symbol selector. */
static bool pylsp_prefix_scan(const CBMTypeRegistry *reg, const char *prefix) {
if (!reg || !prefix)
return false;
CBMFreeFuncIter iter;
cbm_registry_all_funcs_chain(reg, &iter);
size_t length = strlen(prefix);
for (int i; (i = cbm_free_func_iter_next(&iter)) >= 0;) {
const char *qn = iter.reg->funcs[i].qualified_name;
if (qn && strncmp(qn, prefix, length) == 0)
return true;
}
return false;
}
static bool pylsp_prefix_add(CBMTypeRegistry *reg, const char *qn, const char *receiver) {
int before = reg->func_count;
cbm_registry_add_func(
reg,
(CBMRegisteredFunc){.qualified_name = qn, .short_name = "call", .receiver_type = receiver});
return reg->func_count == before + 1;
}
static bool pylsp_prefix_oracles(const CBMTypeRegistry *reg) {
const char *prefixes[] = {NULL, "", "pkg.", "pkg.sub.",
"pkg.sub.call", "pkg.su.", "pkg.submarine.", "pkg.Type.",
"only.tail.", "missing.", "zzzz."};
for (size_t i = 0; i < sizeof(prefixes) / sizeof(prefixes[0]); i++) {
uint64_t visits = 17;
bool indexed = cbm_registry_has_func_qn_prefix(reg, prefixes[i], &visits);
if (indexed != pylsp_prefix_scan(reg, prefixes[i]) || visits < 17 ||
cbm_registry_has_func_qn_prefix(reg, prefixes[i], NULL) != indexed ||
(!prefixes[i] && visits != 17))
return false;
}
return true;
}
TEST(pylsp_prefix_order_matches_iterator_and_chain_shadowing) {
CBMArena arena, scratch;
cbm_arena_init(&arena);
cbm_arena_init(&scratch);
CBMTypeRegistry head, middle, tail, empty;
cbm_registry_init(&head, &arena);
cbm_registry_init(&middle, &arena);
cbm_registry_init(&tail, &arena);
cbm_registry_init(&empty, &arena);
bool ok = pylsp_prefix_add(&head, NULL, NULL) && pylsp_prefix_add(&head, "", NULL) &&
pylsp_prefix_add(&head, "pkg.sub.call", NULL) &&
pylsp_prefix_add(&head, "pkg.sub.call", NULL) &&
pylsp_prefix_add(&head, "pkg.Type.call", "pkg.Type") &&
pylsp_prefix_add(&middle, "pkg.sub.call", "different.receiver") &&
pylsp_prefix_add(&middle, "pkg.submarine.call", NULL) &&
pylsp_prefix_add(&tail, "only.tail.call", NULL);
head.fallback = &middle;
middle.fallback = &tail;
ok = pylsp_prefix_oracles(NULL) && pylsp_prefix_oracles(&empty) &&
pylsp_prefix_oracles(&head) && ok;
cbm_registry_build_func_qn_order(&head, &scratch);
ok = head.func_qn_sorted && head.func_qn_sorted_count == 4 && head.func_qn_sorted_upto == 5 &&
pylsp_prefix_oracles(&head) && ok;
cbm_registry_build_func_qn_order(&tail, &arena);
ok = tail.func_qn_sorted && pylsp_prefix_oracles(&head) && ok;
cbm_registry_build_func_qn_order(&middle, &scratch);
ok = middle.func_qn_sorted && pylsp_prefix_oracles(&head) && ok;
if (head.func_qn_sorted) {
int duplicate_count = 0, previous = -1;
for (int i = 0; i < head.func_qn_sorted_count; i++) {
const CBMRegistryQNEntry *entry = &head.func_qn_sorted[i];
ok = entry->func_index >= 0 && entry->func_index < head.func_count &&
entry->qualified_name == head.funcs[entry->func_index].qualified_name && ok;
if (strcmp(entry->qualified_name, "pkg.sub.call") == 0) {
ok = entry->func_index > previous && ok;
previous = entry->func_index;
duplicate_count++;
}
}
ok = duplicate_count == 2 && ok;
}
size_t owned_before = cbm_arena_total(&arena), scratch_before = cbm_arena_total(&scratch);
ok = pylsp_prefix_oracles(&head) && cbm_arena_total(&arena) == owned_before &&
cbm_arena_total(&scratch) == scratch_before && ok;
/* An indexed fallback also works behind an unindexed empty head. */
empty.fallback = &head;
ok = pylsp_prefix_oracles(&empty) && ok;
cbm_arena_destroy(&scratch);
cbm_arena_destroy(&arena);
ASSERT_TRUE(ok);
PASS();
}
TEST(pylsp_prefix_order_survives_array_growth_rebuild_and_seal) {
CBMArena arena, scratch;
cbm_arena_init(&arena);
cbm_arena_init(&scratch);
CBMTypeRegistry reg, unindexed;
cbm_registry_init(&reg, &arena);
cbm_registry_init(&unindexed, &arena);
bool ok = pylsp_prefix_add(&reg, "pkg.old.call", NULL);
cbm_registry_build_func_qn_order(&reg, &scratch);
CBMRegisteredFunc *original_array = reg.funcs;
const CBMRegistryQNEntry *order = reg.func_qn_sorted;
for (int i = 0; i < 140; i++) {
const char *qn = cbm_arena_sprintf(&arena, "tail.mod%03d.call", i);
ok = qn && pylsp_prefix_add(&reg, qn, NULL) && ok;
}
ok = reg.func_count == 141 && reg.funcs != original_array && reg.func_qn_sorted == order &&
reg.func_qn_sorted_upto == 1 &&
cbm_registry_has_func_qn_prefix(&reg, "tail.mod139.", NULL) &&
!cbm_registry_has_func_qn_prefix(&reg, "tail.mod140.", NULL) && ok;
cbm_registry_build_func_qn_order(&reg, &scratch);
ok = reg.func_qn_sorted && reg.func_qn_sorted_upto == reg.func_count &&
cbm_registry_has_func_qn_prefix(&reg, "tail.mod139.", NULL) && ok;
/* End use of the old snapshot before replacing an identity. No lookup is
* permitted until the complete order has been rebuilt. */
if (reg.func_count == 141) {
reg.funcs[0].qualified_name = "pkg.renamed.call";
cbm_registry_build_func_qn_order(&reg, &scratch);
ok = !cbm_registry_has_func_qn_prefix(&reg, "pkg.old.", NULL) &&
cbm_registry_has_func_qn_prefix(&reg, "pkg.renamed.", NULL) && ok;
}
order = reg.func_qn_sorted;
reg.read_only = true;
cbm_registry_test_qn_order_fail_after(0);
cbm_registry_build_func_qn_order(&reg, &scratch);
cbm_registry_test_qn_order_fail_after(-1);
cbm_registry_add_func(&reg, (CBMRegisteredFunc){.qualified_name = "forbidden.call"});
ok = reg.func_qn_sorted == order && reg.func_count == 141 &&
cbm_registry_has_func_qn_prefix(&reg, "pkg.renamed.", NULL) &&
!cbm_registry_has_func_qn_prefix(&reg, "forbidden.", NULL) && ok;
ok = pylsp_prefix_add(&unindexed, "sealed.call", NULL) && ok;
unindexed.read_only = true;
cbm_registry_build_func_qn_order(&unindexed, &scratch);
uint64_t visits = 0;
ok = !unindexed.func_qn_sorted &&
cbm_registry_has_func_qn_prefix(&unindexed, "sealed.", &visits) && visits == 1 && ok;
cbm_arena_destroy(&scratch);
cbm_arena_destroy(&arena);
ASSERT_TRUE(ok);
PASS();
}
TEST(pylsp_prefix_order_allocation_failure_preserves_oracle) {
CBMArena arena;
cbm_arena_init(&arena);
CBMTypeRegistry reg;
cbm_registry_init(&reg, &arena);
bool ok = pylsp_prefix_add(&reg, "pkg.sub.call", NULL) && pylsp_prefix_add(&reg, NULL, NULL) &&
pylsp_prefix_add(&reg, "only.tail.call", NULL);
cbm_registry_test_qn_order_fail_after(0);
cbm_registry_build_func_qn_order(&reg, &arena);
cbm_registry_test_qn_order_fail_after(-1);
uint64_t visits = 0;
ok = !reg.func_qn_sorted && reg.func_qn_sorted_count == 0 && reg.func_qn_sorted_upto == 0 &&
pylsp_prefix_oracles(&reg) && !cbm_registry_has_func_qn_prefix(&reg, "zzzz.", &visits) &&
visits == 3 && ok;
cbm_registry_test_qn_order_fail_after(1);
cbm_registry_build_func_qn_order(&reg, &arena);
bool first_built = reg.func_qn_sorted && reg.func_qn_sorted_upto == reg.func_count;
cbm_registry_build_func_qn_order(&reg, &arena);
cbm_registry_test_qn_order_fail_after(-1);
ok = first_built && !reg.func_qn_sorted && reg.func_qn_sorted_count == 0 &&
reg.func_qn_sorted_upto == 0 && pylsp_prefix_oracles(&reg) && ok;
cbm_registry_build_func_qn_order(&reg, &arena);
ok = reg.func_qn_sorted && reg.func_qn_sorted_count == 2 && reg.func_qn_sorted_upto == 3 &&
pylsp_prefix_oracles(&reg) && ok;
cbm_arena_destroy(&arena);
ASSERT_TRUE(ok);
PASS();
}
TEST(pylsp_prefix_order_has_absolute_comparison_bound) {
enum { FUNCTIONS = 1024, PROBES = 32, BOUND = 12 };
CBMArena arena;
cbm_arena_init(&arena);
CBMTypeRegistry reg;
cbm_registry_init(&reg, &arena);
bool ok = true;
for (int i = FUNCTIONS - 1; i >= 0; i--) {
const char *qn = cbm_arena_sprintf(&arena, "project.mod%04d.call", i);
ok = qn && pylsp_prefix_add(&reg, qn, NULL) && ok;
}
cbm_registry_build_func_qn_order(&reg, &arena);
ok = reg.func_count == FUNCTIONS && reg.func_qn_sorted_count == FUNCTIONS &&
reg.func_qn_sorted_upto == FUNCTIONS && reg.func_qn_sorted && ok;
for (int i = 0; i < PROBES; i++) {
char hit[64], miss[64];
int a = snprintf(hit, sizeof(hit), "project.mod%04d.", i * 31);
int b = snprintf(miss, sizeof(miss), "project.mod%04dx.", i * 31);
uint64_t positive = 0, negative = 0;
bool ready = a > 0 && (size_t)a < sizeof(hit) && b > 0 && (size_t)b < sizeof(miss);
ok = ready && cbm_registry_has_func_qn_prefix(&reg, hit, &positive) &&
!cbm_registry_has_func_qn_prefix(&reg, miss, &negative) && positive > 0 &&
positive <= BOUND && negative > 0 && negative <= BOUND &&
pylsp_prefix_scan(&reg, hit) && !pylsp_prefix_scan(&reg, miss) && ok;
}
/* Failure restores the reference scan, which must inspect every entry on
* this absent suffix. This is a deterministic contrast, not a timing ratio. */
cbm_registry_test_qn_order_fail_after(0);
cbm_registry_build_func_qn_order(&reg, &arena);
cbm_registry_test_qn_order_fail_after(-1);
uint64_t visits = 0;
ok = !reg.func_qn_sorted && !cbm_registry_has_func_qn_prefix(&reg, "zzzz.", &visits) &&
visits == FUNCTIONS && visits > BOUND && ok;
cbm_arena_destroy(&arena);
ASSERT_TRUE(ok);
PASS();
}
TEST(pylsp_prefix_order_preserves_exact_selection_and_ambiguity) {
CBMArena arena;
cbm_arena_init(&arena);
bool ok = true;
for (int first = 0; first < 2; first++) {
CBMTypeRegistry reg;
cbm_registry_init(&reg, &arena);
ok = pylsp_prefix_add(&reg, "pkg.duplicate", NULL) &&
pylsp_prefix_add(&reg, "pkg.duplicate", NULL) && ok;
const CBMRegisteredFunc *linear = cbm_registry_lookup_func(&reg, "pkg.duplicate");
ok = reg.func_count == 2 && linear == &reg.funcs[0] && ok;
cbm_registry_build_func_qn_order(&reg, &arena);
ok = cbm_registry_lookup_func(&reg, "pkg.duplicate") == linear && ok;
reg.index_first_registered = first != 0;
cbm_registry_finalize(&reg);
const CBMRegisteredFunc *selected = cbm_registry_lookup_func(&reg, "pkg.duplicate");
ok = reg.func_count == 2 && selected == &reg.funcs[first ? 0 : 1] && ok;
cbm_registry_build_func_qn_order(&reg, &arena);
ok = cbm_registry_lookup_func(&reg, "pkg.duplicate") == selected && ok;
}
CBMLSPDef defs[] = {
{.qualified_name = "pkg.duplicate",
.short_name = "duplicate",
.label = "Function",
.def_module_qn = "pkg",
.return_types = "int",
.lang = CBM_LANG_PYTHON},
{.qualified_name = "pkg.duplicate",
.short_name = "duplicate",
.label = "Function",
.def_module_qn = "pkg",
.return_types = "str",
.lang = CBM_LANG_PYTHON},
{.qualified_name = "pkg.unique",
.short_name = "unique",
.label = "Function",
.def_module_qn = "pkg",
.lang = CBM_LANG_PYTHON},
};
for (int fail = 0; fail < 2; fail++) {
cbm_registry_test_qn_order_fail_after(fail ? 0 : -1);
CBMTypeRegistry *reg = cbm_py_build_cross_registry(&arena, defs, 3);
cbm_registry_test_qn_order_fail_after(-1);
ok = reg && reg->read_only && reg->func_count > 3 && ok;
if (reg) {
int duplicate_count = 0, unique_count = 0;
for (int i = 0; i < reg->func_count; i++) {
const CBMRegisteredFunc *f = &reg->funcs[i];
bool ambiguous = (f->flags & CBM_FUNC_FLAG_AMBIGUOUS_BINDING) != 0;
if (fail)
ok = ambiguous && ok;
if (f->qualified_name && strcmp(f->qualified_name, "pkg.duplicate") == 0) {
duplicate_count++;
ok = ambiguous && ok;
}
if (f->qualified_name && strcmp(f->qualified_name, "pkg.unique") == 0) {
unique_count++;
ok = ambiguous == (fail != 0) && ok;
}
}
ok = duplicate_count == 2 && unique_count == 1 &&
(fail ? reg->func_qn_sorted == NULL : reg->func_qn_sorted != NULL) &&
cbm_registry_has_func_qn_prefix(reg, "pkg.", NULL) &&
!cbm_registry_has_func_qn_prefix(reg, "pkg.missing.", NULL) && ok;
}
}
cbm_arena_destroy(&arena);
ASSERT_TRUE(ok);
PASS();
}
/* A negative must have one real extracted source occurrence and no accepted
* resolution; a missing carrier cannot make it pass. */
static bool pylsp_prefix_site_unresolved(const CBMCallArray *calls,
const CBMResolvedCallArray *resolved, const char *source,
const char *site_text, const char *caller) {
const char *site = strstr(source, site_text);
if (!site || strstr(site + 1, site_text))
return false;
uint32_t start = (uint32_t)(site - source), end = start + (uint32_t)strlen(site_text);
const CBMCall *found = NULL;
int count = 0;
for (int i = 0; i < calls->count; i++) {
const CBMCall *call = &calls->items[i];
if (call->enclosing_func_qn && strcmp(call->enclosing_func_qn, caller) == 0 &&
call->site_start_byte == start && call->site_end_byte == end) {
found = call;
count++;
}
}
return count == 1 && cbm_pipeline_find_lsp_resolution(resolved, found, false) == NULL;
}
TEST(pylsp_prefix_raw_per_file_and_shared_keep_exact_calls) {
static const char source[] = "import pkg\nimport vendor\n"
"def ping():\n return 1\n"
"def run():\n"
" pkg.sub.ping()\n"
" pkg.su.ping()\n"
" vendor.external.ping()\n";
CBMLSPDef defs[] = {
{.qualified_name = "pkg.sub.ping",
.short_name = "ping",
.label = "Function",
.def_module_qn = "pkg.sub",
.return_types = "int",
.lang = CBM_LANG_PYTHON},
{.qualified_name = "pkg.submarine.ping",
.short_name = "ping",
.label = "Function",
.def_module_qn = "pkg.submarine",
.return_types = "str",
.lang = CBM_LANG_PYTHON},
};
const char *names[] = {"pkg", "vendor"};
const char *qns[] = {"pkg", "vendor"};
bool ok = true;
for (int fail = 0; fail < 2; fail++) {
uint64_t probes_before =
atomic_load_explicit(&cbm_py_submodule_probes, memory_order_relaxed);
uint64_t visits_before =
atomic_load_explicit(&cbm_py_submodule_probe_visits, memory_order_relaxed);
cbm_registry_test_qn_order_fail_after(fail ? 0 : -1);
CBMFileResult *r = cbm_extract_file(source, (int)strlen(source), CBM_LANG_PYTHON, "pkg",
"sub.py", 0, NULL, NULL);
cbm_registry_test_qn_order_fail_after(-1);
bool healthy = r && !r->has_error && !r->parse_incomplete && !r->lsp_skipped &&
pylsp_module_has_def(r, "pkg.sub.ping") &&
pylsp_module_has_def(r, "pkg.sub.run");
ok = healthy &&
atomic_load_explicit(&cbm_py_submodule_probes, memory_order_relaxed) > probes_before &&
atomic_load_explicit(&cbm_py_submodule_probe_visits, memory_order_relaxed) >
visits_before &&
ok;
if (healthy) {
ok = pylsp_module_site_joins(&r->calls, &r->resolved_calls, source, "pkg.sub.ping()",
"pkg.sub.run", "pkg.sub.ping") &&
pylsp_prefix_site_unresolved(&r->calls, &r->resolved_calls, source,
"pkg.su.ping()", "pkg.sub.run") &&
pylsp_prefix_site_unresolved(&r->calls, &r->resolved_calls, source,
"vendor.external.ping()", "pkg.sub.run") &&
ok;
for (int shared_mode = 0; shared_mode < 2; shared_mode++) {
CBMArena arena, shared;
cbm_arena_init(&arena);
cbm_arena_init(&shared);
CBMTypeRegistry *reg = NULL;
cbm_registry_test_qn_order_fail_after(fail ? 0 : -1);
if (shared_mode)
reg = cbm_py_build_cross_registry(&shared, defs, 2);
CBMResolvedCallArray out = {0};
CBMLSPStatus status = CBM_LSP_SCOPE_FAILED;
probes_before =
atomic_load_explicit(&cbm_py_submodule_probes, memory_order_relaxed);
visits_before =
atomic_load_explicit(&cbm_py_submodule_probe_visits, memory_order_relaxed);
if (!shared_mode || reg) {
status = shared_mode
? cbm_run_py_lsp_cross_with_registry_status(
&arena, source, (int)strlen(source), "pkg.sub", reg, names,
qns, 2, r->cached_tree, &out, NULL)
: cbm_run_py_lsp_cross_status(&arena, source, (int)strlen(source),
"pkg.sub", defs, 2, names, qns, 2,
r->cached_tree, &out, NULL);
}
cbm_registry_test_qn_order_fail_after(-1);
ok = status == CBM_LSP_COMPLETE &&
atomic_load_explicit(&cbm_py_submodule_probes, memory_order_relaxed) >
probes_before &&
atomic_load_explicit(&cbm_py_submodule_probe_visits, memory_order_relaxed) >
visits_before &&
pylsp_module_site_joins(&r->calls, &out, source, "pkg.sub.ping()",
"pkg.sub.run", "pkg.sub.ping") &&
pylsp_prefix_site_unresolved(&r->calls, &out, source, "pkg.su.ping()",
"pkg.sub.run") &&
pylsp_prefix_site_unresolved(&r->calls, &out, source, "vendor.external.ping()",
"pkg.sub.run") &&
ok;
cbm_arena_destroy(&arena);
cbm_arena_destroy(&shared);
}
}
if (r)
cbm_free_result(r);
}
ASSERT_TRUE(ok);
PASS();
}
/* Module-range queries must return the old array sequence, including duplicates. */
static bool pylsp_range_add_type(CBMTypeRegistry *reg, const char *qn, const char *name) {
int before = reg->type_count;
cbm_registry_add_type(reg, (CBMRegisteredType){.qualified_name = qn, .short_name = name});
return reg->type_count == before + 1;
}
static bool pylsp_range_member(const char *qn, const char *prefix, bool direct) {
if (!qn || !prefix)
return false;
size_t n = strlen(prefix);
return strncmp(qn, prefix, n) == 0 && (!direct || !strchr(qn + n, '.'));
}
/* No fixed result buffer and no call to the implementation's member predicate. */
static bool pylsp_range_oracle(const CBMTypeRegistry *reg, bool types, const char *prefix,
bool direct, CBMArena *scratch, uint64_t *visits) {
CBMQNPrefixIter it;
bool opened = types ? cbm_registry_types_with_qn_prefix(reg, prefix, direct, scratch, &it)
: cbm_registry_funcs_with_qn_prefix(reg, prefix, direct, scratch, &it);
if (!opened)
return false;
int total = reg ? (types ? reg->type_count : reg->func_count) : 0;
bool ok = true;
for (int i = 0; i < total; i++) {
const char *qn = types ? reg->types[i].qualified_name : reg->funcs[i].qualified_name;
if (pylsp_range_member(qn, prefix, direct))
ok = cbm_qn_prefix_iter_next(&it) == i && ok;
}
ok = cbm_qn_prefix_iter_next(&it) == -1 && cbm_qn_prefix_iter_next(&it) == -1 && ok;
if (visits)
*visits = it.visits;
return ok;
}
static void pylsp_range_reset_failures(void) {
cbm_registry_test_type_qn_order_fail_after(-1);
cbm_registry_test_qn_prefix_block_fail_after(-1);
cbm_py_test_module_scratch_fail_after(-1);
cbm_py_test_module_prefix_fail_after(-1);
}
TEST(pylsp_range_order_boundaries_and_null_contract) {
static const char *const qns[] = {
"p.foo.B", "p.foo.A.child", NULL, "p.foo.A-b", "p.foobar.C", "p.foo.AB",
"p.foo.A.z", "p.foo.", "", "p.foo.B", "p.foo.C.child", "p.foo.C-plain"};
static const char *const prefixes[] = {NULL, "", "p.foo.", "p.foo", "p.foo.A",
"p.foo.A.", "p.foo.B", "p.foo..", "zzzz."};
CBMArena arena;
cbm_arena_init(&arena);
CBMTypeRegistry reg, empty;
cbm_registry_init(&reg, &arena);
cbm_registry_init(&empty, &arena);
bool ok = true;
for (size_t i = 0; i < sizeof(qns) / sizeof(qns[0]); i++)
ok = pylsp_range_add_type(&reg, qns[i], "Shared") &&
pylsp_prefix_add(&reg, qns[i], i == 1 ? "Receiver" : NULL) && ok;
for (int indexed = 0; indexed < 2; indexed++) {
if (indexed) {
cbm_registry_build_type_qn_order(&reg, &arena);
cbm_registry_build_func_qn_order(&reg, &arena);
ok = reg.type_qn_sorted && reg.func_qn_sorted &&
reg.type_qn_sorted_count == reg.type_count - 1 &&
reg.type_qn_sorted_upto == reg.type_count && ok;
if (reg.type_qn_sorted) {
for (int i = 0; i < reg.type_qn_sorted_count; i++) {
const CBMRegistryTypeQNEntry *e = &reg.type_qn_sorted[i];
ok = e->type_index >= 0 && e->type_index < reg.type_count &&
e->qualified_name == reg.types[e->type_index].qualified_name && ok;
}
}
}
for (int types = 0; types < 2; types++) {
for (int direct = 0; direct < 2; direct++) {
for (size_t p = 0; p < sizeof(prefixes) / sizeof(prefixes[0]); p++) {
CBMArena scratch;
cbm_arena_init(&scratch);
ok = pylsp_range_oracle(&reg, types != 0, prefixes[p], direct != 0, &scratch,
NULL) &&
pylsp_range_oracle(&reg, types != 0, prefixes[p], direct != 0, NULL,
NULL) &&
pylsp_range_oracle(&empty, types != 0, prefixes[p], direct != 0, &scratch,
NULL) &&
pylsp_range_oracle(NULL, types != 0, prefixes[p], direct != 0, &scratch,
NULL) &&
ok;
cbm_arena_destroy(&scratch);
}
}
}
}
ok = !cbm_registry_types_with_qn_prefix(&reg, "p.", true, &arena, NULL) &&
!cbm_registry_funcs_with_qn_prefix(&reg, "p.", false, &arena, NULL) &&
cbm_qn_prefix_iter_next(NULL) == -1 &&
cbm_registry_has_func_qn_prefix(&reg, "p.foo.", NULL) && ok;
cbm_arena_destroy(&arena);
ASSERT_TRUE(ok);
PASS();
}
TEST(pylsp_range_growth_rebuild_seal_and_independent_failure) {
CBMArena arena;
cbm_arena_init(&arena);
CBMTypeRegistry reg, sealed;
cbm_registry_init(&reg, &arena);
cbm_registry_init(&sealed, &arena);
bool ok = pylsp_range_add_type(&reg, "p.old", "Old") && pylsp_prefix_add(&reg, "p.old", NULL);
cbm_registry_build_type_qn_order(&reg, &arena);
cbm_registry_build_func_qn_order(&reg, &arena);
const CBMRegisteredType *old_types = reg.types;
const CBMRegisteredFunc *old_funcs = reg.funcs;
for (int i = 0; i < 140; i++) {
const char *qn = cbm_arena_sprintf(&arena, "p.tail%03d", i);
ok = qn && pylsp_range_add_type(&reg, qn, "Tail") && pylsp_prefix_add(&reg, qn, NULL) && ok;
}
ok = pylsp_range_add_type(&reg, NULL, "NoQN") && pylsp_prefix_add(&reg, NULL, NULL) &&
reg.type_count == 142 && reg.func_count == 142 && reg.types != old_types &&
reg.funcs != old_funcs && reg.type_qn_sorted_upto == 1 && reg.func_qn_sorted_upto == 1 &&
pylsp_range_oracle(&reg, true, "p.", true, &arena, NULL) &&
pylsp_range_oracle(&reg, false, "p.", true, &arena, NULL) && ok;
if (reg.type_count == 142 && reg.func_count == 142) {
/* The previous queries ended before either registered identity changes. */
reg.types[0].qualified_name = "p.renamed";
reg.funcs[0].qualified_name = "p.renamed";
cbm_registry_build_func_qn_order(&reg, &arena);
const CBMRegistryQNEntry *func_order = reg.func_qn_sorted;
cbm_registry_test_type_qn_order_fail_after(0);
cbm_registry_build_type_qn_order(&reg, &arena);
cbm_registry_test_type_qn_order_fail_after(-1);
ok = func_order && reg.func_qn_sorted == func_order && !reg.type_qn_sorted &&
reg.type_qn_sorted_count == 0 && reg.type_qn_sorted_upto == 0 &&
pylsp_range_oracle(&reg, true, "p.", true, &arena, NULL) && ok;
cbm_registry_test_type_qn_order_fail_after(1);
cbm_registry_build_type_qn_order(&reg, &arena);
bool first_built = reg.type_qn_sorted != NULL;
cbm_registry_build_type_qn_order(&reg, &arena);
cbm_registry_test_type_qn_order_fail_after(-1);
ok = first_built && !reg.type_qn_sorted && reg.func_qn_sorted == func_order && ok;
cbm_registry_build_type_qn_order(&reg, &arena);
const CBMRegistryTypeQNEntry *type_order = reg.type_qn_sorted;
cbm_registry_test_qn_order_fail_after(0);
cbm_registry_build_func_qn_order(&reg, &arena);
cbm_registry_test_qn_order_fail_after(-1);
ok = type_order && reg.type_qn_sorted == type_order && !reg.func_qn_sorted &&
pylsp_range_oracle(&reg, false, "p.", true, &arena, NULL) && ok;
cbm_registry_build_func_qn_order(&reg, &arena);
func_order = reg.func_qn_sorted;
reg.read_only = true;
cbm_registry_test_type_qn_order_fail_after(0);
cbm_registry_test_qn_order_fail_after(0);
cbm_registry_build_type_qn_order(&reg, &arena);
cbm_registry_build_func_qn_order(&reg, &arena);
pylsp_range_reset_failures();
cbm_registry_test_qn_order_fail_after(-1);
cbm_registry_add_type(&reg, (CBMRegisteredType){.qualified_name = "forbidden.Type"});
cbm_registry_add_func(&reg, (CBMRegisteredFunc){.qualified_name = "forbidden.call"});
ok = reg.type_qn_sorted == type_order && reg.func_qn_sorted == func_order &&
reg.type_count == 142 && reg.func_count == 142 &&
pylsp_range_oracle(&reg, true, "p.", true, &arena, NULL) &&
pylsp_range_oracle(&reg, false, "p.", true, &arena, NULL) && ok;
}
ok = pylsp_range_add_type(&sealed, "p.sealed", "Sealed") &&
pylsp_prefix_add(&sealed, "p.sealed", NULL) && ok;
sealed.read_only = true;
cbm_registry_build_type_qn_order(&sealed, &arena);
cbm_registry_build_func_qn_order(&sealed, &arena);
ok = !sealed.type_qn_sorted && !sealed.func_qn_sorted &&
pylsp_range_oracle(&sealed, true, "p.", true, &arena, NULL) &&
pylsp_range_oracle(&sealed, false, "p.", true, &arena, NULL) && ok;
pylsp_range_reset_failures();
cbm_arena_destroy(&arena);
ASSERT_TRUE(ok);
PASS();
}
TEST(pylsp_range_invalid_snapshot_uses_complete_scan) {
CBMArena arena;
cbm_arena_init(&arena);
CBMTypeRegistry reg;
cbm_registry_init(&reg, &arena);
bool ok = pylsp_range_add_type(&reg, "p.Z", "Z") && pylsp_range_add_type(&reg, "p.A", "A") &&
pylsp_prefix_add(&reg, "p.Z", NULL) && pylsp_prefix_add(&reg, "p.A", NULL);
cbm_registry_build_type_qn_order(&reg, &arena);
cbm_registry_build_func_qn_order(&reg, &arena);
ok = reg.type_count == 2 && reg.func_count == 2 && reg.type_qn_sorted && reg.func_qn_sorted &&
ok;
size_t allocated = cbm_arena_total(&arena);
for (int types = 0; types < 2; types++) {
int *covered = types ? &reg.type_qn_sorted_upto : &reg.func_qn_sorted_upto;
int *count = types ? &reg.type_qn_sorted_count : &reg.func_qn_sorted_count;
int saved_covered = *covered, saved_count = *count;
for (int invalid = 0; invalid < 3; invalid++) {
*covered = invalid == 0 ? 3 : saved_covered;
*count = invalid == 1 ? -1 : (invalid == 2 ? 3 : saved_count);
int expected_covered = *covered, expected_count = *count;
uint64_t visits = 0;
ok = pylsp_range_oracle(&reg, types != 0, "p.", true, &arena, &visits) && visits >= 2 &&
*covered == expected_covered && *count == expected_count &&
cbm_arena_total(&arena) == allocated && ok;
*covered = saved_covered;
*count = saved_count;
}
ok = pylsp_range_oracle(&reg, types != 0, "p.", true, &arena, NULL) && ok;
allocated = cbm_arena_total(&arena);
}
cbm_arena_destroy(&arena);
ASSERT_TRUE(ok);
PASS();
}
TEST(pylsp_range_block_failure_preserves_each_link_and_tail) {
CBMArena arena;
cbm_arena_init(&arena);
CBMTypeRegistry links[3];
bool ok = true;
for (int j = 0; j < 3; j++) {
cbm_registry_init(&links[j], &arena);
ok = pylsp_range_add_type(&links[j], "p.Z", "Z") &&
pylsp_range_add_type(&links[j], "p.A", "A") &&
pylsp_prefix_add(&links[j], "p.Z", NULL) && pylsp_prefix_add(&links[j], "p.A", NULL) &&
ok;
cbm_registry_build_type_qn_order(&links[j], &arena);
cbm_registry_build_func_qn_order(&links[j], &arena);
ok = links[j].type_qn_sorted && links[j].func_qn_sorted &&
pylsp_range_add_type(&links[j], "p.Tail", "Tail") &&
pylsp_prefix_add(&links[j], "p.Tail", NULL) && ok;
links[j].fallback = j < 2 ? &links[j + 1] : NULL;
}
for (int types = 0; types < 2; types++) {
for (int fail = 0; fail < 3; fail++) {
CBMArena scratch;
cbm_arena_init(&scratch);
cbm_registry_test_qn_prefix_block_fail_after(fail);
for (int j = 0; j < 3; j++) {
uint64_t visits = 0;
ok = pylsp_range_oracle(&links[j], types != 0, "p.", true, &scratch, &visits) &&
visits > 0 && ok;
}
cbm_registry_test_qn_prefix_block_fail_after(-1);
cbm_arena_destroy(&scratch);
}
}
cbm_arena_destroy(&arena);
ASSERT_TRUE(ok);
PASS();
}
TEST(pylsp_range_direct_subtree_pruning_has_absolute_bound) {
enum { NESTED = 1024, BOUND = 100 };
CBMArena arena;
cbm_arena_init(&arena);
CBMTypeRegistry reg;
cbm_registry_init(&reg, &arena);
bool ok = true;
for (int i = NESTED - 1; i >= 0; i--) {
const char *qn = cbm_arena_sprintf(&arena, "p.foo.A.n%04d", i);
ok = qn && pylsp_range_add_type(&reg, qn, "Nested") &&
pylsp_prefix_add(&reg, qn, "p.foo.A") && ok;
}
static const char *const direct[] = {"p.foo.B", "p.foo.A-b", "p.foo.AB"};
for (size_t i = 0; i < sizeof(direct) / sizeof(direct[0]); i++)
ok = pylsp_range_add_type(&reg, direct[i], "Direct") &&
pylsp_prefix_add(&reg, direct[i], NULL) && ok;
cbm_registry_build_type_qn_order(&reg, &arena);
cbm_registry_build_func_qn_order(&reg, &arena);
ok = reg.type_count == NESTED + 3 && reg.func_count == NESTED + 3 && reg.type_qn_sorted &&
reg.func_qn_sorted && ok;
for (int types = 0; types < 2; types++) {
uint64_t indexed = 0, linear = 0, miss = 0;
ok = pylsp_range_oracle(&reg, types != 0, "p.foo.", true, &arena, &indexed) &&
indexed > 0 && indexed <= BOUND &&
pylsp_range_oracle(&reg, types != 0, "zzzz.", true, &arena, &miss) && miss > 0 &&
miss <= BOUND && ok;
cbm_registry_test_qn_prefix_block_fail_after(0);
ok = pylsp_range_oracle(&reg, types != 0, "p.foo.", true, &arena, &linear) &&
linear >= NESTED + 3 && linear > BOUND && ok;
cbm_registry_test_qn_prefix_block_fail_after(-1);
}
cbm_arena_destroy(&arena);
ASSERT_TRUE(ok);
PASS();
}
/* Test scratch storage must be gone before these retained bindings are read. */
TEST(pylsp_range_class_binding_keeps_order_shadow_and_package_filters) {
static const char source[] = "class Local:\n pass\n";
TSParser *parser = ts_parser_new();
bool language = parser && ts_parser_set_language(parser, tree_sitter_python());
TSTree *tree =
language ? ts_parser_parse_string(parser, NULL, source, (uint32_t)strlen(source)) : NULL;
bool ok = tree && !ts_node_has_error(ts_tree_root_node(tree));
for (int package = 0; tree && package < 2; package++) {
for (int failure = 0; failure < 6; failure++) {
CBMArena owner, arena;
cbm_arena_init(&owner);
cbm_arena_init(&arena);
CBMTypeRegistry head, middle, tail;
cbm_registry_init(&head, &owner);
cbm_registry_init(&middle, &owner);
cbm_registry_init(&tail, &owner);
head.fallback = &middle;
middle.fallback = &tail;
bool built = pylsp_range_add_type(&head, "test.pkg.Head", "Shared") &&
pylsp_range_add_type(&head, "test.pkg.Same", "HeadSame") &&
pylsp_range_add_type(&head, "test.pkg.Local", "Local") &&
pylsp_range_add_type(&middle, "test.pkg.Middle", "Shared") &&
pylsp_range_add_type(&middle, "test.pkg.Same", "Masked") &&
pylsp_range_add_type(&middle, "test.pkg.Deep", "MiddleOnly") &&
pylsp_range_add_type(&tail, "test.pkg.Tail", "Shared") &&
pylsp_range_add_type(&tail, "test.pkg.Deep", "TailOnly") &&
pylsp_range_add_type(&tail, "test.pkg.sub.Hidden", "Hidden");
if (failure == 5)
cbm_registry_test_type_qn_order_fail_after(0);
CBMTypeRegistry *links[] = {&head, &middle, &tail};
for (int i = 0; i < 3; i++)
cbm_registry_build_type_qn_order(links[i], &owner);
cbm_registry_test_type_qn_order_fail_after(-1);
if (failure != 5)
built =
head.type_qn_sorted && middle.type_qn_sorted && tail.type_qn_sorted && built;
else
built =
!head.type_qn_sorted && !middle.type_qn_sorted && !tail.type_qn_sorted && built;
/* The two fallback Deep entries remain visible: only the head shadows. */
bool shadowed = built && cbm_registry_type_shadowed(&head, &middle, 1) &&
!cbm_registry_type_shadowed(&head, &tail, 1) &&
!cbm_registry_type_shadowed(&head, &head, 1);
PyLSPContext ctx;
CBMResolvedCallArray out = {0};
py_lsp_init(&ctx, &arena, source, (int)strlen(source), &head,
package ? "test.pkg.__init__" : "test.pkg", &out);
if (failure == 1 || failure == 2)
cbm_registry_test_qn_prefix_block_fail_after(failure - 1);
if (failure == 3)
cbm_py_test_module_prefix_fail_after(0);
if (failure == 4)
cbm_py_test_module_scratch_fail_after(0);
cbm_py_lsp_test_bind_external_classes(&ctx, ts_tree_root_node(tree));
pylsp_range_reset_failures();
ok = shadowed && !ctx.callable_value_proof_disabled &&
ctx.eval_failure == CBM_LSP_COMPLETE &&
pylsp_init_named_type(pylsp_local_type(ctx.current_scope, "Shared"),
"test.pkg.Tail") &&
pylsp_init_named_type(pylsp_local_type(ctx.current_scope, "HeadSame"),
"test.pkg.Same") &&
pylsp_init_named_type(pylsp_local_type(ctx.current_scope, "MiddleOnly"),
"test.pkg.Deep") &&
pylsp_init_named_type(pylsp_local_type(ctx.current_scope, "TailOnly"),
"test.pkg.Deep") &&
!cbm_scope_lookup_local(ctx.current_scope, "Masked") &&
!cbm_scope_lookup_local(ctx.current_scope, "Local") && ok;
const CBMVarBinding *hidden = cbm_scope_lookup_local(ctx.current_scope, "Hidden");
ok = (package ? hidden == NULL
: pylsp_init_named_type(hidden ? hidden->type : NULL,
"test.pkg.sub.Hidden")) &&
ok;
cbm_arena_destroy(&arena);
cbm_arena_destroy(&owner);
}
}
pylsp_range_reset_failures();
if (tree)
ts_tree_delete(tree);
if (parser)
ts_parser_delete(parser);
ASSERT_TRUE(ok);
PASS();
}
TEST(pylsp_range_wildcard_and_scratch_failures_keep_exact_later_calls) {
static const char source[] = "external = External()\nexternal.ping()\n"
"from vendor import *\nshadow_only(72)\nExternal(74)\n"
"def later(v):\n return v\nlater(73)\n";
CBMFileResult *raw = extract_py(source);
TSParser *parser = ts_parser_new();
bool language = parser && ts_parser_set_language(parser, tree_sitter_python());
TSTree *tree =
language ? ts_parser_parse_string(parser, NULL, source, (uint32_t)strlen(source)) : NULL;
bool healthy = raw && !raw->has_error && !raw->parse_incomplete && !raw->lsp_skipped && tree &&
!ts_node_has_error(ts_tree_root_node(tree)) &&
pylsp_module_has_def(raw, "test.main.later");
bool ok = healthy;
for (int failure = 0; healthy && failure < 8; failure++) {
CBMArena owner, arena;
cbm_arena_init(&owner);
cbm_arena_init(&arena);
CBMTypeRegistry head, tail;
cbm_registry_init(&head, &owner);
cbm_registry_init(&tail, &owner);
head.fallback = &tail;
bool built = pylsp_range_add_type(&head, "test.main.External", "External") &&
pylsp_range_add_type(&tail, "test.main.Other", "Other");
cbm_registry_add_func(&head,
(CBMRegisteredFunc){.qualified_name = "test.main.External.ping",
.short_name = "ping",
.receiver_type = "test.main.External"});
cbm_registry_add_func(
&head, (CBMRegisteredFunc){.qualified_name = "test.main.later", .short_name = "later"});
/* A head method must not suppress invalidation of a fallback free function. */
cbm_registry_add_func(&head, (CBMRegisteredFunc){.qualified_name = "test.main.hidden",
.short_name = "hidden",
.receiver_type = "test.main.External"});
cbm_registry_add_func(&tail, (CBMRegisteredFunc){.qualified_name = "test.main.hidden",
.short_name = "shadow_only"});
cbm_registry_build_type_qn_order(&head, &owner);
cbm_registry_build_type_qn_order(&tail, &owner);
cbm_registry_build_func_qn_order(&head, &owner);
cbm_registry_build_func_qn_order(&tail, &owner);
built = built && head.func_count == 3 && tail.func_count == 1 && head.type_qn_sorted &&
tail.type_qn_sorted && head.func_qn_sorted && tail.func_qn_sorted;
head.read_only = true;
tail.read_only = true;
PyLSPContext ctx;
CBMResolvedCallArray out = {0};
py_lsp_init(&ctx, &arena, source, (int)strlen(source), &head, "test.main", &out);
if (failure >= 1 && failure <= 4)
cbm_registry_test_qn_prefix_block_fail_after(failure - 1);
if (failure == 5)
cbm_py_test_module_prefix_fail_after(0);
if (failure == 6)
cbm_py_test_module_scratch_fail_after(0);
if (failure == 7)
cbm_py_test_module_prefix_fail_after(1);
uint64_t before =
atomic_load_explicit(&cbm_py_wildcard_invalidations, memory_order_relaxed);
py_lsp_process_file(&ctx, ts_tree_root_node(tree));
pylsp_range_reset_failures();
const CBMVarBinding *shadow = cbm_scope_lookup_local(ctx.current_scope, "shadow_only");
ok = built && ctx.eval_failure == CBM_LSP_COMPLETE && !ctx.callable_value_proof_disabled &&
shadow && cbm_type_is_unknown(shadow->type) &&
atomic_load_explicit(&cbm_py_wildcard_invalidations, memory_order_relaxed) > before &&
pylsp_module_site_joins(&raw->calls, &out, source, "external.ping()", "test.main",
"test.main.External.ping") &&
pylsp_module_site_joins(&raw->calls, &out, source, "later(73)", "test.main",
"test.main.later") &&
pylsp_prefix_site_unresolved(&raw->calls, &out, source, "shadow_only(72)",
"test.main") &&
pylsp_prefix_site_unresolved(&raw->calls, &out, source, "External(74)", "test.main") &&
ok;
cbm_arena_destroy(&arena);
cbm_arena_destroy(&owner);
}
pylsp_range_reset_failures();
if (tree)
ts_tree_delete(tree);
if (parser)
ts_parser_delete(parser);
if (raw)
cbm_free_result(raw);
ASSERT_TRUE(ok);
PASS();
}
SUITE(py_lsp) {
RUN_TEST(pylsp_range_order_boundaries_and_null_contract);
RUN_TEST(pylsp_range_growth_rebuild_seal_and_independent_failure);
RUN_TEST(pylsp_range_invalid_snapshot_uses_complete_scan);
RUN_TEST(pylsp_range_block_failure_preserves_each_link_and_tail);
RUN_TEST(pylsp_range_direct_subtree_pruning_has_absolute_bound);
RUN_TEST(pylsp_range_class_binding_keeps_order_shadow_and_package_filters);
RUN_TEST(pylsp_range_wildcard_and_scratch_failures_keep_exact_later_calls);
RUN_TEST(pylsp_prefix_order_matches_iterator_and_chain_shadowing);
RUN_TEST(pylsp_prefix_order_survives_array_growth_rebuild_and_seal);
RUN_TEST(pylsp_prefix_order_allocation_failure_preserves_oracle);
RUN_TEST(pylsp_prefix_order_has_absolute_comparison_bound);
RUN_TEST(pylsp_prefix_order_preserves_exact_selection_and_ambiguity);
RUN_TEST(pylsp_prefix_raw_per_file_and_shared_keep_exact_calls);
RUN_TEST(pylsp_classbody_calls_join_extracted_caller);
RUN_TEST(pylsp_classbody_names_visible_to_later_statements_only);
RUN_TEST(pylsp_decorator_and_base_calls_join_extracted_caller);
RUN_TEST(pylsp_nested_and_local_classbody_calls_join_extracted_caller);
RUN_TEST(pylsp_class_comprehensions_separate_iterable_and_body_scopes);
RUN_TEST(pylsp_class_legacy_dispatch_cannot_bypass_or_publish_shadows);
RUN_TEST(pylsp_class_frame_failure_restores_context_and_stops);
RUN_TEST(pylsp_class_frame_failure_rolls_back_raw_and_cross_outputs);
RUN_TEST(pylsp_init_scope_context_and_helper);
RUN_TEST(pylsp_init_scope_raw_exact_occurrences_and_annotations);
RUN_TEST(pylsp_init_scope_cross_signatures_keep_definition_language);
RUN_TEST(pylsp_init_scope_external_classes_and_cross_exact_joins);
RUN_TEST(pylsp_init_scope_failure_preserves_outputs_and_wrappers);
RUN_TEST(pylsp_init_scope_shared_failure_keeps_arena_and_needed_fallback);
RUN_TEST(pylsp_module_callers_join_exact_raw_occurrences);
RUN_TEST(pylsp_module_named_lambda_keeps_exact_dunder_pair);
RUN_TEST(pylsp_module_cross_callers_own_input_and_join_exact_sites);
RUN_TEST(pylsp_issue1527_depth_failure_reports_all_entry_points);
RUN_TEST(pylsp_issue1527_cross_memo_failure_fallback_reports_error);
RUN_TEST(pylsp_issue1527_cross_memo_failure_preserves_completed_rows);
RUN_TEST(pylsp_issue1527_root_class_name_failure_binds_no_external_class);
RUN_TEST(pylsp_issue1527_memo_and_depth_failures_stop_work);
RUN_TEST(pylsp_issue1527_memo_allocation_failure_is_reported);
RUN_TEST(pylsp_issue1527_long_binary_chain_is_not_nesting);
RUN_TEST(pylsp_issue1527_long_receiver_chain_is_not_nesting);
RUN_TEST(pylsp_scale_work_is_linear);
RUN_TEST(pylsp_cross_registry_build_work_is_linear);
/* Phase 2 — smoke */
RUN_TEST(pylsp_smoke_empty);
RUN_TEST(pylsp_smoke_one_function);
RUN_TEST(pylsp_smoke_one_class);
RUN_TEST(pylsp_no_crash_on_syntax_error);
RUN_TEST(pylsp_smoke_imports_passed_through);
/* Phase 3 — imports → scope */
RUN_TEST(pylsp_import_simple);
RUN_TEST(pylsp_import_aliased);
RUN_TEST(pylsp_import_from);
RUN_TEST(pylsp_import_from_aliased);
RUN_TEST(pylsp_import_relative_one_dot);
RUN_TEST(pylsp_import_relative_two_dots);
RUN_TEST(pylsp_import_star_best_effort);
RUN_TEST(pylsp_import_typing_only_still_binds);
RUN_TEST(pylsp_import_multi_pass_through_extract_file);
/* Phases 4-6 — bindings + expression typing + method dispatch */
RUN_TEST(pylsp_direct_function_call);
RUN_TEST(pylsp_method_call_simple);
RUN_TEST(pylsp_method_via_self);
RUN_TEST(pylsp_constructor_call_returns_instance);
RUN_TEST(pylsp_constructor_call_targets_init);
RUN_TEST(pylsp_constructor_same_file_subclass_own_init_only);
RUN_TEST(pylsp_constructor_without_init_stays_on_class);
RUN_TEST(pylsp_callable_instance_call_skips_init);
RUN_TEST(pylsp_constructor_skips_builtin_init);
RUN_TEST(pylsp_method_via_inheritance);
RUN_TEST(pylsp_no_false_positive_on_unknown_method);
/* Phases 7-8 — decorators, super(), multi-inheritance */
RUN_TEST(pylsp_decorated_function_resolves);
RUN_TEST(pylsp_classmethod_resolves);
RUN_TEST(pylsp_staticmethod_resolves);
RUN_TEST(pylsp_dataclass_constructor);
RUN_TEST(pylsp_super_call);
RUN_TEST(pylsp_multi_inheritance_first_base);
RUN_TEST(pylsp_pep695_generic_class);
/* Phase 9 — cross-file + batch */
RUN_TEST(pylsp_crossfile_method_dispatch);
RUN_TEST(pylsp_crossfile_receiver_through_typed_field);
RUN_TEST(pylsp_crossfile_typed_field_shared_registry);
RUN_TEST(pylsp_fused_self_attr_chain_via_overlay);
RUN_TEST(pylsp_cross_registry_interleaved_order);
RUN_TEST(pylsp_crossfile_classmethod_on_class_issue228);
RUN_TEST(pylsp_crossfile_inheritance);
RUN_TEST(pylsp_crossfile_constructor_targets_init);
RUN_TEST(pylsp_crossfile_inherited_init_stays_on_class);
RUN_TEST(pylsp_batch_two_files);
RUN_TEST(pylsp_from_import_alias_equal_module_leaf_targets_imported_member);
RUN_TEST(pylsp_project_prefixed_direct_alias_same_tail_is_not_from_import_reference);
RUN_TEST(pylsp_competing_import_forms_for_same_local_fail_closed);
RUN_TEST(pylsp_later_plain_import_rebinding_same_local_fails_closed);
RUN_TEST(pylsp_top_level_function_shadows_imported_callable);
RUN_TEST(pylsp_decorated_top_level_function_shadows_imported_callable);
RUN_TEST(pylsp_later_import_rebinds_top_level_function_name);
RUN_TEST(pylsp_batch_cross_file_callable_value_preserves_exact_reference_site);
RUN_TEST(pylsp_same_target_calls_have_distinct_exact_semantic_sites);
RUN_TEST(pylsp_batch_preserves_two_same_target_semantic_sites);
RUN_TEST(pylsp_scratch_cross_dunder_carrier_survives_copy);
/* Phase 10 — stdlib resolution */
RUN_TEST(pylsp_stdlib_os_getcwd);
RUN_TEST(pylsp_stdlib_collections_defaultdict);
RUN_TEST(pylsp_stdlib_pathlib_path_method);
RUN_TEST(pylsp_stdlib_logging_getlogger);
/* Round 1 — parity push */
RUN_TEST(pylsp_round1_dotted_import_walk);
RUN_TEST(pylsp_round1_dotted_import_alias_matching_root);
RUN_TEST(pylsp_round1_typing_cast);
RUN_TEST(pylsp_round1_assert_type_passthrough);
RUN_TEST(pylsp_round1_forward_reference);
RUN_TEST(pylsp_round1_self_return_chains);
RUN_TEST(pylsp_round1_generic_subscript_annotation);
/* Round 2 — narrowing */
RUN_TEST(pylsp_round2_isinstance_narrow);
RUN_TEST(pylsp_round2_is_not_none_narrow);
RUN_TEST(pylsp_round2_isinstance_no_false_positive_in_else);
/* Round 2/3 — walrus, comprehension, optional-narrow */
RUN_TEST(pylsp_round2_narrow_after_call);
RUN_TEST(pylsp_round2_walrus_binds);
RUN_TEST(pylsp_round3_listcomp_element_method);
RUN_TEST(pylsp_round3_for_loop_element_method);
RUN_TEST(pylsp_round3_optional_narrow_with_union);
/* Round 3 — match/case + async */
RUN_TEST(pylsp_round3_match_case_class_pattern);
RUN_TEST(pylsp_round3_async_await_pass_through);
/* Round 4 — instance attribute typing */
RUN_TEST(pylsp_round4_instance_attribute_init);
RUN_TEST(pylsp_round4_instance_attribute_class_annotation);
/* Round 5 — subscript, super().__init__, operator dunders */
RUN_TEST(pylsp_round5_dict_subscript_value_type);
RUN_TEST(pylsp_round5_list_subscript_value_type);
RUN_TEST(pylsp_round5_super_init);
RUN_TEST(pylsp_dunder_same_leaf_occurrences_join_by_exact_site);
RUN_TEST(pylsp_lambda_rewalk_dedupes_dunder_occurrence);
RUN_TEST(pylsp_recursive_lambda_rewalk_is_bounded);
RUN_TEST(pylsp_lambda_rewalk_guard_released_between_calls);
RUN_TEST(pylsp_ordinary_same_leaf_calls_join_by_exact_site);
/* Round 6 — generators, dataclasses, properties */
RUN_TEST(pylsp_round6_generator_yields_iterable);
RUN_TEST(pylsp_round6_dataclass_field_access);
RUN_TEST(pylsp_round6_property_access_chains);
/* Evaluator guards — #710/#720 (distilled PRs #732 + #758) */
RUN_TEST(pylsp_issue710_deep_call_chain_resolves);
RUN_TEST(pylsp_issue710_heterogeneous_receiver_chain);
RUN_TEST(pylsp_issue1527_chain_statements_resolve_completely);
RUN_TEST(pylsp_issue1527_large_module_resolves_every_call);
}
Reference in new issue View Git Blame Copy Permalink