zkxw/backend/tests/test_market_protocol_api.py

from datetime import datetime, timedelta
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 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,
    BizProtocolSample,
    BizProtocolSampleTransport,
    BizSampleCode,
    BizSampleReception,
    SysPost,
    SysPostRole,
    SysRole,
    SysRoleDataScopeDetail,
    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,
        tables=[
            BizCollectionProtocol.__table__,
            BizProtocolPartyCustomer.__table__,
            BizProtocolCollectionInfo.__table__,
            BizProtocolSample.__table__,
            BizProtocolSampleTransport.__table__,
            BizProtocolPayment.__table__,
            BizProtocolPaymentDetail.__table__,
            BizProtocolPaymentSchedule.__table__,
            BizProtocolPaymentTransfer.__table__,
            BizSampleCode.__table__,
            BizSampleReception.__table__,
            BizResearchDonation.__table__,
            BizResearchDonationParty.__table__,
            BizResearchDonationCollection.__table__,
            BizResearchDonationSample.__table__,
            BizResearchDonationSampleTransport.__table__,
            BizProtocolDonationConvert.__table__,
            BizCollectionRegionStaff.__table__,
            BizCollectionRegion.__table__,
            BizCollectionHospital.__table__,
            BizCellSampleType.__table__,
            BizItemUnit.__table__,
            BizCustomer.__table__,
            BizAttachment.__table__,
            SysUser.__table__,
            SysRole.__table__,
            SysRoleDataScopeDetail.__table__,
            SysUserRole.__table__,
            SysPost.__table__,
            SysPostRole.__table__,
            SysUserPost.__table__,
        ],
    )
    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-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
            ),
            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_rejects_hidden_update_action():
    from app.services.market_protocol import update_protocol

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

    with pytest.raises(ValueError, match='不可修改'):
        update_protocol(
            db,
            source='cell_storage',
            protocol_id=row.id,
            body=_body(protocolNo='CELL-terminal'),
            ctx=_ctx('specialist'),
            user_info=_user(),
        )


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.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.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': 'manager-a'}),
            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'


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_rejects_same_tenant_unlinked_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()

    with pytest.raises(ValueError, match='客户'):
        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()
    assert db.get(BizCustomer, 'unrelated-c').customer_name == '无关客户'
    assert db.get(BizCustomer, 'linked-c').customer_name == '已关联'


@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_forbids_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()

    with pytest.raises(ValueError, match='客户'):
        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(),
        )


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_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(
    ('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': 'manager-a',
                    },
                }
            ),
            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_scoped_audit_and_delete_but_not_other_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 routes[f'{prefix}/audit'] == {'POST'}
        assert 'DELETE' in routes[prefix]
        assert all(not any(word in path for word in ('void', 'refund', 'convert', 'to-research', 'to-storage')) for path in routes)


def test_protocol_audit_and_delete_recheck_scoped_access_before_crud(monkeypatch):
    from app.services.market_protocol import audit_protocol, delete_protocol

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

    audit_protocol(
        db,
        source='cell_storage',
        protocol_id=own.id,
        audit_status='shtg',
        audit_remark='通过',
        ctx=_ctx('specialist'),
        user_info=_user(),
    )
    delete_protocol(
        db,
        source='cell_storage',
        protocol_id=own.id,
        ctx=_ctx('specialist'),
        user_info=_user(),
    )
    assert audit_calls[0]['collection_protocol_id'] == own.id
    assert delete_calls[0][0][1] == own.id

    with pytest.raises(HTTPException, match='无权'):
        audit_protocol(
            db,
            source='cell_storage',
            protocol_id=other.id,
            audit_status='shtg',
            audit_remark='',
            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'
    )
    assert customer_total == sample_total == 1
    assert by_customer[0]['id'] == by_sample[0]['id'] == row.id


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