File size: 1,771 Bytes
98aa813
df5b35d
98aa813
 
df5b35d
 
 
 
 
 
 
 
 
 
 
 
 
 
98aa813
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
df5b35d
 
 
 
 
98aa813
 
 
 
 
 
 
 
 
 
 
 
df5b35d
 
 
98aa813
 
 
df5b35d
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
"""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