zkxw/backend/tests/test_market_payment_api.py

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

import pytest
from fastapi import HTTPException
from fastapi.encoders import jsonable_encoder
from pydantic import ValidationError
from sqlalchemy import create_engine, event
from sqlalchemy.orm import sessionmaker
from starlette.responses import JSONResponse

from app.models.base import Base
from app.models.models import (
    BizAttachment,
    BizCellSampleType,
    BizCollectionProtocol,
    BizCollectionRegionStaff,
    BizProtocolCollectionInfo,
    BizProtocolPartyCustomer,
    BizProtocolPaymentSchedule,
    BizProtocolPaymentTransfer,
    BizProtocolSample,
    BizProtocolSamplePayment,
    SysPost,
    SysPostRole,
    SysRole,
    SysRoleDataScopeDetail,
    SysUser,
    SysUserPost,
    SysUserRole,
)


def _session():
    engine = create_engine('sqlite:///:memory:')
    Base.metadata.create_all(
        engine,
        tables=[
            BizCollectionProtocol.__table__,
            BizProtocolCollectionInfo.__table__,
            BizProtocolPartyCustomer.__table__,
            BizProtocolSamplePayment.__table__,
            BizProtocolPaymentSchedule.__table__,
            BizProtocolPaymentTransfer.__table__,
            BizProtocolSample.__table__,
            BizCellSampleType.__table__,
            BizAttachment.__table__,
            BizCollectionRegionStaff.__table__,
            SysUser.__table__,
            SysRole.__table__,
            SysRoleDataScopeDetail.__table__,
            SysUserRole.__table__,
            SysPost.__table__,
            SysPostRole.__table__,
            SysUserPost.__table__,
        ],
    )
    return sessionmaker(bind=engine)()


def _ctx(scope_kind='specialist', 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',
        'region_id': 'dept-from-auth',
    }


def _payment_tree(
    db,
    key,
    *,
    sales='u1',
    manager='manager-a',
    region='region-a',
    tenant='tenant-a',
    protocol_status='sxz',
    payment_model_version=2,
    due_date=None,
    due_amount=100,
    schedule_status='open',
    deleted=0,
    transfer_statuses=(),
    transfer_dept='',
    register_date='20261001',
    storage_years=20,
    sample_price=80,
    sample_original_price=100,
):
    protocol = BizCollectionProtocol(
        id=f'p-{key}',
        protocol_no=f'P-{key}',
        protocol_status=protocol_status,
        payment_model_version=payment_model_version,
        register_date=register_date,
        tenant_no=tenant,
        create_dept_id='dept-a',
        create_time=datetime.now(),
        del_flag=deleted,
    )
    collection = BizProtocolCollectionInfo(
        id=f'ci-{key}',
        collection_protocol_id=protocol.id,
        region_id=region,
        sales_user_id=sales,
        region_manager_user_id=manager,
        tenant_no=tenant,
        create_dept_id='dept-a',
        del_flag=deleted,
    )
    party = BizProtocolPartyCustomer(
        id=f'party-{key}',
        collection_protocol_id=protocol.id,
        mother_name=f'客户-{key}',
        tenant_no=tenant,
        create_dept_id='dept-a',
        del_flag=deleted,
    )
    payment = BizProtocolSamplePayment(
        id=f'sp-{key}',
        collection_protocol_id=protocol.id,
        sample_code=f'S-{key}',
        pay_mode='fq',
        contract_amount=due_amount,
        tenant_no=tenant,
        create_dept_id='dept-a',
        create_time=datetime.now(),
        del_flag=deleted,
    )
    schedule = BizProtocolPaymentSchedule(
        id=f'sch-{key}',
        sample_payment_id=payment.id,
        collection_protocol_id=protocol.id,
        sample_code=payment.sample_code,
        schedule_no=1,
        due_amount=due_amount,
        due_date=due_date or date.today().strftime('%Y%m%d'),
        schedule_status=schedule_status,
        tenant_no=tenant,
        create_dept_id='dept-a',
        del_flag=deleted,
    )
    sample_type = BizCellSampleType(
        id=f'type-{key}',
        type_code=f'TYPE-{key}',
        type_name=f'样本类型-{key}',
        tenant_no=tenant,
        create_dept_id='dept-a',
        del_flag=deleted,
    )
    sample = BizProtocolSample(
        id=f'sample-{key}',
        collection_protocol_id=protocol.id,
        sample_type_id=sample_type.id,
        sample_code=payment.sample_code,
        storage_years=storage_years,
        price=sample_price,
        original_price=sample_original_price,
        tenant_no=tenant,
        create_dept_id='dept-a',
        del_flag=deleted,
    )
    db.add_all([protocol, collection, party, payment, schedule, sample_type, sample])
    for index, status in enumerate(transfer_statuses):
        db.add(
            BizProtocolPaymentTransfer(
                id=f't-{key}-{index}',
                schedule_id=schedule.id,
                sample_payment_id=payment.id,
                collection_protocol_id=protocol.id,
                sample_code=payment.sample_code,
                transfer_amount=20,
                transfer_time=datetime.now(),
                transfer_type='bank',
                pay_status=status,
                tenant_no=tenant,
                create_dept_id=transfer_dept,
                del_flag=deleted,
            )
        )
    db.commit()
    return protocol, payment, schedule


def test_chase_is_overdue_union_next_30_days_with_parent_scope():
    from app.services.market_payment import list_payment_chase

    db = _session()
    today = date(2026, 10, 11)
    _payment_tree(db, 'overdue', due_date=(today - timedelta(days=1)).strftime('%Y%m%d'))
    _payment_tree(db, 'today', due_date=today.strftime('%Y%m%d'))
    _payment_tree(db, 'day30', due_date=(today + timedelta(days=30)).strftime('%Y%m%d'))
    _payment_tree(db, 'day31', due_date=(today + timedelta(days=31)).strftime('%Y%m%d'))
    _payment_tree(db, 'other', sales='u2', due_date=today.strftime('%Y%m%d'))
    _payment_tree(db, 'void', protocol_status='yzf', due_date=today.strftime('%Y%m%d'))
    _payment_tree(db, 'bad-date', due_date='not-a-date')

    items, total = list_payment_chase(
        db, ctx=_ctx(), user_info=_user(), today=today
    )

    assert total == 3
    assert {item['scheduleId'] for item in items} == {
        'sch-overdue',
        'sch-today',
        'sch-day30',
    }


def test_payment_chase_uses_historical_nickname_without_user_no_fallback():
    from app.services.market_payment import list_payment_chase

    db = _session()
    db.add_all(
        [
            SysUser(id='u1', user_no='deleted-sales-no', nick_name='历史删除专员', status='0', tenant_no='tenant-a', create_dept_id='', del_flag=1),
            SysUser(id='blank-sales', user_no='raw-sales-no', nick_name='', status='0', tenant_no='tenant-a', create_dept_id='', del_flag=1),
        ]
    )
    _payment_tree(db, 'historical-sales', sales='u1')
    _payment_tree(db, 'blank-sales', sales='blank-sales')

    historical_items, historical_total = list_payment_chase(
        db, ctx=_ctx(user_id='u1'), user_info=_user('u1')
    )
    blank_items, blank_total = list_payment_chase(
        db,
        ctx=_ctx(user_id='blank-sales'),
        user_info=_user('blank-sales'),
    )

    assert historical_total == 1
    assert historical_items[0]['salesUserName'] == '历史删除专员'
    assert blank_total == 1
    assert blank_items[0]['salesUserName'] == ''


def test_chase_counts_only_yjf_and_excludes_fully_confirmed():
    from app.services.market_payment import list_payment_chase

    db = _session()
    _payment_tree(
        db,
        'part-paid',
        due_amount=100,
        transfer_statuses=('dfj', 'jfqrz', 'yjf'),
    )
    _payment_tree(
        db,
        'fully-paid',
        due_amount=60,
        transfer_statuses=('yjf', 'yjf', 'yjf'),
    )

    items, total = list_payment_chase(db, ctx=_ctx(), user_info=_user())

    assert total == 1
    assert items[0]['scheduleId'] == 'sch-part-paid'
    assert items[0]['paidAmount'] == '20.00'
    assert items[0]['arrearsAmount'] == '80.00'


def test_chase_filters_counts_orders_and_pages_in_sql_before_materialization():
    from app.services.market_payment import list_payment_chase

    db = _session()
    today = date(2026, 10, 11)
    for index in range(35):
        _payment_tree(
            db,
            f'history-{index:02d}',
            due_date=(today - timedelta(days=index + 1)).strftime('%Y%m%d'),
            transfer_statuses=('dfj',),
        )
    _payment_tree(db, 'invalid-day', due_date='20260230')
    statements = []

    def capture(_conn, _cursor, statement, _parameters, _context, _many):
        statements.append(statement)

    event.listen(db.get_bind(), 'before_cursor_execute', capture)
    try:
        items, total = list_payment_chase(
            db,
            ctx=_ctx(),
            user_info=_user(),
            today=today,
            page=3,
            page_size=5,
        )
    finally:
        event.remove(db.get_bind(), 'before_cursor_execute', capture)

    assert total == 35
    assert len(items) == 5
    assert [item['scheduleId'] for item in items] == [
        'sch-history-24',
        'sch-history-23',
        'sch-history-22',
        'sch-history-21',
        'sch-history-20',
    ]
    chase_selects = [sql for sql in statements if 'biz_protocol_payment_schedule' in sql]
    assert len(chase_selects) == 2
    assert ' LIMIT ' in chase_selects[-1].upper()
    assert 'SUM(CASE' in chase_selects[-1].upper()
    assert not any(
        'FROM biz_protocol_payment_transfer' in sql and 'SUM(CASE' not in sql.upper()
        for sql in statements
    )


def test_payment_list_is_v2_only_and_uses_parent_scope_not_transfer_department():
    from app.services.market_payment import list_payments

    db = _session()
    _payment_tree(db, 'v2', transfer_statuses=('dfj',), transfer_dept='')
    _payment_tree(db, 'v1', payment_model_version=1)
    _payment_tree(db, 'other', sales='u2')
    _payment_tree(db, 'cross', tenant='tenant-b')
    _payment_tree(db, 'deleted', deleted=1)

    items, total = list_payments(db, ctx=_ctx(), user_info=_user())

    assert total == 1
    assert items[0]['id'] == 'sp-v2'
    assert items[0]['paymentModelVersion'] == 2


def test_manager_scope_is_fixed_by_route_context_and_client_filter_cannot_expand():
    from app.services.market_payment import list_payment_chase

    db = _session()
    _payment_tree(db, 'manager', manager='manager-a', sales='u2')
    _payment_tree(db, 'outside', manager='manager-b', sales='u1')

    items, total = list_payment_chase(
        db,
        ctx=_ctx('manager', 'manager-a'),
        user_info=_user('manager-a'),
        sales_user_id='u1',
    )

    assert total == 0
    assert items == []


def test_payment_detail_rejects_out_of_scope_sample():
    from app.services.market_payment import get_payment_detail

    db = _session()
    _payment_tree(db, 'other', sales='u2')

    with pytest.raises(HTTPException) as exc:
        get_payment_detail(
            db, payment_id='sp-other', ctx=_ctx(), user_info=_user()
        )

    assert exc.value.status_code == 404


def test_payment_detail_identifies_legacy_v1_without_reading_wrong_tables():
    from app.services.market_payment import get_payment_detail

    db = _session()
    _payment_tree(db, 'legacy', payment_model_version=1)

    with pytest.raises(HTTPException) as exc:
        get_payment_detail(
            db, payment_id='sp-legacy', ctx=_ctx(), user_info=_user()
        )

    assert exc.value.status_code == 409
    assert '旧版' in str(exc.value.detail)


def test_market_can_save_submit_and_withdraw_transfer_atomically():
    from app.services.market_payment import (
        create_transfer,
        submit_transfer,
        update_transfer,
        withdraw_transfer,
    )

    db = _session()
    _, _, schedule = _payment_tree(db, 'own')
    db.add(
        BizAttachment(
            id='att-pay',
            business_type='market:specialist:payment:staging',
            original_name='a.jpg',
            file_ext='jpg',
            file_url='a.jpg',
            file_size=10,
            file_type='image/jpeg',
            tenant_no='tenant-a',
            create_by='u1',
            create_dept_id='',
            del_flag=0,
        )
    )
    db.commit()
    created = create_transfer(
        db,
        schedule_id=schedule.id,
        fields={
            'transferAmount': 50,
            'transferTime': '2026-10-11 10:00:00',
            'transferType': 'bank',
            'transferAttachments': 'a.jpg',
        },
        ctx=_ctx(),
        user_info=_user(),
    )
    assert created['payStatus'] == 'dfj'
    row = db.get(BizProtocolPaymentTransfer, created['id'])
    assert row.create_dept_id == 'dept-from-auth'
    attachment = db.get(BizAttachment, 'att-pay')
    assert (attachment.business_type, attachment.business_id) == (
        'market:specialist:payment:bound',
        created['id'],
    )

    updated = update_transfer(
        db,
        transfer_id=created['id'],
        fields={'remark': '补充说明'},
        ctx=_ctx(),
        user_info=_user(),
    )
    assert updated['remark'] == '补充说明'
    assert submit_transfer(
        db, transfer_id=created['id'], ctx=_ctx(), user_info=_user()
    )['payStatus'] == 'jfqrz'
    assert withdraw_transfer(
        db, transfer_id=created['id'], ctx=_ctx(), user_info=_user()
    )['payStatus'] == 'dfj'


def test_transfer_create_rejects_external_attachment_reference():
    from app.services.market_payment import create_transfer

    db = _session()
    _, _, schedule = _payment_tree(db, 'attachment-reject')
    with pytest.raises(ValueError, match='附件'):
        create_transfer(
            db,
            schedule_id=schedule.id,
            fields={
                'transferAmount': 10,
                'transferType': 'bank',
                'transferAttachments': 'https://evil.test/a.jpg',
            },
            ctx=_ctx(),
            user_info=_user(),
        )


def test_transfer_create_rollback_restores_staging_attachment_on_business_failure(
    monkeypatch,
):
    from app.services.market_payment import create_transfer

    db = _session()
    _, _, schedule = _payment_tree(db, 'attachment-rollback')
    attachment = BizAttachment(
        id='att-rollback',
        business_type='market:specialist:payment:staging',
        original_name='rollback.jpg',
        file_ext='jpg',
        file_url='rollback.jpg',
        file_size=10,
        file_type='image/jpeg',
        tenant_no='tenant-a',
        create_by='u1',
        create_dept_id='',
        del_flag=0,
    )
    db.add(attachment)
    db.commit()
    monkeypatch.setattr(
        db, 'commit', lambda: (_ for _ in ()).throw(RuntimeError('db down'))
    )

    with pytest.raises(RuntimeError, match='db down'):
        create_transfer(
            db,
            schedule_id=schedule.id,
            fields={
                'transferAmount': 10,
                'transferType': 'bank',
                'transferAttachments': 'att-rollback',
            },
            ctx=_ctx(),
            user_info=_user(),
        )

    db.expire_all()
    restored = db.get(BizAttachment, 'att-rollback')
    assert (restored.business_type, restored.business_id) == (
        'market:specialist:payment:staging',
        None,
    )
    assert db.query(BizProtocolPaymentTransfer).count() == 0


def test_out_of_scope_write_is_checked_before_any_mutation():
    from app.services.market_payment import update_transfer

    db = _session()
    _payment_tree(db, 'other', sales='u2', transfer_statuses=('dfj',))
    before = db.get(BizProtocolPaymentTransfer, 't-other-0').remark

    with pytest.raises(HTTPException):
        update_transfer(
            db,
            transfer_id='t-other-0',
            fields={'remark': 'must-not-write'},
            ctx=_ctx(),
            user_info=_user(),
        )

    db.expire_all()
    assert db.get(BizProtocolPaymentTransfer, 't-other-0').remark == before


@pytest.mark.parametrize('protocol_status', ['yzf', 'sdfq', 'zky'])
def test_terminal_protocol_rejects_direct_transfer_create_without_mutation(protocol_status):
    from app.services.market_payment import create_transfer

    db = _session()
    _, _, schedule = _payment_tree(
        db, f'terminal-create-{protocol_status}', protocol_status=protocol_status
    )

    with pytest.raises(HTTPException):
        create_transfer(
            db,
            schedule_id=schedule.id,
            fields={'transferAmount': 10, 'transferType': 'bank'},
            ctx=_ctx(),
            user_info=_user(),
        )

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


@pytest.mark.parametrize('schedule_status', ['cleared', 'void'])
def test_non_open_schedule_rejects_direct_transfer_create_without_mutation(schedule_status):
    from app.services.market_payment import create_transfer

    db = _session()
    _, _, schedule = _payment_tree(
        db, f'closed-create-{schedule_status}', schedule_status=schedule_status
    )

    with pytest.raises(HTTPException):
        create_transfer(
            db,
            schedule_id=schedule.id,
            fields={'transferAmount': 10, 'transferType': 'bank'},
            ctx=_ctx(),
            user_info=_user(),
        )

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


@pytest.mark.parametrize(
    ('protocol_status', 'schedule_status', 'pay_status', 'operation'),
    [
        ('yzf', 'open', 'dfj', 'update'),
        ('sdfq', 'open', 'dfj', 'submit'),
        ('zky', 'open', 'jfqrz', 'withdraw'),
        ('sxz', 'cleared', 'dfj', 'update'),
        ('sxz', 'void', 'dfj', 'submit'),
    ],
)
def test_ineligible_parent_rejects_direct_transfer_mutations_without_mutation(
    protocol_status, schedule_status, pay_status, operation
):
    from app.services.market_payment import submit_transfer, update_transfer, withdraw_transfer

    db = _session()
    key = f'ineligible-{protocol_status}-{schedule_status}-{operation}'
    _payment_tree(
        db,
        key,
        protocol_status=protocol_status,
        schedule_status=schedule_status,
        transfer_statuses=(pay_status,),
    )
    transfer_id = f't-{key}-0'
    before = db.get(BizProtocolPaymentTransfer, transfer_id)
    before_values = (before.pay_status, before.remark, before.update_by, before.update_time)

    with pytest.raises(HTTPException):
        if operation == 'update':
            update_transfer(
                db,
                transfer_id=transfer_id,
                fields={'remark': 'must-not-write'},
                ctx=_ctx(),
                user_info=_user(),
            )
        elif operation == 'submit':
            submit_transfer(db, transfer_id=transfer_id, ctx=_ctx(), user_info=_user())
        else:
            withdraw_transfer(db, transfer_id=transfer_id, ctx=_ctx(), user_info=_user())

    db.expire_all()
    after = db.get(BizProtocolPaymentTransfer, transfer_id)
    assert (after.pay_status, after.remark, after.update_by, after.update_time) == before_values


def test_chase_excludes_all_terminal_protocols_and_non_open_schedules():
    from app.services.market_payment import list_payment_chase

    db = _session()
    _payment_tree(db, 'eligible-chase')
    for status in ('yzf', 'sdfq', 'zky'):
        _payment_tree(db, f'terminal-chase-{status}', protocol_status=status)
    _payment_tree(db, 'cleared-chase', schedule_status='cleared')
    _payment_tree(db, 'void-chase', schedule_status='void')

    items, total = list_payment_chase(db, ctx=_ctx(), user_info=_user())

    assert total == 1
    assert [item['scheduleId'] for item in items] == ['sch-eligible-chase']


def test_yjf_is_immutable_and_double_submit_fails_closed():
    from app.services.market_payment import submit_transfer, update_transfer, withdraw_transfer

    db = _session()
    _payment_tree(db, 'draft', transfer_statuses=('dfj',))
    submit_transfer(db, transfer_id='t-draft-0', ctx=_ctx(), user_info=_user())
    with pytest.raises(ValueError, match='状态'):
        submit_transfer(db, transfer_id='t-draft-0', ctx=_ctx(), user_info=_user())

    _payment_tree(db, 'confirmed', transfer_statuses=('yjf',))
    with pytest.raises(ValueError):
        update_transfer(
            db,
            transfer_id='t-confirmed-0',
            fields={'remark': 'no'},
            ctx=_ctx(),
            user_info=_user(),
        )
    with pytest.raises(ValueError):
        withdraw_transfer(
            db,
            transfer_id='t-confirmed-0',
            ctx=_ctx(),
            user_info=_user(),
        )


def test_transfer_dto_forbids_authorization_status_and_finance_fields():
    from app.schemas.market_mobile import MarketPaymentTransferBody

    forbidden = (
        {'tenantNo': 'tenant-b'},
        {'scope': 'manager'},
        {'payStatus': 'yjf'},
        {'financeRemark': 'confirmed'},
        {'scheduleStatus': 'void'},
    )
    for extra in forbidden:
        with pytest.raises(ValidationError):
            MarketPaymentTransferBody.model_validate(
                {'transferAmount': 10, 'transferType': 'bank', **extra}
            )


@pytest.mark.parametrize(
    'amount',
    [float('inf'), float('-inf'), float('nan'), '0.001', '10000000000000000.00'],
)
def test_transfer_amount_rejects_non_finite_over_precision_and_overflow(amount):
    from app.schemas.market_mobile import MarketPaymentTransferBody

    with pytest.raises(ValidationError):
        MarketPaymentTransferBody.model_validate({'transferAmount': amount})


@pytest.mark.parametrize('amount', ['0.01', '9999999999999999.99'])
def test_transfer_amount_accepts_decimal_18_2_boundaries_exactly(amount):
    from app.schemas.market_mobile import MarketPaymentTransferBody

    body = MarketPaymentTransferBody.model_validate({'transferAmount': amount})

    assert body.transfer_amount == Decimal(amount)


def test_task5_money_is_fixed_two_decimal_string_in_actual_json_response():
    from app.core.response import success_response
    from app.schemas.market_mobile import MarketPaymentTransferCreateBody
    from app.services.market_payment import _payment_dict, create_transfer

    db = _session()
    _, _, schedule = _payment_tree(
        db, 'money-json', due_amount=Decimal('9999999999999999.99')
    )
    body = MarketPaymentTransferCreateBody.model_validate(
        {
            'scheduleId': schedule.id,
            'transferAmount': '9999999999999999.99',
            'transferType': 'bank',
        }
    )
    payload = body.model_dump(by_alias=True, exclude_unset=True)
    schedule_id = payload.pop('scheduleId')
    transfer = create_transfer(
        db,
        schedule_id=schedule_id,
        fields=payload,
        ctx=_ctx(),
        user_info=_user(),
    )
    max_money = Decimal('9999999999999999.99')
    payment = _payment_dict(
        SimpleNamespace(
            id='sp-json',
            sample_code='S-json',
            pay_mode='fq',
            contract_amount=max_money,
            remark=None,
        ),
        SimpleNamespace(id='p-json', protocol_no='P-json', protocol_status='sxz'),
        None,
        [
            SimpleNamespace(
                id='sch-json',
                schedule_no=1,
                due_amount=max_money,
                due_date='20261011',
                schedule_status='open',
            )
        ],
        {},
    )

    encoded = jsonable_encoder(
        success_response({'transfer': transfer, 'payment': payment})
    )
    response = JSONResponse(encoded)
    decoded = response.body.decode('utf-8')

    assert transfer['transferAmount'] == '9999999999999999.99'
    assert payment['contractAmount'] == '9999999999999999.99'
    assert payment['schedules'][0]['dueAmount'] == '9999999999999999.99'
    assert '"9999999999999999.99"' in decoded


def test_task5_normal_money_is_always_serialized_with_two_decimals():
    from app.services.market_payment import list_payment_chase

    db = _session()
    _payment_tree(db, 'normal-money', due_amount=Decimal('100.00'))

    items, _ = list_payment_chase(db, ctx=_ctx(), user_info=_user())

    assert items[0]['dueAmount'] == '100.00'
    assert items[0]['paidAmount'] == '0.00'
    assert items[0]['arrearsAmount'] == '100.00'


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

    paths = {route.path for route in router.routes}
    for scope in ('specialist', 'manager'):
        prefix = f'/mini/market/{scope}'
        assert f'{prefix}/payment-chase' in paths
        assert f'{prefix}/payment-chases' not in paths
        assert f'{prefix}/payments' in paths
        assert f'{prefix}/payment-transfers' in paths
        assert f'{prefix}/payment-transfers/{{transfer_id}}/submit' in paths
        assert f'{prefix}/payment-transfers/{{transfer_id}}/withdraw' in paths
        assert f'{prefix}/payment-transfers/{{transfer_id}}/confirm' not in paths
        assert f'{prefix}/payment-transfers/{{transfer_id}}/reject' not in paths
        assert f'{prefix}/payment-transfers/{{transfer_id}}/refund' not in paths
        assert f'{prefix}/payment-transfers/{{transfer_id}}/invoice' not in paths


def test_list_queries_batch_children_without_n_plus_one():
    from app.services.market_payment import list_payments

    db = _session()
    for index in range(8):
        _payment_tree(db, f'row-{index}', transfer_statuses=('dfj', 'yjf'))
    statements = []

    def capture(_conn, _cursor, statement, _parameters, _context, _many):
        statements.append(statement)

    event.listen(db.get_bind(), 'before_cursor_execute', capture)
    try:
        items, total = list_payments(
            db, ctx=_ctx(), user_info=_user(), page=1, page_size=5
        )
    finally:
        event.remove(db.get_bind(), 'before_cursor_execute', capture)

    assert total == 8
    assert len(items) == 5
    assert len(statements) <= 5


def test_payment_list_contains_card_summary_totals():
    from app.services.market_payment import list_payments

    db = _session()
    _payment_tree(db, 'summary', due_amount=Decimal('100.00'), transfer_statuses=('yjf',))

    items, total = list_payments(db, ctx=_ctx(), user_info=_user())

    assert total == 1
    assert items[0]['paidAmount'] == '20.00'
    assert items[0]['remainAmount'] == '80.00'
    assert items[0]['activeScheduleCount'] == 1
    assert items[0]['clearedScheduleCount'] == 0
    assert items[0]['paymentProgress'] == '部分缴费'
    assert items[0]['signDate'] == '2026-10-01'
    assert items[0]['storageYears'] == 20
    assert items[0]['sampleTypeName'] == '样本类型-summary'
    assert items[0]['discountAmount'] == '20.00'
    assert items[0]['receivableAmount'] == '100.00'
    assert items[0]['dueRemainAmount'] == '80.00'
    assert len(items[0]['periods']) == 6
    assert items[0]['periods'][0]['scheduleNo'] == 1


def test_payment_sample_meta_uses_tenant_and_sample_code_after_protocol_change():
    from app.services.market_payment import list_payments

    db = _session()
    _, payment, _ = _payment_tree(db, 'protocol-changed')
    sample = (
        db.query(BizProtocolSample)
        .filter(BizProtocolSample.sample_code == payment.sample_code)
        .one()
    )
    sample.collection_protocol_id = 'p-previous-protocol'
    db.commit()

    items, total = list_payments(db, ctx=_ctx(), user_info=_user())

    assert total == 1
    assert items[0]['storageYears'] == 20
    assert items[0]['sampleTypeName'] == '样本类型-protocol-changed'
    assert items[0]['discountAmount'] == '20.00'