73 lines
2.5 KiB
Python
73 lines
2.5 KiB
Python
import pytest
|
|
|
|
from app.core.contracts import RequestContext
|
|
from app.core.nl2sql_contracts import FinancialNL2SQLInput
|
|
from app.service.financial_nl2sql_service import FinancialNL2SQLService
|
|
|
|
|
|
def context(**updates):
|
|
base = RequestContext(
|
|
user_id="1",
|
|
trace_id="trace-nl2sql",
|
|
roles=("advisor",),
|
|
permissions=("financial:nl2sql:read",),
|
|
data_scope="all",
|
|
)
|
|
return base.model_copy(update=updates)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_generates_read_only_market_sql_with_product_filter() -> None:
|
|
result = await FinancialNL2SQLService().query(
|
|
FinancialNL2SQLInput(question="查询159511近30天行情收盘价", dry_run=True),
|
|
context(),
|
|
)
|
|
assert result["status"] == "ready"
|
|
assert result["sql"].startswith("SELECT ")
|
|
assert "fin_market_price" in result["sql"]
|
|
assert "DROP" not in result["sql"]
|
|
assert result["parameters"]["filter_0"] == "159511"
|
|
assert result["audit"]["permission_check"]["status"] == "passed"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_low_confidence_question_requires_confirmation() -> None:
|
|
result = await FinancialNL2SQLService().query(
|
|
FinancialNL2SQLInput(question="帮我看看这个情况", dry_run=True),
|
|
context(),
|
|
)
|
|
assert result["status"] == "need_confirmation"
|
|
assert result["query_plan"]["confidence"] < 0.85
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_as_of_query_rejects_current_snapshot_tables() -> None:
|
|
result = await FinancialNL2SQLService().query(
|
|
FinancialNL2SQLInput(question="截至某日客户当前持仓市值", dry_run=True),
|
|
context(),
|
|
)
|
|
assert result["status"] == "rejected"
|
|
assert "历史版本" in result["message"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_customer_scope_is_injected_for_non_all_scope() -> None:
|
|
result = await FinancialNL2SQLService().query(
|
|
FinancialNL2SQLInput(question="查询客户账户余额", dry_run=True),
|
|
context(data_scope="own_customers", customer_ids=("7", "8")),
|
|
)
|
|
assert "customer_id IN" in result["sql"]
|
|
assert result["parameters"]["scope_customer_0"] == 7
|
|
assert result["parameters"]["scope_customer_1"] == 8
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cash_ledger_is_in_nl2sql_scope() -> None:
|
|
result = await FinancialNL2SQLService().query(
|
|
FinancialNL2SQLInput(question="查询近7天资金流水变化", dry_run=True),
|
|
context(),
|
|
)
|
|
assert result["status"] == "ready"
|
|
assert "fin_cash_ledger" in result["sql"]
|
|
assert result["query_plan"]["intent"] == "cash_ledger_query"
|