zkxw/backend/tests/test_market_sample_api.py

import asyncio
from datetime import datetime, timedelta
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,
    BizCollectionProtocol,
    BizCollectionRegionStaff,
    BizProtocolCollectionInfo,
    BizProtocolFullmoonTodo,
    BizProtocolPartyCustomer,
    BizProtocolSample,
    BizProtocolStorageCert,
    BizQcWorkOrder,
    BizSampleReception,
    BizStorageReleaseReview,
    SysPost,
    SysPostRole,
    SysRole,
    SysRoleDataScopeDetail,
    SysUser,
    SysUserPost,
    SysUserRole,
)
from app.models.research_donation import (
    BizResearchDonation,
    BizResearchDonationCollection,
    BizResearchDonationParty,
    BizResearchDonationSample,
)


TABLES = [
    BizAttachment.__table__,
    BizCollectionProtocol.__table__,
    BizProtocolCollectionInfo.__table__,
    BizProtocolPartyCustomer.__table__,
    BizProtocolSample.__table__,
    BizResearchDonation.__table__,
    BizResearchDonationCollection.__table__,
    BizResearchDonationParty.__table__,
    BizResearchDonationSample.__table__,
    BizCellSampleType.__table__,
    BizSampleReception.__table__,
    BizQcWorkOrder.__table__,
    BizStorageReleaseReview.__table__,
    BizProtocolStorageCert.__table__,
    BizProtocolFullmoonTodo.__table__,
    BizCollectionRegionStaff.__table__,
    SysUser.__table__,
    SysRole.__table__,
    SysRoleDataScopeDetail.__table__,
    SysUserRole.__table__,
    SysPost.__table__,
    SysPostRole.__table__,
    SysUserPost.__table__,
]


def _session():
    engine = create_engine('sqlite:///:memory:')
    Base.metadata.create_all(engine, tables=TABLES)
    db = sessionmaker(bind=engine)()
    for index, name in enumerate(
        (
            'a.jpg',
            'b.jpg',
            'cert-new.jpg',
            'cert-old.jpg',
            'i.jpg',
            'insurance-new.jpg',
            'insurance-old.jpg',
            'keep.jpg',
            'keep-insurance.jpg',
            'old.jpg',
        )
    ):
        db.add(
            BizAttachment(
                id=f'att-sample-{index}',
                business_type='market:specialist:sample:staging',
                original_name=name,
                file_ext='jpg',
                file_url=name,
                file_size=10,
                file_type='image/jpeg',
                tenant_no='tenant-a',
                create_by='u1',
                create_dept_id='',
                del_flag=0,
            )
        )
    db.commit()
    return db


def _ctx(kind='specialist', user_id='u1'):
    from app.services.market_field_scope import MarketFieldContext

    return MarketFieldContext('tenant-a', user_id, kind)


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


def _assign_region_staff(db, user_id: str, key: str, region_id: str = 'r1'):
    db.add(
        BizCollectionRegionStaff(
            id=f'staff-{key}',
            region_id=region_id,
            user_id=user_id,
            is_leader=0,
            tenant_no='tenant-a',
            create_dept_id='',
            del_flag=0,
        )
    )


def test_sample_handle_rejects_external_attachment_reference():
    from app.services.market_sample import update_storage_cert

    db = _session()
    _, _, sample = _cell_sample(db, 'attachment-reject')
    with pytest.raises(ValueError, match='附件'):
        update_storage_cert(
            db,
            sample_id=f"cell_storage:{sample.sample_code}",
            fields={'attachments': 'https://evil.test/a.jpg'},
            ctx=_ctx(),
            user_info=_user(),
        )


def _cell_sample(
    db,
    key,
    *,
    sample_code=None,
    sales='u1',
    manager='m1',
    region='r1',
    tenant='tenant-a',
    deleted=0,
    created=None,
    row_id=None,
):
    created = created or datetime(2026, 10, 1, 8, 0)
    code = f'C-{key}' if sample_code is None else sample_code
    protocol = BizCollectionProtocol(
        id=f'cp-{key}',
        protocol_no=f'CP-{key}',
        protocol_status='sxz',
        tenant_no=tenant,
        create_dept_id='dept-a',
        create_time=created,
        del_flag=deleted,
    )
    collection = BizProtocolCollectionInfo(
        id=f'cc-{key}',
        collection_protocol_id=protocol.id,
        region_id=region,
        sales_user_id=sales,
        region_manager_user_id=manager,
        sample_collect_time=created,
        tenant_no=tenant,
        create_dept_id='dept-a',
        del_flag=deleted,
    )
    party = BizProtocolPartyCustomer(
        id=f'pc-{key}',
        collection_protocol_id=protocol.id,
        mother_name=f'客户-{key}',
        tenant_no=tenant,
        create_dept_id='dept-a',
        del_flag=deleted,
    )
    sample = BizProtocolSample(
        id=row_id or f'cs-{key}',
        collection_protocol_id=protocol.id,
        sample_code=code,
        sample_status='ycj',
        tenant_no=tenant,
        create_dept_id='dept-a',
        create_time=created,
        del_flag=deleted,
    )
    db.add_all([protocol, collection, party, sample])
    db.commit()
    return protocol, collection, sample


def _research_sample(
    db,
    key,
    *,
    sample_code=None,
    sales='u1',
    manager='m1',
    region='r1',
    tenant='tenant-a',
    deleted=0,
    created=None,
):
    created = created or datetime(2026, 10, 1, 8, 0)
    code = f'R-{key}' if sample_code is None else sample_code
    donation = BizResearchDonation(
        id=f'rd-{key}',
        donation_no=f'RD-{key}',
        donation_status='sxz',
        tenant_no=tenant,
        create_dept_id='dept-a',
        create_time=created,
        del_flag=deleted,
    )
    collection = BizResearchDonationCollection(
        id=f'rc-{key}',
        research_donation_id=donation.id,
        region_id=region,
        sales_user_id=sales,
        region_manager_user_id=manager,
        sample_collect_time=created,
        tenant_no=tenant,
        create_dept_id='dept-a',
        del_flag=deleted,
    )
    party = BizResearchDonationParty(
        id=f'rp-{key}',
        research_donation_id=donation.id,
        mother_name=f'科研客户-{key}',
        tenant_no=tenant,
        create_dept_id='dept-a',
        del_flag=deleted,
    )
    sample = BizResearchDonationSample(
        id=f'rs-{key}',
        research_donation_id=donation.id,
        sample_code=code,
        sample_status='jcz',
        tenant_no=tenant,
        create_dept_id='dept-a',
        create_time=created,
        del_flag=deleted,
    )
    db.add_all([donation, collection, party, sample])
    db.commit()
    return donation, collection, sample


def _downstream(db, key, code, source, *, tenant='tenant-a', when=None, conclusion='tg'):
    when = when or datetime(2026, 10, 2, 8, 0)
    db.add_all(
        [
            BizSampleReception(
                id=f'rec-{key}',
                protocol_sample_id='stale-row-id',
                collection_protocol_id='stale-parent-id',
                reception_time=when,
                accept_conclusion='js',
                reception_source=source,
                sample_code=code,
                tenant_no=tenant,
                create_dept_id='dept-a',
                create_time=when,
                del_flag=0,
            ),
            BizQcWorkOrder(
                id=f'qc-{key}',
                work_order_no=f'QC-{key}',
                sample_code=code,
                reception_source=source,
                trigger_source='reception',
                work_order_status='ywc',
                conclusion=conclusion,
                tenant_no=tenant,
                create_dept_id='dept-a',
                create_time=when,
                update_time=when,
                del_flag=0,
            ),
            BizStorageReleaseReview(
                id=f'st-{key}',
                sample_code=code,
                reception_source=source,
                storage_status='yrk',
                storage_time=when,
                tenant_no=tenant,
                create_dept_id='dept-a',
                create_time=when,
                del_flag=0,
            ),
        ]
    )
    db.commit()


def test_sample_list_resolves_parent_scope_and_uses_global_stable_paging():
    from app.services.market_sample import list_samples

    db = _session()
    base = datetime(2026, 10, 10, 8, 0)
    _cell_sample(db, 'c-old', created=base)
    _research_sample(db, 'r-new', created=base + timedelta(hours=3))
    _cell_sample(db, 'c-mid', created=base + timedelta(hours=2))
    _research_sample(db, 'r-mid', created=base + timedelta(hours=1))
    _cell_sample(db, 'other-user', sales='u2', created=base + timedelta(hours=4))
    _cell_sample(db, 'cross-tenant', tenant='tenant-b', created=base + timedelta(hours=5))
    _research_sample(db, 'deleted', deleted=1, created=base + timedelta(hours=6))

    page1, total = list_samples(db, ctx=_ctx(), user_info=_user(), page=1, page_size=2)
    page2, total2 = list_samples(db, ctx=_ctx(), user_info=_user(), page=2, page_size=2)

    assert total == total2 == 4
    assert [item['sampleCode'] for item in page1] == ['R-r-new', 'C-c-mid']
    assert [item['sampleCode'] for item in page2] == ['R-r-mid', 'C-c-old']
    assert page1[0]['id'] == 'research_donation:R-r-new'
    assert page1[1]['id'] == 'cell_storage:C-c-mid'
    assert all('statusTone' in item and 'tagType' in item for item in page1)


def test_manager_and_dual_role_use_the_fixed_route_field_scope():
    from app.services.market_sample import list_samples

    db = _session()
    db.add(
        BizCollectionRegionStaff(
            id='staff-m1',
            region_id='r-managed',
            user_id='m1',
            is_leader=1,
            tenant_no='tenant-a',
            create_dept_id='',
            del_flag=0,
        )
    )
    db.commit()
    _cell_sample(db, 'managed', sales='u2', manager='other', region='r-managed')
    _cell_sample(db, 'assigned', sales='u2', manager='m1', region='r-other')
    _cell_sample(db, 'hidden', sales='m1', manager='other', region='r-other')

    manager_items, _ = list_samples(db, ctx=_ctx('manager', 'm1'), user_info=_user('m1'))
    specialist_items, _ = list_samples(
        db, ctx=_ctx('specialist', 'm1'), user_info=_user('m1')
    )

    assert {item['sampleCode'] for item in manager_items} == {'C-managed', 'C-assigned'}
    assert {item['sampleCode'] for item in specialist_items} == {'C-hidden'}


def test_detail_matches_downstream_only_by_tenant_code_and_source_and_latest():
    from app.services.market_sample import get_sample_detail

    db = _session()
    _cell_sample(db, 'same', sample_code='SAME', row_id='rebuilt-row-id')
    _research_sample(db, 'same', sample_code='SAME')
    old = datetime(2026, 10, 1, 8, 0)
    new = datetime(2026, 10, 2, 8, 0)
    _downstream(db, 'cell-old', 'SAME', 'cell_storage', when=old, conclusion='btg')
    _downstream(db, 'cell-new', 'SAME', 'cell_storage', when=new, conclusion='tg')
    _downstream(db, 'research', 'SAME', 'research_donation', when=new, conclusion='btg')
    _downstream(db, 'cross', 'SAME', 'cell_storage', tenant='tenant-b', when=new + timedelta(days=1), conclusion='btg')

    cell = get_sample_detail(
        db, sample_id='cell_storage:SAME', ctx=_ctx(), user_info=_user()
    )
    research = get_sample_detail(
        db, sample_id='research_donation:SAME', ctx=_ctx(), user_info=_user()
    )

    assert cell['card']['sampleRowId'] == 'rebuilt-row-id'
    assert cell['progress']['qc']['code'] == 'tg'
    assert research['progress']['qc']['code'] == 'btg'
    assert cell['progress']['reception']['recordId'] == 'rec-cell-new'
    assert cell['progress']['storage']['recordId'] == 'st-cell-new'
    assert cell['sections']


def test_detail_idor_and_malformed_ids_are_not_exposed():
    from app.services.market_sample import get_sample_detail

    db = _session()
    _cell_sample(db, 'other', sales='u2')
    _cell_sample(db, 'cross', tenant='tenant-b')

    for sample_id in ('cell_storage:C-other', 'cell_storage:C-cross', 'bad', 'evil:C-other'):
        with pytest.raises(HTTPException) as exc:
            get_sample_detail(db, sample_id=sample_id, ctx=_ctx(), user_info=_user())
        assert exc.value.status_code == 404


def test_empty_sample_code_has_basic_read_only_detail_without_downstream():
    from app.services.market_sample import get_sample_detail, list_samples

    db = _session()
    _cell_sample(db, 'blank', sample_code='')
    _downstream(db, 'unrelated', 'C-other', 'cell_storage')

    items, total = list_samples(db, ctx=_ctx(), user_info=_user())
    assert total == 1
    assert items[0]['id'] == 'cell_storage:@id:cs-blank'
    detail = get_sample_detail(db, sample_id=items[0]['id'], ctx=_ctx(), user_info=_user())
    assert detail['card']['sampleCode'] == ''
    assert detail['actions'] == []
    assert detail['progress']['reception']['code'] == 'none'


def test_cell_lightweight_storage_cert_and_insurance_writes_validate_handler():
    from app.services.market_sample import update_insurance, update_storage_cert

    db = _session()
    protocol, _, _ = _cell_sample(db, 'write')
    db.add_all(
        [
            SysUser(
                id='handler',
                user_no='handler',
                nick_name='办理人',
                email='',
                phone_no='',
                status='1',
                tenant_no='tenant-a',
                create_dept_id='',
                del_flag=0,
            ),
            BizCollectionRegionStaff(
                id='staff-handler-write',
                region_id='r1',
                user_id='handler',
                is_leader=0,
                tenant_no='tenant-a',
                create_dept_id='',
                del_flag=0,
            ),
            SysUser(
                id='disabled',
                user_no='disabled',
                nick_name='停用人',
                email='',
                phone_no='',
                status='0',
                tenant_no='tenant-a',
                create_dept_id='',
                del_flag=0,
            ),
        ]
    )
    db.commit()
    when = '2026-10-11 09:30:00'
    cert = update_storage_cert(
        db,
        sample_id='cell_storage:C-write',
        fields={
            'handleTime': when,
            'handlerUserId': 'handler',
            'attachments': 'a.jpg,b.jpg',
            'remark': '证书备注',
        },
        ctx=_ctx(),
        user_info=_user(),
    )
    insurance = update_insurance(
        db,
        sample_id='cell_storage:C-write',
        fields={
            'handleTime': when,
            'handlerUserId': 'handler',
            'attachments': 'i.jpg',
            'remark': '保险备注',
        },
        ctx=_ctx(),
        user_info=_user(),
    )

    cert_row = db.query(BizProtocolStorageCert).filter_by(collection_protocol_id=protocol.id).one()
    todo_row = db.query(BizProtocolFullmoonTodo).filter_by(collection_protocol_id=protocol.id).one()
    assert cert['handlerName'] == insurance['handlerName'] == '办理人'
    assert cert_row.cert_image_urls == 'a.jpg,b.jpg'
    assert cert_row.remark == '证书备注'
    assert todo_row.insurance_attachments == 'i.jpg'
    assert todo_row.handle_remark == '保险备注'

    with pytest.raises(ValueError, match='办理人员'):
        update_storage_cert(
            db,
            sample_id='cell_storage:C-write',
            fields={'handlerUserId': 'disabled'},
            ctx=_ctx(),
            user_info=_user(),
        )


def test_sample_form_options_return_only_enabled_staff_in_scoped_document_region():
    from app.services.market_sample import get_sample_form_options

    db = _session()
    _cell_sample(db, 'handler-options', region='r1')
    db.add_all(
        [
            SysUser(id='eligible', user_no='eligible', nick_name='可办理', email='', phone_no='', status='1', tenant_no='tenant-a', create_dept_id='', del_flag=0),
            SysUser(id='other-region', user_no='other-region', nick_name='其他地区', email='', phone_no='', status='1', tenant_no='tenant-a', create_dept_id='', del_flag=0),
            SysUser(id='disabled-staff', user_no='disabled-staff', nick_name='停用人员', email='', phone_no='', status='0', tenant_no='tenant-a', create_dept_id='', del_flag=0),
            SysUser(id='cross-tenant-staff', user_no='cross-tenant-staff', nick_name='跨租户', email='', phone_no='', status='1', tenant_no='tenant-b', create_dept_id='', del_flag=0),
            BizCollectionRegionStaff(id='region-eligible', region_id='r1', user_id='eligible', is_leader=0, tenant_no='tenant-a', create_dept_id='', del_flag=0),
            BizCollectionRegionStaff(id='region-other', region_id='r2', user_id='other-region', is_leader=0, tenant_no='tenant-a', create_dept_id='', del_flag=0),
            BizCollectionRegionStaff(id='region-disabled', region_id='r1', user_id='disabled-staff', is_leader=0, tenant_no='tenant-a', create_dept_id='', del_flag=0),
            BizCollectionRegionStaff(id='region-cross', region_id='r1', user_id='cross-tenant-staff', is_leader=0, tenant_no='tenant-b', create_dept_id='', del_flag=0),
        ]
    )
    db.commit()

    result = get_sample_form_options(
        db,
        sample_id='cell_storage:C-handler-options',
        ctx=_ctx(),
        user_info=_user(),
    )

    assert result == {'staff': [{'value': 'eligible', 'label': '可办理'}]}


def test_sample_detail_and_options_resolve_only_referenced_historical_handlers():
    from app.services.market_sample import get_sample_detail, get_sample_form_options

    db = _session()
    protocol, _, sample = _cell_sample(db, 'historical-handlers', region='r1')
    db.add_all(
        [
            SysUser(id='eligible', user_no='eligible-no', nick_name='当前可办理', email='', phone_no='', status='1', tenant_no='tenant-a', create_dept_id='', del_flag=0),
            SysUser(id='disabled-history', user_no='disabled-no', nick_name='历史停用人员', email='', phone_no='', status='0', tenant_no='tenant-a', create_dept_id='', del_flag=0),
            SysUser(id='deleted-history', user_no='deleted-no', nick_name='历史删除人员', email='', phone_no='', status='1', tenant_no='tenant-a', create_dept_id='', del_flag=1),
            SysUser(id='unrelated-history', user_no='unrelated-no', nick_name='无关历史人员', email='', phone_no='', status='0', tenant_no='tenant-a', create_dept_id='', del_flag=1),
            SysUser(id='cross-history', user_no='cross-no', nick_name='跨租户历史人员', email='', phone_no='', status='0', tenant_no='tenant-b', create_dept_id='', del_flag=1),
            BizCollectionRegionStaff(id='region-eligible-history', region_id='r1', user_id='eligible', is_leader=0, tenant_no='tenant-a', create_dept_id='', del_flag=0),
            BizProtocolStorageCert(id='cert-history', collection_protocol_id=protocol.id, handler_user_id='disabled-history', tenant_no='tenant-a', create_dept_id='', del_flag=0),
            BizProtocolFullmoonTodo(id='todo-history', collection_protocol_id=protocol.id, handler_user_id='deleted-history', tenant_no='tenant-a', create_dept_id='', del_flag=0),
            BizSampleReception(id='reception-history', protocol_sample_id=sample.id, collection_protocol_id=protocol.id, reception_time=datetime(2026, 10, 2, 8, 0), accept_conclusion='js', reception_source='cell_storage', sample_code=sample.sample_code, lab_storage_code='LAB-2026-001', tenant_no='tenant-a', create_dept_id='', del_flag=0),
        ]
    )
    db.commit()

    detail = get_sample_detail(
        db,
        sample_id='cell_storage:C-historical-handlers',
        ctx=_ctx(),
        user_info=_user(),
    )
    handles = next(
        section for section in detail['sections'] if section['key'] == 'market-handles'
    )
    basic = next(section for section in detail['sections'] if section['key'] == 'basic')
    storage_options = get_sample_form_options(
        db,
        sample_id='cell_storage:C-historical-handlers',
        action='storage-cert',
        ctx=_ctx(),
        user_info=_user(),
    )
    insurance_options = get_sample_form_options(
        db,
        sample_id='cell_storage:C-historical-handlers',
        action='insurance',
        ctx=_ctx(),
        user_info=_user(),
    )
    default_options = get_sample_form_options(
        db,
        sample_id='cell_storage:C-historical-handlers',
        ctx=_ctx(),
        user_info=_user(),
    )

    assert handles['storageCert']['handlerUserName'] == '历史停用人员'
    assert handles['insurance']['handlerUserName'] == '历史删除人员'
    assert {'label': '实验室入库编码', 'value': 'LAB-2026-001'} in basic['fields']
    assert {item['value']: item['label'] for item in storage_options['staff']} == {
        'eligible': '当前可办理',
        'disabled-history': '历史停用人员',
    }
    assert {item['value']: item['label'] for item in insurance_options['staff']} == {
        'eligible': '当前可办理',
        'deleted-history': '历史删除人员',
    }
    assert default_options == {
        'staff': [{'value': 'eligible', 'label': '当前可办理'}]
    }


def test_sample_handling_form_options_route_is_not_registered():
    from app.api import market_mobile

    paths = {route.path for route in market_mobile.router.routes}
    assert '/mini/market/specialist/samples/form-options' not in paths
    assert '/mini/market/manager/samples/form-options' not in paths


def test_sample_list_batches_referenced_handler_name_lookup():
    from app.services.market_sample import list_samples

    db = _session()
    db.add(
        SysUser(id='history-handler', user_no='history-no', nick_name='历史办理人', email='', phone_no='', status='0', tenant_no='tenant-a', create_dept_id='', del_flag=1)
    )
    for index in range(5):
        protocol, _, _ = _cell_sample(db, f'handler-batch-{index}')
        db.add(
            BizProtocolStorageCert(
                id=f'cert-handler-batch-{index}',
                collection_protocol_id=protocol.id,
                handler_user_id='history-handler',
                tenant_no='tenant-a',
                create_dept_id='',
                del_flag=0,
            )
        )
    db.commit()
    statements = []

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

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

    assert total == len(items) == 5
    assert {item['storageCert']['handlerUserName'] for item in items} == {'历史办理人'}
    assert sum('FROM sys_user' in statement for statement in statements) == 1


def test_sample_handler_must_be_enabled_staff_in_current_document_region():
    from app.services.market_sample import update_storage_cert

    db = _session()
    _cell_sample(db, 'handler-region', region='r1')
    db.add_all(
        [
            SysUser(id='wrong-region', user_no='wrong-region', nick_name='错区人员', email='', phone_no='', status='1', tenant_no='tenant-a', create_dept_id='', del_flag=0),
            BizCollectionRegionStaff(id='wrong-region-link', region_id='r2', user_id='wrong-region', is_leader=0, tenant_no='tenant-a', create_dept_id='', del_flag=0),
        ]
    )
    db.commit()

    with pytest.raises(ValueError, match='办理人员'):
        update_storage_cert(
            db,
            sample_id='cell_storage:C-handler-region',
            fields={'handleTime': '2026-10-11 09:30:00', 'handlerUserId': 'wrong-region'},
            ctx=_ctx(),
            user_info=_user(),
        )

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


def test_research_and_empty_code_samples_are_read_only_and_scope_is_rechecked():
    from app.services.market_sample import update_storage_cert

    db = _session()
    _research_sample(db, 'readonly')
    _, collection, _ = _cell_sample(db, 'moved')
    _cell_sample(db, 'blank', sample_code='')

    for sample_id in ('research_donation:R-readonly', 'cell_storage:@id:cs-blank'):
        with pytest.raises(ValueError, match='不支持'):
            update_storage_cert(
                db,
                sample_id=sample_id,
                fields={'remark': 'x'},
                ctx=_ctx(),
                user_info=_user(),
            )

    collection.sales_user_id = 'u2'
    db.commit()
    with pytest.raises(HTTPException) as exc:
        update_storage_cert(
            db,
            sample_id='cell_storage:C-moved',
            fields={'remark': 'x'},
            ctx=_ctx(),
            user_info=_user(),
        )
    assert exc.value.status_code == 404


def test_lightweight_dtos_forbid_status_qc_production_lab_and_sample_fields():
    from app.schemas.market_mobile import (
        MarketSampleInsuranceBody,
        MarketSampleStorageCertBody,
    )

    allowed = {
        'handleTime': '2026-10-11 09:30:00',
        'handlerUserId': 'u1',
        'attachments': 'a.jpg',
        'remark': 'ok',
    }
    MarketSampleStorageCertBody.model_validate(allowed)
    MarketSampleInsuranceBody.model_validate(allowed)

    for forbidden in (
        'sampleCode',
        'sampleStatus',
        'storageStatus',
        'qcConclusion',
        'productionResult',
        'labStorageCode',
        'makeStatus',
        'todoStatus',
        'tenantNo',
    ):
        with pytest.raises(ValidationError):
            MarketSampleStorageCertBody.model_validate({**allowed, forbidden: 'bad'})


def test_write_rolls_back_when_flush_fails(monkeypatch):
    from app.services.market_sample import update_storage_cert

    db = _session()
    _cell_sample(db, 'rollback')
    original_flush = db.flush

    def fail_flush(*args, **kwargs):
        original_flush(*args, **kwargs)
        raise RuntimeError('injected failure')

    monkeypatch.setattr(db, 'flush', fail_flush)
    with pytest.raises(RuntimeError, match='injected failure'):
        update_storage_cert(
            db,
            sample_id='cell_storage:C-rollback',
            fields={'remark': 'must rollback'},
            ctx=_ctx(),
            user_info=_user(),
        )
    monkeypatch.setattr(db, 'flush', original_flush)
    assert db.query(BizProtocolStorageCert).count() == 0


def test_list_batch_enrichment_has_bounded_query_count():
    from app.services.market_sample import list_samples

    db = _session()
    for index in range(8):
        source = 'cell_storage' if index % 2 == 0 else 'research_donation'
        if source == 'cell_storage':
            _cell_sample(db, f'q{index}')
            code = f'C-q{index}'
        else:
            _research_sample(db, f'q{index}')
            code = f'R-q{index}'
        _downstream(db, f'q{index}', code, source)

    statements = []

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

    event.listen(db.bind, 'before_cursor_execute', count_statement)
    try:
        items, total = list_samples(db, ctx=_ctx(), user_info=_user(), page_size=20)
    finally:
        event.remove(db.bind, 'before_cursor_execute', count_statement)

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


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

    paths = {route.path for route in router.routes}
    for kind in ('specialist', 'manager'):
        base = f'/mini/market/{kind}/samples'
        assert not any(path.startswith(base) for path in paths)


def test_global_identity_query_deduplicates_legacy_duplicate_child_rows():
    from app.services.market_sample import list_samples

    db = _session()
    protocol, _, _ = _cell_sample(db, 'duplicate-child')
    db.add(
        BizProtocolCollectionInfo(
            id='cc-duplicate-child-legacy',
            collection_protocol_id=protocol.id,
            region_id='r1',
            sales_user_id='u1',
            tenant_no='tenant-a',
            create_dept_id='',
            del_flag=0,
        )
    )
    db.commit()

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

    assert total == len(items) == 1


def test_partial_handle_update_preserves_omitted_lightweight_fields():
    from app.services.market_sample import update_storage_cert

    db = _session()
    _cell_sample(db, 'partial')
    db.add(
        SysUser(
            id='handler',
            user_no='handler',
            nick_name='办理人',
            email='',
            phone_no='',
            status='1',
            tenant_no='tenant-a',
            create_dept_id='',
            del_flag=0,
        )
    )
    _assign_region_staff(db, 'handler', 'partial')
    db.commit()
    update_storage_cert(
        db,
        sample_id='cell_storage:C-partial',
        fields={
            'handleTime': '2026-10-11 09:30:00',
            'handlerUserId': 'handler',
            'attachments': 'old.jpg',
            'remark': '旧备注',
        },
        ctx=_ctx(),
        user_info=_user(),
    )

    update_storage_cert(
        db,
        sample_id='cell_storage:C-partial',
        fields={'remark': '新备注'},
        ctx=_ctx(),
        user_info=_user(),
    )
    row = db.query(BizProtocolStorageCert).one()
    assert _fmt(row.handle_time) == '2026-10-11 09:30:00'
    assert row.handler_user_id == 'handler'
    assert row.cert_image_urls == 'old.jpg'
    assert row.remark == '新备注'


def _fmt(value):
    return value.strftime('%Y-%m-%d %H:%M:%S') if value else ''


def test_materialized_collection_row_itself_must_be_inside_fixed_scope():
    from app.services.market_sample import list_samples

    db = _session()
    protocol, visible, _ = _cell_sample(db, 'mixed-collection', region='visible-region')
    db.add(
        BizProtocolCollectionInfo(
            id='cc-mixed-hidden',
            collection_protocol_id=protocol.id,
            region_id='hidden-region',
            sales_user_id='u2',
            tenant_no='tenant-a',
            create_dept_id='',
            del_flag=0,
        )
    )
    db.commit()

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

    assert total == 1
    assert items[0]['regionId'] == visible.region_id


def test_progress_uses_normalized_codes_and_explicit_semantic_tones():
    from app.services.market_sample import get_sample_detail

    db = _session()
    _cell_sample(db, 'tone')
    detail = get_sample_detail(
        db, sample_id='cell_storage:C-tone', ctx=_ctx(), user_info=_user()
    )

    assert detail['progress']['collection'] == {
        'code': 'collected',
        'label': '已采集',
        'statusTone': 'success',
        'tagType': 'success',
    }
    assert detail['progress']['reception']['statusTone'] == 'info'
    assert detail['progress']['qc']['statusTone'] == 'warning'


def test_legacy_storage_cert_and_insurance_statuses_are_read_as_handled():
    from app.services.market_sample import get_sample_detail

    db = _session()
    protocol, _, _ = _cell_sample(db, 'legacy-handled')
    protocol_alt, _, _ = _cell_sample(db, 'legacy-handled-alt')
    db.add_all(
        [
            BizProtocolStorageCert(
                id='cert-legacy',
                collection_protocol_id=protocol.id,
                make_status='0',
                make_date='20261001',
                tenant_no='tenant-a',
                create_dept_id='',
                del_flag=0,
            ),
            BizProtocolFullmoonTodo(
                id='todo-legacy',
                collection_protocol_id=protocol.id,
                todo_status='1',
                tenant_no='tenant-a',
                create_dept_id='',
                del_flag=0,
            ),
            BizProtocolStorageCert(
                id='cert-legacy-status',
                collection_protocol_id=protocol_alt.id,
                make_status='1',
                tenant_no='tenant-a',
                create_dept_id='',
                del_flag=0,
            ),
            BizProtocolFullmoonTodo(
                id='todo-legacy-date',
                collection_protocol_id=protocol_alt.id,
                todo_status='0',
                handle_date='20261002',
                tenant_no='tenant-a',
                create_dept_id='',
                del_flag=0,
            ),
        ]
    )
    db.commit()

    detail = get_sample_detail(
        db,
        sample_id='cell_storage:C-legacy-handled',
        ctx=_ctx(),
        user_info=_user(),
    )

    section = next(item for item in detail['sections'] if item['key'] == 'market-handles')
    assert section['storageCert']['code'] == 'handled'
    assert section['storageCert']['statusTone'] == 'success'
    assert section['insurance']['code'] == 'handled'
    assert section['insurance']['statusTone'] == 'success'
    alt = get_sample_detail(
        db,
        sample_id='cell_storage:C-legacy-handled-alt',
        ctx=_ctx(),
        user_info=_user(),
    )
    alt_section = next(
        item for item in alt['sections'] if item['key'] == 'market-handles'
    )
    assert alt_section['storageCert']['code'] == 'handled'
    assert alt_section['insurance']['code'] == 'handled'


def test_setting_handle_time_synchronizes_legacy_status_and_date_for_both_actions():
    from app.services.market_sample import update_insurance, update_storage_cert

    db = _session()
    _cell_sample(db, 'sync-set')
    db.add(
        SysUser(
            id='handler',
            user_no='handler',
            nick_name='办理人',
            email='',
            phone_no='',
            status='1',
            tenant_no='tenant-a',
            create_dept_id='',
            del_flag=0,
        )
    )
    _assign_region_staff(db, 'handler', 'sync-set')
    db.commit()
    fields = {
        'handleTime': '2026-10-11 09:30:00',
        'handlerUserId': 'handler',
        'attachments': 'keep.jpg',
        'remark': '保留备注',
    }

    update_storage_cert(
        db,
        sample_id='cell_storage:C-sync-set',
        fields=fields,
        ctx=_ctx(),
        user_info=_user(),
    )
    update_insurance(
        db,
        sample_id='cell_storage:C-sync-set',
        fields={**fields, 'attachments': 'keep-insurance.jpg'},
        ctx=_ctx(),
        user_info=_user(),
    )

    cert = db.query(BizProtocolStorageCert).one()
    todo = db.query(BizProtocolFullmoonTodo).one()
    assert (cert.make_status, cert.make_date) == ('1', '20261011')
    assert (todo.todo_status, todo.handle_date) == ('1', '20261011')


def test_clearing_handle_time_clears_handler_and_legacy_status_but_preserves_content():
    from app.services.market_sample import update_insurance, update_storage_cert

    db = _session()
    _cell_sample(db, 'sync-clear')
    db.add(
        SysUser(
            id='handler',
            user_no='handler',
            nick_name='办理人',
            email='',
            phone_no='',
            status='1',
            tenant_no='tenant-a',
            create_dept_id='',
            del_flag=0,
        )
    )
    _assign_region_staff(db, 'handler', 'sync-clear')
    db.commit()
    initial = {
        'handleTime': '2026-10-11 09:30:00',
        'handlerUserId': 'handler',
        'attachments': 'keep.jpg',
        'remark': '保留备注',
    }
    for updater, attachment in (
        (update_storage_cert, 'keep.jpg'),
        (update_insurance, 'keep-insurance.jpg'),
    ):
        updater(
            db,
            sample_id='cell_storage:C-sync-clear',
            fields={**initial, 'attachments': attachment},
            ctx=_ctx(),
            user_info=_user(),
        )
        updater(
            db,
            sample_id='cell_storage:C-sync-clear',
            fields={'handleTime': None, 'handlerUserId': 'handler'},
            ctx=_ctx(),
            user_info=_user(),
        )

    cert = db.query(BizProtocolStorageCert).one()
    todo = db.query(BizProtocolFullmoonTodo).one()
    assert (cert.handle_time, cert.handler_user_id, cert.make_status, cert.make_date) == (
        None,
        None,
        '0',
        None,
    )
    assert (todo.handle_time, todo.handler_user_id, todo.todo_status, todo.handle_date) == (
        None,
        None,
        '0',
        None,
    )
    assert (cert.cert_image_urls, cert.remark) == ('keep.jpg', '保留备注')
    assert (todo.insurance_attachments, todo.handle_remark) == (
        'keep-insurance.jpg',
        '保留备注',
    )


def test_partial_insurance_update_preserves_omitted_lightweight_fields():
    from app.services.market_sample import update_insurance

    db = _session()
    _cell_sample(db, 'partial-insurance')
    db.add(
        SysUser(
            id='handler',
            user_no='handler',
            nick_name='办理人',
            email='',
            phone_no='',
            status='1',
            tenant_no='tenant-a',
            create_dept_id='',
            del_flag=0,
        )
    )
    _assign_region_staff(db, 'handler', 'partial-insurance')
    db.commit()
    update_insurance(
        db,
        sample_id='cell_storage:C-partial-insurance',
        fields={
            'handleTime': '2026-10-11 09:30:00',
            'handlerUserId': 'handler',
            'attachments': 'old.jpg',
            'remark': '旧备注',
        },
        ctx=_ctx(),
        user_info=_user(),
    )

    update_insurance(
        db,
        sample_id='cell_storage:C-partial-insurance',
        fields={'remark': '新备注'},
        ctx=_ctx(),
        user_info=_user(),
    )
    row = db.query(BizProtocolFullmoonTodo).one()
    assert _fmt(row.handle_time) == '2026-10-11 09:30:00'
    assert row.handler_user_id == 'handler'
    assert row.insurance_attachments == 'old.jpg'
    assert row.handle_remark == '新备注'


def test_partial_update_preserves_legacy_handled_fields_when_time_is_omitted():
    from app.services.market_sample import update_insurance, update_storage_cert

    db = _session()
    protocol, _, _ = _cell_sample(db, 'partial-legacy')
    db.add_all(
        [
            SysUser(
                id='handler',
                user_no='handler',
                nick_name='办理人',
                email='',
                phone_no='',
                status='1',
                tenant_no='tenant-a',
                create_dept_id='',
                del_flag=0,
            ),
            BizProtocolStorageCert(
                id='cert-partial-legacy',
                collection_protocol_id=protocol.id,
                make_status='1',
                make_date='20261001',
                handler_user_id='handler',
                cert_image_urls='cert-old.jpg',
                tenant_no='tenant-a',
                create_dept_id='',
                del_flag=0,
            ),
            BizProtocolFullmoonTodo(
                id='todo-partial-legacy',
                collection_protocol_id=protocol.id,
                todo_status='1',
                handle_date='20261002',
                handler_user_id='handler',
                insurance_attachments='insurance-old.jpg',
                tenant_no='tenant-a',
                create_dept_id='',
                del_flag=0,
            ),
        ]
    )
    db.commit()

    update_storage_cert(
        db,
        sample_id='cell_storage:C-partial-legacy',
        fields={'remark': '证书新备注'},
        ctx=_ctx(),
        user_info=_user(),
    )
    update_insurance(
        db,
        sample_id='cell_storage:C-partial-legacy',
        fields={'remark': '保险新备注'},
        ctx=_ctx(),
        user_info=_user(),
    )

    cert = db.query(BizProtocolStorageCert).one()
    todo = db.query(BizProtocolFullmoonTodo).one()
    assert (cert.make_status, cert.make_date, cert.handler_user_id) == (
        '1',
        '20261001',
        'handler',
    )
    assert cert.cert_image_urls == 'cert-old.jpg'
    assert (todo.todo_status, todo.handle_date, todo.handler_user_id) == (
        '1',
        '20261002',
        'handler',
    )
    assert todo.insurance_attachments == 'insurance-old.jpg'


def test_handler_only_update_changes_legacy_handled_records_without_clearing_status():
    from app.services.market_sample import update_insurance, update_storage_cert

    db = _session()
    protocol, _, _ = _cell_sample(db, 'handler-only-legacy')
    db.add_all(
        [
            SysUser(
                id='new-handler',
                user_no='new-handler',
                nick_name='新办理人',
                email='',
                phone_no='',
                status='1',
                tenant_no='tenant-a',
                create_dept_id='',
                del_flag=0,
            ),
            BizProtocolStorageCert(
                id='cert-handler-only',
                collection_protocol_id=protocol.id,
                make_status='1',
                make_date='20261001',
                tenant_no='tenant-a',
                create_dept_id='',
                del_flag=0,
            ),
            BizProtocolFullmoonTodo(
                id='todo-handler-only',
                collection_protocol_id=protocol.id,
                todo_status='0',
                handle_date='20261002',
                tenant_no='tenant-a',
                create_dept_id='',
                del_flag=0,
            ),
        ]
    )
    _assign_region_staff(db, 'new-handler', 'handler-only-legacy')
    db.commit()

    for updater in (update_storage_cert, update_insurance):
        result = updater(
            db,
            sample_id='cell_storage:C-handler-only-legacy',
            fields={'handlerUserId': 'new-handler'},
            ctx=_ctx(),
            user_info=_user(),
        )
        assert result['handlerName'] == '新办理人'

    cert = db.query(BizProtocolStorageCert).one()
    todo = db.query(BizProtocolFullmoonTodo).one()
    assert (cert.handler_user_id, cert.make_status, cert.make_date) == (
        'new-handler',
        '1',
        '20261001',
    )
    assert (todo.handler_user_id, todo.todo_status, todo.handle_date) == (
        'new-handler',
        '0',
        '20261002',
    )


def test_handler_only_update_rejects_truly_unhandled_cert_and_insurance():
    from app.services.market_sample import update_insurance, update_storage_cert

    db = _session()
    protocol, _, _ = _cell_sample(db, 'handler-only-unhandled')
    db.add_all(
        [
            SysUser(
                id='handler',
                user_no='handler',
                nick_name='办理人',
                email='',
                phone_no='',
                status='1',
                tenant_no='tenant-a',
                create_dept_id='',
                del_flag=0,
            ),
            BizProtocolStorageCert(
                id='cert-unhandled',
                collection_protocol_id=protocol.id,
                make_status='0',
                tenant_no='tenant-a',
                create_dept_id='',
                del_flag=0,
            ),
            BizProtocolFullmoonTodo(
                id='todo-unhandled',
                collection_protocol_id=protocol.id,
                todo_status='0',
                tenant_no='tenant-a',
                create_dept_id='',
                del_flag=0,
            ),
        ]
    )
    db.commit()

    for updater in (update_storage_cert, update_insurance):
        with pytest.raises(ValueError, match='未办理'):
            updater(
                db,
                sample_id='cell_storage:C-handler-only-unhandled',
                fields={'handlerUserId': 'handler'},
                ctx=_ctx(),
                user_info=_user(),
            )


def test_historical_disabled_handler_does_not_block_content_only_updates():
    from app.services.market_sample import update_insurance, update_storage_cert

    db = _session()
    protocol, _, _ = _cell_sample(db, 'disabled-history')
    db.add_all(
        [
            SysUser(
                id='disabled-history-handler',
                user_no='disabled-history-handler',
                nick_name='历史办理人',
                email='',
                phone_no='',
                status='0',
                tenant_no='tenant-a',
                create_dept_id='',
                del_flag=1,
            ),
            BizProtocolStorageCert(
                id='cert-disabled-history',
                collection_protocol_id=protocol.id,
                handle_time=datetime(2026, 10, 1, 8, 0),
                handler_user_id='disabled-history-handler',
                make_status='1',
                make_date='20261001',
                tenant_no='tenant-a',
                create_dept_id='',
                del_flag=0,
            ),
            BizProtocolFullmoonTodo(
                id='todo-disabled-history',
                collection_protocol_id=protocol.id,
                handle_time=datetime(2026, 10, 2, 8, 0),
                handler_user_id='disabled-history-handler',
                todo_status='1',
                handle_date='20261002',
                tenant_no='tenant-a',
                create_dept_id='',
                del_flag=0,
            ),
        ]
    )
    db.commit()

    update_storage_cert(
        db,
        sample_id='cell_storage:C-disabled-history',
        fields={'attachments': 'cert-new.jpg', 'remark': '证书新备注'},
        ctx=_ctx(),
        user_info=_user(),
    )
    update_insurance(
        db,
        sample_id='cell_storage:C-disabled-history',
        fields={'attachments': 'insurance-new.jpg', 'remark': '保险新备注'},
        ctx=_ctx(),
        user_info=_user(),
    )

    cert = db.query(BizProtocolStorageCert).one()
    todo = db.query(BizProtocolFullmoonTodo).one()
    assert cert.handler_user_id == todo.handler_user_id == 'disabled-history-handler'
    assert (cert.cert_image_urls, cert.remark) == ('cert-new.jpg', '证书新备注')
    assert (todo.insurance_attachments, todo.handle_remark) == (
        'insurance-new.jpg',
        '保险新备注',
    )


@pytest.mark.parametrize('handler_state', ['disabled', 'deleted', 'out-of-region'])
@pytest.mark.parametrize('kind', ['storage-cert', 'insurance'])
def test_full_unchanged_historical_handler_payload_allows_content_update(
    handler_state, kind
):
    from app.services.market_sample import update_insurance, update_storage_cert

    db = _session()
    protocol, _, _ = _cell_sample(db, f'historical-full-{handler_state}-{kind}')
    handler_id = f'historical-{handler_state}-{kind}'
    handle_time = datetime(2026, 10, 1, 8, 0)
    user = SysUser(
        id=handler_id,
        user_no=handler_id,
        nick_name='历史办理人',
        email='',
        phone_no='',
        status='0' if handler_state == 'disabled' else '1',
        tenant_no='tenant-a',
        create_dept_id='',
        del_flag=1 if handler_state == 'deleted' else 0,
    )
    staff = BizCollectionRegionStaff(
        id=f'staff-{handler_id}',
        region_id='r2' if handler_state == 'out-of-region' else 'r1',
        user_id=handler_id,
        is_leader=0,
        tenant_no='tenant-a',
        create_dept_id='',
        del_flag=0,
    )
    db.add_all([user, staff])
    if kind == 'storage-cert':
        db.add(
            BizProtocolStorageCert(
                id=f'cert-{handler_id}',
                collection_protocol_id=protocol.id,
                handle_time=handle_time,
                handler_user_id=handler_id,
                cert_image_urls='cert-old.jpg',
                make_status='1',
                tenant_no='tenant-a',
                create_dept_id='',
                del_flag=0,
            )
        )
        updater = update_storage_cert
        attachments = 'cert-new.jpg'
    else:
        db.add(
            BizProtocolFullmoonTodo(
                id=f'todo-{handler_id}',
                collection_protocol_id=protocol.id,
                handle_time=handle_time,
                handler_user_id=handler_id,
                insurance_attachments='insurance-old.jpg',
                todo_status='1',
                tenant_no='tenant-a',
                create_dept_id='',
                del_flag=0,
            )
        )
        updater = update_insurance
        attachments = 'insurance-new.jpg'
    db.commit()

    result = updater(
        db,
        sample_id=f'cell_storage:C-historical-full-{handler_state}-{kind}',
        fields={
            'handleTime': '2026-10-01 08:00:00',
            'handlerUserId': handler_id,
            'attachments': attachments,
            'remark': '只改内容',
        },
        ctx=_ctx(),
        user_info=_user(),
    )

    assert result['handlerUserId'] == handler_id
    assert result['handleTime'] == '2026-10-01 08:00:00'
    assert result['attachments'] == attachments
    assert result['remark'] == '只改内容'


@pytest.mark.parametrize('kind', ['storage-cert', 'insurance'])
def test_changed_handle_time_revalidates_historical_ineligible_handler(kind):
    from app.services.market_sample import update_insurance, update_storage_cert

    db = _session()
    protocol, _, _ = _cell_sample(db, f'historical-time-change-{kind}')
    handler_id = f'disabled-time-change-{kind}'
    db.add(
        SysUser(
            id=handler_id,
            user_no=handler_id,
            nick_name='停用历史办理人',
            email='',
            phone_no='',
            status='0',
            tenant_no='tenant-a',
            create_dept_id='',
            del_flag=0,
        )
    )
    if kind == 'storage-cert':
        row = BizProtocolStorageCert(
            id=f'cert-{handler_id}',
            collection_protocol_id=protocol.id,
            handle_time=datetime(2026, 10, 1, 8, 0),
            handler_user_id=handler_id,
            make_status='1',
            tenant_no='tenant-a',
            create_dept_id='',
            del_flag=0,
        )
        updater = update_storage_cert
    else:
        row = BizProtocolFullmoonTodo(
            id=f'todo-{handler_id}',
            collection_protocol_id=protocol.id,
            handle_time=datetime(2026, 10, 1, 8, 0),
            handler_user_id=handler_id,
            todo_status='1',
            tenant_no='tenant-a',
            create_dept_id='',
            del_flag=0,
        )
        updater = update_insurance
    db.add(row)
    db.commit()

    with pytest.raises(ValueError, match='办理人员'):
        updater(
            db,
            sample_id=f'cell_storage:C-historical-time-change-{kind}',
            fields={
                'handleTime': '2026-10-02 08:00:00',
                'handlerUserId': handler_id,
                'remark': '不得写入',
            },
            ctx=_ctx(),
            user_info=_user(),
        )

    db.expire_all()
    assert row.handle_time == datetime(2026, 10, 1, 8, 0)


def test_first_handle_parent_lock_compiles_to_mysql_for_update():
    from sqlalchemy.dialects import mysql

    from app.services.market_sample import _cell_parent_lock_query

    db = _session()
    query = _cell_parent_lock_query(db, 'cp-lock', _ctx())
    sql = str(
        query.statement.compile(
            dialect=mysql.dialect(), compile_kwargs={'literal_binds': True}
        )
    )

    assert 'FOR UPDATE' in sql
    assert "biz_collection_protocol.id = 'cp-lock'" in sql
    assert "biz_collection_protocol.tenant_no = 'tenant-a'" in sql
    assert 'biz_collection_protocol.del_flag = 0' in sql


def test_repeated_first_handle_requests_reuse_single_child_row():
    from app.services.market_sample import update_insurance, update_storage_cert

    db = _session()
    _cell_sample(db, 'single-child')
    for updater in (update_storage_cert, update_insurance):
        updater(
            db,
            sample_id='cell_storage:C-single-child',
            fields={'remark': 'first'},
            ctx=_ctx(),
            user_info=_user(),
        )
        updater(
            db,
            sample_id='cell_storage:C-single-child',
            fields={'remark': 'second'},
            ctx=_ctx(),
            user_info=_user(),
        )

    assert db.query(BizProtocolStorageCert).count() == 1
    assert db.query(BizProtocolFullmoonTodo).count() == 1
    assert db.query(BizProtocolStorageCert).one().remark == 'second'
    assert db.query(BizProtocolFullmoonTodo).one().handle_remark == 'second'


def test_voided_progress_uses_danger_tone():
    from app.services.market_sample import get_sample_detail

    db = _session()
    _, _, sample = _cell_sample(db, 'voided-tone')
    sample.sample_status = 'yzf'
    db.commit()

    detail = get_sample_detail(
        db,
        sample_id='cell_storage:C-voided-tone',
        ctx=_ctx(),
        user_info=_user(),
    )

    assert detail['progress']['collection']['code'] == 'voided'
    assert detail['progress']['collection']['statusTone'] == 'danger'
    assert detail['progress']['collection']['tagType'] == 'danger'


def test_source_qualified_whitespace_token_returns_uniform_404():
    from app.services.market_sample import get_sample_detail

    db = _session()
    for sample_id in ('cell_storage:   ', 'research_donation:\t'):
        with pytest.raises(HTTPException) as exc:
            get_sample_detail(
                db, sample_id=sample_id, ctx=_ctx(), user_info=_user()
            )
        assert exc.value.status_code == 404
        assert exc.value.detail == '数据不存在或无权访问'


def test_duplicate_active_children_use_same_newest_fields_in_list_and_detail():
    from app.services.market_sample import get_sample_detail, list_samples

    db = _session()
    protocol, collection, _ = _cell_sample(db, 'deterministic')
    collection.create_time = datetime(2026, 10, 1, 8, 0)
    original_party = db.query(BizProtocolPartyCustomer).filter_by(
        collection_protocol_id=protocol.id
    ).one()
    original_party.create_time = datetime(2026, 10, 1, 8, 0)
    db.add_all(
        [
            BizProtocolCollectionInfo(
                id='cc-deterministic-new',
                collection_protocol_id=protocol.id,
                region_id='newest-region',
                sales_user_id='u1',
                tenant_no='tenant-a',
                create_dept_id='',
                create_time=datetime(2026, 10, 2, 8, 0),
                del_flag=0,
            ),
            BizProtocolPartyCustomer(
                id='pc-deterministic-new',
                collection_protocol_id=protocol.id,
                mother_name='最新客户',
                tenant_no='tenant-a',
                create_dept_id='',
                create_time=datetime(2026, 10, 2, 8, 0),
                del_flag=0,
            ),
        ]
    )
    db.commit()

    items, total = list_samples(db, ctx=_ctx(), user_info=_user())
    detail = get_sample_detail(
        db,
        sample_id='cell_storage:C-deterministic',
        ctx=_ctx(),
        user_info=_user(),
    )

    assert total == 1
    assert items[0]['regionId'] == detail['card']['regionId'] == 'newest-region'
    assert items[0]['customerName'] == detail['card']['customerName'] == '最新客户'


def test_write_reasserts_current_scope_after_parent_lock(monkeypatch):
    from app.services import market_sample

    db = _session()
    _, collection, _ = _cell_sample(db, 'scope-race')
    original_find = market_sample._find_sample
    calls = []

    def move_out_of_scope_after_initial_read(*args, **kwargs):
        calls.append(bool(kwargs.get('for_update')))
        item = original_find(*args, **kwargs)
        if len(calls) == 1:
            collection.sales_user_id = 'u2'
            db.commit()
        return item

    monkeypatch.setattr(
        market_sample, '_find_sample', move_out_of_scope_after_initial_read
    )

    with pytest.raises(HTTPException) as exc:
        market_sample.update_storage_cert(
            db,
            sample_id='cell_storage:C-scope-race',
            fields={'remark': 'must not write'},
            ctx=_ctx(),
            user_info=_user(),
        )

    assert exc.value.status_code == 404
    assert calls == [False, True]
    assert not db.in_transaction()
    assert db.query(BizProtocolStorageCert).count() == 0


def test_current_sample_and_child_reads_compile_as_mysql_locking_reads():
    from sqlalchemy.dialects import mysql

    from app.services.market_sample import (
        _cell_sample_lookup_query,
        _handle_child_current_query,
    )

    db = _session()
    sample_query = _cell_sample_lookup_query(
        db,
        token='C-locking',
        by_row_id=False,
        ctx=_ctx(),
        user_info=_user(),
        for_update=True,
    )
    cert_query = _handle_child_current_query(
        db, BizProtocolStorageCert, 'cp-locking', _ctx()
    )
    todo_query = _handle_child_current_query(
        db, BizProtocolFullmoonTodo, 'cp-locking', _ctx()
    )

    def mysql_sql(query):
        return str(
            query.statement.compile(
                dialect=mysql.dialect(), compile_kwargs={'literal_binds': True}
            )
        )

    sample_sql = mysql_sql(sample_query)
    cert_sql = mysql_sql(cert_query)
    todo_sql = mysql_sql(todo_query)
    assert 'FOR UPDATE' in sample_sql
    assert 'FOR UPDATE' in cert_sql
    assert 'FOR UPDATE' in todo_sql
    assert 'ORDER BY biz_protocol_storage_cert.update_time DESC' in cert_sql
    assert 'ORDER BY biz_protocol_fullmoon_todo.update_time DESC' in todo_sql


def test_write_lock_order_is_parent_then_current_sample_then_child(monkeypatch):
    from app.services import market_sample

    db = _session()
    _cell_sample(db, 'lock-order')
    events = []
    original_parent_lock = market_sample._lock_cell_parent
    original_find = market_sample._find_sample
    original_child_query = market_sample._handle_child_current_query

    def record_parent(*args, **kwargs):
        events.append('parent')
        return original_parent_lock(*args, **kwargs)

    def record_find(*args, **kwargs):
        events.append('sample-current' if kwargs.get('for_update') else 'sample-initial')
        return original_find(*args, **kwargs)

    def record_child(*args, **kwargs):
        events.append('child')
        return original_child_query(*args, **kwargs)

    monkeypatch.setattr(market_sample, '_lock_cell_parent', record_parent)
    monkeypatch.setattr(market_sample, '_find_sample', record_find)
    monkeypatch.setattr(
        market_sample, '_handle_child_current_query', record_child
    )

    market_sample.update_storage_cert(
        db,
        sample_id='cell_storage:C-lock-order',
        fields={'remark': 'locked'},
        ctx=_ctx(),
        user_info=_user(),
    )

    assert events == ['sample-initial', 'parent', 'sample-current', 'child']