busya
4d282b43e2
sqlparse raises SQLParseError above MAX_GROUPING_TOKENS=10000 tokens (~25KB of typical SQL). The try/except fallback already handled it, but paid ~1s per oversized SQL for a parse doomed to fail. Add _SQLPARSE_SKIP_THRESHOLD (30k chars) to bypass sqlparse for oversized text (~15x faster, 1.2s->0.08s for a 212KB SQL) while keeping literal filtering for SQL under the threshold. Tests: oversized-SQL skip-threshold behavior.
305 lines
14 KiB
Python
305 lines
14 KiB
Python
# #region Test.SqlTableExtractor.TestSqlTableExtractor [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 -> [Services.SqlTableExtractor.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.SqlTableExtractor.TestSimpleSelect [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.SqlTableExtractor.TestSimpleSelect
|
|
|
|
# #region Test.SqlTableExtractor.TestMultipleTables [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.SqlTableExtractor.TestMultipleTables
|
|
|
|
# #region Test.SqlTableExtractor.TestEmptyInput [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.SqlTableExtractor.TestEmptyInput
|
|
|
|
# #region Test.SqlTableExtractor.TestNoSchemaQualifiedTables [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.SqlTableExtractor.TestNoSchemaQualifiedTables
|
|
|
|
# #region Test.SqlTableExtractor.TestCaseInsensitive [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.SqlTableExtractor.TestCaseInsensitive
|
|
|
|
# #region Test.SqlTableExtractor.TestStringLiteralFalsePositive [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.SqlTableExtractor.TestStringLiteralFalsePositive
|
|
|
|
# #region Test.SqlTableExtractor.TestJinjaStringConcat [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.SqlTableExtractor.TestJinjaStringConcat
|
|
|
|
# #region Test.SqlTableExtractor.TestJinjaSetBlock [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.SqlTableExtractor.TestJinjaSetBlock
|
|
|
|
# #region Test.SqlTableExtractor.TestMixedJinjaAndSql [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.SqlTableExtractor.TestMixedJinjaAndSql
|
|
|
|
# #region Test.SqlTableExtractor.TestCteWithSchemaTables [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.SqlTableExtractor.TestCteWithSchemaTables
|
|
|
|
# #region Test.SqlTableExtractor.TestJinjaCommentExcluded [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.SqlTableExtractor.TestJinjaCommentExcluded
|
|
|
|
|
|
# ── 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
|
|
|
|
def test_huge_sql_falls_back_to_regex_instead_of_raising(self):
|
|
"""SQL exceeding sqlparse's token limit (MAX_GROUPING_TOKENS=10000) must not raise.
|
|
|
|
Production virtual datasets can be enormous; before this fix a SQLParseError
|
|
("Maximum number of tokens exceeded (10000)") aborted the whole maintenance scan.
|
|
It now falls back to regex-only extraction and still finds the target table.
|
|
"""
|
|
from src.services.sql_table_extractor import extract_tables_from_sql
|
|
sql = (
|
|
"SELECT "
|
|
+ " + ".join(f"c{i}" for i in range(15000))
|
|
+ " FROM dm_view.counterparty_td"
|
|
)
|
|
result = extract_tables_from_sql(sql)
|
|
assert "dm_view.counterparty_td" in result
|
|
|
|
def test_oversized_sql_skips_sqlparse_and_keeps_regex_matches(self):
|
|
"""SQL above the skip threshold bypasses sqlparse entirely (fast, no crash).
|
|
|
|
Oversized text is matched with regex only; a real FROM table is still found.
|
|
"""
|
|
from src.services.sql_table_extractor import (
|
|
_SQLPARSE_SKIP_THRESHOLD,
|
|
extract_tables_from_sql,
|
|
)
|
|
sql = (
|
|
"SELECT * FROM raw.sales WHERE x IN ("
|
|
+ ", ".join(f"c{i}" for i in range(12000))
|
|
+ ")"
|
|
)
|
|
assert len(sql) > _SQLPARSE_SKIP_THRESHOLD
|
|
result = extract_tables_from_sql(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.SqlTableExtractor.TestSqlTableExtractor
|