PayrollManagementSystem/backend/tests/integration/test_settlements.py

"""月度结算集成测试（M3 ④：生成快照 → 审批 → 锁定）。

守的还有：
- 确认回款口径（#8 策略点）：仅 is_active=1 且未入结的台账行入结；
  入结后重生成同月不再重复计入
- 通过即锁定（status=3 终态）；锁定后回款写路径 400（走调整单）
- 驳回回草稿可重提；开关两路（关=自动通过锁定；开=审批流转）
- #12 暂口径：经理个人户业绩不入聚合基数（金标准仅业务员个人户，
  聚合用例以「无经理台账」形态验证 items 按人汇总不聚合）
"""

import uuid
from datetime import date
from decimal import Decimal

import pytest
from fastapi.testclient import TestClient
from sqlalchemy import select

from app.core.db import get_session_factory
from app.models import MonthlySettlement, PerformanceLedger
from tests.integration.factories import business_payload, customer

pytestmark = pytest.mark.integration

_SETTLEMENTS = "/api/v1/settlements"
_CALC = "/api/v1/calc"
_BUSINESSES = "/api/v1/business/businesses"
_SHIPMENTS = "/api/v1/business/shipments"
_RECEIPTS = "/api/v1/business/receipts"
_PERSONNEL = "/api/v1/base/personnel"

_GOLD_ITEMS = [
    {
        "product_name": "t_市政_月结",
        "product_category": "municipal_water",
        "quantity": "100",
        "sale_price": "100.00",
        "cost_amount": "3000.00",
    }
]


def _role_id(role_code: str) -> int:
    from sqlalchemy import select

    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 == role_code,
                    BusinessRole.status == 1,
                )
            ).scalar_one()
        )


def _person(
    client: TestClient,
    headers: dict,
    *,
    role_code: str = "salesperson",
    manager_id: int | None = None,
) -> dict:
    resp = client.post(
        _PERSONNEL,
        json={
            "personnel_name": f"t_月结_{uuid.uuid4().hex[:8]}",
            "business_role_id": _role_id(role_code),
            "salary_standard": "3000.00",
            **({"manager_id": manager_id} if manager_id is not None else {}),
        },
        headers=headers,
    )
    assert resp.status_code == 200, resp.text
    return resp.json()["data"]


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


def _calc_one(client: TestClient, headers: dict) -> dict:
    """完整造一笔已核算回款（回款日 2026-10-20 → 归月 2026-10）。"""
    cust = customer(client, headers)
    person = _person(client, headers)
    resp = client.post(
        _BUSINESSES,
        json=business_payload(
            cust["id"],
            owner_personnel_id=person["id"],
            items=[dict(_GOLD_ITEMS[0], product_name=f"t_月结产品_{uuid.uuid4().hex[:6]}")],
        ),
        headers=headers,
    )
    assert resp.status_code == 200, resp.text
    biz = resp.json()["data"]

    resp = client.post(
        _SHIPMENTS,
        json={
            "business_id": biz["id"],
            "ship_date": "2026-10-01",
            "items": [{"product_name": "t_发", "quantity": "1", "sale_price": "1.00"}],
        },
        headers=headers,
    )
    assert resp.status_code == 200, resp.text

    resp = client.post(
        _RECEIPTS,
        json={
            "business_id": biz["id"],
            "receipt_date": "2026-10-20",
            "amount": "20000.00",
            "shipment_id": resp.json()["data"]["id"],
        },
        headers=headers,
    )
    assert resp.status_code == 200, resp.text
    receipt = resp.json()["data"]

    resp = client.post(f"{_CALC}/receipts/{receipt['id']}/calc", headers=headers)
    assert resp.status_code == 200, resp.text
    return biz


@pytest.fixture
def finance(make_user, auth_headers) -> dict:
    user = make_user(["finance"])
    user["headers"] = auth_headers(user)
    return user


class TestGenerate:
    """生成快照：确认回款口径与按人汇总。"""

    def test_生成_按人汇总并回填台账(self, client: TestClient, admin: dict) -> None:
        _calc_one(client, admin["headers"])
        resp = client.post(
            f"{_SETTLEMENTS}/generate", json={"month": "2026-10"}, headers=admin["headers"]
        )
        assert resp.status_code == 200, resp.text
        out = resp.json()["data"]
        assert out["settlement_month"] == "2026-10"
        assert out["status"] == 1  # 草稿
        assert len(out["items"]) >= 1
        item = out["items"][0]
        assert item["is_probation"] in (0, 1)

        # 台账行已回填 settlement_id（入结标记，防重复计入）
        factory = get_session_factory()
        with factory() as session:
            settled = session.execute(
                select(MonthlySettlement).where(MonthlySettlement.settlement_month == "2026-10")
            ).scalar_one()
            assert settled.person_count >= 1
            backfilled = (
                session.execute(
                    select(PerformanceLedger).where(PerformanceLedger.settlement_id == settled.id)
                )
                .scalars()
                .all()
            )
            assert backfilled, "台账行 settlement_id 未回填（「未入结」判据失效）"
            assert all(r.settlement_month == "2026-10" for r in backfilled)

    def test_无核算记录生成400(self, client: TestClient, admin: dict) -> None:
        resp = client.post(
            f"{_SETTLEMENTS}/generate", json={"month": "2099-01"}, headers=admin["headers"]
        )
        assert resp.status_code == 400
        assert "无可结算" in resp.json()["message"]

    def test_已审定月份重生成400(self, client: TestClient, admin: dict) -> None:
        _calc_one(client, admin["headers"])
        assert (
            client.post(
                f"{_SETTLEMENTS}/generate", json={"month": "2026-10"}, headers=admin["headers"]
            ).status_code
            == 200
        )
        # 显式关开关 → 提交即自动通过锁定（用例不依赖种子默认值）
        _set_switch(client, admin["headers"], "approval.monthly_settlement.enabled", "0")
        sid = client.get(f"{_SETTLEMENTS}", headers=admin["headers"]).json()["data"]["list"][0][
            "id"
        ]
        assert (
            client.post(f"{_SETTLEMENTS}/{sid}/submit", headers=admin["headers"]).status_code == 200
        )
        resp = client.post(
            f"{_SETTLEMENTS}/generate", json={"month": "2026-10"}, headers=admin["headers"]
        )
        assert resp.status_code == 400
        assert "已审定锁定" in resp.json()["message"]


class TestSubmitAndLock:
    """提交审批与锁定（开关两路）。"""

    def test_开关关闭_提交即自动通过锁定(
        self, client: TestClient, admin: dict, finance: dict
    ) -> None:
        _calc_one(client, admin["headers"])
        # 显式关开关（V8 起种子默认开启，用例不得依赖默认值）
        _set_switch(client, admin["headers"], "approval.monthly_settlement.enabled", "0")
        resp = client.post(
            f"{_SETTLEMENTS}/generate", json={"month": "2026-10"}, headers=admin["headers"]
        )
        sid = resp.json()["data"]["id"]
        resp = client.post(f"{_SETTLEMENTS}/{sid}/submit", headers=admin["headers"])
        assert resp.status_code == 200, resp.text
        out = resp.json()["data"]
        assert out["status"] == 3  # 自动通过=已审定（锁定）
        assert out["approval_order_no"] and out["approval_order_no"].startswith("AP")
        assert out["locked_at"] is not None

    def test_锁定后回款禁改(self, client: TestClient, admin: dict) -> None:
        from sqlalchemy import select

        biz = _calc_one(client, admin["headers"])
        client.post(f"{_SETTLEMENTS}/generate", json={"month": "2026-10"}, headers=admin["headers"])
        sid = client.get(f"{_SETTLEMENTS}", headers=admin["headers"]).json()["data"]["list"][0][
            "id"
        ]
        client.post(f"{_SETTLEMENTS}/{sid}/submit", headers=admin["headers"])

        # 找该月回款并尝试编辑（400：月已锁定）
        factory = get_session_factory()
        with factory() as session:
            from app.models import Receipt

            receipt_id = int(
                session.execute(
                    select(Receipt.id).where(Receipt.business_id == biz["id"])
                ).scalar_one()
            )
        resp = client.put(
            f"{_RECEIPTS}/{receipt_id}",
            json={
                "business_id": biz["id"],
                "receipt_date": "2026-10-21",
                "amount": "999.00",
            },
            headers=admin["headers"],
        )
        assert resp.status_code == 400
        # 已核算禁改（M2 防线）先于月结锁定拦截生效——两者均 400 拒改
        assert ("已核算" in resp.json()["message"]) or ("锁定" in resp.json()["message"])


class TestSwitchOn:
    """开关开启：审批流转后锁定。"""

    def test_提交待审_通过后锁定(
        self, client: TestClient, admin: dict, finance: dict, make_user, auth_headers
    ) -> None:
        # 开月结审批开关（立即生效）
        resp = client.post(
            "/api/v1/system/params/changes",
            json={
                "change_type": 1,
                "remark": "开月结审批",
                "items": [
                    {
                        "param_code": "approval.monthly_settlement.enabled",
                        "dimension_value": "",
                        "value": "1",
                    }
                ],
            },
            headers=admin["headers"],
        )
        assert resp.status_code == 200, resp.text

        leader = make_user(["leader"])
        leader["headers"] = auth_headers(leader)

        # 造 11 月的台账（避开 10 月已入结月份）
        cust = customer(client, admin["headers"])
        person = _person(client, admin["headers"])
        resp = client.post(
            _BUSINESSES,
            json=business_payload(
                cust["id"],
                owner_personnel_id=person["id"],
                items=[dict(_GOLD_ITEMS[0], product_name=f"t_11月_{uuid.uuid4().hex[:6]}")],
            ),
            headers=admin["headers"],
        )
        assert resp.status_code == 200, resp.text
        biz = resp.json()["data"]
        resp = client.post(
            _SHIPMENTS,
            json={
                "business_id": biz["id"],
                "ship_date": "2026-11-01",
                "items": [{"product_name": "t_发", "quantity": "1", "sale_price": "1.00"}],
            },
            headers=admin["headers"],
        )
        assert resp.status_code == 200, resp.text
        shipment = resp.json()["data"]
        resp = client.post(
            _RECEIPTS,
            json={
                "business_id": biz["id"],
                "receipt_date": "2026-11-20",
                "amount": "20000.00",
                "shipment_id": shipment["id"],
            },
            headers=admin["headers"],
        )
        assert resp.status_code == 200, resp.text
        receipt = resp.json()["data"]
        assert (
            client.post(
                f"{_CALC}/receipts/{receipt['id']}/calc", headers=admin["headers"]
            ).status_code
            == 200
        )

        resp = client.post(
            f"{_SETTLEMENTS}/generate", json={"month": "2026-11"}, headers=admin["headers"]
        )
        assert resp.status_code == 200, resp.text
        sid = resp.json()["data"]["id"]

        # 提交 → 待财务复核
        resp = client.post(f"{_SETTLEMENTS}/{sid}/submit", headers=admin["headers"])
        assert resp.status_code == 200, resp.text
        assert resp.json()["data"]["status"] == 2

        # 财务通过 → 待领导（approval_order_no 关联的审批单流转）
        order_no = resp.json()["data"]["approval_order_no"]
        order = client.get(
            "/api/v1/system/approvals", params={"keyword": order_no}, headers=admin["headers"]
        ).json()["data"]["list"][0]
        resp = client.post(
            f"/api/v1/system/approvals/{order['id']}/action",
            json={"action": 1},
            headers=finance["headers"],
        )
        assert resp.status_code == 200, resp.text

        # 财务节点通过只是流转到待领导——不得提前锁定月结（否则领导驳回后
        # 锁定字段残留、两级审批形同虚设）
        mid = client.get(f"{_SETTLEMENTS}/{sid}", headers=admin["headers"]).json()["data"]
        assert mid["status"] == 2, "财务通过后应仍为待审批，不得提前锁定"
        assert mid["locked_at"] is None
        assert mid["locked_by"] is None

        # 领导通过 → 锁定
        resp = client.post(
            f"/api/v1/system/approvals/{order['id']}/action",
            json={"action": 1},
            headers=leader["headers"],
        )
        assert resp.status_code == 200, resp.text
        final = client.get(f"{_SETTLEMENTS}/{sid}", headers=admin["headers"]).json()["data"]
        assert final["status"] == 3
        assert final["locked_at"] is not None
        assert final["locked_by"] == leader["username"]  # 审批通过人=领导账号


class TestManagerAggregate:
    """经理团队聚合：比例按参数实际值计算（守「百分数再乘 100」放大缺陷）。"""

    def _scene(self, client: TestClient, headers: dict) -> tuple[dict, dict]:
        """经理（带一名业务员）→ 业务 → 发货 → 已核算回款，返回 (经理, 回款)。"""
        manager = _person(client, headers, role_code="sales_manager")
        sales = _person(client, headers, manager_id=manager["id"])

        cust = customer(client, headers)
        resp = client.post(
            _BUSINESSES,
            json=business_payload(
                cust["id"],
                owner_personnel_id=sales["id"],
                items=[dict(_GOLD_ITEMS[0], product_name=f"t_聚合_{uuid.uuid4().hex[:6]}")],
            ),
            headers=headers,
        )
        assert resp.status_code == 200, resp.text
        biz = resp.json()["data"]
        resp = client.post(
            _SHIPMENTS,
            json={
                "business_id": biz["id"],
                "ship_date": "2026-10-01",
                "items": [{"product_name": "t_发", "quantity": "1", "sale_price": "1.00"}],
            },
            headers=headers,
        )
        assert resp.status_code == 200, resp.text
        resp = client.post(
            _RECEIPTS,
            json={
                "business_id": biz["id"],
                "receipt_date": "2026-10-20",
                "amount": "20000.00",
                "shipment_id": resp.json()["data"]["id"],
            },
            headers=headers,
        )
        assert resp.status_code == 200, resp.text
        receipt = resp.json()["data"]
        assert (
            client.post(f"{_CALC}/receipts/{receipt['id']}/calc", headers=headers).status_code
            == 200
        )
        return manager, receipt

    def test_聚合金额按实际比例计入且落库回填(self, client: TestClient, admin: dict) -> None:
        manager, _receipt = self._scene(client, admin["headers"])
        resp = client.post(
            f"{_SETTLEMENTS}/generate", json={"month": "2026-10"}, headers=admin["headers"]
        )
        assert resp.status_code == 200, resp.text
        out = resp.json()["data"]
        items = out["items"]
        manager_items = [i for i in items if i["calc_type"] == 2]
        assert len(manager_items) == 1
        # 业务员绩效 39.11；经理 = 39.11×12.5% + 39.11×5% = 6.84425 → 6.84
        # （若调用方把百分数再 ×100，此处会是 684.43）
        assert manager_items[0]["personnel_id"] == manager["id"]
        assert manager_items[0]["performance_sum"] == "6.84"
        assert manager_items[0]["payable_sum"] == "6.84"

        # 聚合行必须落台账（报告下钻的数据源）且被回填 settlement_id
        factory = get_session_factory()
        with factory() as session:
            rows = (
                session.execute(
                    select(PerformanceLedger).where(
                        PerformanceLedger.settlement_month == "2026-10",
                        PerformanceLedger.calc_type == 2,
                    )
                )
                .scalars()
                .all()
            )
            assert len(rows) == 1, "聚合行必须落库（修复前只在内存参与汇总）"
            assert rows[0].is_active == 1
            assert rows[0].receipt_id is None
            assert rows[0].settlement_id == out["id"], "聚合行未回填 settlement_id"

    def test_草稿态重生成_聚合不重复计入(self, client: TestClient, admin: dict) -> None:
        self._scene(client, admin["headers"])
        first = client.post(
            f"{_SETTLEMENTS}/generate", json={"month": "2026-10"}, headers=admin["headers"]
        ).json()["data"]
        second = client.post(
            f"{_SETTLEMENTS}/generate", json={"month": "2026-10"}, headers=admin["headers"]
        )
        assert second.status_code == 200, second.text
        out = second.json()["data"]
        # 重生成金额不翻倍（聚合行不做废旧行会 45.95→52.79）也不漏行
        assert out["total_payable"] == first["total_payable"] == "45.95"
        assert len([i for i in out["items"] if i["calc_type"] == 2]) == 1
        assert len([i for i in out["items"] if i["calc_type"] == 1]) == 1

    def test_月末取参_预约变更影响本月聚合(self, client: TestClient, admin: dict) -> None:
        """取参锚点=所属月月末：10-25 生效的预约变更应影响 2026-10 月结。"""
        resp = client.post(
            "/api/v1/system/params/changes",
            json={
                "change_type": 2,
                "effective_date": "2026-10-25",
                "remark": "用例：月末取参锚点",
                "items": [
                    {
                        "param_code": "manager.share.team",
                        "dimension_value": "",
                        "value": "20",
                    }
                ],
            },
            headers=admin["headers"],
        )
        assert resp.status_code == 200, resp.text

        self._scene(client, admin["headers"])
        resp = client.post(
            f"{_SETTLEMENTS}/generate", json={"month": "2026-10"}, headers=admin["headers"]
        )
        assert resp.status_code == 200, resp.text
        manager_items = [i for i in resp.json()["data"]["items"] if i["calc_type"] == 2]
        # 39.11×20% + 39.11×5% = 9.7775 → 9.78（按生成日取参会得 6.84）
        assert manager_items[0]["performance_sum"] == "9.78"

        # 台账快照固化月末取参日与实际比例原文
        factory = get_session_factory()
        with factory() as session:
            row = session.execute(
                select(PerformanceLedger).where(
                    PerformanceLedger.settlement_month == "2026-10",
                    PerformanceLedger.calc_type == 2,
                )
            ).scalar_one()
            assert row.snapshot_json["params_as_of"] == "2026-10-31"
            assert row.rate_text == "聚合:20%+5%"
            assert str(row.receipt_date) == "2026-10-31"

    def test_聚合基数开关_经理个人户计入可配置(self, client: TestClient, admin: dict) -> None:
        """#12 答复（2026-10-19）参数化：manager.personal.included=1 时个人户业绩并入基数。

        直插一行经理个人户台账（绩效 100，模拟该类业务经单笔侧计提后的
        结果）——默认 0 时聚合 6.84 不变；开关 1 后基数 39.11+100=139.11。
        """
        manager, _receipt = self._scene(client, admin["headers"])
        factory = get_session_factory()
        with factory() as session:
            session.add(
                PerformanceLedger(
                    receipt_id=None,
                    calc_version=1,
                    calc_type=1,
                    business_id=0,
                    business_no="t_个人户",
                    personnel_id=manager["id"],
                    personnel_name=manager["personnel_name"],
                    role_code="sales_manager",
                    customer_type="manager_personal",
                    product_category="mixed",
                    receipt_date=date(2026, 10, 15),
                    rate_text="0",
                    performance_amount=Decimal("100.00"),
                    barter_applied=0,
                    payable_amount=Decimal("0"),
                    settlement_month="2026-10",
                    is_probation=0,
                )
            )
            session.commit()

        def _manager_share() -> str:
            resp = client.post(
                f"{_SETTLEMENTS}/generate", json={"month": "2026-10"}, headers=admin["headers"]
            )
            assert resp.status_code == 200, resp.text
            items = [i for i in resp.json()["data"]["items"] if i["calc_type"] == 2]
            assert len(items) == 1
            return items[0]["performance_sum"]

        # 默认（开关 0=不计入）：基数 39.11 → 6.84（与既有金标准一致）
        assert _manager_share() == "6.84"

        # 开关 1（立即生效，月末 2026-10-31 取参命中）：139.11×12.5% + ×5% = 24.34425
        resp = client.post(
            "/api/v1/system/params/changes",
            json={
                "change_type": 1,
                "remark": "用例：#12 开关计入",
                "items": [
                    {
                        "param_code": "manager.personal.included",
                        "dimension_value": "",
                        "value": "1",
                    }
                ],
            },
            headers=admin["headers"],
        )
        assert resp.status_code == 200, resp.text
        assert _manager_share() == "24.34"


class TestAdjustment:
    """调整单：已锁定月份补发/扣减的唯一通道。"""

    def test_新建调整单_开关关闭即生效(self, client: TestClient, admin: dict) -> None:
        # 显式关开关（用例不依赖种子默认值）
        _set_switch(client, admin["headers"], "approval.adjustment.enabled", "0")
        person = _person(client, admin["headers"])
        resp = client.post(
            f"{_SETTLEMENTS}/adjustments",
            json={
                "settlement_month": "2026-10",
                "personnel_id": person["id"],
                "direction": 1,
                "amount": "500.00",
                "reason": "补发差额",
            },
            headers=admin["headers"],
        )
        # 守 ENABLE_CODE_BY_TYPE 漏登记 type=3：修复前此处恒 400「未知的审批对象类型」
        assert resp.status_code == 200, resp.text
        out = resp.json()["data"]
        assert out["status"] == 2  # 已生效

    def test_金额恒正校验(self, client: TestClient, admin: dict) -> None:
        """守金额无校验缺陷：负数会反转补发/扣减语义，非数值直落库 500。"""
        _set_switch(client, admin["headers"], "approval.adjustment.enabled", "0")
        person = _person(client, admin["headers"])
        base = {
            "settlement_month": "2026-10",
            "personnel_id": person["id"],
            "direction": 1,
            "reason": "金额校验用例",
        }
        for bad in ("-500.00", "0", "0.00", "abc", "1e3"):
            resp = client.post(
                f"{_SETTLEMENTS}/adjustments",
                json={**base, "amount": bad},
                headers=admin["headers"],
            )
            assert resp.status_code == 400, f"amount={bad!r} 应 400：{resp.text}"


class TestLockInterception:
    """锁定后源数据写路径全量拦截（回款编辑/删除、发货增改删、核算/重算）。"""

    def _lock_scene(self, client: TestClient, admin: dict) -> tuple[dict, dict, dict, dict]:
        """已核算+未核算各一笔回款、两张 10 月发货单，然后锁定 2026-10。

        返回 (月结, 业务, 未核算回款, 无回款引用的发货单)。
        """
        biz = _calc_one(client, admin["headers"])
        resp = client.post(
            _RECEIPTS,
            json={"business_id": biz["id"], "receipt_date": "2026-10-22", "amount": "100.00"},
            headers=admin["headers"],
        )
        assert resp.status_code == 200, resp.text
        unsettled = resp.json()["data"]
        resp = client.post(
            _SHIPMENTS,
            json={
                "business_id": biz["id"],
                "ship_date": "2026-10-05",
                "items": [{"product_name": "t_发2", "quantity": "1", "sale_price": "1.00"}],
            },
            headers=admin["headers"],
        )
        assert resp.status_code == 200, resp.text
        free_shipment = resp.json()["data"]

        _set_switch(client, admin["headers"], "approval.monthly_settlement.enabled", "0")
        resp = client.post(
            f"{_SETTLEMENTS}/generate", json={"month": "2026-10"}, headers=admin["headers"]
        )
        assert resp.status_code == 200, resp.text
        settlement = resp.json()["data"]
        resp = client.post(f"{_SETTLEMENTS}/{settlement['id']}/submit", headers=admin["headers"])
        assert resp.status_code == 200, resp.text
        assert resp.json()["data"]["status"] == 3
        return settlement, biz, unsettled, free_shipment

    def test_锁定后_未核算回款禁改禁删(self, client: TestClient, admin: dict) -> None:
        _settlement, biz, unsettled, _ship = self._lock_scene(client, admin)
        resp = client.put(
            f"{_RECEIPTS}/{unsettled['id']}",
            json={
                "business_id": biz["id"],
                "receipt_date": "2026-10-23",
                "amount": "88.00",
            },
            headers=admin["headers"],
        )
        assert resp.status_code == 400, resp.text
        assert "锁定" in resp.json()["message"]

        resp = client.delete(f"{_RECEIPTS}/{unsettled['id']}", headers=admin["headers"])
        assert resp.status_code == 400, resp.text
        assert "锁定" in resp.json()["message"]

    def test_锁定后_发货增改删400_非锁定月放行(self, client: TestClient, admin: dict) -> None:
        _settlement, biz, _unsettled, free_shipment = self._lock_scene(client, admin)
        items = [{"product_name": "t_发x", "quantity": "1", "sale_price": "1.00"}]

        # 新增：落锁定月 400
        resp = client.post(
            _SHIPMENTS,
            json={"business_id": biz["id"], "ship_date": "2026-10-08", "items": items},
            headers=admin["headers"],
        )
        assert resp.status_code == 400, resp.text
        assert "锁定" in resp.json()["message"]

        # 新增：非锁定月放行（拦截不误伤其它月份）
        resp = client.post(
            _SHIPMENTS,
            json={"business_id": biz["id"], "ship_date": "2026-11-02", "items": items},
            headers=admin["headers"],
        )
        assert resp.status_code == 200, resp.text
        nov = resp.json()["data"]

        # 编辑：把非锁定月的发货单改入锁定月 400
        resp = client.put(
            f"{_SHIPMENTS}/{nov['id']}",
            json={"ship_date": "2026-10-15", "items": items},
            headers=admin["headers"],
        )
        assert resp.status_code == 400, resp.text
        assert "锁定" in resp.json()["message"]

        # 编辑：原月即锁定月 400
        resp = client.put(
            f"{_SHIPMENTS}/{free_shipment['id']}",
            json={"ship_date": "2026-10-06", "items": items},
            headers=admin["headers"],
        )
        assert resp.status_code == 400, resp.text
        assert "锁定" in resp.json()["message"]

        # 删除：锁定月的发货单 400（拦截先于「被回款引用」判定）
        resp = client.delete(f"{_SHIPMENTS}/{free_shipment['id']}", headers=admin["headers"])
        assert resp.status_code == 400, resp.text
        assert "锁定" in resp.json()["message"]

    def test_锁定后_首算与重算400(self, client: TestClient, admin: dict) -> None:
        _settlement, biz, unsettled, _ship = self._lock_scene(client, admin)
        from app.models import Receipt

        with get_session_factory()() as session:
            settled_id = int(
                session.execute(
                    select(Receipt.id).where(
                        Receipt.business_id == biz["id"],
                        Receipt.settlement_status == 1,
                    )
                ).scalar_one()
            )

        resp = client.post(
            f"{_CALC}/receipts/{settled_id}/recalc",
            json={"reason": "锁定月重算"},
            headers=admin["headers"],
        )
        assert resp.status_code == 400, resp.text
        assert "锁定" in resp.json()["message"]

        resp = client.post(f"{_CALC}/receipts/{unsettled['id']}/calc", headers=admin["headers"])
        assert resp.status_code == 400, resp.text
        assert "锁定" in resp.json()["message"]