zkxw/backend/tests/test_market_protocol_api.py

import json
from datetime import datetime, timedelta
from decimal import Decimal
from unittest.mock import patch

import pytest
from fastapi import HTTPException
from pydantic import ValidationError
from sqlalchemy import create_engine, event
from sqlalchemy.orm import Query, sessionmaker

from app.models.base import Base
from app.models.models import (
    BizAttachment,
    BizCellSampleType,
    BizCollectionHospital,
    BizCollectionProtocol,
    BizCollectionRegion,
    BizCollectionRegionStaff,
    BizCustomer,
    BizItemUnit,
    BizProtocolCollectionInfo,
    BizProtocolPartyCustomer,
    BizProtocolPayment,
    BizProtocolPaymentDetail,
    BizProtocolPaymentSchedule,
    BizProtocolPaymentTransfer,
    BizProtocolSamplePayment,
    BizProtocolSample,
    BizProtocolSampleTransport,
    BizPresignOrder,
    BizPresignTransfer,
    BizProtocolAudit,
    BizProtocolStorageCert,
    BizProtocolFullmoonTodo,
    BizSampleCode,
    BizSampleFreight,
    BizSampleFreightItem,
    BizSampleReception,
    SysPost,
    SysPostRole,
    SysRole,
    SysRoleDataScopeDetail,
    SysDictData,
    SysUser,
    SysUserPost,
    SysUserRole,
)
from app.models.research_donation import (
    BizProtocolDonationConvert,
    BizResearchDonation,
    BizResearchDonationCollection,
    BizResearchDonationParty,
    BizResearchDonationSample,
    BizResearchDonationSampleTransport,
)


def _session():
    engine = create_engine('sqlite:///:memory:')
    Base.metadata.create_all(engine)
    return sessionmaker(bind=engine)()


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',
        'region_id': 'dept-from-auth',
    }


def _cell(db, row_id, *, sales='u1', manager='manager-a', region='region-a', created=None):
    row = BizCollectionProtocol(
        id=row_id,
        protocol_no=f'CELL-{row_id}',
        protocol_status='dsx',
        tenant_no='tenant-a',
        create_dept_id='dept-a',
        create_time=created or datetime.now(),
        del_flag=0,
    )
    collection = BizProtocolCollectionInfo(
        id=f'ci-{row_id}',
        collection_protocol_id=row_id,
        region_id=region,
        sales_user_id=sales,
        region_manager_user_id=manager,
        tenant_no='tenant-a',
        create_dept_id='dept-a',
        del_flag=0,
    )
    db.add_all([row, collection])
    return row


def _donation(db, row_id, *, sales='u1', manager='manager-a', region='region-a', created=None):
    row = BizResearchDonation(
        id=row_id,
        donation_no=f'DON-{row_id}',
        donation_status='dsx',
        tenant_no='tenant-a',
        create_dept_id='dept-a',
        create_time=created or datetime.now(),
        del_flag=0,
    )
    collection = BizResearchDonationCollection(
        id=f'dci-{row_id}',
        research_donation_id=row_id,
        region_id=region,
        sales_user_id=sales,
        region_manager_user_id=manager,
        tenant_no='tenant-a',
        create_dept_id='dept-a',
        del_flag=0,
    )
    db.add_all([row, collection])
    return row


def _masters(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
            ),
            BizCollectionHospital(
                id='hospital-a', region_id='region-a', hospital_name='A院', tenant_no='tenant-a', create_dept_id='', del_flag=0
            ),
            BizCellSampleType(
                id='type-a', type_code='A', type_name='样本A', tenant_no='tenant-a', create_dept_id='', del_flag=0
            ),
            BizSampleCode(
                id='code-a', sample_code='S-001', tenant_no='tenant-a', create_dept_id='', del_flag=0
            ),
            BizSampleCode(
                id='code-b', sample_code='S-002', tenant_no='tenant-a', create_dept_id='', del_flag=0
            ),
            BizSampleCode(
                id='code-c', sample_code='S-003', tenant_no='tenant-a', create_dept_id='', del_flag=0
            ),
            BizSampleCode(
                id='code-cross', sample_code='S-CROSS', tenant_no='tenant-b', create_dept_id='', del_flag=0
            ),
            BizSampleCode(
                id='code-deleted', sample_code='S-DELETED', tenant_no='tenant-a', create_dept_id='', del_flag=1
            ),
            BizItemUnit(
                id='unit-a', unit_code='ml', unit_name='毫升', 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='', del_flag=0
            ),
            SysUser(
                id='manager-a', user_no='manager-a', nick_name='经理', status='1', tenant_no='tenant-a', create_dept_id='', del_flag=0
            ),
            SysUser(
                id='u2', user_no='u2', nick_name='其他专员', status='1', tenant_no='tenant-a', create_dept_id='', del_flag=0
            ),
            BizCollectionRegionStaff(
                id='staff-u1-region-a', region_id='region-a', user_id='u1', is_leader=1,
                tenant_no='tenant-a', create_dept_id='', del_flag=0,
            ),
            BizCollectionRegionStaff(
                id='staff-manager-a-region-a', region_id='region-a', user_id='manager-a', is_leader=0,
                tenant_no='tenant-a', create_dept_id='', del_flag=0,
            ),
            BizAttachment(
                id='att-protocol-a',
                business_type='market:specialist:protocol:staging',
                original_name='a.pdf',
                file_ext='pdf',
                file_url='protocol/a.pdf',
                file_size=10,
                file_type='application/pdf',
                tenant_no='tenant-a',
                create_by='u1',
                create_dept_id='',
                del_flag=0,
            ),
        ]
    )
    db.commit()


def _body(source='cell_storage', **overrides):
    from app.schemas.market_mobile import MarketProtocolBody

    payload = {
        'source': source,
        'protocolNo': 'P-001',
        'registerDate': '20261010',
        'attachments': {'protocolAttachments': 'protocol/a.pdf'},
        'customer': {'motherName': '张三', 'motherPhone': '13800000000'},
        'collection': {
            'regionId': 'region-a',
            'hospitalId': 'hospital-a',
            'salesUserId': 'someone-else',
            'regionManagerUserId': 'manager-a',
        },
        'samples': [
            {
                'sampleTypeId': 'type-a',
                'sampleCodeId': 'code-a',
                'sampleCode': 'S-001',
                'unitId': 'unit-a',
                'collectAmount': 10,
            }
        ],
        'remark': '移动端登记',
    }
    payload.update(overrides)
    return MarketProtocolBody.model_validate(payload)


def test_protocol_create_rejects_unknown_attachment_reference():
    from app.services.market_protocol import create_protocol

    db = _session()
    _masters(db)
    body = _body(
        attachments={'protocolAttachments': 'https://evil.test/a.pdf'}
    )

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


def test_terminal_protocol_allows_web_parity_edit_without_status_change():
    from app.services.market_protocol import update_protocol

    db = _session()
    _masters(db)
    row = _cell(db, 'terminal')
    row.protocol_status = 'yzf'
    db.commit()

    update_protocol(
        db,
        source='cell_storage',
        protocol_id=row.id,
        body=_body(protocolNo='CELL-terminal'),
        ctx=_ctx('specialist'),
        user_info=_user(),
    )

    db.expire_all()
    saved = db.get(BizCollectionProtocol, row.id)
    assert saved.protocol_status == 'yzf'
    assert saved.protocol_no == 'CELL-terminal'


def test_protocol_list_combines_cell_and_donation_without_scope_leak():
    from app.services.market_protocol import list_protocols

    db = _session()
    now = datetime.now()
    _cell(db, 'cell-own', created=now - timedelta(minutes=2))
    _cell(db, 'cell-other', sales='u2', created=now)
    _donation(db, 'don-own', created=now - timedelta(minutes=1))
    _donation(db, 'don-other', sales='u2', created=now + timedelta(minutes=1))
    db.commit()

    items, total = list_protocols(db, ctx=_ctx('specialist'), user_info=_user())

    assert total == 2
    assert [(item['source'], item['id']) for item in items] == [
        ('research_donation', 'don-own'),
        ('cell_storage', 'cell-own'),
    ]


def test_protocol_list_large_page_uses_bounded_source_queries_and_global_order():
    from app.services.market_protocol import list_protocols

    db = _session()
    base = datetime(2026, 10, 10, 12, 0, 0)
    expected = []
    for index in range(24):
        created = base + timedelta(seconds=index)
        cell_id = f'cell-{index:02d}'
        donation_id = f'don-{index:02d}'
        _cell(db, cell_id, created=created)
        _donation(db, donation_id, created=created + timedelta(microseconds=1))
        expected.extend([(created, cell_id), (created + timedelta(microseconds=1), donation_id)])
    db.commit()
    statements = []

    def capture(_conn, _cursor, statement, _parameters, _context, _many):
        if 'FROM biz_collection_protocol ' in statement or 'FROM biz_research_donation ' in statement:
            statements.append(statement)

    event.listen(db.get_bind(), 'before_cursor_execute', capture)
    try:
        items, total = list_protocols(
            db,
            ctx=_ctx('specialist'),
            user_info=_user(),
            page=4,
            page_size=7,
        )
    finally:
        event.remove(db.get_bind(), 'before_cursor_execute', capture)

    ordered_ids = [item[1] for item in sorted(expected, key=lambda item: (item[0], item[1]), reverse=True)]
    assert total == 48
    assert [item['id'] for item in items] == ordered_ids[21:28]
    page_selects = [sql for sql in statements if 'count(' not in sql.lower()]
    assert len(page_selects) == 2
    assert all(' LIMIT ' in sql.upper() for sql in page_selects)


def test_protocol_list_stable_tie_breaker_prevents_page_overlap():
    from app.services.market_protocol import list_protocols

    db = _session()
    same_time = datetime(2026, 10, 10, 12, 0, 0)
    for row_id in ('a1', 'a3', 'a5', 'a7'):
        _cell(db, row_id, created=same_time)
    for row_id in ('a2', 'a4', 'a6', 'a8'):
        _donation(db, row_id, created=same_time)
    db.commit()

    page_one, total = list_protocols(
        db, ctx=_ctx('specialist'), user_info=_user(), page=1, page_size=3
    )
    page_two, _ = list_protocols(
        db, ctx=_ctx('specialist'), user_info=_user(), page=2, page_size=3
    )

    assert total == 8
    assert [item['id'] for item in page_one] == ['a8', 'a7', 'a6']
    assert [item['id'] for item in page_two] == ['a5', 'a4', 'a3']
    assert {item['id'] for item in page_one}.isdisjoint(
        {item['id'] for item in page_two}
    )


def test_protocol_detail_rechecks_source_specific_idor():
    from app.services.market_protocol import get_protocol

    db = _session()
    _cell(db, 'cell-other', sales='u2')
    _donation(db, 'don-other', sales='u2')
    db.commit()

    with pytest.raises(HTTPException) as cell_error:
        get_protocol(
            db,
            source='cell_storage',
            protocol_id='cell-other',
            ctx=_ctx('specialist'),
            user_info=_user(),
        )
    with pytest.raises(HTTPException) as donation_error:
        get_protocol(
            db,
            source='research_donation',
            protocol_id='don-other',
            ctx=_ctx('specialist'),
            user_info=_user(),
        )
    assert cell_error.value.status_code == 404
    assert donation_error.value.status_code == 404


def test_protocol_detail_can_resolve_source_from_scoped_record():
    from app.services.market_protocol import get_protocol

    db = _session()
    _donation(db, 'don-own')
    db.commit()

    result = get_protocol(
        db,
        source=None,
        protocol_id='don-own',
        ctx=_ctx('specialist'),
        user_info=_user(),
    )

    assert result['source'] == 'research_donation'
    assert result['basic']['id'] == 'don-own'


def test_protocol_detail_resolves_web_display_names_for_mobile_fields():
    from app.services.market_protocol import get_protocol

    db = _session()
    _masters(db)
    row = _cell(db, 'named')
    row.collector_user_id = 'u1'
    collection = db.query(BizProtocolCollectionInfo).filter_by(
        collection_protocol_id=row.id
    ).one()
    collection.hospital_id = 'hospital-a'
    db.add(
        BizProtocolSample(
            id='sample-named',
            collection_protocol_id=row.id,
            sample_type_id='type-a',
            sample_code='S-NAMED',
            unit_id='unit-a',
            tenant_no='tenant-a',
            create_dept_id='',
            del_flag=0,
        )
    )
    db.commit()

    result = get_protocol(
        db,
        source='cell_storage',
        protocol_id=row.id,
        ctx=_ctx('specialist'),
        user_info=_user(),
    )

    assert result['basic']['collectorUserName'] == '当前用户'
    assert result['collection']['regionName'] == 'A区'
    assert result['collection']['hospitalName'] == 'A院'
    assert result['collection']['salesUserName'] == '当前用户'
    assert result['collection']['regionManagerUserName'] == '经理'
    assert result['samples'][0]['sampleTypeName'] == '样本A'
    assert result['samples'][0]['unitName'] == '毫升'


def test_protocol_detail_resolves_soft_deleted_master_names_with_tenant_scope():
    from app.services.market_protocol import get_protocol

    db = _session()
    _masters(db)
    row = _cell(db, 'historical-master')
    collection = db.query(BizProtocolCollectionInfo).filter_by(
        collection_protocol_id=row.id
    ).one()
    collection.hospital_id = 'hospital-a'
    db.get(BizCollectionRegion, 'region-a').del_flag = 1
    db.get(BizCollectionHospital, 'hospital-a').del_flag = 1
    db.get(BizCellSampleType, 'type-a').del_flag = 1
    db.get(BizItemUnit, 'unit-a').del_flag = 1
    db.add_all(
        [
            BizCellSampleType(id='type-global-history', type_code='GLOBAL', type_name='历史全局样本', tenant_no='', create_dept_id='', del_flag=1),
            BizCellSampleType(id='type-cross-history', type_code='CROSS', type_name='跨租户样本名称', tenant_no='tenant-b', create_dept_id='', del_flag=1),
            BizItemUnit(id='unit-cross-history', unit_code='cross', unit_name='跨租户单位名称', tenant_no='tenant-b', create_dept_id='', del_flag=1),
            BizProtocolSample(id='sample-history-local', collection_protocol_id=row.id, sample_type_id='type-a', sample_code='S-HISTORY-LOCAL', unit_id='unit-a', tenant_no='tenant-a', create_dept_id='', del_flag=0),
            BizProtocolSample(id='sample-history-global', collection_protocol_id=row.id, sample_type_id='type-global-history', sample_code='S-HISTORY-GLOBAL', unit_id='', tenant_no='tenant-a', create_dept_id='', del_flag=0),
            BizProtocolSample(id='sample-history-cross', collection_protocol_id=row.id, sample_type_id='type-cross-history', sample_code='S-HISTORY-CROSS', unit_id='unit-cross-history', tenant_no='tenant-a', create_dept_id='', del_flag=0),
        ]
    )
    db.commit()

    result = get_protocol(
        db,
        source='cell_storage',
        protocol_id=row.id,
        ctx=_ctx('specialist'),
        user_info=_user(),
    )
    samples = {item['sampleCode']: item for item in result['samples']}

    assert result['collection']['regionName'] == 'A区'
    assert result['collection']['hospitalName'] == 'A院'
    assert samples['S-HISTORY-LOCAL']['sampleTypeName'] == '样本A'
    assert samples['S-HISTORY-LOCAL']['unitName'] == '毫升'
    assert samples['S-HISTORY-GLOBAL']['sampleTypeName'] == '历史全局样本'
    assert samples['S-HISTORY-CROSS']['sampleTypeName'] == ''
    assert samples['S-HISTORY-CROSS']['unitName'] == ''


def test_protocol_names_use_exact_historical_nicknames_without_raw_fallbacks():
    from app.services.market_protocol import get_protocol, list_protocols

    db = _session()
    _masters(db)
    row = _cell(db, 'historical-names')
    row.collector_user_id = 'blank-name-user'
    collection = db.query(BizProtocolCollectionInfo).filter_by(
        collection_protocol_id=row.id
    ).one()
    collection.region_manager_user_id = 'manager-a'
    current_user = db.get(SysUser, 'u1')
    current_user.nick_name = '历史删除专员'
    current_user.del_flag = 1
    manager = db.get(SysUser, 'manager-a')
    manager.nick_name = '历史停用经理'
    manager.status = '0'
    db.add(
        SysUser(
            id='blank-name-user',
            user_no='raw-collector-no',
            nick_name='',
            status='0',
            tenant_no='tenant-a',
            create_dept_id='',
            del_flag=1,
        )
    )
    sample_type = db.get(BizCellSampleType, 'type-a')
    sample_type.type_name = ''
    unit = db.get(BizItemUnit, 'unit-a')
    unit.unit_name = ''
    db.add(
        BizProtocolSample(
            id='sample-historical-names',
            collection_protocol_id=row.id,
            sample_type_id='type-a',
            sample_code='S-HISTORY',
            unit_id='unit-a',
            tenant_no='tenant-a',
            create_dept_id='',
            del_flag=0,
        )
    )
    db.commit()

    items, total = list_protocols(
        db, ctx=_ctx('specialist'), user_info=_user()
    )
    detail = get_protocol(
        db,
        source='cell_storage',
        protocol_id=row.id,
        ctx=_ctx('specialist'),
        user_info=_user(),
    )

    assert total == 1
    assert items[0]['salesUserName'] == '历史删除专员'
    assert items[0]['collectorUserName'] == ''
    assert items[0]['managerName'] == '历史停用经理'
    assert items[0]['sampleTypeName'] == ''
    assert detail['basic']['collectorUserName'] == ''
    assert detail['collection']['salesUserName'] == '历史删除专员'
    assert detail['collection']['regionManagerUserName'] == '历史停用经理'
    assert detail['samples'][0]['sampleTypeName'] == ''
    assert detail['samples'][0]['unitName'] == ''


def test_specialist_cell_create_forces_current_collection_and_collector_users():
    from app.services.market_protocol import create_protocol

    db = _session()
    _masters(db)

    result = create_protocol(
        db, body=_body(), ctx=_ctx('specialist'), user_info=_user()
    )

    main = db.get(BizCollectionProtocol, result['basic']['id'])
    collection = db.query(BizProtocolCollectionInfo).filter_by(collection_protocol_id=main.id, del_flag=0).one()
    party = db.query(BizProtocolPartyCustomer).filter_by(collection_protocol_id=main.id, del_flag=0).one()
    sample = db.query(BizProtocolSample).filter_by(collection_protocol_id=main.id, del_flag=0).one()
    assert main.protocol_status == 'sxz'
    assert main.collector_user_id == 'u1'
    assert main.create_dept_id == 'dept-from-auth'
    assert collection.sales_user_id == 'u1'
    assert collection.create_dept_id == 'dept-from-auth'
    assert party.collector_user_id == 'u1'
    assert party.customer_id is None
    assert db.query(BizCustomer).count() == 0
    assert sample.sample_code == 'S-001'


def test_specialist_donation_create_forces_current_users_and_has_no_payment_section():
    from app.services.market_protocol import create_protocol

    db = _session()
    _masters(db)

    result = create_protocol(
        db,
        body=_body('research_donation', donationNo='D-001', protocolNo=None),
        ctx=_ctx('specialist'),
        user_info=_user(),
    )

    donation = db.get(BizResearchDonation, result['basic']['id'])
    collection = db.query(BizResearchDonationCollection).filter_by(research_donation_id=donation.id, del_flag=0).one()
    party = db.query(BizResearchDonationParty).filter_by(research_donation_id=donation.id, del_flag=0).one()
    sample = db.query(BizResearchDonationSample).filter_by(research_donation_id=donation.id, del_flag=0).one()
    assert donation.donation_status == 'sxz'
    assert donation.collector_user_id == 'u1'
    assert donation.create_dept_id == 'dept-from-auth'
    assert collection.sales_user_id == 'u1'
    assert collection.create_dept_id == 'dept-from-auth'
    assert party.collector_user_id == 'u1'
    assert party.create_dept_id == 'dept-from-auth'
    assert sample.create_dept_id == 'dept-from-auth'
    assert party.customer_id is None
    assert db.query(BizCustomer).count() == 0
    assert 'payment' not in result


def test_manager_update_cannot_move_document_out_of_scope_and_rolls_back():
    from app.services.market_protocol import update_protocol

    db = _session()
    _masters(db)
    _cell(db, 'cell-manager', manager='u1')
    db.commit()

    with pytest.raises(ValueError, match='范围'):
        update_protocol(
            db,
            source='cell_storage',
            protocol_id='cell-manager',
            body=_body(collection={'regionId': 'region-b', 'regionManagerUserId': 'u1'}),
            ctx=_ctx('manager'),
            user_info=_user(),
        )

    db.expire_all()
    collection = db.query(BizProtocolCollectionInfo).filter_by(collection_protocol_id='cell-manager', del_flag=0).one()
    assert collection.region_id == 'region-a'
    assert collection.region_manager_user_id == 'u1'


@pytest.mark.parametrize('source', ['cell_storage', 'research_donation'])
def test_manager_create_cannot_self_assign_an_unmanaged_region(source):
    from app.services.market_protocol import create_protocol

    db = _session()
    _masters(db)
    body = _body(
        source,
        **(
            {'donationNo': 'D-UNMANAGED', 'protocolNo': None}
            if source == 'research_donation'
            else {}
        ),
        collection={
            'regionId': 'region-b',
            'regionManagerUserId': 'u1',
            'salesUserId': 'u2',
        },
        attachments=None,
    )

    with pytest.raises(ValueError, match='范围'):
        create_protocol(
            db,
            body=body,
            ctx=_ctx('manager'),
            user_info=_user(),
        )


@pytest.mark.parametrize('source', ['cell_storage', 'research_donation'])
@pytest.mark.parametrize('operation', ['update', 'delete'])
def test_protocol_mutations_lock_the_scoped_parent_row(
    monkeypatch, source, operation
):
    from app.schemas.market_mobile import MarketProtocolBody
    from app.services.market_protocol import delete_protocol, update_protocol

    db = _session()
    _masters(db)
    row_id = f'{source}-{operation}-lock'
    if source == 'cell_storage':
        _cell(db, row_id)
    else:
        _donation(db, row_id)
    db.commit()
    lock_calls = []
    original = Query.with_for_update

    def record_lock(query, *args, **kwargs):
        lock_calls.append(True)
        return original(query, *args, **kwargs)

    monkeypatch.setattr(Query, 'with_for_update', record_lock)
    if operation == 'update':
        update_protocol(
            db,
            source=source,
            protocol_id=row_id,
            body=MarketProtocolBody.model_validate(
                {'source': source, 'remark': '锁后更新'}
            ),
            ctx=_ctx('specialist'),
            user_info=_user(),
        )
    else:
        delete_protocol(
            db,
            source=source,
            protocol_id=row_id,
            ctx=_ctx('specialist'),
            user_info=_user(),
        )

    assert lock_calls


def test_v1_cell_protocol_mobile_update_keeps_version_and_saves_legacy_payment():
    from app.schemas.market_mobile import MarketProtocolBody
    from app.services.market_protocol import update_protocol

    db = _session()
    _masters(db)
    row = _cell(db, 'legacy-v1')
    row.payment_model_version = 1
    db.commit()

    result = update_protocol(
        db,
        source='cell_storage',
        protocol_id=row.id,
        body=MarketProtocolBody.model_validate(
            {
                'source': 'cell_storage',
                'remark': '移动端可编辑历史协议',
                'payment': {
                    'payMode': 'dj',
                    'contractAmount': '999999999999.99',
                    'lumpSumDeadline': '20261231',
                },
                'paymentDetails': [
                    {
                        'installmentNo': 1,
                        'installmentDueDate': '20261231',
                        'dueAmount': '999999999999.99',
                        'transferType': 'wx',
                        'transferAmount': '1.01',
                        'transferTime': '2026-10-14 10:00:00',
                        'transferAttachments': 'protocol/a.pdf',
                        'remark': '移动草稿',
                    }
                ],
            }
        ),
        ctx=_ctx('specialist'),
        user_info=_user(),
    )

    db.expire_all()
    saved = db.get(BizCollectionProtocol, row.id)
    assert saved.payment_model_version == 1
    assert saved.remark == '移动端可编辑历史协议'
    assert result['paymentModelVersion'] == 1
    assert result['payment']['contractAmount'] == '999999999999.99'
    assert result['paymentDetails'][0]['transferAmount'] == '1.01'


def test_v1_market_update_accepts_complete_generated_installment_amounts():
    from app.schemas.market_mobile import MarketProtocolBody
    from app.services.market_protocol import update_protocol

    db = _session()
    _masters(db)
    row = _cell(db, 'legacy-v1-generated-installments')
    row.payment_model_version = 1
    db.commit()

    update_protocol(
        db,
        source='cell_storage',
        protocol_id=row.id,
        body=MarketProtocolBody.model_validate(
            {
                'source': 'cell_storage',
                'payment': {
                    'payMode': 'fq',
                    'contractAmount': '100.01',
                    'installmentCount': 2,
                    'installmentIntervalMonths': '1',
                    'installmentStartDate': '20260131',
                    'installmentAdvanceDays': 0,
                    'installmentRatios': ['50.00', '50.01'],
                },
                'paymentDetails': [
                    {
                        'installmentNo': 1,
                        'installmentDueDate': '20260131',
                        'dueAmount': '50.00',
                    },
                    {
                        'installmentNo': 2,
                        'installmentDueDate': '20260228',
                        'dueAmount': '50.01',
                    },
                ],
            }
        ),
        ctx=_ctx('specialist'),
        user_info=_user(),
    )

    db.expire_all()
    payment = db.query(BizProtocolPayment).filter_by(
        collection_protocol_id=row.id, del_flag=0
    ).one()
    assert payment.installment_ratios == '["50.00", "50.01"]'
    details = db.query(BizProtocolPaymentDetail).filter_by(
        payment_id=payment.id, del_flag=0
    ).order_by(BizProtocolPaymentDetail.installment_no).all()
    assert [detail.due_amount for detail in details] == [
        Decimal('50.00'), Decimal('50.01')
    ]
    assert [detail.installment_due_date for detail in details] == [
        '20260131', '20260228'
    ]


def test_protocol_body_legacy_payment_is_typed_and_rejects_sensitive_detail_fields():
    from app.schemas.market_mobile import MarketProtocolBody
    from fastapi.encoders import jsonable_encoder

    body = MarketProtocolBody.model_validate(
        {
            'source': 'cell_storage',
            'payment': {
                'payMode': 'fq',
                'contractAmount': '1000.01',
                'installmentCount': 2,
                'installmentIntervalMonths': '12.50',
                'installmentStartDate': '20261014',
                'installmentAdvanceDays': 30,
                'installmentRatios': ['400.00', '600.01'],
            },
            'paymentDetails': [
                {
                    'id': 'detail-a',
                    'installmentNo': 1,
                    'dueAmount': '400.00',
                    'transferType': 'bank',
                    'transferAmount': '1.01',
                    'transferTime': '2026-10-14 10:00:00',
                    'transferAttachments': 'proof.pdf',
                    'remark': '市场备注',
                }
            ],
        }
    )

    assert body.payment.contract_amount == Decimal('1000.01')
    assert body.payment.installment_ratios == [Decimal('400.00'), Decimal('600.01')]
    assert body.payment_details[0].transfer_amount == Decimal('1.01')
    attachment_alias = MarketProtocolBody.model_validate(
        {
            'source': 'cell_storage',
            'paymentDetails': [{'attachments': 'protocol/proof.pdf'}],
        }
    )
    assert attachment_alias.payment_details[0].transfer_attachments == 'protocol/proof.pdf'
    max_amount = MarketProtocolBody.model_validate(
        {
            'source': 'cell_storage',
            'payment': {
                'payMode': 'dj',
                'contractAmount': '9999999999999999.99',
            },
        }
    )
    assert max_amount.payment.contract_amount == Decimal('9999999999999999.99')

    with pytest.raises(ValidationError):
        MarketProtocolBody.model_validate(
            {
                'source': 'cell_storage',
                'payment': {'payMode': 'evil', 'contractAmount': '1.00'},
            }
        )

    from app.crud import collection_protocol as cell_crud

    serialized_payment = cell_crud._payment_to_dict(
        BizProtocolPayment(
            id='max-payment', collection_protocol_id='max-protocol',
            contract_amount=Decimal('9999999999999999.99'),
            installment_ratios='["4999999999999999.99", "5000000000000000.00"]',
            tenant_no='tenant-a', create_dept_id='', del_flag=0,
        )
    )
    serialized_detail = cell_crud._detail_to_dict(
        BizProtocolPaymentDetail(
            id='max-detail', payment_id='max-payment',
            due_amount=Decimal('4999999999999999.99'),
            transfer_amount=Decimal('0.01'), tenant_no='tenant-a',
            create_dept_id='', del_flag=0,
        )
    )
    assert serialized_payment['contractAmount'] == '9999999999999999.99'
    assert jsonable_encoder(serialized_payment)['installmentRatios'] == [
        '4999999999999999.99', '5000000000000000.00'
    ]
    assert serialized_detail['dueAmount'] == '4999999999999999.99'
    assert serialized_detail['transferAmount'] == '0.01'

    for forbidden in (
        {'payStatus': 'dfj'},
        {'financeRemark': '越权'},
        {'refundAmount': '1.00'},
        {'financeConfirmAttachments': 'finance.pdf'},
    ):
        with pytest.raises(ValidationError):
            MarketProtocolBody.model_validate(
                {
                    'source': 'cell_storage',
                    'paymentDetails': [{'dueAmount': '1.00', **forbidden}],
                }
            )


@pytest.mark.parametrize(
    ('section', 'field', 'value'),
    [
        ('payment', 'lumpSumDeadline', '20261399'),
        ('payment', 'installmentStartDate', '20230229'),
        ('paymentDetails', 'installmentDueDate', '20260230'),
    ],
)
def test_protocol_body_legacy_payment_rejects_invalid_calendar_dates(
    section, field, value
):
    from app.schemas.market_mobile import MarketProtocolBody

    payload = {'source': 'cell_storage'}
    if section == 'paymentDetails':
        payload[section] = [{field: value}]
    else:
        payload[section] = {'payMode': 'dj', field: value}
    with pytest.raises(ValidationError, match='YYYYMMDD'):
        MarketProtocolBody.model_validate(payload)


def test_protocol_body_legacy_payment_accepts_month_end_and_leap_dates():
    from app.schemas.market_mobile import MarketProtocolBody

    body = MarketProtocolBody.model_validate(
        {
            'source': 'cell_storage',
            'payment': {
                'payMode': 'fq',
                'lumpSumDeadline': '20260228',
                'installmentStartDate': '20240229',
            },
            'paymentDetails': [{'installmentDueDate': '20260228'}],
        }
    )
    assert body.payment.installment_start_date == '20240229'
    assert body.payment_details[0].installment_due_date == '20260228'


def test_protocol_payment_models_are_mutually_exclusive_by_stored_version():
    from app.schemas.market_mobile import MarketProtocolBody
    from app.services.market_protocol import update_protocol

    db = _session()
    _masters(db)
    v1 = _cell(db, 'legacy-mixed')
    v1.payment_model_version = 1
    v2 = _cell(db, 'sample-mixed')
    v2.payment_model_version = 2
    db.commit()

    with pytest.raises(ValueError, match='v1.*samplePayments|旧版.*样本缴费'):
        update_protocol(
            db,
            source='cell_storage',
            protocol_id=v1.id,
            body=MarketProtocolBody.model_validate(
                {
                    'source': 'cell_storage',
                    'samplePayments': [
                        {'sampleCode': 'S-001', 'payMode': 'dj'}
                    ],
                }
            ),
            ctx=_ctx('specialist'),
            user_info=_user(),
        )

    with pytest.raises(ValueError, match='v2.*payment|新版.*整单缴费'):
        update_protocol(
            db,
            source='cell_storage',
            protocol_id=v2.id,
            body=MarketProtocolBody.model_validate(
                {
                    'source': 'cell_storage',
                    'payment': {'payMode': 'dj', 'contractAmount': '1.00'},
                    'paymentDetails': [],
                }
            ),
            ctx=_ctx('specialist'),
            user_info=_user(),
        )

    donation = _donation(db, 'donation-payment-mixed')
    db.commit()
    with pytest.raises(ValueError, match='科研捐赠.*缴费'):
        update_protocol(
            db,
            source='research_donation',
            protocol_id=donation.id,
            body=MarketProtocolBody.model_validate(
                {
                    'source': 'research_donation',
                    'payment': {'payMode': 'dj', 'contractAmount': '1.00'},
                }
            ),
            ctx=_ctx('specialist'),
            user_info=_user(),
        )


def test_v1_market_update_preserves_locked_rows_and_mutates_only_drafts():
    from app.schemas.market_mobile import MarketProtocolBody
    from app.services.market_protocol import update_protocol

    db = _session()
    _masters(db)
    row = _cell(db, 'legacy-status-guard')
    row.payment_model_version = 1
    payment = BizProtocolPayment(
        id='legacy-payment', collection_protocol_id=row.id, pay_mode='dj',
        contract_amount=Decimal('1000.00'), tenant_no='tenant-a',
        create_dept_id='', del_flag=0,
    )
    details = [
        BizProtocolPaymentDetail(
            id=f'detail-{status_name}', payment_id=payment.id,
            pay_status=status, due_amount=Decimal('100.00'),
            transfer_amount=Decimal('10.00'), remark='原值',
            tenant_no='tenant-a', create_dept_id='', del_flag=0,
        )
        for status_name, status in (
            ('paid', 'yjf'), ('confirming', 'jfqrz'),
            ('empty', ''), ('spaces', '   '), ('draft', 'dfj'),
            ('padded-draft', ' dfj '), ('delete', 'dfj'),
        )
    ]
    db.add_all([payment, *details])
    db.commit()

    update_protocol(
        db,
        source='cell_storage',
        protocol_id=row.id,
        body=MarketProtocolBody.model_validate(
            {
                'source': 'cell_storage',
                'payment': {'payMode': 'dj', 'contractAmount': '1000.00'},
                'paymentDetails': [
                    {'id': 'detail-paid', 'dueAmount': '1.00', 'remark': '篡改'},
                    {'id': 'detail-confirming', 'dueAmount': '1.00', 'remark': '篡改'},
                    {'id': 'detail-empty', 'dueAmount': '1.00', 'remark': '篡改'},
                    {'id': 'detail-spaces', 'dueAmount': '1.00', 'remark': '篡改'},
                    {'id': 'detail-draft', 'dueAmount': '250.00', 'remark': '已更新'},
                    {'dueAmount': '350.00', 'remark': '新增草稿'},
                ],
            }
        ),
        ctx=_ctx('specialist'),
        user_info=_user(),
    )

    db.expire_all()
    for locked_id in (
        'detail-paid', 'detail-confirming', 'detail-empty',
        'detail-spaces', 'detail-padded-draft',
    ):
        locked = db.get(BizProtocolPaymentDetail, locked_id)
        assert locked.due_amount == Decimal('100.00')
        assert locked.remark == '原值'
        assert locked.del_flag == 0
    assert db.get(BizProtocolPaymentDetail, 'detail-draft').due_amount == Decimal('250.00')
    assert db.get(BizProtocolPaymentDetail, 'detail-delete').del_flag == 1
    created = db.query(BizProtocolPaymentDetail).filter_by(
        payment_id=payment.id, remark='新增草稿', del_flag=0
    ).one()
    assert created.pay_status == 'dfj'


def test_v1_market_update_rolls_back_protocol_and_payment_on_detail_failure():
    from app.schemas.market_mobile import MarketProtocolBody
    from app.services.market_protocol import update_protocol

    db = _session()
    _masters(db)
    row = _cell(db, 'legacy-atomic')
    row.payment_model_version = 1
    payment = BizProtocolPayment(
        id='legacy-atomic-payment', collection_protocol_id=row.id,
        pay_mode='fq', contract_amount=Decimal('2000.00'),
        installment_count=2, installment_ratios='[1000, 1000]',
        remark='原方案', tenant_no='tenant-a', create_dept_id='', del_flag=0,
    )
    locked = BizProtocolPaymentDetail(
        id='legacy-atomic-locked', payment_id=payment.id, pay_status='yjf',
        installment_no=2, due_amount=Decimal('1000.00'),
        tenant_no='tenant-a', create_dept_id='', del_flag=0,
    )
    db.add_all([payment, locked])
    db.commit()

    with pytest.raises(ValueError, match='明细条数'):
        update_protocol(
            db,
            source='cell_storage',
            protocol_id=row.id,
            body=MarketProtocolBody.model_validate(
                {
                    'source': 'cell_storage',
                    'remark': '不应提交',
                    'payment': {
                        'payMode': 'fq', 'contractAmount': '700.00',
                        'installmentCount': 1, 'installmentRatios': ['700.00'],
                    },
                    'paymentDetails': [],
                }
            ),
            ctx=_ctx('specialist'),
            user_info=_user(),
        )

    db.expire_all()
    assert db.get(BizCollectionProtocol, row.id).remark is None
    saved_payment = db.get(BizProtocolPayment, payment.id)
    assert saved_payment.contract_amount == Decimal('2000.00')
    assert saved_payment.installment_count == 2
    assert saved_payment.remark == '原方案'
    assert db.get(BizProtocolPaymentDetail, locked.id).del_flag == 0


def test_v1_market_update_rejects_locked_plan_inconsistency_and_rolls_back_everything():
    from app.schemas.market_mobile import MarketProtocolBody
    from app.services.market_protocol import update_protocol

    db = _session()
    _masters(db)
    row = _cell(db, 'legacy-locked-plan')
    row.payment_model_version = 1
    payment = BizProtocolPayment(
        id='legacy-locked-plan-payment', collection_protocol_id=row.id,
        pay_mode='fq', contract_amount=Decimal('2000.00'),
        installment_count=2, installment_interval_months=Decimal('12'),
        installment_start_date='20261001', installment_advance_days=30,
        installment_ratios='[1000, 1000]', tenant_no='tenant-a',
        create_dept_id='', del_flag=0,
    )
    locked = BizProtocolPaymentDetail(
        id='legacy-locked-plan-paid', payment_id=payment.id, pay_status='yjf',
        installment_no=1, due_amount=Decimal('1000.00'),
        tenant_no='tenant-a', create_dept_id='', del_flag=0,
    )
    db.add_all([payment, locked])
    db.commit()

    with pytest.raises(ValueError, match='已缴|缴清|锁定|合同金额|缴费计划'):
        update_protocol(
            db,
            source='cell_storage',
            protocol_id=row.id,
            body=MarketProtocolBody.model_validate(
                {
                    'source': 'cell_storage',
                    'remark': '不应提交',
                    'payment': {'payMode': 'dj', 'contractAmount': '1.00'},
                    'paymentDetails': [
                        {
                            'id': locked.id,
                            'installmentNo': 1,
                            'dueAmount': '1000.00',
                        }
                    ],
                }
            ),
            ctx=_ctx('specialist'),
            user_info=_user(),
        )

    db.expire_all()
    assert db.get(BizCollectionProtocol, row.id).remark is None
    saved_payment = db.get(BizProtocolPayment, payment.id)
    assert saved_payment.pay_mode == 'fq'
    assert saved_payment.contract_amount == Decimal('2000.00')
    assert saved_payment.installment_count == 2
    assert db.get(BizProtocolPaymentDetail, locked.id).del_flag == 0


def test_v1_market_update_blocks_plan_changes_during_finance_confirmation():
    from app.schemas.market_mobile import MarketProtocolBody
    from app.services.market_protocol import update_protocol

    db = _session()
    _masters(db)
    row = _cell(db, 'legacy-confirming-plan')
    row.payment_model_version = 1
    payment = BizProtocolPayment(
        id='legacy-confirming-plan-payment', collection_protocol_id=row.id,
        pay_mode='fq', contract_amount=Decimal('2000.00'),
        installment_count=2, installment_interval_months=Decimal('12'),
        installment_start_date='20261001', installment_advance_days=30,
        installment_ratios='[1000, 1000]', tenant_no='tenant-a',
        create_dept_id='', del_flag=0,
    )
    confirming = BizProtocolPaymentDetail(
        id='legacy-confirming-plan-locked', payment_id=payment.id,
        pay_status='jfqrz', installment_no=1, due_amount=Decimal('1000.00'),
        tenant_no='tenant-a', create_dept_id='', del_flag=0,
    )
    draft = BizProtocolPaymentDetail(
        id='legacy-confirming-plan-draft', payment_id=payment.id,
        pay_status='dfj', installment_no=2, due_amount=Decimal('1000.00'),
        tenant_no='tenant-a', create_dept_id='', del_flag=0,
    )
    db.add_all([payment, confirming, draft])
    db.commit()

    with pytest.raises(ValueError, match='确认中|锁定|缴费计划'):
        update_protocol(
            db,
            source='cell_storage',
            protocol_id=row.id,
            body=MarketProtocolBody.model_validate(
                {
                    'source': 'cell_storage',
                    'remark': '不应提交',
                    'payment': {
                        'payMode': 'fq',
                        'contractAmount': '2000.00',
                        'installmentCount': 2,
                        'installmentIntervalMonths': '24',
                        'installmentStartDate': '20261001',
                        'installmentAdvanceDays': 30,
                        'installmentRatios': ['1000.00', '1000.00'],
                    },
                    'paymentDetails': [
                        {
                            'id': confirming.id,
                            'installmentNo': 1,
                            'dueAmount': '1000.00',
                        },
                        {
                            'id': draft.id,
                            'installmentNo': 2,
                            'dueAmount': '1000.00',
                        },
                    ],
                }
            ),
            ctx=_ctx('specialist'),
            user_info=_user(),
        )

    db.expire_all()
    assert db.get(BizCollectionProtocol, row.id).remark is None
    assert db.get(BizProtocolPayment, payment.id).installment_interval_months == Decimal('12')


def test_v1_market_update_rejects_stale_draft_schedule_after_paid_plan_change():
    from app.schemas.market_mobile import MarketProtocolBody
    from app.services.market_protocol import update_protocol

    db = _session()
    _masters(db)
    row = _cell(db, 'legacy-stale-plan')
    row.payment_model_version = 1
    payment = BizProtocolPayment(
        id='legacy-stale-plan-payment', collection_protocol_id=row.id,
        pay_mode='fq', contract_amount=Decimal('2000.00'),
        installment_count=2, installment_interval_months=Decimal('12'),
        installment_start_date='20261001', installment_advance_days=30,
        installment_ratios='[1000, 1000]', tenant_no='tenant-a',
        create_dept_id='', del_flag=0,
    )
    paid = BizProtocolPaymentDetail(
        id='legacy-stale-plan-paid', payment_id=payment.id, pay_status='yjf',
        installment_no=1, installment_due_date='20261001',
        due_amount=Decimal('1000.00'), tenant_no='tenant-a',
        create_dept_id='', del_flag=0,
    )
    draft = BizProtocolPaymentDetail(
        id='legacy-stale-plan-draft', payment_id=payment.id, pay_status='dfj',
        installment_no=2, installment_due_date='20270901',
        due_amount=Decimal('1000.00'), tenant_no='tenant-a',
        create_dept_id='', del_flag=0,
    )
    db.add_all([payment, paid, draft])
    db.commit()

    with pytest.raises(ValueError, match='缴费计划|应缴时间'):
        update_protocol(
            db,
            source='cell_storage',
            protocol_id=row.id,
            body=MarketProtocolBody.model_validate(
                {
                    'source': 'cell_storage',
                    'remark': '不应提交',
                    'payment': {
                        'payMode': 'fq', 'contractAmount': '2000.00',
                        'installmentCount': 2, 'installmentIntervalMonths': '24',
                        'installmentStartDate': '20261001',
                        'installmentAdvanceDays': 30,
                        'installmentRatios': ['1000.00', '1000.00'],
                    },
                    'paymentDetails': [
                        {
                            'id': paid.id, 'installmentNo': 1,
                            'installmentDueDate': '20261001',
                            'dueAmount': '1000.00',
                        },
                        {
                            'id': draft.id, 'installmentNo': 2,
                            'installmentDueDate': '20270901',
                            'dueAmount': '1000.00',
                        },
                    ],
                }
            ),
            ctx=_ctx('specialist'),
            user_info=_user(),
        )

    db.expire_all()
    assert db.get(BizCollectionProtocol, row.id).remark is None
    assert db.get(BizProtocolPayment, payment.id).installment_interval_months == Decimal('12')
    assert db.get(BizProtocolPaymentDetail, draft.id).installment_due_date == '20270901'

    result = update_protocol(
        db,
        source='cell_storage',
        protocol_id=row.id,
        body=MarketProtocolBody.model_validate(
            {
                'source': 'cell_storage',
                'remark': '已对齐新计划',
                'payment': {
                    'payMode': 'fq', 'contractAmount': '2000.00',
                    'installmentCount': 2, 'installmentIntervalMonths': '24',
                    'installmentStartDate': '20261001',
                    'installmentAdvanceDays': 30,
                    'installmentRatios': ['1000.00', '1000.00'],
                },
                'paymentDetails': [
                    {
                        'id': paid.id, 'installmentNo': 1,
                        'installmentDueDate': '20261001',
                        'dueAmount': '1000.00',
                    },
                    {
                        'id': draft.id, 'installmentNo': 2,
                        'installmentDueDate': '20280901',
                        'dueAmount': '1000.00',
                    },
                ],
            }
        ),
        ctx=_ctx('specialist'),
        user_info=_user(),
    )
    assert result['payment']['installmentIntervalMonths'] == 24.0
    db.expire_all()
    assert db.get(BizCollectionProtocol, row.id).remark == '已对齐新计划'
    assert db.get(BizProtocolPaymentDetail, draft.id).installment_due_date == '20280901'


def test_v1_market_update_requires_rebuilt_lump_sum_after_paid_installment_switch():
    from app.schemas.market_mobile import MarketProtocolBody
    from app.services.market_protocol import update_protocol

    db = _session()
    _masters(db)
    row = _cell(db, 'legacy-switch-to-lump-sum')
    row.payment_model_version = 1
    payment = BizProtocolPayment(
        id='legacy-switch-to-lump-sum-payment', collection_protocol_id=row.id,
        pay_mode='fq', contract_amount=Decimal('2000.00'),
        installment_count=2, installment_interval_months=Decimal('12'),
        installment_start_date='20261001', installment_advance_days=30,
        installment_ratios='[1000, 1000]', tenant_no='tenant-a',
        create_dept_id='', del_flag=0,
    )
    paid = BizProtocolPaymentDetail(
        id='legacy-switch-to-lump-sum-paid', payment_id=payment.id,
        pay_status='yjf', installment_no=1, installment_due_date='20261001',
        due_amount=Decimal('1000.00'), tenant_no='tenant-a',
        create_dept_id='', del_flag=0,
    )
    old_draft = BizProtocolPaymentDetail(
        id='legacy-switch-to-lump-sum-draft', payment_id=payment.id,
        pay_status='dfj', installment_no=2, installment_due_date='20270901',
        due_amount=Decimal('1000.00'), tenant_no='tenant-a',
        create_dept_id='', del_flag=0,
    )
    db.add_all([payment, paid, old_draft])
    db.commit()

    payment_payload = {
        'payMode': 'dj', 'contractAmount': '2000.00',
        'lumpSumDeadline': '20261231',
    }
    with pytest.raises(ValueError, match='趸交|待缴|期号|应缴时间'):
        update_protocol(
            db,
            source='cell_storage',
            protocol_id=row.id,
            body=MarketProtocolBody.model_validate(
                {
                    'source': 'cell_storage',
                    'remark': '不应提交',
                    'payment': payment_payload,
                    'paymentDetails': [
                        {
                            'id': paid.id, 'installmentNo': 1,
                            'installmentDueDate': '20261001',
                            'dueAmount': '1000.00',
                        },
                        {
                            'id': old_draft.id, 'installmentNo': 2,
                            'installmentDueDate': '20270901',
                            'dueAmount': '1000.00',
                        },
                    ],
                }
            ),
            ctx=_ctx('specialist'),
            user_info=_user(),
        )

    db.expire_all()
    assert db.get(BizCollectionProtocol, row.id).remark is None
    assert db.get(BizProtocolPayment, payment.id).pay_mode == 'fq'
    assert db.get(BizProtocolPaymentDetail, old_draft.id).del_flag == 0

    result = update_protocol(
        db,
        source='cell_storage',
        protocol_id=row.id,
        body=MarketProtocolBody.model_validate(
            {
                'source': 'cell_storage',
                'remark': '转趸交',
                'payment': payment_payload,
                'paymentDetails': [
                    {
                        'id': paid.id, 'installmentNo': 1,
                        'installmentDueDate': '20261001',
                        'dueAmount': '1000.00',
                    },
                    {
                        'installmentNo': None,
                        'installmentDueDate': '20261231',
                        'dueAmount': '1000.00',
                    },
                ],
            }
        ),
        ctx=_ctx('specialist'),
        user_info=_user(),
    )

    editable = [item for item in result['paymentDetails'] if item['payStatus'] == 'dfj']
    assert len(editable) == 1
    assert editable[0]['installmentNo'] is None
    assert editable[0]['installmentDueDate'] == '20261231'
    assert editable[0]['dueAmount'] == '1000.00'
    db.expire_all()
    assert db.get(BizProtocolPaymentDetail, old_draft.id).del_flag == 1


def test_v1_market_update_rejects_details_without_payment_and_keeps_attachment_staging():
    from app.schemas.market_mobile import MarketProtocolBody
    from app.services.market_protocol import update_protocol

    db = _session()
    _masters(db)
    row = _cell(db, 'legacy-details-without-parent')
    row.payment_model_version = 1
    db.commit()

    with pytest.raises(ValueError, match='缴费主表|缴费方案'):
        update_protocol(
            db,
            source='cell_storage',
            protocol_id=row.id,
            body=MarketProtocolBody.model_validate(
                {
                    'source': 'cell_storage',
                    'remark': '不应提交',
                    'paymentDetails': [
                        {
                            'dueAmount': '100.00',
                            'attachments': 'protocol/a.pdf',
                        }
                    ],
                }
            ),
            ctx=_ctx('specialist'),
            user_info=_user(),
        )

    db.expire_all()
    assert db.get(BizCollectionProtocol, row.id).remark is None
    assert db.query(BizProtocolPayment).count() == 0
    assert db.query(BizProtocolPaymentDetail).count() == 0
    attachment = db.get(BizAttachment, 'att-protocol-a')
    assert attachment.business_type == 'market:specialist:protocol:staging'
    assert attachment.business_id in (None, '')


@pytest.mark.parametrize(
    'attachment',
    [
        {
            'id': 'unknown-ref', 'file_url': 'protocol/unknown.pdf',
            'business_type': 'market:specialist:protocol:staging',
            'tenant_no': 'tenant-a', 'create_by': 'u1',
        },
        {
            'id': 'cross-tenant-ref', 'file_url': 'protocol/cross.pdf',
            'business_type': 'market:specialist:protocol:staging',
            'tenant_no': 'tenant-b', 'create_by': 'u1',
        },
        {
            'id': 'wrong-module-ref', 'file_url': 'payment/wrong-module.pdf',
            'business_type': 'market:specialist:payment:staging',
            'tenant_no': 'tenant-a', 'create_by': 'u1',
        },
        {
            'id': 'other-user-ref', 'file_url': 'protocol/other-user.pdf',
            'business_type': 'market:specialist:protocol:staging',
            'tenant_no': 'tenant-a', 'create_by': 'u2',
        },
    ],
)
def test_v1_market_update_rejects_untrusted_payment_detail_attachment_and_rolls_back(attachment):
    from app.schemas.market_mobile import MarketProtocolBody
    from app.services.market_protocol import update_protocol

    db = _session()
    _masters(db)
    row = _cell(db, f"legacy-{attachment['id']}")
    row.payment_model_version = 1
    payment = BizProtocolPayment(
        id=f"payment-{attachment['id']}", collection_protocol_id=row.id,
        pay_mode='dj', contract_amount=Decimal('100.00'),
        tenant_no='tenant-a', create_dept_id='', del_flag=0,
    )
    detail = BizProtocolPaymentDetail(
        id=f"detail-{attachment['id']}", payment_id=payment.id,
        pay_status='dfj', due_amount=Decimal('100.00'),
        tenant_no='tenant-a', create_dept_id='', del_flag=0,
    )
    if attachment['id'] != 'unknown-ref':
        db.add(
            BizAttachment(
                id=attachment['id'], business_type=attachment['business_type'],
                original_name='proof.pdf', file_ext='pdf',
                file_url=attachment['file_url'], file_size=10,
                file_type='application/pdf', tenant_no=attachment['tenant_no'],
                create_by=attachment['create_by'], create_dept_id='', del_flag=0,
            )
        )
    db.add_all([payment, detail])
    db.commit()

    with pytest.raises(ValueError, match='附件'):
        update_protocol(
            db,
            source='cell_storage',
            protocol_id=row.id,
            body=MarketProtocolBody.model_validate(
                {
                    'source': 'cell_storage',
                    'remark': '不应提交',
                    'paymentDetails': [
                        {
                            'id': detail.id,
                            'dueAmount': '100.00',
                            'transferAttachments': attachment['file_url'],
                        }
                    ],
                }
            ),
            ctx=_ctx('specialist'),
            user_info=_user(),
        )

    db.expire_all()
    assert db.get(BizCollectionProtocol, row.id).remark is None
    assert db.get(BizProtocolPaymentDetail, detail.id).transfer_attachments is None


def test_v1_market_update_binds_payment_detail_attachment_to_protocol_business():
    from app.schemas.market_mobile import MarketProtocolBody
    from app.services.market_protocol import update_protocol

    db = _session()
    _masters(db)
    row = _cell(db, 'legacy-payment-attachment')
    row.payment_model_version = 1
    payment = BizProtocolPayment(
        id='legacy-payment-attachment-parent', collection_protocol_id=row.id,
        pay_mode='dj', contract_amount=Decimal('100.00'),
        tenant_no='tenant-a', create_dept_id='', del_flag=0,
    )
    detail = BizProtocolPaymentDetail(
        id='legacy-payment-attachment-detail', payment_id=payment.id,
        pay_status='dfj', due_amount=Decimal('100.00'),
        tenant_no='tenant-a', create_dept_id='', del_flag=0,
    )
    db.add_all([payment, detail])
    db.commit()

    update_protocol(
        db,
        source='cell_storage',
        protocol_id=row.id,
        body=MarketProtocolBody.model_validate(
            {
                'source': 'cell_storage',
                'paymentDetails': [
                    {
                        'id': detail.id,
                        'dueAmount': '100.00',
                        'attachments': 'protocol/a.pdf',
                    }
                ],
            }
        ),
        ctx=_ctx('specialist'),
        user_info=_user(),
    )

    db.expire_all()
    saved_detail = db.get(BizProtocolPaymentDetail, detail.id)
    saved_attachment = db.get(BizAttachment, 'att-protocol-a')
    assert saved_detail.transfer_attachments == 'protocol/a.pdf'
    assert saved_attachment.business_type == 'market:specialist:protocol:bound'
    assert saved_attachment.business_id == row.id


def test_protocol_attachment_download_authorizes_exact_payment_detail_relation_only():
    from app.services.market_support import get_market_attachment_download_url

    class FakeStorage:
        @staticmethod
        def presigned_download_url(file_url, file_name):
            return f'https://download.test/{file_url}?name={file_name}'

    db = _session()
    own = _cell(db, 'payment-attachment-own')
    hidden = _cell(db, 'payment-attachment-hidden', sales='u2')
    own_payment = BizProtocolPayment(
        id='payment-attachment-own-parent', collection_protocol_id=own.id,
        pay_mode='dj', tenant_no='tenant-a', create_dept_id='', del_flag=0,
    )
    hidden_payment = BizProtocolPayment(
        id='payment-attachment-hidden-parent', collection_protocol_id=hidden.id,
        pay_mode='dj', tenant_no='tenant-a', create_dept_id='', del_flag=0,
    )
    db.add_all(
        [
            own_payment,
            hidden_payment,
            BizProtocolPaymentDetail(
                id='payment-attachment-own-detail', payment_id=own_payment.id,
                transfer_attachments='protocol/exact.pdf', pay_status='dfj',
                tenant_no='tenant-a', create_dept_id='', del_flag=0,
            ),
            BizProtocolPaymentDetail(
                id='payment-attachment-hidden-detail', payment_id=hidden_payment.id,
                transfer_attachments='protocol/hidden.pdf', pay_status='dfj',
                tenant_no='tenant-a', create_dept_id='', del_flag=0,
            ),
            BizProtocolPaymentDetail(
                id='payment-attachment-substring-detail', payment_id=own_payment.id,
                transfer_attachments='protocol/exact.pdf.evil', pay_status='dfj',
                tenant_no='tenant-a', create_dept_id='', del_flag=0,
            ),
            BizAttachment(
                id='payment-attachment-exact',
                business_type='market:specialist:protocol:bound', business_id=own.id,
                original_name='exact.pdf', file_ext='pdf', file_url='protocol/exact.pdf',
                file_size=10, file_type='application/pdf', tenant_no='tenant-a',
                create_by='u1', create_dept_id='', del_flag=0,
            ),
            BizAttachment(
                id='payment-attachment-hidden',
                business_type='market:specialist:protocol:bound', business_id=hidden.id,
                original_name='hidden.pdf', file_ext='pdf', file_url='protocol/hidden.pdf',
                file_size=10, file_type='application/pdf', tenant_no='tenant-a',
                create_by='u2', create_dept_id='', del_flag=0,
            ),
            BizAttachment(
                id='payment-attachment-substring',
                business_type='market:specialist:protocol:bound', business_id=own.id,
                original_name='exact-short.pdf', file_ext='pdf', file_url='protocol/exact.pdf.evil',
                file_size=10, file_type='application/pdf', tenant_no='tenant-a',
                create_by='u1', create_dept_id='', del_flag=0,
            ),
        ]
    )
    db.commit()

    result = get_market_attachment_download_url(
        db,
        file_url='protocol/exact.pdf',
        file_name=None,
        module='protocol',
        ctx=_ctx('specialist'),
        user_info=_user(),
        storage_backend=FakeStorage(),
    )
    assert result['downloadUrl'].startswith('https://download.test/protocol/exact.pdf?')

    with pytest.raises(PermissionError):
        get_market_attachment_download_url(
            db,
            file_url='protocol/hidden.pdf',
            file_name=None,
            module='protocol',
            ctx=_ctx('specialist'),
            user_info=_user(),
            storage_backend=FakeStorage(),
        )

    db.query(BizProtocolPaymentDetail).filter_by(
        id='payment-attachment-own-detail'
    ).delete()
    db.commit()
    with pytest.raises(PermissionError):
        get_market_attachment_download_url(
            db,
            file_url='protocol/exact.pdf',
            file_name=None,
            module='protocol',
            ctx=_ctx('specialist'),
            user_info=_user(),
            storage_backend=FakeStorage(),
        )


@pytest.mark.parametrize('boundary', ['tenant', 'scope'])
def test_v1_market_update_cannot_cross_tenant_or_manager_scope(boundary):
    from app.schemas.market_mobile import MarketProtocolBody
    from app.services.market_protocol import update_protocol

    db = _session()
    _masters(db)
    row = _cell(
        db,
        f'legacy-{boundary}-hidden',
        sales='u2',
        manager='manager-a',
        region='region-b' if boundary == 'scope' else 'region-a',
    )
    row.payment_model_version = 1
    payment = BizProtocolPayment(
        id=f'legacy-{boundary}-payment',
        collection_protocol_id=row.id,
        pay_mode='dj',
        contract_amount=Decimal('100.00'),
        tenant_no='tenant-a',
        create_dept_id='',
        del_flag=0,
    )
    if boundary == 'tenant':
        row.tenant_no = 'tenant-b'
        db.query(BizProtocolCollectionInfo).filter_by(
            collection_protocol_id=row.id
        ).update({'tenant_no': 'tenant-b'})
        payment.tenant_no = 'tenant-b'
    db.add(payment)
    db.commit()

    with pytest.raises(HTTPException):
        update_protocol(
            db,
            source='cell_storage',
            protocol_id=row.id,
            body=MarketProtocolBody.model_validate(
                {
                    'source': 'cell_storage',
                    'payment': {'payMode': 'dj', 'contractAmount': '999.00'},
                    'paymentDetails': [],
                }
            ),
            ctx=_ctx('specialist' if boundary == 'tenant' else 'manager'),
            user_info=_user(),
        )

    db.expire_all()
    assert db.get(BizProtocolPayment, payment.id).contract_amount == Decimal('100.00')


def test_cell_sample_partial_edit_preserves_hidden_lifecycle_fields():
    from app.schemas.market_mobile import MarketProtocolBody
    from app.services.market_protocol import update_protocol

    db = _session()
    _masters(db)
    row = _cell(db, 'sample-merge')
    db.add(
        BizProtocolSample(
            id='sample-merge-old',
            collection_protocol_id=row.id,
            sample_type_id='type-a',
            sample_code_id='code-a',
            sample_code='S-001',
            collect_amount=12.5,
            unit_id='unit-a',
            price=88,
            original_price=99,
            storage_years=20,
            sample_status='yrk',
            storage_in_date='20261001',
            detect_start_date='20260929',
            void_date='20261002',
            storage_expire_date='20461001',
            is_special=1,
            special_remark='历史特殊说明',
            remark='历史样本备注',
            tenant_no='tenant-a',
            create_dept_id='',
            del_flag=0,
        )
    )
    db.commit()

    with patch(
        'app.crud.collection_protocol._migrate_sample_child_refs_on_replace'
    ):
        update_protocol(
            db,
            source='cell_storage',
            protocol_id=row.id,
            body=MarketProtocolBody.model_validate(
                {
                    'source': 'cell_storage',
                    'samples': [
                        {
                            'sampleTypeId': 'type-a',
                            'sampleCodeId': 'code-a',
                            'sampleCode': 'S-001',
                            'price': 77,
                        }
                    ],
                }
            ),
            ctx=_ctx('specialist'),
            user_info=_user(),
        )

    saved = db.query(BizProtocolSample).filter_by(
        collection_protocol_id=row.id, del_flag=0
    ).one()
    assert float(saved.collect_amount) == 12.5
    assert saved.unit_id == 'unit-a'
    assert saved.remark == '历史样本备注'
    assert saved.storage_in_date == '20261001'
    assert saved.detect_start_date == '20260929'
    assert saved.void_date == '20261002'
    assert saved.storage_expire_date == '20461001'
    assert float(saved.price) == 77


def test_donation_sample_partial_edit_preserves_hidden_lifecycle_fields():
    from app.schemas.market_mobile import MarketProtocolBody
    from app.services.market_protocol import update_protocol

    db = _session()
    _masters(db)
    row = _donation(db, 'donation-sample-merge')
    db.add(
        BizResearchDonationSample(
            id='donation-sample-merge-old',
            research_donation_id=row.id,
            sample_type_id='type-a',
            sample_code_id='code-a',
            sample_code='S-001',
            collect_amount=9.5,
            unit_id='unit-a',
            storage_years=10,
            sample_status='jcz',
            storage_in_date='20261001',
            detect_start_date='20260929',
            void_date='20261002',
            storage_expire_date='20361001',
            is_special=1,
            special_remark='历史捐赠说明',
            remark='历史捐赠备注',
            tenant_no='tenant-a',
            create_dept_id='',
            del_flag=0,
        )
    )
    db.commit()

    with patch(
        'app.crud.research_donation._migrate_donation_sample_child_refs_on_replace'
    ):
        update_protocol(
            db,
            source='research_donation',
            protocol_id=row.id,
            body=MarketProtocolBody.model_validate(
                {
                    'source': 'research_donation',
                    'samples': [
                        {
                            'sampleTypeId': 'type-a',
                            'sampleCodeId': 'code-a',
                            'sampleCode': 'S-001',
                            'storageYears': 11,
                        }
                    ],
                }
            ),
            ctx=_ctx('specialist'),
            user_info=_user(),
        )

    saved = db.query(BizResearchDonationSample).filter_by(
        research_donation_id=row.id, del_flag=0
    ).one()
    assert float(saved.collect_amount) == 9.5
    assert saved.unit_id == 'unit-a'
    assert saved.remark == '历史捐赠备注'
    assert saved.storage_in_date == '20261001'
    assert saved.detect_start_date == '20260929'
    assert saved.void_date == '20261002'
    assert saved.storage_expire_date == '20361001'
    assert saved.storage_years == 11


def _protocol_sample_row(source, *, row_id, protocol_id, code_id, code, amount, remark):
    common = {
        'id': row_id,
        'sample_type_id': 'type-a',
        'sample_code_id': code_id,
        'sample_code': code,
        'collect_amount': amount,
        'unit_id': 'unit-a',
        'sample_status': 'yrk',
        'remark': remark,
        'tenant_no': 'tenant-a',
        'create_dept_id': '',
        'del_flag': 0,
    }
    if source == 'cell_storage':
        return BizProtocolSample(collection_protocol_id=protocol_id, **common)
    return BizResearchDonationSample(research_donation_id=protocol_id, **common)


def _active_protocol_samples(db, source, protocol_id):
    if source == 'cell_storage':
        return (
            db.query(BizProtocolSample)
            .filter_by(collection_protocol_id=protocol_id, del_flag=0)
            .order_by(BizProtocolSample.order_num.asc())
            .all()
        )
    return (
        db.query(BizResearchDonationSample)
        .filter_by(research_donation_id=protocol_id, del_flag=0)
        .order_by(BizResearchDonationSample.order_num.asc())
        .all()
    )


@pytest.mark.parametrize('source', ['cell_storage', 'research_donation'])
def test_protocol_sample_update_uses_id_for_insert_and_reorder(source):
    from app.schemas.market_mobile import MarketProtocolBody
    from app.services.market_protocol import update_protocol

    db = _session()
    _masters(db)
    row_id = f'{source}-sample-reorder'
    (_cell if source == 'cell_storage' else _donation)(db, row_id)
    db.add_all(
        [
            _protocol_sample_row(
                source,
                row_id='sample-a',
                protocol_id=row_id,
                code_id='code-a',
                code='S-001',
                amount=11,
                remark='隐藏-A',
            ),
            _protocol_sample_row(
                source,
                row_id='sample-b',
                protocol_id=row_id,
                code_id='code-b',
                code='S-002',
                amount=22,
                remark='隐藏-B',
            ),
        ]
    )
    db.commit()

    migrate_target = (
        'app.crud.collection_protocol._migrate_sample_child_refs_on_replace'
        if source == 'cell_storage'
        else 'app.crud.research_donation._migrate_donation_sample_child_refs_on_replace'
    )
    with patch(migrate_target):
        update_protocol(
            db,
            source=source,
            protocol_id=row_id,
            body=MarketProtocolBody.model_validate(
                {
                    'source': source,
                    'samples': [
                        {
                            'sampleTypeId': 'type-a',
                            'sampleCodeId': 'code-c',
                            'sampleCode': 'S-003',
                        },
                        {
                            'id': 'sample-b',
                            'sampleTypeId': 'type-a',
                            'sampleCodeId': 'code-b',
                            'sampleCode': 'S-002',
                        },
                        {
                            'id': 'sample-a',
                            'sampleTypeId': 'type-a',
                            'sampleCodeId': 'code-a',
                            'sampleCode': 'S-001',
                        },
                    ],
                }
            ),
            ctx=_ctx('specialist'),
            user_info=_user(),
        )

    saved = _active_protocol_samples(db, source, row_id)
    assert [row.sample_code for row in saved] == ['S-003', 'S-002', 'S-001']
    assert [row.remark for row in saved] == [None, '隐藏-B', '隐藏-A']
    assert [float(row.collect_amount) if row.collect_amount is not None else None for row in saved] == [
        None,
        22.0,
        11.0,
    ]


@pytest.mark.parametrize('source', ['cell_storage', 'research_donation'])
def test_protocol_sample_update_uses_id_when_replacing_code(source):
    from app.schemas.market_mobile import MarketProtocolBody
    from app.services.market_protocol import update_protocol

    db = _session()
    _masters(db)
    row_id = f'{source}-sample-recode'
    (_cell if source == 'cell_storage' else _donation)(db, row_id)
    db.add(
        _protocol_sample_row(
            source,
            row_id='sample-recode',
            protocol_id=row_id,
            code_id='code-a',
            code='S-001',
            amount=15,
            remark='必须保留',
        )
    )
    db.commit()

    migrate_target = (
        'app.crud.collection_protocol._migrate_sample_child_refs_on_replace'
        if source == 'cell_storage'
        else 'app.crud.research_donation._migrate_donation_sample_child_refs_on_replace'
    )
    with patch(migrate_target):
        update_protocol(
            db,
            source=source,
            protocol_id=row_id,
            body=MarketProtocolBody.model_validate(
                {
                    'source': source,
                    'samples': [
                        {
                            'sampleId': 'sample-recode',
                            'sampleTypeId': 'type-a',
                            'sampleCodeId': 'code-c',
                            'sampleCode': 'S-003',
                        }
                    ],
                }
            ),
            ctx=_ctx('specialist'),
            user_info=_user(),
        )

    saved = _active_protocol_samples(db, source, row_id)[0]
    assert saved.sample_code_id == 'code-c'
    assert saved.sample_code == 'S-003'
    assert float(saved.collect_amount) == 15
    assert saved.remark == '必须保留'


@pytest.mark.parametrize('source', ['cell_storage', 'research_donation'])
@pytest.mark.parametrize('duplicate_field', ['sampleCode', 'sampleCodeId'])
def test_protocol_sample_write_rejects_duplicate_identity(source, duplicate_field):
    from app.services.market_protocol import create_protocol

    db = _session()
    _masters(db)
    if duplicate_field == 'sampleCode':
        samples = [
            {'sampleTypeId': 'type-a', 'sampleCode': 'DUP-001'},
            {'sampleTypeId': 'type-a', 'sampleCode': 'DUP-001'},
        ]
    else:
        samples = [
            {'sampleTypeId': 'type-a', 'sampleCodeId': 'code-a', 'sampleCode': 'S-001'},
            {'sampleTypeId': 'type-a', 'sampleCodeId': 'code-a', 'sampleCode': 'S-ALT'},
        ]
    overrides = {'samples': samples}
    if source == 'research_donation':
        overrides.update({'donationNo': 'D-DUP', 'protocolNo': None})

    with pytest.raises(ValueError, match=f'{duplicate_field}.*重复'):
        create_protocol(
            db,
            body=_body(source, **overrides),
            ctx=_ctx('specialist'),
            user_info=_user(),
        )


@pytest.mark.parametrize('source', ['cell_storage', 'research_donation'])
def test_protocol_sample_update_rejects_duplicate_sample_id(source):
    from app.schemas.market_mobile import MarketProtocolBody
    from app.services.market_protocol import update_protocol

    db = _session()
    _masters(db)
    row_id = f'{source}-duplicate-sample-id'
    (_cell if source == 'cell_storage' else _donation)(db, row_id)
    db.add(
        _protocol_sample_row(
            source,
            row_id='sample-duplicate-id',
            protocol_id=row_id,
            code_id='code-a',
            code='S-001',
            amount=10,
            remark='不能双重继承',
        )
    )
    db.commit()
    migrate_target = (
        'app.crud.collection_protocol._migrate_sample_child_refs_on_replace'
        if source == 'cell_storage'
        else 'app.crud.research_donation._migrate_donation_sample_child_refs_on_replace'
    )

    with patch(migrate_target), pytest.raises(ValueError, match='sampleId.*重复'):
        update_protocol(
            db,
            source=source,
            protocol_id=row_id,
            body=MarketProtocolBody.model_validate(
                {
                    'source': source,
                    'samples': [
                        {
                            'id': 'sample-duplicate-id',
                            'sampleTypeId': 'type-a',
                            'sampleCodeId': 'code-b',
                            'sampleCode': 'S-002',
                        },
                        {
                            'sampleId': 'sample-duplicate-id',
                            'sampleTypeId': 'type-a',
                            'sampleCodeId': 'code-c',
                            'sampleCode': 'S-003',
                        },
                    ],
                }
            ),
            ctx=_ctx('specialist'),
            user_info=_user(),
        )


@pytest.mark.parametrize('source', ['cell_storage', 'research_donation'])
def test_protocol_sample_update_rejects_sample_id_from_another_protocol(source):
    from app.schemas.market_mobile import MarketProtocolBody
    from app.services.market_protocol import update_protocol

    db = _session()
    _masters(db)
    owner_id = f'{source}-owner'
    other_id = f'{source}-other'
    creator = _cell if source == 'cell_storage' else _donation
    creator(db, owner_id)
    creator(db, other_id)
    db.add_all(
        [
            _protocol_sample_row(
                source,
                row_id='sample-owner',
                protocol_id=owner_id,
                code_id='code-a',
                code='S-001',
                amount=10,
                remark='本协议',
            ),
            _protocol_sample_row(
                source,
                row_id='sample-foreign',
                protocol_id=other_id,
                code_id='code-b',
                code='S-002',
                amount=20,
                remark='其他协议',
            ),
        ]
    )
    db.commit()

    with pytest.raises(ValueError, match='样本.*不属于当前协议'):
        update_protocol(
            db,
            source=source,
            protocol_id=owner_id,
            body=MarketProtocolBody.model_validate(
                {
                    'source': source,
                    'samples': [
                        {
                            'id': 'sample-foreign',
                            'sampleTypeId': 'type-a',
                            'sampleCodeId': 'code-a',
                            'sampleCode': 'S-001',
                        }
                    ],
                }
            ),
            ctx=_ctx('specialist'),
            user_info=_user(),
        )


def test_partial_collection_update_validates_against_resulting_region():
    from app.schemas.market_mobile import MarketProtocolBody
    from app.services.market_protocol import update_protocol

    db = _session()
    _masters(db)
    _cell(db, 'cell-partial', sales='u1')
    db.commit()

    update_protocol(
        db,
        source='cell_storage',
        protocol_id='cell-partial',
        body=MarketProtocolBody.model_validate(
            {
                'source': 'cell_storage',
                'collection': {'hospitalId': 'hospital-a'},
            }
        ),
        ctx=_ctx('specialist'),
        user_info=_user(),
    )

    collection = db.query(BizProtocolCollectionInfo).filter_by(
        collection_protocol_id='cell-partial', del_flag=0
    ).one()
    assert collection.region_id == 'region-a'
    assert collection.hospital_id == 'hospital-a'


def test_cell_partial_customer_update_preserves_ids_and_unspecified_fields():
    from app.schemas.market_mobile import MarketProtocolBody
    from app.services.market_protocol import update_protocol

    db = _session()
    _masters(db)
    _cell(db, 'cell-party')
    db.add_all(
        [
            BizCustomer(id='mother-c', customer_name='母客户', sex='1', tenant_no='tenant-a', create_dept_id='', del_flag=0),
            BizCustomer(id='father-c', customer_name='父客户', sex='0', tenant_no='tenant-a', create_dept_id='', del_flag=0),
            BizCustomer(id='newborn-c', customer_name='子客户', sex='2', tenant_no='tenant-a', create_dept_id='', del_flag=0),
            BizProtocolPartyCustomer(
                id='party-cell', collection_protocol_id='cell-party',
                customer_id='mother-c', father_customer_id='father-c', newborn_customer_id='newborn-c',
                mother_name='母亲', mother_phone='旧电话', mother_id_card='ID-M',
                father_name='父亲', father_phone='父电话', newborn_name='宝宝',
                tenant_no='tenant-a', create_dept_id='', del_flag=0,
            ),
        ]
    )
    db.commit()

    update_protocol(
        db,
        source='cell_storage',
        protocol_id='cell-party',
        body=MarketProtocolBody.model_validate(
            {
                'source': 'cell_storage',
                'customer': {'motherPhone': '新电话', 'remark': '只改电话'},
            }
        ),
        ctx=_ctx('specialist'),
        user_info=_user(),
    )

    party = db.get(BizProtocolPartyCustomer, 'party-cell')
    assert (party.customer_id, party.father_customer_id, party.newborn_customer_id) == (
        'mother-c', 'father-c', 'newborn-c'
    )
    assert party.mother_name == '母亲'
    assert party.mother_id_card == 'ID-M'
    assert party.father_name == '父亲'
    assert party.father_phone == '父电话'
    assert party.newborn_name == '宝宝'
    assert party.mother_phone == '新电话'
    assert party.remark == '只改电话'


@pytest.mark.parametrize('source', ['cell_storage', 'research_donation'])
def test_protocol_update_can_rebind_same_tenant_customer_id(source):
    from app.schemas.market_mobile import MarketProtocolBody
    from app.services.market_protocol import update_protocol

    db = _session()
    _masters(db)
    db.add_all(
        [
            BizCustomer(id='linked-c', customer_name='已关联', sex='1', tenant_no='tenant-a', create_dept_id='', del_flag=0),
            BizCustomer(id='unrelated-c', customer_name='无关客户', sex='1', tenant_no='tenant-a', create_dept_id='', del_flag=0),
        ]
    )
    if source == 'cell_storage':
        _cell(db, 'customer-idor')
        db.add(BizProtocolPartyCustomer(id='linked-party', collection_protocol_id='customer-idor', customer_id='linked-c', mother_name='原姓名', tenant_no='tenant-a', create_dept_id='', del_flag=0))
    else:
        _donation(db, 'customer-idor')
        db.add(BizResearchDonationParty(id='linked-party', research_donation_id='customer-idor', customer_id='linked-c', mother_name='原姓名', tenant_no='tenant-a', create_dept_id='', del_flag=0))
    db.commit()

    update_protocol(
        db,
        source=source,
        protocol_id='customer-idor',
        body=MarketProtocolBody.model_validate(
            {
                'source': source,
                'customer': {'customerId': 'unrelated-c', 'motherName': '无关客户'},
            }
        ),
        ctx=_ctx('specialist'),
        user_info=_user(),
    )

    db.expire_all()
    if source == 'cell_storage':
        party = (
            db.query(BizProtocolPartyCustomer)
            .filter_by(collection_protocol_id='customer-idor', del_flag=0)
            .one()
        )
    else:
        party = (
            db.query(BizResearchDonationParty)
            .filter_by(research_donation_id='customer-idor', del_flag=0)
            .one()
        )
    assert party.customer_id == 'unrelated-c'
    assert party.mother_name == '无关客户'


@pytest.mark.parametrize('source', ['cell_storage', 'research_donation'])
def test_protocol_update_rejects_missing_or_cross_tenant_customer_id(source):
    from app.schemas.market_mobile import MarketProtocolBody
    from app.services.market_protocol import update_protocol

    db = _session()
    _masters(db)
    db.add(
        BizCustomer(
            id='other-tenant-c',
            customer_name='外租户',
            sex='1',
            tenant_no='tenant-b',
            create_dept_id='',
            del_flag=0,
        )
    )
    if source == 'cell_storage':
        _cell(db, 'customer-cross')
        db.add(
            BizProtocolPartyCustomer(
                id='cross-party',
                collection_protocol_id='customer-cross',
                mother_name='原姓名',
                tenant_no='tenant-a',
                create_dept_id='',
                del_flag=0,
            )
        )
    else:
        _donation(db, 'customer-cross')
        db.add(
            BizResearchDonationParty(
                id='cross-party',
                research_donation_id='customer-cross',
                mother_name='原姓名',
                tenant_no='tenant-a',
                create_dept_id='',
                del_flag=0,
            )
        )
    db.commit()

    with pytest.raises(ValueError, match='客户'):
        update_protocol(
            db,
            source=source,
            protocol_id='customer-cross',
            body=MarketProtocolBody.model_validate(
                {
                    'source': source,
                    'customer': {'customerId': 'other-tenant-c', 'motherName': '外租户'},
                }
            ),
            ctx=_ctx('specialist'),
            user_info=_user(),
        )


@pytest.mark.parametrize('source', ['cell_storage', 'research_donation'])
def test_protocol_update_does_not_mutate_unrelated_customer_by_id_card(source):
    from app.schemas.market_mobile import MarketProtocolBody
    from app.services.market_protocol import update_protocol

    db = _session()
    _masters(db)
    db.add(BizCustomer(id='unrelated-card', customer_name='身份证客户', sex='1', id_card_no='CARD-1', tenant_no='tenant-a', create_dept_id='', del_flag=0))
    if source == 'cell_storage':
        _cell(db, 'card-idor')
        db.add(BizProtocolPartyCustomer(id='card-party', collection_protocol_id='card-idor', mother_name='原姓名', tenant_no='tenant-a', create_dept_id='', del_flag=0))
    else:
        _donation(db, 'card-idor')
        db.add(BizResearchDonationParty(id='card-party', research_donation_id='card-idor', mother_name='原姓名', tenant_no='tenant-a', create_dept_id='', del_flag=0))
    db.commit()

    update_protocol(
        db,
        source=source,
        protocol_id='card-idor',
        body=MarketProtocolBody.model_validate(
            {
                'source': source,
                'customer': {'motherName': '协议姓名', 'motherIdCard': 'CARD-1'},
            }
        ),
        ctx=_ctx('specialist'),
        user_info=_user(),
    )

    db.expire_all()
    assert db.get(BizCustomer, 'unrelated-card').customer_name == '身份证客户'


@pytest.mark.parametrize('source', ['cell_storage', 'research_donation'])
def test_protocol_create_allows_existing_customer_ids(source):
    from app.services.market_protocol import create_protocol

    db = _session()
    _masters(db)
    db.add(BizCustomer(id='existing-c', customer_name='已有客户', sex='1', tenant_no='tenant-a', create_dept_id='', del_flag=0))
    db.commit()

    created = create_protocol(
        db,
        body=_body(
            source,
            **({'donationNo': 'D-CUSTOMER', 'protocolNo': None} if source == 'research_donation' else {}),
            customer={'customerId': 'existing-c', 'motherName': '已有客户'},
        ),
        ctx=_ctx('specialist'),
        user_info=_user(),
    )
    assert (created.get('customer') or {}).get('customerId') == 'existing-c'


def test_market_protocol_body_accepts_sample_payments():
    from app.schemas.market_mobile import MarketProtocolBody

    body = MarketProtocolBody.model_validate(
        {
            'source': 'cell_storage',
            'protocolNo': 'P-PAY-1',
            'customer': {'motherName': '李四'},
            'samplePayments': [
                {
                    'sampleCode': 'S-001',
                    'payMode': 'dj',
                    'contractAmount': 20000,
                    'lumpSumDeadline': '20261231',
                    'schedules': [
                        {
                            'scheduleNo': 1,
                            'dueAmount': 20000,
                            'dueDate': '20261231',
                        }
                    ],
                }
            ],
        }
    )
    assert body.sample_payments is not None
    assert len(body.sample_payments) == 1
    assert body.sample_payments[0].sample_code == 'S-001'
    assert body.sample_payments[0].pay_mode == 'dj'


def test_market_protocol_money_json_keeps_decimal_18_2_through_service_payload():
    from app.schemas.market_mobile import MarketProtocolBody
    from app.services.market_protocol import (
        _market_sample_payment_payload,
        _persist_sample_payments_if_any,
    )

    amount = '9999999999999999.99'
    body = MarketProtocolBody.model_validate_json(json.dumps({
        'source': 'cell_storage',
        'protocolTotalAmount': amount,
        'samplePayments': [{
            'sampleCode': 'S-001',
            'payMode': 'fq',
            'contractAmount': amount,
            'installmentRatios': [amount],
            'schedules': [{
                'scheduleNo': 1,
                'dueAmount': amount,
                'transfers': [{'transferAmount': amount, 'transferType': 'bank'}],
            }],
        }],
    }))
    assert body.protocol_total_amount == Decimal(amount)
    assert body.sample_payments[0].contract_amount == Decimal(amount)
    assert body.sample_payments[0].schedules[0].due_amount == Decimal(amount)
    assert body.sample_payments[0].schedules[0].transfers[0].transfer_amount == Decimal(amount)
    db = _session()
    row = _cell(db, 'decimal-boundary')
    row.payment_model_version = 2
    sample = BizProtocolSample(
        id='sample-decimal', collection_protocol_id=row.id,
        sample_code='S-001', tenant_no='tenant-a', del_flag=0,
    )
    db.add(sample)
    db.commit()
    payload = _market_sample_payment_payload(
        db,
        row=row,
        body=body,
        ctx=_ctx('specialist'),
        saved_samples=[sample],
    )
    assert payload[0]['contractAmount'] == Decimal(amount)
    assert payload[0]['schedules'][0]['dueAmount'] == Decimal(amount)
    with patch(
        'app.services.market_protocol.psp_crud.save_sample_payments_for_protocol'
    ) as save:
        _persist_sample_payments_if_any(
            db,
            row=row,
            body=body,
            ctx=_ctx('specialist'),
            saved_samples=[sample],
        )
    crud_payload = save.call_args.kwargs['sample_payments']
    assert crud_payload[0]['contractAmount'] == Decimal(amount)
    assert crud_payload[0]['schedules'][0]['dueAmount'] == Decimal(amount)


def test_mobile_payment_response_serializes_decimal_as_fixed_string():
    from app.crud.protocol_sample_payment import _transfer_to_dict

    transfer = BizProtocolPaymentTransfer(
        id='t-boundary',
        transfer_amount=Decimal('9999999999999999.99'),
        tenant_no='tenant-a',
        del_flag=0,
    )
    response = _transfer_to_dict(_session(), transfer, 'tenant-a', money_as_string=True)
    encoded = json.dumps(response)
    assert response['transferAmount'] == '9999999999999999.99'
    assert '1e+16' not in encoded.lower()


def test_protocol_create_persists_sample_payments():
    from app.services.market_protocol import create_protocol, get_protocol

    db = _session()
    _masters(db)
    body = _body(
        samplePayments=[
            {
                'sampleCode': 'S-001',
                'payMode': 'dj',
                'contractAmount': 20000,
                'lumpSumDeadline': '20261231',
                'schedules': [
                    {
                        'scheduleNo': 1,
                        'dueAmount': 20000,
                        'dueDate': '20261231',
                        'scheduleStatus': 'open',
                        'transfers': [],
                    }
                ],
            }
        ]
    )
    created = create_protocol(
        db, body=body, ctx=_ctx('specialist'), user_info=_user()
    )
    protocol_id = created['basic']['id']
    pays = created.get('samplePayments') or []
    assert len(pays) == 1
    assert pays[0]['sampleCode'] == 'S-001'
    assert pays[0]['payMode'] == 'dj'
    assert pays[0]['contractAmount'] == '20000.00'
    assert pays[0]['schedules'][0]['dueAmount'] == '20000.00'
    assert str(pays[0]['schedules'][0].get('dueDate') or '').replace('-', '')[:8] == '20261231'

    detail = get_protocol(
        db,
        source='cell_storage',
        protocol_id=protocol_id,
        ctx=_ctx('specialist'),
        user_info=_user(),
    )
    assert len(detail.get('samplePayments') or []) == 1
    assert detail['samplePayments'][0]['sampleCode'] == 'S-001'
    assert detail['samplePayments'][0]['contractAmount'] == '20000.00'


@pytest.mark.parametrize('scope_kind', ['specialist', 'manager'])
def test_protocol_update_rejects_foreign_sample_payment_id(scope_kind):
    from app.services.market_protocol import _market_sample_payment_payload

    db = _session()
    _masters(db)
    owner = _cell(db, 'pay-owner')
    _cell(db, 'pay-foreign')
    owner_sample = BizProtocolSample(
        id='sample-owner', collection_protocol_id='pay-owner',
        sample_type_id='type-a', sample_code='S-001', tenant_no='tenant-a',
        create_dept_id='', del_flag=0,
    )
    db.add_all(
        [
            owner_sample,
            BizProtocolSamplePayment(
                id='payment-foreign', collection_protocol_id='pay-foreign',
                protocol_sample_id='sample-foreign', sample_code='S-FOREIGN',
                pay_mode='dj', contract_amount=999, tenant_no='tenant-a',
                create_dept_id='', del_flag=0,
            ),
        ]
    )
    db.commit()

    body = _body(
        samples=None,
        attachments=None,
        samplePayments=[
            {
                'id': 'payment-foreign',
                'sampleCode': 'S-001',
                'protocolSampleId': 'sample-foreign',
                'payMode': 'dj',
                'contractAmount': 100,
                'schedules': [
                    {'scheduleNo': 1, 'dueAmount': 100, 'dueDate': '20261031'}
                ],
            }
        ]
    )

    with pytest.raises(ValueError, match='缴费方案'):
        _market_sample_payment_payload(
            db,
            row=owner,
            body=body,
            ctx=_ctx(scope_kind, 'u1' if scope_kind == 'specialist' else 'manager-a'),
            saved_samples=[owner_sample],
        )

    db.expire_all()
    foreign = db.get(BizProtocolSamplePayment, 'payment-foreign')
    assert foreign.collection_protocol_id == 'pay-foreign'
    assert foreign.sample_code == 'S-FOREIGN'
    assert float(foreign.contract_amount) == 999


def test_protocol_payment_payload_uses_saved_sample_and_preserves_submitted_transfer():
    from app.crud.protocol_sample_payment import save_sample_payments_for_protocol
    from app.services.market_protocol import _market_sample_payment_payload

    db = _session()
    _masters(db)
    owner = _cell(db, 'pay-safe')
    sample = BizProtocolSample(
        id='sample-safe', collection_protocol_id='pay-safe',
        sample_type_id='type-a', sample_code='S-SAFE', tenant_no='tenant-a',
        create_dept_id='', del_flag=0,
    )
    payment = BizProtocolSamplePayment(
        id='payment-safe', collection_protocol_id='pay-safe',
        protocol_sample_id='sample-safe-old', sample_code='S-SAFE', pay_mode='dj',
        contract_amount=100, tenant_no='tenant-a', create_dept_id='', del_flag=0,
    )
    schedule = BizProtocolPaymentSchedule(
        id='schedule-safe', sample_payment_id='payment-safe',
        collection_protocol_id='pay-safe', sample_code='S-SAFE', schedule_no=1,
        due_amount=100, due_date='20261031', schedule_status='open',
        tenant_no='tenant-a', create_dept_id='', del_flag=0,
    )
    transfer = BizProtocolPaymentTransfer(
        id='transfer-safe', schedule_id='schedule-safe',
        sample_payment_id='payment-safe', collection_protocol_id='pay-safe',
        sample_code='S-SAFE', transfer_amount=20, transfer_type='bank',
        transfer_attachments='payment/existing.jpg', pay_status='jfqrz',
        tenant_no='tenant-a', create_dept_id='', del_flag=0,
    )
    db.add_all([sample, payment, schedule, transfer])
    db.commit()
    body = _body(
        samples=None,
        attachments=None,
        samplePayments=[
            {
                'id': 'payment-safe', 'sampleCode': 'S-SAFE',
                'protocolSampleId': 'foreign-sample', 'payMode': 'dj',
                'contractAmount': 100,
                'schedules': [
                    {
                        'id': 'foreign-schedule', 'scheduleNo': 1,
                        'dueAmount': 100, 'dueDate': '20261031',
                        'scheduleStatus': 'void',
                        'transfers': [
                            {
                                'id': 'transfer-safe', 'transferAmount': 999,
                                'transferAttachments': 'evil.jpg',
                            }
                        ],
                    }
                ],
            }
        ],
    )

    payload = _market_sample_payment_payload(
        db,
        row=owner,
        body=body,
        ctx=_ctx('specialist'),
        saved_samples=[sample],
    )
    assert payload[0]['protocolSampleId'] == 'sample-safe'
    assert payload[0]['schedules'][0]['scheduleStatus'] == 'open'
    assert payload[0]['schedules'][0]['transfers'] == [{'id': 'transfer-safe'}]

    save_sample_payments_for_protocol(
        db,
        collection_protocol_id='pay-safe',
        tenant_no='tenant-a',
        user_id='u1',
        sample_payments=payload,
    )
    db.expire_all()
    saved_transfer = db.get(BizProtocolPaymentTransfer, 'transfer-safe')
    assert float(saved_transfer.transfer_amount) == 20
    assert saved_transfer.transfer_attachments == 'payment/existing.jpg'
    assert saved_transfer.pay_status == 'jfqrz'
    assert db.get(BizProtocolSamplePayment, 'payment-safe').protocol_sample_id == 'sample-safe'


def test_save_sample_payments_serializes_decimal_installment_ratios_as_fixed_strings():
    from app.crud.protocol_sample_payment import save_sample_payments_for_protocol

    db = _session()
    _cell(db, 'decimal-ratios')
    save_sample_payments_for_protocol(
        db,
        collection_protocol_id='decimal-ratios',
        tenant_no='tenant-a',
        user_id='u1',
        sample_payments=[
            {
                'sampleCode': 'S-DECIMAL',
                'payMode': 'fq',
                'contractAmount': Decimal('100.00'),
                'installmentCount': 2,
                'installmentRatios': [
                    Decimal('20.00'),
                    Decimal('80.00'),
                    Decimal('9999999999999999.99'),
                ],
                'schedules': [
                    {'scheduleNo': 1, 'dueAmount': Decimal('20.00'), 'dueDate': '20261101'},
                    {'scheduleNo': 2, 'dueAmount': Decimal('80.00'), 'dueDate': '20261201'},
                ],
            },
            {
                'sampleCode': 'S-STRING',
                'payMode': 'fq',
                'contractAmount': Decimal('100.00'),
                'installmentCount': 1,
                'installmentRatios': 'legacy-string',
                'schedules': [
                    {'scheduleNo': 1, 'dueAmount': Decimal('100.00'), 'dueDate': '20261101'},
                ],
            },
        ],
    )
    saved = (
        db.query(BizProtocolSamplePayment)
        .filter(
            BizProtocolSamplePayment.collection_protocol_id == 'decimal-ratios',
            BizProtocolSamplePayment.sample_code == 'S-DECIMAL',
        )
        .one()
    )
    assert json.loads(saved.installment_ratios) == [
        '20.00',
        '80.00',
        '9999999999999999.99',
    ]
    legacy = (
        db.query(BizProtocolSamplePayment)
        .filter(
            BizProtocolSamplePayment.collection_protocol_id == 'decimal-ratios',
            BizProtocolSamplePayment.sample_code == 'S-STRING',
        )
        .one()
    )
    assert legacy.installment_ratios == 'legacy-string'
def test_protocol_update_rejects_unscoped_freight_target():
    from app.services.market_protocol import update_protocol

    db = _session()
    _masters(db)
    _cell(db, 'freight-owner')
    _cell(db, 'freight-hidden', sales='u2', manager='manager-other', region='region-b')
    db.add_all(
        [
            BizProtocolSample(
                id='sample-owner', collection_protocol_id='freight-owner',
                sample_type_id='type-a', sample_code='S-001', tenant_no='tenant-a',
                create_dept_id='', del_flag=0,
            ),
            BizProtocolSample(
                id='sample-hidden', collection_protocol_id='freight-hidden',
                sample_type_id='type-a', sample_code='S-HIDDEN', tenant_no='tenant-a',
                create_dept_id='', del_flag=0,
            ),
            BizSampleFreight(
                id='freight-hidden', freight_no='F-HIDDEN', transport_mode='zxc',
                freight_amount=80, tenant_no='tenant-a', create_dept_id='', del_flag=0,
            ),
            BizSampleFreightItem(
                id='freight-item-hidden', freight_id='freight-hidden',
                sample_source='cell_storage', sample_id='sample-hidden',
                protocol_id='freight-hidden', sample_code='S-HIDDEN',
                tenant_no='tenant-a', create_dept_id='', del_flag=0,
            ),
        ]
    )
    db.commit()

    body = _body(
        protocolNo='CELL-freight-owner',
        samples=[
            {
                'sampleTypeId': 'type-a', 'sampleCodeId': 'code-a',
                'sampleCode': 'S-001', 'unitId': 'unit-a',
                'collectAmount': 10, 'freightId': 'freight-hidden',
            }
        ]
    )

    with pytest.raises(HTTPException):
        update_protocol(
            db,
            source='cell_storage',
            protocol_id='freight-owner',
            body=body,
            ctx=_ctx('specialist'),
            user_info=_user(),
        )

    db.expire_all()
    assert db.get(BizProtocolSample, 'sample-owner').del_flag == 0
    hidden_item = db.get(BizSampleFreightItem, 'freight-item-hidden')
    assert hidden_item.sample_id == 'sample-hidden'


def test_protocol_body_rejects_status_finance_and_conversion_fields():
    from app.schemas.market_mobile import MarketProtocolBody

    forbidden = [
        {'auditStatus': 'shtg'},
        {'protocolStatus': 'yzf'},
        {'delFlag': 1},
        {'refundAmount': 100},
        {'financeRemark': '通过'},
        {'toResearchTime': '2026-10-10'},
    ]
    for field in forbidden:
        with pytest.raises(ValidationError):
            MarketProtocolBody.model_validate({'source': 'cell_storage', **field})


def test_cross_tenant_or_disabled_referenced_user_is_rejected():
    from app.services.market_protocol import create_protocol

    db = _session()
    _masters(db)
    db.add(
        SysUser(
            id='disabled', user_no='disabled', nick_name='停用', status='0', tenant_no='tenant-a', create_dept_id='', del_flag=0
        )
    )
    db.commit()

    with pytest.raises(ValueError, match='用户'):
        create_protocol(
            db,
            body=_body(collection={'regionId': 'region-a', 'hospitalId': 'hospital-a', 'regionManagerUserId': 'disabled'}),
            ctx=_ctx('specialist'),
            user_info=_user(),
        )


@pytest.mark.parametrize('source', ['cell_storage', 'research_donation'])
@pytest.mark.parametrize(
    ('field_label', 'scope_kind', 'collection'),
    [
        (
            '市场专员',
            'manager',
            {
                'regionId': 'region-a',
                'hospitalId': 'hospital-a',
                'salesUserId': 'u2',
                'regionManagerUserId': 'u1',
            },
        ),
        (
            '地区经理',
            'specialist',
            {
                'regionId': 'region-a',
                'hospitalId': 'hospital-a',
                'salesUserId': 'u1',
                'regionManagerUserId': 'u2',
            },
        ),
    ],
)
def test_protocol_create_allows_assignment_without_target_region_staff_binding(
    source, field_label, scope_kind, collection
):
    from app.services.market_protocol import create_protocol

    db = _session()
    _masters(db)
    overrides = {'collection': collection, 'attachments': {}}
    if source == 'research_donation':
        overrides.update({'donationNo': 'D-OUTSIDER', 'protocolNo': None})

    created = create_protocol(
        db,
        body=_body(source, **overrides),
        ctx=_ctx(scope_kind),
        user_info=_user(),
    )
    saved = created['collection']
    if field_label == '市场专员':
        assert saved['salesUserId'] == 'u2'
    else:
        assert saved['regionManagerUserId'] == 'u2'


@pytest.mark.parametrize('source', ['cell_storage', 'research_donation'])
@pytest.mark.parametrize(
    ('field_name', 'scope_kind'),
    [('salesUserId', 'manager'), ('regionManagerUserId', 'specialist')],
)
def test_protocol_update_allows_changed_assignment_without_target_region_staff_binding(
    source, field_name, scope_kind
):
    from app.schemas.market_mobile import MarketProtocolBody
    from app.services.market_protocol import update_protocol

    db = _session()
    _masters(db)
    row_id = f'{source}-{field_name}-outsider'
    if source == 'cell_storage':
        _cell(
            db,
            row_id,
            sales='u1',
            manager='u1' if scope_kind == 'manager' else 'manager-a',
        )
    else:
        _donation(
            db,
            row_id,
            sales='u1',
            manager='u1' if scope_kind == 'manager' else 'manager-a',
        )
    db.commit()

    updated = update_protocol(
        db,
        source=source,
        protocol_id=row_id,
        body=MarketProtocolBody.model_validate(
            {'source': source, 'collection': {field_name: 'u2'}}
        ),
        ctx=_ctx(scope_kind),
        user_info=_user(),
    )
    assert updated['collection'][field_name] == 'u2'


@pytest.mark.parametrize('source', ['cell_storage', 'research_donation'])
@pytest.mark.parametrize(
    ('historical_field', 'scope_kind'),
    [('sales_user_id', 'manager'), ('region_manager_user_id', 'specialist')],
)
def test_protocol_update_keeps_unchanged_historical_assignment_without_binding(
    source, historical_field, scope_kind
):
    from app.schemas.market_mobile import MarketProtocolBody
    from app.services.market_protocol import update_protocol

    db = _session()
    _masters(db)
    row_id = f'{source}-{historical_field}-legacy'
    kwargs = {'sales': 'u1', 'manager': 'manager-a'}
    if historical_field == 'sales_user_id':
        kwargs.update({'sales': 'u2', 'manager': 'u1'})
    else:
        kwargs['manager'] = 'u2'
    if source == 'cell_storage':
        _cell(db, row_id, **kwargs)
    else:
        _donation(db, row_id, **kwargs)
    db.commit()

    result = update_protocol(
        db,
        source=source,
        protocol_id=row_id,
        body=MarketProtocolBody.model_validate(
            {'source': source, 'remark': '仅修改非归属字段'}
        ),
        ctx=_ctx(scope_kind),
        user_info=_user(),
    )

    assert result['basic']['remark'] == '仅修改非归属字段'


@pytest.mark.parametrize('source', ['cell_storage', 'research_donation'])
@pytest.mark.parametrize(
    ('sample_code_id', 'sample_code'),
    [
        ('code-cross', 'S-CROSS'),
        ('code-deleted', 'S-DELETED'),
        ('code-a', 'S-WRONG'),
    ],
)
def test_protocol_rejects_invalid_or_inconsistent_sample_code_reference(
    source, sample_code_id, sample_code
):
    from app.services.market_protocol import create_protocol

    db = _session()
    _masters(db)
    payload = {
        'samples': [
            {
                'sampleTypeId': 'type-a',
                'sampleCodeId': sample_code_id,
                'sampleCode': sample_code,
                'unitId': 'unit-a',
            }
        ]
    }
    if source == 'research_donation':
        payload.update({'donationNo': 'D-CODE', 'protocolNo': None})

    with pytest.raises(ValueError, match='样本编码'):
        create_protocol(
            db,
            body=_body(source, **payload),
            ctx=_ctx('specialist'),
            user_info=_user(),
        )


def test_cell_update_rolls_back_main_and_existing_children_after_late_failure():
    from app.schemas.market_mobile import MarketProtocolBody
    from app.services.market_protocol import update_protocol

    db = _session()
    _masters(db)
    _cell(db, 'cell-update')
    db.add_all(
        [
            BizProtocolPartyCustomer(
                id='party-old', collection_protocol_id='cell-update', mother_name='旧姓名', tenant_no='tenant-a', create_dept_id='', del_flag=0
            ),
            BizProtocolSample(
                id='sample-old', collection_protocol_id='cell-update', sample_code='OLD-CODE', tenant_no='tenant-a', create_dept_id='', del_flag=0
            ),
        ]
    )
    db.commit()
    body = MarketProtocolBody.model_validate(
        {
            'source': 'cell_storage',
            'remark': '新备注',
            'customer': {'motherName': '新姓名'},
            'collection': {'medicalStaff': '新医护'},
            'samples': [{'sampleTypeId': 'type-a', 'sampleCodeId': 'code-a', 'sampleCode': 'S-001'}],
        }
    )

    with patch(
        'app.services.market_protocol.cell_crud.save_samples',
        side_effect=RuntimeError('late failure'),
    ):
        with pytest.raises(RuntimeError, match='late'):
            update_protocol(
                db,
                source='cell_storage',
                protocol_id='cell-update',
                body=body,
                ctx=_ctx('specialist'),
                user_info=_user(),
            )

    db.expire_all()
    assert db.get(BizCollectionProtocol, 'cell-update').remark is None
    assert db.get(BizProtocolPartyCustomer, 'party-old').mother_name == '旧姓名'
    assert db.get(BizProtocolPartyCustomer, 'party-old').del_flag == 0
    assert db.get(BizProtocolSample, 'sample-old').sample_code == 'OLD-CODE'
    assert db.get(BizProtocolCollectionInfo, 'ci-cell-update').medical_staff is None


def test_donation_update_rolls_back_main_and_old_children_after_soft_delete():
    from app.schemas.market_mobile import MarketProtocolBody
    from app.services.market_protocol import update_protocol

    db = _session()
    _masters(db)
    _donation(db, 'don-update')
    db.add_all(
        [
            BizResearchDonationParty(
                id='dparty-old', research_donation_id='don-update', mother_name='旧姓名', tenant_no='tenant-a', create_dept_id='', del_flag=0
            ),
            BizResearchDonationSample(
                id='dsample-old', research_donation_id='don-update', sample_code='OLD-CODE', tenant_no='tenant-a', create_dept_id='', del_flag=0
            ),
        ]
    )
    db.commit()
    body = MarketProtocolBody.model_validate(
        {
            'source': 'research_donation',
            'donationNo': 'DON-don-update',
            'remark': '新备注',
            'customer': {'motherName': '新姓名'},
            'collection': {'medicalStaff': '新医护'},
            'samples': [{'sampleTypeId': 'type-a', 'sampleCodeId': 'code-a', 'sampleCode': 'S-001'}],
        }
    )

    with patch(
        'app.crud.research_donation._create_samples',
        side_effect=RuntimeError('late failure'),
    ):
        with pytest.raises(RuntimeError, match='late'):
            update_protocol(
                db,
                source='research_donation',
                protocol_id='don-update',
                body=body,
                ctx=_ctx('specialist'),
                user_info=_user(),
            )

    db.expire_all()
    assert db.get(BizResearchDonation, 'don-update').remark is None
    assert db.get(BizResearchDonationParty, 'dparty-old').mother_name == '旧姓名'
    assert db.get(BizResearchDonationParty, 'dparty-old').del_flag == 0
    assert db.get(BizResearchDonationSample, 'dsample-old').del_flag == 0
    assert db.get(BizResearchDonationCollection, 'dci-don-update').medical_staff is None


def test_manager_can_update_donation_in_scope_but_cannot_move_it_out():
    from app.schemas.market_mobile import MarketProtocolBody
    from app.services.market_protocol import update_protocol

    db = _session()
    _masters(db)
    _donation(db, 'don-manager', sales='u2', manager='u1')
    db.add(
        BizResearchDonationParty(
            id='dparty-manager', research_donation_id='don-manager', mother_name='旧姓名', tenant_no='tenant-a', create_dept_id='', del_flag=0
        )
    )
    db.commit()

    result = update_protocol(
        db,
        source='research_donation',
        protocol_id='don-manager',
        body=MarketProtocolBody.model_validate(
            {
                'source': 'research_donation',
                'donationNo': 'DON-don-manager',
                'remark': '经理已修改',
                'customer': {'motherName': '新姓名'},
                'collection': {'medicalStaff': '新医护'},
            }
        ),
        ctx=_ctx('manager'),
        user_info=_user(),
    )
    assert result['basic']['remark'] == '经理已修改'
    assert result['customer']['motherName'] == '新姓名'
    assert result['collection']['medicalStaff'] == '新医护'

    with pytest.raises(ValueError, match='范围'):
        update_protocol(
            db,
            source='research_donation',
            protocol_id='don-manager',
            body=MarketProtocolBody.model_validate(
                {
                    'source': 'research_donation',
                    'collection': {
                        'regionId': 'region-b',
                        'regionManagerUserId': 'u1',
                    },
                }
            ),
            ctx=_ctx('manager'),
            user_info=_user(),
        )
    db.expire_all()
    collection = db.query(BizResearchDonationCollection).filter_by(
        research_donation_id='don-manager', del_flag=0
    ).one()
    assert collection.region_id == 'region-a'
    assert collection.region_manager_user_id == 'u1'


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

    for scope in ('specialist', 'manager'):
        prefix = f'/mini/market/{scope}/protocols/{{protocol_id}}'
        routes = {
            route.path: route.methods
            for route in router.routes
            if route.path.startswith(f'/mini/market/{scope}/protocols')
        }
        assert 'DELETE' in routes[prefix]
        assert f'{prefix}/audit' not in routes
        assert all(not any(word in path for word in ('void', 'refund', 'convert', 'to-research', 'to-storage')) for path in routes)
        assert not any(
            route.path.startswith(f'/mini/market/{scope}/samples')
            or route.path.startswith(f'/mini/market/{scope}/sample/attachments')
            for route in router.routes
        )


def test_protocol_delete_rechecks_scoped_access_before_crud(monkeypatch):
    from app.services.market_protocol import delete_protocol

    db = _session()
    _masters(db)
    own = _cell(db, 'own', sales='u1')
    other = _cell(db, 'other', sales='u2')
    db.commit()
    delete_calls = []
    monkeypatch.setattr(
        'app.services.market_protocol.cell_crud.soft_delete_collection_protocol',
        lambda *args, **kwargs: delete_calls.append((args, kwargs)),
    )

    delete_protocol(
        db,
        source='cell_storage',
        protocol_id=own.id,
        ctx=_ctx('specialist'),
        user_info=_user(),
    )
    assert delete_calls[0][0][1] == own.id

    with pytest.raises(HTTPException, match='无权'):
        delete_protocol(
            db,
            source='cell_storage',
            protocol_id=other.id,
            ctx=_ctx('specialist'),
            user_info=_user(),
        )


def test_protocol_list_contains_mobile_card_fields():
    from app.services.market_protocol import list_protocols

    db = _session()
    _masters(db)
    row = _cell(db, 'card')
    row.protocol_total_amount = 20000
    db.add_all([
        BizProtocolPartyCustomer(
            id='party-card', collection_protocol_id=row.id, mother_name='李安然',
            mother_phone='13800138000', mother_id_card='340100', father_name='王先生',
            newborn_sex='1', delivery_mode='sc', newborn_weight_gram=3200,
            tenant_no='tenant-a', create_dept_id='', del_flag=0,
        ),
        BizProtocolSample(
            id='sample-card', collection_protocol_id=row.id, sample_type_id='type-a',
            sample_code='S-CARD', tenant_no='tenant-a', create_dept_id='', del_flag=0,
        ),
    ])
    db.commit()

    items, total = list_protocols(db, ctx=_ctx('specialist'), user_info=_user())

    assert total == 1
    assert items[0]['motherName'] == '李安然'
    assert items[0]['sampleCode'] == 'S-CARD'
    assert items[0]['sampleTypeName'] == '样本A'
    assert items[0]['regionName'] == 'A区'
    assert items[0]['motherPhone'] == '13800138000'
    assert items[0]['motherIdCard'] == '340100'
    assert items[0]['fatherName'] == '王先生'
    assert items[0]['newbornSex'] == '1'
    assert items[0]['deliveryMode'] == 'sc'
    assert items[0]['newbornWeightGram'] == 3200
    assert items[0]['salesUserName'] == '当前用户'

    by_customer, customer_total = list_protocols(
        db, ctx=_ctx('specialist'), user_info=_user(), keyword='李安然'
    )
    by_sample, sample_total = list_protocols(
        db, ctx=_ctx('specialist'), user_info=_user(), keyword='S-CARD'
    )
    by_phone, phone_total = list_protocols(
        db, ctx=_ctx('specialist'), user_info=_user(), keyword='13800138000'
    )
    by_id_card, id_card_total = list_protocols(
        db, ctx=_ctx('specialist'), user_info=_user(), keyword='340100'
    )
    assert customer_total == sample_total == phone_total == id_card_total == 1
    assert {
        by_customer[0]['id'],
        by_sample[0]['id'],
        by_phone[0]['id'],
        by_id_card[0]['id'],
    } == {row.id}


@pytest.mark.parametrize('source', ['cell_storage', 'research_donation'])
def test_protocol_list_and_detail_resolve_tenant_void_type_label(source):
    from app.services.market_protocol import get_protocol, list_protocols

    db = _session()
    _masters(db)
    db.add(
        SysDictData(
            id=f'void-dict-{source}',
            dict_type='xyzflx',
            dict_value='tenant-custom',
            dict_label='租户自定义作废',
            tenant_no='tenant-a',
            create_dept_id='',
            del_flag=0,
        )
    )
    if source == 'cell_storage':
        row = _cell(db, 'void-label-cell')
        row.protocol_status = 'yzf'
        row.void_type = 'tenant-custom'
    else:
        row = _donation(db, 'void-label-donation')
        row.donation_status = 'yzf'
        row.void_type = 'tenant-custom'
    db.commit()

    items, total = list_protocols(
        db,
        ctx=_ctx('specialist'),
        user_info=_user(),
        source=source,
    )
    detail = get_protocol(
        db,
        source=source,
        protocol_id=row.id,
        ctx=_ctx('specialist'),
        user_info=_user(),
    )

    assert total == 1
    assert items[0]['voidType'] == 'tenant-custom'
    assert items[0]['voidTypeLabel'] == '租户自定义作废'
    assert detail['basic']['voidType'] == 'tenant-custom'
    assert detail['basic']['voidTypeLabel'] == '租户自定义作废'


def test_unknown_void_type_label_preserves_original_code():
    from app.services.market_protocol import list_protocols

    db = _session()
    _masters(db)
    row = _cell(db, 'void-label-unknown')
    row.protocol_status = 'yzf'
    row.void_type = 'legacy-unknown'
    db.commit()

    items, _ = list_protocols(
        db,
        ctx=_ctx('specialist'),
        user_info=_user(),
        source='cell_storage',
    )

    assert items[0]['voidType'] == 'legacy-unknown'
    assert items[0]['voidTypeLabel'] == 'legacy-unknown'


def test_protocol_list_filters_status_create_time_and_sample_collect_time():
    from app.services.market_protocol import list_protocols

    db = _session()
    _masters(db)
    hit = _cell(db, 'hit', created=datetime(2026, 10, 8, 9, 0))
    hit.protocol_status = 'sxz'
    db.query(BizProtocolCollectionInfo).filter_by(
        collection_protocol_id=hit.id
    ).update({'sample_collect_time': datetime(2026, 10, 9, 10, 0)})
    miss_status = _cell(db, 'miss-status', created=datetime(2026, 10, 8, 9, 0))
    miss_status.protocol_status = 'dsx'
    db.query(BizProtocolCollectionInfo).filter_by(
        collection_protocol_id=miss_status.id
    ).update({'sample_collect_time': datetime(2026, 10, 9, 10, 0)})
    miss_date = _cell(db, 'miss-date', created=datetime(2026, 9, 1, 9, 0))
    miss_date.protocol_status = 'sxz'
    db.query(BizProtocolCollectionInfo).filter_by(
        collection_protocol_id=miss_date.id
    ).update({'sample_collect_time': datetime(2026, 9, 2, 10, 0)})
    db.commit()

    items, total = list_protocols(
        db,
        ctx=_ctx('specialist'),
        user_info=_user(),
        source='cell_storage',
        protocol_status='sxz',
        create_time_start='2026-10-01',
        create_time_end='2026-10-13',
        sample_collect_time_start='2026-10-01',
        sample_collect_time_end='2026-10-13',
    )

    assert total == 1
    assert [item['id'] for item in items] == ['hit']


def test_cell_create_rolls_back_main_customer_and_children_on_child_failure():
    from app.services.market_protocol import create_protocol

    db = _session()
    _masters(db)

    with patch(
        'app.services.market_protocol.cell_crud.upsert_collection_info',
        side_effect=RuntimeError('injected child failure'),
    ):
        with pytest.raises(RuntimeError, match='injected'):
            create_protocol(
                db, body=_body(), ctx=_ctx('specialist'), user_info=_user()
            )

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

def test_donation_create_rolls_back_customer_and_children_on_child_failure():
    from app.services.market_protocol import create_protocol

    db = _session()
    _masters(db)

    with patch(
        'app.crud.research_donation._create_collection',
        side_effect=RuntimeError('injected child failure'),
    ):
        with pytest.raises(RuntimeError, match='injected'):
            create_protocol(
                db,
                body=_body(
                    'research_donation', donationNo='D-ROLLBACK', protocolNo=None
                ),
                ctx=_ctx('specialist'),
                user_info=_user(),
            )

    assert db.query(BizResearchDonation).count() == 0
    assert db.query(BizResearchDonationParty).count() == 0
    assert db.query(BizCustomer).count() == 0


def test_cell_protocol_detail_includes_storage_insurance_and_audits():
    from app.services.market_protocol import get_protocol

    db = _session()
    _masters(db)
    _cell(db, 'cell-enrich')
    db.add(
        BizProtocolStorageCert(
            id='sc-1',
            collection_protocol_id='cell-enrich',
            make_status='1',
            make_date='20261001',
            tenant_no='tenant-a',
            create_dept_id='',
            del_flag=0,
        )
    )
    db.add(
        BizProtocolFullmoonTodo(
            id='fm-1',
            collection_protocol_id='cell-enrich',
            insurance_company_name='测试保险',
            insurance_policy_no='POL-1',
            tenant_no='tenant-a',
            create_dept_id='',
            del_flag=0,
        )
    )
    db.add(
        BizProtocolAudit(
            id='au-1',
            collection_protocol_id='cell-enrich',
            audit_status='shtg',
            audit_remark='通过',
            auditor_user_id='u1',
            tenant_no='tenant-a',
            create_dept_id='',
            del_flag=0,
        )
    )
    db.commit()

    detail = get_protocol(
        db,
        source='cell_storage',
        protocol_id='cell-enrich',
        ctx=_ctx('specialist'),
        user_info=_user(),
    )
    assert detail['storageCert']['makeStatus'] == '1'
    assert detail['fullmoonTodo']['insurancePolicyNo'] == 'POL-1'
    assert len(detail['audits']) == 1
    assert detail['audits'][0]['auditStatus'] == 'shtg'


def test_cell_protocol_detail_hides_cross_tenant_and_deleted_handler_names():
    from app.services.market_protocol import get_protocol

    db = _session()
    _masters(db)
    _cell(db, 'cell-handler-isolation')
    db.add_all(
        [
            SysUser(
                id='cross-tenant-handler',
                user_no='cross-handler',
                nick_name='异租户经办人',
                status='1',
                tenant_no='tenant-b',
                create_dept_id='',
                del_flag=0,
            ),
            SysUser(
                id='deleted-handler',
                user_no='deleted-handler',
                nick_name='已删除经办人',
                status='1',
                tenant_no='tenant-a',
                create_dept_id='',
                del_flag=1,
            ),
            BizProtocolStorageCert(
                id='sc-handler-isolation',
                collection_protocol_id='cell-handler-isolation',
                handler_user_id='cross-tenant-handler',
                tenant_no='tenant-a',
                create_dept_id='',
                del_flag=0,
            ),
            BizProtocolFullmoonTodo(
                id='fm-handler-isolation',
                collection_protocol_id='cell-handler-isolation',
                handler_user_id='deleted-handler',
                tenant_no='tenant-a',
                create_dept_id='',
                del_flag=0,
            ),
        ]
    )
    db.commit()

    detail = get_protocol(
        db,
        source='cell_storage',
        protocol_id='cell-handler-isolation',
        ctx=_ctx('specialist'),
        user_info=_user(),
    )

    assert detail['storageCert']['handlerUserName'] == ''
    assert detail['fullmoonTodo']['handlerUserName'] == ''


def test_cell_protocol_detail_includes_presign_deposit_hint_and_transfers():
    from app.services.market_protocol import get_protocol

    db = _session()
    _masters(db)
    row = _cell(db, 'cell-deposit')
    row.source_presign_order_id = 'presign-d1'
    db.add(
        BizPresignOrder(
            id='presign-d1',
            protocol_no='PS-D1',
            mother_name='张敏',
            order_status='yzh',
            tenant_no='tenant-a',
            create_dept_id='',
            del_flag=0,
        )
    )
    db.add(
        BizPresignTransfer(
            id='tr-d1',
            presign_order_id='presign-d1',
            transfer_amount=1000,
            transfer_type='wx',
            pay_status='yjf',
            tenant_no='tenant-a',
            create_dept_id='',
            del_flag=0,
        )
    )
    db.add(
        SysDictData(
            id='dict-deposit-wx',
            dict_label='微信支付',
            dict_value='wx',
            dict_type='zzfs',
            tenant_no='tenant-a',
            create_dept_id='',
            del_flag=0,
        )
    )
    db.commit()

    detail = get_protocol(
        db,
        source='cell_storage',
        protocol_id='cell-deposit',
        ctx=_ctx('specialist'),
        user_info=_user(),
    )
    assert detail['presignDepositHint']['confirmedDepositAmount'] == '1000.00'
    assert detail['presignDepositHint']['confirmedTransferCount'] == 1
    assert len(detail['depositTransfers']) == 1
    assert detail['depositTransfers'][0]['id'] == 'tr-d1'
    assert detail['depositTransfers'][0]['transferAmount'] == '1000.00'
    assert detail['depositTransfers'][0]['transferTypeLabel'] == '微信支付'


def test_protocol_customer_search_empty_keyword_returns_recent_list():
    """空关键字仍返回近期客户列表，供移动端「选档案」弹窗默认展示。"""
    from app.services.market_support import search_customers_for_protocol

    db = _session()
    db.add(
        BizCustomer(
            id='c-empty-kw',
            customer_name='可见客户',
            sex='1',
            tenant_no='tenant-a',
            create_dept_id='',
            del_flag=0,
        )
    )
    db.commit()

    data = search_customers_for_protocol(
        db, ctx=_ctx('specialist'), user_info=_user(), keyword='  '
    )
    assert [item['id'] for item in data['list']] == ['c-empty-kw']
    assert data['total'] == 1


def test_protocol_customer_search_matches_name_phone_and_isolates_tenant():
    from app.services.market_support import search_customers_for_protocol

    db = _session()
    db.add_all(
        [
            BizCustomer(
                id='c-hit',
                customer_name='张敏',
                phone_no='13800001111',
                id_card_no='ID-HIT',
                sex='1',
                tenant_no='tenant-a',
                create_dept_id='',
                del_flag=0,
            ),
            BizCustomer(
                id='c-other-tenant',
                customer_name='张敏',
                phone_no='13800001111',
                sex='1',
                tenant_no='tenant-b',
                create_dept_id='',
                del_flag=0,
            ),
            BizCustomer(
                id='c-miss',
                customer_name='李四',
                phone_no='13900002222',
                sex='1',
                tenant_no='tenant-a',
                create_dept_id='',
                del_flag=0,
            ),
        ]
    )
    db.commit()

    by_name = search_customers_for_protocol(
        db, ctx=_ctx('specialist'), user_info=_user(), keyword='张敏'
    )
    assert [item['id'] for item in by_name['list']] == ['c-hit']
    assert by_name['list'][0]['phoneNo'] == '13800001111'
    assert by_name['list'][0]['idCardNo'] == 'ID-HIT'

    by_phone = search_customers_for_protocol(
        db, ctx=_ctx('manager'), user_info=_user(), keyword='1380000'
    )
    assert [item['id'] for item in by_phone['list']] == ['c-hit']


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

    for scope in ('specialist', 'manager'):
        paths = {
            route.path
            for route in router.routes
            if route.path.startswith(f'/mini/market/{scope}/protocols')
        }
        assert f'/mini/market/{scope}/protocols/customers' in paths


def test_protocol_sample_body_accepts_status_and_freight_fields():
    from app.schemas.market_mobile import MarketProtocolBody

    body = MarketProtocolBody.model_validate(
        {
            'source': 'cell_storage',
            'samples': [
                {
                    'sampleTypeId': 'type-a',
                    'sampleCode': 'S-001',
                    'sampleStatus': 'ycj',
                    'etaTime': '2026-10-13 12:00:00',
                    'freightId': 'fr-1',
                    'freightDraft': {
                        'clientKey': 'fd_1',
                        'transportMode': 'zxc',
                        'freightAmount': 120,
                    },
                }
            ],
        }
    )
    sample = body.samples[0]
    assert sample.sample_status == 'ycj'
    assert sample.freight_id == 'fr-1'
    assert sample.freight_draft.client_key == 'fd_1'
    assert sample.freight_draft.transport_mode == 'zxc'
    assert sample.freight_draft.freight_amount == 120


def test_cell_protocol_detail_reuses_web_sample_lifecycle_bundle(monkeypatch):
    from app.services.market_protocol import get_protocol

    db = _session()
    _masters(db)
    _cell(db, 'cell-lifecycle')
    db.add(
        BizProtocolSample(
            id='ps-lifecycle',
            collection_protocol_id='cell-lifecycle',
            sample_type_id='type-a',
            sample_code='S-LIFE-1',
            sample_status='yrk',
            tenant_no='tenant-a',
            create_dept_id='',
            del_flag=0,
        )
    )
    db.commit()
    lifecycle_sample = {
        'id': 'ps-lifecycle',
        'sampleCode': 'S-LIFE-1',
        'sampleStatus': 'yrk',
        'sampleReception': {'receptionTime': '2026-10-01', 'receiveAttachmentFiles': [{'fileUrl': 'receive.pdf'}]},
        'productionReception': {'processTime': '2026-10-02'},
        'qcInfo': {'workOrderNo': 'QC-1'},
        'processQcInfoList': [{'workOrderNo': 'PQC-1'}],
        'productionInfo': {'workOrderNo': 'PR-1'},
        'storageReleaseInfo': {'storageStatus': 'yrk'},
    }
    monkeypatch.setattr(
        'app.services.market_protocol.cell_crud.get_protocol_detail_bundle',
        lambda *args, **kwargs: {'samples': [lifecycle_sample]},
    )

    detail = get_protocol(
        db,
        source='cell_storage',
        protocol_id='cell-lifecycle',
        ctx=_ctx('specialist'),
        user_info=_user(),
    )

    sample = detail['samples'][0]
    for key in [
        'sampleReception', 'productionReception', 'qcInfo',
        'processQcInfoList', 'productionInfo', 'storageReleaseInfo',
    ]:
        assert sample[key] == lifecycle_sample[key]


def test_cell_v1_detail_reuses_web_payment_details_and_transfer_type_dict(monkeypatch):
    from app.services.market_protocol import get_protocol

    db = _session()
    _masters(db)
    row = _cell(db, 'cell-v1-payment')
    row.payment_model_version = 1
    db.add(
        SysDictData(
            id='dict-wx',
            dict_label='微信支付',
            dict_value='wx',
            dict_type='zzfs',
            tenant_no='tenant-a',
            create_dept_id='',
            del_flag=0,
        )
    )
    db.commit()
    monkeypatch.setattr(
        'app.services.market_protocol.cell_crud.get_protocol_detail_bundle',
        lambda *args, **kwargs: {
            'samples': [],
            'paymentDetails': [{
                'id': 'pd-1',
                'installmentNo': 1,
                'dueAmount': 1000,
                'transferAmount': 800,
                'installmentDueDate': '20261031',
                'transferTime': '2026-10-13 12:00:00',
                'transferType': 'wx',
                'transferAttachments': 'proof.pdf',
                'payStatus': 'yjf',
                'financeRemark': '已确认',
            }],
        },
    )

    detail = get_protocol(
        db,
        source='cell_storage',
        protocol_id='cell-v1-payment',
        ctx=_ctx('specialist'),
        user_info=_user(),
    )

    assert detail['paymentModelVersion'] == 1
    assert detail['samplePayments'] == []
    assert detail['paymentDetails'][0]['installmentNo'] == 1
    assert detail['paymentDetails'][0]['transferTypeLabel'] == '微信支付'


def test_cell_v2_detail_labels_all_transfer_types_with_unknown_fallback(monkeypatch):
    from app.services.market_protocol import get_protocol

    db = _session()
    _masters(db)
    row = _cell(db, 'cell-v2-labels')
    row.payment_model_version = 2
    db.add(
        SysDictData(
            id='dict-v2-zfb',
            dict_label='支付宝',
            dict_value='zfb',
            dict_type='zzfs',
            tenant_no='tenant-a',
            create_dept_id='',
            del_flag=0,
        )
    )
    db.commit()
    monkeypatch.setattr(
        'app.services.market_protocol.cell_crud.get_protocol_detail_bundle',
        lambda *args, **kwargs: {'samples': [], 'paymentDetails': []},
    )
    monkeypatch.setattr(
        'app.services.market_protocol.psp_crud.get_sample_payments_bundle',
        lambda *args, **kwargs: [{
            'id': 'sp-1',
            'schedules': [{
                'id': 'sch-1',
                'transfers': [
                    {'id': 'tr-1', 'transferType': 'zfb'},
                    {'id': 'tr-2', 'transferType': 'future-code'},
                ],
            }],
        }],
    )

    detail = get_protocol(
        db,
        source='cell_storage',
        protocol_id='cell-v2-labels',
        ctx=_ctx('specialist'),
        user_info=_user(),
    )
    transfers = detail['samplePayments'][0]['schedules'][0]['transfers']
    assert transfers[0]['transferTypeLabel'] == '支付宝'
    assert transfers[1]['transferTypeLabel'] == 'future-code'


def test_donation_detail_reuses_web_bundle_lifecycle_audits_and_convert_aliases(monkeypatch):
    from app.services.market_protocol import get_protocol

    db = _session()
    _masters(db)
    _donation(db, 'don-web-detail')
    db.commit()
    lifecycle_sample = {
        'id': 'ds-1',
        'sampleCode': 'D-S-1',
        'sampleReception': {'receptionTime': '2026-10-01'},
        'productionReception': {'processTime': '2026-10-02'},
        'qcInfo': {'workOrderNo': 'D-QC-1'},
        'processQcInfoList': [{'workOrderNo': 'D-PQC-1'}],
        'productionInfo': {'workOrderNo': 'D-PR-1'},
        'storageReleaseInfo': {'storageStatus': 'yrk'},
    }
    monkeypatch.setattr(
        'app.services.market_protocol.donation_crud.get_donation_detail',
        lambda *args, **kwargs: {
            'donation': {
                'id': 'don-web-detail',
                'donationNo': 'DON-WEB',
                'donationStatus': 'zckc',
                'toStorageTime': '2026-10-10 10:00:00',
                'toStorageDesc': '已转细胞存储',
                'consentFormUrls': 'consent.pdf',
            },
            'party': {'motherName': '张敏'},
            'collection': {'regionName': 'A区'},
            'samples': [lifecycle_sample],
        },
    )
    monkeypatch.setattr(
        'app.services.market_protocol.donation_crud.list_audits',
        lambda *args, **kwargs: {
            'list': [{'id': 'da-1', 'auditStatus': 'shtg', 'auditRemark': '通过'}]
        },
    )

    detail = get_protocol(
        db,
        source='research_donation',
        protocol_id='don-web-detail',
        ctx=_ctx('specialist'),
        user_info=_user(),
    )

    assert detail['samples'][0] == lifecycle_sample
    assert detail['audits'][0]['id'] == 'da-1'
    assert detail['basic']['convertTime'] == '2026-10-10 10:00:00'
    assert detail['basic']['convertDesc'] == '已转细胞存储'


def test_cell_protocol_detail_attaches_sample_freight():
    from app.services.market_protocol import get_protocol

    db = _session()
    _masters(db)
    _cell(db, 'cell-fr')
    db.add(
        BizProtocolSample(
            id='ps-fr',
            collection_protocol_id='cell-fr',
            sample_type_id='type-a',
            sample_code='S-FR-1',
            sample_status='ycj',
            tenant_no='tenant-a',
            create_dept_id='',
            del_flag=0,
        )
    )
    db.add(
        SysDictData(
            id='dict-zxc',
            dict_label='专线车',
            dict_value='zxc',
            dict_type='sample_transport_mode',
            tenant_no='tenant-a',
            create_dept_id='',
            del_flag=0,
        )
    )
    db.add(
        BizSampleFreight(
            id='fr-1',
            freight_no='F-001',
            transport_mode='zxc',
            freight_amount=88,
            tenant_no='tenant-a',
            create_dept_id='',
            del_flag=0,
        )
    )
    db.add(
        BizSampleFreightItem(
            id='fri-1',
            freight_id='fr-1',
            sample_source='cell_storage',
            sample_id='ps-fr',
            protocol_id='cell-fr',
            sample_code='S-FR-1',
            tenant_no='tenant-a',
            create_dept_id='',
            del_flag=0,
        )
    )
    db.commit()

    detail = get_protocol(
        db,
        source='cell_storage',
        protocol_id='cell-fr',
        ctx=_ctx('specialist'),
        user_info=_user(),
    )
    sample = detail['samples'][0]
    assert sample['freightId'] == 'fr-1'
    assert sample['freight']['transportMode'] == 'zxc'
    assert sample['freight']['freightAmount'] == '88.00'


def test_protocol_update_replaces_freight_item_without_duplicate_active_links():
    from app.services.market_protocol import update_protocol

    db = _session()
    _masters(db)
    _cell(db, 'cell-edit')
    _cell(db, 'cell-sibling')
    db.add_all(
        [
            BizProtocolSample(
                id='sample-old', collection_protocol_id='cell-edit',
                sample_type_id='type-a', sample_code_id='code-a', sample_code='S-001',
                tenant_no='tenant-a', create_dept_id='', del_flag=0,
            ),
            BizProtocolSample(
                id='sample-sibling', collection_protocol_id='cell-sibling',
                sample_type_id='type-a', sample_code_id='code-b', sample_code='S-002',
                tenant_no='tenant-a', create_dept_id='', del_flag=0,
            ),
            BizSampleFreight(
                id='freight-shared', freight_no='F-SHARED', transport_mode='zxc',
                freight_amount=88, tenant_no='tenant-a', create_dept_id='', del_flag=0,
            ),
            BizSampleFreightItem(
                id='item-old', freight_id='freight-shared', sample_source='cell_storage',
                sample_id='sample-old', protocol_id='cell-edit', sample_code='S-001',
                tenant_no='tenant-a', create_dept_id='', del_flag=0,
            ),
            BizSampleFreightItem(
                id='item-sibling', freight_id='freight-shared', sample_source='cell_storage',
                sample_id='sample-sibling', protocol_id='cell-sibling', sample_code='S-002',
                tenant_no='tenant-a', create_dept_id='', del_flag=0,
            ),
        ]
    )
    db.commit()
    body = _body(
        protocolNo='CELL-cell-edit',
        samples=[{
            'sampleTypeId': 'type-a', 'sampleCodeId': 'code-a',
            'sampleCode': 'S-001', 'unitId': 'unit-a',
            'collectAmount': 10, 'freightId': 'freight-shared',
        }],
    )

    first = update_protocol(
        db, source='cell_storage', protocol_id='cell-edit', body=body,
        ctx=_ctx('specialist'), user_info=_user(),
    )
    second = update_protocol(
        db, source='cell_storage', protocol_id='cell-edit', body=body,
        ctx=_ctx('specialist'), user_info=_user(),
    )

    active_sample = db.query(BizProtocolSample).filter_by(
        collection_protocol_id='cell-edit', tenant_no='tenant-a', del_flag=0,
    ).one()
    active_items = db.query(BizSampleFreightItem).filter_by(
        freight_id='freight-shared', tenant_no='tenant-a', del_flag=0,
    ).all()
    assert {item.sample_id for item in active_items} == {
        active_sample.id, 'sample-sibling'
    }
    assert db.get(BizSampleFreightItem, 'item-old').del_flag == 1
    assert first['samples'][0]['id'] != second['samples'][0]['id']
    assert second['samples'][0]['freight']['linkedSampleCount'] == 2


def test_protocol_update_cancels_then_relinks_freight_without_touching_other_samples():
    from app.services.market_protocol import update_protocol

    db = _session()
    _masters(db)
    _cell(db, 'cell-move')
    _cell(db, 'cell-other-a')
    _cell(db, 'cell-other-b')
    samples = [
        BizProtocolSample(
            id='sample-move', collection_protocol_id='cell-move',
            sample_type_id='type-a', sample_code_id='code-a', sample_code='S-001',
            tenant_no='tenant-a', create_dept_id='', del_flag=0,
        ),
        BizProtocolSample(
            id='sample-other-a', collection_protocol_id='cell-other-a',
            sample_type_id='type-a', sample_code_id='code-b', sample_code='S-002',
            tenant_no='tenant-a', create_dept_id='', del_flag=0,
        ),
        BizProtocolSample(
            id='sample-other-b', collection_protocol_id='cell-other-b',
            sample_type_id='type-a', sample_code_id='code-c', sample_code='S-003',
            tenant_no='tenant-a', create_dept_id='', del_flag=0,
        ),
    ]
    freights = [
        BizSampleFreight(
            id='freight-a', freight_no='F-A', transport_mode='zxc', freight_amount=10,
            tenant_no='tenant-a', create_dept_id='', del_flag=0,
        ),
        BizSampleFreight(
            id='freight-b', freight_no='F-B', transport_mode='zxc', freight_amount=20,
            tenant_no='tenant-a', create_dept_id='', del_flag=0,
        ),
    ]
    items = [
        BizSampleFreightItem(
            id='item-move', freight_id='freight-a', sample_source='cell_storage',
            sample_id='sample-move', protocol_id='cell-move', sample_code='S-001',
            tenant_no='tenant-a', create_dept_id='', del_flag=0,
        ),
        BizSampleFreightItem(
            id='item-other-a', freight_id='freight-a', sample_source='cell_storage',
            sample_id='sample-other-a', protocol_id='cell-other-a', sample_code='S-002',
            tenant_no='tenant-a', create_dept_id='', del_flag=0,
        ),
        BizSampleFreightItem(
            id='item-other-b', freight_id='freight-b', sample_source='cell_storage',
            sample_id='sample-other-b', protocol_id='cell-other-b', sample_code='S-003',
            tenant_no='tenant-a', create_dept_id='', del_flag=0,
        ),
    ]
    role = SysRole(
        id='role-all', role_name='市场人员', data_scope='1', status='1',
        tenant_no='tenant-a', create_dept_id='', del_flag=0,
    )
    user_role = SysUserRole(
        id='user-role-all', user_id='u1', role_id='role-all',
        tenant_no='tenant-a', create_dept_id='', del_flag=0,
    )
    db.add_all([*samples, *freights, *items, role, user_role])
    db.commit()
    base_sample = {
        'sampleTypeId': 'type-a', 'sampleCodeId': 'code-a',
        'sampleCode': 'S-001', 'unitId': 'unit-a', 'collectAmount': 10,
    }

    cancelled = update_protocol(
        db, source='cell_storage', protocol_id='cell-move',
        body=_body(protocolNo='CELL-cell-move', samples=[base_sample]),
        ctx=_ctx('specialist'), user_info=_user(),
    )
    assert cancelled['samples'][0]['freightId'] == ''
    assert db.get(BizSampleFreightItem, 'item-move').del_flag == 1
    assert db.get(BizSampleFreightItem, 'item-other-a').del_flag == 0

    relinked = update_protocol(
        db, source='cell_storage', protocol_id='cell-move',
        body=_body(
            protocolNo='CELL-cell-move',
            samples=[{**base_sample, 'freightId': 'freight-b'}],
        ),
        ctx=_ctx('specialist'), user_info=_user(),
    )
    current_id = relinked['samples'][0]['id']
    assert relinked['samples'][0]['freight']['linkedSampleCount'] == 2
    assert {
        item.sample_id
        for item in db.query(BizSampleFreightItem).filter_by(
            freight_id='freight-b', tenant_no='tenant-a', del_flag=0,
        ).all()
    } == {'sample-other-b', current_id}
    assert db.get(BizSampleFreightItem, 'item-other-a').del_flag == 0


def test_protocol_update_freight_reconciliation_rolls_back_on_later_failure(monkeypatch):
    from app.services.market_protocol import update_protocol

    db = _session()
    _masters(db)
    _cell(db, 'cell-rollback')
    db.add_all(
        [
            BizProtocolSample(
                id='sample-rollback', collection_protocol_id='cell-rollback',
                sample_type_id='type-a', sample_code_id='code-a', sample_code='S-001',
                tenant_no='tenant-a', create_dept_id='', del_flag=0,
            ),
            BizSampleFreight(
                id='freight-rollback', freight_no='F-ROLLBACK', transport_mode='zxc',
                freight_amount=88, tenant_no='tenant-a', create_dept_id='', del_flag=0,
            ),
            BizSampleFreightItem(
                id='item-rollback', freight_id='freight-rollback',
                sample_source='cell_storage', sample_id='sample-rollback',
                protocol_id='cell-rollback', sample_code='S-001',
                tenant_no='tenant-a', create_dept_id='', del_flag=0,
            ),
        ]
    )
    db.commit()
    monkeypatch.setattr(
        'app.services.market_protocol._persist_sample_payments_if_any',
        lambda *args, **kwargs: (_ for _ in ()).throw(RuntimeError('later failure')),
    )

    with pytest.raises(RuntimeError, match='later failure'):
        update_protocol(
            db, source='cell_storage', protocol_id='cell-rollback',
            body=_body(
                protocolNo='CELL-cell-rollback',
                samples=[{
                    'sampleTypeId': 'type-a', 'sampleCodeId': 'code-a',
                    'sampleCode': 'S-001', 'unitId': 'unit-a',
                    'collectAmount': 10, 'freightId': 'freight-rollback',
                }],
            ),
            ctx=_ctx('specialist'), user_info=_user(),
        )

    db.expire_all()
    assert db.get(BizProtocolSample, 'sample-rollback').del_flag == 0
    assert db.get(BizSampleFreightItem, 'item-rollback').del_flag == 0
    assert db.query(BizProtocolSample).filter_by(
        collection_protocol_id='cell-rollback', tenant_no='tenant-a', del_flag=0,
    ).count() == 1
    assert db.query(BizSampleFreightItem).filter_by(
        protocol_id='cell-rollback', tenant_no='tenant-a', del_flag=0,
    ).count() == 1