import os import shutil import subprocess from decimal import Decimal from io import BytesIO import pytest from openpyxl import load_workbook from finance.exporting import CUSTOM_COLUMN_TOLERANCE, RECONCILE_TOLERANCES, build_formula_workbook from finance.formula_columns import to_decimal pytestmark = pytest.mark.django_db def custom_column(name="收入成本差"): return { "name": name, "meaning": "实际收入减成本", "scope": "按所选分组及本期、基期计算", "missing": "结果不可用(#N/A);零分母不可用", "expression": { "operator": "subtract", "left": {"metric": "revenue"}, "right": {"metric": "cost"}, }, } def test_to_decimal_evaluator_golden_samples(): values = { "revenue": Decimal("999999999999.99"), "cost": Decimal("0.0000001"), "customers": Decimal("7"), } # 大数与多层表达式:((收入 - 收入) + 成本×2) ÷ (成本×2) = 1,抵消后不丢失精度 cancellation = { "operator": "divide", "left": { "operator": "add", "left": { "operator": "subtract", "left": {"metric": "revenue"}, "right": {"metric": "revenue"}, }, "right": {"operator": "multiply", "left": {"metric": "cost"}, "right": {"number": 2}}, }, "right": {"operator": "multiply", "left": {"metric": "cost"}, "right": {"number": 2}}, } assert to_decimal(cancellation, values) == Decimal("1") # 极小值与比率 assert to_decimal({"metric": "cost"}, values) == Decimal("0.0000001") assert to_decimal( {"operator": "divide", "left": {"metric": "revenue"}, "right": {"metric": "customers"}}, values ) == Decimal("999999999999.99") / Decimal("7") # 除零与缺失指标返回 None(对应 Excel #N/A) assert ( to_decimal({"operator": "divide", "left": {"metric": "revenue"}, "right": {"number": 0}}, values) is None ) assert ( to_decimal( {"operator": "add", "left": {"metric": "cost"}, "right": {"metric": "customers"}}, {**values, "customers": None}, ) is None ) # 负数与常数组合 assert to_decimal( {"operator": "subtract", "left": {"number": -2.5}, "right": {"metric": "cost"}}, values ) == Decimal("-2.5") - Decimal("0.0000001") def test_reconcile_uses_independent_tolerances_per_metric_kind(workspace): ds, member, *_ = workspace 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) sheet = workbook["对账检查"] rows = list(sheet.iter_rows(values_only=True)) assert list(rows[0]) == [ "指标", "分组", "期间侧", "服务器 Decimal 参考", "公式结果", "差异", "容差", "状态", ] def find(metric, side="本期"): return next(row for row in rows[1:] if row[0] == metric and row[2] == side) assert find("消费人数")[6] == 0 # 人数严格相等 assert find("实际收入")[6] == pytest.approx(0.01) assert find("毛利贡献")[6] == pytest.approx(0.01) assert find("毛利率")[6] == pytest.approx(0.0001) assert find("ROI")[6] == pytest.approx(0.0001) assert find("客单价")[6] == pytest.approx(0.0001) # 状态公式按容差列比较(<=G),不再统一 <0.01 status = next( cell.value for cell in sheet[2] if isinstance(cell.value, str) and cell.value.startswith("=IF(D") ) assert "<=G2" in status # 参数页写明对账规则,容差调整需财务批准 parameters = {str(row[0]): row for row in workbook["参数与口径"].iter_rows(values_only=True) if row[0]} assert "人数严格相等" in parameters["对账容差"][1] assert parameters["对账容差"][3] == "否" def test_custom_columns_reconcile_against_decimal_reference(workspace): ds, member, *_ = workspace 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) rows = list(workbook["对账检查"].iter_rows(values_only=True)) custom_rows = [row for row in rows[1:] if row[0] == "收入成本差"] assert custom_rows pediatrics = next(row for row in custom_rows if row[1] == "儿科" and row[2] == "本期") assert Decimal(pediatrics[3]) == Decimal("60") # 100 - 40 assert pediatrics[6] == pytest.approx(float(CUSTOM_COLUMN_TOLERANCE)) status = next( row[7] for row in workbook["对账检查"].iter_rows(min_row=2, values_only=True) if row[0] == "收入成本差" and row[1] == "儿科" and row[2] == "本期" ) assert "<=G" in status and "通过" in status # 旧的占位文案不再出现 assert all("自定义公式无服务器参考" not in str(row[3]) for row in custom_rows) assert all(row[7] != "需业务验收" for row in custom_rows) def test_custom_column_reference_missing_when_dependency_unavailable(workspace): from finance.models import Fact ds, member, *_ = workspace # 外科 2026-08 成本置空 → cost 不可用 → 收入成本差参考为 None → 无参考 Fact.objects.filter(dataset=ds, period="2026-08", dimensions__department="外科").update(cost=None) 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) surgery = next( row for row in workbook["对账检查"].iter_rows(min_row=2, values_only=True) if row[0] == "收入成本差" and row[1] == "外科" and row[2] == "本期" ) assert surgery[3] in ("", None) # 无服务器参考 assert "无参考" in surgery[7] def test_reconcile_tolerances_cover_every_metric(): from finance.analysis import METRICS assert set(RECONCILE_TOLERANCES) == set(METRICS) assert RECONCILE_TOLERANCES["customers"] == 0 assert all( tolerance in (Decimal("0.01"), Decimal("0.0001")) for key, tolerance in RECONCILE_TOLERANCES.items() if key != "customers" ) def test_reconcile_passes_in_libreoffice_with_per_metric_tolerances(workspace, tmp_path): executable = os.environ.get("SOFFICE_BIN") or shutil.which("soffice") if not executable: pytest.skip("set SOFFICE_BIN to run the LibreOffice recalculation check") ds, member, *_ = workspace source, output = tmp_path / "source", tmp_path / "out" source.mkdir() output.mkdir() workbook_path = source / "reconcile.xlsx" with workbook_path.open("wb") as file: build_formula_workbook( ds, member, { "start": "2026-08", "end": "2026-08", "comparison": "yoy", "group": "department", "custom_columns": [custom_column()], }, file, ) profile = tmp_path / "profile" result = subprocess.run( [ executable, "--headless", f"-env:UserInstallation={profile.as_uri()}", "--convert-to", "xlsx", "--outdir", str(output), str(workbook_path), ], capture_output=True, text=True, timeout=120, check=False, ) assert result.returncode == 0, result.stderr or result.stdout values = load_workbook(output / workbook_path.name, data_only=True) statuses = [ (row[0], row[1], row[2], row[7]) for row in values["对账检查"].iter_rows(min_row=2, values_only=True) ] assert statuses assert all(status in ("通过", "无参考") for *_head, status in statuses), statuses