cma-management/backend/tests/test_kpi_causality.py

"""KPI因果链模块测试 — 因果网络 + 模拟推演 + CRUD

覆盖 kpi_causality.py 全部8个端点：
full-network / kpi{id}/network / simulate / list / get / create / update / delete
权限：读需 ceo/finance/business/it，写需 ceo/finance/it
"""
import hashlib

import pytest
from fastapi.testclient import TestClient
from sqlalchemy.orm import Session

from app.models import KPIDefinition, KPICausality, KPIValue, Entity, User, UserEntity
from tests.conftest import create_test_user, get_token_for_user, auth_header

BASE = "/api/cma/kpi-causality"


def _seed_kpi(db: Session, code: str, name: str = None, dimension: str = "finance",
              entity_id: int = 2) -> KPIDefinition:
    kpi = KPIDefinition(
        kpi_code=code,
        kpi_name=name or code,
        dimension=dimension,
        entity_id=entity_id,
        status="active",
        target_value=100.0,
    )
    db.add(kpi)
    db.commit()
    db.refresh(kpi)
    return kpi


def _seed_entity2(db: Session) -> None:
    """博海(id=2)为主测试实体，酣客(id=1)已有（conftest）"""
    ent = db.query(Entity).filter(Entity.id == 2).first()
    if not ent:
        db.add(Entity(id=2, name="博海网络科技", short_name="博海", status="active"))
        db.commit()


def _seed_chain(db: Session):
    """造一条因果链: 收入 → 净利润 (positive, 0.5)"""
    src = _seed_kpi(db, "BH_REVENUE", "营业收入")
    tgt = _seed_kpi(db, "BH_NET_PROFIT", "净利润")
    c = KPICausality(source_kpi_id=src.id, target_kpi_id=tgt.id,
                     strength=0.5, lag_months=1, direction="positive",
                     formula="净利润 = 收入 × 10%")
    db.add(c)
    db.commit()
    db.refresh(c)
    # 实际值（模拟推演用）
    db.add(KPIValue(kpi_id=src.id, period="2026-07", actual_value=100.0))
    db.add(KPIValue(kpi_id=tgt.id, period="2026-07", actual_value=10.0))
    db.commit()
    return src, tgt, c


class TestFullNetwork:
    def test_empty_network(self, client: TestClient, db: Session):
        """无数据时网络为空"""
        create_test_user(db)
        token = get_token_for_user(client)
        resp = client.get(f"{BASE}/full-network", headers=auth_header(token))
        assert resp.status_code == 200
        data = resp.json()
        assert data["nodes"] == []
        assert data["edges"] == []
        assert data["total_edges"] == 0

    def test_full_network_with_chain(self, client: TestClient, db: Session):
        """有因果链时返回节点和边"""
        create_test_user(db)
        token = get_token_for_user(client)
        src, tgt, c = _seed_chain(db)

        resp = client.get(f"{BASE}/full-network", headers=auth_header(token))
        assert resp.status_code == 200
        data = resp.json()
        assert data["total_edges"] == 1
        assert data["edges"][0]["source"] == src.id
        assert data["edges"][0]["target"] == tgt.id
        # 两个节点都带KPI信息
        codes = {n["kpi_code"] for n in data["nodes"]}
        assert codes == {"BH_REVENUE", "BH_NET_PROFIT"}


class TestKpiNetwork:
    def test_kpi_network(self, client: TestClient, db: Session):
        """单KPI上下游网络"""
        create_test_user(db)
        token = get_token_for_user(client)
        src, tgt, c = _seed_chain(db)

        # 源KPI的下游
        resp = client.get(f"{BASE}/kpi/{src.id}/network", headers=auth_header(token))
        assert resp.status_code == 200
        data = resp.json()
        assert data["kpi"]["kpi_code"] == "BH_REVENUE"
        assert len(data["downstream"]) == 1
        assert data["downstream"][0]["kpi_code"] == "BH_NET_PROFIT"
        assert len(data["upstream"]) == 0

        # 目标KPI的上游
        resp2 = client.get(f"{BASE}/kpi/{tgt.id}/network", headers=auth_header(token))
        assert resp2.status_code == 200
        data2 = resp2.json()
        assert len(data2["upstream"]) == 1
        assert data2["upstream"][0]["kpi_code"] == "BH_REVENUE"

    def test_kpi_network_not_found(self, client: TestClient, db: Session):
        """KPI不存在 → 404"""
        create_test_user(db)
        token = get_token_for_user(client)
        resp = client.get(f"{BASE}/kpi/99999/network", headers=auth_header(token))
        assert resp.status_code == 404


class TestSimulate:
    def test_simulate_simple(self, client: TestClient, db: Session):
        """模拟推演：收入+10% → 净利润受影响"""
        create_test_user(db)
        token = get_token_for_user(client)
        src, tgt, c = _seed_chain(db)

        resp = client.post(f"{BASE}/simulate", headers=auth_header(token), json={
            "kpi_id": src.id,
            "new_value": 110.0,
            "period": "2026-07",
        })
        assert resp.status_code == 200
        data = resp.json()
        assert data["source"]["kpi_code"] == "BH_REVENUE"
        assert data["source"]["change_pct"] == 10.0
        assert data["total_impacted"] == 1
        impact = data["impacts"][0]
        assert impact["kpi_code"] == "BH_NET_PROFIT"
        # 10% × 0.5(强度) × 1(正向) = 5% 影响
        assert impact["change_pct"] == 5.0
        # 预测值 = 10 × 1.05 = 10.5
        assert impact["predicted_value"] == 10.5

    def test_simulate_negative_direction(self, client: TestClient, db: Session):
        """负向因果：成本↑ → 净利润↓"""
        create_test_user(db)
        token = get_token_for_user(client)
        cost = _seed_kpi(db, "BH_COST", "成本")
        profit = _seed_kpi(db, "BH_NET_PROFIT", "净利润")
        c = KPICausality(source_kpi_id=cost.id, target_kpi_id=profit.id,
                         strength=0.8, lag_months=0, direction="negative")
        db.add(c)
        db.add(KPIValue(kpi_id=cost.id, period="2026-07", actual_value=50.0))
        db.add(KPIValue(kpi_id=profit.id, period="2026-07", actual_value=100.0))
        db.commit()

        resp = client.post(f"{BASE}/simulate", headers=auth_header(token), json={
            "kpi_id": cost.id, "new_value": 60.0, "period": "2026-07",
        })
        assert resp.status_code == 200
        data = resp.json()
        assert data["source"]["change_pct"] == 20.0
        impact = data["impacts"][0]
        # 20% × 0.8 × (-1) = -16%
        assert impact["change_pct"] == -16.0
        # 100 × 0.84 = 84.0
        assert impact["predicted_value"] == 84.0

    def test_simulate_missing_params(self, client: TestClient, db: Session):
        """缺 kpi_id/new_value → 400"""
        create_test_user(db)
        token = get_token_for_user(client)
        resp = client.post(f"{BASE}/simulate", headers=auth_header(token), json={"kpi_id": 1})
        assert resp.status_code == 400

    def test_simulate_kpi_not_found(self, client: TestClient, db: Session):
        """KPI不存在 → 404"""
        create_test_user(db)
        token = get_token_for_user(client)
        resp = client.post(f"{BASE}/simulate", headers=auth_header(token),
                           json={"kpi_id": 99999, "new_value": 10.0})
        assert resp.status_code == 404


class TestCausalityCRUD:
    def test_list_empty(self, client: TestClient, db: Session):
        """空列表"""
        create_test_user(db)
        token = get_token_for_user(client)
        resp = client.get(BASE, headers=auth_header(token))
        assert resp.status_code == 200
        assert resp.json()["data"] == []
        assert resp.json()["total"] == 0

    def test_create_and_get(self, client: TestClient, db: Session):
        """创建因果链 + 按ID查询"""
        create_test_user(db)
        token = get_token_for_user(client)
        src = _seed_kpi(db, "BH_REVENUE", "营业收入")
        tgt = _seed_kpi(db, "BH_NET_PROFIT", "净利润")

        resp = client.post(BASE, headers=auth_header(token), json={
            "source_kpi_id": src.id,
            "target_kpi_id": tgt.id,
            "strength": 0.6,
            "lag_months": 2,
            "direction": "positive",
            "formula": "净利润 = 收入 × 10%",
        })
        assert resp.status_code == 200
        cid = resp.json()["id"]
        assert resp.json()["strength"] == 0.6

        get_resp = client.get(f"{BASE}/{cid}", headers=auth_header(token))
        assert get_resp.status_code == 200
        assert get_resp.json()["source"]["kpi_code"] == "BH_REVENUE"
        assert get_resp.json()["target"]["kpi_code"] == "BH_NET_PROFIT"

    def test_create_duplicate(self, client: TestClient, db: Session):
        """重复创建同一条因果链 → 400"""
        create_test_user(db)
        token = get_token_for_user(client)
        src = _seed_kpi(db, "BH_REVENUE", "营业收入")
        tgt = _seed_kpi(db, "BH_NET_PROFIT", "净利润")

        client.post(BASE, headers=auth_header(token),
                    json={"source_kpi_id": src.id, "target_kpi_id": tgt.id})
        resp = client.post(BASE, headers=auth_header(token),
                           json={"source_kpi_id": src.id, "target_kpi_id": tgt.id})
        assert resp.status_code == 400
        assert "已存在" in resp.json()["detail"]

    def test_create_missing_kpis(self, client: TestClient, db: Session):
        """缺源/目标 → 400"""
        create_test_user(db)
        token = get_token_for_user(client)
        resp = client.post(BASE, headers=auth_header(token), json={"source_kpi_id": 1})
        assert resp.status_code == 400

    def test_create_same_kpi(self, client: TestClient, db: Session):
        """源=目标 → 400"""
        create_test_user(db)
        token = get_token_for_user(client)
        src = _seed_kpi(db, "BH_REVENUE", "营业收入")
        resp = client.post(BASE, headers=auth_header(token),
                           json={"source_kpi_id": src.id, "target_kpi_id": src.id})
        assert resp.status_code == 400

    def test_create_kpi_not_found(self, client: TestClient, db: Session):
        """KPI不存在 → 404"""
        create_test_user(db)
        token = get_token_for_user(client)
        src = _seed_kpi(db, "BH_REVENUE", "营业收入")
        resp = client.post(BASE, headers=auth_header(token),
                           json={"source_kpi_id": src.id, "target_kpi_id": 99999})
        assert resp.status_code == 404

    def test_update(self, client: TestClient, db: Session):
        """更新强度/滞后期/方向"""
        create_test_user(db)
        token = get_token_for_user(client)
        src, tgt, c = _seed_chain(db)

        resp = client.put(f"{BASE}/{c.id}", headers=auth_header(token), json={
            "strength": 0.9, "lag_months": 3, "direction": "negative",
        })
        assert resp.status_code == 200
        assert resp.json()["strength"] == 0.9
        assert resp.json()["lag_months"] == 3
        assert resp.json()["direction"] == "negative"

    def test_update_not_found(self, client: TestClient, db: Session):
        """更新不存在的因果链 → 404"""
        create_test_user(db)
        token = get_token_for_user(client)
        resp = client.put(f"{BASE}/99999", headers=auth_header(token), json={"strength": 0.5})
        assert resp.status_code == 404

    def test_delete(self, client: TestClient, db: Session):
        """删除因果链"""
        create_test_user(db)
        token = get_token_for_user(client)
        src, tgt, c = _seed_chain(db)

        resp = client.delete(f"{BASE}/{c.id}", headers=auth_header(token))
        assert resp.status_code == 200
        assert resp.json()["message"] == "已删除"

        # 列表验证已删除
        list_resp = client.get(BASE, headers=auth_header(token))
        assert list_resp.json()["total"] == 0

    def test_delete_not_found(self, client: TestClient, db: Session):
        """删除不存在的因果链 → 幂等返回已删除"""
        create_test_user(db)
        token = get_token_for_user(client)
        resp = client.delete(f"{BASE}/99999", headers=auth_header(token))
        assert resp.status_code == 200

    def test_list_filter(self, client: TestClient, db: Session):
        """列表按源/目标KPI过滤"""
        create_test_user(db)
        token = get_token_for_user(client)
        src, tgt, c = _seed_chain(db)

        resp = client.get(f"{BASE}?source_kpi_id={src.id}", headers=auth_header(token))
        assert resp.status_code == 200
        assert resp.json()["total"] == 1

        resp2 = client.get(f"{BASE}?source_kpi_id=99999", headers=auth_header(token))
        assert resp2.json()["total"] == 0


class TestPermissions:
    def test_write_requires_ceo_finance_it(self, client: TestClient, db: Session):
        """business角色无写权限 → 403"""
        # business用户
        business = User(
            username="business_user",
            password_hash=hashlib.sha256("pass123".encode()).hexdigest(),
            name="业务员",
            role="business",
        )
        db.add(business)
        db.commit()
        token = get_token_for_user(client, username="business_user", password="pass123")

        src = _seed_kpi(db, "BH_REVENUE", "营业收入")
        tgt = _seed_kpi(db, "BH_NET_PROFIT", "净利润")
        resp = client.post(BASE, headers=auth_header(token),
                           json={"source_kpi_id": src.id, "target_kpi_id": tgt.id})
        assert resp.status_code == 403

    def test_read_allowed_for_business(self, client: TestClient, db: Session):
        """business角色可读"""
        business = User(
            username="business_user2",
            password_hash=hashlib.sha256("pass123".encode()).hexdigest(),
            name="业务员",
            role="business",
        )
        db.add(business)
        db.commit()
        token = get_token_for_user(client, username="business_user2", password="pass123")

        resp = client.get(BASE, headers=auth_header(token))
        assert resp.status_code == 200

    def test_no_token_denied(self, client: TestClient):
        """无token → 403"""
        resp = client.get(BASE)
        assert resp.status_code == 403