34d2ccc576
Build spec and host brief carried in from C:\Claude and WRPS/02-env; the plant model (equipment, tags, alarm bitmask, enums, unit conversions) is derived from WRPS/04-plc/register-map.csv, WRPS/05-scada/modbus/scada-points.csv and WRPS-CTL-003. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
90 lines
2.6 KiB
Python
90 lines
2.6 KiB
Python
"""SQL allow-list and Cube query caps.
|
|
|
|
The Phase 8 gate demands zero SQL executed outside the allow-list. That is a
|
|
property of this module, so it is tested here rather than inferred from the
|
|
eval run.
|
|
"""
|
|
|
|
import pytest
|
|
|
|
from guardrails import GuardrailViolation, check_cube_query, check_sql
|
|
|
|
MAX_ROWS = 5000
|
|
|
|
|
|
def test_plain_select_is_allowed_and_capped():
|
|
out = check_sql("SELECT tag_id FROM tags", max_rows=MAX_ROWS)
|
|
assert "LIMIT 5000" in out.upper()
|
|
|
|
|
|
def test_smaller_limit_is_honoured():
|
|
out = check_sql("SELECT tag_id FROM tags LIMIT 10", max_rows=MAX_ROWS)
|
|
assert "LIMIT 10" in out.upper()
|
|
|
|
|
|
def test_larger_limit_is_capped():
|
|
out = check_sql("SELECT tag_id FROM tags LIMIT 999999", max_rows=MAX_ROWS)
|
|
assert "LIMIT 5000" in out.upper()
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"sql",
|
|
[
|
|
"INSERT INTO equipment VALUES ('X')",
|
|
"UPDATE tags SET display_name = 'x'",
|
|
"DELETE FROM doc_chunks",
|
|
"DROP TABLE tags",
|
|
"CREATE TABLE t (i int)",
|
|
],
|
|
)
|
|
def test_writes_are_refused(sql):
|
|
with pytest.raises(GuardrailViolation):
|
|
check_sql(sql, max_rows=MAX_ROWS)
|
|
|
|
|
|
def test_stacked_statements_are_refused():
|
|
with pytest.raises(GuardrailViolation) as caught:
|
|
check_sql("SELECT 1 FROM tags; DROP TABLE tags", max_rows=MAX_ROWS)
|
|
assert caught.value.rule in {"multiple_statements", "write_operation"}
|
|
|
|
|
|
def test_table_outside_the_allow_list_is_refused():
|
|
with pytest.raises(GuardrailViolation) as caught:
|
|
check_sql("SELECT * FROM pg_shadow", max_rows=MAX_ROWS)
|
|
assert caught.value.rule == "table_not_allowed"
|
|
|
|
|
|
def test_subquery_tables_are_checked_too():
|
|
with pytest.raises(GuardrailViolation):
|
|
check_sql(
|
|
"SELECT * FROM tags WHERE tag_id IN (SELECT usename FROM pg_shadow)",
|
|
max_rows=MAX_ROWS,
|
|
)
|
|
|
|
|
|
def test_cube_query_needs_a_pinned_window():
|
|
with pytest.raises(GuardrailViolation) as caught:
|
|
check_cube_query({"measures": ["alarm_activity.alarm_count"]}, max_rows=MAX_ROWS)
|
|
assert caught.value.rule == "unpinned_time_window"
|
|
|
|
|
|
def test_cube_query_needs_a_date_range():
|
|
with pytest.raises(GuardrailViolation):
|
|
check_cube_query(
|
|
{"timeDimensions": [{"dimension": "alarm_activity.event_time"}]},
|
|
max_rows=MAX_ROWS,
|
|
)
|
|
|
|
|
|
def test_cube_query_is_capped():
|
|
out = check_cube_query(
|
|
{
|
|
"timeDimensions": [
|
|
{"dimension": "alarm_activity.event_time",
|
|
"dateRange": ["2026-08-13", "2026-08-20"]}
|
|
],
|
|
"limit": 1_000_000,
|
|
},
|
|
max_rows=MAX_ROWS,
|
|
)
|
|
assert out["limit"] == MAX_ROWS
|