"""业务管理接口集成测试（需要 MySQL 与 Redis）。

守的是业务单的五处汇总冗余一致性与状态机/归属留痕 —— 明细/发货/回款任何
写路径漏算汇总，业务行的 total_* 就会与源数据漂移，M3 起付判定直接读冗余
列，漂移就是算错钱；状态留痕的「变更前状态」若取到赋值后的值（old==new
失真）或归属变更拆成「主表已换、留痕缺失」，事后追溯无据；编辑端点若能
改归属，就绕过留痕产生「改了归属却无留痕」的旁路。毛利润视图为只读派生
值，断言锁定金标准口径（销售额 0 防除零、成本缺失行不进类别汇总）。

运行前置：``docker compose -f docker/docker-compose.dev.yml up -d``

数据卫生：人员/公司/产品名一律 ``t_`` 前缀由 conftest 兜底清理；业务/
发货/回款编号系统生成（YW/SH/HK 前缀），conftest 按编号前缀全量清；
删除语义的用例自验 DELETE 端点。
"""

import re
import uuid
from datetime import date
from decimal import Decimal

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

from app.core.constants import (
    ACTION_CREATE,
    BUSINESS_ROLE_SALESPERSON,
    MODULE_BUSINESS,
    PRODUCT_CATEGORY_MUNICIPAL_WATER,
    PRODUCT_CATEGORY_RAIL_TRANSIT,
)
from app.core.db import get_session_factory
from app.models import (
    Business,
    BusinessItem,
    BusinessRole,
    OperationLog,
    OwnershipChangeLog,
    ParamChangeBatch,
    ParamValue,
)
from app.models.business import BUSINESS_STATUS_ACTIVE
from app.models.customer import OWNER_TYPE_BUSINESS

pytestmark = pytest.mark.integration

_BUSINESSES = "/api/v1/business/businesses"
_PERSONNEL = "/api/v1/base/personnel"
_CUSTOMERS = "/api/v1/base/customers"
_PARAMS = "/api/v1/system/params"


def _company() -> str:
    return f"t_公司_{uuid.uuid4().hex[:8]}"


def _personnel(client: TestClient, headers: dict) -> dict:
    """建一名 t_ 前缀的在职营销人员（归属变更的被引用方）。"""
    factory = get_session_factory()
    with factory() as session:
        role_id = int(
            session.execute(
                select(BusinessRole.id).where(
                    BusinessRole.role_code == BUSINESS_ROLE_SALESPERSON,
                    BusinessRole.status == 1,
                )
            ).scalar_one()
        )
    resp = client.post(
        _PERSONNEL,
        json={
            "personnel_name": f"t_人员_{uuid.uuid4().hex[:8]}",
            "business_role_id": role_id,
        },
        headers=headers,
    )
    assert resp.status_code == 200, resp.text
    return resp.json()["data"]


def _customer(client: TestClient, headers: dict) -> dict:
    """建一名 t_ 前缀测试客户（业务的必选引用）。"""
    resp = client.post(
        _CUSTOMERS,
        json={
            "company_name": _company(),
            "owner_name": "张老板",
            "customer_type": "personal",
            "address": "本溪市平山区",
        },
        headers=headers,
    )
    assert resp.status_code == 200, resp.text
    return resp.json()["data"]


def _item(**overrides: object) -> dict:
    """业务明细行入参（默认：数量 100 × 单价 100.00 = 销售额 10000，成本 30）。"""
    body: dict = {
        "product_name": f"t_产品_{uuid.uuid4().hex[:8]}",
        "product_category": PRODUCT_CATEGORY_MUNICIPAL_WATER,
        "quantity": "100",
        "sale_price": "100.00",
        "cost_amount": "30.00",
    }
    body.update(overrides)
    return body


def _payload(customer_id: int, **overrides: object) -> dict:
    body: dict = {
        "customer_id": customer_id,
        "business_date": str(date.today()),
        "items": [_item()],
    }
    body.update(overrides)
    return body


def _create(client: TestClient, headers: dict, customer_id: int, **overrides: object) -> dict:
    resp = client.post(_BUSINESSES, json=_payload(customer_id, **overrides), headers=headers)
    assert resp.status_code == 200, resp.text
    return resp.json()["data"]


def _terminate(client: TestClient, headers: dict, business_id: int) -> None:
    """把业务流转到已终止（终态禁写场景的造数步骤）。"""
    resp = client.put(
        f"{_BUSINESSES}/{business_id}/status",
        json={"status": 3, "reason": f"t_终止_{uuid.uuid4().hex[:6]}"},
        headers=headers,
    )
    assert resp.status_code == 200, resp.text


def _logs_of(username: str) -> list[OperationLog]:
    """取某操作人的审计行（按 id 倒序，最新在前）。"""
    factory = get_session_factory()
    with factory() as session:
        return list(
            session.execute(
                select(OperationLog)
                .where(OperationLog.username == username)
                .order_by(OperationLog.id.desc())
            ).scalars()
        )


@pytest.fixture
def threshold_batch(client: TestClient, admin: dict):
    """提交「立即生效」的产品阈值参数变更批次，用例结束整批清理。

    走参数接口的真实路径（system params 要求管理员）；清理按 batch_id 删
    条目与批次头（test_param_management.track_batches 同款）。
    """
    created: list[int] = []

    def _factory(dimension_value: str, value: str) -> dict:
        resp = client.post(
            f"{_PARAMS}/changes",
            json={
                "change_type": 1,
                "remark": "t_毛利润阈值变更",
                "items": [
                    {
                        "param_code": "product.threshold",
                        "dimension_value": dimension_value,
                        "value": value,
                    }
                ],
            },
            headers=admin["headers"],
        )
        assert resp.status_code == 200, resp.text
        batch = resp.json()["data"]
        created.append(batch["id"])
        return batch

    yield _factory

    factory = get_session_factory()
    with factory() as session:
        if created:
            session.execute(delete(ParamValue).where(ParamValue.batch_id.in_(created)))
            session.execute(delete(ParamChangeBatch).where(ParamChangeBatch.id.in_(created)))
        session.commit()


class TestAuth:
    """鉴权：业务数据属业务操作不要求管理员，但仍是登录后接口。"""

    def test_未登录访问业务列表_401(self, client: TestClient) -> None:
        resp = client.get(_BUSINESSES)
        assert resp.status_code == 401, resp.text


class TestCreate:
    """创建：编号契约、行校验与销售额固化 —— 编号序号错乱或行金额经前端
    传值，列表检索与 M3 取数都会错乱。"""

    def test_创建业务_编号格式与当日序号递增(self, client: TestClient, normal: dict) -> None:
        headers = normal["headers"]
        customer = _customer(client, headers)
        first = _create(client, headers, customer["id"])
        second = _create(client, headers, customer["id"])

        # 编号格式：YW{YYYYMMDD}-{4位当日序号}（日期取当日，序号连续 +1）
        expected_prefix = f"YW{date.today():%Y%m%d}-"
        for no in (first["business_no"], second["business_no"]):
            assert re.fullmatch(r"YW\d{8}-\d{4}", no), no
            assert no.startswith(expected_prefix), (no, expected_prefix)
        assert int(second["business_no"].rsplit("-", 1)[1]) == (
            int(first["business_no"].rsplit("-", 1)[1]) + 1
        ), (first["business_no"], second["business_no"])

        # 回包契约：默认执行中、客户名直出
        assert first["status"] == BUSINESS_STATUS_ACTIVE
        assert first["customer_name"] == customer["company_name"]

    def test_空明细_400(self, client: TestClient, normal: dict) -> None:
        customer = _customer(client, normal["headers"])
        resp = client.post(
            _BUSINESSES, json=_payload(customer["id"], items=[]), headers=normal["headers"]
        )
        assert resp.status_code == 400, resp.text

    def test_行金额乘积显式两位固化_回显与库值一致(self, client: TestClient, normal: dict) -> None:
        """数量(2位)×单价(4位) 可产 6 位小数：显式 HALF_UP 固化，与 MySQL 的
        DECIMAL(14,2) 舍入方向一致 —— 直写不 quantize 会被库端静默舍入，
        创建响应（读内存值）与复查（读库值）前后显示不一致。"""
        headers = normal["headers"]
        customer = _customer(client, headers)
        data = _create(
            client,
            headers,
            customer["id"],
            items=[_item(quantity="1.55", sale_price="3.3333", cost_amount=None)],
        )
        # 1.55 × 3.3333 = 5.166615 → 5.17（HALF_UP；MySQL 同值同向）
        item = client.get(f"{_BUSINESSES}/{data['id']}", headers=headers).json()["data"]["items"][0]
        assert item["sale_amount"] == "5.17"
        assert Decimal(data["total_sale_amount"]) == Decimal("5.17")

        # 库值与回显一致（无「内存 5.166615 / 库里 5.17」的漂移）
        factory = get_session_factory()
        with factory() as session:
            db_amount = session.execute(
                select(BusinessItem.sale_amount).where(BusinessItem.business_id == data["id"])
            ).scalar_one()
        assert db_amount == Decimal("5.17")

    def test_明细行数量为0_400(self, client: TestClient, normal: dict) -> None:
        customer = _customer(client, normal["headers"])
        resp = client.post(
            _BUSINESSES,
            json=_payload(customer["id"], items=[_item(quantity="0")]),
            headers=normal["headers"],
        )
        assert resp.status_code == 400, resp.text
        assert "数量" in resp.json()["message"], resp.text

    def test_非法产品类别_400(self, client: TestClient, normal: dict) -> None:
        """类别是折算阈值的行级锚点：脏类别取不到阈值参数。"""
        customer = _customer(client, normal["headers"])
        resp = client.post(
            _BUSINESSES,
            json=_payload(customer["id"], items=[_item(product_category="water")]),
            headers=normal["headers"],
        )
        assert resp.status_code == 400, resp.text
        assert "产品类别不合法" in resp.json()["message"], resp.text

    def test_销售单价非数字_400(self, client: TestClient, normal: dict) -> None:
        customer = _customer(client, normal["headers"])
        resp = client.post(
            _BUSINESSES,
            json=_payload(customer["id"], items=[_item(sale_price="abc")]),
            headers=normal["headers"],
        )
        assert resp.status_code == 400, resp.text
        assert "须为有效数字" in resp.json()["message"], resp.text

    def test_销售额由服务层固化_入参sale_amount不生效(
        self, client: TestClient, normal: dict
    ) -> None:
        """schema 无 sale_amount 入参，塞了也必须被忽略 —— 行金额只能由
        数量×单价计算，否则前端传值直接改业务销售额。"""
        headers = normal["headers"]
        customer = _customer(client, headers)
        item = _item(quantity="3", sale_price="12.34", sale_amount="88888.00")
        data = _create(client, headers, customer["id"], items=[item])

        detail = client.get(f"{_BUSINESSES}/{data['id']}", headers=headers).json()["data"]
        assert Decimal(detail["items"][0]["sale_amount"]) == Decimal("37.02")
        assert Decimal(detail["total_sale_amount"]) == Decimal("37.02")

    def test_特殊业务缺说明_400(self, client: TestClient, normal: dict) -> None:
        customer = _customer(client, normal["headers"])
        resp = client.post(
            _BUSINESSES,
            json=_payload(customer["id"], is_special=1),
            headers=normal["headers"],
        )
        assert resp.status_code == 400, resp.text
        assert "特殊业务必须填写说明" in resp.json()["message"], resp.text

    def test_特殊业务有说明创建成功_回显说明(self, client: TestClient, normal: dict) -> None:
        headers = normal["headers"]
        customer = _customer(client, headers)
        data = _create(client, headers, customer["id"], is_special=1, special_note="公司参与结算")
        assert data["is_special"] == 1
        assert data["special_note"] == "公司参与结算"


class TestEdit:
    """编辑：全量替换明细并单点重算汇总 —— 三处冗余（行 SUM / 业务冗余列 /
    数据库 SUM）不一致就说明重算有旁路。"""

    def test_编辑全量替换明细_三处汇总一致(self, client: TestClient, normal: dict) -> None:
        headers = normal["headers"]
        customer = _customer(client, headers)
        data = _create(
            client,
            headers,
            customer["id"],
            items=[
                _item(quantity="100", sale_price="100.00", cost_amount="30.00"),
                _item(quantity="50", sale_price="200.00", cost_amount="40.00"),
            ],
        )
        before = client.get(f"{_BUSINESSES}/{data['id']}", headers=headers).json()["data"]
        assert Decimal(before["total_sale_amount"]) == Decimal("20000")
        assert Decimal(before["total_cost_amount"]) == Decimal("70")

        updated = client.put(
            f"{_BUSINESSES}/{data['id']}",
            json=_payload(
                customer["id"],
                items=[_item(quantity="200", sale_price="50.00", cost_amount="30.00")],
            ),
            headers=headers,
        )
        assert updated.status_code == 200, updated.text
        assert Decimal(updated.json()["data"]["total_sale_amount"]) == Decimal("10000")

        detail = client.get(f"{_BUSINESSES}/{data['id']}", headers=headers).json()["data"]
        assert len(detail["items"]) == 1, "编辑是全量替换，旧行不得残留"
        item_sum_sale = sum(Decimal(i["sale_amount"]) for i in detail["items"])
        item_sum_cost = sum(Decimal(i["cost_amount"]) for i in detail["items"] if i["cost_amount"])
        assert item_sum_sale == Decimal(detail["total_sale_amount"])
        assert item_sum_cost == Decimal(detail["total_cost_amount"])

        # 数据库侧再核一次 SUM（接口回包来自内存计算，两处同源才可靠）
        factory = get_session_factory()
        with factory() as session:
            db_business = session.get(Business, data["id"])
            db_sale = (
                session.execute(
                    select(BusinessItem.sale_amount).where(BusinessItem.business_id == data["id"])
                )
                .scalars()
                .all()
            )
        assert sum(db_sale) == db_business.total_sale_amount == Decimal("10000.00")

    def test_编辑端点塞owner不生效_归属唯一走ownership(
        self, client: TestClient, normal: dict
    ) -> None:
        """归属刻意不出编辑端点：否则「改归属无留痕」的旁路破坏 append-only。"""
        headers = normal["headers"]
        customer = _customer(client, headers)
        owner = _personnel(client, headers)
        someone = _personnel(client, headers)
        data = _create(client, headers, customer["id"], owner_personnel_id=owner["id"])

        resp = client.put(
            f"{_BUSINESSES}/{data['id']}",
            json=_payload(customer["id"], owner_personnel_id=someone["id"]),
            headers=headers,
        )
        assert resp.status_code == 200, resp.text
        assert resp.json()["data"]["owner_personnel_id"] == owner["id"], resp.text

        logs = client.get(f"{_BUSINESSES}/{data['id']}/ownership-logs", headers=headers).json()[
            "data"
        ]
        assert logs["total"] == 0, "编辑端点改归属却无留痕"

    def test_终态业务编辑_400且明细未动(self, client: TestClient, normal: dict) -> None:
        """终态门槛与发货/回款写准入同口径：明细/汇总是 M3 核算基线，终态
        业务禁改 —— 父单据若放行，发货/回款服务「禁写终态」的防线就被绕过。"""
        headers = normal["headers"]
        customer = _customer(client, headers)
        data = _create(client, headers, customer["id"])
        _terminate(client, headers, data["id"])

        resp = client.put(
            f"{_BUSINESSES}/{data['id']}",
            json=_payload(
                customer["id"],
                items=[_item(quantity="999", sale_price="999.00")],
            ),
            headers=headers,
        )
        assert resp.status_code == 400, resp.text
        assert "已完成或已终止" in resp.json()["message"]

        # 明细与汇总未被改写（护栏：校验在 replace_items 之前生效）
        detail = client.get(f"{_BUSINESSES}/{data['id']}", headers=headers).json()["data"]
        assert Decimal(detail["total_sale_amount"]) == Decimal("10000")
        assert len(detail["items"]) == 1
        assert Decimal(detail["items"][0]["quantity"]) == Decimal("100")


class TestStatus:
    """状态机：终态不可逆是终态单据禁写的前提；留痕从创建首条记起。"""

    def test_终止留痕_终态禁再变更(self, client: TestClient, normal: dict) -> None:
        headers = normal["headers"]
        customer = _customer(client, headers)
        data = _create(client, headers, customer["id"])

        # 终止必填原因（schema 层校验，400）
        no_reason = client.put(
            f"{_BUSINESSES}/{data['id']}/status", json={"status": 3}, headers=headers
        )
        assert no_reason.status_code == 400, no_reason.text
        assert "终止业务必须填写原因" in no_reason.json()["message"], no_reason.text

        done = client.put(
            f"{_BUSINESSES}/{data['id']}/status",
            json={"status": 3, "reason": "客户取消订单"},
            headers=headers,
        )
        assert done.status_code == 200, done.text
        assert done.json()["data"]["status"] == 3

        # 终态不可再变更
        again = client.put(
            f"{_BUSINESSES}/{data['id']}/status",
            json={"status": 2, "reason": "强行翻转"},
            headers=headers,
        )
        assert again.status_code == 400, again.text
        assert "不可再变更状态" in again.json()["message"], again.text

        logs = client.get(f"{_BUSINESSES}/{data['id']}/status-logs", headers=headers).json()["data"]
        assert logs["total"] == 2, logs["total"]
        rows = logs["list"]  # 正序：创建首条在前
        assert (rows[0]["old_status"], rows[0]["new_status"]) == (1, 1)
        assert (rows[1]["old_status"], rows[1]["new_status"]) == (1, 3)
        assert rows[1]["change_reason"] == "客户取消订单"
        assert rows[1]["operator_name"] == normal["username"]


class TestOwnership:
    """归属变更：主表更新 + 留痕同事务；两者全相同则属无效操作。"""

    def test_归属变更_主表更新与留痕同生(self, client: TestClient, normal: dict) -> None:
        headers = normal["headers"]
        customer = _customer(client, headers)
        old_owner = _personnel(client, headers)
        data = _create(client, headers, customer["id"], owner_personnel_id=old_owner["id"])

        new_owner = _personnel(client, headers)
        new_assistant = _personnel(client, headers)
        resp = client.put(
            f"{_BUSINESSES}/{data['id']}/ownership",
            json={
                "new_owner_personnel_id": new_owner["id"],
                "new_assistant_personnel_id": new_assistant["id"],
                "reason": "人员调动",
            },
            headers=headers,
        )
        assert resp.status_code == 200, resp.text
        body = resp.json()["data"]
        assert body["owner_personnel_id"] == new_owner["id"]
        assert body["owner_personnel_name"] == new_owner["personnel_name"]
        assert body["assistant_personnel_id"] == new_assistant["id"]

        logs = client.get(f"{_BUSINESSES}/{data['id']}/ownership-logs", headers=headers).json()[
            "data"
        ]
        assert logs["total"] == 1, "首次归属变更只产生一条留痕"
        row = logs["list"][0]
        assert row["old_owner_personnel_id"] == old_owner["id"]
        assert row["new_owner_personnel_id"] == new_owner["id"]
        assert row["old_assistant_personnel_id"] is None
        assert row["new_assistant_personnel_id"] == new_assistant["id"]
        assert row["old_owner_name"] == old_owner["personnel_name"]
        assert row["new_owner_name"] == new_owner["personnel_name"]
        assert row["change_reason"] == "人员调动"
        assert row["operator_name"] == normal["username"]

        # owner_type 由 DB 固化断言（响应 DTO 不含该列；2=业务归属）
        factory = get_session_factory()
        with factory() as session:
            log = session.execute(
                select(OwnershipChangeLog).where(OwnershipChangeLog.business_id == data["id"])
            ).scalar_one()
        assert log.owner_type == OWNER_TYPE_BUSINESS, (
            f"owner_type 应为业务归属(2)：{log.owner_type}"
        )

    def test_归属未变化_400(self, client: TestClient, normal: dict) -> None:
        headers = normal["headers"]
        customer = _customer(client, headers)
        owner = _personnel(client, headers)
        data = _create(client, headers, customer["id"], owner_personnel_id=owner["id"])
        resp = client.put(
            f"{_BUSINESSES}/{data['id']}/ownership",
            json={"new_owner_personnel_id": owner["id"], "reason": "无变化"},
            headers=headers,
        )
        assert resp.status_code == 400, resp.text
        assert "归属未变化" in resp.json()["message"], resp.text

    def test_创建时负责人员已离职_400(self, client: TestClient, normal: dict) -> None:
        headers = normal["headers"]
        quit_owner = _personnel(client, headers)
        off = client.put(
            f"{_PERSONNEL}/{quit_owner['id']}/status", json={"status": 0}, headers=headers
        )
        assert off.status_code == 200, off.text
        customer = _customer(client, headers)
        resp = client.post(
            _BUSINESSES,
            json=_payload(customer["id"], owner_personnel_id=quit_owner["id"]),
            headers=headers,
        )
        assert resp.status_code == 400, resp.text
        assert "负责人员不存在或已离职" in resp.json()["message"], resp.text

    def test_归属端点负责人员不存在_400(self, client: TestClient, normal: dict) -> None:
        headers = normal["headers"]
        customer = _customer(client, headers)
        data = _create(client, headers, customer["id"])
        resp = client.put(
            f"{_BUSINESSES}/{data['id']}/ownership",
            json={"new_owner_personnel_id": 999999999, "reason": "填错人"},
            headers=headers,
        )
        assert resp.status_code == 400, resp.text
        assert "负责人员不存在或已离职" in resp.json()["message"], resp.text


class TestCustomerProtect:
    """客户引用保护：发货/回款单据 FK 挂在业务上，换客户或删除会让单据
    语义断裂 —— 服务层检查是唯一防线。"""

    def test_有发货或回款时换客户_400(self, client: TestClient, normal: dict) -> None:
        headers = normal["headers"]
        customer = _customer(client, headers)
        other = _customer(client, headers)
        data = _create(client, headers, customer["id"])

        # 有发货记录：禁换
        ship = client.post(
            "/api/v1/business/shipments",
            json={
                "business_id": data["id"],
                "ship_date": str(date.today()),
                "items": [_item(product_name="t_货", quantity="1", sale_price="10.00")],
            },
            headers=headers,
        )
        assert ship.status_code == 200, ship.text
        blocked_ship = client.put(
            f"{_BUSINESSES}/{data['id']}",
            json=_payload(other["id"]),
            headers=headers,
        )
        assert blocked_ship.status_code == 400, blocked_ship.text
        assert "已有发货或回款记录" in blocked_ship.json()["message"], blocked_ship.text

        # 解除发货后仍有回款：禁换（同一路径 has_docs）
        client.delete(f"/api/v1/business/shipments/{ship.json()['data']['id']}", headers=headers)
        receipt = client.post(
            "/api/v1/business/receipts",
            json={
                "business_id": data["id"],
                "receipt_date": str(date.today()),
                "amount": "100.00",
            },
            headers=headers,
        )
        assert receipt.status_code == 200, receipt.text
        blocked_receipt = client.put(
            f"{_BUSINESSES}/{data['id']}",
            json=_payload(other["id"]),
            headers=headers,
        )
        assert blocked_receipt.status_code == 400, blocked_receipt.text
        assert "已有发货或回款记录" in blocked_receipt.json()["message"], blocked_receipt.text

    def test_有发货或回款禁删_400(self, client: TestClient, normal: dict) -> None:
        headers = normal["headers"]
        customer = _customer(client, headers)
        data = _create(client, headers, customer["id"])

        ship = client.post(
            "/api/v1/business/shipments",
            json={
                "business_id": data["id"],
                "ship_date": str(date.today()),
                "items": [_item(product_name="t_货", quantity="1", sale_price="10.00")],
            },
            headers=headers,
        )
        assert ship.status_code == 200, ship.text
        blocked_ship = client.delete(f"{_BUSINESSES}/{data['id']}", headers=headers)
        assert blocked_ship.status_code == 400, blocked_ship.text
        assert "请改为已终止而非删除" in blocked_ship.json()["message"], blocked_ship.text

        # 解除发货后仍有回款：禁删（文案同一路径）
        client.delete(f"/api/v1/business/shipments/{ship.json()['data']['id']}", headers=headers)
        receipt = client.post(
            "/api/v1/business/receipts",
            json={
                "business_id": data["id"],
                "receipt_date": str(date.today()),
                "amount": "100.00",
            },
            headers=headers,
        )
        assert receipt.status_code == 200, receipt.text
        blocked_receipt = client.delete(f"{_BUSINESSES}/{data['id']}", headers=headers)
        assert blocked_receipt.status_code == 400, blocked_receipt.text
        assert "请改为已终止而非删除" in blocked_receipt.json()["message"], blocked_receipt.text

    def test_空业务删除成功_连带清理明细与留痕(self, client: TestClient, normal: dict) -> None:
        headers = normal["headers"]
        customer = _customer(client, headers)
        owner = _personnel(client, headers)
        new_owner = _personnel(client, headers)
        data = _create(
            client,
            headers,
            customer["id"],
            owner_personnel_id=owner["id"],
            items=[_item(quantity="1", sale_price="5.00")],
        )
        # 先造一次归属留痕，删除时须连带清理（无外键固化表）
        moved = client.put(
            f"{_BUSINESSES}/{data['id']}/ownership",
            json={"new_owner_personnel_id": new_owner["id"], "reason": "调动"},
            headers=headers,
        )
        assert moved.status_code == 200, moved.text

        deleted = client.delete(f"{_BUSINESSES}/{data['id']}", headers=headers)
        assert deleted.status_code == 200, deleted.text
        assert client.get(f"{_BUSINESSES}/{data['id']}", headers=headers).status_code == 404

        factory = get_session_factory()
        with factory() as session:
            assert session.get(Business, data["id"]) is None
            assert (
                session.execute(select(BusinessItem).where(BusinessItem.business_id == data["id"]))
                .scalars()
                .first()
                is None
            )
            # 状态留痕（business_status_logs）与归属留痕（owner_type=2）均无
            # 外键固化，删除时必须显式清
            assert (
                session.execute(
                    select(OwnershipChangeLog).where(
                        OwnershipChangeLog.business_id == data["id"],
                        OwnershipChangeLog.owner_type == OWNER_TYPE_BUSINESS,
                    )
                )
                .scalars()
                .first()
                is None
            )


class TestGrossProfit:
    """毛利润数据视图（只读派生）：锁定金标准口径 —— 率用精确分数、
    折算用化简式（率 0 / 销售额 0 防除零）、成本缺失行不进类别汇总。"""

    def test_毛利润基准值_率与折算锁定精确分数(self, client: TestClient, normal: dict) -> None:
        """销售 10000 / 成本 30 → 利润 9970、率 997/1000、市政阈值 30% →
        折算 10000 ×(997/1000 − 30/100) = 6970。"""
        headers = normal["headers"]
        customer = _customer(client, headers)
        data = _create(
            client,
            headers,
            customer["id"],
            items=[_item(quantity="1", sale_price="10000.00", cost_amount="30.00")],
        )
        resp = client.get(f"{_BUSINESSES}/{data['id']}/gross-profit", headers=headers)
        assert resp.status_code == 200, resp.text
        view = resp.json()["data"]
        assert view["as_of"] == str(date.today())

        row = view["rows"][0]
        assert Decimal(row["sale_amount"]) == Decimal("10000")
        assert Decimal(row["cost_amount"]) == Decimal("30")
        assert Decimal(row["profit_amount"]) == Decimal("9970")
        assert row["profit_rate"] == "997/1000", "率须精确分数串（约分后），不得截断"
        assert row["threshold"] == "30"
        assert Decimal(row["converted_profit"]) == Decimal("6970")
        assert row["missing_reason"] is None
        assert view["missing_count"] == 0

        cats = view["categories"]
        assert [c["product_category"] for c in cats] == ["municipal_water"]
        assert Decimal(cats[0]["sale_amount"]) == Decimal("10000")
        assert Decimal(cats[0]["profit_amount"]) == Decimal("9970")
        assert cats[0]["profit_rate"] == "997/1000"
        assert Decimal(cats[0]["converted_profit"]) == Decimal("6970")

    def test_毛利润_销售额0_全0不抛错_防除零(self, client: TestClient, normal: dict) -> None:
        """单价空 → 行销售额 0：率分母为 0 时必须返回 0 而非抛除零异常，
        否则录入期（补价前）打开毛利润视图就是 500。"""
        headers = normal["headers"]
        customer = _customer(client, headers)
        data = _create(
            client,
            headers,
            customer["id"],
            items=[_item(quantity="5", sale_price=None, cost_amount="0.00")],
        )
        resp = client.get(f"{_BUSINESSES}/{data['id']}/gross-profit", headers=headers)
        assert resp.status_code == 200, resp.text
        view = resp.json()["data"]
        row = view["rows"][0]
        assert Decimal(row["sale_amount"]) == Decimal("0")
        assert Decimal(row["profit_amount"]) == Decimal("0")
        assert row["profit_rate"] == "0"
        assert Decimal(row["converted_profit"]) == Decimal("0")
        assert view["missing_count"] == 0

        cats = view["categories"]
        assert [c["product_category"] for c in cats] == ["municipal_water"]
        assert Decimal(cats[0]["sale_amount"]) == Decimal("0")
        assert Decimal(cats[0]["converted_profit"]) == Decimal("0")

    def test_毛利润_成本缺失行_派生为空不计入类别汇总(
        self, client: TestClient, normal: dict
    ) -> None:
        headers = normal["headers"]
        customer = _customer(client, headers)
        data = _create(
            client,
            headers,
            customer["id"],
            items=[
                _item(
                    product_name="t_产品_市政",
                    quantity="1",
                    sale_price="5000.00",
                    cost_amount="100.00",
                ),
                _item(
                    product_name="t_产品_轨交",
                    product_category=PRODUCT_CATEGORY_RAIL_TRANSIT,
                    quantity="2",
                    sale_price=None,
                    cost_amount=None,
                ),
            ],
        )
        resp = client.get(f"{_BUSINESSES}/{data['id']}/gross-profit", headers=headers)
        assert resp.status_code == 200, resp.text
        view = resp.json()["data"]
        assert view["missing_count"] == 1

        rows = view["rows"]
        assert rows[1]["cost_amount"] is None
        assert rows[1]["profit_amount"] is None
        assert rows[1]["profit_rate"] is None
        assert rows[1]["converted_profit"] is None
        assert "成本缺失" in (rows[1]["missing_reason"] or ""), rows[1]
        assert rows[1]["threshold"] == "20"
        # 可计算行正常派生
        assert Decimal(rows[0]["profit_amount"]) == Decimal("4900")
        assert rows[0]["profit_rate"] == "49/50"

        cats = view["categories"]
        assert [c["product_category"] for c in cats] == ["municipal_water"], "缺失行不得进类别汇总"
        assert Decimal(cats[0]["sale_amount"]) == Decimal("5000")
        assert Decimal(cats[0]["cost_amount"]) == Decimal("100")
        assert Decimal(cats[0]["profit_amount"]) == Decimal("4900")
        assert cats[0]["profit_rate"] == "49/50"
        assert Decimal(cats[0]["converted_profit"]) == Decimal("3400")

    def test_毛利润asOf按业务日期取参数时点值_旧日期取旧值(
        self, client: TestClient, normal: dict, threshold_batch
    ) -> None:
        """改参数后按旧业务日期取值必须回溯旧阈值 —— 落库即漂移的「假快照」
        靠 asOf 回溯参数快照，M3 台账才是法定固化点。"""
        headers = normal["headers"]
        customer = _customer(client, headers)
        data = _create(
            client,
            headers,
            customer["id"],
            items=[_item(quantity="1", sale_price="10000.00", cost_amount="30.00")],
        )
        threshold_batch("municipal_water", "25")  # 立即生效（今日）

        today = client.get(
            f"{_BUSINESSES}/{data['id']}/gross-profit",
            params={"as_of": str(date.today())},
            headers=headers,
        )
        assert today.status_code == 200, today.text
        row_today = today.json()["data"]["rows"][0]
        assert row_today["threshold"] == "25"
        # 25%：10000 ×(997/1000 − 25/100) = 7470
        assert Decimal(row_today["converted_profit"]) == Decimal("7470")

        past = client.get(
            f"{_BUSINESSES}/{data['id']}/gross-profit",
            params={"as_of": "2026-02-17"},
            headers=headers,
        )
        assert past.status_code == 200, past.text
        row_past = past.json()["data"]["rows"][0]
        assert row_past["threshold"] == "30", "旧业务日期须回溯变更前的阈值"
        assert Decimal(row_past["converted_profit"]) == Decimal("6970")


class TestAudit:
    """写操作（含校验失败的）都在 operation_logs 留痕 —— 审计是验收项。"""

    def test_新增业务成功留审计(self, client: TestClient, normal: dict) -> None:
        headers = normal["headers"]
        customer = _customer(client, headers)
        _create(client, headers, customer["id"])
        logs = [
            log
            for log in _logs_of(normal["username"])
            if log.module == MODULE_BUSINESS and log.action == ACTION_CREATE
        ]
        assert logs, "新增业务未留操作日志"
        latest = logs[0]
        assert latest.success == 1
        assert "创建业务" in latest.summary

    def test_归属人员不存在失败留审计(self, client: TestClient, normal: dict) -> None:
        customer = _customer(client, normal["headers"])
        resp = client.post(
            _BUSINESSES,
            json=_payload(customer["id"], owner_personnel_id=999999999),
            headers=normal["headers"],
        )
        assert resp.status_code == 400, resp.text
        failed = [
            log
            for log in _logs_of(normal["username"])
            if log.module == MODULE_BUSINESS and log.success == 0
        ]
        assert failed, "失败操作未留痕"
        assert "负责人员不存在或已离职" in (failed[0].error_message or ""), failed[0]