group_xinghuo_jinrong/tests/test_convert_integration.py

"""T-9 真 MySQL 端到端集成测试（架构 §8/§10 · 开发计划 §7.2 DoD · §12 R15 迁入）。

**链路（T+1 受理/确认分离模型，T-9 起）**——两段，缺一不可：

```
T 日受理段：TestClient(main app) → POST /api/simulate/trade
            → trade_gateway 分派 → convert_service.accept_convert
            （只校验 + 落受理单 → **202 + 受理回执**，不扣份额 / 不折算 / 不写流水）
T+1 确认段：POST /api/admin/convert/confirm?accept_date=<业务日>
            → confirm_service.confirm_batch → confirm_one
            （扣批次 + 两条流水 + 两端持仓 + 计费明细 → **200 + 折算结果**）
查询：      GET /api/simulate/trade/convert/{gid}（PRD §5.7）
撤单：      POST /api/simulate/trade/convert/{gid}/cancel（PRD §5.5）
```

**为什么 v1.0 的断言整批失效**：v1.0 是「受理即确认」的实时链路——一次 POST 里
完成折算并回 200，故旧用例直接断言 `out_amount` / `lot_breakdown` / `core_trade` 两条
流水。T+1 模型下**受理段这三样全都没有**（这是模型的核心，PRD §2.6），断言必须
挪到确认之后，否则测的就不是本项目要实现的东西。

**为什么必须有真库这一层**：单测层只验到「路由层不加工、原样透传」（见
`test_trade_gateway.py`），而折算数字是否与 PRD §5.3 逐项吻合、两条流水是否
真的同组同事务、持仓/批次是否真的扣减——只有真库能证明。

隔离策略（架构 §10）——**三条同时成立，缺一即污染种子**：

1. **id 前缀**：`CNV-TEST-` / `TRD-TEST-`（monkeypatch `convert_service._new_id`
   与 `confirm_service.new_id` 两个**单号生成点**）；
2. **数据自建**：客户/产品/费率/净值/批次全部 `CNVTEST` 前缀自建，**绝不碰种子**——
   `risk_demo_env` 的 teardown 只清 `TRD-TEST-` 前缀与时间窗，**不还原
   `core_share_lot`/`core_holding`**；若借种子客户跑转换，扣掉的份额会跨用例污染
   `test_integration_risk.py`；
3. **teardown 全清**：函数级 fixture 按前缀删两库全部自建行（幂等，seed 失败亦可清）。

⚠️ **净值日期必须锚在「受理日」**：确认段按 `get_nav_on(pid, accept_date)` **精确匹配
T 日净值**（T-2b：不匹配即 `nav_pending`，而恒回退旧净值的 `get_nav_as_of` 会让
`nav_pending` 永不触发——见开发计划 v2.0 关键裁定）。故种子净值与批次日期都由
`_accept_date_now()`（**调生产函数**，不手算）推出，与受理段落库的受理日同源。
"""

from __future__ import annotations

from datetime import date, datetime, time, timedelta
from decimal import Decimal
from uuid import uuid4

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

from conftest import ensure_risk_demo_ready

from app.config.settings import settings  # noqa: E402
from app.gateway import trade_gateway  # noqa: E402
from app.main import app  # noqa: E402
from app.repository.core_ro import CoreReadOnlyRepository  # noqa: E402
from app.service.convert import confirm_service  # noqa: E402
from app.service.convert import convert_service as cs  # noqa: E402
from app.service.convert.calc import (  # noqa: E402
    convert_amount,
    diff_fee,
    in_qty,
    lot_amount,
    lot_fee,
)
from app.service.convert.trading_calendar import (  # noqa: E402
    parse_cutoff,
    resolve_accept_date,
)
from app.service.risk import redis_gateway  # noqa: E402

ensure_risk_demo_ready()

#: 交易 owner 头（撤单用）——`risk_demo` 演示角色。
DEMO = {"X-Debug-Role": "risk_demo", "X-Debug-Actor": "STAFF-DEMO"}
#: admin 级头（确认 / 查询用）——`risk_officer` 是 `STAFF_FULL_ACCESS_ROLES` 唯一成员，
#: 也是**查询 scope 闸门**的合法主体（`risk_demo` 查不了：`assert_customer_access`
#: 只认 customer/advisor/risk_officer，见 `app/api/deps.py`）。
OPS = {"X-Debug-Role": "risk_officer", "X-Debug-Actor": "STAFF-OPS"}

# ── 红线 7 的两道闸门：各自的主体（撤单=交易 owner / 查询=查询 scope）──
#: 非本人客户（**不存在于任何库**——`assert_customer_access` 的 customer 分支只比
#: `customer_id`、不查库，故无需自建行；仅用于触发 403 与留痕）。
STRANGER_ACTOR = "CUST-OTHERTEST"
STRANGER = {"X-Debug-Role": "customer", "X-Debug-Actor": STRANGER_ACTOR}
#: 已分配顾问（seed 里为自建客户插了归属行，见 `_seed`）。
ADVISOR = {"X-Debug-Role": "advisor", "X-Debug-Actor": "STAFF-10086"}

#: 路径模板（PRD §5.5 / §5.7 —— 见开发计划 §1.3 裁定 11）。
_CANCEL = "/api/simulate/trade/convert/{gid}/cancel"
_QUERY = "/api/simulate/trade/convert/{gid}"

# ── 隔离种子常量（全部带 CNVTEST 前缀）──────────────────────────────
CUSTOMER = "CUST-CNVTEST"
#: 客户本人头（撤单与查询都合法 —— 交易 owner 闸门与查询 scope 闸门的交集）。
SELF = {"X-Debug-Role": "customer", "X-Debug-Actor": CUSTOMER}
PROD_OUT = "PROD-CNVTESTO"  # 债基，申购费率 0.0030
PROD_IN = "PROD-CNVTESTI"  # 股基，申购费率 0.0080（**高于**转出端 → 补差费非零）
PROD_CROSS = "PROD-161725"  # 真库既有：易方模拟基金 / TA-CN-002（跨主体负例）
COMPANY = "华夏模拟基金"
TA = "TA-CN-001"
OUT_RATE = Decimal("0.0030")
IN_RATE = Decimal("0.0080")
OUT_NAV = Decimal("1.3604")  # T-2b：真实净值（06-seed-nav.sql，下同）
IN_NAV = Decimal("1.9194")
FEE_TIERS = [
    (0, 7, "0.0150"),
    (7, 30, "0.0100"),
    (30, 180, "0.0050"),
    (180, 365, "0.0025"),
    (365, None, "0.0000"),
]
#: PRD §5.3 场景：两批次 30000 份持 100 天 + 20000 份持 3 天 → 费用档 0.0050 / 0.0150
LOT_SPEC = [("LOT-CNVTEST-A1", "30000", 100), ("LOT-CNVTEST-A2", "20000", 3)]
TOTAL_QTY = sum(Decimal(q) for _, q, _ in LOT_SPEC)

#: 受理回执与**未确认查询**中**绝不允许出现**的折算字段（Q2 / PRD §5.3.1、§5.3.2：
#: T 日未知价法不折算）。与 `app/api/convert_admin.py::_CONFIRMED_ONLY_FIELDS` 同源
#: 同义——此处独立列一份是有意的：若哪天有人把折算字段漏进受理响应/未确认查询，
#: 这条用例要能独立发现。
#: （`forced_full_transfer` 不在列 —— 它是**受理段**的决策，受理即已确定并回传。）
NOT_IN_ACCEPT = (
    "out_trade_id", "out_nav", "out_amount", "lot_count", "lot_breakdown",
    "redeem_fee", "in_trade_id", "in_nav", "convert_amount", "diff_fee",
    "in_amount", "in_qty", "rounding_diff", "nav_date", "nav_stale",
)

_GROUP_LIKE = "CNV-TEST-%"


class FakePub:
    def __init__(self) -> None:
        self.messages: list = []
        self.deletes: list = []

    def publish(self, channel, payload):
        self.messages.append((channel, payload))

    def delete(self, *keys):
        self.deletes.append(keys)


def _test_new_id(prefix: str, now: datetime) -> str:
    """单号工厂替身：`CNV-TEST-xxxx` / `TRD-TEST-xxxx`（架构 §10 前缀约定）。

    受理段（`convert_service._new_id`）与确认段（`confirm_service.new_id`）是**两个
    独立生成点**（T-7 抽到 `format.new_id` 后仍各自持有模块级名字），故 fixture 需
    同时打两处；只打一处会让确认段落下的 `TRD-` 流水逃出 teardown 前缀清理。
    """
    return f"{prefix}-TEST-{uuid4().hex[:8].upper()}"


# ── 种子与清理 ──────────────────────────────────────────────────────
def _accept_date_now() -> date:
    """此刻提交会落在哪个**受理日**——调生产函数（禁手算 · 自检第 13 问）。

    运行时刻可能已过 15:00 截点或逢非交易日（R-5）→ 受理日顺延到下一交易日；
    种子净值与批次日期都必须锚在这个日期上，否则确认段 `get_nav_on(accept_date)`
    取不到净值 → `nav_pending`，用例会以「像业务 bug」的方式失败。

    交易日历判定用**只读仓储**（`fixture` 里的 `risk_demo_env["core"]` 是裸 `Engine`，
    没有 `is_open`）：与生产侧 `accept_convert` 取日历的口径同源。
    """
    return resolve_accept_date(
        datetime.now(),
        CoreReadOnlyRepository().is_open,
        parse_cutoff(settings.convert_cutoff_time),
    )


def _seed(core) -> date:
    """自建 PRD §5.3 场景（客户 + 双产品 + 费率 + 净值 + 双批次 + 持仓）。

    返回本场景的**受理日**（fixture 透出给用例，避免用例再算一次）。
    """
    accept_date = _accept_date_now()
    base = datetime.combine(accept_date, time(10, 0))  # 固定钟点 → hold_days 恒等于 LOT_SPEC 天数
    with core.begin() as conn:
        conn.execute(
            text(
                "INSERT INTO core_customer (customer_id, display_name, open_date)"
                " VALUES (:c, 'T9集成测试', :d)"
            ),
            {"c": CUSTOMER, "d": accept_date},
        )
        conn.execute(
            text(
                "INSERT INTO core_customer_risk (customer_id, risk_code, evaluated_at, expires_at)"
                " VALUES (:c, 'C5', :t, :exp)"
            ),
            {"c": CUSTOMER, "t": base - timedelta(days=30), "exp": base + timedelta(days=300)},
        )
        # 客户-代理人归属：为「查询 scope 闸门」的 advisor 分支备料（红线 7 用例）。
        # advisor 必须已存在于 `core_staff`（FK `fk_ca_advisor`）→ 复用种子 `STAFF-10086`
        # （`staff_type='advisor'`，02-seed-base.sql:40）；**只新增一行归属**、不碰种子行。
        conn.execute(
            text(
                "INSERT INTO core_customer_advisor (customer_id, advisor_id, rel_status,"
                " effective_from) VALUES (:c, 'STAFF-10086', 'active', :d)"
            ),
            {"c": CUSTOMER, "d": accept_date},
        )
        for pid, name, ptype, rate in [
            (PROD_OUT, "T9转出基金", "bond", OUT_RATE),
            (PROD_IN, "T9转入基金", "stock", IN_RATE),
        ]:
            conn.execute(
                text(
                    "INSERT INTO core_product (product_id, product_name, min_risk_code,"
                    " product_type, can_subscribe, can_redeem, subscribe_fee_rate,"
                    " fund_company, ta_code)"
                    " VALUES (:p, :n, 'R2', :t, 1, 1, :r, :co, :ta)"
                ),
                {"p": pid, "n": name, "t": ptype, "r": str(rate), "co": COMPANY, "ta": TA},
            )
        for mh, mh_max, rate in FEE_TIERS:
            conn.execute(
                text(
                    "INSERT INTO core_fee_rule (product_id, fee_type, min_hold_days,"
                    " max_hold_days, rate) VALUES (:p, 'redeem', :mh, :mm, :r)"
                ),
                {"p": PROD_OUT, "mh": mh, "mm": mh_max, "r": rate},
            )
        # 两端净值都落在**受理日**（确认段 `get_nav_on(pid, accept_date)` 精确匹配）：
        # 旧版只给转入端插净值——v1.0 转出端用批次自身 nav、不需要净值表，T+1 需要。
        for pid, nav in ((PROD_OUT, OUT_NAV), (PROD_IN, IN_NAV)):
            conn.execute(
                text(
                    "INSERT INTO core_product_nav (product_id, nav, daily_chg_pct, nav_date)"
                    " VALUES (:p, :n, 0, :d)"
                ),
                {"p": pid, "n": str(nav), "d": accept_date},
            )
        for lot_id, qty, days in LOT_SPEC:
            conn.execute(
                text(
                    "INSERT INTO core_share_lot (lot_id, customer_id, product_id, qty,"
                    " remain_qty, nav, confirmed_at) VALUES (:l, :c, :p, :q, :q, :n, :cat)"
                ),
                {
                    "l": lot_id,
                    "c": CUSTOMER,
                    "p": PROD_OUT,
                    "q": qty,
                    "n": str(OUT_NAV),
                    "cat": base - timedelta(days=days),
                },
            )
        conn.execute(
            text(
                "INSERT INTO core_holding (customer_id, product_id, qty, cost_amount,"
                " market_value, pnl_pct, as_of) VALUES (:c, :p, :q, :cost, :mv, 0, :d)"
            ),
            {
                "c": CUSTOMER,
                "p": PROD_OUT,
                "q": str(TOTAL_QTY),
                "cost": str(TOTAL_QTY * OUT_NAV),
                "mv": str(TOTAL_QTY * OUT_NAV),
                "d": accept_date,
            },
        )
    return accept_date


def _cleanup(core, agent) -> None:
    """按前缀清两库全部自建行（幂等）。

    ⚠️ 删除顺序受 FK 约束：`core_convert_request` 引用 `core_product` 与
    `core_customer` —— **T-9 起 convert 请求会落受理单**（此前 v1.0 实时链路
    不落此表），故必须先于 `core_product` 删除，否则
    `ft_creq_to_product` 外键会挡住删产品（T-9 实测：残留受理单导致
    fixture teardown 报 IntegrityError）。
    """
    core_sqls = [
        "DELETE FROM core_convert_request WHERE customer_id = :c",
        f"DELETE FROM core_convert_lot_detail WHERE convert_group_id LIKE '{_GROUP_LIKE}'",
        # T-16 起确认事务写 core_cash_flow（remark = 'convert:{gid}'），
        # fk_cf_customer 挡在 core_customer 之前，必须先清（同 T-9 的 FK 坑）
        f"DELETE FROM core_cash_flow WHERE remark LIKE 'convert:{_GROUP_LIKE}'",
        "DELETE FROM core_trade WHERE customer_id = :c",
        "DELETE FROM core_share_lot WHERE customer_id = :c",
        "DELETE FROM core_holding WHERE customer_id = :c",
        "DELETE FROM core_customer_advisor WHERE customer_id = :c",
        "DELETE FROM core_customer_risk WHERE customer_id = :c",
        "DELETE FROM core_fee_rule WHERE product_id LIKE 'PROD-CNVTEST%'",
        "DELETE FROM core_product_nav WHERE product_id LIKE 'PROD-CNVTEST%'",
        "DELETE FROM core_product WHERE product_id LIKE 'PROD-CNVTEST%'",
        "DELETE FROM core_customer WHERE customer_id = :c",
    ]
    with core.begin() as conn:
        for sql in core_sqls:
            conn.execute(text(sql), {"c": CUSTOMER})
    with agent.begin() as conn:
        conn.execute(text(f"DELETE FROM risk_convert_detail WHERE convert_group_id LIKE '{_GROUP_LIKE}'"))
        conn.execute(text("DELETE FROM risk_alert WHERE customer_id = :c"), {"c": CUSTOMER})
        conn.execute(text("DELETE FROM risk_suitability_log WHERE customer_id = :c"), {"c": CUSTOMER})
        conn.execute(text("DELETE FROM audit_log WHERE customer_id = :c"), {"c": CUSTOMER})
        # 鉴权拒绝的**双写**第二站（`input_guard_log` 无 customer_id 列，只能按 actor 清）：
        # 仅「非本人客户查询」会走到（撤单拒绝走 agent_type='platform'，只写 audit_log）。
        conn.execute(
            text("DELETE FROM input_guard_log WHERE actor_id = :a"), {"a": STRANGER_ACTOR}
        )


@pytest.fixture()
def conv_env(risk_demo_env, monkeypatch):
    """自建隔离种子 + 注入 TEST 前缀 id 工厂（两处生成点） + 函数级全清。"""
    core = risk_demo_env["core"]
    agent = risk_demo_env["agent"]
    monkeypatch.setattr(cs, "_new_id", _test_new_id)
    monkeypatch.setattr(confirm_service, "new_id", _test_new_id)
    monkeypatch.setattr(trade_gateway, "_new_trade_id", lambda now: _test_new_id("TRD", now))
    _cleanup(core, agent)  # 先清后建：上一轮异常退出也能自愈
    accept_date = _seed(core)
    try:
        yield {"core": core, "agent": agent, "accept_date": accept_date}
    finally:
        _cleanup(core, agent)


@pytest.fixture()
def client(conv_env, monkeypatch):
    fake = FakePub()
    with TestClient(app) as c:
        monkeypatch.setattr(redis_gateway, "_gateway", fake)
        yield c


def _body(qty="50000", cid=None, frm=PROD_OUT, to=PROD_IN):
    payload = {
        "customer_id": CUSTOMER,
        "trade_type": "convert",
        "from_product_id": frm,
        "to_product_id": to,
        "qty": qty,
    }
    if cid:
        payload["client_request_id"] = cid
    return payload


def _one(engine, sql: str, **params):
    with engine.connect() as conn:
        return conn.execute(text(sql), params).mappings().first()


def _all(engine, sql: str, **params):
    with engine.connect() as conn:
        return conn.execute(text(sql), params).mappings().all()


def _n(engine, sql: str, **params) -> int:
    with engine.connect() as conn:
        return conn.execute(text(sql), params).scalar_one()


# ── 两段链路的统一入口 ──────────────────────────────────────────────
def _accept(client, *, expect_status: str = "accepted", **kw) -> dict:
    """T 日受理：**必须 202 + 受理回执**（PRD §5.3.1 / 开发计划 T-9）。

    `expect_status`：幂等命中会**原样透出既有单的状态**（可能是终态 `confirmed`），
    故重试场景需显式放行 —— 这也是 PRD「客户端据此得知该请求已处理完」的落点。
    """
    r = client.post("/api/simulate/trade", json=_body(**kw), headers=DEMO)
    assert r.status_code == 202, f"受理应回 202，实际 {r.status_code}：{r.text}"
    body = r.json()
    assert body["accepted"] is True
    assert body["status"] == expect_status
    assert body["convert_group_id"].startswith("CNV-TEST-")
    for key in NOT_IN_ACCEPT:
        assert key not in body, f"受理回执不得含折算字段 {key}（Q2：T 日未知价法不折算）"
    return body


def _confirm(client, as_of: str) -> dict:
    """T+1 确认：**必须 200**；`accept_date` 传**业务确认日**（裁定 12）。"""
    r = client.post(f"/api/admin/convert/confirm?accept_date={as_of}", headers=OPS)
    assert r.status_code == 200, f"确认批处理应回 200，实际 {r.status_code}：{r.text}"
    return r.json()


def _confirm_accepted(client, accepted: dict) -> dict:
    """按受理回执的 `confirm_date`（T+1 交易日）触发确认，返回单笔结果。

    `confirm_date` 为 `None` 只在**交易日历数据边界**（种子用尽）出现；
    此时回退用受理日 —— 批处理捞单窗口上界是闭区间，仍能捞到该单。
    """
    as_of = accepted.get("confirm_date") or accepted["accept_date"]
    batch = _confirm(client, as_of)
    assert batch["locked"] is True, "测试串行执行不应撞批处理锁"
    assert batch["scanned"] >= 1, f"确认批处理应捞到本用例的受理单：{batch}"
    hit = [x for x in batch["results"] if x["convert_group_id"] == accepted["convert_group_id"]]
    assert hit, f"确认结果中缺本用例的单：{batch['results']}"
    assert hit[0]["status"] == "confirmed", f"确认应成功：{hit[0]}"
    return {"batch": batch, "one": hit[0]}


def _query(client, gid: str) -> dict:
    """查询受理单（PRD §5.7）——**查询 scope 闸门**，用 `risk_officer`。"""
    r = client.get(f"/api/simulate/trade/convert/{gid}", headers=OPS)
    assert r.status_code == 200, f"查询应回 200，实际 {r.status_code}：{r.text}"
    return r.json()


# ── 1. 端到端：与 PRD §5.3 示例逐项吻合 ─────────────────────────────
def test_convert_end_to_end_matches_prd_5_3(client, conv_env):
    """跨两批次转换走通，四段金额与 PRD §5.3 示例**逐项吻合**。

    期望值一律**调生产纯函数**得出（禁手算 · 自检第 13 问），再与 PRD 示例数字对表。
    折算结果有**两条读取路径**（首次=确认响应，重放=查询接口 rebuild），本用例走
    查询路径（顺带验 Q2 字段白名单），两条路径的逐字节一致性由用例 5 单独把关。
    """
    accepted = _accept(client)
    assert accepted["requested_qty"] == "50000.00"
    assert accepted["qty"] == "50000.00"
    assert accepted["estimated"] is True  # T 日未知价法：受理承诺量为预估值
    assert accepted["confirm_date"] and accepted["available_date"], "T+1/T+2 应由交易日历推出"

    # 受理段**不写流水**（PRD §2.6 核心）
    core = conv_env["core"]
    assert _n(core, "SELECT COUNT(*) FROM core_trade WHERE customer_id = :c", c=CUSTOMER) == 0

    _confirm_accepted(client, accepted)
    body = _query(client, accepted["convert_group_id"])

    assert body["status"] == "confirmed"
    assert body["confirmed"] is True
    assert body["actual_qty"] == "50000.00"  # 全额转出 → 实际=申请
    assert body["forced_full_transfer"] is False
    assert body["lot_count"] == 2

    by_days = {b["hold_days"]: b for b in body["lot_breakdown"]}
    assert set(by_days) == {100, 3}, "两批次持有期应为 100 / 3 天"

    exp_out = Decimal("0")
    exp_fee = Decimal("0")
    for days, rate in ((100, Decimal("0.0050")), (3, Decimal("0.0150"))):
        leg = by_days[days]
        assert Decimal(leg["fee_rate"]) == rate, f"持有 {days} 天的费率档"
        leg_amount = lot_amount(Decimal(leg["qty"]), Decimal(leg["nav"]))
        leg_fee = lot_fee(leg_amount, rate)
        assert Decimal(leg["fee_amount"]) == leg_fee, f"持有 {days} 天的逐批费用"
        exp_out += leg_amount
        exp_fee += leg_fee  # 逐批舍入后求和（PRD §2.5）

    exp_conv = convert_amount(exp_out, exp_fee)
    exp_diff = diff_fee(exp_conv, OUT_RATE, IN_RATE)
    exp_in_amount = convert_amount(exp_conv, exp_diff)

    assert Decimal(body["out_amount"]) == exp_out
    assert Decimal(body["redeem_fee"]) == exp_fee
    assert Decimal(body["convert_amount"]) == exp_conv
    assert Decimal(body["diff_fee"]) == exp_diff
    assert Decimal(body["in_amount"]) == exp_in_amount
    assert Decimal(body["in_qty"]) == in_qty(exp_in_amount, IN_NAV)

    # 与 PRD §5.3 示例数字对表（该示例由 calc_convert_demo.py 实算回填，验收 20）
    assert (exp_out, exp_fee, exp_conv, exp_diff, exp_in_amount) == (
        Decimal("68020.00"),
        Decimal("612.18"),
        Decimal("67407.82"),
        Decimal("333.36"),
        Decimal("67074.46"),
    )
    assert Decimal(body["in_qty"]) == Decimal("34945.54")
    # 转入端费率高于转出端 → 补差费应 > 0（验收 19 的非零场景）
    assert exp_diff > 0
    # 转出端成交净值取 **T 日净值**（不是批次自身净值）—— T+1 相对 v1.0 的实质变更
    assert Decimal(body["out_nav"]) == OUT_NAV
    assert body["nav_date"] == str(conv_env["accept_date"])


# ── 2. 两条流水同组、id 前缀正确（**确认后才有**）───────────────────
def test_convert_writes_two_trades_in_same_group(client, conv_env):
    core = conv_env["core"]
    accepted = _accept(client)
    gid = accepted["convert_group_id"]

    # 受理段不写流水（T+1 核心：T 日只落受理单）
    assert _n(
        core, "SELECT COUNT(*) FROM core_trade WHERE customer_id = :c", c=CUSTOMER
    ) == 0, "受理段不得写 core_trade（PRD §2.6）"

    _confirm_accepted(client, accepted)

    rows = {
        row["trade_type"]: row
        for row in [
            dict(r)
            for r in _all(
                core,
                "SELECT trade_type, trade_id, amount, convert_group_id FROM core_trade"
                " WHERE customer_id = :c",
                c=CUSTOMER,
            )
        ]
    }
    assert set(rows) == {"redeem", "subscribe"}, "一次转换落两条流水（R-b）"
    assert {r["convert_group_id"] for r in rows.values()} == {gid}
    assert all(r["trade_id"].startswith("TRD-TEST-") for r in rows.values())
    # 转出端金额 = 原始转出额（未扣费）
    assert Decimal(str(rows["redeem"]["amount"])) == Decimal("68020.00")


# ── 3. 持仓与批次如实变动（**确认后**）─────────────────────────────
def test_convert_updates_holding_and_lots(client, conv_env):
    core = conv_env["core"]
    accepted = _accept(client)

    # 受理段：批次与持仓**分毫不动**（占用由「未终态受理单」推导，R-3，不写份额）
    remain_before = _one(
        core,
        "SELECT SUM(remain_qty) AS s FROM core_share_lot"
        " WHERE customer_id = :c AND product_id = :p",
        c=CUSTOMER,
        p=PROD_OUT,
    )
    assert Decimal(str(remain_before["s"])) == TOTAL_QTY
    assert _n(
        core,
        "SELECT COUNT(*) FROM core_holding WHERE customer_id = :c AND product_id = :p",
        c=CUSTOMER,
        p=PROD_IN,
    ) == 0, "受理段不得建转入端持仓"

    _confirm_accepted(client, accepted)

    # 转出端批次：两批 remain_qty 归零（全部转出）—— 必须限定 product_id，
    # 否则会把**转入端新建批次**算进同一客户的聚合（首版断言即栽在这里）
    remain = _one(
        core,
        "SELECT SUM(remain_qty) AS s FROM core_share_lot"
        " WHERE customer_id = :c AND product_id = :p",
        c=CUSTOMER,
        p=PROD_OUT,
    )
    assert Decimal(str(remain["s"])) == Decimal("0.0000")

    # 转出端持仓归零但**行保留**（D9/P2）
    out_holding = _one(
        core,
        "SELECT qty FROM core_holding WHERE customer_id = :c AND product_id = :p",
        c=CUSTOMER,
        p=PROD_OUT,
    )
    assert out_holding is not None, "转出归零保留行（不可 DELETE）"
    assert Decimal(str(out_holding["qty"])) == Decimal("0.0000")

    # 转入端持仓新建且份额 = 响应 in_qty
    in_holding = _one(
        core,
        "SELECT qty FROM core_holding WHERE customer_id = :c AND product_id = :p",
        c=CUSTOMER,
        p=PROD_IN,
    )
    assert in_holding is not None
    assert Decimal(str(in_holding["qty"])) == Decimal("34945.54")


# ── 4. 两段镜像进度：pending → completed，审计如实落库 ─────────────
def test_convert_detail_completed_and_audited(client, conv_env):
    agent = conv_env["agent"]
    accepted = _accept(client)
    gid = accepted["convert_group_id"]

    # 受理段：镜像为 pending（agent 写入进度，非业务状态 —— T-4 契约）
    pending = _one(
        agent,
        "SELECT status FROM risk_convert_detail WHERE convert_group_id = :g",
        g=gid,
    )
    assert pending is not None and pending["status"] == "pending", "受理段镜像应为 pending"

    _confirm_accepted(client, accepted)

    detail = _one(
        agent,
        "SELECT status, fee_amount FROM risk_convert_detail WHERE convert_group_id = :g",
        g=gid,
    )
    assert detail["status"] == "completed"
    assert Decimal(str(detail["fee_amount"])) == Decimal("612.18")

    # 审计以 convert_group_id 为关联主键（L-2：一次转换 = 一条主审计）
    # 两段各一条：受理段 `accepted` / 确认段 `confirmed`（`write_convert_audit` 唯一写入点）
    assert _n(
        agent,
        "SELECT COUNT(*) FROM audit_log WHERE customer_id = :c AND decision = 'accepted'"
        " AND event_type = 'convert_request'",
        c=CUSTOMER,
    ) == 1
    assert _n(
        agent,
        "SELECT COUNT(*) FROM audit_log WHERE customer_id = :c AND decision = 'confirmed'"
        " AND event_type = 'convert_request'",
        c=CUSTOMER,
    ) == 1
    # 网关不重复写 trade_request（T-9：convert 审计归 convert_service）
    assert _n(
        agent,
        "SELECT COUNT(*) FROM audit_log WHERE customer_id = :c AND event_type = 'trade_request'",
        c=CUSTOMER,
    ) == 0


# ── 5. 幂等 + 「同一结果两条路径」逐字节一致 ────────────────────────
#: 首次（确认段按公式算）与重放（查询接口按 Core 重建）必须**逐字节相同**的字段。
#: 这份清单是「同一响应、两条路径」的一致性契约：数值相等但字符串不等
#: （如 `34945.54` vs `34945.5400`）会被前端/对账按字符串比对误判为两笔。
REPLAY_IDENTICAL_FIELDS = (
    "requested_qty", "actual_qty",
    "out_nav", "out_amount", "redeem_fee",
    "in_nav", "convert_amount", "diff_fee", "in_amount", "in_qty",
    "rounding_diff", "out_subscribe_fee_rate", "in_subscribe_fee_rate",
)


def test_convert_idempotent_retry_returns_byte_identical_response(client, conv_env):
    """同键重试 → 命中幂等、**不重开受理单**；确认后查询结果与确认响应逐字节相同。

    **T-9 与 v1.0 的差异**：v1.0 的「重放」是同一 POST 再走一次实时链路，直接比对
    两次响应体。T+1 下同一 POST 重试只该命中**受理幂等**（受理单已存在即原样透出），
    故一致性契约落在「确认响应（首次算）vs 查询响应（按 Core 重建）」这一对上——
    这正是展示位数回归闸门的真实落点（`DECIMAL(18,4)` 重建后不收敛会吐出 4 位）。
    """
    core = conv_env["core"]
    first = _accept(client, cid="T9-IT-IDEM-1")
    second = _accept(client, cid="T9-IT-IDEM-1")
    assert second["convert_group_id"] == first["convert_group_id"], "同键必须命中同一受理单"
    assert second["idempotent"] is True
    assert second["requested_qty"] == first["requested_qty"]
    assert _n(
        core, "SELECT COUNT(*) FROM core_convert_request WHERE customer_id = :c", c=CUSTOMER
    ) == 1, "同键重试不得重开受理单（uk_idem 守卫生效）"

    confirmed = _confirm_accepted(client, first)["one"]
    queried = _query(client, first["convert_group_id"])
    for field in REPLAY_IDENTICAL_FIELDS:
        assert queried[field] == confirmed[field], (
            f"查询重建的 {field} 应与确认响应**逐字节**相同："
            f"{queried[field]!r} != {confirmed[field]!r}"
        )
    # 逐批明细同样逐字段一致（qty / nav / fee_rate 的位数也是展示规格的一部分）
    assert queried["lot_breakdown"] == confirmed["lot_breakdown"]

    # 已确认的单再带同键重试 → 原样透出终态（`_accept_idempotent` 透传既有状态）
    third = _accept(client, cid="T9-IT-IDEM-1", expect_status="confirmed")
    assert third["convert_group_id"] == first["convert_group_id"]
    assert third["status"] == "confirmed", "终态原样透出，客户端据此得知请求已处理完"

    assert _n(core, "SELECT COUNT(*) FROM core_trade WHERE customer_id = :c", c=CUSTOMER) == 2
    assert _n(
        core,
        "SELECT COUNT(*) FROM core_share_lot"
        " WHERE customer_id = :c AND product_id = :p AND remain_qty > 0",
        c=CUSTOMER,
        p=PROD_OUT,
    ) == 0, "重试不得再扣一次转出端份额"
    assert _n(
        core,
        "SELECT COUNT(*) FROM core_share_lot WHERE customer_id = :c AND product_id = :p",
        c=CUSTOMER,
        p=PROD_IN,
    ) == 1, "重试不得再建一条转入端批次"


# ── 5b. 展示位数规格（真实公告口径，见 convert_service._q / format.q）──
def test_convert_response_field_scales(client, conv_env):
    """响应的**展示位数**按字段分类收敛（真实公告口径），不随数据源漂移。

    真实依据：金额/份额「四舍五入保留至小数点后两位」；净值保留 4 位、
    第 5 位四舍五入；申购费率以百分比 2 位表示（小数形式即 4 位）。

    两段都要验：受理回执（份额类）与确认后查询（全量，含折算 + 逐批明细）。
    """
    accepted = _accept(client)
    # 受理回执：份额 2 位。申请份额即使**不带小数**传入，也必须补齐（不能回显 "50000"）
    assert accepted["requested_qty"] == "50000.00"
    assert accepted["qty"] == "50000.00"

    _confirm_accepted(client, accepted)
    body = _query(client, accepted["convert_group_id"])

    def scale(key: str) -> int:
        return len(body[key].split(".")[1])

    # 金额 / 份额 → 2 位
    for key in ("out_amount", "redeem_fee", "convert_amount", "diff_fee", "in_amount",
                "in_qty", "requested_qty", "actual_qty"):
        assert scale(key) == 2, f"{key} 应为金额/份额口径（2 位），实际 {body[key]}"
    # 净值 / 费率 / 尾差 → 4 位
    for key in ("out_nav", "in_nav", "out_subscribe_fee_rate",
                "in_subscribe_fee_rate", "rounding_diff"):
        assert scale(key) == 4, f"{key} 应为净值/费率口径（4 位），实际 {body[key]}"

    for leg in body["lot_breakdown"]:
        assert len(leg["qty"].split(".")[1]) == 2, f"逐批份额 2 位，实际 {leg['qty']}"
        assert len(leg["fee_amount"].split(".")[1]) == 2, f"逐批费用 2 位，实际 {leg['fee_amount']}"
        assert len(leg["nav"].split(".")[1]) == 4, f"逐批净值 4 位，实际 {leg['nav']}"
        assert len(leg["fee_rate"].split(".")[1]) == 4, f"逐批费率 4 位，实际 {leg['fee_rate']}"


# ── 6. 跨主体拒绝（验收 14）─────────────────────────────────────────
def test_convert_cross_entity_rejected(client, conv_env):
    """转出华夏/TA-CN-001 → 转入易方/TA-CN-002 → 400 `CROSS_ENTITY_NOT_SUPPORTED`。"""
    r = client.post("/api/simulate/trade", json=_body(to=PROD_CROSS), headers=DEMO)
    assert r.status_code == 400, "跨主体在**受理段**即拒（校验前置于落单）"
    body = r.json()
    assert body["error_code"] == "CROSS_ENTITY_NOT_SUPPORTED"
    assert body["trace_id"]
    # 校验失败不留任何痕迹（4xx 不落库）
    core = conv_env["core"]
    assert _n(core, "SELECT COUNT(*) FROM core_trade WHERE customer_id = :c", c=CUSTOMER) == 0
    assert _n(
        core, "SELECT COUNT(*) FROM core_convert_request WHERE customer_id = :c", c=CUSTOMER
    ) == 0


# ── 7. 未知类型仍 400（R3 的集成侧对应）─────────────────────────────
def test_unknown_trade_type_returns_400_over_http(client):
    """走 HTTP 的未知类型仍是 400（`TradeRequest` 分支校验**放行**未知类型给网关兜底拒绝）。"""
    r = client.post(
        "/api/simulate/trade",
        json={"customer_id": CUSTOMER, "trade_type": "purchase", "amount": "1000"},
        headers=DEMO,
    )
    assert r.status_code == 400
    assert r.json()["error_code"] == "BAD_REQUEST"


# ── 8. T 日撤单（PRD §5.5 / R-9）· 占用自然释放、确认段天然跳过 ─────
def test_cancel_before_cutoff_releases_and_skips_confirm(client, conv_env):
    """受理日截点前撤单 → `cancelled`；确认批处理捞不到（终态天然幂等）。

    撤单走的闸门是**交易 owner**（本人 / `risk_demo`）—— 与查询的 **scope 闸门**
    是两道独立闸门（红线 7），故此处用 `DEMO` 头。
    """
    core = conv_env["core"]
    accepted = _accept(client)
    gid = accepted["convert_group_id"]

    r = client.post(f"/api/simulate/trade/convert/{gid}/cancel", headers=DEMO)
    assert r.status_code == 200, r.text
    body = r.json()
    assert body["status"] == "cancelled"
    assert body["released_qty"] == "50000.00"

    # 受理单状态已迁；批次/流水分毫未动（占用由状态推导，不写份额 · R-3）
    assert _one(
        core, "SELECT status FROM core_convert_request WHERE convert_group_id = :g", g=gid
    )["status"] == "cancelled"
    assert _n(core, "SELECT COUNT(*) FROM core_trade WHERE customer_id = :c", c=CUSTOMER) == 0

    # 撤单后再触发确认：捞单 SQL 只取 accepted/nav_pending → **一张也捞不到**
    batch = _confirm(client, accepted["confirm_date"] or accepted["accept_date"])
    assert batch["scanned"] == 0, f"终态单不应被确认批处理捞到：{batch}"
    assert batch["confirmed"] == 0

    # 重复撤单 → 409（仅 accepted 可撤，状态闸门）
    again = client.post(f"/api/simulate/trade/convert/{gid}/cancel", headers=DEMO)
    assert again.status_code == 409
    assert again.json()["error_code"] == "CANCEL_NOT_ALLOWED"


# ══ 以下为 T-9 的 **API 契约 + 鉴权** 用例（DoD 第 2/3/6 条）══════════
# ── 9. Q2：确认前查询不返回任何折算金额 ─────────────────────────────
def test_query_before_confirm_hides_settled_fields(client):
    """Q2 / PRD §5.3.2：**未确认**的单只回受理字段，绝不回折算金额。

    与用例 1 的分工：用例 1 验「受理**回执**」不含折算字段，本条验「查询**接口**」
    在未确认时同样不含 —— 两条路径是不同代码（`accept_convert` vs `_query_response`），
    漏一处就是「受理不折算、查询却先算给你看」的口径泄漏。
    """
    accepted = _accept(client)
    body = _query(client, accepted["convert_group_id"])

    assert body["status"] == "accepted"
    assert body["confirmed"] is False
    for key in NOT_IN_ACCEPT:
        assert key not in body, f"未确认的查询不得返回折算字段 {key}（Q2）"
    # 受理字段齐备（客户端据此展示「已受理 / 预计确认日 / 可撤单截止」）
    assert body["requested_qty"] == "50000.00"
    assert body["actual_qty"] is None  # 确认段才写（部分成交时 < 申请量）
    assert body["accept_date"] == str(accepted["accept_date"])
    assert body["confirm_date"] and body["available_date"]
    assert body["cancel_deadline"]
    assert body["forced_full_transfer"] is False


# ── 10. 两类 202 分辨（PRD §5.3.1 / §8.3）───────────────────────────
def test_two_kinds_of_202_are_distinguishable(client, monkeypatch):
    """`status='accepted'`（受理成功）与 `status='processing'`（并发执行中）都回 202，
    客户端**只能靠 `status` 分辨** —— 两者语义完全不同（前者已落受理单，后者未必）。

    `processing` 是 v1.0 实时链路的遗留形态（T-9 改分派后本链路不再产生该状态），
    保留兼容分支故需一条用例守住它 —— 否则将来有人「顺手清理」时无人发现。
    """
    _accept(client)  # 主形态：202 + `status='accepted'`（`_accept` 内已断言）

    from app.api import simulate as simulate_api

    monkeypatch.setattr(
        simulate_api,
        "submit_trade",
        lambda *a, **kw: {
            "blocked": False,
            "status": "processing",
            "convert_group_id": "CNV-TEST-PROC",
            "client_request_id": None,
        },
    )
    r = client.post("/api/simulate/trade", json=_body(), headers=DEMO)
    assert r.status_code == 202, r.text
    assert r.json()["status"] == "processing"


# ── 11. 撤单的两道闸门：窗口外 409 / 确认后 409（R-9）────────────────
def test_cancel_after_deadline_and_after_confirm_are_409(client, conv_env):
    """R-9 的两道判据：**状态闸门**（仅 accepted）与**时间闸门**（受理日截点前）。

    两道判据在真库上等价（`nav_pending` 只在 T+1 出现，彼时截点已过），故实现按
    更严的一条 —— 但**两条各自的拒绝分支**都要有用例，否则「窗口外」这条会无人守护。
    """
    core = conv_env["core"]
    accepted = _accept(client)
    gid = accepted["convert_group_id"]

    # ① 时间闸门：把撤单截止时刻改到过去（受理日 15:00 截点已过）
    with core.begin() as conn:
        conn.execute(
            text(
                "UPDATE core_convert_request SET cancel_before = :past"
                " WHERE convert_group_id = :g"
            ),
            {"past": datetime.now() - timedelta(hours=1), "g": gid},
        )
    r = client.post(_CANCEL.format(gid=gid), headers=DEMO)
    assert r.status_code == 409, r.text
    assert r.json()["error_code"] == "CANCEL_NOT_ALLOWED"

    # ② 状态闸门：恢复窗口后走确认，已 confirmed 的单再撤 → 409（终态不可逆）
    with core.begin() as conn:
        conn.execute(
            text(
                "UPDATE core_convert_request SET cancel_before = :future"
                " WHERE convert_group_id = :g"
            ),
            {"future": datetime.now() + timedelta(hours=5), "g": gid},
        )
    _confirm_accepted(client, accepted)
    r2 = client.post(_CANCEL.format(gid=gid), headers=DEMO)
    assert r2.status_code == 409, r2.text
    assert r2.json()["error_code"] == "CANCEL_NOT_ALLOWED"


# ── 12. 撤单 / 查询的鉴权：两道闸门**不可合并**（红线 7）─────────────
def test_convert_two_auth_gates_are_separate(client):
    """撤单 = **交易 owner** 闸门（本人 / `risk_demo`）；查询 = **查询 scope** 闸门
    （本人 / 已分配顾问 / `risk_officer`）。代理人可查**不可撤** —— 这是两闸门分离的
    核心判据，合并即意味着「顾问能替客户动钱」。
    """
    accepted = _accept(client)
    gid = accepted["convert_group_id"]

    # ① 非本人客户：两闸门都拒（403）
    assert client.post(_CANCEL.format(gid=gid), headers=STRANGER).status_code == 403
    assert client.get(_QUERY.format(gid=gid), headers=STRANGER).status_code == 403

    # ② 已分配顾问：**可查**（查询 scope）但**不可撤**（交易 owner 闸门）
    assert client.get(_QUERY.format(gid=gid), headers=ADVISOR).status_code == 200
    r_adv = client.post(_CANCEL.format(gid=gid), headers=ADVISOR)
    assert r_adv.status_code == 403, r_adv.text
    assert r_adv.json()["error_code"] == "AUTH_403_ROLE"

    # ③ `risk_demo` 有交易权限但**不在查询 scope**（它是演示角色，非真实归属主体）
    assert client.get(_QUERY.format(gid=gid), headers=DEMO).status_code == 403
    assert client.get(_QUERY.format(gid=gid), headers=OPS).status_code == 200

    # ④ 本人两闸门皆通；前序 403 未改动任何状态，此刻仍可撤
    assert client.get(_QUERY.format(gid=gid), headers=SELF).status_code == 200
    r_self = client.post(_CANCEL.format(gid=gid), headers=SELF)
    assert r_self.status_code == 200, r_self.text
    assert r_self.json()["status"] == "cancelled"