110 lines
4.0 KiB
Python
110 lines
4.0 KiB
Python
import json
|
|
from io import BytesIO
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
from openpyxl import load_workbook
|
|
|
|
from finance.exporting import build_formula_workbook
|
|
from finance.formula_columns import validate_column
|
|
from finance.formula_views import design_column
|
|
from finance.views import APIError
|
|
|
|
pytestmark = pytest.mark.django_db
|
|
|
|
|
|
def custom_column():
|
|
return {
|
|
"name": "收入成本差",
|
|
"meaning": "实际收入减成本",
|
|
"scope": "按所选分组及本期、基期计算",
|
|
"missing": "结果不可用（#N/A）；零分母不可用",
|
|
"expression": {
|
|
"operator": "subtract",
|
|
"left": {"metric": "revenue"},
|
|
"right": {"metric": "cost"},
|
|
},
|
|
}
|
|
|
|
|
|
def model_response(design):
|
|
return type(
|
|
"Response",
|
|
(),
|
|
{
|
|
"raise_for_status": lambda self: None,
|
|
"json": lambda self: {
|
|
"status": "completed",
|
|
"output": [{"content": [{"type": "output_text", "text": json.dumps(design)}]}],
|
|
},
|
|
},
|
|
)()
|
|
|
|
|
|
def test_design_column_asks_for_missing_details_then_returns_safe_definition(workspace, settings):
|
|
ds, *_ = workspace
|
|
settings.AI_ENABLED = True
|
|
settings.OPENAI_API_KEY = "test-key"
|
|
settings.OPENAI_MODEL = "test-model"
|
|
ctx = {"current": ["2026-08"], "baseline": ["2025-08"], "filters": {}, "group": "department"}
|
|
missing = {"complete": False, "questions": ["请补充缺失值和零分母的处理方式。"], "column": None}
|
|
complete = {"complete": True, "questions": [], "column": custom_column()}
|
|
with patch("finance.formula_views.httpx.Client") as client:
|
|
client.return_value.__enter__.return_value.post.return_value = model_response(missing)
|
|
first = design_column([{"role": "user", "content": "增加收入减成本的新列"}], ds, ctx)
|
|
assert first == missing
|
|
|
|
messages = [
|
|
{"role": "user", "content": "增加收入减成本的新列"},
|
|
{"role": "assistant", "content": first["questions"][0]},
|
|
{"role": "user", "content": "缺失或零分母显示不可用"},
|
|
]
|
|
client.return_value.__enter__.return_value.post.return_value = model_response(complete)
|
|
result = design_column(messages, ds, ctx)
|
|
|
|
assert result == complete
|
|
payload = client.return_value.__enter__.return_value.post.call_args.kwargs["json"]
|
|
assert payload["input"][-1] == messages[-1]
|
|
validate_column(result["column"])
|
|
|
|
|
|
def test_design_column_rejects_unsafe_model_definitions(workspace, settings):
|
|
ds, *_ = workspace
|
|
settings.AI_ENABLED = True
|
|
settings.OPENAI_API_KEY = "test-key"
|
|
settings.OPENAI_MODEL = "test-model"
|
|
ctx = {"current": ["2026-08"], "baseline": ["2025-08"], "filters": {}, "group": "department"}
|
|
invalid = {
|
|
"complete": True,
|
|
"questions": [],
|
|
"column": {**custom_column(), "expression": "=HYPERLINK()"},
|
|
}
|
|
with patch("finance.formula_views.httpx.Client") as client:
|
|
client.return_value.__enter__.return_value.post.return_value = model_response(invalid)
|
|
with pytest.raises(APIError, match="未创建工作簿"):
|
|
design_column([{"role": "user", "content": "随意算一个值"}], ds, ctx)
|
|
|
|
|
|
def test_formula_labels_from_ai_are_written_as_plain_text(workspace):
|
|
ds, member, *_ = workspace
|
|
definition = custom_column()
|
|
definition["name"] = '=HYPERLINK("https://invalid")'
|
|
validate_column(definition)
|
|
workbook_bytes = BytesIO()
|
|
build_formula_workbook(
|
|
ds,
|
|
member,
|
|
{
|
|
"start": "2026-08",
|
|
"end": "2026-08",
|
|
"comparison": "yoy",
|
|
"group": "department",
|
|
"custom_columns": [definition],
|
|
},
|
|
workbook_bytes,
|
|
)
|
|
workbook = load_workbook(BytesIO(workbook_bytes.getvalue()), data_only=False)
|
|
summary = workbook["分析汇总"]
|
|
cell = next(cell for row in summary.iter_rows() for cell in row if cell.value == definition["name"])
|
|
assert cell.data_type == "s"
|