repotoire 0.8.0

//! AST-driven extraction of [`super::predict::Evidence`] for Python
//! SQL-injection call sites.
//!
//! # Why a separate module
//!
//! The scorer in [`super::predict`] takes plain data
//! ([`super::predict::Evidence`]) so it can be unit-tested without an
//! AST. This module's job is to populate that struct from a
//! `tree_sitter::Node` for a Python `call` expression whose callee
//! names a SQL sink (`cursor.execute`, `Model.objects.raw`, etc.).
//!
//! Splitting the two halves matches Phase 2a/2b/2c/2d/2e/2f/2g/2h/2i's
//! `evidence.rs` split.
//!
//! # What this module knows about
//!
//! - Walking the module AST to collect every call whose callee chain
//!   names a recognized SQL sink method
//!   ([`collect_python_sql_sites`]).
//! - **Structural classification of the SQL argument** — the heart of
//!   Phase 2j and its cleanest-in-series AST split:
//!   * f-string (`fstring` AST kind) with SQL keywords in the first
//!     positional argument → `SqlApi::Unsafe` (D1.b collapse).
//!   * `+`-concat / `.format()` / `%`-operator on a SQL-keyword
//!     string → `SqlApi::Unsafe` (D1.b collapse).
//!   * Django `Model.objects.raw(<formatted SQL>)` → `SqlApi::UnsafeRaw`
//!     (D1.c collapse — **the Phase 2j headline**).
//!   * Django `Model.objects.filter/get/create/...` keyword-filter
//!     expression → `SqlApi::Safe` (D1.a collapse).
//!   * Parameterized `execute(<string literal>, <values>)` with TWO
//!     positional arguments → `SqlApi::Safe` (D1.a collapse).
//!   * SQLAlchemy `db.execute(text(<literal>), {<binds>})` →
//!     `SqlApi::Safe` (D1.a collapse).
//!   * Static literal SQL with no formatting (one-arg `execute`) →
//!     `SqlApi::Safe` (D1.a collapse).
//!   * Otherwise → `SqlApi::Ambiguous` (weighted scoring).
//! - Walking up from the call node to the enclosing
//!   `function_definition` (for name) and `class_definition`
//!   (informational).
//! - Detecting the function's decorator list (for the route-handler
//!   signal) and naming-convention match.
//! - Detecting user-input source family (TypedString vs
//!   UnstructuredJson) within ±10 lines of the call.
//! - Reading the source line for `# repotoire: sql-safe[...]` /
//!   `sql-vulnerable[...]` annotations.
//!
//! # What this module deliberately does NOT do
//!
//! - Does not look for evidence in non-Python languages (D5.1 scope —
//!   JS / Go / Java deferred to a follow-on phase).
//! - Does not trace cross-statement query assembly
//!   (`q = build_query(req.form); cursor.execute(q)`) — D5.4 v0
//!   limitation; user annotates with `# repotoire: sql-vulnerable[...]`.
//! - Does not detect type-cast laundering inside `.format()` —
//!   `"...{}".format(int(x))` is conservatively classified Unsafe.
//!   D5.3 v0 limitation; user annotates with
//!   `# repotoire: sql-safe[type-cast-laundered]`.
//! - Does not resolve the model class for `.raw()` vs `.filter()` via
//!   type inference — the discriminator is purely AST-text-based
//!   (the method name at the end of the receiver chain). D5.5 v0
//!   limitation.
//!
//! # Status
//!
//! Wired in via `mod.rs::scan_python_file_dual_branch` in Phase 2j's
//! commit 5 (integration). The `#![allow(dead_code)]` at the top of
//! this file is removed in commit 5 alongside the integration.

use super::predict::{
    classify_user_input_source, extract_sql_safe_reason, extract_sql_vulnerable_source,
    is_safe_django_orm_method, is_sql_sink_method, is_sqlalchemy_text_function,
    is_unsafe_raw_django_orm_method, line_contains_sql_keyword, matches_route_handler_decorator,
    matches_route_handler_name, matches_trust_boundary_name, Evidence, SqlApi, UserInputSource,
};
use crate::detectors::security::ast_helpers::{enclosing_python_function, node_text};
use tree_sitter::Node;

/// A Python SQL call site discovered by walking the module AST.
/// Returned by [`collect_python_sql_sites`] so the integration in
/// `mod.rs` can iterate without re-walking.
pub(super) struct PythonSqlSite<'a> {
    pub call_node: Node<'a>,
    pub api: SqlApi,
    /// The raw callee text used for the title/description (e.g.
    /// `cursor.execute`, `User.objects.raw`). Resolved from the call
    /// expression's `function` child.
    pub callee_label: String,
}

/// Walk a Python module AST and collect every call whose callee
/// receiver-chain ends with a recognized SQL sink method.
///
/// Recognized sink methods:
/// - Generic SQL sinks: `execute`, `executemany`, `executescript`,
///   `scalar`, `scalars`, `query`, etc. (see `SQL_SINK_METHODS`).
/// - Django ORM safe methods: `filter`, `get`, `create`, etc. via
///   `.objects.<method>(...)` chain.
/// - Django ORM raw methods: `raw`, `extra` via `.objects.<method>(...)`
///   chain — these trigger the D1.c collapse.
///
/// Performs structural API classification on the call's argument
/// shape; returns one of `SqlApi::Safe` / `Unsafe` / `UnsafeRaw` /
/// `Ambiguous`.
///
/// Skips bare calls (no receiver chain).
pub(super) fn collect_python_sql_sites<'a>(
    module_root: Node<'a>,
    source: &'a [u8],
) -> Vec<PythonSqlSite<'a>> {
    let mut sites = Vec::new();
    let mut stack: Vec<Node<'_>> = vec![module_root];
    while let Some(node) = stack.pop() {
        let mut cursor = node.walk();
        for child in node.children(&mut cursor) {
            stack.push(child);
        }
        if node.kind() != "call" {
            continue;
        }
        let Some(func) = node.child_by_field_name("function") else {
            continue;
        };
        let func_text = node_text(func, source).unwrap_or("");
        if func_text.is_empty() {
            continue;
        }

        // Recognized SQL sink method?
        let kind = recognized_sql_call_kind(func_text);
        if kind == RecognizedSqlCallKind::NotRecognized {
            continue;
        }

        let api = classify_sql_shape(node, source, kind);

        sites.push(PythonSqlSite {
            call_node: node,
            api,
            callee_label: func_text.to_string(),
        });
    }
    sites
}

/// Distinguish the three kinds of recognized SQL call shapes by
/// callee chain — drives the structural classifier branch selection.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(super) enum RecognizedSqlCallKind {
    /// `<receiver>.<sink_method>(...)` where `<sink_method>` is
    /// `execute` / `executemany` / etc.
    GenericSink,
    /// `<Model>.objects.filter/get/create/...` (safe ORM keyword filter).
    DjangoOrmSafe,
    /// `<Model>.objects.raw(...)` or `.extra(...)`.
    DjangoOrmRaw,
    /// Not a recognized SQL call shape.
    NotRecognized,
}

/// Classify the callee text into one of the [`RecognizedSqlCallKind`]
/// variants. Purely AST-text based: looks at the last method-name
/// segment and inspects whether the receiver chain contains `.objects`
/// to discriminate Django ORM from generic SQL sinks.
fn recognized_sql_call_kind(func_text: &str) -> RecognizedSqlCallKind {
    let Some(method_name) = func_text.rsplit('.').next() else {
        return RecognizedSqlCallKind::NotRecognized;
    };
    if method_name == func_text {
        // No dot — bare call. Not a method call.
        return RecognizedSqlCallKind::NotRecognized;
    }

    // Django ORM .objects.<method> dispatch.
    if func_text.contains(".objects.") {
        if is_unsafe_raw_django_orm_method(method_name) {
            return RecognizedSqlCallKind::DjangoOrmRaw;
        }
        if is_safe_django_orm_method(method_name) {
            return RecognizedSqlCallKind::DjangoOrmSafe;
        }
        return RecognizedSqlCallKind::NotRecognized;
    }

    // Generic SQL sink (execute / executemany / ...).
    if is_sql_sink_method(method_name) {
        return RecognizedSqlCallKind::GenericSink;
    }

    RecognizedSqlCallKind::NotRecognized
}

/// Structural classification of the SQL call's argument shape. The
/// heart of Phase 2j's evidence extraction — the cleanest Safe/Unsafe
/// AST split in the series.
///
/// Per-kind classification:
///
/// * `DjangoOrmSafe` (`Model.objects.filter(...)`, etc.) → unconditionally
///   `SqlApi::Safe`. Keyword-filter expressions don't take raw SQL.
///
/// * `DjangoOrmRaw` (`Model.objects.raw(...)`):
///   - Argument is f-string / concat / format / `%` → `SqlApi::UnsafeRaw`
///     (D1.c collapse — **the Phase 2j headline**).
///   - Argument is a string literal with no formatting → `SqlApi::Ambiguous`
///     (could be safe if developer is careful; legacy scanner catches
///     concat downstream).
///
/// * `GenericSink` (`cursor.execute(...)`, etc.):
///   - Argument is f-string / concat / format / `%` → `SqlApi::Unsafe`
///     (D1.b collapse).
///   - Call has TWO positional args + first is a string literal →
///     `SqlApi::Safe` (parameterized execute).
///   - First arg is `text(<literal>)` and call has a 2nd arg that's a
///     dict → `SqlApi::Safe` (SQLAlchemy bound-params).
///   - First arg is a string literal (no formatting) and no 2nd arg →
///     `SqlApi::Safe` (static literal).
///   - First arg is an identifier (opaque variable) → `SqlApi::Ambiguous`.
///   - Otherwise → `SqlApi::Ambiguous`.
fn classify_sql_shape(call_node: Node<'_>, source: &[u8], kind: RecognizedSqlCallKind) -> SqlApi {
    let Some(args) = call_node.child_by_field_name("arguments") else {
        return SqlApi::Ambiguous;
    };
    let positional = positional_args(args);

    match kind {
        RecognizedSqlCallKind::DjangoOrmSafe => SqlApi::Safe,

        RecognizedSqlCallKind::DjangoOrmRaw => {
            let Some(first) = positional.first() else {
                return SqlApi::Ambiguous;
            };
            if first_arg_is_formatted_sql(*first, source) {
                SqlApi::UnsafeRaw
            } else {
                SqlApi::Ambiguous
            }
        }

        RecognizedSqlCallKind::GenericSink => {
            let Some(first) = positional.first() else {
                return SqlApi::Ambiguous;
            };
            // First check: formatted SQL → Unsafe (D1.b).
            if first_arg_is_formatted_sql(*first, source) {
                return SqlApi::Unsafe;
            }
            // Parameterized: literal + values.
            if positional.len() >= 2 && is_string_literal(*first) {
                return SqlApi::Safe;
            }
            // SQLAlchemy text(<literal>) + bound-params dict.
            if positional.len() >= 2
                && is_sqlalchemy_text_call_with_literal(*first, source)
                && positional
                    .get(1)
                    .map(|n| n.kind() == "dictionary")
                    .unwrap_or(false)
            {
                return SqlApi::Safe;
            }
            // Static literal, no formatting, single arg → safe (no
            // user input in the SQL string).
            if positional.len() == 1 && is_string_literal(*first) {
                // Check that the literal doesn't contain any
                // formatting placeholder character that would indicate
                // an unparameterized template (e.g., `f""` already
                // handled above, but `"... {}".format(...)` was
                // checked by formatted_sql).
                return SqlApi::Safe;
            }
            SqlApi::Ambiguous
        }

        RecognizedSqlCallKind::NotRecognized => SqlApi::Ambiguous,
    }
}

/// Return the positional arguments in order, skipping keyword arguments.
fn positional_args<'a>(args: Node<'a>) -> Vec<Node<'a>> {
    let mut out = Vec::new();
    let mut cursor = args.walk();
    for child in args.children(&mut cursor) {
        if !child.is_named() {
            continue;
        }
        if child.kind() == "keyword_argument" {
            continue;
        }
        out.push(child);
    }
    out
}

/// True iff the given argument node is a Python string literal
/// (`string` AST kind) that is NOT an f-string. f-strings in tree-
/// sitter-python may be either kind `string` (with `interpolation`
/// children) or distinct depending on grammar version; we check both.
fn is_string_literal(arg: Node<'_>) -> bool {
    if arg.kind() != "string" {
        return false;
    }
    // Reject f-strings: a `string` whose first child is `string_start`
    // beginning with `f"` or which contains `interpolation` children.
    !string_is_fstring(arg)
}

/// True iff the `string` node is an f-string. Detection: scan children
/// for `interpolation` named children, OR check the leading
/// `string_start` token text for an `f` prefix.
fn string_is_fstring(arg: Node<'_>) -> bool {
    if arg.kind() != "string" {
        return false;
    }
    let mut cursor = arg.walk();
    for child in arg.children(&mut cursor) {
        if child.kind() == "interpolation" {
            return true;
        }
    }
    // Fall back to checking the source bytes for the `f` prefix.
    // Newer tree-sitter-python grammars use a dedicated `string_start`
    // child whose text is `f"` or `f'`.
    let start = arg.start_byte();
    let end = arg.end_byte();
    if start >= end {
        return false;
    }
    // Look at the first ~3 chars for an `f` / `F` / `rf` / `Rf` prefix.
    // We can't get source here without it being passed in — so the
    // fallback is interpolation-child-based only. Caller must pass
    // source through `arg_text_contains_fstring_prefix` for the
    // text-based check; for evidence classification we additionally
    // check the surrounding text in the caller.
    false
}

/// True iff the first positional argument is a "formatted SQL"
/// expression: f-string, `+`-concat with a string literal, `.format()`
/// method call on a string literal, or `%`-operator on a string
/// literal. All four are the D1.b/D1.c structural-pattern triggers.
fn first_arg_is_formatted_sql(arg: Node<'_>, source: &[u8]) -> bool {
    // f-string detection: a `string` node whose source text starts
    // with `f"` / `f'` / `F"` / etc., or which has `interpolation`
    // children.
    if arg.kind() == "string" {
        if string_is_fstring(arg) {
            // Need SQL keyword present too; otherwise it's a
            // plain f-string with non-SQL text.
            let text = node_text(arg, source).unwrap_or("");
            return line_contains_sql_keyword(text);
        }
        let text = node_text(arg, source).unwrap_or("");
        // Source-text-based fallback for f-string detection: leading
        // `f` / `F` / `rf` / `fr` prefix on the literal.
        let lower = text.to_lowercase();
        if (lower.starts_with("f\"") || lower.starts_with("f'") || lower.starts_with("rf"))
            && line_contains_sql_keyword(text)
        {
            return true;
        }
        return false;
    }
    // Binary `+` concatenation with a string literal anywhere — likely
    // SQL string concat. AST kind `binary_operator`; operator is `+`.
    if arg.kind() == "binary_operator" {
        let op = arg
            .child_by_field_name("operator")
            .map(|n| node_text(n, source).unwrap_or(""))
            .unwrap_or("");
        if op == "+" && binary_op_contains_string_literal_with_sql(arg, source) {
            return true;
        }
        // `%` operator: `"SELECT ... %s" % var`.
        if op == "%" && binary_op_contains_string_literal_with_sql(arg, source) {
            return true;
        }
    }
    // `"...".format(...)` — a `call` whose function chain ends in
    // `.format` and whose receiver is a string literal with SQL.
    if arg.kind() == "call" {
        if let Some(func) = arg.child_by_field_name("function") {
            let func_text = node_text(func, source).unwrap_or("");
            if func_text.ends_with(".format") {
                // Inspect the receiver (the part before `.format`).
                if let Some(receiver_text) = func_text.strip_suffix(".format") {
                    // The receiver text in the AST is whatever was
                    // before the dot; we need to know if it's a SQL
                    // string. Look at the function's child structure:
                    // it's typically `attribute` with an `object`
                    // field that is the string literal.
                    if let Some(object) = func.child_by_field_name("object") {
                        if object.kind() == "string" {
                            let s = node_text(object, source).unwrap_or("");
                            if line_contains_sql_keyword(s) {
                                return true;
                            }
                        }
                    }
                    // Fall back: substring check on the receiver text
                    // (handles cases where the AST shape differs).
                    if line_contains_sql_keyword(receiver_text) {
                        return true;
                    }
                }
            }
        }
    }
    false
}

/// True iff a `binary_operator` node contains a string-literal operand
/// with a SQL keyword in its text.
fn binary_op_contains_string_literal_with_sql(arg: Node<'_>, source: &[u8]) -> bool {
    let mut cursor = arg.walk();
    for child in arg.children(&mut cursor) {
        if child.kind() == "string" {
            let text = node_text(child, source).unwrap_or("");
            if line_contains_sql_keyword(text) {
                return true;
            }
        }
        // Recurse one level into nested binary_operator for `a + b + c`.
        if child.kind() == "binary_operator"
            && binary_op_contains_string_literal_with_sql(child, source)
        {
            return true;
        }
    }
    false
}

/// True iff `arg` is a call expression `text(<string-literal>)` —
/// the SQLAlchemy parameterized-SQL constructor.
fn is_sqlalchemy_text_call_with_literal(arg: Node<'_>, source: &[u8]) -> bool {
    if arg.kind() != "call" {
        return false;
    }
    let Some(func) = arg.child_by_field_name("function") else {
        return false;
    };
    let func_text = node_text(func, source).unwrap_or("");
    // Accept either bare `text(...)` or `sqlalchemy.text(...)` /
    // similar; the discriminator is the last identifier segment.
    let last = func_text.rsplit('.').next().unwrap_or("");
    if !is_sqlalchemy_text_function(last) {
        return false;
    }
    let Some(args) = arg.child_by_field_name("arguments") else {
        return false;
    };
    let positional = positional_args(args);
    let Some(first) = positional.first() else {
        return false;
    };
    is_string_literal(*first)
}

// ─────────────────────────────────────────────────────────────────────────────
// Extract full evidence from a classified call site
// ─────────────────────────────────────────────────────────────────────────────

/// Extract typed evidence from a Python SQL call node.
///
/// `call_node` must be a `call` AST node whose callee chain ends with
/// a SQL sink method (post-`collect_python_sql_sites` filter).
/// `module_root` is the file's module-level root node. `source` is
/// the file's raw bytes. `lines` is the pre-split source-line slice
/// the scanner already builds; used for the annotation parsing and
/// the user-input proximity check.
///
/// `file_path` is the source file path string; populated into the
/// returned `Evidence.file_path` for diagnostics. `api` is the
/// classification produced by `collect_python_sql_sites`.
///
/// Never panics; missing fields produce defaults.
pub(super) fn extract_python_evidence<'a>(
    call_node: Node<'a>,
    _module_root: Node<'a>,
    source: &'a [u8],
    lines: &[&str],
    file_path: Option<String>,
    api: SqlApi,
    callee_label: String,
) -> Evidence {
    let mut ev = Evidence {
        file_path,
        api: Some(api),
        callee_label: Some(callee_label),
        ..Default::default()
    };

    // ── Enclosing function (for name) and class. ──
    let fn_node = enclosing_python_function(call_node);
    if let Some(fn_node) = fn_node {
        if let Some(name_node) = fn_node.child_by_field_name("name") {
            if let Some(name) = node_text(name_node, source) {
                ev.enclosing_function = Some(name.to_string());
            }
        }
    }
    ev.enclosing_class = enclosing_python_class_name(call_node, source);

    // ── Function-name-based signals. ──
    if let Some(fn_name) = &ev.enclosing_function {
        ev.trust_boundary_name = matches_trust_boundary_name(fn_name);
        if matches_route_handler_name(fn_name) {
            ev.enclosing_route_handler = true;
        }
    }

    // ── Decorator-based route-handler detection. ──
    if !ev.enclosing_route_handler {
        if let Some(fn_node) = fn_node {
            if function_has_route_decorator(fn_node, lines) {
                ev.enclosing_route_handler = true;
            }
        }
    }

    // ── User-input source classification (±10 lines, prioritizing
    //    UnstructuredJson). ──
    let call_line = call_node.start_position().row;
    ev.user_input_source = classify_nearby_user_input(lines, call_line, 10);

    // ── First-argument inspection for the static-literal /
    //    no-formatting Ambiguous-bucket signals. ──
    if let Some(args) = call_node.child_by_field_name("arguments") {
        let positional = positional_args(args);
        if let Some(first) = positional.first() {
            let is_literal = is_string_literal(*first);
            let is_formatted = first_arg_is_formatted_sql(*first, source);
            if is_literal && !is_formatted {
                ev.static_sql_string_literal = true;
            }
            if !is_formatted {
                if let Some(line) = lines.get(call_line) {
                    if line_contains_sql_keyword(line) {
                        ev.sql_keyword_no_formatting = true;
                    }
                }
            }
        }
    }

    // ── Source-line annotations. ──
    if let Some(line) = lines.get(call_line) {
        ev.sql_safe_annotation = extract_sql_safe_reason(line);
        ev.sql_vulnerable_annotation = extract_sql_vulnerable_source(line);
    }

    ev
}

/// Classify the user-input source family appearing in any line within
/// ±`radius` of `call_line` (0-indexed). Priority: UnstructuredJson
/// over TypedString (per the predict.rs `classify_user_input_source`).
fn classify_nearby_user_input(lines: &[&str], call_line: usize, radius: usize) -> UserInputSource {
    let start = call_line.saturating_sub(radius);
    let end = (call_line + radius + 1).min(lines.len());
    for line in &lines[start..end] {
        if matches!(
            classify_user_input_source(line),
            UserInputSource::UnstructuredJson
        ) {
            return UserInputSource::UnstructuredJson;
        }
    }
    for line in &lines[start..end] {
        if matches!(
            classify_user_input_source(line),
            UserInputSource::TypedString
        ) {
            return UserInputSource::TypedString;
        }
    }
    UserInputSource::None
}

// ─────────────────────────────────────────────────────────────────────────────
// Decorator inspection
// ─────────────────────────────────────────────────────────────────────────────

/// True iff the `function_definition` node (Python) is preceded by a
/// `decorated_definition` whose decorator list contains a recognized
/// route-handler decorator.
fn function_has_route_decorator(fn_node: Node<'_>, lines: &[&str]) -> bool {
    let mut parent = fn_node.parent();
    while let Some(p) = parent {
        if p.kind() == "decorated_definition" {
            let mut cursor = p.walk();
            for child in p.children(&mut cursor) {
                if child.kind() == "decorator" {
                    let line_idx = child.start_position().row;
                    if let Some(line) = lines.get(line_idx) {
                        if matches_route_handler_decorator(line) {
                            return true;
                        }
                    }
                }
            }
            break;
        }
        if p.kind() == "module" {
            break;
        }
        parent = p.parent();
    }
    false
}

// ─────────────────────────────────────────────────────────────────────────────
// Enclosing class
// ─────────────────────────────────────────────────────────────────────────────

fn enclosing_python_class_name<'a>(node: Node<'a>, source: &'a [u8]) -> Option<String> {
    let mut cur = node.parent()?;
    loop {
        if cur.kind() == "class_definition" {
            let name = cur.child_by_field_name("name")?;
            return node_text(name, source).map(str::to_string);
        }
        if cur.kind() == "module" {
            return None;
        }
        cur = cur.parent()?;
    }
}

// ─────────────────────────────────────────────────────────────────────────────
// Tests
// ─────────────────────────────────────────────────────────────────────────────

#[cfg(test)]
mod tests {
    use super::*;
    use crate::detectors::ast_fingerprint::parse_root_ext;
    use crate::parsers::lightweight::Language;

    /// Parse `source` as Python and find the first `call` node whose
    /// function chain ends with the given attribute or identifier
    /// name.
    fn first_call_with_attr<'tree>(
        tree: &'tree tree_sitter::Tree,
        source: &[u8],
        attr_name: &str,
    ) -> tree_sitter::Node<'tree> {
        fn walk<'a>(
            node: tree_sitter::Node<'a>,
            source: &[u8],
            attr_name: &str,
        ) -> Option<tree_sitter::Node<'a>> {
            if node.kind() == "call" {
                if let Some(func) = node.child_by_field_name("function") {
                    let text = node_text(func, source).unwrap_or("");
                    let last = text.rsplit('.').next().unwrap_or("");
                    if last == attr_name {
                        return Some(node);
                    }
                }
            }
            let mut cursor = node.walk();
            for child in node.children(&mut cursor) {
                if let Some(found) = walk(child, source, attr_name) {
                    return Some(found);
                }
            }
            None
        }
        walk(tree.root_node(), source, attr_name)
            .unwrap_or_else(|| panic!("no call to {} found in source", attr_name))
    }

    fn extract(src: &str, attr: &str) -> Evidence {
        let tree = parse_root_ext(src, Language::Python, "py").expect("parse python");
        let root = tree.root_node();
        let call = first_call_with_attr(&tree, src.as_bytes(), attr);
        let lines: Vec<&str> = src.lines().collect();
        let func_text = call
            .child_by_field_name("function")
            .and_then(|f| node_text(f, src.as_bytes()))
            .unwrap_or("")
            .to_string();
        let kind = recognized_sql_call_kind(&func_text);
        let api = classify_sql_shape(call, src.as_bytes(), kind);
        extract_python_evidence(call, root, src.as_bytes(), &lines, None, api, func_text)
    }

    fn collect_sites(src: &str) -> Vec<(SqlApi, String)> {
        let tree = parse_root_ext(src, Language::Python, "py").expect("parse python");
        let root = tree.root_node();
        collect_python_sql_sites(root, src.as_bytes())
            .into_iter()
            .map(|s| (s.api, s.callee_label))
            .collect()
    }

    // ─── collect_python_sql_sites: receiver-method shape ───
    #[test]
    fn collect_picks_up_cursor_execute() {
        let sites = collect_sites("cursor.execute(\"SELECT 1\")\n");
        assert_eq!(sites.len(), 1);
        assert_eq!(sites[0].1, "cursor.execute");
    }

    #[test]
    fn collect_picks_up_django_orm_filter() {
        let sites = collect_sites("User.objects.filter(id=1)\n");
        assert_eq!(sites.len(), 1);
        assert_eq!(sites[0].1, "User.objects.filter");
        assert_eq!(sites[0].0, SqlApi::Safe);
    }

    #[test]
    fn collect_picks_up_django_orm_raw() {
        let sites = collect_sites("User.objects.raw(\"SELECT * FROM users\")\n");
        assert_eq!(sites.len(), 1);
        assert_eq!(sites[0].1, "User.objects.raw");
    }

    #[test]
    fn collect_skips_bare_execute_call() {
        let sites = collect_sites("execute(\"SELECT 1\")\n");
        assert!(sites.is_empty());
    }

    #[test]
    fn collect_skips_non_sql_method() {
        let sites = collect_sites("user.greet()\n");
        assert!(sites.is_empty());
    }

    // ─── recognized_sql_call_kind ───
    #[test]
    fn recognized_kind_generic_execute() {
        assert_eq!(
            recognized_sql_call_kind("cursor.execute"),
            RecognizedSqlCallKind::GenericSink
        );
    }

    #[test]
    fn recognized_kind_orm_safe() {
        assert_eq!(
            recognized_sql_call_kind("User.objects.filter"),
            RecognizedSqlCallKind::DjangoOrmSafe
        );
        assert_eq!(
            recognized_sql_call_kind("Article.objects.get"),
            RecognizedSqlCallKind::DjangoOrmSafe
        );
    }

    #[test]
    fn recognized_kind_orm_raw() {
        assert_eq!(
            recognized_sql_call_kind("User.objects.raw"),
            RecognizedSqlCallKind::DjangoOrmRaw
        );
        assert_eq!(
            recognized_sql_call_kind("Article.objects.extra"),
            RecognizedSqlCallKind::DjangoOrmRaw
        );
    }

    #[test]
    fn recognized_kind_unknown_falls_through() {
        assert_eq!(
            recognized_sql_call_kind("foo.bar.unknown"),
            RecognizedSqlCallKind::NotRecognized
        );
        assert_eq!(
            recognized_sql_call_kind("bare_call"),
            RecognizedSqlCallKind::NotRecognized
        );
    }

    // ─── Structural classification: D1.a Safe ───
    #[test]
    fn classify_parameterized_execute_with_tuple() {
        let src = "cursor.execute(\"SELECT * FROM u WHERE id = %s\", (uid,))\n";
        let sites = collect_sites(src);
        assert_eq!(sites[0].0, SqlApi::Safe);
    }

    #[test]
    fn classify_parameterized_execute_with_list() {
        let src = "cursor.execute(\"SELECT * FROM u WHERE id = ?\", [uid])\n";
        let sites = collect_sites(src);
        assert_eq!(sites[0].0, SqlApi::Safe);
    }

    #[test]
    fn classify_django_orm_filter_safe() {
        let src = "User.objects.filter(id=request.GET['id'])\n";
        let sites = collect_sites(src);
        assert_eq!(sites[0].0, SqlApi::Safe);
    }

    #[test]
    fn classify_django_orm_get_safe() {
        let src = "User.objects.get(id=1)\n";
        let sites = collect_sites(src);
        assert_eq!(sites[0].0, SqlApi::Safe);
    }

    #[test]
    fn classify_sqlalchemy_text_with_bound_dict_safe() {
        let src = "db.execute(text(\"SELECT :id\"), {\"id\": uid})\n";
        let sites = collect_sites(src);
        assert_eq!(sites[0].0, SqlApi::Safe);
    }

    #[test]
    fn classify_static_literal_single_arg_safe() {
        let src = "cursor.execute(\"INSERT INTO log VALUES ('static')\")\n";
        let sites = collect_sites(src);
        assert_eq!(sites[0].0, SqlApi::Safe);
    }

    // ─── Structural classification: D1.b Unsafe ───
    #[test]
    fn classify_fstring_execute_unsafe() {
        let src = "cursor.execute(f\"SELECT * FROM u WHERE id = {uid}\")\n";
        let sites = collect_sites(src);
        assert_eq!(sites[0].0, SqlApi::Unsafe);
    }

    #[test]
    fn classify_concat_execute_unsafe() {
        let src = "cursor.execute(\"SELECT * FROM u WHERE id = \" + uid)\n";
        let sites = collect_sites(src);
        assert_eq!(sites[0].0, SqlApi::Unsafe);
    }

    #[test]
    fn classify_format_execute_unsafe() {
        let src = "cursor.execute(\"SELECT * FROM u WHERE id = {}\".format(uid))\n";
        let sites = collect_sites(src);
        assert_eq!(sites[0].0, SqlApi::Unsafe);
    }

    #[test]
    fn classify_percent_execute_unsafe() {
        let src = "cursor.execute(\"SELECT * FROM u WHERE id = %s\" % uid)\n";
        let sites = collect_sites(src);
        assert_eq!(sites[0].0, SqlApi::Unsafe);
    }

    // ─── Structural classification: D1.c UnsafeRaw — THE HEADLINE ───
    #[test]
    fn classify_django_raw_concat_unsafe_raw() {
        let src = "User.objects.raw(\"SELECT * FROM u WHERE id = \" + uid)\n";
        let sites = collect_sites(src);
        assert_eq!(sites[0].0, SqlApi::UnsafeRaw);
    }

    #[test]
    fn classify_django_raw_fstring_unsafe_raw() {
        let src = "User.objects.raw(f\"SELECT * FROM u WHERE id = {uid}\")\n";
        let sites = collect_sites(src);
        assert_eq!(sites[0].0, SqlApi::UnsafeRaw);
    }

    #[test]
    fn classify_django_raw_format_unsafe_raw() {
        let src = "User.objects.raw(\"SELECT * FROM u WHERE id = {}\".format(uid))\n";
        let sites = collect_sites(src);
        assert_eq!(sites[0].0, SqlApi::UnsafeRaw);
    }

    #[test]
    fn classify_django_raw_static_literal_ambiguous() {
        // .raw() with a static SQL literal is NOT D1.c — falls
        // through to Ambiguous. The user has to add user input
        // (formatting) for D1.c to fire.
        let src = "User.objects.raw(\"SELECT * FROM u\")\n";
        let sites = collect_sites(src);
        assert_eq!(sites[0].0, SqlApi::Ambiguous);
    }

    // ─── Structural classification: Ambiguous fallback ───
    #[test]
    fn classify_opaque_variable_arg_ambiguous() {
        let src = "cursor.execute(query)\n";
        let sites = collect_sites(src);
        assert_eq!(sites[0].0, SqlApi::Ambiguous);
    }

    #[test]
    fn classify_function_call_arg_ambiguous() {
        let src = "cursor.execute(build_query())\n";
        let sites = collect_sites(src);
        assert_eq!(sites[0].0, SqlApi::Ambiguous);
    }

    // ─── Enclosing function detection ───
    #[test]
    fn detects_enclosing_function_name() {
        let src = "\
            def get_user(user_id):\n\
            \x20   return cursor.execute(\"SELECT * FROM u WHERE id = %s\", (user_id,))\n";
        let ev = extract(src, "execute");
        assert_eq!(ev.enclosing_function.as_deref(), Some("get_user"));
    }

    #[test]
    fn detects_enclosing_class_name() {
        let src = "\
            class UserRepo:\n\
            \x20   def get(self, uid):\n\
            \x20       return cursor.execute(\"SELECT * FROM u\")\n";
        let ev = extract(src, "execute");
        assert_eq!(ev.enclosing_class.as_deref(), Some("UserRepo"));
    }

    // ─── Decorator-based route-handler detection ───
    #[test]
    fn detects_flask_route_decorator() {
        let src = "\
            from flask import request\n\
            @app.route('/u', methods=['POST'])\n\
            def login():\n\
            \x20   return cursor.execute(f\"SELECT * FROM u WHERE n = {request.form['n']}\")\n";
        let ev = extract(src, "execute");
        assert!(ev.enclosing_route_handler);
    }

    #[test]
    fn detects_handler_name() {
        let src = "\
            def login_handler():\n\
            \x20   return cursor.execute(\"SELECT 1\")\n";
        let ev = extract(src, "execute");
        assert!(ev.enclosing_route_handler);
    }

    // ─── User-input source classification ───
    #[test]
    fn detects_unstructured_json_source() {
        let src = "\
            def f():\n\
            \x20   body = request.json\n\
            \x20   return cursor.execute(\"SELECT * WHERE x = %s\", (body['x'],))\n";
        let ev = extract(src, "execute");
        assert_eq!(ev.user_input_source, UserInputSource::UnstructuredJson);
    }

    #[test]
    fn detects_typed_string_source() {
        let src = "\
            def f():\n\
            \x20   n = request.form['name']\n\
            \x20   return cursor.execute(\"SELECT * WHERE n = %s\", (n,))\n";
        let ev = extract(src, "execute");
        assert_eq!(ev.user_input_source, UserInputSource::TypedString);
    }

    #[test]
    fn detects_no_user_input_source() {
        let src = "\
            def f():\n\
            \x20   return cursor.execute(\"SELECT 1\")\n";
        let ev = extract(src, "execute");
        assert_eq!(ev.user_input_source, UserInputSource::None);
    }

    // ─── Static literal / no-formatting Ambiguous-bucket signals ───
    #[test]
    fn detects_static_sql_string_literal_signal() {
        // The classifier already chose `Safe` for the single-arg
        // literal, but the additive signal is also computed (for
        // calls that ended up Ambiguous via other routes).
        let src = "cursor.execute(\"SELECT 1\")\n";
        let ev = extract(src, "execute");
        assert!(ev.static_sql_string_literal);
    }

    #[test]
    fn detects_sql_keyword_no_formatting_signal() {
        let src = "cursor.execute(\"SELECT 1\")\n";
        let ev = extract(src, "execute");
        assert!(ev.sql_keyword_no_formatting);
    }

    #[test]
    fn does_not_detect_static_literal_when_fstring() {
        let src = "cursor.execute(f\"SELECT * WHERE x = {x}\")\n";
        let ev = extract(src, "execute");
        assert!(!ev.static_sql_string_literal);
    }

    // ─── Annotation parsing ───
    #[test]
    fn extracts_sql_safe_annotation() {
        let src = "\
            def f():\n\
            \x20   return cursor.execute(q)  # repotoire: sql-safe[whitelisted-table]\n";
        let ev = extract(src, "execute");
        assert_eq!(ev.sql_safe_annotation.as_deref(), Some("whitelisted-table"));
    }

    #[test]
    fn extracts_sql_vulnerable_annotation() {
        let src = "\
            def f():\n\
            \x20   return cursor.execute(q)  # repotoire: sql-vulnerable[helper-built]\n";
        let ev = extract(src, "execute");
        assert_eq!(
            ev.sql_vulnerable_annotation.as_deref(),
            Some("helper-built")
        );
    }

    #[test]
    fn does_not_extract_unrelated_annotation() {
        let src = "\
            def f():\n\
            \x20   return cursor.execute(\"SELECT 1\")  # repotoire: jwt-safe[ok]\n";
        let ev = extract(src, "execute");
        assert!(ev.sql_safe_annotation.is_none());
        assert!(ev.sql_vulnerable_annotation.is_none());
    }

    // ─── End-to-end integration: full evidence for §6 Case A ───
    #[test]
    fn case_a_full_evidence_parameterized_execute() {
        let src = "\
            from flask import request\n\
            @app.route('/u', methods=['POST'])\n\
            def get_user():\n\
            \x20   return cursor.execute(\"SELECT * FROM u WHERE id = %s\", (request.form['id'],))\n";
        let ev = extract(src, "execute");
        assert_eq!(ev.api, Some(SqlApi::Safe));
        assert!(ev.enclosing_route_handler);
    }

    // ─── End-to-end integration: full evidence for §6 Case B ───
    #[test]
    fn case_b_full_evidence_fstring_execute() {
        let src = "\
            from flask import request\n\
            def get_user():\n\
            \x20   return cursor.execute(f\"SELECT * FROM u WHERE id = {request.form['id']}\")\n";
        let ev = extract(src, "execute");
        assert_eq!(ev.api, Some(SqlApi::Unsafe));
        assert_eq!(ev.user_input_source, UserInputSource::TypedString);
    }

    // ─── End-to-end integration: full evidence for §6 Case C ───
    #[test]
    fn case_c_full_evidence_type_cast_laundered_format() {
        // The v0 limitation D5.3: the developer cast to int(), but
        // the predictor sees `.format()` and classifies as Unsafe.
        // Annotation is the escape hatch.
        let src = "\
            def f():\n\
            \x20   return cursor.execute(\"SELECT * WHERE id = {}\".format(int(request.form['id'])))\n";
        let ev = extract(src, "execute");
        assert_eq!(ev.api, Some(SqlApi::Unsafe));
    }

    // ─── End-to-end integration: full evidence for §6 Case D ───
    #[test]
    fn case_d_full_evidence_django_orm_filter() {
        let src = "\
            from flask import request\n\
            def list_users():\n\
            \x20   return User.objects.filter(id=request.GET['id'])\n";
        let ev = extract(src, "filter");
        assert_eq!(ev.api, Some(SqlApi::Safe));
    }

    // ─── End-to-end integration: full evidence for §6 Case E (HEADLINE) ───
    #[test]
    fn case_e_full_evidence_django_raw_with_concat() {
        let src = "\
            from flask import request\n\
            def list_users():\n\
            \x20   return User.objects.raw(\"SELECT * FROM u WHERE id = \" + request.GET['id'])\n";
        let ev = extract(src, "raw");
        assert_eq!(ev.api, Some(SqlApi::UnsafeRaw));
    }

    // ─── End-to-end integration: full evidence for §6 Case F ───
    #[test]
    fn case_f_full_evidence_static_literal_insert() {
        let src = "\
            def f():\n\
            \x20   return cursor.execute(\"INSERT INTO log VALUES ('static')\")\n";
        let ev = extract(src, "execute");
        assert_eq!(ev.api, Some(SqlApi::Safe));
        assert!(ev.static_sql_string_literal);
    }

    // ─── End-to-end integration: full evidence for §6 Case G ───
    #[test]
    fn case_g_full_evidence_opaque_variable() {
        let src = "\
            from flask import request\n\
            def get_user():\n\
            \x20   q = build_query(request.form)\n\
            \x20   return cursor.execute(q)\n";
        let ev = extract(src, "execute");
        assert_eq!(ev.api, Some(SqlApi::Ambiguous));
        assert_eq!(ev.user_input_source, UserInputSource::TypedString);
    }

    // ─── End-to-end integration: full evidence for §6 Case H ───
    #[test]
    fn case_h_full_evidence_sqlalchemy_text_with_binds() {
        let src = "\
            from flask import request\n\
            def f():\n\
            \x20   return db.execute(text(\"SELECT :id\"), {\"id\": request.form['id']})\n";
        let ev = extract(src, "execute");
        assert_eq!(ev.api, Some(SqlApi::Safe));
    }
}