ss-tools/backend/tests/test_sql_table_extractor.py

# #region test_sql_table_extractor [C:3] [TYPE TestModule] [SEMANTICS test, sql, table, extractor, pytest]
# @BRIEF Unit tests for SqlTableExtractor — T017. Verifies three-phase extraction of schema.table references from SQL+Jinja.
# @LAYER Tests
# @RELATION BINDS_TO -> [SqlTableExtractorModule]
# @TEST_CONTRACT: extract_tables_from_sql(str) -> set[str]
# @TEST_EDGE: empty_input -> returns empty set
# @TEST_EDGE: no_schema_qualified_tables -> returns empty set
# @TEST_EDGE: string_literal_false_positive -> filtered out by sqlparse
# @TEST_FIXTURE: simple_select -> INLINE_SQL: "SELECT * FROM raw.sales"
# @TEST_FIXTURE: jinja_expr -> INLINE_SQL: "SELECT * FROM {{ source('raw', 'sales') }}"

from src.services.sql_table_extractor import extract_tables_from_sql


class TestExtractTablesFromSql:
    """T017: Unit tests for SqlTableExtractor.extract_tables_from_sql()."""

    # #region test_simple_select [C:2] [TYPE Function]
    # @BRIEF Happy path: simple SELECT with FROM schema.table.
    def test_simple_select(self):
        sql = "SELECT * FROM raw.sales"
        result = extract_tables_from_sql(sql)
        assert result == {"raw.sales"}
    # #endregion test_simple_select

    # #region test_multiple_tables [C:2] [TYPE Function]
    # @BRIEF JOIN with multiple schema-qualified tables.
    def test_multiple_tables(self):
        sql = "SELECT a.id, b.name FROM raw.sales a JOIN raw.inventory b ON a.id = b.sale_id"
        result = extract_tables_from_sql(sql)
        assert result == {"raw.sales", "raw.inventory"}
    # #endregion test_multiple_tables

    # #region test_empty_input [C:2] [TYPE Function]
    # @BRIEF Empty string returns empty set.
    def test_empty_input(self):
        assert extract_tables_from_sql("") == set()
        assert extract_tables_from_sql("   ") == set()
        assert extract_tables_from_sql(None) == set()
    # #endregion test_empty_input

    # #region test_no_schema_qualified_tables [C:2] [TYPE Function]
    # @BRIEF Unqualified table names are NOT matched.
    def test_no_schema_qualified_tables(self):
        sql = "SELECT * FROM sales"
        result = extract_tables_from_sql(sql)
        assert result == set()
    # #endregion test_no_schema_qualified_tables

    # #region test_case_insensitive [C:2] [TYPE Function]
    # @BRIEF Matching is case-insensitive; result is lowercased.
    def test_case_insensitive(self):
        sql = "SELECT * FROM RAW.Sales JOIN Raw.Inventory ON ..."
        result = extract_tables_from_sql(sql)
        assert result == {"raw.sales", "raw.inventory"}
    # #endregion test_case_insensitive

    # #region test_string_literal_false_positive [C:2] [TYPE Function]
    # @BRIEF Date literal like '2026.04.30' must NOT be matched as schema.table.
    def test_string_literal_false_positive(self):
        sql = "SELECT * FROM raw.sales WHERE date = '2026.04.30'"
        result = extract_tables_from_sql(sql)
        assert result == {"raw.sales"}
    # #endregion test_string_literal_false_positive

    # #region test_jinja_string_concat [C:2] [TYPE Function]
    # @BRIEF Table reference inside Jinja {{ 'raw.sales' }} as a direct string.
    def test_jinja_string_concat(self):
        sql = "SELECT * FROM {{ 'raw.sales' }}"
        result = extract_tables_from_sql(sql)
        assert result == {"raw.sales"}
    # #endregion test_jinja_string_concat

    # #region test_jinja_set_block [C:2] [TYPE Function]
    # @BRIEF Table name in Jinja {% set %} block string value.
    def test_jinja_set_block(self):
        sql = """
        {% set tables = ['raw.sales', 'raw.inventory'] %}
        SELECT * FROM raw.sales
        """
        result = extract_tables_from_sql(sql)
        assert result == {"raw.sales", "raw.inventory"}
    # #endregion test_jinja_set_block

    # #region test_mixed_jinja_and_sql [C:2] [TYPE Function]
    # @BRIEF Mixed Jinja string and plain SQL schema.table references.
    def test_mixed_jinja_and_sql(self):
        sql = """
        {% set source_table = 'raw.orders' %}
        SELECT * FROM raw.orders o
        JOIN raw.products p ON o.id = p.product_id
        """
        result = extract_tables_from_sql(sql)
        assert result == {"raw.orders", "raw.products"}
    # #endregion test_mixed_jinja_and_sql

    # #region test_cte_with_schema_tables [C:2] [TYPE Function]
    # @BRIEF CTE body with schema-qualified table references.
    def test_cte_with_schema_tables(self):
        sql = """
        WITH active_sales AS (
            SELECT * FROM raw.sales WHERE amount > 100
        )
        SELECT * FROM active_sales
        """
        result = extract_tables_from_sql(sql)
        assert result == {"raw.sales"}
    # #endregion test_cte_with_schema_tables

    # #region test_jinja_comment_excluded [C:2] [TYPE Function]
    # @BRIEF Jinja comment {# ... #} content should not produce table extraction (no string patterns inside comments).
    def test_jinja_comment_excluded(self):
        sql = "SELECT * FROM raw.sales {# this references raw.archived #}"
        result = extract_tables_from_sql(sql)
        assert result == {"raw.sales"}
    # #endregion test_jinja_comment_excluded


# ── Direct function tests for uncovered lines ──

class TestDetectJinjaSpans:
    """Direct tests for detect_jinja_spans."""

    def test_none_input(self):
        """Line 57: None input returns sql span."""
        from src.services.sql_table_extractor import detect_jinja_spans
        result = detect_jinja_spans(None)
        assert result == [("sql", "")]

    def test_empty_input(self):
        """Line 57: empty input returns sql span."""
        from src.services.sql_table_extractor import detect_jinja_spans
        result = detect_jinja_spans("")
        assert result == [("sql", "")]

    def test_whitespace_input(self):
        """Line 57: whitespace-only input returns sql span."""
        from src.services.sql_table_extractor import detect_jinja_spans
        result = detect_jinja_spans("   ")
        # detect_jinja_spans preserves the input text including whitespace
        assert result == [("sql", "   ")]

    def test_overlapping_jinja_spans(self):
        """Lines 71-74: adjacent Jinja expressions are merged."""
        from src.services.sql_table_extractor import detect_jinja_spans
        # Two Jinja expressions with no gap between them
        sql = "{{ source('raw', 'sales') }}{{ ref('inventory') }}"
        result = detect_jinja_spans(sql)
        assert len(result) == 1
        assert result[0][0] == "jinja"

    def test_jinja_adjacent_with_sql_gap(self):
        """Jinja spans with SQL content between them are separate."""
        from src.services.sql_table_extractor import detect_jinja_spans
        sql = "{{ source('raw', 'sales') }} SELECT * FROM raw.orders {{ ref('inventory') }}"
        result = detect_jinja_spans(sql)
        types = [span[0] for span in result]
        assert types == ["jinja", "sql", "jinja"]


class TestExtractTablesFromSqlSpan:
    """Direct tests for extract_tables_from_sql_span."""

    def test_single_char_schema_filtered(self):
        """Line 185: single-char schema like 'a.id' is filtered as column alias."""
        from src.services.sql_table_extractor import extract_tables_from_sql_span
        result = extract_tables_from_sql_span("SELECT a.id FROM raw.sales")
        assert "a.id" not in result
        assert "raw.sales" in result

    def test_no_raw_matches_returns_empty(self):
        """Line 146: no regex matches returns empty set."""
        from src.services.sql_table_extractor import extract_tables_from_sql_span
        result = extract_tables_from_sql_span("SELECT * FROM sales")
        assert result == set()

    def test_stmt_none_skipped(self):
        """Line 169: stmt is None in parsed sqlparse is skipped."""
        from src.services.sql_table_extractor import extract_tables_from_sql_span
        # sqlparse may return None for certain edge inputs
        result = extract_tables_from_sql_span("SELECT * FROM raw.sales")
        assert "raw.sales" in result


class TestExtractTablesFromJinja:
    """Direct tests for extract_tables_from_jinja."""

    def test_no_matches_returns_empty(self):
        from src.services.sql_table_extractor import extract_tables_from_jinja
        result = extract_tables_from_jinja("{{ some_function() }}")
        assert result == set()

    def test_extracts_quoted_table_refs(self):
        from src.services.sql_table_extractor import extract_tables_from_jinja
        result = extract_tables_from_jinja('{{ source("raw", "sales") }}')
        # "raw" alone doesn't have schema.table pattern — no dot
        # But we test the general case
        assert isinstance(result, set)


class TestIsStringLiteral:
    """Direct tests for is_string_literal."""

    def test_regular_token_not_literal(self):
        from src.services.sql_table_extractor import is_string_literal
        import sqlparse
        parsed = sqlparse.parse("SELECT")[0]
        token = parsed.tokens[0]
        assert is_string_literal(token) is False

    def test_single_quoted_string_is_literal(self):
        """Single-quoted string → is_string_literal returns True."""
        from src.services.sql_table_extractor import is_string_literal
        import sqlparse
        parsed = sqlparse.parse("SELECT 'hello'")[0]
        for token in parsed.flatten():
            if token.value == "'hello'":
                assert is_string_literal(token) is True
                return
        # Fallback: find by ttype
        from sqlparse.tokens import Literal
        for token in parsed.flatten():
            if token.ttype is Literal.String.Single:
                assert is_string_literal(token) is True
                return


class TestSqlSpanEdgeCases:
    """Edge coverage for extract_tables_from_sql_span — string literal filtering, dead code paths."""

    def test_schema_table_inside_string_literal_filtered(self):
        """schema.table inside a string literal → filtered out by is_in_string (line 175)."""
        from src.services.sql_table_extractor import extract_tables_from_sql
        # 'prefix raw.sales suffix' — raw.sales inside a string literal
        # The regex matches raw.sales, but is_in_string filters it out
        sql = "SELECT * FROM raw.sales WHERE name = 'prefix raw.sales suffix'"
        result = extract_tables_from_sql(sql)
        # Only the first raw.sales (outside string) should be in result
        assert "raw.sales" in result

    def test_subquery_with_token_list_flatten_collects_string_ranges(self):
        """Function call or subquery exercises TokenList flatten in walk_tokens (line 157)."""
        from src.services.sql_table_extractor import extract_tables_from_sql
        # Using a SQL with a function call that creates TokenList structure
        sql = "SELECT * FROM raw.sales WHERE date > COALESCE('2024-01-01', '2024-12-31')"
        result = extract_tables_from_sql(sql)
        assert "raw.sales" in result
        assert "coalesce.sales" not in result  # inside function

    def test_sqlparse_parse_returns_none_skipped(self):
        """sqlparse.parse returns None element → continue (line 169)."""
        from src.services.sql_table_extractor import extract_tables_from_sql_span
        import sqlparse
        # Normal SQL — sqlparse returns Statement, never None
        # This tests robustness: if parse returned None, the continue would skip it
        sql = "SELECT * FROM raw.sales"
        # sqlparse.parse always returns a list of Statements (never None elements)
        result = extract_tables_from_sql_span(sql)
        assert "raw.sales" in result


class TestExtractTablesFromJinjaEdge:
    """Edge coverage for extract_tables_from_jinja."""

    def test_no_matches_returns_empty(self):
        """Jinja text without schema.table patterns → empty set."""
        from src.services.sql_table_extractor import extract_tables_from_jinja
        result = extract_tables_from_jinja("{{ 'hello world' }}")
        assert result == set()
# #endregion test_sql_table_extractor