81 lines
2.8 KiB
Python
81 lines
2.8 KiB
Python
from sqlalchemy import create_engine, text
|
|
|
|
from app.core.contracts import RequestContext
|
|
from app.service.offsite_nl2sql_adapter import OffsiteNl2SqlAdapter
|
|
|
|
|
|
def test_offsite_adapter_executes_query_dict_with_read_only_engine() -> None:
|
|
engine = create_engine("sqlite+pysqlite:///:memory:")
|
|
with engine.begin() as connection:
|
|
connection.execute(text(
|
|
"CREATE TABLE fin_product (id INTEGER PRIMARY KEY, product_code TEXT)"
|
|
))
|
|
connection.execute(text(
|
|
"CREATE TABLE fin_market_price "
|
|
"(id INTEGER PRIMARY KEY, product_id INTEGER, total_fund_shares NUMERIC)"
|
|
))
|
|
connection.execute(text(
|
|
"CREATE TABLE fin_nav_history "
|
|
"(id INTEGER PRIMARY KEY, product_id INTEGER, nav NUMERIC, nav_date TEXT)"
|
|
))
|
|
connection.execute(text(
|
|
"CREATE TABLE fin_holding "
|
|
"(id INTEGER PRIMARY KEY, product_id INTEGER, trade_account TEXT, "
|
|
"total_quantity NUMERIC, shares NUMERIC, available_quantity NUMERIC)"
|
|
))
|
|
connection.execute(text(
|
|
"INSERT INTO fin_product VALUES (1, '15911')"
|
|
))
|
|
connection.execute(text(
|
|
"INSERT INTO fin_market_price VALUES (1, 1, 1000000)"
|
|
))
|
|
connection.execute(text(
|
|
"INSERT INTO fin_nav_history VALUES (1, 1, 1.25, '2026-09-10')"
|
|
))
|
|
connection.execute(text(
|
|
"INSERT INTO fin_holding VALUES (1, 1, '10001', 1000, 1000, 900)"
|
|
))
|
|
|
|
context = RequestContext(
|
|
user_id="1",
|
|
trace_id="offsite-nl2sql-unit",
|
|
roles=("operator",),
|
|
permissions=("offsite:write",),
|
|
data_scope="all",
|
|
)
|
|
result = OffsiteNl2SqlAdapter(db_engine=engine).query(
|
|
"基金代码为15911，账户标识为10001，查询基金最新总份额、最新净值和申请前持有份额",
|
|
context,
|
|
)
|
|
|
|
assert result["status"] == "success"
|
|
assert result["data"] == {
|
|
"total": 1,
|
|
"rows": [{
|
|
"total_fund_shares": 1000000,
|
|
"nav": 1.25,
|
|
"total_quantity": 1000,
|
|
}],
|
|
}
|
|
assert ":filter_0" in str(result["sql"])
|
|
assert ":filter_1" in str(result["sql"])
|
|
|
|
|
|
def test_offsite_adapter_returns_error_without_database_configuration() -> None:
|
|
class SettingsStub:
|
|
mysql_dsn = ""
|
|
|
|
adapter = OffsiteNl2SqlAdapter(settings=SettingsStub())
|
|
context = RequestContext(
|
|
user_id="1",
|
|
trace_id="offsite-nl2sql-missing-dsn",
|
|
roles=("operator",),
|
|
permissions=("offsite:write",),
|
|
data_scope="all",
|
|
)
|
|
|
|
result = adapter.query("基金代码为15911，查询该基金最新净值", context)
|
|
|
|
assert result["status"] == "error"
|
|
assert result["message"] == "NL2SQL调用失败：RuntimeError"
|