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}" assert "含义：实际收入减成本" in workbook["使用说明"]["B12"].value assert "范围：按所选分组及本期、基期计算" in workbook["使用说明"]["B12"].value 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