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

fix(python): resolve class-body and decorator calls

Browse Source 
Signed-off-by: Martin Vogel <martin.vogel.tech@gmail.com>
This commit is contained in:
Martin Vogel committed 2026-10-12 22:05:21 +02:00
1 parent 82c2b8320c
commit 78aea85b68
4 files changed
+863 -14
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

internal/cbm/lsp/py_lsp.c
+230 -12
View File
@@ -44,6 +44,10 @@
static _Thread_local int py_test_memo_allocations_left = -1;
static _Thread_local bool py_test_depth_failure;
static _Thread_local int py_test_scope_allocations_left = -1;
static _Thread_local int py_test_class_frames_left = -1;
void cbm_py_lsp_test_class_frame_fail_after(int successful_allocations) {
py_test_class_frames_left = successful_allocations;
}
void cbm_py_lsp_test_scope_fail_after(int successful_allocations) {
py_test_scope_allocations_left = successful_allocations;
}
@@ -227,6 +231,7 @@ static void py_resolve_calls_in(PyLSPContext *ctx, TSNode node) {
static const CBMType *py_eval_expr_type(PyLSPContext *ctx, TSNode node);
static const CBMType *py_eval_expr_type_uncached(PyLSPContext *ctx, TSNode node);
static void py_process_statement(PyLSPContext *ctx, TSNode node);
static void py_exec_class_definition(PyLSPContext *ctx, TSNode class_node);
static void py_invalidate_possible_bindings(PyLSPContext *ctx, TSNode node, int depth);
static const CBMRegisteredFunc *py_lookup_attribute(PyLSPContext *ctx, const char *type_qn,
const char *member_name);
@@ -288,6 +293,25 @@ static CBMScope *py_scope_push_checked(PyLSPContext *ctx) {
return child;
}
/* New class execution frames must never fall back to their parent: body
* bindings would escape and subsequent calls could use those stale bindings. */
static CBMScope *py_class_frame_push(PyLSPContext *ctx, CBMScope *parent) {
#ifdef CBM_ENABLE_TEST_SEAMS
if (py_test_class_frames_left == 0) {
ctx->eval_failure = CBM_LSP_SCOPE_FAILED;
return NULL;
}
if (py_test_class_frames_left > 0)
py_test_class_frames_left--;
#endif
CBMScope *frame = cbm_scope_push(ctx->arena, parent);
if (!frame || frame == parent) {
ctx->eval_failure = CBM_LSP_SCOPE_FAILED;
return NULL;
}
return frame;
}
static void py_scope_clear_callable(PyLSPContext *ctx, const char *name) {
if (cbm_scope_update_callable(ctx->current_scope, name, NULL)) {
ctx->type_cache_gen++;
@@ -2565,13 +2589,16 @@ static void py_process_statement(PyLSPContext *ctx, TSNode node) {
} else {
py_scope_bind(ctx, name, rhs_type);
}
// Class attributes must not replace file-wide dispatch entries.
// Lambda registry: `fn = lambda x: ...`.
if (!ts_node_is_null(right) && strcmp(ts_node_type(right), "lambda") == 0) {
if (!ctx->class_body_outer_scope && !ts_node_is_null(right) &&
strcmp(ts_node_type(right), "lambda") == 0) {
py_register_lambda(ctx, name, right);
}
// Dict literal dispatch table: `funcs = {"a": foo, "b": bar}`
// with all values being known function QNs.
if (!ts_node_is_null(right) && strcmp(ts_node_type(right), "dictionary") == 0) {
if (!ctx->class_body_outer_scope && !ts_node_is_null(right) &&
strcmp(ts_node_type(right), "dictionary") == 0) {
py_register_dict_literal(ctx, name, right);
}
}
@@ -2779,7 +2806,11 @@ static void py_emit_call_for(PyLSPContext *ctx, TSNode call_node) {
// Walk the lambda body for any nested call sites with the params
// bound to the call's arg types. Resolved calls get caller_qn
// set to a synthetic <lambda> child of the enclosing function.
TSNode lambda_node = py_lookup_lambda(ctx, fname);
/* These legacy entries have no lexical frame identity. During class
* execution, leave lambda replay unresolved instead of bypassing a
* local shadow or running the lambda against class locals. */
TSNode lambda_node =
ctx->class_body_outer_scope ? (TSNode){0} : py_lookup_lambda(ctx, fname);
if (!ts_node_is_null(lambda_node)) {
TSNode params = ts_node_child_by_field_name(lambda_node, "parameters", 10);
TSNode body = ts_node_child_by_field_name(lambda_node, "body", 4);
@@ -2890,7 +2921,7 @@ static void py_emit_call_for(PyLSPContext *ctx, TSNode call_node) {
// Subscript-as-call: `funcs["a"]()` where funcs is a registered
// dict-literal dispatch table.
if (strcmp(fk, "subscript") == 0) {
if (!ctx->class_body_outer_scope && strcmp(fk, "subscript") == 0) {
TSNode container = ts_node_child_by_field_name(fn, "value", 5);
TSNode key = ts_node_child_by_field_name(fn, "subscript", 9);
if (!ts_node_is_null(container) && !ts_node_is_null(key) &&
@@ -3465,6 +3496,65 @@ static void py_emit_dunder_call(PyLSPContext *ctx, const CBMType *recv, const ch
}
}
/* A class comprehension evaluates its leftmost iterable in the class frame,
* but the implicit body scope skips class locals. Nested comprehensions keep
* the surrounding comprehension frame. Keep class execution marked active so
* legacy bare-name dispatch cannot bypass either frame's lexical bindings. */
static void py_resolve_class_comprehension(PyLSPContext *ctx, TSNode node) {
if (ctx->eval_failure)
return;
PyKids kids = py_kids(ctx->arena, node);
uint32_t first_for = kids.n;
TSNode first_right = {0};
for (uint32_t i = 0; i < kids.n && !ctx->eval_failure; i++) {
TSNode child = py_kid(&kids, i);
if (strcmp(ts_node_type(child), "for_in_clause") == 0) {
first_for = i;
first_right = ts_node_child_by_field_name(child, "right", 5);
break;
}
}
if (first_for == kids.n)
return;
const CBMType *first_type = py_eval_expr_type(ctx, first_right);
py_resolve_calls_in(ctx, first_right);
if (ctx->eval_failure)
return;
const CBMType *first_elem = py_iterable_element_type(ctx, first_type);
CBMScope *saved_scope = ctx->current_scope;
bool saved_comprehension = ctx->class_body_in_comprehension;
CBMScope *parent = saved_comprehension ? saved_scope : ctx->class_body_outer_scope;
CBMScope *frame = py_class_frame_push(ctx, parent);
if (!frame)
return;
py_scope_restore(ctx, frame);
ctx->class_body_in_comprehension = true;
/* Clauses precede the body at evaluation time, although tree-sitter stores
* the body first. Walk each iterable/filter only once, before later binds. */
for (uint32_t i = first_for; i < kids.n && !ctx->eval_failure; i++) {
TSNode child = py_kid(&kids, i);
if (strcmp(ts_node_type(child), "for_in_clause") == 0) {
TSNode left = ts_node_child_by_field_name(child, "left", 4);
const CBMType *elem = first_elem;
if (i != first_for) {
TSNode right = ts_node_child_by_field_name(child, "right", 5);
const CBMType *iter = py_eval_expr_type(ctx, right);
py_resolve_calls_in(ctx, right);
if (ctx->eval_failure)
break;
elem = py_iterable_element_type(ctx, iter);
}
py_bind_for_target(ctx, left, elem);
} else {
py_resolve_calls_in(ctx, child);
}
}
for (uint32_t i = 0; i < first_for && !ctx->eval_failure; i++)
py_resolve_calls_in(ctx, py_kid(&kids, i));
ctx->class_body_in_comprehension = saved_comprehension;
py_scope_restore(ctx, saved_scope);
}
static void py_resolve_calls_in_inner(PyLSPContext *ctx, TSNode node) {
if (!ctx || ts_node_is_null(node))
return;
@@ -3667,6 +3757,10 @@ static void py_resolve_calls_in_inner(PyLSPContext *ctx, TSNode node) {
// iterable's element type, then walk inner expressions.
if (strcmp(k, "list_comprehension") == 0 || strcmp(k, "dictionary_comprehension") == 0 ||
strcmp(k, "set_comprehension") == 0 || strcmp(k, "generator_expression") == 0) {
if (ctx->class_body_outer_scope) {
py_resolve_class_comprehension(ctx, node);
return;
}
CBMScope *saved = ctx->current_scope;
ctx->current_scope = py_scope_push_checked(ctx);
uint32_t cnc = ts_node_named_child_count(node);
@@ -3699,15 +3793,19 @@ static void py_resolve_calls_in_inner(PyLSPContext *ctx, TSNode node) {
// Recurse: children. We don't push scope for control-flow blocks
// here (Python scoping is function-level apart from comprehension /
// lambda / class), but we do for nested function / class / lambda.
if (strcmp(k, "function_definition") == 0 || strcmp(k, "class_definition") == 0 ||
strcmp(k, "lambda") == 0) {
if (strcmp(k, "function_definition") == 0 || strcmp(k, "lambda") == 0) {
// These are processed by the top-level pass; skip recursion to
// avoid double-walking their bodies.
return;
}
if (strcmp(k, "class_definition") == 0) {
py_exec_class_definition(ctx, node);
return;
}
PyKids nk = py_kids(ctx->arena, node);
for (uint32_t i = 0; i < nk.n; i++) {
for (uint32_t i = 0; i < nk.n && !ctx->eval_failure; i++) {
py_resolve_calls_in(ctx, py_kid(&nk, i));
}
}
@@ -4328,6 +4426,120 @@ static bool py_is_init_method(PyLSPContext *ctx, TSNode func_node) {
return nm && (strcmp(nm, "__init__") == 0 || strcmp(nm, "__post_init__") == 0);
}
/* Decorator expressions of a decorated definition, in source order. They
* evaluate in the scope enclosing the definition, before it is bound. A bare
* `@name` is no call and emits nothing; `@name(args)` is one. */
static void py_resolve_decorators(PyLSPContext *ctx, TSNode decorated) {
if (!ctx || ctx->eval_failure)
return;
PyKids kids = py_kids(ctx->arena, decorated);
for (uint32_t i = 0; i < kids.n && !ctx->eval_failure; i++) {
TSNode child = py_kid(&kids, i);
if (strcmp(ts_node_type(child), "decorator") == 0)
py_resolve_calls_in(ctx, child);
}
}
/* Bind the name a class-body `def`/`class` statement defines in the class
* frame, so later class-body statements see it (and never the same-named
* module symbol). Its value is exact only for an undecorated function or a
* registered class whose QN is known; anything else is an UNKNOWN shadow. */
static void py_bind_class_body_def(PyLSPContext *ctx, TSNode stmt) {
if (ctx->eval_failure)
return;
TSNode def = stmt;
bool decorated = strcmp(ts_node_type(stmt), "decorated_definition") == 0;
if (decorated) {
def = ts_node_child_by_field_name(stmt, "definition", 10);
if (ts_node_is_null(def))
return;
}
const char *kind = ts_node_type(def);
bool is_class = strcmp(kind, "class_definition") == 0;
if (!is_class && strcmp(kind, "function_definition") != 0)
return;
char *name = py_node_text(ctx, ts_node_child_by_field_name(def, "name", 4));
if (!name || !name[0]) {
py_disable_callable_value_proof(ctx);
return;
}
const char *qn = ctx->class_body_qn
? cbm_arena_sprintf(ctx->arena, "%s.%s", ctx->class_body_qn, name)
: NULL;
if (is_class) {
const CBMRegisteredType *type = qn ? cbm_registry_lookup_type(ctx->registry, qn) : NULL;
py_scope_bind(ctx, name,
type ? cbm_type_named(ctx->arena, type->qualified_name) : cbm_type_unknown());
return;
}
const CBMRegisteredFunc *func = qn ? cbm_registry_lookup_func(ctx->registry, qn) : NULL;
if (!decorated && py_func_is_exact_callable_value(func)) {
py_scope_bind_callable(ctx, name, cbm_type_unknown(), func->qualified_name);
} else {
py_scope_bind(ctx, name, cbm_type_unknown());
}
}
/* Execute a class definition where it stands: its bases in the enclosing
* scope, then its body statements in source order in a class frame. Calls
* keep the caller the extraction gives them: the enclosing function, or the
* module QN at module level (enclosing_func_qn is left unchanged).
*
* Scoping follows Python: a name bound earlier in the body is visible to
* later body statements (and to decorators of methods), but the class frame
* is not an enclosing scope for anything nested in it. Method bodies are
* resolved later by py_process_class against the module scope, and a nested
* class body chains to the nearest enclosing non-class scope. Each body
* statement is walked exactly once. */
static void py_exec_class_definition(PyLSPContext *ctx, TSNode class_node) {
if (!ctx || ctx->eval_failure || ts_node_is_null(class_node))
return;
TSNode supers = ts_node_child_by_field_name(class_node, "superclasses", 12);
if (!ts_node_is_null(supers))
py_resolve_calls_in(ctx, supers);
TSNode body = ts_node_child_by_field_name(class_node, "body", 4);
if (ctx->eval_failure || ts_node_is_null(body))
return;
char *cname = py_node_text(ctx, ts_node_child_by_field_name(class_node, "name", 4));
const char *class_qn = NULL;
if (cname && cname[0]) {
if (ctx->class_body_outer_scope) {
class_qn = ctx->class_body_qn
? cbm_arena_sprintf(ctx->arena, "%s.%s", ctx->class_body_qn, cname)
: NULL;
} else if (ctx->module_qn && ctx->enclosing_func_qn && ctx->file_module_qn &&
strcmp(ctx->enclosing_func_qn, ctx->file_module_qn) == 0) {
class_qn = cbm_arena_sprintf(ctx->arena, "%s.%s", ctx->module_qn, cname);
}
}
CBMScope *saved_scope = ctx->current_scope;
CBMScope *saved_outer = ctx->class_body_outer_scope;
const char *saved_qn = ctx->class_body_qn;
bool saved_comprehension = ctx->class_body_in_comprehension;
CBMScope *outer = saved_outer ? saved_outer : saved_scope;
CBMScope *frame = py_class_frame_push(ctx, outer);
if (!frame)
return;
py_scope_restore(ctx, frame);
ctx->class_body_outer_scope = outer;
ctx->class_body_qn = class_qn;
ctx->class_body_in_comprehension = false;
PyKids bk = py_kids(ctx->arena, body);
for (uint32_t i = 0; i < bk.n && !ctx->eval_failure; i++) {
TSNode stmt = py_kid(&bk, i);
py_resolve_calls_in(ctx, stmt);
py_bind_class_body_def(ctx, stmt);
}
ctx->class_body_outer_scope = saved_outer;
ctx->class_body_qn = saved_qn;
ctx->class_body_in_comprehension = saved_comprehension;
py_scope_restore(ctx, saved_scope);
}
static void py_process_class(PyLSPContext *ctx, TSNode class_node) {
if (!ctx || ctx->eval_failure)
return;
@@ -4990,15 +5202,21 @@ void py_lsp_process_file(PyLSPContext *ctx, TSNode root) {
} else if (strcmp(ck, "function_definition") == 0) {
py_bind_module_function(ctx, c);
} else if (strcmp(ck, "class_definition") == 0) {
py_bind_module_class(ctx, c);
py_exec_class_definition(ctx, c);
if (!ctx->eval_failure)
py_bind_module_class(ctx, c);
} else if (strcmp(ck, "decorated_definition") == 0) {
py_resolve_decorators(ctx, c);
TSNode def = ts_node_child_by_field_name(c, "definition", 10);
if (!ts_node_is_null(def)) {
if (!ctx->eval_failure && !ts_node_is_null(def)) {
const char *dk = ts_node_type(def);
if (strcmp(dk, "function_definition") == 0)
if (strcmp(dk, "function_definition") == 0) {
py_bind_module_function(ctx, def);
else if (strcmp(dk, "class_definition") == 0)
py_bind_module_class(ctx, def);
} else if (strcmp(dk, "class_definition") == 0) {
py_exec_class_definition(ctx, def);
if (!ctx->eval_failure)
py_bind_module_class(ctx, def);
}
}
} else if (strcmp(ck, "expression_statement") == 0) {
/* An expression statement is resolved for calls, not scanned as a
internal/cbm/lsp/py_lsp.h
+9 -2
View File
@@ -61,8 +61,13 @@ typedef struct {
// caller QNs.
const char *enclosing_func_qn;
const char *enclosing_class_qn;
const char *module_qn; // Python symbol scope (package __init__ omitted)
const char *file_module_qn; // raw file identity; borrowed for this API call
// Active class execution: nested class bodies skip this class frame.
// Local classes have no extracted graph QN, so class_body_qn stays NULL.
CBMScope *class_body_outer_scope;
const char *class_body_qn;
bool class_body_in_comprehension; // retain enclosing comprehension bindings
const char *module_qn; // Python symbol scope (package __init__ omitted)
const char *file_module_qn; // raw file identity; borrowed for this API call
// Output: resolved calls accumulate here.
CBMResolvedCallArray *resolved_calls;
@@ -180,6 +185,8 @@ void cbm_py_lsp_record_failure(CBMFileResult *result, CBMLSPStatus status);
void cbm_py_lsp_test_memo_fail_after(int successful_allocations);
void cbm_py_lsp_test_depth_fail(bool enabled);
void cbm_py_lsp_test_scope_fail_after(int successful_allocations);
// Separate from the package symbol-scope seam; only new class execution frames.
void cbm_py_lsp_test_class_frame_fail_after(int successful_allocations);
bool cbm_py_lsp_test_register_defs(CBMArena *arena, CBMTypeRegistry *reg, CBMLSPDef *defs,
int def_count);
const CBMType *cbm_py_lsp_test_eval(PyLSPContext *ctx, TSNode node);
tests/test_pipeline.c
+149
View File
@@ -10892,6 +10892,153 @@ TEST(pipeline_python_module_callers_parallel) {
return python_module_callers_pipeline_case(true);
}
/* Distinct targets keep each newly walked class/decorator site observable
* even though the stored graph aggregates calls by caller/target pair. */
static int python_class_body_pipeline_case(bool parallel) {
static const char helpers[] =
"def top_field():\n return 1\n"
"def nested_field():\n return 2\n"
"def class_decorator(value):\n return lambda target: target\n"
"def class_argument():\n return 3\n"
"def method_decorator(value):\n return lambda target: target\n"
"def method_argument():\n return 4\n"
"def method_control():\n return 5\n"
"def local_field():\n return 6\n"
"def method_local_field():\n return 7\n"
"def base_factory():\n return object\n"
"def package_field():\n return 8\n"
"def package_decorator(value):\n return lambda target: target\n"
"def package_argument():\n return 9\n";
static const char app[] =
"from helpers import top_field, nested_field, class_decorator, class_argument\n"
"from helpers import method_decorator, method_argument, method_control\n"
"from helpers import local_field, method_local_field, base_factory\n"
"class Base(base_factory()):\n"
" value = top_field()\n"
" @method_decorator(method_argument())\n"
" def run(self):\n"
" method_control()\n"
" class MethodLocal:\n"
" value = method_local_field()\n"
"@class_decorator(class_argument())\n"
"class Admin:\n"
" class Meta:\n"
" value = nested_field()\n"
"def outer():\n"
" class Local:\n"
" value = local_field()\n"
" return Local\n";
static const char package[] =
"from helpers import package_field, package_decorator, package_argument\n"
"@package_decorator(package_argument())\n"
"class Package:\n"
" value = package_field()\n";
static const struct {
const char *caller;
const char *caller_label;
const char *target;
} expected[] = {
{"app", "Module", "helpers.base_factory"},
{"app", "Module", "helpers.top_field"},
{"app", "Module", "helpers.nested_field"},
{"app", "Module", "helpers.class_decorator"},
{"app", "Module", "helpers.class_argument"},
{"app", "Module", "helpers.method_decorator"},
{"app", "Module", "helpers.method_argument"},
{"app.Base.run", "Method", "helpers.method_control"},
{"app.Base.run", "Method", "helpers.method_local_field"},
{"app.outer", "Function", "helpers.local_field"},
{"pkg.__init__", "Module", "helpers.package_decorator"},
{"pkg.__init__", "Module", "helpers.package_argument"},
{"pkg.__init__", "Module", "helpers.package_field"},
};
char root[512], path[768];
int n = snprintf(root, sizeof(root), "%s/cbm-python-class-body-XXXXXX", cbm_tmpdir());
if (n <= 0 || (size_t)n >= sizeof(root) || !cbm_mkdtemp(root))
FAIL("failed to create Python class-body fixture");
n = snprintf(path, sizeof(path), "%s/pkg", root);
bool written = n > 0 && (size_t)n < sizeof(path);
if (written) {
cbm_mkdir_p(path, 0755);
written = python_module_write_fixture_file(root, "helpers.py", helpers) &&
python_module_write_fixture_file(root, "app.py", app) &&
python_module_write_fixture_file(root, "pkg/__init__.py", package);
}
for (int i = 0; written && parallel && i < 52; i++) {
char name[64];
snprintf(name, sizeof(name), "class_body_pad_%02d.py", i);
written = python_module_write_fixture_file(root, name, "# inert parallel routing file\n");
}
if (!written) {
th_rmtree(root);
FAIL("failed to write Python class-body fixture");
}
const char *keys[] = {"CBM_WORKERS", "CBM_INDEX_SINGLE_THREAD"};
char *saved[2] = {0};
bool saved_ok = true;
for (int i = 0; i < 2; i++) {
const char *old = getenv(keys[i]);
saved[i] = old ? strdup(old) : NULL;
saved_ok = saved_ok && (!old || saved[i]);
}
if (!saved_ok) {
free(saved[0]);
free(saved[1]);
th_rmtree(root);
FAIL("failed to save Python class-body fixture environment");
}
int set_rc = cbm_setenv(keys[0], parallel ? "4" : "1", 1);
set_rc |= parallel ? cbm_unsetenv(keys[1]) : cbm_setenv(keys[1], "1", 1);
n = snprintf(path, sizeof(path), "%s/class-body.db", root);
cbm_pipeline_t *pipeline = set_rc == 0 && n > 0 && (size_t)n < sizeof(path)
? cbm_pipeline_new(root, path, CBM_MODE_FULL)
: NULL;
int run_rc = pipeline ? cbm_pipeline_run(pipeline) : -1;
cbm_file_error_t *errors = NULL;
int error_count = 0;
cbm_pipeline_get_file_errors(pipeline, &errors, &error_count);
cbm_store_t *store = run_rc == 0 ? cbm_store_open_path(path) : NULL;
bool correct = store != NULL;
if (store) {
const char *project = cbm_pipeline_project_name(pipeline);
cbm_node_t *files = NULL;
int file_count = 0;
correct = project &&
cbm_store_find_nodes_by_label(store, project, "File", &files, &file_count) ==
CBM_STORE_OK &&
file_count == (parallel ? 55 : 3);
cbm_store_free_nodes(files, file_count);
for (size_t i = 0; i < sizeof(expected) / sizeof(expected[0]); i++) {
bool edge = project && python_module_exact_lsp_edge(store, project, expected[i].caller,
expected[i].caller_label,
expected[i].target, "Function");
correct = correct && edge;
}
cbm_store_close(store);
}
cbm_pipeline_free(pipeline);
int restore_rc = 0;
for (int i = 0; i < 2; i++) {
restore_rc |= saved[i] ? cbm_setenv(keys[i], saved[i], 1) : cbm_unsetenv(keys[i]);
free(saved[i]);
}
int cleanup_rc = th_rmtree(root);
ASSERT_EQ(set_rc, 0);
ASSERT_EQ(restore_rc, 0);
ASSERT_EQ(cleanup_rc, 0);
ASSERT_EQ(run_rc, 0);
ASSERT_EQ(error_count, 0);
ASSERT_TRUE(correct);
PASS();
}
TEST(pipeline_python_class_body_calls_sequential) {
return python_class_body_pipeline_case(false);
}
TEST(pipeline_python_class_body_calls_parallel) {
return python_class_body_pipeline_case(true);
}
/* Python file identity keeps .__init__; its exported symbols do not. */
static bool python_init_local_inheritance(cbm_store_t *store, const char *project,
const char *module) {
@@ -20940,6 +21087,8 @@ TEST(pipeline_calls_args_overlong_argument_stops_cleanly) {
SUITE(pipeline) {
RUN_TEST(pipeline_python_module_callers_sequential);
RUN_TEST(pipeline_python_module_callers_parallel);
RUN_TEST(pipeline_python_class_body_calls_sequential);
RUN_TEST(pipeline_python_class_body_calls_parallel);
RUN_TEST(pipeline_python_init_scope_sequential);
RUN_TEST(pipeline_python_init_scope_parallel);
RUN_TEST(pipeline_python_init_raw_scope_failure_sequential);
tests/test_py_lsp.c
+475
View File
@@ -3435,7 +3435,482 @@ TEST(pylsp_init_scope_shared_failure_keeps_arena_and_needed_fallback) {
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);
bool correct = pylsp_class_sites(r, source, sites, 7) &&
!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(correct);
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();
}
SUITE(py_lsp) {
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);