Jan Doubravský
116 lines
3.6 KiB
Python
116 lines
3.6 KiB
Python
import pytest
|
|
|
|
from sqlmem.exceptions import ReadOnlyError, UnsupportedQueryError
|
|
from sqlmem.parser import parse
|
|
|
|
|
|
def test_simple_select():
|
|
result = parse("SELECT name, email FROM users WHERE status = 'active'")
|
|
assert result.tables == ["users"]
|
|
cols = result.columns_by_table["users"]
|
|
# WHERE columns are also extracted — needed for in-memory SQLite filtering
|
|
assert {"name", "email"}.issubset(set(cols))
|
|
assert "status" in cols
|
|
|
|
|
|
def test_multiple_columns():
|
|
result = parse("SELECT a, b, c FROM orders")
|
|
assert result.tables == ["orders"]
|
|
assert set(result.columns_by_table["orders"]) == {"a", "b", "c"}
|
|
|
|
|
|
def test_columns_deduplicated_in_order():
|
|
result = parse("SELECT a, a, b FROM t WHERE a > 1")
|
|
assert result.columns_by_table["t"] == ["a", "b"]
|
|
|
|
|
|
def test_insert_raises_readonly():
|
|
with pytest.raises(ReadOnlyError):
|
|
parse("INSERT INTO users (name) VALUES ('alice')")
|
|
|
|
|
|
def test_update_raises_readonly():
|
|
with pytest.raises(ReadOnlyError):
|
|
parse("UPDATE users SET name = 'bob' WHERE id = 1")
|
|
|
|
|
|
def test_delete_raises_readonly():
|
|
with pytest.raises(ReadOnlyError):
|
|
parse("DELETE FROM users WHERE id = 1")
|
|
|
|
|
|
def test_select_without_from_raises():
|
|
with pytest.raises(UnsupportedQueryError):
|
|
parse("SELECT 1")
|
|
|
|
|
|
# --- R1: parameters ---------------------------------------------------------
|
|
|
|
|
|
def test_params_stored():
|
|
result = parse("SELECT name FROM users WHERE id = ?", ("7189790",))
|
|
assert result.params == ("7189790",)
|
|
assert "?" in result.sqlite_sql
|
|
|
|
|
|
def test_named_params_preserved():
|
|
result = parse("SELECT name FROM users WHERE id = :id", {"id": 1})
|
|
assert ":id" in result.sqlite_sql
|
|
|
|
|
|
# --- R2: JOINs --------------------------------------------------------------
|
|
|
|
|
|
def test_join_extracts_all_tables():
|
|
result = parse(
|
|
"SELECT a.id, b.title FROM users a "
|
|
"JOIN orders b ON a.id = b.user_id WHERE a.id = ?",
|
|
(1,),
|
|
)
|
|
assert set(result.tables) == {"users", "orders"}
|
|
assert "id" in result.columns_by_table["users"]
|
|
assert "title" in result.columns_by_table["orders"]
|
|
# join + where columns resolved to their tables via alias
|
|
assert "user_id" in result.columns_by_table["orders"]
|
|
|
|
|
|
def test_join_unqualified_column_is_ambiguous():
|
|
with pytest.raises(UnsupportedQueryError):
|
|
parse("SELECT name FROM users a JOIN orders b ON a.id = b.user_id")
|
|
|
|
|
|
# --- R3: SELECT * -----------------------------------------------------------
|
|
|
|
|
|
def test_wildcard_marks_table_full():
|
|
result = parse("SELECT * FROM users")
|
|
assert result.wildcard_tables == {"users"}
|
|
assert result.columns_by_table == {}
|
|
|
|
|
|
def test_qualified_wildcard_marks_only_that_table():
|
|
result = parse(
|
|
"SELECT u.*, o.total FROM users u JOIN orders o ON u.id = o.user_id"
|
|
)
|
|
assert "users" in result.wildcard_tables
|
|
assert "orders" not in result.wildcard_tables
|
|
assert "total" in result.columns_by_table["orders"]
|
|
|
|
|
|
# --- R4: three-part names (MSSQL brackets) ----------------------------------
|
|
|
|
|
|
def test_three_part_name_uses_base_table():
|
|
result = parse(
|
|
"SELECT [PRODUCT_PRODUCTNR], [PRAT_NAME] "
|
|
"FROM [DP_PIM].[dbo].[VW_P_PRATVALUES] WHERE PRODUCT_PRODUCTNR = ?",
|
|
("7189790",),
|
|
)
|
|
assert result.tables == ["VW_P_PRATVALUES"]
|
|
cols = result.columns_by_table["VW_P_PRATVALUES"]
|
|
assert {"PRODUCT_PRODUCTNR", "PRAT_NAME"}.issubset(set(cols))
|
|
# in-memory SQL must drop the catalog/schema prefix
|
|
assert "DP_PIM" not in result.sqlite_sql
|
|
assert "dbo" not in result.sqlite_sql
|
|
assert "VW_P_PRATVALUES" in result.sqlite_sql
|