zkxw/backend/tests/test_market_presign_api.py

import pytest
from fastapi import HTTPException
from sqlalchemy import create_engine, event
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import sessionmaker

from app.models.base import Base
from app.models.models import (
    BizAttachment,
    BizCellSampleType,
    BizCollectionHospital,
    BizCollectionProtocol,
    BizCollectionRegion,
    BizCollectionRegionStaff,
    BizBizNoSequence,
    BizPresignOrder,
    BizPresignOrderSampleType,
    BizPresignTransfer,
    BizProtocolCollectionInfo,
    BizProtocolPartyCustomer,
    BizProtocolSample,
    SysPost,
    SysPostRole,
    SysRole,
    SysRoleDataScopeDetail,
    SysUser,
    SysUserPost,
    SysUserRole,
)
from app.models.research_donation import (
    BizResearchDonation,
    BizResearchDonationSample,
)


_TEST_TABLES = [
    BizCollectionRegionStaff.__table__,
    BizCollectionRegion.__table__,
    BizCollectionHospital.__table__,
    BizBizNoSequence.__table__,
    BizPresignOrder.__table__,
    BizPresignOrderSampleType.__table__,
    BizPresignTransfer.__table__,
    BizCellSampleType.__table__,
    BizCollectionProtocol.__table__,
    BizProtocolPartyCustomer.__table__,
    BizProtocolCollectionInfo.__table__,
    BizProtocolSample.__table__,
    BizResearchDonation.__table__,
    BizResearchDonationSample.__table__,
    BizAttachment.__table__,
    SysUser.__table__,
    SysRole.__table__,
    SysRoleDataScopeDetail.__table__,
    SysUserRole.__table__,
    SysPost.__table__,
    SysPostRole.__table__,
    SysUserPost.__table__,
]


def _session_factory(database_url="sqlite:///:memory:"):
    engine = create_engine(database_url)
    Base.metadata.create_all(
        engine,
        tables=_TEST_TABLES,
    )
    return sessionmaker(bind=engine)


def _session():
    return _session_factory()()


def _order(
    order_id,
    *,
    tenant_no="tenant-a",
    region_id="region-a",
    sales_user_id="sales-a",
    manager_user_id="manager-a",
):
    return BizPresignOrder(
        id=order_id,
        tenant_no=tenant_no,
        region_id=region_id,
        sales_user_id=sales_user_id,
        region_manager_user_id=manager_user_id,
        order_status="active",
        create_by="creator-a",
        create_dept_id="dept-a",
        del_flag=0,
    )


def _transfer(
    transfer_id,
    *,
    order_id="own",
    tenant_no="tenant-a",
    pay_status="dfj",
):
    return BizPresignTransfer(
        id=transfer_id,
        presign_order_id=order_id,
        transfer_amount=100,
        transfer_type="wechat",
        pay_status=pay_status,
        tenant_no=tenant_no,
        create_by="creator-a",
        create_dept_id="dept-a",
        del_flag=0,
    )


def _ctx(scope_kind, user_id="u1"):
    from app.services.market_field_scope import MarketFieldContext

    return MarketFieldContext("tenant-a", user_id, scope_kind)


def _user(user_id="u1"):
    return {
        "user_id": user_id,
        "tenant_no": "tenant-a",
        "role": "super_admin",
        "dept_id": "dept-a",
    }


def _real_auth_user(user_id="u1"):
    return {
        "user_id": user_id,
        "tenant_no": "tenant-a",
        "role": "super_admin",
        "region_id": "dept-from-auth",
    }


def test_converted_presign_rejects_update_and_convert_actions():
    from app.schemas.market_mobile import MarketPresignOrderBody
    from app.services.market_presign import convert_presign_order, update_presign_order

    db = _session()
    _add_master_data(db)
    db.add(_order('terminal', sales_user_id='u1'))
    db.commit()
    row = db.query(BizPresignOrder).filter_by(id='terminal').one()
    row.order_status = 'converted'
    db.commit()

    with pytest.raises(ValueError, match='不可|已转'):
        update_presign_order(
            db,
            order_id='terminal',
            body=MarketPresignOrderBody(remark='不应保存'),
            ctx=_ctx('specialist'),
            user_info=_user(),
        )
    with pytest.raises(ValueError, match='不可|已转'):
        convert_presign_order(
            db,
            order_id='terminal',
            ctx=_ctx('specialist'),
            user_info=_user(),
        )


def test_void_presign_allows_update_but_rejects_convert():
    from app.schemas.market_mobile import MarketPresignOrderBody
    from app.services.market_presign import convert_presign_order, update_presign_order

    db = _session()
    _add_master_data(db)
    db.add(_order('void-order', sales_user_id='u1'))
    db.commit()
    row = db.query(BizPresignOrder).filter_by(id='void-order').one()
    row.order_status = 'void'
    db.commit()

    data = update_presign_order(
        db,
        order_id='void-order',
        body=MarketPresignOrderBody(remark='作废单仍可修订'),
        ctx=_ctx('specialist'),
        user_info=_user(),
    )

    assert data['remark'] == '作废单仍可修订'
    with pytest.raises(ValueError, match='不可|作废'):
        convert_presign_order(
            db,
            order_id='void-order',
            ctx=_ctx('specialist'),
            user_info=_user(),
        )


def test_presign_create_rejects_unregistered_transfer_attachment_url():
    from app.schemas.market_mobile import MarketPresignOrderBody
    from app.services.market_presign import create_presign_order

    db = _session()
    _add_master_data(db)
    body = MarketPresignOrderBody.model_validate(
        {
            'regionId': 'region-a',
            'hospitalId': 'hospital-a',
            'transfers': [
                {
                    'transferAmount': 100,
                    'transferAttachments': 'https://evil.test/proof.pdf',
                }
            ],
        }
    )

    with pytest.raises(ValueError, match='附件'):
        create_presign_order(
            db,
            body=body,
            ctx=_ctx('specialist'),
            user_info=_user(),
        )


@pytest.mark.parametrize('order_status', ['converted', 'void'])
def test_presign_create_rejects_embedded_transfer_for_non_writable_order(
    order_status,
):
    from app.schemas.market_mobile import MarketPresignOrderBody
    from app.services.market_presign import create_presign_order

    db = _session()
    _add_master_data(db)
    body = MarketPresignOrderBody.model_validate(
        {
            'regionId': 'region-a',
            'hospitalId': 'hospital-a',
            'orderStatus': order_status,
            'transfers': [
                {
                    'transferAmount': 100,
                    'transferType': 'bank',
                }
            ],
        }
    )

    with pytest.raises(ValueError, match='订单|状态|转账'):
        create_presign_order(
            db,
            body=body,
            ctx=_ctx('specialist'),
            user_info=_user(),
        )

    assert db.query(BizPresignOrder).count() == 0
    assert db.query(BizPresignTransfer).count() == 0


def _add_master_data(db):
    db.add_all(
        [
            BizCollectionRegion(
                id="region-a",
                region_name="A区",
                tenant_no="tenant-a",
                create_dept_id="",
                del_flag=0,
            ),
            BizCollectionRegion(
                id="region-b",
                region_name="B区",
                tenant_no="tenant-a",
                create_dept_id="",
                del_flag=0,
            ),
            BizCollectionRegion(
                id="region-cross",
                region_name="跨租户地区",
                tenant_no="tenant-b",
                create_dept_id="",
                del_flag=0,
            ),
            BizCollectionHospital(
                id="hospital-a",
                region_id="region-a",
                hospital_name="A医院",
                tenant_no="tenant-a",
                create_dept_id="",
                del_flag=0,
            ),
            BizCollectionHospital(
                id="hospital-b",
                region_id="region-b",
                hospital_name="B医院",
                tenant_no="tenant-a",
                create_dept_id="",
                del_flag=0,
            ),
            BizCellSampleType(
                id="sample-a",
                type_code="A",
                type_name="样本A",
                tenant_no="tenant-a",
                create_dept_id="",
                del_flag=0,
            ),
            SysUser(
                id="u1",
                user_no="u1",
                nick_name="当前用户",
                status="1",
                tenant_no="tenant-a",
                create_dept_id="dept-a",
                del_flag=0,
            ),
            SysUser(
                id="collector-a",
                user_no="collector-a",
                nick_name="有效采集员",
                status="1",
                tenant_no="tenant-a",
                create_dept_id="dept-a",
                del_flag=0,
            ),
            SysUser(
                id="manager-a",
                user_no="manager-a",
                nick_name="有效经理",
                status="1",
                tenant_no="tenant-a",
                create_dept_id="dept-a",
                del_flag=0,
            ),
            SysUser(
                id="cross-user",
                user_no="cross-secret",
                nick_name="跨租户秘密姓名",
                status="1",
                tenant_no="tenant-b",
                create_dept_id="dept-b",
                del_flag=0,
            ),
            SysUser(
                id="deleted-user",
                user_no="deleted-secret",
                nick_name="已删除秘密姓名",
                status="1",
                tenant_no="tenant-a",
                create_dept_id="dept-a",
                del_flag=1,
            ),
            SysUser(
                id="disabled-user",
                user_no="disabled-secret",
                nick_name="停用秘密姓名",
                status="0",
                tenant_no="tenant-a",
                create_dept_id="dept-a",
                del_flag=0,
            ),
            BizCellSampleType(
                id="sample-cross",
                type_code="X",
                type_name="跨租户样本",
                tenant_no="tenant-b",
                create_dept_id="",
                del_flag=0,
            ),
        ]
    )
    db.commit()


def test_specialist_presign_list_only_returns_own_sales_orders():
    from app.services.market_presign import list_presign_orders

    db = _session()
    db.add_all(
        [
            _order("own", sales_user_id="u1"),
            _order("other", sales_user_id="u2"),
            _order("cross-tenant", tenant_no="tenant-b", sales_user_id="u1"),
        ]
    )
    db.commit()

    items, total = list_presign_orders(
        db, ctx=_ctx("specialist"), user_info=_user()
    )

    assert total == 1
    assert [item["id"] for item in items] == ["own"]


def test_manager_presign_list_uses_region_or_manager_union():
    from app.services.market_presign import list_presign_orders

    db = _session()
    db.add(
        BizCollectionRegionStaff(
            id="leader",
            region_id="region-led",
            user_id="u1",
            is_leader=1,
            tenant_no="tenant-a",
            create_dept_id="dept-a",
            del_flag=0,
        )
    )
    db.add_all(
        [
            _order("by-region", region_id="region-led", manager_user_id="u2"),
            _order("by-manager", region_id="region-other", manager_user_id="u1"),
            _order("outside", region_id="region-other", manager_user_id="u2"),
        ]
    )
    db.commit()

    items, total = list_presign_orders(
        db, ctx=_ctx("manager"), user_info=_user()
    )

    assert total == 2
    assert {item["id"] for item in items} == {"by-region", "by-manager"}


def test_presign_list_batches_deposits_and_user_names_without_n_plus_one():
    from app.services.market_presign import list_presign_orders

    db = _session()
    db.add(
        SysUser(
            id="u1",
            user_no="u1",
            nick_name="当前用户",
            status="1",
            tenant_no="tenant-a",
            create_dept_id="dept-a",
            del_flag=0,
        )
    )
    db.add_all(
        [_order(f"order-{index}", sales_user_id="u1") for index in range(3)]
    )
    db.commit()
    statements = []

    def capture_statement(*args):
        statements.append(args[2])

    event.listen(db.get_bind(), "before_cursor_execute", capture_statement)
    try:
        items, total = list_presign_orders(
            db, ctx=_ctx("specialist"), user_info=_user(), page_size=10
        )
    finally:
        event.remove(db.get_bind(), "before_cursor_execute", capture_statement)

    assert total == 3
    assert len(items) == 3
    assert len(statements) <= 6


def test_presign_list_and_detail_resolve_referenced_master_names_in_batches():
    from app.services.market_presign import get_presign_order, list_presign_orders

    db = _session()
    _add_master_data(db)
    db.add_all(
        [
            BizCollectionRegion(id='region-deleted', region_name='历史地区', tenant_no='tenant-a', create_dept_id='', del_flag=1),
            BizCollectionHospital(id='hospital-deleted', region_id='region-deleted', hospital_name='历史医院', tenant_no='tenant-a', create_dept_id='', del_flag=1),
            BizCellSampleType(id='sample-deleted', type_code='D', type_name='历史样本', tenant_no='tenant-a', create_dept_id='', del_flag=1),
            BizCollectionHospital(id='hospital-cross', region_id='region-cross', hospital_name='跨租户医院', tenant_no='tenant-b', create_dept_id='', del_flag=1),
        ]
    )
    normal = _order('normal-master', region_id='region-a', sales_user_id='u1')
    normal.hospital_id = 'hospital-a'
    historical = _order(
        'historical-master', region_id='region-deleted', sales_user_id='u1'
    )
    historical.hospital_id = 'hospital-deleted'
    cross = _order('cross-master', region_id='region-cross', sales_user_id='u1')
    cross.hospital_id = 'hospital-cross'
    db.add_all(
        [
            normal,
            historical,
            cross,
            BizPresignOrderSampleType(id='normal-master-type', presign_order_id=normal.id, sample_type_id='sample-a', tenant_no='tenant-a', create_dept_id='', order_num=1, del_flag=0),
            BizPresignOrderSampleType(id='historical-master-type', presign_order_id=historical.id, sample_type_id='sample-deleted', tenant_no='tenant-a', create_dept_id='', order_num=1, del_flag=0),
            BizPresignOrderSampleType(id='cross-master-type', presign_order_id=cross.id, sample_type_id='sample-cross', tenant_no='tenant-a', create_dept_id='', order_num=1, del_flag=0),
        ]
    )
    db.commit()
    statements = []

    def capture_statement(*args):
        statements.append(args[2])

    event.listen(db.get_bind(), 'before_cursor_execute', capture_statement)
    try:
        items, total = list_presign_orders(
            db, ctx=_ctx('specialist'), user_info=_user(), page_size=10
        )
    finally:
        event.remove(db.get_bind(), 'before_cursor_execute', capture_statement)
    by_id = {item['id']: item for item in items}
    normal_detail = get_presign_order(
        db,
        order_id=normal.id,
        ctx=_ctx('specialist'),
        user_info=_user(),
    )
    historical_detail = get_presign_order(
        db,
        order_id=historical.id,
        ctx=_ctx('specialist'),
        user_info=_user(),
    )

    assert total == 3
    assert by_id[normal.id]['regionName'] == 'A区'
    assert by_id[normal.id]['hospitalName'] == 'A医院'
    assert by_id[normal.id]['sampleTypeIds'] == ['sample-a']
    assert by_id[normal.id]['sampleTypeNames'] == ['样本A']
    assert normal_detail['regionName'] == 'A区'
    assert normal_detail['hospitalName'] == 'A医院'
    assert normal_detail['sampleTypeNames'] == ['样本A']
    assert by_id[historical.id]['regionName'] == '历史地区'
    assert by_id[historical.id]['hospitalName'] == '历史医院'
    assert by_id[historical.id]['sampleTypeNames'] == ['历史样本']
    assert historical_detail['regionName'] == '历史地区'
    assert historical_detail['hospitalName'] == '历史医院'
    assert historical_detail['sampleTypeNames'] == ['历史样本']
    assert by_id[cross.id]['regionName'] == ''
    assert by_id[cross.id]['hospitalName'] == ''
    assert by_id[cross.id]['sampleTypeNames'] == ['']
    assert len(statements) <= 8


def test_presign_response_uses_historical_nicknames_without_raw_fallbacks():
    from app.services.market_presign import list_presign_orders

    db = _session()
    _add_master_data(db)
    db.get(SysUser, 'u1').nick_name = ''
    db.add_all(
        [
            _order(
                "own",
                sales_user_id="u1",
                manager_user_id="disabled-user",
            ),
            _order(
                "cross-reference",
                sales_user_id="u1",
                manager_user_id="cross-user",
            ),
        ]
    )
    row = db.query(BizPresignOrder).filter_by(id="own").one_or_none()
    if row is None:
        db.flush()
        row = db.query(BizPresignOrder).filter_by(id="own").one()
    row.collector_user_id = "deleted-user"
    db.commit()

    items, total = list_presign_orders(
        db, ctx=_ctx("specialist"), user_info=_user()
    )
    by_id = {item['id']: item for item in items}

    assert total == 2
    assert by_id['own']["salesUserName"] == ""
    assert by_id['own']["collectorUserName"] == "已删除秘密姓名"
    assert by_id['own']["regionManagerUserName"] == "停用秘密姓名"
    assert by_id['cross-reference']["regionManagerUserName"] == ""
    assert "跨租户秘密姓名" not in str(items)


def test_presign_detail_rejects_guessed_out_of_scope_id():
    from app.services.market_presign import get_presign_order

    db = _session()
    db.add(_order("other", sales_user_id="u2"))
    db.commit()

    with pytest.raises(HTTPException) as exc_info:
        get_presign_order(
            db,
            order_id="other",
            ctx=_ctx("specialist"),
            user_info=_user(),
        )

    assert exc_info.value.status_code == 404


def test_presign_body_rejects_client_controlled_sales_user():
    from pydantic import ValidationError

    from app.schemas.market_mobile import MarketPresignOrderBody

    with pytest.raises(ValidationError):
        MarketPresignOrderBody.model_validate({'salesUserId': 'attacker-selected-user'})


@pytest.mark.parametrize('payload', [{'motherName': '   '}, {'sampleTypeIds': []}])
def test_presign_update_body_rejects_explicitly_empty_required_fields(payload):
    from pydantic import ValidationError

    from app.schemas.market_mobile import MarketPresignOrderBody

    with pytest.raises(ValidationError):
        MarketPresignOrderBody.model_validate(payload)


def test_specialist_create_forces_current_sales_user_and_accepts_order_status():
    from app.schemas.market_mobile import MarketPresignOrderBody
    from app.services.market_presign import create_presign_order

    db = _session()
    _add_master_data(db)
    body = MarketPresignOrderBody.model_validate(
        {
            "motherName": "测试客户",
            "regionId": "region-a",
            "hospitalId": "hospital-a",
            "sampleTypeIds": ["sample-a"],
            "orderStatus": "void",
        }
    )

    data = create_presign_order(
        db,
        body=body,
        ctx=_ctx("specialist"),
        user_info=_user(),
    )
    row = db.query(BizPresignOrder).filter_by(id=data["id"]).one()

    assert row.tenant_no == "tenant-a"
    assert row.sales_user_id == "u1"
    assert row.order_status == "void"


def test_create_rejects_client_requested_converted_status_without_writing():
    from app.schemas.market_mobile import MarketPresignOrderBody
    from app.services.market_presign import create_presign_order

    db = _session()
    _add_master_data(db)

    with pytest.raises(ValueError, match='转正式|converted'):
        create_presign_order(
            db,
            body=MarketPresignOrderBody.model_validate(
                {
                    'motherName': '不可直转客户',
                    'regionId': 'region-a',
                    'hospitalId': 'hospital-a',
                    'orderStatus': 'converted',
                }
            ),
            ctx=_ctx('specialist'),
            user_info=_user(),
        )

    assert db.query(BizPresignOrder).count() == 0


@pytest.mark.parametrize('current_status', ['active', 'draft', 'void'])
def test_update_rejects_client_requested_converted_status_without_writing(
    current_status,
):
    from app.schemas.market_mobile import MarketPresignOrderBody
    from app.services.market_presign import update_presign_order

    db = _session()
    _add_master_data(db)
    order = _order('own', sales_user_id='u1')
    order.order_status = current_status
    order.remark = 'before'
    db.add(order)
    db.commit()

    with pytest.raises(ValueError, match='转正式|converted'):
        update_presign_order(
            db,
            order_id='own',
            body=MarketPresignOrderBody.model_validate(
                {'orderStatus': 'converted', 'remark': 'after'}
            ),
            ctx=_ctx('specialist'),
            user_info=_user(),
        )

    db.expire_all()
    unchanged = db.query(BizPresignOrder).filter_by(id='own').one()
    assert unchanged.order_status == current_status
    assert unchanged.remark == 'before'


def test_update_rolls_back_when_response_serialization_fails(monkeypatch):
    from app.crud import presign_order as presign_crud
    from app.schemas.market_mobile import MarketPresignOrderBody
    from app.services.market_presign import update_presign_order

    db = _session()
    _add_master_data(db)
    order = _order('own', sales_user_id='u1')
    order.remark = 'before'
    db.add(order)
    db.commit()

    monkeypatch.setattr(
        presign_crud,
        'order_to_dict',
        lambda *args, **kwargs: (_ for _ in ()).throw(
            RuntimeError('serialize failed')
        ),
    )

    with pytest.raises(RuntimeError, match='serialize failed'):
        update_presign_order(
            db,
            order_id='own',
            body=MarketPresignOrderBody(remark='after'),
            ctx=_ctx('specialist'),
            user_info=_user(),
        )

    db.expire_all()
    assert db.query(BizPresignOrder).filter_by(id='own').one().remark == 'before'


def test_create_rolls_back_order_children_and_transfers_when_data_scope_rejects():
    from app.schemas.market_mobile import MarketPresignOrderBody
    from app.services.market_presign import create_presign_order

    db = _session()
    _add_master_data(db)
    restricted_user = {
        "user_id": "u1",
        "tenant_no": "tenant-a",
        "role": "ordinary",
        "dept_id": "dept-a",
    }

    with pytest.raises(HTTPException):
        create_presign_order(
            db,
            body=MarketPresignOrderBody.model_validate(
                {
                    "regionId": "region-a",
                    "hospitalId": "hospital-a",
                    "sampleTypeIds": ["sample-a"],
                    "transfers": [{"transferAmount": 100}],
                }
            ),
            ctx=_ctx("specialist"),
            user_info=restricted_user,
        )

    assert db.query(BizPresignOrder).count() == 0
    assert db.query(BizPresignOrderSampleType).count() == 0
    assert db.query(BizPresignTransfer).count() == 0


def test_create_uses_real_auth_region_id_as_department_for_all_rows():
    from app.schemas.market_mobile import MarketPresignOrderBody
    from app.services.market_presign import create_presign_order

    db = _session()
    _add_master_data(db)

    data = create_presign_order(
        db,
        body=MarketPresignOrderBody.model_validate(
            {
                "regionId": "region-a",
                "hospitalId": "hospital-a",
                "sampleTypeIds": ["sample-a"],
                "transfers": [{"transferAmount": 100}],
            }
        ),
        ctx=_ctx("specialist"),
        user_info=_real_auth_user(),
    )

    assert db.query(BizPresignOrder).filter_by(id=data["id"]).one().create_dept_id == "dept-from-auth"
    assert db.query(BizPresignOrderSampleType).one().create_dept_id == "dept-from-auth"
    assert db.query(BizPresignTransfer).one().create_dept_id == "dept-from-auth"


@pytest.mark.parametrize("user_id", ["cross-user", "deleted-user", "disabled-user"])
def test_create_rejects_cross_tenant_deleted_or_disabled_person(user_id):
    from app.schemas.market_mobile import MarketPresignOrderBody
    from app.services.market_presign import create_presign_order

    db = _session()
    _add_master_data(db)

    with pytest.raises(ValueError, match="用户"):
        create_presign_order(
            db,
            body=MarketPresignOrderBody.model_validate(
                {"regionId": "region-a", "collectorUserId": user_id}
            ),
            ctx=_ctx("specialist"),
            user_info=_user(),
        )

    assert db.query(BizPresignOrder).count() == 0


def test_manager_create_cannot_self_assign_an_unmanaged_region():
    from app.schemas.market_mobile import MarketPresignOrderBody
    from app.services.market_presign import create_presign_order

    db = _session()
    _add_master_data(db)
    db.add(
        BizCollectionRegionStaff(
            id='manager-region-a',
            region_id='region-a',
            user_id='u1',
            is_leader=1,
            tenant_no='tenant-a',
            create_dept_id='dept-a',
            del_flag=0,
        )
    )
    db.commit()

    with pytest.raises(ValueError, match="地区经理范围"):
        create_presign_order(
            db,
            body=MarketPresignOrderBody.model_validate(
                {
                    "motherName": "越权订单",
                    "regionId": "region-b",
                    "regionManagerUserId": "u1",
                    "sampleTypeIds": ["sample-a"],
                }
            ),
            ctx=_ctx("manager"),
            user_info=_user(),
        )

    assert db.query(BizPresignOrder).count() == 0


@pytest.mark.parametrize(
    'payload',
    [
        {'sampleTypeIds': ['sample-a']},
        {'motherName': '   ', 'sampleTypeIds': ['sample-a']},
        {'motherName': '测试客户'},
        {'motherName': '测试客户', 'sampleTypeIds': []},
    ],
)
def test_create_requires_mother_name_and_sample_types(payload):
    from pydantic import ValidationError

    from app.schemas.market_mobile import MarketPresignOrderCreateBody

    with pytest.raises(ValidationError, match='motherName|sampleTypeIds|母亲姓名'):
        MarketPresignOrderCreateBody.model_validate(payload)


@pytest.mark.parametrize(
    ("payload", "message"),
    [
        ({"regionId": "region-cross"}, "采集地区"),
        ({"regionId": "region-missing"}, "采集地区"),
        (
            {"regionId": "region-a", "hospitalId": "hospital-b"},
            "医院",
        ),
        (
            {"regionId": "region-a", "sampleTypeIds": ["sample-cross"]},
            "样本类型",
        ),
        (
            {"regionId": "region-a", "sampleTypeIds": ["sample-missing"]},
            "样本类型",
        ),
    ],
)
def test_create_rejects_invalid_or_cross_tenant_master_data(payload, message):
    from app.schemas.market_mobile import MarketPresignOrderBody
    from app.services.market_presign import create_presign_order

    db = _session()
    _add_master_data(db)

    with pytest.raises(ValueError, match=message):
        create_presign_order(
            db,
            body=MarketPresignOrderBody.model_validate(payload),
            ctx=_ctx("specialist"),
            user_info=_user(),
        )

    assert db.query(BizPresignOrder).count() == 0


def test_update_validates_resulting_hospital_region_and_sample_types_before_write():
    from app.schemas.market_mobile import MarketPresignOrderBody
    from app.services.market_presign import update_presign_order

    db = _session()
    _add_master_data(db)
    db.add(
        _order(
            "own",
            region_id="region-a",
            sales_user_id="u1",
        )
    )
    db.commit()

    with pytest.raises(ValueError, match="医院"):
        update_presign_order(
            db,
            order_id="own",
            body=MarketPresignOrderBody.model_validate(
                {"hospitalId": "hospital-b", "sampleTypeIds": ["sample-cross"]}
            ),
            ctx=_ctx("specialist"),
            user_info=_user(),
        )

    row = db.query(BizPresignOrder).filter_by(id="own").one()
    assert row.hospital_id is None
    assert db.query(BizPresignOrderSampleType).count() == 0


def test_update_rebuilt_sample_types_keep_current_user_department():
    from app.schemas.market_mobile import MarketPresignOrderBody
    from app.services.market_presign import update_presign_order

    db = _session()
    _add_master_data(db)
    db.add(_order("own", region_id="region-a", sales_user_id="u1"))
    db.commit()

    update_presign_order(
        db,
        order_id="own",
        body=MarketPresignOrderBody.model_validate(
            {"sampleTypeIds": ["sample-a"]}
        ),
        ctx=_ctx("specialist"),
        user_info=_real_auth_user(),
    )

    child = db.query(BizPresignOrderSampleType).one()
    assert child.create_dept_id == "dept-from-auth"


def test_update_rejects_disabled_person_before_write():
    from app.schemas.market_mobile import MarketPresignOrderBody
    from app.services.market_presign import update_presign_order

    db = _session()
    _add_master_data(db)
    db.add(_order("own", region_id="region-a", sales_user_id="u1"))
    db.commit()

    with pytest.raises(ValueError, match="用户"):
        update_presign_order(
            db,
            order_id="own",
            body=MarketPresignOrderBody.model_validate(
                {"collectorUserId": "disabled-user"}
            ),
            ctx=_ctx("specialist"),
            user_info=_user(),
        )

    assert db.query(BizPresignOrder).filter_by(id="own").one().collector_user_id is None


def test_update_rejects_resulting_assignment_outside_manager_scope():
    from app.schemas.market_mobile import MarketPresignOrderBody
    from app.services.market_presign import update_presign_order

    db = _session()
    _add_master_data(db)
    db.add(_order("managed", region_id="region-a", manager_user_id="u1"))
    db.commit()

    with pytest.raises(ValueError, match="地区经理范围"):
        update_presign_order(
            db,
            order_id="managed",
            body=MarketPresignOrderBody.model_validate(
                {
                    "regionId": "region-b",
                    "regionManagerUserId": "u2",
                }
            ),
            ctx=_ctx("manager"),
            user_info=_user(),
        )

    row = db.query(BizPresignOrder).filter_by(id="managed").one()
    assert row.region_id == "region-a"
    assert row.region_manager_user_id == "u1"


def test_convert_rechecks_scope_before_reusing_presign_conversion(monkeypatch):
    from app.crud import presign_order as presign_crud
    from app.services.market_presign import convert_presign_order

    db = _session()
    _add_master_data(db)
    db.add_all(
        [
            _order("own", sales_user_id="u1", manager_user_id="manager-a"),
            _order("other", sales_user_id="u2", manager_user_id="manager-b"),
        ]
    )
    db.commit()
    calls = []

    def fake_convert(db, *, order_id, tenant_no, user_id, create_dept_id=""):
        calls.append((order_id, tenant_no, user_id, create_dept_id))
        return {"protocolId": "protocol-1"}

    monkeypatch.setattr(presign_crud, "convert_to_protocol", fake_convert)

    with pytest.raises(HTTPException) as exc_info:
        convert_presign_order(
            db,
            order_id="other",
            ctx=_ctx("specialist"),
            user_info=_user(),
        )
    assert exc_info.value.status_code == 404
    assert calls == []

    assert convert_presign_order(
        db,
        order_id="own",
        ctx=_ctx("specialist"),
        user_info=_real_auth_user(),
    ) == {"protocolId": "protocol-1"}
    assert calls == [("own", "tenant-a", "u1", "dept-from-auth")]


def test_convert_rolls_back_protocol_children_and_presign_status_on_midway_failure(
    monkeypatch,
):
    from app.crud import collection_protocol as protocol_crud
    from app.services.market_presign import convert_presign_order

    db = _session()
    _add_master_data(db)
    order = _order(
        "own",
        region_id="region-a",
        sales_user_id="u1",
        manager_user_id="manager-a",
    )
    order.protocol_no = "P-ATOMIC-1"
    order.hospital_id = "hospital-a"
    db.add(order)
    db.add(
        BizPresignOrderSampleType(
            id="presign-sample",
            presign_order_id="own",
            sample_type_id="sample-a",
            tenant_no="tenant-a",
            create_dept_id="dept-a",
            del_flag=0,
        )
    )
    db.commit()

    def fail_after_collection(*args, **kwargs):
        raise RuntimeError("injected sample failure")

    monkeypatch.setattr(protocol_crud, "save_samples", fail_after_collection)

    with pytest.raises(RuntimeError, match="injected sample failure"):
        convert_presign_order(
            db,
            order_id="own",
            ctx=_ctx("specialist"),
            user_info=_user(),
        )

    db.expire_all()
    unchanged = db.query(BizPresignOrder).filter_by(id="own").one()
    assert unchanged.order_status == "active"
    assert unchanged.converted_protocol_id is None
    assert db.query(BizCollectionProtocol).count() == 0
    assert db.query(BizProtocolPartyCustomer).count() == 0
    assert db.query(BizProtocolCollectionInfo).count() == 0
    assert db.query(BizProtocolSample).count() == 0


def test_convert_creates_protocol_in_effective_status():
    """预签约转正式后，正式协议状态应为生效中(sxz)。"""
    from app.services.market_presign import convert_presign_order

    db = _session()
    _add_master_data(db)
    order = _order(
        "own",
        region_id="region-a",
        sales_user_id="u1",
        manager_user_id="manager-a",
    )
    order.protocol_no = "P-CONVERT-SXZ"
    order.hospital_id = "hospital-a"
    db.add(order)
    db.commit()

    result = convert_presign_order(
        db,
        order_id="own",
        ctx=_ctx("specialist"),
        user_info=_user(),
    )

    protocol = db.query(BizCollectionProtocol).filter_by(id=result["protocolId"]).one()
    assert protocol.protocol_status == "sxz"


def test_convert_rejects_invalid_person_before_creating_protocol():
    from app.services.market_presign import convert_presign_order

    db = _session()
    _add_master_data(db)
    order = _order("own", sales_user_id="u1")
    order.protocol_no = "P-INVALID-USER"
    order.collector_user_id = "cross-user"
    db.add(order)
    db.commit()

    with pytest.raises(ValueError, match="用户"):
        convert_presign_order(
            db,
            order_id="own",
            ctx=_ctx("specialist"),
            user_info=_user(),
        )

    assert db.query(BizCollectionProtocol).count() == 0


def test_convert_fetches_presign_row_with_for_update(monkeypatch):
    from app.crud import presign_order as presign_crud

    class QueryProbe:
        def __init__(self):
            self.locked = False

        def filter(self, *args):
            return self

        def with_for_update(self):
            self.locked = True
            return self

        def populate_existing(self):
            return self

        def first(self):
            return object()

    class DbProbe:
        def __init__(self):
            self.query_probe = QueryProbe()

        def query(self, model):
            return self.query_probe

    db = DbProbe()
    assert presign_crud._get_order_or_raise(
        db, order_id="order-1", tenant_no="tenant-a", for_update=True
    ) is not None
    assert db.query_probe.locked is True


def test_locked_convert_read_refreshes_preloaded_stale_identity(tmp_path):
    from app.crud import presign_order as presign_crud

    engine = create_engine(f"sqlite:///{tmp_path / 'stale.db'}")
    Base.metadata.create_all(engine, tables=[BizPresignOrder.__table__])
    Session = sessionmaker(bind=engine)
    first = Session()
    second = Session()
    try:
        first.add(_order("race", sales_user_id="u1"))
        first.commit()
        cached = first.query(BizPresignOrder).filter_by(id="race").one()
        assert cached.order_status == "active"

        concurrent = second.query(BizPresignOrder).filter_by(id="race").one()
        concurrent.order_status = "converted"
        concurrent.converted_protocol_id = "protocol-existing"
        second.commit()

        locked = presign_crud._get_order_or_raise(
            first,
            order_id="race",
            tenant_no="tenant-a",
            for_update=True,
        )
        assert locked is cached
        assert locked.order_status == "converted"
        assert locked.converted_protocol_id == "protocol-existing"
        with pytest.raises(ValueError, match="已转正式"):
            presign_crud.assert_presign_convertible(
                order_status=locked.order_status
            )
    finally:
        first.close()
        second.close()


def test_convert_translates_unique_race_to_already_converted_domain_error(
    monkeypatch,
):
    from app.crud import collection_protocol as protocol_crud
    from app.services.market_presign import convert_presign_order

    db = _session()
    _add_master_data(db)
    order = _order("own", sales_user_id="u1", manager_user_id="manager-a")
    order.protocol_no = "P-RACE"
    db.add(order)
    db.commit()

    def duplicate_source(*args, **kwargs):
        raise IntegrityError(
            "INSERT INTO biz_collection_protocol",
            {},
            Exception("Duplicate entry for key 'uk_collection_protocol_presign_source'"),
        )

    monkeypatch.setattr(
        protocol_crud, "create_collection_protocol", duplicate_source
    )

    with pytest.raises(ValueError, match="已转正式"):
        convert_presign_order(
            db,
            order_id="own",
            ctx=_ctx("specialist"),
            user_info=_user(),
        )

    assert db.query(BizPresignOrder).filter_by(id="own").one().order_status == "active"


def test_market_presign_bodies_forbid_unknown_fields_and_validate_transfer_limits():
    from pydantic import ValidationError

    from app.schemas.market_mobile import (
        MarketPresignOrderBody,
        MarketPresignTransferBody,
    )

    with pytest.raises(ValidationError, match='tenantNo'):
        MarketPresignOrderBody.model_validate({'tenantNo': 'tenant-b'})
    assert (
        MarketPresignOrderBody.model_validate(
            {'orderStatus': 'void'}
        ).order_status
        == 'void'
    )

    valid = MarketPresignTransferBody.model_validate(
        {
            'transferAmount': '123.45',
            'transferTime': '2026-10-11 12:34:56',
            'transferType': 'bank',
            'transferAttachments': 'a' * 4000,
            'remark': 'r' * 512,
        }
    )
    assert str(valid.transfer_amount) == '123.45'
    assert valid.transfer_time == '2026-10-11 12:34:56'
    assert (
        MarketPresignTransferBody.model_validate(
            {'transferTime': '2026-10-11T12:34'}
        ).transfer_time
        == '2026-10-11T12:34'
    )
    for payload in (
        {'unknown': 'value'},
        {'transferAmount': 0},
        {'transferTime': 'not-a-datetime'},
        {'transferTime': '2026-10-11'},
        {'transferTime': '2026-10-11 12:34'},
        {'transferTime': '2026-10-11T12'},
        {'transferTime': '2026-10-11T12:34+08:00'},
        {'transferType': 'x' * 17},
        {'transferAttachments': 'x' * 4001},
        {'remark': 'x' * 513},
    ):
        with pytest.raises(ValidationError):
            MarketPresignTransferBody.model_validate(payload)


def test_market_presign_order_body_accepts_database_sized_values():
    from decimal import Decimal

    from app.schemas.market_mobile import MarketPresignOrderBody

    body = MarketPresignOrderBody.model_validate(
        {
            'orderDate': '20261011',
            'protocolNo': 'P' * 64,
            'motherName': '母' * 64,
            'expectedDeliveryDate': '20261231',
            'regionId': 'r' * 64,
            'hospitalId': 'h' * 64,
            'sampleCollectTime': '2026-10-11 12:34:56',
            'payMode': 'x' * 8,
            'contractAmount': '9999999999999999.99',
            'collectorUserId': 'c' * 64,
            'regionManagerUserId': 'm' * 64,
            'remark': '备' * 512,
            'sampleTypeIds': [f'sample-{index}' for index in range(100)],
        }
    )

    assert body.order_date == '20261011'
    assert body.expected_delivery_date == '20261231'
    assert body.contract_amount == Decimal('9999999999999999.99')
    assert len(body.sample_type_ids or []) == 100


@pytest.mark.parametrize(
    'payload',
    [
        {'orderDate': '2026-10-11'},
        {'orderDate': '20260230'},
        {'expectedDeliveryDate': '20261301'},
        {'sampleCollectTime': 'not-a-datetime'},
        {'protocolNo': 'p' * 65},
        {'motherName': 'm' * 65},
        {'regionId': 'r' * 65},
        {'hospitalId': 'h' * 65},
        {'collectorUserId': 'c' * 65},
        {'salesUserId': 's' * 65},
        {'regionManagerUserId': 'm' * 65},
        {'payMode': 'p' * 9},
        {'remark': 'r' * 513},
        {'sampleTypeIds': ['s' * 65]},
        {'sampleTypeIds': [f'sample-{index}' for index in range(101)]},
        {'contractAmount': 'NaN'},
        {'contractAmount': 'Infinity'},
        {'contractAmount': '-0.01'},
        {'contractAmount': '10000000000000000.00'},
        {'contractAmount': '1.001'},
    ],
)
def test_market_presign_order_body_rejects_values_outside_database_bounds(
    payload,
):
    from pydantic import ValidationError

    from app.schemas.market_mobile import MarketPresignOrderBody

    with pytest.raises(ValidationError):
        MarketPresignOrderBody.model_validate(payload)


@pytest.mark.parametrize('status', ['active', 'void'])
def test_specialist_can_soft_delete_own_non_converted_presign(status):
    from app.services.market_presign import delete_presign_order

    db = _session()
    order = _order('own', sales_user_id='u1')
    order.order_status = status
    db.add(order)
    db.commit()

    delete_presign_order(
        db,
        order_id='own',
        ctx=_ctx('specialist'),
        user_info=_user(),
    )

    deleted = db.query(BizPresignOrder).filter_by(id='own').one()
    assert deleted.del_flag == 1
    assert deleted.update_by == 'u1'


def test_specialist_cannot_delete_another_specialists_presign():
    from app.services.market_presign import delete_presign_order

    db = _session()
    db.add(_order('other', sales_user_id='u2'))
    db.commit()

    with pytest.raises(HTTPException) as exc_info:
        delete_presign_order(
            db,
            order_id='other',
            ctx=_ctx('specialist'),
            user_info=_user(),
        )

    assert exc_info.value.status_code == 404
    assert db.query(BizPresignOrder).filter_by(id='other').one().del_flag == 0


def test_manager_delete_is_limited_to_manager_scope():
    from app.services.market_presign import delete_presign_order

    db = _session()
    db.add(
        BizCollectionRegionStaff(
            id='manager-region',
            region_id='region-led',
            user_id='u1',
            is_leader=1,
            tenant_no='tenant-a',
            create_dept_id='dept-a',
            del_flag=0,
        )
    )
    db.add_all(
        [
            _order(
                'managed',
                region_id='region-led',
                sales_user_id='u2',
                manager_user_id='u2',
            ),
            _order(
                'outside',
                region_id='region-other',
                sales_user_id='u2',
                manager_user_id='u2',
            ),
        ]
    )
    db.commit()

    delete_presign_order(
        db,
        order_id='managed',
        ctx=_ctx('manager'),
        user_info=_user(),
    )
    with pytest.raises(HTTPException) as exc_info:
        delete_presign_order(
            db,
            order_id='outside',
            ctx=_ctx('manager'),
            user_info=_user(),
        )

    assert exc_info.value.status_code == 404
    assert db.query(BizPresignOrder).filter_by(id='managed').one().del_flag == 1
    assert db.query(BizPresignOrder).filter_by(id='outside').one().del_flag == 0


def test_converted_presign_cannot_be_deleted():
    from app.services.market_presign import delete_presign_order

    db = _session()
    order = _order('converted', sales_user_id='u1')
    order.order_status = 'converted'
    db.add(order)
    db.commit()

    with pytest.raises(ValueError, match='已转|converted'):
        delete_presign_order(
            db,
            order_id='converted',
            ctx=_ctx('specialist'),
            user_info=_user(),
        )

    assert db.query(BizPresignOrder).filter_by(id='converted').one().del_flag == 0


@pytest.mark.parametrize('scope_kind', ['specialist', 'manager'])
@pytest.mark.parametrize(
    ('action', 'pay_status'),
    [
        ('create', 'dfj'),
        ('update', 'dfj'),
        ('delete', 'dfj'),
        ('submit', 'dfj'),
        ('withdraw', 'jfqrz'),
    ],
)
def test_presign_transfer_writes_recheck_parent_order_scope(
    scope_kind, action, pay_status
):
    from app.schemas.market_mobile import MarketPresignTransferBody
    from app.services.market_presign import (
        create_presign_transfer,
        delete_presign_transfer,
        submit_presign_transfer,
        update_presign_transfer,
        withdraw_presign_transfer,
    )

    db = _session()
    db.add(
        _order(
            'outside',
            region_id='region-outside',
            sales_user_id='u2',
            manager_user_id='u2',
        )
    )
    db.add(_transfer('outside-transfer', order_id='outside', pay_status=pay_status))
    db.commit()
    body = MarketPresignTransferBody(
        transferAmount=200,
        transferType='bank',
    )

    calls = {
        'create': lambda: create_presign_transfer(
            db,
            order_id='outside',
            body=body,
            ctx=_ctx(scope_kind),
            user_info=_user(),
        ),
        'update': lambda: update_presign_transfer(
            db,
            transfer_id='outside-transfer',
            body=body,
            ctx=_ctx(scope_kind),
            user_info=_user(),
        ),
        'delete': lambda: delete_presign_transfer(
            db,
            transfer_id='outside-transfer',
            ctx=_ctx(scope_kind),
            user_info=_user(),
        ),
        'submit': lambda: submit_presign_transfer(
            db,
            transfer_id='outside-transfer',
            ctx=_ctx(scope_kind),
            user_info=_user(),
        ),
        'withdraw': lambda: withdraw_presign_transfer(
            db,
            transfer_id='outside-transfer',
            ctx=_ctx(scope_kind),
            user_info=_user(),
        ),
    }

    with pytest.raises(HTTPException) as exc_info:
        calls[action]()

    assert exc_info.value.status_code == 404
    row = db.query(BizPresignTransfer).filter_by(id='outside-transfer').one()
    assert row.del_flag == 0
    assert row.pay_status == pay_status


def test_dfj_presign_transfer_can_edit_submit_withdraw_and_soft_delete():
    from app.schemas.market_mobile import MarketPresignTransferBody
    from app.services.market_presign import (
        create_presign_transfer,
        delete_presign_transfer,
        submit_presign_transfer,
        update_presign_transfer,
        withdraw_presign_transfer,
    )

    db = _session()
    db.add(_order('own', sales_user_id='u1'))
    db.commit()

    created = create_presign_transfer(
        db,
        order_id='own',
        body=MarketPresignTransferBody(
            transferAmount='100.50',
            transferType='bank',
        ),
        ctx=_ctx('specialist'),
        user_info=_real_auth_user(),
    )
    transfer_id = created['id']
    updated = update_presign_transfer(
        db,
        transfer_id=transfer_id,
        body=MarketPresignTransferBody(transferAmount='200.25', remark='更新'),
        ctx=_ctx('specialist'),
        user_info=_user(),
    )
    assert updated['transferAmount'] == '200.25'
    assert updated['remark'] == '更新'
    assert (
        db.query(BizPresignTransfer).filter_by(id=transfer_id).one().create_dept_id
        == 'dept-from-auth'
    )

    submitted = submit_presign_transfer(
        db,
        transfer_id=transfer_id,
        ctx=_ctx('specialist'),
        user_info=_user(),
    )
    assert submitted['payStatus'] == 'jfqrz'
    with pytest.raises(ValueError, match='状态|待缴费|已缴费|已付款'):
        delete_presign_transfer(
            db,
            transfer_id=transfer_id,
            ctx=_ctx('specialist'),
            user_info=_user(),
        )

    withdrawn = withdraw_presign_transfer(
        db,
        transfer_id=transfer_id,
        ctx=_ctx('specialist'),
        user_info=_user(),
    )
    assert withdrawn['payStatus'] == 'dfj'
    delete_presign_transfer(
        db,
        transfer_id=transfer_id,
        ctx=_ctx('specialist'),
        user_info=_user(),
    )
    deleted = db.query(BizPresignTransfer).filter_by(id=transfer_id).one()
    assert deleted.del_flag == 1
    assert deleted.update_by == 'u1'


def test_jfqrz_presign_transfer_only_allows_withdraw():
    from app.schemas.market_mobile import MarketPresignTransferBody
    from app.services.market_presign import (
        delete_presign_transfer,
        submit_presign_transfer,
        update_presign_transfer,
        withdraw_presign_transfer,
    )

    db = _session()
    db.add(_order('own', sales_user_id='u1'))
    db.add(_transfer('pending', pay_status='jfqrz'))
    db.commit()

    for action in (
        lambda: update_presign_transfer(
            db,
            transfer_id='pending',
            body=MarketPresignTransferBody(remark='不允许'),
            ctx=_ctx('specialist'),
            user_info=_user(),
        ),
        lambda: delete_presign_transfer(
            db,
            transfer_id='pending',
            ctx=_ctx('specialist'),
            user_info=_user(),
        ),
        lambda: submit_presign_transfer(
            db,
            transfer_id='pending',
            ctx=_ctx('specialist'),
            user_info=_user(),
        ),
    ):
        with pytest.raises(ValueError):
            action()

    assert withdraw_presign_transfer(
        db,
        transfer_id='pending',
        ctx=_ctx('specialist'),
        user_info=_user(),
    )['payStatus'] == 'dfj'


@pytest.mark.parametrize('action', ['update', 'delete', 'submit', 'withdraw'])
def test_yjf_presign_transfer_rejects_all_market_writes(action):
    from app.schemas.market_mobile import MarketPresignTransferBody
    from app.services.market_presign import (
        delete_presign_transfer,
        submit_presign_transfer,
        update_presign_transfer,
        withdraw_presign_transfer,
    )

    db = _session()
    db.add(_order('own', sales_user_id='u1'))
    db.add(_transfer('paid', pay_status='yjf'))
    db.commit()
    calls = {
        'update': lambda: update_presign_transfer(
            db,
            transfer_id='paid',
            body=MarketPresignTransferBody(remark='不允许'),
            ctx=_ctx('specialist'),
            user_info=_user(),
        ),
        'delete': lambda: delete_presign_transfer(
            db,
            transfer_id='paid',
            ctx=_ctx('specialist'),
            user_info=_user(),
        ),
        'submit': lambda: submit_presign_transfer(
            db,
            transfer_id='paid',
            ctx=_ctx('specialist'),
            user_info=_user(),
        ),
        'withdraw': lambda: withdraw_presign_transfer(
            db,
            transfer_id='paid',
            ctx=_ctx('specialist'),
            user_info=_user(),
        ),
    }

    with pytest.raises(ValueError):
        calls[action]()


@pytest.mark.parametrize('order_status', ['converted', 'void'])
@pytest.mark.parametrize('action', ['create', 'update', 'delete', 'submit', 'withdraw'])
def test_non_writable_presign_order_rejects_every_transfer_write(
    order_status, action
):
    from app.schemas.market_mobile import MarketPresignTransferBody
    from app.services.market_presign import (
        create_presign_transfer,
        delete_presign_transfer,
        submit_presign_transfer,
        update_presign_transfer,
        withdraw_presign_transfer,
    )

    db = _session()
    order = _order('own', sales_user_id='u1')
    order.order_status = order_status
    db.add(order)
    db.add(
        _transfer(
            'transfer',
            pay_status='jfqrz' if action == 'withdraw' else 'dfj',
        )
    )
    db.commit()
    body = MarketPresignTransferBody(
        transferAmount=100,
        transferType='bank',
    )
    calls = {
        'create': lambda: create_presign_transfer(
            db,
            order_id='own',
            body=body,
            ctx=_ctx('specialist'),
            user_info=_user(),
        ),
        'update': lambda: update_presign_transfer(
            db,
            transfer_id='transfer',
            body=body,
            ctx=_ctx('specialist'),
            user_info=_user(),
        ),
        'delete': lambda: delete_presign_transfer(
            db,
            transfer_id='transfer',
            ctx=_ctx('specialist'),
            user_info=_user(),
        ),
        'submit': lambda: submit_presign_transfer(
            db,
            transfer_id='transfer',
            ctx=_ctx('specialist'),
            user_info=_user(),
        ),
        'withdraw': lambda: withdraw_presign_transfer(
            db,
            transfer_id='transfer',
            ctx=_ctx('specialist'),
            user_info=_user(),
        ),
    }

    with pytest.raises(ValueError, match='订单|状态|转账'):
        calls[action]()


@pytest.mark.parametrize('action', ['create', 'update'])
def test_presign_transfer_rejects_unregistered_attachment_url(action):
    from app.schemas.market_mobile import MarketPresignTransferBody
    from app.services.market_presign import (
        create_presign_transfer,
        update_presign_transfer,
    )

    db = _session()
    db.add(_order('own', sales_user_id='u1'))
    db.add(_transfer('draft'))
    db.commit()
    body = MarketPresignTransferBody(
        transferAttachments='https://evil.test/proof.pdf'
    )

    with pytest.raises(ValueError, match='附件'):
        if action == 'create':
            create_presign_transfer(
                db,
                order_id='own',
                body=body,
                ctx=_ctx('specialist'),
                user_info=_user(),
            )
        else:
            update_presign_transfer(
                db,
                transfer_id='draft',
                body=body,
                ctx=_ctx('specialist'),
                user_info=_user(),
            )


def test_presign_transfer_attachments_bind_to_parent_order():
    from app.schemas.market_mobile import MarketPresignTransferBody
    from app.services.market_presign import (
        create_presign_transfer,
        update_presign_transfer,
    )

    db = _session()
    db.add(_order('own', sales_user_id='u1'))
    db.add_all(
        [
            BizAttachment(
                id='attachment-create',
                business_type='market:specialist:presign:staging',
                original_name='create.pdf',
                file_url='presign/create.pdf',
                create_by='u1',
                create_dept_id='dept-a',
                tenant_no='tenant-a',
                del_flag=0,
            ),
            BizAttachment(
                id='attachment-update',
                business_type='market:specialist:presign:staging',
                original_name='update.pdf',
                file_url='presign/update.pdf',
                create_by='u1',
                create_dept_id='dept-a',
                tenant_no='tenant-a',
                del_flag=0,
            ),
        ]
    )
    db.commit()

    created = create_presign_transfer(
        db,
        order_id='own',
        body=MarketPresignTransferBody(
            transferAmount=100,
            transferType='bank',
            transferAttachments='attachment-create',
        ),
        ctx=_ctx('specialist'),
        user_info=_user(),
    )
    updated = update_presign_transfer(
        db,
        transfer_id=created['id'],
        body=MarketPresignTransferBody(
            transferAttachments='attachment-create,attachment-update'
        ),
        ctx=_ctx('specialist'),
        user_info=_user(),
    )

    assert updated['transferAttachments'] == (
        'presign/create.pdf,presign/update.pdf'
    )
    attachments = db.query(BizAttachment).order_by(BizAttachment.id).all()
    assert {row.business_id for row in attachments} == {'own'}
    assert {row.business_type for row in attachments} == {
        'market:specialist:presign:bound'
    }


def test_stale_withdraw_does_not_overwrite_later_finance_confirmation(tmp_path):
    from app.crud import presign_order as presign_crud
    from app.services.market_presign import withdraw_presign_transfer

    Session = _session_factory(f"sqlite:///{tmp_path / 'withdraw-race.db'}")
    market_db = Session()
    finance_db = Session()
    try:
        market_db.add(_order('own', sales_user_id='u1'))
        market_db.add(_transfer('transfer', pay_status='jfqrz'))
        market_db.commit()
        stale = market_db.query(BizPresignTransfer).filter_by(id='transfer').one()
        assert stale.pay_status == 'jfqrz'

        presign_crud.confirm_receipt(
            finance_db,
            transfer_id='transfer',
            tenant_no='tenant-a',
            user_id='finance-user',
        )

        with pytest.raises(ValueError, match='状态|撤回'):
            withdraw_presign_transfer(
                market_db,
                transfer_id='transfer',
                ctx=_ctx('specialist'),
                user_info=_user(),
            )

        finance_db.expire_all()
        current = finance_db.query(BizPresignTransfer).filter_by(id='transfer').one()
        assert current.pay_status == 'yjf'
        assert current.update_by == 'finance-user'
    finally:
        market_db.close()
        finance_db.close()


def test_stale_finance_confirmation_does_not_overwrite_later_mobile_withdraw(
    tmp_path,
):
    from app.crud import presign_order as presign_crud
    from app.services.market_presign import withdraw_presign_transfer

    Session = _session_factory(
        f"sqlite:///{tmp_path / 'finance-confirm-after-withdraw-race.db'}"
    )
    finance_db = Session()
    market_db = Session()
    try:
        finance_db.add(_order('own', sales_user_id='u1'))
        finance_db.add(_transfer('transfer', pay_status='jfqrz'))
        finance_db.commit()
        stale = finance_db.query(BizPresignTransfer).filter_by(id='transfer').one()
        assert stale.pay_status == 'jfqrz'

        withdrawn = withdraw_presign_transfer(
            market_db,
            transfer_id='transfer',
            ctx=_ctx('specialist'),
            user_info=_user(),
        )
        assert withdrawn['payStatus'] == 'dfj'

        with pytest.raises(ValueError, match='状态'):
            presign_crud.confirm_receipt(
                finance_db,
                transfer_id='transfer',
                tenant_no='tenant-a',
                user_id='finance-user',
            )

        market_db.expire_all()
        current = market_db.query(BizPresignTransfer).filter_by(id='transfer').one()
        assert current.pay_status == 'dfj'
        assert current.update_by == 'u1'
    finally:
        finance_db.close()
        market_db.close()


@pytest.mark.parametrize('action', ['update', 'submit'])
def test_stale_web_draft_action_does_not_mutate_later_paid_transfer(
    tmp_path, action
):
    from app.crud import presign_order as presign_crud

    Session = _session_factory(
        f"sqlite:///{tmp_path / f'web-{action}-after-paid-race.db'}"
    )
    web_db = Session()
    finance_db = Session()
    try:
        web_db.add(_order('own', sales_user_id='u1'))
        transfer = _transfer('transfer', pay_status='dfj')
        transfer.remark = 'original'
        web_db.add(transfer)
        web_db.commit()
        stale = web_db.query(BizPresignTransfer).filter_by(id='transfer').one()
        assert stale.pay_status == 'dfj'

        presign_crud.submit_receipt_finance(
            finance_db,
            transfer_id='transfer',
            tenant_no='tenant-a',
            user_id='finance-user',
        )
        presign_crud.confirm_receipt(
            finance_db,
            transfer_id='transfer',
            tenant_no='tenant-a',
            user_id='finance-user',
        )

        calls = {
            'update': lambda: presign_crud.update_transfer_draft(
                web_db,
                transfer_id='transfer',
                tenant_no='tenant-a',
                user_id='web-user',
                fields={'remark': 'stale-write'},
            ),
            'submit': lambda: presign_crud.submit_receipt_finance(
                web_db,
                transfer_id='transfer',
                tenant_no='tenant-a',
                user_id='web-user',
            ),
        }
        with pytest.raises(ValueError, match='状态'):
            calls[action]()

        finance_db.expire_all()
        current = finance_db.query(BizPresignTransfer).filter_by(id='transfer').one()
        assert current.pay_status == 'yjf'
        assert current.remark == 'original'
        assert current.update_by == 'finance-user'
    finally:
        web_db.close()
        finance_db.close()


@pytest.mark.parametrize('action', ['withdraw', 'confirm', 'reject'])
def test_stale_shared_pending_action_does_not_mutate_later_draft_transfer(
    tmp_path, action
):
    from app.crud import presign_order as presign_crud
    from app.services.market_presign import withdraw_presign_transfer

    Session = _session_factory(
        f"sqlite:///{tmp_path / f'shared-{action}-after-withdraw-race.db'}"
    )
    stale_db = Session()
    market_db = Session()
    try:
        stale_db.add(_order('own', sales_user_id='u1'))
        stale_db.add(_transfer('transfer', pay_status='jfqrz'))
        stale_db.commit()
        stale = stale_db.query(BizPresignTransfer).filter_by(id='transfer').one()
        assert stale.pay_status == 'jfqrz'

        withdraw_presign_transfer(
            market_db,
            transfer_id='transfer',
            ctx=_ctx('specialist'),
            user_info=_user(),
        )
        calls = {
            'withdraw': lambda: presign_crud.withdraw_receipt(
                stale_db,
                transfer_id='transfer',
                tenant_no='tenant-a',
                user_id='stale-user',
            ),
            'confirm': lambda: presign_crud.confirm_receipt(
                stale_db,
                transfer_id='transfer',
                tenant_no='tenant-a',
                user_id='stale-user',
            ),
            'reject': lambda: presign_crud.reject_receipt(
                stale_db,
                transfer_id='transfer',
                tenant_no='tenant-a',
                user_id='stale-user',
            ),
        }
        with pytest.raises(ValueError, match='状态'):
            calls[action]()

        market_db.expire_all()
        current = market_db.query(BizPresignTransfer).filter_by(id='transfer').one()
        assert current.pay_status == 'dfj'
        assert current.update_by == 'u1'
    finally:
        stale_db.close()
        market_db.close()


def test_mobile_transfer_update_serializes_before_commit(monkeypatch):
    from app.crud import presign_order as presign_crud
    from app.schemas.market_mobile import MarketPresignTransferBody
    from app.services.market_presign import update_presign_transfer

    db = _session()
    db.add(_order('own', sales_user_id='u1'))
    db.add(_transfer('transfer', pay_status='dfj'))
    db.commit()

    committed = False
    original_commit = db.commit
    original_to_dict = presign_crud._transfer_to_dict

    def tracked_commit():
        nonlocal committed
        committed = True
        original_commit()

    def snapshot_before_commit(*args, **kwargs):
        assert not committed
        return original_to_dict(*args, **kwargs)

    monkeypatch.setattr(db, 'commit', tracked_commit)
    monkeypatch.setattr(presign_crud, '_transfer_to_dict', snapshot_before_commit)

    result = update_presign_transfer(
        db,
        transfer_id='transfer',
        body=MarketPresignTransferBody(remark='updated'),
        ctx=_ctx('specialist'),
        user_info=_user(),
    )

    assert committed
    assert result['remark'] == 'updated'


@pytest.mark.parametrize('action', ['update', 'delete', 'submit'])
def test_stale_dfj_action_does_not_mutate_later_paid_transfer(tmp_path, action):
    from app.crud import presign_order as presign_crud
    from app.schemas.market_mobile import MarketPresignTransferBody
    from app.services.market_presign import (
        delete_presign_transfer,
        submit_presign_transfer,
        update_presign_transfer,
    )

    Session = _session_factory(
        f"sqlite:///{tmp_path / f'transfer-{action}-race.db'}"
    )
    market_db = Session()
    finance_db = Session()
    try:
        market_db.add(_order('own', sales_user_id='u1'))
        transfer = _transfer('transfer', pay_status='dfj')
        transfer.remark = 'original'
        market_db.add(transfer)
        market_db.commit()
        stale = market_db.query(BizPresignTransfer).filter_by(id='transfer').one()
        assert stale.pay_status == 'dfj'

        presign_crud.submit_receipt_finance(
            finance_db,
            transfer_id='transfer',
            tenant_no='tenant-a',
            user_id='finance-user',
        )
        presign_crud.confirm_receipt(
            finance_db,
            transfer_id='transfer',
            tenant_no='tenant-a',
            user_id='finance-user',
        )
        calls = {
            'update': lambda: update_presign_transfer(
                market_db,
                transfer_id='transfer',
                body=MarketPresignTransferBody(remark='stale-write'),
                ctx=_ctx('specialist'),
                user_info=_user(),
            ),
            'delete': lambda: delete_presign_transfer(
                market_db,
                transfer_id='transfer',
                ctx=_ctx('specialist'),
                user_info=_user(),
            ),
            'submit': lambda: submit_presign_transfer(
                market_db,
                transfer_id='transfer',
                ctx=_ctx('specialist'),
                user_info=_user(),
            ),
        }

        with pytest.raises(ValueError, match='状态|待缴费|提交'):
            calls[action]()

        finance_db.expire_all()
        current = finance_db.query(BizPresignTransfer).filter_by(id='transfer').one()
        assert current.pay_status == 'yjf'
        assert current.del_flag == 0
        assert current.remark == 'original'
    finally:
        market_db.close()
        finance_db.close()


@pytest.mark.parametrize('action', ['update', 'delete'])
def test_stale_order_write_does_not_mutate_later_conversion(tmp_path, action):
    from app.schemas.market_mobile import MarketPresignOrderBody
    from app.services.market_presign import (
        convert_presign_order,
        delete_presign_order,
        update_presign_order,
    )

    Session = _session_factory(
        f"sqlite:///{tmp_path / f'order-{action}-race.db'}"
    )
    market_db = Session()
    convert_db = Session()
    try:
        _add_master_data(market_db)
        order = _order('own', sales_user_id='u1')
        order.protocol_no = f'P-RACE-{action}'
        market_db.add(order)
        market_db.commit()
        stale = market_db.query(BizPresignOrder).filter_by(id='own').one()
        assert stale.order_status == 'active'

        convert_presign_order(
            convert_db,
            order_id='own',
            ctx=_ctx('specialist'),
            user_info=_user(),
        )
        calls = {
            'update': lambda: update_presign_order(
                market_db,
                order_id='own',
                body=MarketPresignOrderBody(remark='stale-write'),
                ctx=_ctx('specialist'),
                user_info=_user(),
            ),
            'delete': lambda: delete_presign_order(
                market_db,
                order_id='own',
                ctx=_ctx('specialist'),
                user_info=_user(),
            ),
        }

        with pytest.raises(ValueError, match='不可|已转'):
            calls[action]()

        convert_db.expire_all()
        current = convert_db.query(BizPresignOrder).filter_by(id='own').one()
        assert current.order_status == 'converted'
        assert current.converted_protocol_id
        assert current.del_flag == 0
        assert current.remark is None
    finally:
        market_db.close()
        convert_db.close()


def test_conversion_unique_index_migration_is_mysql57_idempotent():
    from pathlib import Path

    sql = (
        Path(__file__).parents[2]
        / "docs/sql/migrations/2026-10-10-presign-convert-unique.sql"
    ).read_text(encoding="utf-8")
    normalized = " ".join(sql.lower().replace("`", "").split())

    assert "information_schema.statistics" in normalized
    assert "create procedure ensure_presign_convert_unique" in normalized
    assert "call ensure_presign_convert_unique()" in normalized
    assert "group by tenant_no, source_presign_order_id having count(*) > 1" in normalized
    assert "signal sqlstate '45000'" in normalized
    assert "duplicate" in normalized
    assert "non_unique" in normalized
    assert "group_concat(column_name order by seq_in_index" in normalized
    assert "tenant_no,source_presign_order_id" in normalized
    assert "wrong index" in normalized
    assert "unique index uk_collection_protocol_presign_source" in normalized


def test_market_presign_routes_are_fixed_and_expose_scoped_transfer_actions():
    from app.api.market_mobile import router

    actual = {
        (method, route.path)
        for route in router.routes
        if "/presign-orders" in route.path
        for method in route.methods
    }
    expected = set()
    for role in ("specialist", "manager"):
        base = f"/mini/market/{role}/presign-orders"
        expected.update(
            {
                ("GET", base),
                ("POST", base),
                ("GET", f"{base}/form-options"),
                ("GET", f"{base}/{{order_id}}"),
                ("PUT", f"{base}/{{order_id}}"),
                ("DELETE", f"{base}/{{order_id}}"),
                ("POST", f"{base}/{{order_id}}/convert"),
                ("POST", f"{base}/{{order_id}}/transfers"),
            }
        )

    assert actual == expected

    transfer_actual = {
        (method, route.path)
        for route in router.routes
        if "/presign-transfers" in route.path
        for method in route.methods
    }
    transfer_expected = set()
    for role in ("specialist", "manager"):
        base = f"/mini/market/{role}/presign-transfers/{{transfer_id}}"
        transfer_expected.update(
            {
                ("PUT", base),
                ("DELETE", base),
                ("POST", f"{base}/submit"),
                ("POST", f"{base}/withdraw"),
            }
        )
    assert transfer_actual == transfer_expected


def test_mobile_presign_money_dto_preserves_decimal_strings_for_presenter(
    monkeypatch,
):
    import json
    import subprocess
    from decimal import Decimal
    from pathlib import Path

    from app.crud import presign_order as presign_crud
    from app.services import market_presign

    db = _session()
    normal = _order('money-normal', sales_user_id='u1')
    normal.contract_amount = Decimal('1234.50')
    empty = _order('money-empty', sales_user_id='u1')
    empty.contract_amount = None
    paid = _transfer(
        'money-paid', order_id=normal.id, pay_status='yjf'
    )
    paid.transfer_amount = Decimal('200.25')
    draft = _transfer(
        'money-draft', order_id=normal.id, pay_status='dfj'
    )
    draft.transfer_amount = Decimal('12.30')
    db.add_all([normal, empty, paid, draft])
    db.commit()

    items, total = market_presign.list_presign_orders(
        db,
        ctx=_ctx('specialist'),
        user_info=_user(),
        page=1,
        page_size=10,
    )
    by_id = {item['id']: item for item in items}
    assert total == 2
    assert by_id['money-normal']['contractAmount'] == '1234.50'
    assert by_id['money-normal']['confirmedDepositAmount'] == '200.25'
    assert by_id['money-empty']['contractAmount'] is None
    assert by_id['money-empty']['confirmedDepositAmount'] == '0.00'

    normal_detail = market_presign.get_presign_order(
        db,
        order_id=normal.id,
        ctx=_ctx('specialist'),
        user_info=_user(),
    )
    transfer_amounts = {
        transfer['id']: transfer['transferAmount']
        for transfer in normal_detail['transfers']
    }
    assert transfer_amounts == {
        'money-paid': '200.25',
        'money-draft': '12.30',
    }

    maximum = Decimal('9999999999999999.99')
    maximum_order = _order('money-maximum', sales_user_id='u1')
    maximum_order.contract_amount = maximum
    monkeypatch.setattr(
        market_presign,
        'assert_presign_access',
        lambda *_args, **_kwargs: maximum_order,
    )
    monkeypatch.setattr(
        presign_crud,
        'confirmed_deposit_amount',
        lambda *_args, **_kwargs: maximum,
    )
    maximum_detail = market_presign.get_presign_order(
        db,
        order_id=maximum_order.id,
        ctx=_ctx('specialist'),
        user_info=_user(),
    )
    assert maximum_detail['contractAmount'] == str(maximum)
    assert maximum_detail['confirmedDepositAmount'] == str(maximum)
    maximum_transfer = _transfer(
        'money-maximum-transfer', order_id=maximum_order.id
    )
    maximum_transfer.transfer_amount = maximum
    maximum_transfer_dto = market_presign._mobile_transfer_to_dict(
        db, maximum_transfer, 'tenant-a'
    )
    assert maximum_transfer_dto['transferAmount'] == str(maximum)

    serialized = json.dumps(
        {
            'maximum': maximum_detail,
            'maximumTransfer': maximum_transfer_dto,
            'normal': normal_detail,
            'empty': by_id['money-empty'],
        },
        ensure_ascii=False,
    )
    presenter_url = (
        Path(__file__).resolve().parents[2]
        / 'mobile-oa-h5/src/market/presign/presign-presenter.js'
    ).as_uri()
    script = """
import fs from 'node:fs'
const { formatPresignMoney } = await import(process.argv[1])
const dto = JSON.parse(fs.readFileSync(0, 'utf8'))
const values = [
  dto.maximum.contractAmount,
  dto.maximum.confirmedDepositAmount,
  dto.maximumTransfer.transferAmount,
  dto.normal.contractAmount,
  dto.normal.transfers.find((item) => item.id === 'money-paid').transferAmount,
  dto.empty.contractAmount,
  dto.empty.confirmedDepositAmount
]
process.stdout.write(JSON.stringify(values.map(formatPresignMoney)))
"""
    formatted = subprocess.run(
        ['node', '--input-type=module', '-e', script, presenter_url],
        input=serialized,
        text=True,
        capture_output=True,
        check=True,
    )
    assert json.loads(formatted.stdout) == [
        '¥9,999,999,999,999,999.99',
        '¥9,999,999,999,999,999.99',
        '¥9,999,999,999,999,999.99',
        '¥1,234.50',
        '¥200.25',
        '-',
        '¥0.00',
    ]


def test_mobile_presign_convert_response_preserves_decimal_for_presenter(
    monkeypatch,
):
    import json
    import subprocess
    from decimal import Decimal
    from pathlib import Path

    from app.crud import presign_order as presign_crud
    from app.services import market_presign

    db = _session()
    _add_master_data(db)
    db.commit()
    maximum = Decimal('9999999999999999.99')
    order = _order('convert-money', sales_user_id='u1')
    order.protocol_no = 'P-CONVERT-MONEY'
    order.contract_amount = maximum

    monkeypatch.setattr(
        market_presign,
        '_scoped_order_for_update',
        lambda *_args, **_kwargs: order,
    )
    monkeypatch.setattr(
        presign_crud,
        '_get_order_or_raise',
        lambda *_args, **_kwargs: order,
    )
    monkeypatch.setattr(
        presign_crud,
        'confirmed_deposit_amount',
        lambda *_args, **_kwargs: maximum,
    )
    # SQLite binds Numeric through float. Keep the exact Decimal on the real ORM
    # instance while exercising the complete core conversion and mobile response.
    monkeypatch.setattr(db, 'commit', db.flush)
    monkeypatch.setattr(db, 'refresh', lambda *_args, **_kwargs: None)

    response = market_presign.convert_presign_order(
        db,
        order_id=order.id,
        ctx=_ctx('specialist'),
        user_info=_user(),
    )
    nested = response['presignOrder']
    assert response['protocolId']
    assert order.order_status == 'converted'
    assert nested['contractAmount'] == str(maximum)
    assert nested['confirmedDepositAmount'] == str(maximum)

    presenter_url = (
        Path(__file__).resolve().parents[2]
        / 'mobile-oa-h5/src/market/presign/presign-presenter.js'
    ).as_uri()
    script = """
import fs from 'node:fs'
const { formatPresignMoney } = await import(process.argv[1])
const dto = JSON.parse(fs.readFileSync(0, 'utf8'))
process.stdout.write(JSON.stringify([
  formatPresignMoney(dto.contractAmount),
  formatPresignMoney(dto.confirmedDepositAmount)
]))
"""
    formatted = subprocess.run(
        ['node', '--input-type=module', '-e', script, presenter_url],
        input=json.dumps(nested),
        text=True,
        capture_output=True,
        check=True,
    )
    assert json.loads(formatted.stdout) == [
        '¥9,999,999,999,999,999.99',
        '¥9,999,999,999,999,999.99',
    ]