fa-agent/tests/test_formula_reconcile_tolerances.py

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