| """Tests de sécurité SQL (validateur + exécuteur SQLite).""" |
| from app.crag.data_query import is_safe_query, run_structured_query |
| from app.sql import is_safe_sql, run_sql_query |
| from app.sql.executor import execute_readonly |
|
|
|
|
| def test_rejects_unknown_column_filter(): |
| assert not is_safe_query({"mot_de_passe": "x"}, "sum", None) |
|
|
|
|
| def test_rejects_unknown_aggregation(): |
| assert not is_safe_query({}, "delete", None) |
|
|
|
|
| def test_accepts_whitelisted_query(): |
| assert is_safe_query({"mois": "Juin"}, "sum", "client") |
|
|
|
|
| def test_is_safe_sql_blocks_mutations(): |
| assert is_safe_sql("SELECT client FROM sales LIMIT 10") |
| assert not is_safe_sql("DELETE FROM sales") |
| assert not is_safe_sql("DROP TABLE sales") |
| assert not is_safe_sql("SELECT * FROM sales; DROP TABLE sales") |
|
|
|
|
| def test_execute_readonly_rejects_unsafe(): |
| result = execute_readonly("DELETE FROM sales") |
| assert "error" in result |
|
|
|
|
| def test_run_sql_best_client_june(): |
| result = run_sql_query( |
| metric="chiffre_affaires", |
| aggregation="sum", |
| filters={"mois": "Juin"}, |
| group_by="client", |
| top_n=1, |
| ) |
| assert "result" in result |
| top_client, top_value = result["result"][0] |
| assert top_client == "Client F" |
| assert top_value == 18200 |
| assert "sql" in result |
|
|
|
|
| def test_run_structured_query_compat_alias(): |
| result = run_structured_query( |
| metric="chiffre_affaires", |
| aggregation="sum", |
| filters={"mois": "Juin"}, |
| group_by="client", |
| top_n=1, |
| ) |
| assert result["result"][0][0] == "Client F" |
|
|
|
|
| def test_run_structured_query_refuses_unsafe_params(): |
| result = run_structured_query( |
| metric="chiffre_affaires", aggregation="drop", filters={} |
| ) |
| assert "error" in result |
|
|
