mirror of
https://github.com/apache/superset.git
synced 2026-04-10 11:55:24 +00:00
feat: improve _extract_tables_from_sql (#26748)
This commit is contained in:
@@ -25,7 +25,10 @@ from sqlalchemy import text
|
||||
from sqlparse.sql import Identifier, Token, TokenList
|
||||
from sqlparse.tokens import Name
|
||||
|
||||
from superset.exceptions import QueryClauseValidationException
|
||||
from superset.exceptions import (
|
||||
QueryClauseValidationException,
|
||||
SupersetSecurityException,
|
||||
)
|
||||
from superset.sql_parse import (
|
||||
add_table_name,
|
||||
extract_table_references,
|
||||
@@ -267,13 +270,34 @@ def test_extract_tables_illdefined() -> None:
|
||||
"""
|
||||
Test that ill-defined tables return an empty set.
|
||||
"""
|
||||
assert extract_tables("SELECT * FROM schemaname.") == set()
|
||||
assert extract_tables("SELECT * FROM catalogname.schemaname.") == set()
|
||||
assert extract_tables("SELECT * FROM catalogname..") == set()
|
||||
with pytest.raises(SupersetSecurityException) as excinfo:
|
||||
extract_tables("SELECT * FROM schemaname.")
|
||||
assert (
|
||||
str(excinfo.value) == "Unable to parse SQL (generic): SELECT * FROM schemaname."
|
||||
)
|
||||
|
||||
with pytest.raises(SupersetSecurityException) as excinfo:
|
||||
extract_tables("SELECT * FROM catalogname.schemaname.")
|
||||
assert (
|
||||
str(excinfo.value)
|
||||
== "Unable to parse SQL (generic): SELECT * FROM catalogname.schemaname."
|
||||
)
|
||||
|
||||
with pytest.raises(SupersetSecurityException) as excinfo:
|
||||
extract_tables("SELECT * FROM catalogname..")
|
||||
assert (
|
||||
str(excinfo.value)
|
||||
== "Unable to parse SQL (generic): SELECT * FROM catalogname.."
|
||||
)
|
||||
|
||||
with pytest.raises(SupersetSecurityException) as excinfo:
|
||||
extract_tables('SELECT * FROM "tbname')
|
||||
assert str(excinfo.value) == 'Unable to parse SQL (generic): SELECT * FROM "tbname'
|
||||
|
||||
# odd edge case that works
|
||||
assert extract_tables("SELECT * FROM catalogname..tbname") == {
|
||||
Table(table="tbname", schema=None, catalog="catalogname")
|
||||
}
|
||||
assert extract_tables('SELECT * FROM "tbname') == set()
|
||||
|
||||
|
||||
def test_extract_tables_show_tables_from() -> None:
|
||||
|
||||
Reference in New Issue
Block a user