group_xinghuo_jinrong/tests/test_convert_accept.py

"""T-6 `accept_convert` 受理事务单测（T+1 受理/确认分离模型 · D21）。

**为什么单独一个文件**：受理段（T-6）与确认段（T-7）是两条独立链路 ——
受理只「校验 + 落单」，绝不扣份额/折算/写流水；v1.0 的 `test_convert_service.py`
测的是旧三阶段流（T-7 会整体重写），两者混在一个文件会让「受理无副作用」
这条最关键的断言淹没在旧流程用例里。

sqlite 内存库（`conftest.sqlite_engine`，T-1 已建 3 张新表 + 适当性矩阵种子）。
**真库侧**（InnoDB 唯一约束 / ENUM 严校验 / DATETIME(3)）由
`scripts/dev/verify_convert_accept.py` 承担 —— sqlite 永远看不见这些。

覆盖（开发计划 v2.0 T-6 DoD 的 sqlite 侧）：
1. 受理成功：status=accepted、**份额不变、无流水**、镜像 pending、审计 accepted
2. 同 client_request_id 重复提交 → 同一 group_id、仍 1 行（验收 12）
3. 命中终态 → 原样透出、**不重开**（再次提交不产生新行）
4. 无幂等键 → 每次都是独立转换（可重复受理）
5. 在途占用（R-3）：受理后再超量申请 → InsufficientShares（验收 22）
6. 非交易日 / 过 15:00 截点 → 受理日顺延下一交易日（验收 28）
7. 适当性不匹配 → blocked 且受理单 0 行（验收 4）
8. 强制全转 → 受理 qty 记实际全转量（验收 13 前半）
9. 低于最低转出份额 → BelowMinQty
10. uk_idem 竞态兜底：查时无、插时有 → 转幂等返回（R-2 守卫）
11. 产品校验 4xx（同产品 / 不可赎回 / 跨管理人）
"""

from __future__ import annotations

from datetime import date, datetime, timedelta
from decimal import Decimal

import pytest
from sqlalchemy import text

from app.repository.convert_repository import ConvertRepository
from app.repository.convert_request_repository import ConvertRequestRepository
from app.repository.core_ro import CoreReadOnlyRepository
from app.repository.risk_repository import RiskRepository
from app.service.convert.convert_service import accept_convert
from app.service.convert.errors import (
    BelowMinQty,
    CrossEntityNotSupported,
    InsufficientShares,
    ProductNotRedeemable,
    SameProduct,
)

CUST = "CUST-AC"  # C3 客户 → 转入 R4 = allowed_with_disclosure（放行）
CUST_LOW = "CUST-ACL"  # C1 客户 → 转入 R4 = forbidden（唯一阻断点）
PROD_OUT = "PROD-AC-OUT"
PROD_IN = "PROD-AC-IN"
COMPANY = "华夏模拟基金"
TA = "TA-CN-001"

TODAY = date(2026, 9, 4)  # 周五
NEXT_BIZ = date(2026, 9, 7)  # 下周一
NOW = datetime(2026, 9, 4, 10, 0, 0)


# ── 种子 ────────────────────────────────────────────────────────────
def _exec(engine, sql: str, **params) -> None:
    with engine.begin() as conn:
        conn.execute(text(sql), params)


def _rows(engine, sql: str, **params) -> list[dict]:
    with engine.connect() as conn:
        return [dict(r) for r in conn.execute(text(sql), params).mappings()]


def _seed_calendar(engine, start: date = date(2026, 9, 1), days: int = 60) -> None:
    """按真实交易日历灌种子：周一~周五开市，周末**不插行**（`is_open` 无行即休市）。"""
    cursor = start
    for _ in range(days):
        if cursor.weekday() < 5:
            _exec(
                engine,
                "INSERT INTO core_trade_calendar (cal_date, is_open, remark) "
                "VALUES (:d, 1, 'test')",
                d=cursor,
            )
        cursor += timedelta(days=1)


def _seed(engine, *, min_redeem: str = "0", min_hold: str = "0") -> None:
    _exec(
        engine,
        "INSERT INTO core_customer (customer_id, display_name, age, is_active) "
        "VALUES (:c, 'AC客户', 40, 1)",
        c=CUST,
    )
    _exec(
        engine,
        "INSERT INTO core_customer (customer_id, display_name, age, is_active) "
        "VALUES (:c, 'AC低风险客户', 40, 1)",
        c=CUST_LOW,
    )
    for cid, code in ((CUST, "C3"), (CUST_LOW, "C1")):
        _exec(
            engine,
            "INSERT INTO core_customer_risk "
            "(customer_id, risk_code, evaluated_at, expires_at) "
            "VALUES (:c, :r, :t, :exp)",
            c=cid, r=code,
            t=NOW - timedelta(days=30), exp=NOW + timedelta(days=300),
        )
    _exec(
        engine,
        "INSERT INTO core_product (product_id, product_name, min_risk_code, product_type, "
        "can_subscribe, can_redeem, subscribe_fee_rate, min_redeem_qty, min_hold_qty, "
        "fund_company, ta_code) "
        "VALUES (:p, '转出债基', 'R2', 'bond', 1, 1, 0.0030, :mr, :mh, :co, :ta)",
        p=PROD_OUT, mr=float(min_redeem), mh=float(min_hold), co=COMPANY, ta=TA,
    )
    _exec(
        engine,
        "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, '转入股基', 'R4', 'stock', 1, 1, 0.0080, :co, :ta)",
        p=PROD_IN, co=COMPANY, ta=TA,
    )
    # 最近净值（仅用于适当性阻断预警的**金额估算**，受理不折算）
    _exec(
        engine,
        "INSERT INTO core_product_nav (product_id, nav, daily_chg_pct, nav_date) "
        "VALUES (:p, 1.0300, 0, :d)",
        p=PROD_OUT, d=TODAY,
    )
    for cid, tag in ((CUST, "AC"), (CUST_LOW, "ACL")):
        _exec(
            engine,
            "INSERT INTO core_share_lot (lot_id, customer_id, product_id, qty, remain_qty, "
            "nav, confirmed_at) VALUES (:l, :c, :p, 100, 100, 1.0300, :cat)",
            l=f"LOT-{tag}-1", c=cid, p=PROD_OUT, cat=datetime(2026, 8, 1, 10, 0, 0),
        )
        _exec(
            engine,
            "INSERT INTO core_share_lot (lot_id, customer_id, product_id, qty, remain_qty, "
            "nav, confirmed_at) VALUES (:l, :c, :p, 50, 50, 1.0000, :cat)",
            l=f"LOT-{tag}-2", c=cid, p=PROD_OUT, cat=datetime(2026, 9, 1, 10, 0, 0),
        )
        _exec(
            engine,
            "INSERT INTO core_holding (customer_id, product_id, qty, cost_amount, "
            "market_value, pnl_pct, as_of) VALUES (:c, :p, 150, 150, 154.5, 0, :d)",
            c=cid, p=PROD_OUT, d=TODAY,
        )
    _seed_calendar(engine)


def _services(engine) -> dict:
    return dict(
        core_ro=CoreReadOnlyRepository(engine=engine),
        risk_repo=RiskRepository(engine=engine),
        convert_repo=ConvertRepository(engine=engine),
        request_repo=ConvertRequestRepository(engine=engine),
    )


def _reqs(engine, **where) -> list[dict]:
    sql = "SELECT * FROM core_convert_request"
    if where:
        sql += " WHERE " + " AND ".join(f"{k} = :{k}" for k in where)
    return _rows(engine, sql, **where)


def _physical_qty(engine, cid: str = CUST) -> Decimal:
    rows = _rows(
        engine,
        "SELECT COALESCE(SUM(remain_qty), 0) AS s FROM core_share_lot "
        "WHERE customer_id = :c AND product_id = :p",
        c=cid, p=PROD_OUT,
    )
    return Decimal(str(rows[0]["s"]))


def _call(engine, **over) -> dict:
    req = {
        "customer_id": CUST,
        "from_product_id": PROD_OUT,
        "to_product_id": PROD_IN,
        "qty": Decimal("100"),
    }
    req.update(over)
    return accept_convert(req, now=NOW, **_services(engine))


# ── 1. 受理成功：无任何副作用 ────────────────────────────────────────
def test_accept_success_has_no_side_effects(sqlite_engine):
    """受理成功 = 只落受理单；**份额不变、无流水**、镜像 pending、审计 accepted。"""
    _seed(sqlite_engine)
    before = _physical_qty(sqlite_engine)

    result = _call(sqlite_engine, client_request_id="REQ-AC-1")

    assert result["blocked"] is False
    assert result["accepted"] is True
    assert result["status"] == "accepted"
    assert result["idempotent"] is False
    assert result["convert_group_id"].startswith("CNV-")
    assert result["qty"] == "100.00"
    assert result["forced_full_transfer"] is False
    # 受理回执字段（PRD §5.3.1）：T / T+1 / T+2 按交易日历推算
    assert result["accept_date"] == str(TODAY)
    assert result["confirm_date"] == str(NEXT_BIZ)  # T+1 = 下一交易日
    assert result["available_date"] == str(date(2026, 9, 8))  # T+2
    assert result["cancel_deadline"] == f"{TODAY} 15:00:00"
    assert result["accepted_at"].startswith(str(NOW))

    row = _reqs(sqlite_engine, convert_group_id=result["convert_group_id"])[0]
    assert row["status"] == "accepted"
    assert row["actual_qty"] is None  # 确认后才有
    assert Decimal(str(row["qty"])) == Decimal("100")
    assert str(row["requested_at"])[:10] == str(TODAY)  # 受理日
    assert str(row["cancel_before"])[:10] == str(TODAY)  # 撤单窗口落在受理日
    assert str(row["cancel_before"])[11:16] == "15:00"

    # 副作用断言（受理段最核心的三条）
    assert _physical_qty(sqlite_engine) == before  # 份额一分未动
    assert _rows(sqlite_engine, "SELECT 1 FROM core_trade LIMIT 1") == []  # 无流水

    mirror = _rows(
        sqlite_engine,
        "SELECT status FROM risk_convert_detail WHERE convert_group_id = :g",
        g=result["convert_group_id"],
    )
    assert len(mirror) == 1 and mirror[0]["status"] == "pending"

    audits = _rows(
        sqlite_engine,
        "SELECT decision FROM audit_log WHERE event_type = 'convert_request' "
        "AND input_summary LIKE :pat",
        pat=f"%{result['convert_group_id']}%",
    )
    assert [a["decision"] for a in audits] == ["accepted"]

    # 引擎时机（T-8 / FR-C28 · 验收 29 前半）：受理段**不得**触规则引擎 ——
    # 引擎只在 T+1 确认事务提交后跑一次（`confirm_service` → `engine_call`）。
    assert _rows(sqlite_engine, "SELECT 1 FROM risk_alert LIMIT 1") == []
    assert _rows(
        sqlite_engine,
        "SELECT 1 FROM audit_log WHERE decision IN ('engine_error', 'pass')",
    ) == []


# ── 2/3/4. 幂等语义 ─────────────────────────────────────────────────
def test_accept_same_client_request_id_returns_same_group(sqlite_engine):
    """同 `client_request_id` 重复提交 → 同一 group_id，受理单仍 1 行（验收 12）。"""
    _seed(sqlite_engine)
    first = _call(sqlite_engine, client_request_id="REQ-AC-DUP")
    second = _call(sqlite_engine, client_request_id="REQ-AC-DUP")

    assert second["idempotent"] is True
    assert second["convert_group_id"] == first["convert_group_id"]
    assert len(_reqs(sqlite_engine)) == 1
    # 镜像不得因重复提交多出一行
    assert len(_rows(sqlite_engine, "SELECT 1 FROM risk_convert_detail")) == 1


def test_accept_final_state_is_not_reopened(sqlite_engine):
    """命中终态 → 原样透出且**不重开**（T+1 模型下终态不可逆）。"""
    _seed(sqlite_engine)
    first = _call(sqlite_engine, client_request_id="REQ-AC-FIN")
    _exec(
        sqlite_engine,
        "UPDATE core_convert_request SET status = 'confirmed' "
        "WHERE convert_group_id = :g",
        g=first["convert_group_id"],
    )

    again = _call(sqlite_engine, client_request_id="REQ-AC-FIN")

    assert again["idempotent"] is True
    assert again["status"] == "confirmed"
    assert again["convert_group_id"] == first["convert_group_id"]
    assert len(_reqs(sqlite_engine)) == 1  # 没有新行


def test_accept_without_idempotency_key_allows_repeat(sqlite_engine):
    """无幂等键 → 每次都是独立转换（PRD §7.3：无幂等语义）。

    注意两次申请量之和须不超可用余额 —— 在途占用对**无键请求同样生效**
    （占用是数据行层面的事实，与是否带幂等键无关）。
    """
    _seed(sqlite_engine)
    a = _call(sqlite_engine, qty=Decimal("60"))
    b = _call(sqlite_engine, qty=Decimal("60"))

    assert a["convert_group_id"] != b["convert_group_id"]
    assert len(_reqs(sqlite_engine)) == 2


# ── 5. 在途占用（R-3）───────────────────────────────────────────────
def test_accept_inflight_occupancy_blocks_over_request(sqlite_engine):
    """受理占 100 → 可用 = 150 − 100 = 50；再申请 60 → InsufficientShares（验收 22）。"""
    _seed(sqlite_engine)
    _call(sqlite_engine, qty=Decimal("100"), client_request_id="REQ-AC-OCC")

    lot_repo = CoreReadOnlyRepository(engine=sqlite_engine)
    assert lot_repo.sum_inflight_qty(CUST, PROD_OUT) == Decimal("100")

    with pytest.raises(InsufficientShares):
        _call(sqlite_engine, qty=Decimal("60"))
    assert len(_reqs(sqlite_engine)) == 1  # 超量那笔未落单

    # 恰好等于可用余额 → 放行（边界：`>` 比较，等于不拒）
    ok = _call(sqlite_engine, qty=Decimal("50"))
    assert ok["accepted"] is True
    assert len(_reqs(sqlite_engine)) == 2


def test_accept_cancelled_releases_occupancy(sqlite_engine):
    """终态释放占用（R-3）：单子被撤 → 在途归零，同量可再受理。"""
    _seed(sqlite_engine)
    first = _call(sqlite_engine, qty=Decimal("150"))
    with pytest.raises(InsufficientShares):
        _call(sqlite_engine, qty=Decimal("10"))

    _exec(
        sqlite_engine,
        "UPDATE core_convert_request SET status = 'cancelled' WHERE convert_group_id = :g",
        g=first["convert_group_id"],
    )
    again = _call(sqlite_engine, qty=Decimal("150"))
    assert again["accepted"] is True


# ── 6. 受理日判定（R-5）─────────────────────────────────────────────
def test_accept_on_non_trading_day_rolls_to_next(sqlite_engine):
    """非交易日提交 → 受理日顺延下一交易日（验收 28）。"""
    _seed(sqlite_engine)
    _exec(
        sqlite_engine,
        "DELETE FROM core_trade_calendar WHERE cal_date = :d",
        d=TODAY,
    )

    result = _call(sqlite_engine)

    row = _reqs(sqlite_engine, convert_group_id=result["convert_group_id"])[0]
    assert str(row["requested_at"])[:10] == str(NEXT_BIZ)
    assert str(row["cancel_before"])[:10] == str(NEXT_BIZ)
    # 顺延后的受理日 → T+1/T+2 也随之顺延（PRD §5.3.1「全部按交易日历推算」）
    assert result["accept_date"] == str(NEXT_BIZ)
    assert result["confirm_date"] == str(date(2026, 9, 8))
    assert result["available_date"] == str(date(2026, 9, 9))


def test_accept_after_cutoff_rolls_to_next(sqlite_engine):
    """交易日但过 15:00 截点 → 顺延（15:00:00 整即算超时，PRD §2.6.1）。"""
    _seed(sqlite_engine)
    result = accept_convert(
        {
            "customer_id": CUST,
            "from_product_id": PROD_OUT,
            "to_product_id": PROD_IN,
            "qty": Decimal("100"),
        },
        now=datetime(2026, 9, 4, 16, 0, 0),
        **_services(sqlite_engine),
    )
    row = _reqs(sqlite_engine, convert_group_id=result["convert_group_id"])[0]
    assert str(row["requested_at"])[:10] == str(NEXT_BIZ)


def test_accept_before_cutoff_keeps_same_day(sqlite_engine):
    """交易日且截点前 → 受理日即当日（14:59 提交）。"""
    _seed(sqlite_engine)
    result = accept_convert(
        {
            "customer_id": CUST,
            "from_product_id": PROD_OUT,
            "to_product_id": PROD_IN,
            "qty": Decimal("100"),
        },
        now=datetime(2026, 9, 4, 14, 59, 0),
        **_services(sqlite_engine),
    )
    row = _reqs(sqlite_engine, convert_group_id=result["convert_group_id"])[0]
    assert str(row["requested_at"])[:10] == str(TODAY)


def test_accept_dates_none_when_calendar_exhausted(sqlite_engine):
    """日历数据边界：受理日之后再无交易日 → T+1/T+2 日期置空，但**受理必须成功**。

    真库 verify 首轮实跑暴露的真实缺陷：T+1 日期原本写在返回语句里直接调
    `next_biz_day`，日历用尽时抛 ValueError，**已经落库的受理单在调用方眼里变成 500**
    —— 客户端会重试，无幂等键的请求还会重复受理。展示性字段不得影响受理结果。
    """
    _seed(sqlite_engine)
    _exec(
        sqlite_engine,
        "DELETE FROM core_trade_calendar WHERE cal_date > :d",
        d=TODAY,
    )

    result = _call(sqlite_engine)

    assert result["accepted"] is True
    assert result["status"] == "accepted"
    assert result["confirm_date"] is None  # 推算不出 → 置空，不抛
    assert result["available_date"] is None
    assert result["accept_date"] == str(TODAY)  # 受理日本身仍有值（当日是交易日）
    assert len(_reqs(sqlite_engine)) == 1  # 受理单照常落地


# ── 7. 适当性阻断（唯一阻断点）──────────────────────────────────────
def test_accept_suitability_blocked_leaves_no_request(sqlite_engine):
    """适当性不匹配 → blocked，**受理单 0 行、镜像 0 行**（验收 4）。"""
    _seed(sqlite_engine)
    result = _call(sqlite_engine, customer_id=CUST_LOW)

    assert result["blocked"] is True
    assert result["accepted"] is False
    assert result["block_reason"]
    assert _reqs(sqlite_engine) == []
    assert _rows(sqlite_engine, "SELECT 1 FROM risk_convert_detail") == []

    audits = _rows(
        sqlite_engine,
        "SELECT decision FROM audit_log WHERE event_type = 'convert_request' "
        "AND decision = 'suitability_blocked'",
    )
    assert len(audits) >= 1


# ── 8/9. 强制全转与最低转出份额 ─────────────────────────────────────
def test_accept_forced_full_transfer_records_actual_qty(sqlite_engine):
    """强制全转：申请 148 留 2 不足 min_hold=100 → 实际全转 150（验收 13 前半）。"""
    _seed(sqlite_engine, min_hold="100")
    result = _call(sqlite_engine, qty=Decimal("148"))

    assert result["forced_full_transfer"] is True
    assert result["requested_qty"] == "148.00"
    assert result["qty"] == "150.00"
    row = _reqs(sqlite_engine, convert_group_id=result["convert_group_id"])[0]
    assert Decimal(str(row["qty"])) == Decimal("150")


def test_accept_below_min_redeem_raises(sqlite_engine):
    """低于最低转出份额且非全额 → BelowMinQty；不落单。"""
    _seed(sqlite_engine, min_redeem="200")
    with pytest.raises(BelowMinQty):
        _call(sqlite_engine, qty=Decimal("100"))
    assert _reqs(sqlite_engine) == []

    # 全额转出豁免（PRD §12 I-3）：申请 150 = 全部持有 → 放行
    ok = _call(sqlite_engine, qty=Decimal("150"))
    assert ok["accepted"] is True


# ── 10. uk_idem 竞态兜底（R-2 守卫）────────────────────────────────
def test_accept_uk_idem_race_falls_back_to_idempotent(sqlite_engine, monkeypatch):
    """查时无、插时有（并发窗口）→ 撞 `uk_idem` 转幂等返回，不抛 IntegrityError。

    模拟方式：第一次 `get_by_client_request_id` 返回 None 的同时，让「抢先者」把行插进去
    —— 这正是 T-13 真库实测过的交错窗口。若无兜底，未映射的 IntegrityError 会直穿 503。

    本笔申请量取 50：抢占者已占 100，可用余额恰为 150 − 100 = 50，
    这样既穿过了 ⑤ 占用校验，又必然在 ⑥ 撞上 `uk_idem`（否则测不到兜底分支）。
    """
    _seed(sqlite_engine)
    winner = {}

    def _race(engine):
        winner["gid"] = "CNV-WINNER"
        ConvertRequestRepository(engine=engine).insert(
            "CNV-WINNER", CUST, PROD_OUT, PROD_IN, Decimal("100"),
            client_request_id="REQ-AC-RACE",
            requested_at=datetime.combine(TODAY, datetime.min.time()),
        )

    real = ConvertRequestRepository.get_by_client_request_id
    calls = {"n": 0}

    def flaky(self, cid):
        calls["n"] += 1
        if calls["n"] == 1:
            _race(sqlite_engine)  # 抢占者在「查」与「插」之间落地
            return None
        return real(self, cid)

    monkeypatch.setattr(ConvertRequestRepository, "get_by_client_request_id", flaky)

    result = _call(sqlite_engine, qty=Decimal("50"), client_request_id="REQ-AC-RACE")

    assert result["idempotent"] is True
    assert result["convert_group_id"] == "CNV-WINNER"
    assert len(_reqs(sqlite_engine)) == 1  # 本笔未新增行


# ── 11. 产品校验 4xx ────────────────────────────────────────────────
def test_accept_product_validation_errors(sqlite_engine):
    """同产品 / 不可赎回 / 跨管理人 → 各自 4xx，且均不落单。"""
    _seed(sqlite_engine)

    with pytest.raises(SameProduct):
        _call(sqlite_engine, to_product_id=PROD_OUT)
    assert _reqs(sqlite_engine) == []

    _exec(
        sqlite_engine,
        "UPDATE core_product SET can_redeem = 0 WHERE product_id = :p",
        p=PROD_OUT,
    )
    with pytest.raises(ProductNotRedeemable):
        _call(sqlite_engine)
    assert _reqs(sqlite_engine) == []

    _exec(
        sqlite_engine,
        "UPDATE core_product SET can_redeem = 1 WHERE product_id = :p",
        p=PROD_OUT,
    )
    _exec(
        sqlite_engine,
        "UPDATE core_product SET ta_code = 'TA-OTHER' WHERE product_id = :p",
        p=PROD_IN,
    )
    with pytest.raises(CrossEntityNotSupported):
        _call(sqlite_engine)
    assert _reqs(sqlite_engine) == []


def test_accept_zero_or_negative_qty_raises(sqlite_engine):
    """申请份额必须为正（受理不落单）。"""
    _seed(sqlite_engine)
    with pytest.raises(BelowMinQty):
        _call(sqlite_engine, qty=Decimal("0"))
    assert _reqs(sqlite_engine) == []