Bertrand Benjamin
d42bfecab4
- Config ruff dans pyproject.toml : règles E/W/F/I/UP/B, whitelist des appels d'injection FastAPI (Depends/File/Form/Query) pour B008, B904 ignoré (traduction volontaire des exceptions en réponses HTTP) - Auto-fixes : tri des imports, suppression d'imports inutilisés, annotations PEP 604, f-strings sans placeholder, modes open redondants - Suppression de variables inutilisées (config.reset_setting, parser locataires) ruff check . : All checks passed ; 61 tests OK Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
94 lines
2.7 KiB
Python
94 lines
2.7 KiB
Python
"""Tests de l'exécution SQL read-only de l'assistant IA."""
|
|
|
|
import sqlite3
|
|
|
|
import pytest
|
|
|
|
from plesna_gerance.services.sql_executor import (
|
|
_SQLITE_DENY,
|
|
_SQLITE_OK,
|
|
_SQLITE_PRAGMA,
|
|
_SQLITE_READ,
|
|
_SQLITE_SELECT,
|
|
MAX_ROWS,
|
|
_authorizer,
|
|
_ensure_limit,
|
|
_validate_sql,
|
|
execute_readonly_sql,
|
|
)
|
|
|
|
# --- Garde lexicale -------------------------------------------------------
|
|
|
|
|
|
def test_validate_rejects_non_select():
|
|
with pytest.raises(ValueError):
|
|
_validate_sql("DELETE FROM tags")
|
|
|
|
|
|
def test_validate_rejects_stacked_statements():
|
|
with pytest.raises(ValueError):
|
|
_validate_sql("SELECT 1; DROP TABLE tags")
|
|
|
|
|
|
def test_validate_rejects_unknown_pragma():
|
|
with pytest.raises(ValueError):
|
|
_validate_sql("PRAGMA writable_schema = ON")
|
|
|
|
|
|
def test_validate_allows_select_and_with():
|
|
_validate_sql("SELECT * FROM tags")
|
|
_validate_sql("WITH t AS (SELECT 1) SELECT * FROM t")
|
|
_validate_sql("PRAGMA table_info(tags)")
|
|
|
|
|
|
def test_ensure_limit_adds_limit():
|
|
assert _ensure_limit("SELECT * FROM tags").endswith(f"LIMIT {MAX_ROWS}")
|
|
|
|
|
|
def test_ensure_limit_preserves_existing():
|
|
q = "SELECT * FROM tags LIMIT 5"
|
|
assert _ensure_limit(q) == q
|
|
|
|
|
|
# --- Autorisation SQLite (unitaire) ---------------------------------------
|
|
|
|
|
|
def test_authorizer_allows_reads():
|
|
assert _authorizer(_SQLITE_SELECT, None, None, None, None) == _SQLITE_OK
|
|
assert _authorizer(_SQLITE_READ, "tags", "nom", "main", None) == _SQLITE_OK
|
|
|
|
|
|
def test_authorizer_allows_whitelisted_pragma():
|
|
assert _authorizer(_SQLITE_PRAGMA, "table_info", "tags", None, None) == _SQLITE_OK
|
|
|
|
|
|
def test_authorizer_denies_unknown_pragma():
|
|
assert _authorizer(_SQLITE_PRAGMA, "writable_schema", "ON", None, None) == _SQLITE_DENY
|
|
|
|
|
|
def test_authorizer_denies_unknown_action():
|
|
# 9 = SQLITE_DELETE, doit être refusé
|
|
assert _authorizer(9, "tags", None, "main", None) == _SQLITE_DENY
|
|
|
|
|
|
# --- Exécution réelle (nécessite une base) --------------------------------
|
|
|
|
|
|
def test_execute_select_returns_rows(db_session):
|
|
# init_db seed des tags prédéfinis
|
|
result = execute_readonly_sql("SELECT nom FROM tags ORDER BY nom")
|
|
assert "nom" in result["columns"]
|
|
assert result["row_count"] >= 1
|
|
|
|
|
|
def test_execute_write_blocked(db_session):
|
|
# Bloqué par la garde lexicale (ValueError) ou, à défaut, par mode=ro /
|
|
# l'autorisation au niveau SQLite (DatabaseError). Dans tous les cas : refusé.
|
|
with pytest.raises((ValueError, sqlite3.DatabaseError)):
|
|
execute_readonly_sql("DELETE FROM tags")
|
|
|
|
|
|
def test_execute_attach_blocked(db_session):
|
|
with pytest.raises((ValueError, sqlite3.DatabaseError)):
|
|
execute_readonly_sql("ATTACH DATABASE 'x.db' AS x")
|