busya
ce0369ae5c
Subagents delivered tests across all uncovered backend modules: Schemas (100%): agent, auth, health, profile, settings, validation Services (98-100%): auth, profile, health, llm, mapping, resource, security, git, superset_lookup, sql_table_extractor, rbac API routes (new): auth, admin, health, environments, plugins, dashboards (helpers, projection, actions, listing), git (config, deps, env, helpers) Clean Release (100%): DTO, facade, policy_engine, stages, repos, preparation, source_isolation, compliance Git services: base, remote_providers Agent module: app, run, middleware, langgraph_setup Core: trace, cleanup, ws_log_handler, timezone, auth (config/oauth/security), matching Reports: normalizer, report_service, type_profiles Notifications: service, providers Also: - .gitignore: add .coverage, *.cover, coverage-* dirs - src/schemas/auth.py: fix AD group DN regex (comma in CN=...) - Remove co-located src/services/__tests__/ (caused pytest module collision)
210 lines
8.8 KiB
Python
210 lines
8.8 KiB
Python
# #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(" ")
|
|
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
|
|
# #endregion test_sql_table_extractor
|