79 lines
2.5 KiB
Python
79 lines
2.5 KiB
Python
from contextlib import contextmanager
|
|
from unittest.mock import MagicMock
|
|
|
|
import pytest
|
|
|
|
from core import postgres
|
|
|
|
|
|
def test_safe_identifier_quotes_names_and_rejects_sql_expressions():
|
|
assert postgres._safe_id("scheduled_jobs") == '"scheduled_jobs"'
|
|
|
|
for unsafe in ["jobs.id", "jobs; DROP TABLE jobs", "two words", "", 7, None]:
|
|
with pytest.raises(ValueError, match="Invalid SQL identifier"):
|
|
postgres._safe_id(unsafe)
|
|
|
|
|
|
def test_order_clause_allows_only_identifiers_and_directions():
|
|
assert postgres._order_clause(
|
|
["created_at desc", ("id", "ASC")]
|
|
) == '"created_at" DESC, "id" ASC'
|
|
|
|
unsafe_values = [
|
|
"created_at DESC NULLS LAST",
|
|
"created_at;drop DESC",
|
|
[("created_at", "SIDEWAYS")],
|
|
[("created_at", "ASC", "extra")],
|
|
]
|
|
for value in unsafe_values:
|
|
with pytest.raises(ValueError):
|
|
postgres._order_clause(value)
|
|
|
|
|
|
def test_select_builds_parameterized_where_and_safe_order(monkeypatch):
|
|
cursor = MagicMock()
|
|
cursor.fetchall.return_value = [{"id": "job-1"}]
|
|
|
|
@contextmanager
|
|
def fake_cursor():
|
|
yield cursor
|
|
|
|
monkeypatch.setattr(postgres, "get_cursor", fake_cursor)
|
|
|
|
rows = postgres.select(
|
|
"scheduled_jobs",
|
|
where={"user_uuid": "user-1", "status": ("IN", ["pending", "running"])},
|
|
order_by=[("run_at", "ASC"), ("id", "DESC")],
|
|
limit=10,
|
|
)
|
|
|
|
assert rows == [{"id": "job-1"}]
|
|
query, params = cursor.execute.call_args.args
|
|
assert 'FROM "scheduled_jobs"' in query
|
|
assert '"user_uuid" = %(user_uuid_0)s' in query
|
|
assert '"status" IN (%(status_1_0)s, %(status_1_1)s)' in query
|
|
assert 'ORDER BY "run_at" ASC, "id" DESC' in query
|
|
assert "LIMIT %(query_limit)s" in query
|
|
assert params == {
|
|
"user_uuid_0": "user-1",
|
|
"status_1_0": "pending",
|
|
"status_1_1": "running",
|
|
"query_limit": 10,
|
|
}
|
|
|
|
|
|
def test_empty_update_and_delete_conditions_fail_before_opening_cursor(monkeypatch):
|
|
cursor_factory = MagicMock(
|
|
side_effect=AssertionError("a database cursor must not be opened")
|
|
)
|
|
monkeypatch.setattr(postgres, "get_cursor", cursor_factory)
|
|
|
|
with pytest.raises(ValueError, match="update data cannot be empty"):
|
|
postgres.update("users", {}, {"id": "user-1"})
|
|
with pytest.raises(ValueError, match="non-empty where"):
|
|
postgres.update("users", {"timezone": "UTC"}, {})
|
|
with pytest.raises(ValueError, match="non-empty where"):
|
|
postgres.delete("users", {})
|
|
|
|
cursor_factory.assert_not_called()
|