Unit tests cover the SQL gate, broadcast codec, Hive writer layout, and flight kinematics. CI runs pytest first; the simulator image build runs tests in a dedicated stage before the final layer.
36 lines
1.2 KiB
Python
36 lines
1.2 KiB
Python
"""Read-only SQL gate — same rules as the peer-query forced command."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "explorer"))
|
|
|
|
from server import gate # noqa: E402
|
|
|
|
|
|
def test_allows_read_statements() -> None:
|
|
assert gate("SELECT 1") is None
|
|
assert gate("WITH t AS (SELECT 1) SELECT * FROM t") is None
|
|
assert gate("DESCRIBE telemetry") is None
|
|
assert gate("SUMMARIZE state") is None
|
|
assert gate("SHOW TABLES") is None
|
|
|
|
|
|
def test_rejects_writes_and_config() -> None:
|
|
assert gate("DROP TABLE telemetry") is not None
|
|
assert gate("INSERT INTO t VALUES (1)") is not None
|
|
assert gate("UPDATE t SET x=1") is not None
|
|
assert gate("DELETE FROM t") is not None
|
|
assert gate("CREATE TABLE t (x INT)") is not None
|
|
assert gate("INSTALL httpfs") is not None
|
|
assert gate("SET memory_limit='1GB'") is not None
|
|
assert gate("PRAGMA threads=4") is not None
|
|
|
|
|
|
def test_rejects_multi_statement_and_injection() -> None:
|
|
assert gate("SELECT 1; SELECT 2") is not None
|
|
assert gate("/* sneaky */ COPY t TO 'x'") is not None
|
|
assert gate("") is not None
|