cma-management/backend/tests/test_reports.py

"""报表中心模块测试 — 管理利润表/预算执行/KPI趋势/BSC评分卡/新30号准则三表/杜邦/自动报告生成

覆盖 reports.py 全部18个端点：
profit-summary / budget-execution / kpi-trends / bsc-scorecard /
profit-statement(old/new/dual) / mpm-calculate / restatement / category-map /
balance-sheet / cash-flow / statutory / statutory/export / dupont /
generate(weekly/monthly/special) / history / history/{id}
"""
import pytest
from fastapi.testclient import TestClient
from sqlalchemy.orm import Session
from tests.conftest import create_test_user, get_token_for_user, auth_header
from app.models import (
    KPIDefinition, KPIValue, BudgetPlan, StrategicMap, KPIAlert, ActionPlan,
    Subject, ReportHistory,
)
from app.models.budget_plan import BudgetPlan as BP


# ── 测试数据工厂（报表口径KPI） ──

def mk_kpi(db: Session, code: str, dim: str = "finance", target: float = 100.0,
           entity_id: int = 1, name: str = None, unit: str = "万元") -> KPIDefinition:
    kpi = KPIDefinition(
        kpi_code=code, kpi_name=name or code, dimension=dim,
        target_value=target, unit=unit, status="active", frequency="monthly",
        entity_id=entity_id,
    )
    db.add(kpi)
    db.commit()
    db.refresh(kpi)
    return kpi


def mk_val(db: Session, kpi: KPIDefinition, period: str, val: float) -> KPIValue:
    v = KPIValue(
        kpi_id=kpi.id, period=period, actual_value=val,
        source_type="test", source_batch="test", data_status="verified",
    )
    db.add(v)
    db.commit()
    db.refresh(v)
    return v


def mk_budget(db: Session, kpi: KPIDefinition, period: str, val: float) -> BudgetPlan:
    y, m = period.split("-")
    b = BP(
        kpi_id=kpi.id, period=period, budget_value=val,
        budget_year=int(y), budget_month=int(m), status="active",
    )
    db.add(b)
    db.commit()
    db.refresh(b)
    return b


def seed_profit_kpis(db: Session, period: str = "2026-06", prev: str = "2026-05"):
    """利润表口径KPI：收入/毛利率/净利率/成本率，含上期环比数据"""
    pairs = [
        ("F_REVENUE", 100.0, 80.0),
        ("F_PROFIT_RATE", 20.0, 15.0),
        ("F_NET_PROFIT_RATE", 10.0, 8.0),
        ("F_COST_RATIO", 60.0, 65.0),
    ]
    for code, cur, pv in pairs:
        k = mk_kpi(db, code)
        mk_val(db, k, period, cur)
        mk_val(db, k, prev, pv)


def seed_subject_kpis(db: Session, period: str = "2026-06"):
    """新30号准则科目映射KPI（经 _get_subject_amount 的 kpi_code_map）"""
    data = {
        "F_REVENUE": 100.0, "F_COST": 60.0, "F_SELLING_EXP": 5.0,
        "F_ADMIN_EXP": 10.0, "F_RD_EXP": 4.0, "F_FINANCE_EXP": 0.5,
        "F_INTEREST_INCOME": 3.0, "F_INVEST_INCOME": 3.0,
        "F_INTEREST_EXP": 2.0, "F_TAX_EXP": 1.0, "F_FX_LOSS": 0.5,
    }
    kpis = {}
    for code, val in data.items():
        k = mk_kpi(db, code)
        mk_val(db, k, period, val)
        kpis[code] = k
    return kpis


class TestProfitSummary:
    """报表1：管理利润表"""

    BASE = "/api/cma/reports/profit-summary"

    def test_profit_summary_with_data(self, client: TestClient, db: Session):
        """有数据时：营收/变动成本/边际贡献/固定成本/息税前利润 + 环比"""
        create_test_user(db)
        token = get_token_for_user(client)
        seed_profit_kpis(db)

        resp = client.get(f"{self.BASE}?period=2026-06", headers=auth_header(token))
        assert resp.status_code == 200
        data = resp.json()
        assert data["period"] == "2026-06"
        assert data["prev_period"] == "2026-05"
        items = {i["name"]: i for i in data["items"]}
        # 营业收入 100 vs 上期80 → +25%
        assert items["营业收入"]["value"] == 100.0
        assert items["营业收入"]["prev_value"] == 80.0
        assert items["营业收入"]["change_rate"] == 25.0
        # 变动成本 = 营收×50% = 50
        assert items["减：变动成本"]["value"] == 50.0
        # 边际贡献 = 毛利 = 100×20% = 20
        assert items["＝ 边际贡献"]["value"] == 20.0
        assert items["＝ 边际贡献"]["is_subtotal"] is True
        # 固定成本 = 总成本60 - 变动成本50 = 10
        assert items["减：固定成本"]["value"] == 10.0
        # 息税前利润 = 100×10% = 10
        assert items["＝ 息税前利润"]["value"] == 10.0
        assert items["＝ 息税前利润"]["is_total"] is True

    def test_profit_summary_empty_db(self, client: TestClient, db: Session):
        """无任何KPI数据 → 200，值全部为None，不报错"""
        create_test_user(db)
        token = get_token_for_user(client)
        resp = client.get(self.BASE, headers=auth_header(token))
        assert resp.status_code == 200
        data = resp.json()
        assert len(data["items"]) == 5
        assert data["items"][0]["value"] is None

    def test_profit_summary_year_boundary(self, client: TestClient, db: Session):
        """1月 → 上期跨年（2026-01 → 2025-12）"""
        create_test_user(db)
        token = get_token_for_user(client)
        resp = client.get(f"{self.BASE}?period=2026-01", headers=auth_header(token))
        assert resp.status_code == 200
        assert resp.json()["prev_period"] == "2025-12"


class TestBudgetExecution:
    """报表2：预算执行报告"""

    BASE = "/api/cma/reports/budget-execution"

    def _seed(self, db: Session):
        # 超预算红(30%) / 正常(5%) / 超支黄(-25%→红? abs>20=red) 细分场景
        k1 = mk_kpi(db, "BH_REVENUE", name="营业收入")
        mk_val(db, k1, "2026-06", 130.0)
        mk_budget(db, k1, "2026-06", 100.0)          # +30% → red over
        k2 = mk_kpi(db, "BH_PROFIT", name="净利润")
        mk_val(db, k2, "2026-06", 105.0)
        mk_budget(db, k2, "2026-06", 100.0)          # +5% → normal
        k3 = mk_kpi(db, "BH_COST", name="成本", dim="customer")
        mk_val(db, k3, "2026-06", 75.0)
        mk_budget(db, k3, "2026-06", 100.0)          # -25% → red under
        k4 = mk_kpi(db, "BH_ZERO_TARGET", name="零目标KPI")
        mk_val(db, k4, "2026-06", 50.0)
        mk_budget(db, k4, "2026-06", 0.0)            # 预算0 → gray
        k5 = mk_kpi(db, "BH_EMPTY", name="无数据KPI", target=None)  # 完全无数据 → 跳过
        return [k1, k2, k3, k4, k5]

    def test_budget_execution_summary(self, client: TestClient, db: Session):
        """summary统计与alert_level分级"""
        create_test_user(db)
        token = get_token_for_user(client)
        self._seed(db)

        resp = client.get(f"{self.BASE}?period=2026-06", headers=auth_header(token))
        assert resp.status_code == 200
        data = resp.json()
        assert data["period"] == "2026-06"
        s = data["summary"]
        assert s["total"] == 4           # 无数据KPI被跳过
        assert s["with_budget"] == 4
        assert s["over_budget"] == 1     # +30%
        assert s["under_budget"] == 1    # -25%
        assert s["normal"] == 2          # +5% 和 gray
        levels = {i["kpi_code"]: i["alert_level"] for i in data["items"]}
        assert levels["BH_REVENUE"] == "red"
        assert levels["BH_PROFIT"] == "normal"
        assert levels["BH_COST"] == "red"
        assert levels["BH_ZERO_TARGET"] == "gray"

    def test_budget_execution_filters(self, client: TestClient, db: Session):
        """dimension / alert_level 过滤"""
        create_test_user(db)
        token = get_token_for_user(client)
        self._seed(db)

        r1 = client.get(f"{self.BASE}?period=2026-06&dimension=finance", headers=auth_header(token))
        assert {i["kpi_code"] for i in r1.json()["items"]} == {"BH_REVENUE", "BH_PROFIT", "BH_ZERO_TARGET"}

        r2 = client.get(f"{self.BASE}?period=2026-06&alert_level=red", headers=auth_header(token))
        assert {i["kpi_code"] for i in r2.json()["items"]} == {"BH_REVENUE", "BH_COST"}
        assert r2.json()["summary"]["total"] == 4  # summary不过滤


class TestKpiTrends:
    """报表3：KPI趋势报告"""

    BASE = "/api/cma/reports/kpi-trends"

    def _seed(self, db: Session):
        k = mk_kpi(db, "BH_REVENUE", name="营业收入", target=120.0)
        for i, p in enumerate(["2026-01", "2026-02", "2026-03", "2026-04", "2026-05", "2026-06"]):
            mk_val(db, k, p, 10 + i * 2)   # 10,12,14,16,18,20 上升
        k2 = mk_kpi(db, "BH_EMPTY_TREND", name="无数据KPI")
        return k, k2

    def test_trend_up(self, client: TestClient, db: Session):
        """上升趋势 + 统计值"""
        create_test_user(db)
        token = get_token_for_user(client)
        k, _ = self._seed(db)

        resp = client.get(f"{self.BASE}?kpi_id={k.id}", headers=auth_header(token))
        assert resp.status_code == 200
        rows = resp.json()["data"]
        assert len(rows) == 1
        r = rows[0]
        assert r["kpi_code"] == "BH_REVENUE"
        assert r["trend_dir"] == "up"
        assert len(r["trend"]) == 6
        assert r["avg"] == 15.0
        assert r["max"] == 20.0
        assert r["min"] == 10.0
        assert r["target_value"] == 120.0

    def test_trend_months_limit(self, client: TestClient, db: Session):
        """months 限制条数"""
        create_test_user(db)
        token = get_token_for_user(client)
        k, _ = self._seed(db)
        resp = client.get(f"{self.BASE}?kpi_id={k.id}&months=3", headers=auth_header(token))
        assert resp.status_code == 200
        assert len(resp.json()["data"][0]["trend"]) == 3

    def test_trend_no_data(self, client: TestClient, db: Session):
        """无数据KPI → 空trend + stable"""
        create_test_user(db)
        token = get_token_for_user(client)
        _, k2 = self._seed(db)
        resp = client.get(f"{self.BASE}?kpi_id={k2.id}", headers=auth_header(token))
        assert resp.status_code == 200
        r = resp.json()["data"][0]
        assert r["trend"] == []
        assert r["trend_dir"] == "stable"
        assert r["avg"] is None

    def test_trend_months_out_of_range(self, client: TestClient, db: Session):
        """months 越界(2 / 37) → 422"""
        create_test_user(db)
        token = get_token_for_user(client)
        for m in (2, 37):
            resp = client.get(f"{self.BASE}?months={m}", headers=auth_header(token))
            assert resp.status_code == 422

    def test_trend_dimension_filter(self, client: TestClient, db: Session):
        """dimension 过滤"""
        create_test_user(db)
        token = get_token_for_user(client)
        self._seed(db)
        resp = client.get(f"{self.BASE}?dimension=customer", headers=auth_header(token))
        assert resp.status_code == 200
        assert resp.json()["data"] == []


class TestBscScorecard:
    """报表4：四维度绩效评分卡"""

    BASE = "/api/cma/reports/bsc-scorecard"

    def test_scorecard_from_kpis(self, client: TestClient, db: Session):
        """无发布地图 → 按维度聚合KPI算分"""
        create_test_user(db)
        token = get_token_for_user(client)
        k1 = mk_kpi(db, "BH_REVENUE", dim="finance", target=100.0)
        mk_val(db, k1, "2026-06", 95.0)    # 95分 green
        k2 = mk_kpi(db, "BH_NET_PROFIT", dim="finance", target=100.0)
        mk_val(db, k2, "2026-06", 50.0)    # 50分 red
        k3 = mk_kpi(db, "BH_SATISFACTION", dim="customer", target=10.0)
        mk_val(db, k3, "2026-06", 9.0)     # 90分 green
        k4 = mk_kpi(db, "BH_ZERO_TARGET", dim="learning", target=0.0)
        mk_val(db, k4, "2026-06", 5.0)     # 目标0 → gray 无分

        resp = client.get(f"{self.BASE}?period=2026-06", headers=auth_header(token))
        assert resp.status_code == 200
        data = resp.json()
        assert data["map_id"] is None
        dims = {d["key"]: d for d in data["dimensions"]}
        assert dims["finance"]["score"] == 72.5      # (95+50)/2
        assert dims["customer"]["score"] == 90.0
        assert dims["learning"]["score"] == 0.0      # 无有效数据
        assert data["overall_score"] == 54.2         # (72.5+90+0)/3
        # level 判定
        fin_kpis = {k["code"]: k for k in dims["finance"]["objectives"][0]["kpis"]}
        assert fin_kpis["BH_REVENUE"]["level"] == "green"
        assert fin_kpis["BH_NET_PROFIT"]["level"] == "red"
        learn_kpis = {k["code"]: k for k in dims["learning"]["objectives"][0]["kpis"]}
        assert learn_kpis["BH_ZERO_TARGET"]["level"] == "gray"

    def test_scorecard_from_map(self, client: TestClient, db: Session):
        """有已发布战略地图 → 走地图维度路径"""
        create_test_user(db)
        token = get_token_for_user(client)
        k = mk_kpi(db, "BH_REVENUE", dim="finance", target=100.0)
        mk_val(db, k, "2026-06", 95.0)
        dims = [{
            "key": "finance", "name": "财务维度", "icon": "💰", "color": "#409eff",
            "objectives": [{"name": "增收", "kpis": ["BH_REVENUE"]}],
        }]
        sm = StrategicMap(title="已发布地图", status="published", dimensions=dims,
                          canvas_data={"connections": []})
        db.add(sm)
        db.commit()

        resp = client.get(f"{self.BASE}?period=2026-06&map_id={sm.id}", headers=auth_header(token))
        assert resp.status_code == 200
        data = resp.json()
        assert data["map_id"] == sm.id
        assert data["map_title"] == "已发布地图"
        assert data["dimensions"][0]["score"] == 95.0
        assert data["dimensions"][0]["objectives"][0]["kpis"][0]["level"] == "green"

    def test_scorecard_string_dimensions(self, client: TestClient, db: Session):
        """dimensions 存为JSON字符串（旧数据兼容）"""
        create_test_user(db)
        token = get_token_for_user(client)
        k = mk_kpi(db, "BH_REVENUE", dim="finance", target=100.0)
        mk_val(db, k, "2026-06", 88.0)
        import json as _json
        sm = StrategicMap(title="字符串维度地图", status="published",
                          dimensions=_json.dumps([{
                              "key": "finance", "name": "财务", "icon": "💰",
                              "objectives": [{"name": "增收", "kpis": ["BH_REVENUE"]}],
                          }]),
                          canvas_data={"connections": []})
        db.add(sm)
        db.commit()
        resp = client.get(f"{self.BASE}?period=2026-06", headers=auth_header(token))
        assert resp.status_code == 200
        assert resp.json()["map_id"] == sm.id


class TestProfitStatement:
    """利润表：旧/新/双列格式"""

    BASE = "/api/cma/reports/profit-statement"

    def test_old_format(self, client: TestClient, db: Session):
        """format=old → 复用管理利润表"""
        create_test_user(db)
        token = get_token_for_user(client)
        seed_profit_kpis(db)
        resp = client.get(f"{self.BASE}?period=2026-06&format=old", headers=auth_header(token))
        assert resp.status_code == 200
        data = resp.json()
        assert "items" in data and len(data["items"]) == 5

    def test_new_format_blocks(self, client: TestClient, db: Session):
        """format=new → 五板块 + 附注，净利润=板块之和"""
        create_test_user(db)
        token = get_token_for_user(client)
        seed_subject_kpis(db)
        resp = client.get(f"{self.BASE}?period=2026-06&format=new", headers=auth_header(token))
        assert resp.status_code == 200
        data = resp.json()
        assert data["format"] == "new"
        assert len(data["blocks"]) == 5
        keys = {b["key"] for b in data["blocks"]}
        assert keys == {"operating", "investing", "financing", "tax", "discontinued"}
        # 经营 21.5(100-60-5-10-4+0.5) + 投资 6(3+3) - 筹资 1.5(2-0.5) - 所得税 1 = 25.0
        assert data["net_profit"] == 25.0
        op = next(b for b in data["blocks"] if b["key"] == "operating")
        assert op["has_real_data"] is True
        assert data["notes"]["revenue_total"] == 100.0
        assert len(data["notes"]["key_ratios"]) >= 2

    def test_dual_format(self, client: TestClient, db: Session):
        """format=dual → 新旧双列对比"""
        create_test_user(db)
        token = get_token_for_user(client)
        seed_subject_kpis(db)
        resp = client.get(f"{self.BASE}?period=2026-06&format=dual", headers=auth_header(token))
        assert resp.status_code == 200
        data = resp.json()
        assert data["format"] == "dual"
        assert "old_format" in data and "new_format" in data
        assert data["new_format"]["net_profit"] == 25.0


class TestMpmCalculate:
    """MPM管理层指标计算器"""

    BASE = "/api/cma/reports/mpm-calculate"

    def test_invalid_type(self, client: TestClient, db: Session):
        """不支持的指标类型 → 400"""
        create_test_user(db)
        token = get_token_for_user(client)
        resp = client.post(self.BASE, headers=auth_header(token),
                           json={"indicator_type": "not_a_type", "period": "2026-06"})
        assert resp.status_code == 400

    def test_default_empty_db(self, client: TestClient, db: Session):
        """空库 → 基准0，has_real_data False"""
        create_test_user(db)
        token = get_token_for_user(client)
        resp = client.post(self.BASE, headers=auth_header(token),
                           json={"indicator_type": "adjusted_net_profit", "period": "2026-06"})
        assert resp.status_code == 200
        data = resp.json()
        assert data["has_real_data"] is False
        assert data["base_value"] == 0
        assert data["final_value"] == 0

    def test_adjusted_net_profit_with_data(self, client: TestClient, db: Session):
        """有净利润+投资收益数据 → 调整后净利润 = 21.5 - 3(非经常性投资) = 18.5"""
        create_test_user(db)
        token = get_token_for_user(client)
        seed_subject_kpis(db)
        resp = client.post(self.BASE, headers=auth_header(token),
                           json={"indicator_type": "adjusted_net_profit", "period": "2026-06"})
        assert resp.status_code == 200
        data = resp.json()
        assert data["has_real_data"] is True
        assert data["base_value"] == 25.0
        assert data["final_value"] == 22.0  # 25.0 - 3(非经常性投资收益)
        assert data["adjustment_count"] == 3

    def test_ebitda(self, client: TestClient, db: Session):
        """EBITDA = 净利润25 + 所得税1 + 利息支出2 = 28.0"""
        create_test_user(db)
        token = get_token_for_user(client)
        seed_subject_kpis(db)
        resp = client.post(self.BASE, headers=auth_header(token),
                           json={"indicator_type": "ebitda", "period": "2026-06"})
        assert resp.status_code == 200
        assert resp.json()["final_value"] == 28.0

    def test_free_cash_flow(self, client: TestClient, db: Session):
        """自由现金流 = 经营现金流30（有值时优先）"""
        create_test_user(db)
        token = get_token_for_user(client)
        k = mk_kpi(db, "F_OPERATING_CF")
        mk_val(db, k, "2026-06", 30.0)
        resp = client.post(self.BASE, headers=auth_header(token),
                           json={"indicator_type": "free_cash_flow", "period": "2026-06"})
        assert resp.status_code == 200
        data = resp.json()
        assert data["base_value"] == 30.0
        assert data["final_value"] == 30.0

    def test_custom_adjustments(self, client: TestClient, db: Session):
        """自定义指标+显式调整金额"""
        create_test_user(db)
        token = get_token_for_user(client)
        resp = client.post(self.BASE, headers=auth_header(token),
                           json={
                               "indicator_type": "custom",
                               "period": "2026-06",
                               "adjustments": [
                                   {"code": "adjustment_1", "name": "加回项", "sign": 1,
                                    "checked": True, "amount": 100.0},
                                   {"code": "adjustment_2", "name": "扣除项", "sign": -1,
                                    "checked": True, "amount": 30.0},
                               ],
                           })
        assert resp.status_code == 200
        assert resp.json()["final_value"] == 70.0  # 0 + 100 - 30


class TestRestatement:
    """追溯调整：新旧口径对比"""

    BASE = "/api/cma/reports/restatement"

    def test_restatement_with_data(self, client: TestClient, db: Session):
        """旧口径KPI + 新口径映射 → 对比行与净利润对比"""
        create_test_user(db)
        token = get_token_for_user(client)
        # 旧口径
        for code, val in [("F_REVENUE", 100.0), ("F_COST", 50.0),
                          ("F_SELLING_EXP", 3.0), ("F_ADMIN_EXP", 20.0),
                          ("F_RD_EXP", 5.0)]:
            k = mk_kpi(db, code)
            mk_val(db, k, "2026-06", val)

        resp = client.get(f"{self.BASE}?period=2026-06", headers=auth_header(token))
        assert resp.status_code == 200
        data = resp.json()
        assert data["period"] == "2026-06"
        names = [i["item_name"] for i in data["items"]]
        # 管理费用：新旧口径共用F_ADMIN_EXP → 新值=旧值=20（剥离逻辑在KPI口径下不触发）
        admin = next(i for i in data["items"] if i["item_name"] == "减：管理费用")
        assert admin["old_value"] == 20.0
        assert admin["new_value"] == 20.0
        assert admin["needs_adjustment"] is False
        # 研发费用单独列示行（新30号准则）
        rd = next(i for i in data["items"] if i["item_name"] == "减：研发费用（单独列示）")
        assert rd["new_value"] == 5.0
        assert rd["needs_adjustment"] is True
        assert rd["adjustment_reason"] == "新30号准则单独列示"
        # 旧口径净利润 = 100-50-3-20 = 27；新口径 = 100-50-3-20-5 = 22
        npc = data["net_profit_comparison"]
        assert npc["old_net_profit"] == 27.0
        assert npc["new_net_profit"] == 22.0
        assert npc["difference"] == -5.0

    def test_restatement_empty(self, client: TestClient, db: Session):
        """空库 → 200 正常返回"""
        create_test_user(db)
        token = get_token_for_user(client)
        resp = client.get(f"{self.BASE}?period=2026-06", headers=auth_header(token))
        assert resp.status_code == 200
        assert "summary" in resp.json()


class TestCategoryMap:
    """科目→新30号准则板块映射"""

    BASE = "/api/cma/reports/category-map"

    def test_fallback_map(self, client: TestClient, db: Session):
        """无科目数据 → 硬编码映射回退"""
        create_test_user(db)
        token = get_token_for_user(client)
        resp = client.get(self.BASE, headers=auth_header(token))
        assert resp.status_code == 200
        data = resp.json()
        assert data["total"] >= 15
        assert data["mapping"]["6001"] == "operating"
        assert "operating" in data["grouped"]
        assert "tax" in data["grouped"]

    def test_subject_map(self, client: TestClient, db: Session):
        """有科目数据 → 用科目实际分类"""
        create_test_user(db)
        token = get_token_for_user(client)
        db.add(Subject(subject_code="6001", subject_name="主营业务收入",
                       new_standard_category="operating", is_active=1))
        db.add(Subject(subject_code="6801", subject_name="所得税费用",
                       new_standard_category="tax", is_active=1))
        db.add(Subject(subject_code="9999", subject_name="停用科目",
                       new_standard_category="operating", is_active=0))  # 不启用 → 排除
        db.commit()
        resp = client.get(self.BASE, headers=auth_header(token))
        assert resp.status_code == 200
        data = resp.json()
        codes = {m["subject_code"] for m in data["subjects"]}
        assert codes == {"6001", "6801"}


class TestBalanceSheet:
    """资产负债表（新30号准则）"""

    BASE = "/api/cma/reports/balance-sheet"

    def test_balance_sheet_demo(self, client: TestClient, db: Session):
        """无凭证数据 → 示例数据，勾稽平衡"""
        create_test_user(db)
        token = get_token_for_user(client)
        resp = client.get(f"{self.BASE}?period=2026-06", headers=auth_header(token))
        assert resp.status_code == 200
        data = resp.json()
        assert len(data["sections"]) == 5
        t = data["totals"]
        assert t["assets"]["end"] == t["liab_equity"]["end"]
        assert t["balanced"] is True
        assert data["all_items_have_data"] is False
        # 行项目含新准则分类
        first_section_lines = data["sections"][0]["lines"]
        assert all("ns_category" in l for l in first_section_lines)

    def test_balance_sheet_prev_period_boundary(self, client: TestClient, db: Session):
        """1月 → 期初期间跨年"""
        create_test_user(db)
        token = get_token_for_user(client)
        resp = client.get(f"{self.BASE}?period=2026-01", headers=auth_header(token))
        assert resp.status_code == 200
        assert resp.json()["prev_period"] == "2025-12"


class TestCashFlow:
    """现金流量表"""

    BASE = "/api/cma/reports/cash-flow"

    def test_cash_flow_demo(self, client: TestClient, db: Session):
        """无数据 → 示例数据三活动"""
        create_test_user(db)
        token = get_token_for_user(client)
        resp = client.get(f"{self.BASE}?period=2026-06", headers=auth_header(token))
        assert resp.status_code == 200
        data = resp.json()
        assert len(data["sections"]) == 3
        assert set(data["summary"].keys()) == {"net_increase", "begin_cash", "end_cash"}
        assert "fx_effect" in data

    def test_cash_flow_op_kpi_override(self, client: TestClient, db: Session):
        """F_OP_CFLOW 存在 → 经营净额被KPI覆盖"""
        create_test_user(db)
        token = get_token_for_user(client)
        k = mk_kpi(db, "F_OP_CFLOW")
        mk_val(db, k, "2026-06", 999.0)
        resp = client.get(f"{self.BASE}?period=2026-06", headers=auth_header(token))
        assert resp.status_code == 200
        data = resp.json()
        assert data["sections"][0]["net"] == 999.0

    def test_cash_flow_fx_kpi(self, client: TestClient, db: Session):
        """F_FX_LOSS 存在 → 汇率影响非demo"""
        create_test_user(db)
        token = get_token_for_user(client)
        k = mk_kpi(db, "F_FX_LOSS")
        mk_val(db, k, "2026-06", 2.5)
        resp = client.get(f"{self.BASE}?period=2026-06", headers=auth_header(token))
        assert resp.status_code == 200
        fx = resp.json()["fx_effect"]
        assert fx["value"] == 2.5
        assert fx["is_demo"] is False


class TestStatutory:
    """对外法定报表 组合视图 + 导出"""

    BASE = "/api/cma/reports/statutory"

    def test_statutory_combined(self, client: TestClient, db: Session):
        """三表合一"""
        create_test_user(db)
        token = get_token_for_user(client)
        seed_subject_kpis(db)
        resp = client.get(f"{self.BASE}?period=2026-06", headers=auth_header(token))
        assert resp.status_code == 200
        data = resp.json()
        assert "profit" in data and "balance_sheet" in data and "cash_flow" in data
        assert data["profit"]["net_profit"] == 25.0

    def test_statutory_export(self, client: TestClient, db: Session):
        """导出xlsx"""
        create_test_user(db)
        token = get_token_for_user(client)
        resp = client.get(f"{self.BASE}/export?period=2026-06", headers=auth_header(token))
        assert resp.status_code == 200
        assert "spreadsheetml" in resp.headers.get("content-type", "")
        assert len(resp.content) > 1000


class TestDupont:
    """杜邦分析"""

    BASE = "/api/cma/reports/dupont"

    def test_bohai_fallback(self, client: TestClient, db: Session):
        """博海无DB数据 → 回退文档常量"""
        create_test_user(db)
        token = get_token_for_user(client)
        resp = client.get(f"{self.BASE}?entity=bohai", headers=auth_header(token))
        assert resp.status_code == 200
        data = resp.json()
        assert data["entity_name"] == "陕西博海科技（IT服务）"
        assert data["roe"] > 0
        assert set(data["factors"].keys()) == {"net_profit_margin", "asset_turnover", "financial_leverage"}
        assert data["raw_data"]["net_profit"] == 14.13

    def test_bohai_with_db_data(self, client: TestClient, db: Session):
        """博海(entity 2) DB有verified KPI → 用DB值"""
        create_test_user(db)
        token = get_token_for_user(client)
        k1 = mk_kpi(db, "F_NET_PROFIT", entity_id=2, target=0)
        mk_val(db, k1, "2026-H1", 20.0)
        k2 = mk_kpi(db, "F_REVENUE", entity_id=2, target=0)
        mk_val(db, k2, "2026-H1", 400.0)
        resp = client.get(f"{self.BASE}?entity=bohai", headers=auth_header(token))
        assert resp.status_code == 200
        data = resp.json()
        assert data["raw_data"]["net_profit"] == 20.0
        assert data["raw_data"]["revenue"] == 400.0
        assert data["roe"] > 0

    def test_hanke_no_data(self, client: TestClient, db: Session):
        """酣客无数据 → roe None，不报错"""
        create_test_user(db)
        token = get_token_for_user(client)
        resp = client.get(f"{self.BASE}?entity=hanke", headers=auth_header(token))
        assert resp.status_code == 200
        data = resp.json()
        assert data["entity_name"] == "陕西酣客文化传媒（白酒经销）"
        assert data["roe"] is None
        assert data["factors"]["net_profit_margin"]["value"] is None

    def test_hanke_with_loss_data(self, client: TestClient, db: Session):
        """酣客亏损数据 → 负ROE"""
        create_test_user(db)
        token = get_token_for_user(client)
        k1 = mk_kpi(db, "F_REVENUE", entity_id=1, target=0)
        mk_val(db, k1, "2026-H1", 100.0)
        k2 = mk_kpi(db, "F_NET_PROFIT", entity_id=1, target=0)
        mk_val(db, k2, "2026-H1", -10.0)
        resp = client.get(f"{self.BASE}?entity=hanke", headers=auth_header(token))
        assert resp.status_code == 200
        data = resp.json()
        assert data["roe"] == -10.0
        assert data["factors"]["net_profit_margin"]["status"] == "🔴"

    def test_unknown_entity(self, client: TestClient, db: Session):
        """未知实体 → error"""
        create_test_user(db)
        token = get_token_for_user(client)
        resp = client.get(f"{self.BASE}?entity=xxx", headers=auth_header(token))
        assert resp.status_code == 200
        assert "error" in resp.json()


class TestGenerateReports:
    """自动报告生成（周报/月报/专项）+ 历史"""

    BASE = "/api/cma/reports"

    def _seed_kpis(self, db: Session):
        k1 = mk_kpi(db, "BH_REVENUE", target=100.0)
        mk_val(db, k1, "2026-06", 120.0)
        mk_val(db, k1, "2026-05", 100.0)   # 环比+20%
        k2 = mk_kpi(db, "BH_NET_PROFIT", target=50.0)
        mk_val(db, k2, "2026-06", 30.0)
        # 预警
        db.add(KPIAlert(kpi_id=k1.id, alert_level="red", alert_message="收入偏离目标", status="pending"))
        db.add(KPIAlert(kpi_id=k2.id, alert_level="yellow", alert_message="利润预警", status="pending"))
        # 改善行动
        db.add(ActionPlan(kpi_id=k1.id, title="提升毛利率", status="in_progress", progress=50))
        db.commit()
        return k1, k2

    def test_generate_invalid_type(self, client: TestClient, db: Session):
        """非法报告类型 → 400"""
        create_test_user(db)
        token = get_token_for_user(client)
        resp = client.post(f"{self.BASE}/generate", headers=auth_header(token),
                           json={"report_type": "daily"})
        assert resp.status_code == 400

    def test_generate_monthly(self, client: TestClient, db: Session):
        """月报生成 + 入库历史"""
        create_test_user(db)
        token = get_token_for_user(client)
        self._seed_kpis(db)
        resp = client.post(f"{self.BASE}/generate", headers=auth_header(token),
                           json={"report_type": "monthly", "period": "2026-06",
                                 "trigger_type": "manual"})
        assert resp.status_code == 200
        data = resp.json()
        assert data["report_type"] == "monthly"
        assert data["id"] > 0
        assert "经营分析月报" in data["title"]
        assert "经营分析月报" in data["markdown"]
        assert data["json"]["overview"]["red_alerts"] == 1
        assert len(data["json"]["budget_execution"]) >= 0

    def test_generate_weekly(self, client: TestClient, db: Session):
        """周报生成（ISO周期间）"""
        create_test_user(db)
        token = get_token_for_user(client)
        self._seed_kpis(db)
        resp = client.post(f"{self.BASE}/generate", headers=auth_header(token),
                           json={"report_type": "weekly", "period": "2026-W30"})
        assert resp.status_code == 200
        data = resp.json()
        assert data["report_type"] == "weekly"
        assert "经营分析周报" in data["markdown"]
        assert len(data["json"]["top_changes"]) >= 1   # 有环比数据

    def test_generate_special_with_alert(self, client: TestClient, db: Session):
        """专项报告（事件触发）"""
        create_test_user(db)
        token = get_token_for_user(client)
        k1, _ = self._seed_kpis(db)
        alert = db.query(KPIAlert).filter_by(kpi_id=k1.id).first()
        resp = client.post(f"{self.BASE}/generate", headers=auth_header(token),
                           json={"report_type": "special", "period": "2026-06",
                                 "trigger_type": "event", "alert_ref": str(alert.id)})
        assert resp.status_code == 200
        data = resp.json()
        assert data["report_type"] == "special"
        assert "专项报告" in data["title"]
        assert data["json"]["focus_kpi"] == "BH_REVENUE"
        assert data["json"]["alert_ref"] == str(alert.id)

    def test_generate_special_no_alerts(self, client: TestClient, db: Session):
        """专项报告（无预警数据）"""
        create_test_user(db)
        token = get_token_for_user(client)
        resp = client.post(f"{self.BASE}/generate", headers=auth_header(token),
                           json={"report_type": "special", "period": "2026-06"})
        assert resp.status_code == 200
        assert resp.json()["json"]["focus_kpi"] is None

    def test_history_empty_then_generated(self, client: TestClient, db: Session):
        """历史：空 → 生成后有条目 → 详情可取"""
        create_test_user(db)
        token = get_token_for_user(client)
        # 空
        r = client.get(f"{self.BASE}/history", headers=auth_header(token))
        assert r.status_code == 200 and r.json()["total"] == 0
        # 生成
        self._seed_kpis(db)
        gen = client.post(f"{self.BASE}/generate", headers=auth_header(token),
                          json={"report_type": "monthly", "period": "2026-06"})
        rid = gen.json()["id"]
        # 历史列表
        r = client.get(f"{self.BASE}/history", headers=auth_header(token))
        assert r.json()["total"] >= 1
        assert r.json()["data"][0]["id"] == rid
        # 类型过滤
        r = client.get(f"{self.BASE}/history?report_type=monthly", headers=auth_header(token))
        assert all(x["report_type"] == "monthly" for x in r.json()["data"])
        # 详情
        r = client.get(f"{self.BASE}/history/{rid}", headers=auth_header(token))
        assert r.status_code == 200
        assert "markdown" in r.json()
        # 404
        r = client.get(f"{self.BASE}/history/99999", headers=auth_header(token))
        assert r.status_code == 404


class TestReportsPermissions:
    """权限：报表中心仅 ceo/finance/business 可访问"""

    BASE = "/api/cma/reports"

    def test_it_role_denied(self, client: TestClient, db: Session):
        """it角色访问报表 → 403"""
        from app.models import User
        import hashlib
        it = User(username="it_reports", password_hash=hashlib.sha256("pass123".encode()).hexdigest(),
                  name="IT运维", role="it")
        db.add(it)
        db.commit()
        token = get_token_for_user(client, username="it_reports", password="pass123")
        resp = client.get(f"{self.BASE}/profit-summary", headers=auth_header(token))
        assert resp.status_code == 403

    def test_no_token_denied(self, client: TestClient, db: Session):
        """无token → 403（router级 require_role 直接拒绝）"""
        resp = client.get(f"{self.BASE}/profit-summary")
        assert resp.status_code == 403