Jayway Liu
509b406254
- F01: _safe_unlink_export 不跟随符号链接、逐层校验父目录，导出失败路径与 清理命令统一使用，只清理受控目录内本次创建的文件 - F02: 清理失败记录告警日志；清理命令回收 failed 任务残留文件 - F05: 定义保存事务内锁数据集行，配额原子检查；revision 冲突返回 409； 保留最近 20 版历史；支持个人/团队共享；审计与变更同事务 - F06: 工作簿参数页毛利率改为只读快照；可编辑单元格解锁并加数值校验 - F07: 对账按指标分容差（人数 0、比率 0.0001、金额 0.01）；自定义列使用 服务端 Decimal 表达式求值参考 - F08: 导出任务执行/下载阶段重校验账号有效性与角色 - 对话草稿会话恢复；载入定义后追问携带原定义（F04）；下载改浏览器直传 - 迁移 0006-0008；PostgreSQL 咨询锁语义修正（import_worker）
269 lines
9.7 KiB
Python
269 lines
9.7 KiB
Python
import json
|
|
from decimal import Decimal
|
|
from io import BytesIO
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
from django.contrib.auth.models import User
|
|
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.models import Dataset, Fact, Membership, Organization
|
|
|
|
pytestmark = pytest.mark.django_db
|
|
|
|
|
|
@pytest.fixture
|
|
def formula_data():
|
|
org = Organization.objects.create(name="Formula tests")
|
|
user = User.objects.create_user("formula", password="Formula-TestPassword991!")
|
|
member = Membership.objects.create(user=user, organization=org, role="admin")
|
|
mapping = {
|
|
"sheet": "CSV",
|
|
"columns": [
|
|
{"index": 0, "name": "Period", "role": "period"},
|
|
{"index": 1, "name": "Department", "role": "dimension", "key": "department"},
|
|
{"index": 2, "name": "Revenue", "role": "revenue"},
|
|
{"index": 3, "name": "Cost", "role": "cost"},
|
|
],
|
|
"roi_formula": "profit_over_cost",
|
|
}
|
|
ds = Dataset.objects.create(
|
|
organization=org,
|
|
created_by=user,
|
|
name="formula.csv",
|
|
sha256="b" * 64,
|
|
status="ready",
|
|
mapping=mapping,
|
|
periods=["2025-08", "2026-08"],
|
|
)
|
|
for period, revenue, cost in (("2025-08", 80, 30), ("2026-08", 100, 40)):
|
|
Fact.objects.create(
|
|
dataset=ds,
|
|
period=period,
|
|
dimensions={"department": "儿科"},
|
|
revenue=Decimal(revenue),
|
|
cost=Decimal(cost),
|
|
gross=Decimal(revenue - cost),
|
|
customer=f"{period}-patient",
|
|
source_row=1 if period == "2025-08" else 2,
|
|
raw=[period, "儿科", revenue, cost],
|
|
)
|
|
return ds, member
|
|
|
|
|
|
def custom_column():
|
|
return {
|
|
"name": "收入成本差",
|
|
"meaning": "实际收入减成本",
|
|
"scope": "按所选分组及本期、基期计算",
|
|
"missing": "结果不可用（#N/A）；零分母不可用",
|
|
"expression": {
|
|
"operator": "subtract",
|
|
"left": {"metric": "revenue"},
|
|
"right": {"metric": "cost"},
|
|
},
|
|
}
|
|
|
|
|
|
def test_formula_workbook_exposes_custom_metric_as_recalculable_column(formula_data):
|
|
ds, member = formula_data
|
|
output = BytesIO()
|
|
build_formula_workbook(
|
|
ds,
|
|
member,
|
|
{
|
|
"start": "2026-08",
|
|
"end": "2026-08",
|
|
"comparison": "yoy",
|
|
"group": "department",
|
|
"custom_columns": [custom_column()],
|
|
},
|
|
output,
|
|
)
|
|
workbook = load_workbook(BytesIO(output.getvalue()), data_only=False)
|
|
assert workbook["计算明细"]["M2"].value == '=IF(H2="direct",F2,IF(H2="lookup",D2*G2,""))'
|
|
assert workbook["计算明细"]["I2"].data_type == "n"
|
|
summary = workbook["分析汇总"]
|
|
metric_row = next(
|
|
row for row in range(6, summary.max_row + 1) if summary.cell(row, 1).value == "收入成本差"
|
|
)
|
|
formula = summary.cell(metric_row, 3).value
|
|
assert formula.startswith("=IF(OR(ISNA(C")
|
|
assert "IFERROR((C" in formula and "NA())" in formula
|
|
assert "自定义指标" in workbook.sheetnames
|
|
custom = workbook["自定义指标"]
|
|
assert custom["B1"].value == "收入成本差（本期）"
|
|
assert custom["B2"].value == f"='分析汇总'!C{metric_row}"
|
|
instruction_text = " ".join(str(cell.value) for row in workbook["使用说明"].iter_rows() for cell in row)
|
|
assert "含义：实际收入减成本" in instruction_text
|
|
assert "范围：按所选分组及本期、基期计算" in instruction_text
|
|
|
|
|
|
def test_formula_workbook_protects_structure_and_unlocks_calculation_inputs(formula_data):
|
|
ds, member = formula_data
|
|
output = BytesIO()
|
|
build_formula_workbook(
|
|
ds,
|
|
member,
|
|
{"start": "2026-08", "end": "2026-08", "comparison": "yoy", "group": "department"},
|
|
output,
|
|
)
|
|
workbook = load_workbook(BytesIO(output.getvalue()), data_only=False)
|
|
data = workbook["计算明细"]
|
|
assert data.protection.sheet
|
|
assert data["D2"].protection.locked is False
|
|
assert data["F2"].protection.locked is False
|
|
assert data["G2"].protection.locked is False
|
|
assert data["I2"].protection.locked is False
|
|
assert data["C2"].protection.locked is True
|
|
assert data["M2"].protection.locked is True
|
|
assert workbook["原始证据"].protection.sheet
|
|
|
|
|
|
def test_custom_expression_rejects_formula_text_and_unknown_metrics():
|
|
from finance.formula_columns import metrics_used, to_excel
|
|
|
|
expression = {
|
|
"operator": "divide",
|
|
"left": {"metric": "revenue"},
|
|
"right": {
|
|
"operator": "add",
|
|
"left": {"metric": "cost"},
|
|
"right": {"number": 2.5},
|
|
},
|
|
}
|
|
validate_column({**custom_column(), "expression": expression})
|
|
assert metrics_used(expression) == {"revenue", "cost"}
|
|
assert to_excel(expression, {"revenue": "C6", "cost": "D6"}) == "(C6/(D6+2.5))"
|
|
column = custom_column()
|
|
column["expression"] = {"metric": "__import__('os')"}
|
|
with pytest.raises(ValueError, match="不支持的指标"):
|
|
validate_column(column)
|
|
|
|
column = custom_column()
|
|
column["expression"] = {"formula": "=HYPERLINK(...)"}
|
|
with pytest.raises(ValueError, match="不支持的字段或运算"):
|
|
validate_column(column)
|
|
|
|
|
|
def test_design_column_preserves_clarification_branch_and_full_conversation(formula_data, settings):
|
|
ds, _ = formula_data
|
|
settings.AI_ENABLED = True
|
|
settings.OPENAI_API_KEY = "test-key"
|
|
settings.OPENAI_MODEL = "test-model"
|
|
|
|
clarification = {
|
|
"status": "completed",
|
|
"output": [
|
|
{
|
|
"content": [
|
|
{
|
|
"type": "output_text",
|
|
"text": '{"complete":false,"questions":["请说明零分母如何处理。"],"column":null}',
|
|
}
|
|
]
|
|
}
|
|
],
|
|
}
|
|
with patch("finance.formula_views.httpx.Client") as http:
|
|
http.return_value.__enter__.return_value.post.return_value = type(
|
|
"Response",
|
|
(),
|
|
{"raise_for_status": lambda self: None, "json": lambda self: clarification},
|
|
)()
|
|
incomplete = design_column(
|
|
[{"role": "user", "content": "新增利润率列"}],
|
|
ds,
|
|
{"current": ["2026-08"], "baseline": ["2025-08"], "filters": {}, "group": "department"},
|
|
)
|
|
assert incomplete == {
|
|
"complete": False,
|
|
"questions": ["请说明零分母如何处理。"],
|
|
"column": None,
|
|
}
|
|
|
|
class Response:
|
|
def raise_for_status(self):
|
|
pass
|
|
|
|
def json(self):
|
|
return {
|
|
"status": "completed",
|
|
"output": [
|
|
{
|
|
"content": [
|
|
{
|
|
"type": "output_text",
|
|
"text": '{"complete":false,"questions":["零分母如何处理？"],"column":null}',
|
|
}
|
|
]
|
|
}
|
|
],
|
|
}
|
|
|
|
messages = [
|
|
{"role": "user", "content": "增加收入成本差列"},
|
|
{"role": "assistant", "content": "请说明计算方式和缺失值处理。"},
|
|
{"role": "user", "content": "收入减成本，缺失时不可用"},
|
|
]
|
|
with patch("finance.formula_views.httpx.Client") as client:
|
|
client.return_value.__enter__.return_value.post.return_value = Response()
|
|
result = design_column(
|
|
messages,
|
|
ds,
|
|
{"current": ["2026-08"], "baseline": ["2025-08"], "filters": {}, "group": "department"},
|
|
)
|
|
|
|
assert result == {"complete": False, "questions": ["零分母如何处理？"], "column": None}
|
|
payload = client.return_value.__enter__.return_value.post.call_args.kwargs["json"]
|
|
assert payload["input"][1:] == messages
|
|
|
|
|
|
def test_design_column_returns_only_validated_formula_definition(formula_data, settings):
|
|
ds, _ = formula_data
|
|
settings.AI_ENABLED = True
|
|
settings.OPENAI_API_KEY = "test-key"
|
|
settings.OPENAI_MODEL = "test-model"
|
|
definition = custom_column()
|
|
response = type(
|
|
"Response",
|
|
(),
|
|
{
|
|
"raise_for_status": lambda self: None,
|
|
"json": lambda self: {
|
|
"status": "completed",
|
|
"output": [
|
|
{
|
|
"content": [
|
|
{
|
|
"type": "output_text",
|
|
"text": json.dumps(
|
|
{"complete": True, "questions": [], "column": definition},
|
|
ensure_ascii=False,
|
|
),
|
|
}
|
|
]
|
|
}
|
|
],
|
|
},
|
|
},
|
|
)()
|
|
with patch("finance.formula_views.httpx.Client") as http:
|
|
http.return_value.__enter__.return_value.post.return_value = response
|
|
result = design_column(
|
|
[
|
|
{"role": "user", "content": "增加收入成本差列"},
|
|
{"role": "assistant", "content": "请补充计算口径"},
|
|
{"role": "user", "content": "收入减成本，按分组选期间计算"},
|
|
],
|
|
ds,
|
|
{"current": ["2026-08"], "baseline": ["2025-08"], "filters": {}, "group": "department"},
|
|
)
|
|
assert result == {"complete": True, "questions": [], "column": definition}
|
|
payload = http.return_value.__enter__.return_value.post.call_args.kwargs["json"]
|
|
assert payload["store"] is False
|
|
assert payload["text"]["format"]["strict"] is True
|