PayrollManagementSystem/backend/tests/integration/test_secondary_allocation.py

"""二次分配方案录入（业务创建/编辑同事务）集成测试。

守的是（计划 T3 验收）：
- 创建带方案 → 详情回显含 personnel_name 冗余与比例
- 编辑全量替换（空数组=清空）
- 离职/不存在人员 → 400「分摊参与人员不存在或已离职」
- 终态业务禁改方案（与编辑终态门槛同一口径）
- 合计不强制 100%：hint 展示合计与余额（余额归属待澄清 #2）
- 删除保护：被方案引用的人员 RESTRICT 禁删（400 而非静默剥离）
- 人员删除后方案回显仍有名（personnel_name 固化的意义）

人员造数走 /api/v1/base/personnel（种子营销员岗位）；离职用状态端点。
"""

import uuid

import pytest
from fastapi.testclient import TestClient

from tests.integration.factories import business_payload, customer

pytestmark = pytest.mark.integration

_BUSINESSES = "/api/v1/business/businesses"
_PERSONNEL = "/api/v1/base/personnel"


def _person(client: TestClient, headers: dict, **overrides: object) -> dict:
    role_id = overrides.pop("business_role_id", None)
    body = {
        "personnel_name": f"t_分摊_{uuid.uuid4().hex[:8]}",
        "business_role_id": role_id or _any_role_id(),
        "salary_standard": "3000.00",
        **overrides,
    }
    resp = client.post(_PERSONNEL, json=body, headers=headers)
    assert resp.status_code == 200, resp.text
    return resp.json()["data"]


def _any_role_id() -> int:
    """种子营销员岗位 ID（内置启用）。"""
    from sqlalchemy import select

    from app.core.constants import BUSINESS_ROLE_SALESPERSON
    from app.core.db import get_session_factory
    from app.models import BusinessRole

    factory = get_session_factory()
    with factory() as session:
        return int(
            session.execute(
                select(BusinessRole.id).where(
                    BusinessRole.role_code == BUSINESS_ROLE_SALESPERSON,
                    BusinessRole.status == 1,
                )
            ).scalar_one()
        )


def _create_with_allocations(client: TestClient, headers: dict, allocations: list[dict]) -> dict:
    cust = customer(client, headers)
    resp = client.post(
        _BUSINESSES,
        json=business_payload(cust["id"], secondary_allocations=allocations),
        headers=headers,
    )
    assert resp.status_code == 200, resp.text
    return resp.json()["data"]


def _set_switch(client: TestClient, headers: dict, value: str) -> None:
    """显式设置二次分配审批开关（V8 起默认开启，用例不得依赖种子默认值）。"""
    resp = client.post(
        "/api/v1/system/params/changes",
        json={
            "change_type": 1,
            "remark": f"用例设置二次分配审批开关={value}",
            "items": [
                {
                    "param_code": "approval.secondary_allocation.enabled",
                    "dimension_value": "",
                    "value": value,
                }
            ],
        },
        headers=headers,
    )
    assert resp.status_code == 200, resp.text


def _detail(client: TestClient, headers: dict, business_id: int) -> dict:
    resp = client.get(f"{_BUSINESSES}/{business_id}", headers=headers)
    assert resp.status_code == 200, resp.text
    return resp.json()["data"]


class TestCreateAndEcho:
    """创建带方案与详情回显。"""

    def test_带方案创建_详情回显姓名与比例(self, client: TestClient, admin: dict) -> None:
        p1 = _person(client, admin["headers"])
        p2 = _person(client, admin["headers"])
        data = _create_with_allocations(
            client,
            admin["headers"],
            [
                {"personnel_id": p1["id"], "ratio": "30.00"},
                {"personnel_id": p2["id"], "ratio": "20.00"},
            ],
        )
        detail = _detail(client, admin["headers"], data["id"])
        assert len(detail["secondary_allocations"]) == 2
        names = {a["personnel_name"] for a in detail["secondary_allocations"]}
        assert p1["personnel_name"] in names and p2["personnel_name"] in names
        ratios = sorted(a["ratio"] for a in detail["secondary_allocations"])
        assert ratios == ["20.00", "30.00"]
        # 合计 50%，余额 50% —— 只提示不落库（待澄清 #2）
        assert "50.00" in (detail["allocation_hint"] or "")
        assert "余额" in (detail["allocation_hint"] or "")

    def test_不建方案_详情为空且无hint(self, client: TestClient, admin: dict) -> None:
        data = _create_with_allocations(client, admin["headers"], [])
        detail = _detail(client, admin["headers"], data["id"])
        assert detail["secondary_allocations"] == []
        assert detail["allocation_hint"] is None

    def test_合计满额_hint只报合计(self, client: TestClient, admin: dict) -> None:
        p1 = _person(client, admin["headers"])
        p2 = _person(client, admin["headers"])
        data = _create_with_allocations(
            client,
            admin["headers"],
            [
                {"personnel_id": p1["id"], "ratio": "60"},
                {"personnel_id": p2["id"], "ratio": "40.00"},
            ],
        )
        hint = _detail(client, admin["headers"], data["id"])["allocation_hint"]
        assert hint is not None and "余额" not in hint


class TestGuards:
    """校验链：在职/终态/去重已由 schema 层覆盖，此处打服务层口径。"""

    def test_离职人员参与被拒(self, client: TestClient, admin: dict) -> None:
        p = _person(client, admin["headers"])
        resp = client.put(
            f"{_PERSONNEL}/{p['id']}/status", json={"status": 0}, headers=admin["headers"]
        )
        assert resp.status_code == 200, resp.text

        cust = customer(client, admin["headers"])
        resp = client.post(
            _BUSINESSES,
            json=business_payload(
                cust["id"],
                secondary_allocations=[{"personnel_id": p["id"], "ratio": "50.00"}],
            ),
            headers=admin["headers"],
        )
        assert resp.status_code == 400
        assert "分摊参与" in resp.json()["message"]

    def test_离职历史参与人_编辑其余字段放行(self, client: TestClient, admin: dict) -> None:
        """历史参与人离职后：编辑业务（保留该行）不应被 400 卡死——
        personnel_name 冗余固化的设计本就允许离职后回显；仅新增参与人
        才做在职校验。"""
        p = _person(client, admin["headers"])
        data = _create_with_allocations(
            client, admin["headers"], [{"personnel_id": p["id"], "ratio": "50.00"}]
        )
        resp = client.put(
            f"{_PERSONNEL}/{p['id']}/status", json={"status": 0}, headers=admin["headers"]
        )
        assert resp.status_code == 200

        detail = _detail(client, admin["headers"], data["id"])
        payload = business_payload(
            detail["customer_id"],
            secondary_allocations=[{"personnel_id": p["id"], "ratio": "50.00"}],
        )
        resp = client.put(f"{_BUSINESSES}/{data['id']}", json=payload, headers=admin["headers"])
        assert resp.status_code == 200, resp.text
        after = _detail(client, admin["headers"], data["id"])
        assert after["secondary_allocations"][0]["personnel_name"] == p["personnel_name"]

        # 新增一名离职人员仍要拦（新增才做在职校验）
        p2 = _person(client, admin["headers"])
        client.put(f"{_PERSONNEL}/{p2['id']}/status", json={"status": 0}, headers=admin["headers"])
        payload = business_payload(
            detail["customer_id"],
            secondary_allocations=[
                {"personnel_id": p["id"], "ratio": "30.00"},
                {"personnel_id": p2["id"], "ratio": "20.00"},
            ],
        )
        resp = client.put(f"{_BUSINESSES}/{data['id']}", json=payload, headers=admin["headers"])
        assert resp.status_code == 400
        assert "分摊参与" in resp.json()["message"]

    def test_终态业务禁改方案(self, client: TestClient, admin: dict) -> None:
        p = _person(client, admin["headers"])
        data = _create_with_allocations(
            client, admin["headers"], [{"personnel_id": p["id"], "ratio": "100"}]
        )
        resp = client.put(
            f"{_BUSINESSES}/{data['id']}/status",
            json={"status": 2},
            headers=admin["headers"],
        )
        assert resp.status_code == 200, resp.text

        detail = _detail(client, admin["headers"], data["id"])
        payload = business_payload(detail["customer_id"], secondary_allocations=[])
        for key in ("business_no", "status", "owner_personnel_id", "total_sale_amount"):
            payload.pop(key, None)
        resp = client.put(f"{_BUSINESSES}/{data['id']}", json=payload, headers=admin["headers"])
        assert resp.status_code == 400
        assert "禁止编辑" in resp.json()["message"]

    def test_编辑全量替换与清空(self, client: TestClient, admin: dict) -> None:
        p1 = _person(client, admin["headers"])
        p2 = _person(client, admin["headers"])
        data = _create_with_allocations(
            client, admin["headers"], [{"personnel_id": p1["id"], "ratio": "40.00"}]
        )

        detail = _detail(client, admin["headers"], data["id"])
        payload = business_payload(
            detail["customer_id"],
            secondary_allocations=[{"personnel_id": p2["id"], "ratio": "70.00"}],
        )
        resp = client.put(f"{_BUSINESSES}/{data['id']}", json=payload, headers=admin["headers"])
        assert resp.status_code == 200, resp.text

        replaced = _detail(client, admin["headers"], data["id"])
        assert len(replaced["secondary_allocations"]) == 1
        assert replaced["secondary_allocations"][0]["personnel_id"] == p2["id"]

        payload = business_payload(detail["customer_id"], secondary_allocations=[])
        resp = client.put(f"{_BUSINESSES}/{data['id']}", json=payload, headers=admin["headers"])
        assert resp.status_code == 200, resp.text
        assert _detail(client, admin["headers"], data["id"])["secondary_allocations"] == []


class TestDeleteProtection:
    """RESTRICT 外键：被方案引用的人员禁删，方案随业务删除而清。"""

    def test_同一会话内旧方案行同步为历史态(self, client: TestClient, admin: dict) -> None:
        """守「置历史后同一会话内的旧行必须同步刷新」这一行为：Core UPDATE
        走的是 SQLAlchemy 的会话同步策略，条件复杂化时可能退化为不刷新，
        届时行对象会停留在 is_active=1 的过期内存值（库中实际已置历史），
        被序列化或参与判断就会呈现「生效态」的假象。"""
        from app.core.db import get_session_factory
        from app.models import SecondaryAllocation
        from app.repositories import business_repository

        p = _person(client, admin["headers"])
        data = _create_with_allocations(
            client, admin["headers"], [{"personnel_id": p["id"], "ratio": "50.00"}]
        )

        factory = get_session_factory()
        with factory() as session:
            loaded = business_repository.list_allocations(session, data["id"])
            assert loaded and all(a.is_active == 1 for a in loaded)
            business_repository.deactivate_allocations(session, data["id"])
            # 同一 session 内的行对象必须已是历史态，不能停留在过期内存值
            assert all(a.is_active == 0 for a in loaded), "旧行未随置历史同步刷新"
            # 库中同样已置历史
            rows = (
                session.query(SecondaryAllocation)
                .filter(SecondaryAllocation.business_id == data["id"])
                .all()
            )
            assert rows and all(r.is_active == 0 for r in rows)
            session.rollback()

    def test_被方案引用的人员禁删(self, client: TestClient, admin: dict) -> None:
        p = _person(client, admin["headers"])
        data = _create_with_allocations(
            client, admin["headers"], [{"personnel_id": p["id"], "ratio": "50.00"}]
        )
        resp = client.delete(f"{_PERSONNEL}/{p['id']}", headers=admin["headers"])
        assert resp.status_code == 400
        # 方案仍在，回显有名（RESTRICT 保证未静默剥离）
        detail = _detail(client, admin["headers"], data["id"])
        assert detail["secondary_allocations"][0]["personnel_name"] == p["personnel_name"]

    def test_删除业务后可删人员(self, client: TestClient, admin: dict) -> None:
        # 显式关开关：开启时审批单在途、删业务被 400 拦截（产品行为，另有用例覆盖）
        _set_switch(client, admin["headers"], "0")
        p = _person(client, admin["headers"])
        data = _create_with_allocations(
            client, admin["headers"], [{"personnel_id": p["id"], "ratio": "50.00"}]
        )
        resp = client.delete(f"{_BUSINESSES}/{data['id']}", headers=admin["headers"])
        assert resp.status_code == 200, resp.text  # 方案 CASCADE 随业务删除
        assert client.delete(f"{_PERSONNEL}/{p['id']}", headers=admin["headers"]).status_code == 200