cma-management/backend/tests/test_ar_aging_overdue_20261020.py

# -*- coding: utf-8 -*-
"""应收账龄看板 + 逾期预警（T1/T2 · 2026-10-20 R2）单测

任务：`project-20261020-ar-overdue-alert-and-aging-board-fullstack`（阶段A）

口径与边界（全部为硬断言）：
  · 六档互斥：逐行 `b_0_30+…+b_150p == total_amt`；
  · 逾期额 `o_i = total_amt − b_0_30`，负档位**不** ABS/COALESCE（DQ-AR-04 杨创 -20）；
  · 三口径分列、禁跨口径合计（`caliber=all` → 400）；
  · 缺期（2026-10）⇒ `has_data=false` 且金额 null（禁补 0）；
  · §6 裁决：`red.share_90plus_gt=null` ⇒ 90 天占比红线停用（不回落 0.25）；
    新增「单户 >30 天且 >5 万」逐户红线 + 「150+ 占比环比上升」结构恶化红线；
  · §6.2 assignee：手工点名 → 近 3 个月主业务员（employee_dict / 同库摘要降级）
    → 部门负责人（渠道 118 / 商务 114 / 直属酒窖 117）；两路皆失败 ⇒ skipped[]，不落空行；
  · 阈值/责任人全部走 `alert_rules.params`（本套件在本地 SQLite 夹具里改参数复跑，
    证明「改参数不改代码」）；
  · 写通道只 INSERT，历史行零 UPDATE/DELETE，同 `alert_message` 幂等跳过。

本套件**不连任何生产库**：`tests/conftest.py` 已把引擎替换为内存 SQLite；
`hanke_ar_aging` 非 ORM 模型，故用原始 DDL 在夹具里建表。
"""
from __future__ import annotations

import json

import pytest
from sqlalchemy import text
from sqlalchemy.orm import Session

from app.api.alert_rules import AlertRule
from app.models import KPIAlert
from app.services import ar_overdue_alert
from app.utils import ar_caliber
from tests.conftest import create_test_kpi, create_test_user, get_token_for_user, auth_header

ENT = 1
DDL = """
CREATE TABLE hanke_ar_aging (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    entity_id INTEGER NOT NULL,
    period VARCHAR(7) NOT NULL,
    caliber VARCHAR(24) NOT NULL,
    unit_code VARCHAR(20) NOT NULL,
    unit_name VARCHAR(100) NOT NULL,
    opening_amt NUMERIC(18,2) NOT NULL DEFAULT 0,
    total_amt NUMERIC(18,2) NOT NULL DEFAULT 0,
    b_0_30 NUMERIC(18,2) NOT NULL DEFAULT 0,
    b_31_60 NUMERIC(18,2) NOT NULL DEFAULT 0,
    b_61_90 NUMERIC(18,2) NOT NULL DEFAULT 0,
    b_90_120 NUMERIC(18,2) NOT NULL DEFAULT 0,
    b_120_150 NUMERIC(18,2) NOT NULL DEFAULT 0,
    b_150p NUMERIC(18,2) NOT NULL DEFAULT 0,
    source_file VARCHAR(160) NOT NULL DEFAULT 'fixture.csv',
    source_sha256 VARCHAR(64) NOT NULL DEFAULT 'fixture',
    loaded_at DATETIME DEFAULT CURRENT_TIMESTAMP
)
"""

# §6.1 参数最终形状（含 aggregate_assignee，仅测试用显式给值以验证聚合可落库）
BASE_PARAMS = {
    "kpi_code": "F_AR_002",
    "caliber": ar_caliber.CALIBER_DEBIT,
    "red": {"share_90plus_gt": None, "single_overdue_amt_gt": 50000,
            "single_overdue_min_days": 30, "count_mom_pct_gte": 0.20,
            "structure_worsen": True},
    "yellow": {"share_90plus_min": 0.15, "share_90plus_max": 0.25,
               "single_overdue_amt_min": 10000, "single_overdue_amt_max": 50000},
    "assignee_source": {
        "mode": "salesman_then_dept",
        "lookback_months": 3,
        "dept_fallback": {"渠道": 118, "商务": 114, "直属酒窖": 117},
        "dept_by_unit": {"U3": "商务"},
        "aggregate_assignee": "财务应收会计",
    },
    "assignee_by_unit": {},
    "exclude_unit_codes": [],
    "action_texts": {"red": "催收+停供评估", "yellow": "对账+催收"},
}

ZXT_DDL = """
CREATE TABLE zxt_masterbill (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    entity_id INTEGER,
    unit_id INTEGER,
    emp_id INTEGER,
    bill_type INTEGER DEFAULT 1,
    bill_date DATE,
    sum_money NUMERIC(18,2) DEFAULT 0,
    abst VARCHAR(500)
)
"""


def _reset_table(db: Session) -> None:
    db.execute(text("DROP TABLE IF EXISTS hanke_ar_aging"))
    db.execute(text(DDL))
    db.commit()


def _add(db: Session, period: str, code: str, name: str, total: float, buckets: list,
         opening: float = 0.0, caliber: str = ar_caliber.CALIBER_DEBIT, entity_id: int = ENT) -> None:
    assert abs(sum(buckets) - total) < 0.005, "夹具必须满足六档和 = 合计"
    db.execute(text(
        "INSERT INTO hanke_ar_aging (entity_id, period, caliber, unit_code, unit_name, opening_amt,"
        " total_amt, b_0_30, b_31_60, b_61_90, b_90_120, b_120_150, b_150p)"
        " VALUES (:e,:p,:c,:code,:name,:op,:tot,:b0,:b1,:b2,:b3,:b4,:b5)"
    ), {"e": entity_id, "p": period, "c": caliber, "code": code, "name": name, "op": opening,
        "tot": total, "b0": buckets[0], "b1": buckets[1], "b2": buckets[2],
        "b3": buckets[3], "b4": buckets[4], "b5": buckets[5]})
    db.commit()


@pytest.fixture()
def aging(db: Session):
    """两期夹具（2026-08 / 2026-09），含负档位行 + 未逾期户 + 贷方口径干扰行。

    2026-09：合计 173,159；90+ = 63,179（36.48%）；逾期户 = U1/U3/U4 共 3 户。
    2026-08：合计 163,159；90+ = 43,179（26.46%）；逾期户 3 户 ⇒ 环比 0%。
    150+ 占比：2026-08 = 14.21% → 2026-09 = 16.27%（上升 ⇒ 结构恶化红）。
    """
    _reset_table(db)
    _add(db, "2026-09", "U1", "甲公司", 100000, [20000, 10000, 10000, 20000, 15000, 25000], opening=5000)
    _add(db, "2026-09", "U2", "乙公司", 40000, [40000, 0, 0, 0, 0, 0])
    _add(db, "2026-09", "U3", "丙公司", 30000, [0, 30000, 0, 0, 0, 0])
    # 负档位（DQ-AR-04 同形）：total=3159, b_0_30=-20, b_150p=3179 ⇒ o_i=3179（不得 ABS 抹成 3139/3159）
    _add(db, "2026-09", "U4", "杨创", 3159, [-20, 0, 0, 0, 0, 3179])
    # 干扰行：若查询漏 caliber 过滤 → 会污染借方口径合计
    _add(db, "2026-09", "C1", "贷方户", -34010, [0, 0, 0, 0, 0, -34010],
         caliber=ar_caliber.CALIBER_CREDIT_SEPARATE)
    _add(db, "2026-08", "U1", "甲公司", 90000, [30000, 10000, 10000, 10000, 10000, 20000])
    _add(db, "2026-08", "U2", "乙公司", 50000, [50000, 0, 0, 0, 0, 0])
    _add(db, "2026-08", "U3", "丙公司", 20000, [0, 20000, 0, 0, 0, 0])
    _add(db, "2026-08", "U4", "杨创", 3159, [-20, 0, 0, 0, 0, 3179])
    return db


def _staff(db: Session, *, with_employee_dict: bool = True, with_units_dict: bool = True,
           with_zxt: bool = True) -> None:
    """维度夹具：units_dict(U1/U3/U4) + employee_dict(14/32/114) + zxt_masterbill。"""
    for t in ("units_dict", "employee_dict", "zxt_masterbill"):
        db.execute(text("DROP TABLE IF EXISTS %s" % t))
    if with_units_dict:
        db.execute(text("CREATE TABLE units_dict (unit_code VARCHAR(20), unit_id INTEGER)"))
        for code, uid in (("U1", 101), ("U3", 103), ("U4", 104)):
            db.execute(text("INSERT INTO units_dict VALUES (:c,:u)"), {"c": code, "u": uid})
    if with_employee_dict:
        db.execute(text("CREATE TABLE employee_dict (id INTEGER, u_Name VARCHAR(64))"))
        for i, n in ((14, "李世昌"), (32, "兀晨博"), (114, "王经理"), (118, "渠道负责人"),
                     (117, "酒窖负责人")):
            db.execute(text("INSERT INTO employee_dict VALUES (:i,:n)"), {"i": i, "n": n})
    if with_zxt:
        db.execute(text(ZXT_DDL))
        rows = [
            (ENT, 101, 14, "2026-08-10", "向【甲公司】出售商品：李世昌"),
            (ENT, 101, 14, "2026-08-11", "向【甲公司】出售商品：李世昌"),
            (ENT, 101, 14, "2026-09-05", "向【甲公司】出售商品：李世昌"),
            (ENT, 101, 32, "2026-09-20", "向【甲公司】出售商品：兀晨博"),
        ]
        for e, u, m, d, a in rows:
            db.execute(text("INSERT INTO zxt_masterbill (entity_id, unit_id, emp_id, bill_date, abst)"
                            " VALUES (:e,:u,:m,:d,:a)"),
                       {"e": e, "u": u, "m": m, "d": d, "a": a})
    db.commit()


@pytest.fixture()
def staff(aging: Session):
    _staff(aging)
    return aging


# ─────────────────────── 数据口径 ───────────────────────

def test_six_buckets_equal_total_per_row(aging: Session):
    rows = aging.execute(text(
        "SELECT period, unit_code, total_amt, b_0_30+b_31_60+b_61_90+b_90_120+b_120_150+b_150p AS s"
        " FROM hanke_ar_aging WHERE entity_id=1 AND caliber=:c"
    ), {"c": ar_caliber.CALIBER_DEBIT}).mappings().all()
    assert len(rows) == 8
    for r in rows:
        assert abs(float(r["s"]) - float(r["total_amt"])) < 0.005, r


def test_caliber_literals_are_exact():
    assert ar_caliber.CALIBERS == ("全户口径", "借方应收户口径", "贷方户单列(净额)")
    with pytest.raises(ar_caliber.CaliberError):
        ar_caliber.validate_caliber("贷方户单列")  # 少「(净额)」⇒ 会漏 42 行
    with pytest.raises(ar_caliber.CaliberError):
        ar_caliber.validate_caliber("all")
    assert ar_caliber.validate_caliber(None, allow_none=True) == ar_caliber.CALIBER_DEBIT


# ─────────────────────── 计算与分级（§6 阈值） ───────────────────────

def test_share_90plus_red_disabled_and_structure_worsen(staff: Session):
    """§6.1：90 天占比红线置 null ⇒ 不生成该聚合项；150+ 环比上升 ⇒ 结构恶化红。"""
    res = ar_overdue_alert.compute(staff, ENT, "2026-09", BASE_PARAMS)
    assert res["skipped_reason"] is None
    m = res["metrics"]
    assert m["total_amt"] == pytest.approx(173159.0, abs=0.01)      # 贷方口径行未被混入
    assert m["overdue_90plus"] == pytest.approx(63179.0, abs=0.01)
    assert m["share_90plus"] == pytest.approx(63179 / 173159, rel=1e-6)
    assert m["share_150p"] == pytest.approx(28179 / 173159, abs=1e-6)   # metrics 保留 6 位
    assert m["prev_share_150p"] == pytest.approx(23179 / 163159, abs=1e-6)

    keys = [a["key"] for a in res["aggregates"]]
    assert "share_90plus" not in keys, "红线停用后 36.48% 不应再出聚合项（无区分度）"
    assert "structure_150p" in keys
    agg = [a for a in res["aggregates"] if a["key"] == "structure_150p"][0]
    assert agg["level"] == "red"
    assert agg["assignee"] == "财务应收会计"
    for token in ("2026-09", "150+ 档占比", "16.27%", "14.21%", "结构恶化", "催收+停供评估"):
        assert token in agg["alert_message"], token
    assert len(agg["alert_message"]) <= 500
    assert any("90 天占比红线停用" in n for n in res["notes"])


def test_single_overdue_red_and_yellow_households(staff: Session):
    res = ar_overdue_alert.compute(staff, ENT, "2026-09", BASE_PARAMS)
    by_code = {c["unit_code"]: c for c in res["per_customer"]}
    # 单户 >30 天且 >5 万 ⇒ 红
    assert by_code["U1"]["level"] == "red"
    assert by_code["U1"]["overdue_amt"] == pytest.approx(80000.0)
    assert by_code["U1"]["oldest_days"] == 151
    # 单户 1–5 万 ⇒ 黄
    assert by_code["U3"]["level"] == "yellow" and by_code["U3"]["oldest_days"] == 31
    # U4 逾期 3,179 < 1 万 ⇒ 不生成；U2 全额 0-30 ⇒ 不生成
    assert "U4" not in by_code and "U2" not in by_code
    for c in res["per_customer"]:
        assert c["assignee"], "禁落空 assignee"
        assert c["alert_message"].count("｜建议：") == 1
        assert "｜责任人：" in c["alert_message"]


def test_assignee_chain_salesman_then_dept(staff: Session):
    res = ar_overdue_alert.compute(staff, ENT, "2026-09", BASE_PARAMS)
    by_code = {c["unit_code"]: c for c in res["per_customer"]}
    # U1：units_dict → unit_id=101 → 主业务员 emp14 单据最多（3 vs 1）→ 李世昌
    assert by_code["U1"]["assignee"] == "李世昌"
    assert "salesman" in by_code["U1"]["assignee_source"]
    assert by_code["U1"]["emp_id"] == 14
    # U3：近 3 个月无单据 → 部门负责人 商务 114 → 王经理
    assert by_code["U3"]["assignee"] == "王经理"
    assert by_code["U3"]["assignee_source"].startswith("dept:商务")


def test_assignee_manual_map_has_priority(staff: Session):
    p = json.loads(json.dumps(BASE_PARAMS))
    p["assignee_by_unit"] = {"U1": "财务点名人甲"}
    res = ar_overdue_alert.compute(staff, ENT, "2026-09", p)
    by_code = {c["unit_code"]: c for c in res["per_customer"]}
    assert by_code["U1"]["assignee"] == "财务点名人甲"
    assert by_code["U1"]["assignee_source"] == "manual:assignee_by_unit"


def test_assignee_abst_fallback_when_employee_dict_absent(aging: Session):
    """employee_dict 缺表 ⇒ 同库单据摘要尾段制单人姓名降级（仍非空、可审计）。"""
    _staff(aging, with_employee_dict=False)
    res = ar_overdue_alert.compute(aging, ENT, "2026-09", BASE_PARAMS)
    by_code = {c["unit_code"]: c for c in res["per_customer"]}
    assert by_code["U1"]["assignee"] == "李世昌"
    assert "abst_operator" in by_code["U1"]["assignee_source"]


def test_assignee_unresolved_goes_to_skipped(aging: Session):
    """维度表全缺 + 部门无法归类 ⇒ 该户不生成，skipped[] 写原因（禁落空 assignee）。"""
    _staff(aging, with_employee_dict=False, with_units_dict=False, with_zxt=False)
    p = json.loads(json.dumps(BASE_PARAMS))
    p["assignee_source"].pop("dept_by_unit")
    res = ar_overdue_alert.compute(aging, ENT, "2026-09", p)
    assert res["per_customer"] == []
    skipped = {s["unit_code"]: s for s in res["skipped"] if s["scope"] == "customer"}
    assert set(skipped) == {"U1", "U3"}
    assert "unit_link_unresolved" in skipped["U1"]["reason"]
    assert "dept_unclassified" in skipped["U1"]["reason"]
    assert [h for h in ar_overdue_alert.hits(res) if h.get("unit_code")] == []


def test_assignee_missing_is_fail_closed(aging: Session):
    p = {"red": {}, "yellow": {}, "assignee_default": "", "assignee_by_unit": {}}
    with pytest.raises(ar_overdue_alert.AssigneeMissing):
        ar_overdue_alert.ensure_assignee_configured(p)
    # 配了逐户派生模式即视为已配置（派生失败按户 skipped，不整体 400）
    p["assignee_source"] = {"mode": "salesman_then_dept"}
    ar_overdue_alert.ensure_assignee_configured(p)


def test_level_boundaries_are_param_driven(aging: Session):
    """改 DB 参数复跑 ⇒ 命中集合变化，代码零改动（§6.3 第 11 项：single_overdue_amt_gt）。"""
    _staff(aging)
    p = json.loads(json.dumps(BASE_PARAMS))
    assert {c["unit_code"] for c in ar_overdue_alert.compute(aging, ENT, "2026-09", p)["per_customer"]} == {"U1", "U3"}
    p["red"]["single_overdue_amt_gt"] = 300000          # 5 万 → 30 万 ⇒ U1(8 万) 掉出红线且高于黄线上限
    res = ar_overdue_alert.compute(aging, ENT, "2026-09", p)
    assert {c["unit_code"] for c in res["per_customer"]} == {"U3"}
    p["red"]["single_overdue_amt_gt"] = 50000
    p["yellow"]["single_overdue_amt_min"] = 100000      # 黄线下限 10 万 ⇒ U3 消失
    assert {c["unit_code"] for c in ar_overdue_alert.compute(aging, ENT, "2026-09", p)["per_customer"]} == {"U1"}


def test_missing_period_is_skipped_not_zero(aging: Session):
    res = ar_overdue_alert.compute(aging, ENT, "2026-10", BASE_PARAMS)
    assert res["skipped_reason"] == "no_data_for_period:2026-10"
    assert res["aggregates"] == [] and res["per_customer"] == []


def test_default_period_is_max_and_mom_computed(staff: Session):
    res = ar_overdue_alert.compute(staff, ENT, None, BASE_PARAMS)
    assert res["period"] == "2026-09"
    assert res["metrics"]["prev_period"] == "2026-08"
    assert res["metrics"]["prev_overdue_customers"] == 3
    assert res["metrics"]["overdue_customers"] == 3
    assert res["metrics"]["count_mom_pct"] == pytest.approx(0.0)
    assert "count_mom" not in {a["key"] for a in res["aggregates"]}


def test_no_previous_period_no_mom(aging: Session):
    res = ar_overdue_alert.compute(aging, ENT, "2026-08", BASE_PARAMS)
    assert res["metrics"]["prev_period"] is None
    assert res["metrics"]["count_mom_pct"] is None
    assert any("不计算环比" in n for n in res["notes"])


def test_exclude_unit_codes(aging: Session):
    _staff(aging)
    p = json.loads(json.dumps(BASE_PARAMS))
    p["exclude_unit_codes"] = ["U1"]
    res = ar_overdue_alert.compute(aging, ENT, "2026-09", p)
    assert "U1" not in {c["unit_code"] for c in res["per_customer"]}
    assert res["metrics"]["total_amt"] < 173159


# ─────────────────────── 落库（只 INSERT） ───────────────────────

def test_write_alerts_idempotent_and_history_intact(staff: Session):
    kpi = create_test_kpi(staff, kpi_code="F_AR_002", kpi_name="逾期应收账款率")
    hist = KPIAlert(entity_id=ENT, kpi_id=kpi.id, alert_level="red",
                    alert_message="历史行-不得被改写", alert_type="actual", status="pending",
                    assignee="old-owner")
    staff.add(hist)
    staff.commit()
    hist_id = hist.id

    res = ar_overdue_alert.compute(staff, ENT, "2026-09", BASE_PARAMS)
    first = ar_overdue_alert.write_alerts(staff, ENT, kpi.id, res)
    assert first["inserted"] == len(ar_overdue_alert.hits(res)) > 0
    second = ar_overdue_alert.write_alerts(staff, ENT, kpi.id, res)
    assert second == {"inserted": 0, "skipped_duplicate": first["inserted"]}

    rows = staff.query(KPIAlert).filter(KPIAlert.alert_type == "overdue").all()
    assert len(rows) == first["inserted"]
    assert all(r.assignee and r.status == "pending" and r.kpi_value_id is None for r in rows)
    assert all(len(r.alert_message) <= 500 for r in rows)
    old = staff.get(KPIAlert, hist_id)
    assert old.alert_message == "历史行-不得被改写" and old.assignee == "old-owner"


# ─────────────────────── HTTP 面（TestClient + 内存库） ───────────────────────

@pytest.fixture()
def fin_headers(client, db):
    create_test_user(db, username="fin_probe", role="finance")
    return auth_header(get_token_for_user(client, username="fin_probe"))


def test_api_meta_trend_overview(staff, client, fin_headers):
    r = client.get("/api/cma/ar-aging/meta", headers=fin_headers)
    assert r.status_code == 200, r.text
    body = r.json()
    assert body["available"] is True
    assert body["calibers"] == list(ar_caliber.CALIBERS)
    assert body["latest_period"] == "2026-09"
    assert body["row_count"] == 9                      # 8 借方 + 1 贷方（三口径不混算，仅计数）

    r = client.get("/api/cma/ar-aging/trend", headers=fin_headers,
                   params={"caliber": ar_caliber.CALIBER_DEBIT})
    assert r.status_code == 200
    pts = {p["period"]: p for p in r.json()["points"]}
    assert pts["2026-09"]["has_data"] is True
    assert pts["2026-10"]["has_data"] is False
    assert pts["2026-10"]["total_amt"] is None and pts["2026-10"]["customers"] is None

    r = client.get("/api/cma/ar-aging/overview", headers=fin_headers,
                   params={"period": "2026-09", "caliber": ar_caliber.CALIBER_DEBIT})
    ov = r.json()
    assert ov["has_data"] is True
    assert ov["buckets_sum_matches_total"] is True
    assert ov["opening_amt"] == pytest.approx(5000.0)
    assert ov["opening_amt_non_aging_base"] is True
    assert ov["total_amt"] == pytest.approx(173159.0, abs=0.01)


def test_api_caliber_all_rejected(aging, client, fin_headers):
    for ep in ("overview", "trend", "customers"):
        r = client.get(f"/api/cma/ar-aging/{ep}", headers=fin_headers, params={"caliber": "all"})
        assert r.status_code == 400, (ep, r.text)
        assert "禁止跨口径合计" in r.json()["detail"]


def test_api_customers_sort_and_export(aging, client, fin_headers):
    r = client.get("/api/cma/ar-aging/customers", headers=fin_headers,
                   params={"period": "2026-09", "caliber": ar_caliber.CALIBER_DEBIT,
                           "overdue_only": "true", "sort": "overdue_amt", "order": "desc"})
    items = r.json()["items"]
    assert [i["unit_code"] for i in items][:3] == ["U1", "U3", "U4"]
    assert items[0]["overdue_amt"] == pytest.approx(80000.0)

    r = client.get("/api/cma/ar-aging/customers/export", headers=fin_headers,
                   params={"period": "2026-09", "caliber": ar_caliber.CALIBER_DEBIT})
    assert r.status_code == 200
    assert r.content[:3] == b"\xef\xbb\xbf"                     # UTF-8 BOM（Excel 直开）
    lines = r.content.decode("utf-8-sig").splitlines()
    assert lines[0].startswith("截止期,2026-09,口径,借方应收户口径,导出时刻,")
    assert "U1" in lines[2] or "U1" in lines[3]


def test_api_run_ar_overdue_rule_params_parameterized(staff, client, db, fin_headers):
    """端到端：规则参数来自 DB 行 ⇒ 改 `single_overdue_amt_gt` 复跑命中变化，代码零改动。"""
    kpi = create_test_kpi(db, kpi_code="F_AR_002", kpi_name="逾期应收账款率")
    rule = AlertRule(entity_id=ENT, kpi_id=kpi.id, rule_type=ar_overdue_alert.RULE_TYPE,
                     trigger_on="actual", enabled=1, params=BASE_PARAMS)
    db.add(rule)
    db.commit()

    r = client.post("/api/cma/alert-rules/run-ar-overdue",
                    headers=fin_headers, params={"period": "2026-09", "dry_run": "true"})
    assert r.status_code == 200, r.text
    j = r.json()
    assert j["params_source"] == "alert_rules#%s" % rule.id
    assert j["kpi_id"] == kpi.id
    assert j["write_blockers"] == []
    assert j["hit_count"] == 3                       # 结构恶化聚合红 + U1 红 + U3 黄
    assert {c["unit_code"] for c in j["per_customer"]} == {"U1", "U3"}
    assert all(c["assignee"] for c in j["per_customer"])
    assert j["skipped"] == []
    assert "structure_150p" in {a["key"] for a in j["aggregates"]}

    params2 = json.loads(json.dumps(BASE_PARAMS))
    params2["red"]["single_overdue_amt_gt"] = 300000   # 5 万 → 30 万
    rule.params = params2
    db.commit()
    r = client.post("/api/cma/alert-rules/run-ar-overdue",
                    headers=fin_headers, params={"period": "2026-09", "dry_run": "true"})
    assert r.status_code == 200
    j2 = r.json()
    assert {c["unit_code"] for c in j2["per_customer"]} == {"U3"}
    assert j2["hit_count"] == 2 and j2["hit_count"] < j["hit_count"]   # 聚合红 + U3 黄


def test_api_run_ar_overdue_assignee_empty_400(aging, client, db, fin_headers):
    kpi = create_test_kpi(db, kpi_code="F_AR_002x", kpi_name="逾期应收账款率")
    p = {"red": {}, "yellow": {}, "assignee_default": "", "assignee_by_unit": {}}
    db.add(AlertRule(entity_id=ENT, kpi_id=kpi.id, rule_type=ar_overdue_alert.RULE_TYPE,
                     trigger_on="actual", enabled=1, params=p))
    db.commit()
    r = client.post("/api/cma/alert-rules/run-ar-overdue",
                    headers=fin_headers, params={"period": "2026-09", "dry_run": "true"})
    assert r.status_code == 400
    assert "责任人" in r.json()["detail"]

    # dry_run=false + 参数覆盖 → 400（防绕过 DB 配置）
    r = client.post("/api/cma/alert-rules/run-ar-overdue", headers=fin_headers,
                    params={"period": "2026-09", "dry_run": "false"},
                    json={"params": BASE_PARAMS})
    assert r.status_code == 400
    assert "禁止参数/KPI 覆盖" in r.json()["detail"]


def test_api_meta_degrades_when_table_missing(client, db, fin_headers, monkeypatch):
    _reset_table(db)
    db.execute(text("DROP TABLE hanke_ar_aging"))
    db.commit()
    import app.api.ar_aging as aa
    monkeypatch.setattr(aa, "_TABLE", "hanke_ar_aging_missing_zzz")
    r = client.get("/api/cma/ar-aging/meta", headers=fin_headers)
    assert r.status_code == 200
    body = r.json()
    assert body["available"] is False and "不可用" in body["reason"]
    r = client.get("/api/cma/ar-aging/trend", headers=fin_headers)
    assert r.status_code == 200 and r.json()["available"] is False