use tree_sitter::Node;
use crate::model::{Edge, EdgeType, Language, Node as ModelNode, NodeLabel};
use crate::resolve::{FqnGenerator, ScopeContext, ScopeResolverRegistry};
use super::dedupe_qn;
use super::error::{ParseError, Result};
use super::extractor::{
AssignInfo, CallInfo, ExtractResult, Extractor, ImportInfo, ReadInfo, WriteInfo,
};
use super::parser_factory::ParserFactory;
pub struct PythonExtractor {
_priv: (),
}
impl PythonExtractor {
#[must_use]
pub const fn new() -> Self {
Self { _priv: () }
}
}
impl Default for PythonExtractor {
fn default() -> Self {
Self::new()
}
}
impl Extractor for PythonExtractor {
fn language(&self) -> Language {
Language::Python
}
fn extract(&self, source: &str, file_path: &str, project: &str) -> Result<ExtractResult> {
let mut result = ExtractResult::new(file_path, Language::Python);
let mut parser = ParserFactory::create_parser(Language::Python)?;
let tree = parser
.parse(source, None)
.ok_or_else(|| ParseError::ParseFailed {
file_path: file_path.to_string(),
})?;
let root = tree.root_node();
let registry = ScopeResolverRegistry::new();
let ctx = VisitContext {
file_path,
project,
current_func: None,
current_parent: None,
resolver: ®istry,
};
for i in 0..root.named_child_count() as u32 {
if let Some(child) = root.named_child(i) {
visit_node(child, source, &ctx, &mut result);
}
}
Ok(result)
}
}
struct VisitContext<'a> {
file_path: &'a str,
project: &'a str,
current_func: Option<&'a str>,
current_parent: Option<&'a str>,
resolver: &'a ScopeResolverRegistry,
}
fn visit_node(node: Node, source: &str, ctx: &VisitContext<'_>, result: &mut ExtractResult) {
match node.kind() {
"function_definition" => {
extract_function(node, source, ctx, result);
let scope_ctx = ScopeContext {
source,
file_path: ctx.file_path,
project: ctx.project,
current_parent: ctx.current_parent,
};
let scope = ctx
.resolver
.get(Language::Python)
.and_then(|r| r.resolve(node, &scope_ctx));
let func_name = scope.as_ref().map(|s| s.name.as_str());
let child_ctx = VisitContext {
file_path: ctx.file_path,
project: ctx.project,
current_func: func_name,
current_parent: ctx.current_parent,
resolver: ctx.resolver,
};
visit_children(node, source, &child_ctx, result);
}
"class_definition" => {
extract_class(node, source, ctx, result);
let scope_ctx = ScopeContext {
source,
file_path: ctx.file_path,
project: ctx.project,
current_parent: ctx.current_parent,
};
let scope = ctx
.resolver
.get(Language::Python)
.and_then(|r| r.resolve(node, &scope_ctx));
let class_name = scope.as_ref().map(|s| s.name.as_str());
let combined = combine_scope(ctx.current_parent, class_name);
let child_ctx = VisitContext {
file_path: ctx.file_path,
project: ctx.project,
current_func: None,
current_parent: combined.as_deref(),
resolver: ctx.resolver,
};
visit_children(node, source, &child_ctx, result);
}
"import_statement" => {
extract_import(node, source, result);
}
"import_from_statement" => {
extract_import_from(node, source, result);
}
"call" => {
extract_call(node, source, ctx, result);
visit_children(node, source, ctx, result);
}
"assignment" => {
extract_assignment(node, source, result);
if let Some(func) = ctx.current_func {
if let Some(left) = node.child_by_field_name("left") {
if let Some(name) = identifier_text(left, source) {
result.writes.push(WriteInfo {
writer_qn: Some(make_qn(
ctx.file_path,
func,
ctx.project,
ctx.current_parent,
)),
var_name: name,
line: node.start_position().row as u32 + 1,
});
}
}
}
visit_children(node, source, ctx, result);
}
"augmented_assignment" => {
if let Some(func) = ctx.current_func {
if let Some(left) = node.child_by_field_name("left") {
if let Some(name) = identifier_text(left, source) {
result.writes.push(WriteInfo {
writer_qn: Some(make_qn(
ctx.file_path,
func,
ctx.project,
ctx.current_parent,
)),
var_name: name,
line: node.start_position().row as u32 + 1,
});
}
}
}
visit_children(node, source, ctx, result);
}
"for_statement" => {
if let Some(func) = ctx.current_func {
if let Some(left) = node.child_by_field_name("left") {
if let Some(name) = identifier_text(left, source) {
result.writes.push(WriteInfo {
writer_qn: Some(make_qn(
ctx.file_path,
func,
ctx.project,
ctx.current_parent,
)),
var_name: name,
line: node.start_position().row as u32 + 1,
});
}
}
}
visit_children(node, source, ctx, result);
}
"identifier" => {
if let Some(func) = ctx.current_func {
if is_python_read_position(node) {
if let Some(name) = node_text(node, source).map(String::from) {
result.reads.push(ReadInfo {
reader_qn: Some(make_qn(
ctx.file_path,
func,
ctx.project,
ctx.current_parent,
)),
var_name: name,
line: node.start_position().row as u32 + 1,
});
}
}
}
visit_children(node, source, ctx, result);
}
_ => {
visit_children(node, source, ctx, result);
}
}
}
fn visit_children(node: Node, source: &str, ctx: &VisitContext<'_>, result: &mut ExtractResult) {
for i in 0..node.named_child_count() as u32 {
if let Some(child) = node.named_child(i) {
visit_node(child, source, ctx, result);
}
}
}
fn extract_function(node: Node, source: &str, ctx: &VisitContext<'_>, result: &mut ExtractResult) {
let Some(name_node) = node.child_by_field_name("name") else {
return;
};
let Some(name) = node_text(name_node, source).map(String::from) else {
return;
};
let is_method = is_inside_class(node);
if !is_method && has_ancestor_function(node) {
return;
}
let label = if is_method {
NodeLabel::Method
} else {
NodeLabel::Function
};
let qn = dedupe_qn(
make_qn(ctx.file_path, &name, ctx.project, ctx.current_parent),
node.start_position().row as u32 + 1,
result,
);
let signature = function_signature(node, source);
let mut builder = ModelNode::builder(label, name, qn)
.file_path(ctx.file_path)
.start_line(node.start_position().row as u32 + 1)
.end_line(node.end_position().row as u32 + 1)
.language(Language::Python)
.project(ctx.project)
.is_global(!is_method);
if is_method {
if let Some(parent) = ctx.current_parent {
builder = builder.parent_qn(parent);
}
}
if let Some(sig) = signature {
builder = builder.signature(sig);
}
let model_node = builder.build();
add_definition_edges(ctx.file_path, ctx.project, &model_node, result);
result.push_node(model_node);
}
fn extract_class(node: Node, source: &str, ctx: &VisitContext<'_>, result: &mut ExtractResult) {
let Some(name_node) = node.child_by_field_name("name") else {
return;
};
let Some(name) = node_text(name_node, source).map(String::from) else {
return;
};
let qn = dedupe_qn(
make_qn(ctx.file_path, &name, ctx.project, None),
node.start_position().row as u32 + 1,
result,
);
let model_node = ModelNode::builder(NodeLabel::Class, name, qn.clone())
.file_path(ctx.file_path)
.start_line(node.start_position().row as u32 + 1)
.end_line(node.end_position().row as u32 + 1)
.language(Language::Python)
.project(ctx.project)
.is_global(true)
.build();
add_definition_edges(ctx.file_path, ctx.project, &model_node, result);
result.push_node(model_node);
if let Some(superclasses) = node.child_by_field_name("superclasses") {
for i in 0..superclasses.named_child_count() as u32 {
if let Some(base) = superclasses.named_child(i) {
if base.kind() == "keyword_argument" {
continue;
}
if let Some(parent_name) = base_class_name(base, source) {
let parent_qn =
make_qn(ctx.file_path, &parent_name, ctx.project, ctx.current_parent);
result.edges.push(Edge::new(
qn.clone(),
parent_qn,
EdgeType::Extends,
ctx.project,
));
}
}
}
}
}
fn base_class_name(node: Node, source: &str) -> Option<String> {
match node.kind() {
"identifier" => node_text(node, source).map(String::from),
"attribute" => {
let attr = node.child_by_field_name("attribute")?;
node_text(attr, source).map(String::from)
}
"call" => {
let func = node.child_by_field_name("function")?;
base_class_name(func, source)
}
_ => None,
}
}
fn extract_import(node: Node, source: &str, result: &mut ExtractResult) {
for i in 0..node.named_child_count() as u32 {
if let Some(child) = node.named_child(i) {
if child.kind() == "dotted_name" {
if let Some(name) = dotted_name_text(child, source) {
result.imports.push(ImportInfo {
source_file: name,
imported_names: Vec::new(),
line: node.start_position().row as u32 + 1,
});
}
}
}
}
}
fn extract_import_from(node: Node, source: &str, result: &mut ExtractResult) {
let mut source_module = None;
let mut names = Vec::new();
for i in 0..node.named_child_count() as u32 {
if let Some(child) = node.named_child(i) {
if child.kind() == "dotted_name" {
if source_module.is_none() {
source_module = dotted_name_text(child, source);
} else if let Some(n) = dotted_name_text(child, source) {
names.push(n);
}
} else if child.kind() == "aliased_import" {
if let Some(name) = aliased_import_name(child, source) {
names.push(name);
}
} else if child.kind() == "wildcard_import" {
names.push("*".to_string());
}
}
}
let Some(source_module) = source_module else {
return;
};
result.imports.push(ImportInfo {
source_file: source_module,
imported_names: names,
line: node.start_position().row as u32 + 1,
});
}
fn extract_call(node: Node, source: &str, ctx: &VisitContext<'_>, result: &mut ExtractResult) {
let Some(func_node) = node.child_by_field_name("function") else {
return;
};
let Some(callee) = callee_name(func_node, source) else {
return;
};
let args = call_arguments(node, source);
let caller_qn = ctx
.current_func
.map(|name| make_qn(ctx.file_path, name, ctx.project, None));
result.calls.push(CallInfo {
caller_qn,
callee_name: callee,
line: node.start_position().row as u32 + 1,
args,
});
}
fn extract_assignment(node: Node, source: &str, result: &mut ExtractResult) {
let Some(left_node) = node.child_by_field_name("left") else {
return;
};
let Some(target) = assignment_target_name(left_node, source) else {
return;
};
let right_node = node.child_by_field_name("right");
let (source_name, is_return_assign) = match right_node {
Some(v) => {
let is_call = v.kind() == "call";
let name = if is_call {
v.child_by_field_name("function")
.and_then(|f| callee_name(f, source))
.unwrap_or_default()
} else {
callee_name(v, source).unwrap_or_default()
};
(name, is_call)
}
None => (String::new(), false),
};
result.assignments.push(AssignInfo {
target_name: target,
source_name,
line: node.start_position().row as u32 + 1,
is_return_assign,
});
}
enum FunctionScope {
Class,
Function,
Module,
}
fn function_scope(node: Node) -> FunctionScope {
let mut cur = node.parent();
while let Some(p) = cur {
match p.kind() {
"class_definition" => return FunctionScope::Class,
"function_definition" => return FunctionScope::Function,
_ => cur = p.parent(),
}
}
FunctionScope::Module
}
fn is_inside_class(node: Node) -> bool {
matches!(function_scope(node), FunctionScope::Class)
}
fn has_ancestor_function(node: Node) -> bool {
matches!(function_scope(node), FunctionScope::Function)
}
fn function_signature(node: Node, source: &str) -> Option<String> {
let start = node.start_position();
let end = node.end_position();
if start.row == end.row {
node_text(node, source).map(String::from)
} else {
let line_end = source.lines().nth(start.row).map(|l| l.len()).unwrap_or(0);
let start_byte = node.start_byte();
let line_end_byte = start_byte + line_end;
if line_end_byte <= source.len() {
Some(source[start_byte..line_end_byte].to_string())
} else {
node_text(node, source).map(String::from)
}
}
}
fn dotted_name_text(node: Node, source: &str) -> Option<String> {
let text = node_text(node, source)?;
Some(text.to_string())
}
fn aliased_import_name(node: Node, source: &str) -> Option<String> {
if let Some(alias) = node.child_by_field_name("alias") {
return node_text(alias, source).map(String::from);
}
node.child_by_field_name("name")
.and_then(|n| node_text(n, source).map(String::from))
}
fn callee_name(node: Node, source: &str) -> Option<String> {
match node.kind() {
"identifier" => node_text(node, source).map(String::from),
"attribute" => {
let attr = node.child_by_field_name("attribute")?;
node_text(attr, source).map(String::from)
}
"call" => {
let func = node.child_by_field_name("function")?;
callee_name(func, source)
}
"parenthesized_expression" => {
let inner = node.named_child(0)?;
callee_name(inner, source)
}
_ => None,
}
}
fn assignment_target_name(node: Node, source: &str) -> Option<String> {
match node.kind() {
"identifier" => node_text(node, source).map(String::from),
"attribute" => {
let attr = node.child_by_field_name("attribute")?;
node_text(attr, source).map(String::from)
}
"tuple" | "list" | "pattern_list" => {
for i in 0..node.named_child_count() as u32 {
if let Some(child) = node.named_child(i) {
if let Some(name) = assignment_target_name(child, source) {
return Some(name);
}
}
}
None
}
"subscript" => {
let value = node.child_by_field_name("value")?;
assignment_target_name(value, source)
}
_ => {
let text = node_text(node, source)?;
if text.chars().all(|c| c.is_alphanumeric() || c == '_')
&& text
.chars()
.next()
.is_some_and(|c| c.is_alphabetic() || c == '_')
{
Some(text.to_string())
} else {
None
}
}
}
}
fn call_arguments(node: Node, source: &str) -> Vec<String> {
let Some(args_node) = node.child_by_field_name("arguments") else {
return Vec::new();
};
let mut args = Vec::new();
for i in 0..args_node.named_child_count() as u32 {
if let Some(arg) = args_node.named_child(i) {
if let Ok(text) = arg.utf8_text(source.as_bytes()) {
args.push(text.to_string());
}
}
}
args
}
fn node_text<'a>(node: Node<'a>, source: &'a str) -> Option<&'a str> {
node.utf8_text(source.as_bytes()).ok()
}
fn identifier_text(node: Node, source: &str) -> Option<String> {
if node.kind() == "identifier" {
node_text(node, source).map(String::from)
} else {
None
}
}
fn is_python_read_position(node: Node) -> bool {
let Some(parent) = node.parent() else {
return false;
};
match parent.kind() {
"binary_operator"
| "boolean_operator"
| "comparison_operator"
| "parenthesized_expression"
| "return_statement"
| "argument_list"
| "subscript"
| "conditional_expression"
| "list"
| "tuple"
| "set"
| "keyword_argument" => true,
"call" => !is_at_field(node, parent, "function"),
"assignment" => !is_at_field(node, parent, "left"),
"attribute" => is_at_field(node, parent, "object"),
"augmented_assignment"
| "for_statement"
| "function_definition"
| "class_definition"
| "parameters"
| "lambda"
| "import_statement"
| "import_from_statement"
| "dotted_name"
| "aliased_import"
| "wildcard_import" => false,
_ => false,
}
}
fn is_at_field(node: Node, parent: Node, field: &str) -> bool {
parent
.child_by_field_name(field)
.is_some_and(|f| f.byte_range() == node.byte_range())
}
fn make_qn(file_path: &str, name: &str, project: &str, parent: Option<&str>) -> String {
FqnGenerator::generate(project, file_path, name, Language::Python, parent)
}
fn combine_scope(parent: Option<&str>, child: Option<&str>) -> Option<String> {
match (parent, child) {
(Some(p), Some(c)) => Some(format!("{p}_{c}")),
(None, Some(c)) => Some(c.to_string()),
(Some(p), None) => Some(p.to_string()),
(None, None) => None,
}
}
fn add_definition_edges(
file_path: &str,
project: &str,
node: &ModelNode,
result: &mut ExtractResult,
) {
result.edges.push(Edge::new(
file_path.to_string(),
node.id.clone(),
EdgeType::Defines,
project,
));
}
#[cfg(test)]
mod tests {
use super::*;
use crate::model::NodeLabel;
const PYTHON_SOURCE: &str = r#"import os
from typing import List
def add(a, b):
return a + b
class Point:
def __init__(self, x, y):
self.x = x
self.y = y
def distance(self):
return self.x + self.y
result = add(1, 2)
"#;
fn extract(source: &str) -> ExtractResult {
let ext = PythonExtractor::new();
ext.extract(source, "test.py", "proj")
.expect("extraction should succeed")
}
#[test]
fn language_returns_python() {
assert_eq!(PythonExtractor::new().language(), Language::Python);
}
#[test]
fn default_creates_extractor() {
let ext = PythonExtractor::default();
assert_eq!(ext.language(), Language::Python);
}
#[test]
fn extracts_imports() {
let result = extract(PYTHON_SOURCE);
assert_eq!(result.imports.len(), 2, "should extract 2 imports");
assert_eq!(result.imports[0].source_file, "os");
assert_eq!(result.imports[1].source_file, "typing");
assert!(
result.imports[1]
.imported_names
.contains(&"List".to_string()),
"from typing import List should have List in imported_names"
);
}
#[test]
fn extracts_top_level_function() {
let result = extract(PYTHON_SOURCE);
let funcs: Vec<_> = result
.nodes
.iter()
.filter(|n| n.label == NodeLabel::Function)
.collect();
assert_eq!(funcs.len(), 1, "should extract 1 top-level function (add)");
assert_eq!(funcs[0].name, "add");
assert_eq!(funcs[0].language, Some(Language::Python));
assert_eq!(funcs[0].project, "proj");
assert_eq!(funcs[0].file_path.as_deref(), Some("test.py"));
assert!(funcs[0].is_global, "top-level function should be global");
}
#[test]
fn extracts_class() {
let result = extract(PYTHON_SOURCE);
let classes: Vec<_> = result
.nodes
.iter()
.filter(|n| n.label == NodeLabel::Class)
.collect();
assert_eq!(classes.len(), 1);
assert_eq!(classes[0].name, "Point");
}
#[test]
fn p2_2_extends_edge_for_single_inheritance() {
let src = r#"class Parent:
pass
class Child(Parent):
pass
"#;
let result = extract(src);
let extends: Vec<_> = result
.edges
.iter()
.filter(|e| e.edge_type == EdgeType::Extends)
.collect();
assert_eq!(
extends.len(),
1,
"should create 1 EXTENDS edge: {:?}",
extends
);
assert!(
extends[0].source.contains("Child"),
"EXTENDS source should be Child FQN: {}",
extends[0].source
);
assert!(
extends[0].target.contains("Parent"),
"EXTENDS target should be Parent FQN: {}",
extends[0].target
);
}
#[test]
fn p2_2_extends_edge_for_multiple_bases() {
let src = r#"class Base1:
pass
class Base2:
pass
class Derived(Base1, Base2):
pass
"#;
let result = extract(src);
let extends: Vec<_> = result
.edges
.iter()
.filter(|e| e.edge_type == EdgeType::Extends)
.collect();
assert_eq!(
extends.len(),
2,
"should create 2 EXTENDS edges: {:?}",
extends
);
}
#[test]
fn p2_2_extends_edge_skips_keyword_argument() {
let src = r#"class Meta:
pass
class Foo(metaclass=Meta):
pass
"#;
let result = extract(src);
let extends: Vec<_> = result
.edges
.iter()
.filter(|e| e.edge_type == EdgeType::Extends)
.collect();
assert_eq!(
extends.len(),
0,
"should skip keyword_argument: {:?}",
extends
);
}
#[test]
fn extracts_methods() {
let result = extract(PYTHON_SOURCE);
let methods: Vec<_> = result
.nodes
.iter()
.filter(|n| n.label == NodeLabel::Method)
.collect();
let names: Vec<_> = methods.iter().map(|n| n.name.as_str()).collect();
assert!(
names.contains(&"__init__"),
"should extract __init__ method: {:?}",
names
);
assert!(
names.contains(&"distance"),
"should extract distance method: {:?}",
names
);
assert!(!methods[0].is_global, "methods should not be global");
}
#[test]
fn extracts_call_to_add() {
let result = extract(PYTHON_SOURCE);
let callees: Vec<_> = result
.calls
.iter()
.map(|c| c.callee_name.as_str())
.collect();
assert!(
callees.contains(&"add"),
"should extract call to add: {:?}",
callees
);
}
#[test]
fn call_has_line_and_args() {
let result = extract(PYTHON_SOURCE);
let call = result
.calls
.iter()
.find(|c| c.callee_name == "add")
.expect("call to add should exist");
assert_eq!(call.line, 15);
assert_eq!(call.args.len(), 2, "add(1, 2) should have 2 args");
}
#[test]
fn extracts_assignment() {
let result = extract(PYTHON_SOURCE);
let assign = result
.assignments
.iter()
.find(|a| a.target_name == "result")
.expect("should find `result = add(1, 2)` assignment");
assert_eq!(assign.source_name, "add");
assert!(
assign.is_return_assign,
"assignment from function call should be return assign"
);
}
#[test]
fn creates_defines_edges() {
let result = extract(PYTHON_SOURCE);
let defines_count = result
.edges
.iter()
.filter(|e| e.edge_type == EdgeType::Defines)
.count();
let node_count = result.nodes.len();
assert_eq!(defines_count, node_count);
let contains_count = result
.edges
.iter()
.filter(|e| e.edge_type == EdgeType::Contains)
.count();
assert_eq!(
contains_count, 0,
"B1 fix: no CONTAINS edges should be emitted"
);
}
#[test]
fn qualified_name_uses_file_path_and_name() {
let result = extract(PYTHON_SOURCE);
let add = result.nodes.iter().find(|n| n.name == "add").unwrap();
assert_eq!(add.qualified_name, "proj.test.py.add");
}
#[test]
fn empty_source_returns_empty_result() {
let result = extract("");
assert!(result.is_empty());
}
#[test]
fn function_has_signature() {
let result = extract(PYTHON_SOURCE);
let add = result.nodes.iter().find(|n| n.name == "add").unwrap();
assert!(add.signature.is_some(), "function should have a signature");
assert!(add.signature.as_deref().unwrap().contains("add"));
}
#[test]
fn handles_from_import_with_multiple_names() {
let src = "from typing import List, Dict, Optional\n";
let result = extract(src);
assert_eq!(result.imports.len(), 1);
assert_eq!(result.imports[0].source_file, "typing");
assert_eq!(result.imports[0].imported_names.len(), 3);
assert!(result.imports[0]
.imported_names
.contains(&"List".to_string()));
assert!(result.imports[0]
.imported_names
.contains(&"Dict".to_string()));
assert!(result.imports[0]
.imported_names
.contains(&"Optional".to_string()));
}
#[test]
fn handles_wildcard_import() {
let src = "from os import *\n";
let result = extract(src);
assert_eq!(result.imports.len(), 1);
assert_eq!(result.imports[0].source_file, "os");
assert!(result.imports[0].imported_names.contains(&"*".to_string()));
}
#[test]
fn handles_dotted_import() {
let src = "import os.path\n";
let result = extract(src);
assert_eq!(result.imports.len(), 1);
assert_eq!(result.imports[0].source_file, "os.path");
}
#[test]
fn handles_method_call() {
let src = "class A:\n def foo(self):\n self.bar()\n";
let result = extract(src);
let callees: Vec<_> = result
.calls
.iter()
.map(|c| c.callee_name.as_str())
.collect();
assert!(callees.contains(&"bar"), "should extract self.bar() call");
}
#[test]
fn handles_attribute_assignment() {
let src = "class A:\n def foo(self):\n self.x = 5\n";
let result = extract(src);
let assign = result
.assignments
.iter()
.find(|a| a.target_name == "x")
.expect("should find self.x = 5 assignment");
assert!(!assign.is_return_assign, "5 is not a call");
}
#[test]
fn result_language_is_python() {
let result = extract(PYTHON_SOURCE);
assert_eq!(result.language, Language::Python);
assert_eq!(result.file_path, "test.py");
}
#[test]
fn nested_function_definitions() {
let src = "def outer():\n def inner():\n return 1\n return inner()\n";
let result = extract(src);
let funcs: Vec<_> = result
.nodes
.iter()
.filter(|n| n.label == NodeLabel::Function)
.collect();
let names: Vec<_> = funcs.iter().map(|n| n.name.as_str()).collect();
assert!(
names.contains(&"outer"),
"should extract top-level outer function"
);
assert!(
!names.contains(&"inner"),
"nested inner function must NOT be promoted to a Function node (P2-5)"
);
let inner_call = result.calls.iter().find(|c| c.callee_name == "inner");
assert!(
inner_call.is_some(),
"call to inner() should still be recorded"
);
}
#[test]
fn call_in_function_has_dotted_fqn_caller_qn() {
let src = "def caller():\n callee()\n";
let ext = PythonExtractor::new();
let result = ext
.extract(src, "/tmp/demo/main.py", "proj")
.expect("extraction should succeed");
let call = result
.calls
.iter()
.find(|c| c.callee_name == "callee")
.expect("should find call to callee");
assert_eq!(
call.caller_qn.as_deref(),
Some("proj.tmp.demo.main.py.caller"),
"caller_qn should be the dotted FQN of the enclosing function"
);
let caller_node = result
.nodes
.iter()
.find(|n| n.name == "caller")
.expect("should find caller function node");
assert_eq!(
call.caller_qn.as_deref(),
Some(caller_node.qualified_name.as_str()),
"caller_qn must match the caller function node id"
);
}
#[test]
fn top_level_call_has_none_caller_qn() {
let src = "callee()\n";
let ext = PythonExtractor::new();
let result = ext
.extract(src, "main.py", "proj")
.expect("extraction should succeed");
let call = result
.calls
.iter()
.find(|c| c.callee_name == "callee")
.expect("should find top-level call to callee");
assert!(
call.caller_qn.is_none(),
"top-level call should have None caller_qn"
);
}
#[test]
fn function_in_if_block_at_module_scope_is_indexed() {
let src = "if True:\n def conditional_fn():\n return 1\n";
let result = extract(src);
let funcs: Vec<_> = result
.nodes
.iter()
.filter(|n| n.label == NodeLabel::Function && n.name == "conditional_fn")
.collect();
assert_eq!(
funcs.len(),
1,
"function inside if-block at module scope should be indexed (P2-5)"
);
}
#[test]
fn function_in_try_block_at_module_scope_is_indexed() {
let src = "try:\n def try_fn():\n return 1\nexcept Exception:\n pass\n";
let result = extract(src);
let funcs: Vec<_> = result
.nodes
.iter()
.filter(|n| n.label == NodeLabel::Function && n.name == "try_fn")
.collect();
assert_eq!(
funcs.len(),
1,
"function inside try-block at module scope should be indexed (P2-5)"
);
}
#[test]
fn read_in_function_has_dotted_fqn_reader_qn() {
let src = "def caller(x):\n y = x + 1\n return y\n";
let ext = PythonExtractor::new();
let result = ext
.extract(src, "/tmp/demo/main.py", "proj")
.expect("extraction should succeed");
let read = result
.reads
.iter()
.find(|r| r.var_name == "x")
.expect("should find a read of x");
assert_eq!(
read.reader_qn.as_deref(),
Some("proj.tmp.demo.main.py.caller"),
"reader_qn should be the dotted FQN of the enclosing function"
);
let caller_node = result
.nodes
.iter()
.find(|n| n.name == "caller")
.expect("should find caller function node");
assert_eq!(
read.reader_qn.as_deref(),
Some(caller_node.qualified_name.as_str()),
"reader_qn must match the caller function node id"
);
}
#[test]
fn write_in_function_assignment_has_dotted_fqn_writer_qn() {
let src = "def caller(x):\n y = x + 1\n return y\n";
let ext = PythonExtractor::new();
let result = ext
.extract(src, "/tmp/demo/main.py", "proj")
.expect("extraction should succeed");
let write = result
.writes
.iter()
.find(|w| w.var_name == "y")
.expect("should find a write of y");
assert_eq!(
write.writer_qn.as_deref(),
Some("proj.tmp.demo.main.py.caller"),
"writer_qn should be the dotted FQN of the enclosing function"
);
let caller_node = result
.nodes
.iter()
.find(|n| n.name == "caller")
.expect("should find caller function node");
assert_eq!(
write.writer_qn.as_deref(),
Some(caller_node.qualified_name.as_str()),
"writer_qn must match the caller function node id"
);
}
#[test]
fn augmented_assignment_is_write() {
let src = "def caller(x):\n y = x\n y += 1\n return y\n";
let ext = PythonExtractor::new();
let result = ext
.extract(src, "/tmp/demo/main.py", "proj")
.expect("extraction should succeed");
let y_writes: Vec<_> = result.writes.iter().filter(|w| w.var_name == "y").collect();
assert!(
y_writes.len() >= 2,
"y should be written at least twice (assignment + augmented): {:?}",
y_writes
);
for w in y_writes {
assert_eq!(
w.writer_qn.as_deref(),
Some("proj.tmp.demo.main.py.caller"),
"writer_qn should be the dotted FQN of the enclosing function"
);
}
}
#[test]
fn for_loop_target_is_write() {
let src =
"def looper():\n s = 0\n for i in range(10):\n s = s + i\n return s\n";
let ext = PythonExtractor::new();
let result = ext
.extract(src, "/tmp/demo/main.py", "proj")
.expect("extraction should succeed");
let i_write = result
.writes
.iter()
.find(|w| w.var_name == "i")
.expect("should find a write of loop variable i");
assert_eq!(
i_write.writer_qn.as_deref(),
Some("proj.tmp.demo.main.py.looper"),
"loop variable writer_qn should be the dotted FQN of the enclosing function"
);
assert!(
result.writes.iter().any(|w| w.var_name == "s"),
"loop body assignment should write s"
);
assert!(
result.reads.iter().any(|r| r.var_name == "i"),
"loop body should read i"
);
}
#[test]
fn call_to_parenthesized_callee_is_extracted() {
let src = "def get_fn():\n pass\nresult = (get_fn)()\n";
let result = extract(src);
let callees: Vec<_> = result
.calls
.iter()
.map(|c| c.callee_name.as_str())
.collect();
assert!(
callees.contains(&"get_fn"),
"should extract call to parenthesized get_fn: {:?}",
callees
);
}
#[test]
fn call_to_chained_invocation_is_extracted() {
let src = "def factory():\n pass\nfactory()()\n";
let result = extract(src);
assert!(
result.calls.iter().any(|c| c.callee_name == "factory"),
"chained call should record the inner factory() call: {:?}",
result.calls
);
}
#[test]
fn tuple_assignment_extracts_first_target() {
let src = "a, b = 1, 2\n";
let result = extract(src);
let assign = result
.assignments
.iter()
.find(|a| a.target_name == "a")
.expect("should find tuple assignment to a");
assert_eq!(assign.line, 1);
}
#[test]
fn list_assignment_does_not_panic() {
let src = "[a, b] = [1, 2]\n";
let result = extract(src);
let _ = result.assignments;
}
#[test]
fn subscript_assignment_extracts_container_name() {
let src = "arr = [1, 2, 3]\narr[0] = 99\n";
let result = extract(src);
assert!(
result.assignments.iter().any(|a| a.target_name == "arr"),
"should find subscript assignment to arr: {:?}",
result.assignments
);
}
#[test]
fn from_import_with_alias_records_alias_name() {
let src = "from numpy import array as arr\n";
let result = extract(src);
assert_eq!(result.imports.len(), 1);
assert_eq!(result.imports[0].source_file, "numpy");
assert!(
result.imports[0]
.imported_names
.contains(&"arr".to_string()),
"aliased import should record the alias name: {:?}",
result.imports[0].imported_names
);
}
#[test]
fn multi_line_function_signature_uses_first_line() {
let src = "def add(a,\n b):\n return a + b\n";
let result = extract(src);
let add = result.nodes.iter().find(|n| n.name == "add").expect("add");
let sig = add.signature.as_deref().expect("signature should be set");
assert!(
!sig.contains('\n'),
"signature must be a single line, got: {sig:?}"
);
assert!(
sig.contains("add"),
"signature should contain the function name"
);
}
#[test]
fn class_with_attribute_base_class_creates_extends_edge() {
let src = "class Foo(module.Bar):\n pass\n";
let result = extract(src);
let extends: Vec<_> = result
.edges
.iter()
.filter(|e| e.edge_type == EdgeType::Extends)
.collect();
assert_eq!(
extends.len(),
1,
"should create 1 EXTENDS edge for attribute base: {:?}",
extends
);
assert!(
extends[0].target.contains("Bar"),
"EXTENDS target should contain 'Bar': {}",
extends[0].target
);
}
#[test]
fn class_with_call_base_class_creates_extends_edge() {
let src = "class Meta:\n pass\nclass Foo(Meta()):\n pass\n";
let result = extract(src);
let extends: Vec<_> = result
.edges
.iter()
.filter(|e| e.edge_type == EdgeType::Extends)
.collect();
assert!(
extends.iter().any(|e| e.target.contains("Meta")),
"should create EXTENDS edge with 'Meta' target for call base: {:?}",
extends
);
}
#[test]
fn class_with_unknown_base_class_type_creates_no_extends_edge() {
let src = "class Foo(123):\n pass\n";
let result = extract(src);
let extends: Vec<_> = result
.edges
.iter()
.filter(|e| e.edge_type == EdgeType::Extends)
.collect();
assert!(
extends.is_empty(),
"should NOT create EXTENDS edge for integer base: {:?}",
extends
);
}
#[test]
fn nested_class_combines_parent_and_child_scope() {
let src = "class Outer:\n class Inner:\n def method(self):\n pass\n";
let result = extract(src);
let method = result
.nodes
.iter()
.find(|n| n.name == "method")
.expect("should find method node");
assert!(
method.qualified_name.contains("Outer"),
"qualified_name should contain Outer scope: {}",
method.qualified_name
);
assert!(
method.qualified_name.contains("Inner"),
"qualified_name should contain Inner scope: {}",
method.qualified_name
);
}
#[test]
fn identifier_in_assert_statement_is_not_read() {
let src = "def f():\n assert x\n";
let result = extract(src);
let reads: Vec<_> = result.reads.iter().filter(|r| r.var_name == "x").collect();
assert!(
reads.is_empty(),
"identifier in assert_statement should not be a read: {:?}",
reads
);
}
#[test]
fn decorator_on_function_does_not_break_extraction() {
let src = "@staticmethod\ndef foo():\n return 1\n";
let result = extract(src);
let funcs: Vec<_> = result
.nodes
.iter()
.filter(|n| n.label == NodeLabel::Function)
.collect();
assert_eq!(funcs.len(), 1, "should still extract foo function");
assert_eq!(funcs[0].name, "foo");
}
#[test]
fn decorator_on_class_does_not_break_extraction() {
let src = "@dataclass\nclass Foo:\n x: int\n";
let result = extract(src);
let classes: Vec<_> = result
.nodes
.iter()
.filter(|n| n.label == NodeLabel::Class)
.collect();
assert_eq!(classes.len(), 1, "should still extract Foo class");
assert_eq!(classes[0].name, "Foo");
}
#[test]
fn async_function_does_not_break_extraction() {
let src = "async def fetch():\n return 1\n";
let result = extract(src);
let _ = result;
}
#[test]
fn function_with_type_hints() {
let src = "def add(a: int, b: int) -> int:\n return a + b\n";
let result = extract(src);
let func = result
.nodes
.iter()
.find(|n| n.name == "add")
.expect("should find add function");
assert_eq!(func.label, NodeLabel::Function);
assert!(func.signature.is_some());
assert!(func.signature.as_deref().unwrap().contains("add"));
}
#[test]
fn walrus_operator_does_not_break_extraction() {
let src = "def f():\n if (n := 10) > 5:\n return n\n";
let result = extract(src);
let _ = result;
}
#[test]
fn match_statement_does_not_break_extraction() {
let src = "def f(x):\n match x:\n case 1:\n return 'one'\n case _:\n return 'other'\n";
let result = extract(src);
let _ = result;
}
#[test]
fn comment_only_source_returns_empty_result() {
let result = extract("# just a comment\n");
assert!(
result.is_empty(),
"comment-only file should produce no nodes"
);
}
#[test]
fn lambda_does_not_break_extraction() {
let src = "f = lambda x: x + 1\n";
let result = extract(src);
let _ = result;
}
#[test]
fn list_comprehension_does_not_break_extraction() {
let src = "squares = [x**2 for x in range(10)]\n";
let result = extract(src);
let _ = result;
}
#[test]
fn try_except_block_does_not_break_extraction() {
let src = "def f():\n try:\n x = 1\n except Exception:\n pass\n";
let result = extract(src);
assert!(
result.nodes.iter().any(|n| n.name == "f"),
"should extract f function"
);
}
#[test]
fn with_statement_does_not_break_extraction() {
let src = "def f():\n with open('file') as fh:\n return fh.read()\n";
let result = extract(src);
assert!(
result.nodes.iter().any(|n| n.name == "f"),
"should extract f function"
);
}
#[test]
fn multiple_classes_with_same_method_name() {
let src = "class A:\n def run(self):\n pass\nclass B:\n def run(self):\n pass\n";
let result = extract(src);
let methods: Vec<_> = result
.nodes
.iter()
.filter(|n| n.label == NodeLabel::Method && n.name == "run")
.collect();
assert_eq!(methods.len(), 2, "should extract 2 run methods");
assert_ne!(
methods[0].qualified_name, methods[1].qualified_name,
"same-name methods on different classes must have distinct FQNs"
);
}
#[test]
fn global_variable_assignment() {
let src = "MAX_SIZE = 1024\n";
let result = extract(src);
let assign = result
.assignments
.iter()
.find(|a| a.target_name == "MAX_SIZE")
.expect("should find MAX_SIZE assignment");
assert_eq!(assign.source_name, "");
assert!(!assign.is_return_assign);
}
#[test]
fn class_with_docstring() {
let src = "class Foo:\n \"\"\"This is a docstring.\"\"\"\n pass\n";
let result = extract(src);
let classes: Vec<_> = result
.nodes
.iter()
.filter(|n| n.label == NodeLabel::Class)
.collect();
assert_eq!(classes.len(), 1);
assert_eq!(classes[0].name, "Foo");
}
fn parse_source(source: &str) -> tree_sitter::Tree {
let mut parser =
crate::parse::parser_factory::ParserFactory::create_parser(Language::Python)
.expect("parser");
parser.parse(source, None).expect("parse")
}
fn find_first_by_kind<'a>(
node: tree_sitter::Node<'a>,
kind: &str,
) -> Option<tree_sitter::Node<'a>> {
if node.kind() == kind {
return Some(node);
}
for i in 0..node.named_child_count() as u32 {
if let Some(child) = node.named_child(i) {
if let Some(found) = find_first_by_kind(child, kind) {
return Some(found);
}
}
}
None
}
#[test]
fn function_signature_single_line_function() {
let src = "def foo(): pass\n";
let tree = parse_source(src);
let root = tree.root_node();
let func = find_first_by_kind(root, "function_definition")
.expect("should find function_definition");
let sig = function_signature(func, src);
assert!(sig.is_some(), "single-line function should have signature");
assert!(
sig.as_deref().unwrap().contains("foo"),
"signature should contain foo: {sig:?}"
);
}
#[test]
fn aliased_import_name_without_alias_returns_name() {
let src = "import os\n";
let tree = parse_source(src);
let root = tree.root_node();
if let Some(aliased) = find_first_by_kind(root, "aliased_import") {
let name = aliased_import_name(aliased, src);
assert_eq!(
name,
Some("os".to_string()),
"aliased_import_name without alias should return name field"
);
}
}
#[test]
fn assignment_target_name_empty_tuple_returns_none() {
let src = "def f():\n () = ()\n";
let tree = parse_source(src);
let root = tree.root_node();
if let Some(assign) = find_first_by_kind(root, "assignment") {
if let Some(left) = assign.child_by_field_name("left") {
if left.kind() == "tuple" {
let name = assignment_target_name(left, src);
assert!(name.is_none(), "empty tuple should return None: {name:?}");
}
}
}
}
#[test]
fn assignment_target_name_fallback_accepts_valid_identifier_text() {
let src = "x = 1\n";
let tree = parse_source(src);
let root = tree.root_node();
let name = assignment_target_name(root, src);
let _ = name;
}
#[test]
fn call_arguments_returns_empty_when_no_arguments_field() {
let src = "x = 1\n";
let tree = parse_source(src);
let root = tree.root_node();
if let Some(assign) = find_first_by_kind(root, "assignment") {
let args = call_arguments(assign, src);
assert!(
args.is_empty(),
"call_arguments on assignment should return empty"
);
}
}
#[test]
fn is_python_read_position_returns_false_for_root_node() {
let src = "x = 1\n";
let tree = parse_source(src);
let root = tree.root_node();
assert!(
!is_python_read_position(root),
"root node has no parent, should return false"
);
}
#[test]
fn combine_scope_only_parent_returns_parent() {
assert_eq!(
combine_scope(Some("parent"), None),
Some("parent".to_string())
);
}
#[test]
fn combine_scope_neither_returns_none() {
assert_eq!(combine_scope(None, None), None);
}
}