cma-management/backend/tests/test_maps_review_hold_20261022.py

"""P2 回归：GET /api/cma/maps/{id}/review?level=strategic 不再 500

任务 project-20261022-cma-map-canvas-connections-not-drawn-fullstack（缺陷2）
根因：maps.py 的 total_hold 只在「目标含 DATA_HOLD_KPIS」分支被 += 累加、从未初始化，
      首个待核目标即 UnboundLocalError → 500（线上 map65 栈顶实测）。
修复：初始化 total_hold = 0，并在 summary.hold 透出计数（纯增量，消费方 green/yellow/red/health_score 不受影响）。

覆盖：
  ① 含待核 KPI 的地图 → 200，summary.hold=1，该目标 data_hold=True 且 actual 不外泄
  ② 普通地图 → 200，summary.hold=0
  ③ dimensions 缺失（None）/ 空数组 → 200，不抛（total=0，health_score=0）
"""
import pytest
from fastapi.testclient import TestClient
from sqlalchemy.orm import Session

from app.models import StrategicMap, MapObjective, KPIDefinition, KPIValue
from app.api.maps import DATA_HOLD_KPIS
from tests.conftest import create_test_user, get_token_for_user, auth_header, create_test_kpi

HOLD_CODE = "C_CUST_CONCENTRATION"
PLAIN_CODE = "TEST_REVIEW_PLAIN"


def _seed_review_map(db: Session, dims, title="P2 review 回归", map_id=None) -> StrategicMap:
    kwargs = dict(entity_id=1, title=title, status="active", dimensions=dims,
                  canvas_data={"connections": []})
    if map_id is not None:
        kwargs["id"] = map_id
    m = StrategicMap(**kwargs)
    db.add(m)
    db.commit()
    db.refresh(m)
    return m


@pytest.fixture(autouse=True)
def _clean(db: Session):
    yield
    db.rollback()
    db.query(KPIValue).filter(KPIValue.kpi_id.in_(
        [k.id for k in db.query(KPIDefinition).filter(
            KPIDefinition.kpi_code.in_([HOLD_CODE, PLAIN_CODE])).all()]
    )).delete(synchronize_session=False)
    db.query(KPIDefinition).filter(KPIDefinition.kpi_code.in_([HOLD_CODE, PLAIN_CODE])).delete(
        synchronize_session=False)
    db.query(MapObjective).filter(MapObjective.map_id >= 900000).delete(synchronize_session=False)
    db.query(StrategicMap).filter(StrategicMap.id >= 900000).delete(synchronize_session=False)
    db.commit()


def _hold_dims():
    # 一个待核目标（含 C_CUST_CONCENTRATION）+ 一个普通目标
    return [{
        "key": "customer", "name": "客户层", "icon": "👥", "color": "#409EFF",
        "objectives": [
            {"name": "客户集中度", "kpis": [HOLD_CODE]},
            {"name": "客户满意", "kpis": [PLAIN_CODE]},
        ],
    }]


def _seed_kpis(db: Session):
    hold = create_test_kpi(db, kpi_code=HOLD_CODE, kpi_name="客户集中度", dimension="customer",
                           target_value=60.0, kpi_level="strategic", unit="%")
    plain = create_test_kpi(db, kpi_code=PLAIN_CODE, kpi_name="客户满意", dimension="customer",
                            target_value=90.0, kpi_level="strategic", unit="分")
    return hold, plain


class TestReviewNo500:
    def test_data_hold_kpi_returns_200_and_counts_hold(self, client: TestClient, db: Session):
        """含待核 KPI：修复前此处 UnboundLocalError → 500；修复后 200 且 summary.hold=1。"""
        create_test_user(db)
        token = get_token_for_user(client)
        hold_kpi, plain_kpi = _seed_kpis(db)
        db.add(KPIValue(kpi_id=hold_kpi.id, period="2026-09", actual_value=72.0, source_type="actual"))
        db.add(KPIValue(kpi_id=plain_kpi.id, period="2026-09", actual_value=93.0, source_type="actual"))
        db.commit()
        m = _seed_review_map(db, _hold_dims(), map_id=900101)

        resp = client.get(f"/api/cma/maps/{m.id}/review", params={"level": "strategic"},
                          headers=auth_header(token))
        assert resp.status_code == 200, resp.text
        body = resp.json()
        assert body["map_id"] == m.id
        s = body["summary"]
        assert s["hold"] == 1, s
        assert s["total"] == 2, s
        assert s["green"] == 1, s               # 待核目标不进任何档
        assert isinstance(s["health_score"], (int, float))

        objs = {o["name"]: o for o in body["dimensions"][0]["objectives"]}
        held = objs["客户集中度"]
        assert held["data_hold"] is True and held["hold_reason"]
        assert held["level"] == "pending"
        # 数据前置：待核目标的实测值不得外泄
        assert all(k["actual_value"] is None for k in held["kpis"])
        assert objs["客户满意"]["data_hold"] is False

    def test_plain_map_returns_200_hold_zero(self, client: TestClient, db: Session):
        create_test_user(db)
        token = get_token_for_user(client)
        _seed_kpis(db)
        m = _seed_review_map(db, _hold_dims(), title="P2 普通地图", map_id=900102)
        # 去掉待核目标，只留普通目标
        m.dimensions = [{"key": "customer", "name": "客户层", "icon": "", "color": "#409EFF",
                         "objectives": [{"name": "客户满意", "kpis": [PLAIN_CODE]}]}]
        db.commit()

        resp = client.get(f"/api/cma/maps/{m.id}/review", params={"level": "strategic"},
                          headers=auth_header(token))
        assert resp.status_code == 200, resp.text
        assert resp.json()["summary"]["hold"] == 0

    @pytest.mark.parametrize("dims", [None, []])
    def test_empty_dimensions_returns_200(self, client: TestClient, db: Session, dims):
        create_test_user(db)
        token = get_token_for_user(client)
        m = _seed_review_map(db, dims, title="P2 空地图", map_id=900103)
        resp = client.get(f"/api/cma/maps/{m.id}/review", params={"level": "strategic"},
                          headers=auth_header(token))
        assert resp.status_code == 200, resp.text
        s = resp.json()["summary"]
        assert s["total"] == 0 and s["hold"] == 0 and s["health_score"] == 0

    def test_level_none_still_200(self, client: TestClient, db: Session):
        """不传 level（向后兼容路径）同样不得 500。"""
        create_test_user(db)
        token = get_token_for_user(client)
        hold_kpi, _ = _seed_kpis(db)
        db.add(KPIValue(kpi_id=hold_kpi.id, period="2026-09", actual_value=72.0, source_type="actual"))
        db.commit()
        m = _seed_review_map(db, _hold_dims(), title="P2 无 level", map_id=900104)
        resp = client.get(f"/api/cma/maps/{m.id}/review", headers=auth_header(token))
        assert resp.status_code == 200, resp.text
        assert resp.json()["summary"]["hold"] == 1