"""市场移动端运费批次：作用域、脱敏与原子写规则。"""

from datetime import datetime, timedelta
from decimal import Decimal
from pathlib import Path

import pytest
from fastapi import HTTPException
from pydantic import ValidationError
from sqlalchemy import create_engine
from sqlalchemy.dialects import mysql
from sqlalchemy.orm import sessionmaker

from app.models.base import Base
from app.models.models import (
    BizCellSampleType, BizCollectionProtocol, BizCollectionRegionStaff, BizProtocolCollectionInfo,
    BizProtocolPartyCustomer, BizProtocolSample, BizSampleFreight,
    BizSampleFreightItem, SysDictData, SysDictType, SysPost, SysPostRole,
    SysDept, SysRole, SysRoleDataScopeDetail, SysUser, SysUserPost, SysUserRole,
)
from app.models.research_donation import (
    BizResearchDonation, BizResearchDonationCollection,
    BizResearchDonationParty, BizResearchDonationSample,
)
from app.services.market_field_scope import MarketFieldContext

TABLES = [
    BizCollectionProtocol.__table__, BizProtocolCollectionInfo.__table__,
    BizProtocolPartyCustomer.__table__, BizProtocolSample.__table__,
    BizResearchDonation.__table__, BizResearchDonationCollection.__table__,
    BizResearchDonationParty.__table__, BizResearchDonationSample.__table__,
    BizSampleFreight.__table__, BizSampleFreightItem.__table__,
    BizCellSampleType.__table__,
    BizCollectionRegionStaff.__table__, SysDictType.__table__,
    SysDictData.__table__, SysUser.__table__, SysRole.__table__,
    SysRoleDataScopeDetail.__table__, SysUserRole.__table__, SysPost.__table__,
    SysPostRole.__table__, SysUserPost.__table__, SysDept.__table__,
]


def _session():
    engine = create_engine('sqlite:///:memory:')
    Base.metadata.create_all(engine, tables=TABLES)
    db = sessionmaker(bind=engine)()
    now = datetime(2026, 10, 11, 8, 0)
    db.add_all([
        SysDictType(id='dt', dict_name='运输方式', dict_type='sample_transport_mode', tenant_no='tenant-a', create_dept_id='', create_time=now, update_time=now, del_flag=0),
        SysDictData(id='dd', dict_label='顺丰', dict_value='sf', dict_type='sample_transport_mode', tenant_no='tenant-a', create_dept_id='', create_time=now, update_time=now, del_flag=0),
        SysUser(id='u1', dept_id='', user_no='u1', nick_name='专员', password='', tenant_no='tenant-a', create_dept_id='', create_time=now, update_time=now, del_flag=0),
        SysUser(id='m1', dept_id='', user_no='m1', nick_name='经理', password='', tenant_no='tenant-a', create_dept_id='', create_time=now, update_time=now, del_flag=0),
        SysRole(id='fixture-super-role', role_name='超级管理员', data_scope='1', status='1', tenant_no='tenant-a', create_dept_id='', create_time=now, update_time=now, del_flag=0),
        SysUserRole(id='fixture-super-u1', user_id='u1', role_id='fixture-super-role', tenant_no='tenant-a', create_dept_id='', create_time=now, update_time=now, del_flag=0),
        SysUserRole(id='fixture-super-m1', user_id='m1', role_id='fixture-super-role', tenant_no='tenant-a', create_dept_id='', create_time=now, update_time=now, del_flag=0),
    ])
    db.commit()
    return db


def _ctx(kind='specialist', user_id='u1'):
    return MarketFieldContext('tenant-a', user_id, kind)


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


def _sample(db, key, *, source='cell_storage', sales='u1', manager='m1', region='r1', tenant='tenant-a', deleted=0, created=None):
    created = created or datetime(2026, 10, 10, 8, 0)
    if source == 'cell_storage':
        parent = BizCollectionProtocol(id=f'cp-{key}', protocol_no=f'CP-{key}', tenant_no=tenant, create_dept_id='', create_time=created, del_flag=deleted)
        collection = BizProtocolCollectionInfo(id=f'cc-{key}', collection_protocol_id=parent.id, region_id=region, sales_user_id=sales, region_manager_user_id=manager, tenant_no=tenant, create_dept_id='', create_time=created, del_flag=deleted)
        party = BizProtocolPartyCustomer(id=f'pc-{key}', collection_protocol_id=parent.id, mother_name=f'客户-{key}', tenant_no=tenant, create_dept_id='', create_time=created, del_flag=deleted)
        sample = BizProtocolSample(id=f'cs-{key}', collection_protocol_id=parent.id, sample_code=f'C-{key}', tenant_no=tenant, create_dept_id='', create_time=created, del_flag=deleted)
    else:
        parent = BizResearchDonation(id=f'rd-{key}', donation_no=f'RD-{key}', tenant_no=tenant, create_dept_id='', create_time=created, del_flag=deleted)
        collection = BizResearchDonationCollection(id=f'rc-{key}', research_donation_id=parent.id, region_id=region, sales_user_id=sales, region_manager_user_id=manager, tenant_no=tenant, create_dept_id='', create_time=created, del_flag=deleted)
        party = BizResearchDonationParty(id=f'rp-{key}', research_donation_id=parent.id, mother_name=f'客户-{key}', tenant_no=tenant, create_dept_id='', create_time=created, del_flag=deleted)
        sample = BizResearchDonationSample(id=f'rs-{key}', research_donation_id=parent.id, sample_code=f'R-{key}', tenant_no=tenant, create_dept_id='', create_time=created, del_flag=deleted)
    db.add_all([parent, collection, party, sample])
    db.commit()
    return sample, parent


def _freight(db, key, samples, *, tenant='tenant-a', amount='12.30', deleted=0, create_by='u1'):
    now = datetime(2026, 10, 11, 8, 0)
    freight = BizSampleFreight(id=f'f-{key}', freight_no=f'YF-{key}', transport_mode='sf', freight_amount=Decimal(amount), tenant_no=tenant, create_by=create_by, create_dept_id='', create_time=now, update_time=now, del_flag=deleted)
    db.add(freight)
    for pos, (source, sample, parent) in enumerate(samples):
        db.add(BizSampleFreightItem(id=f'fi-{key}-{pos}', freight_id=freight.id, sample_source=source, sample_id=sample.id, sample_code=sample.sample_code, protocol_id=parent.id, tenant_no=tenant, create_dept_id='', create_time=now, update_time=now, del_flag=0))
    db.commit()
    return freight


def _items(*pairs):
    return [{'sampleSource': source, 'sampleId': sample.id} for source, sample in pairs]


def test_specialist_sees_shared_batch_but_only_own_items_and_count():
    from app.services.market_freight import get_freight_detail, list_freights
    db = _session()
    own, own_parent = _sample(db, 'own')
    other, other_parent = _sample(db, 'other', sales='u2')
    freight = _freight(db, 'mixed', [('cell_storage', own, own_parent), ('cell_storage', other, other_parent)])
    rows, total = list_freights(db, ctx=_ctx(), user_info=_user(), include_items=True)
    detail = get_freight_detail(db, freight.id, ctx=_ctx(), user_info=_user())
    assert total == 1 and rows[0]['visibleItemCount'] == 1
    assert [item['sampleCode'] for item in rows[0]['items']] == ['C-own']
    assert [item['sampleCode'] for item in detail['items']] == ['C-own']
    assert rows[0]['createBy'] == detail['createBy'] == 'u1'
    assert rows[0]['createByName'] == detail['createByName'] == '专员'
    assert 'linkedSampleCount' not in rows[0] and 'linkedSampleCount' not in detail
    assert 'hasHiddenItems' not in detail


def test_freight_creator_name_uses_historical_nickname_without_id_fallback():
    from app.services.market_freight import get_freight_detail, list_freights

    db = _session()
    now = datetime(2026, 10, 11, 8, 0)
    db.add_all(
        [
            SysUser(id='deleted-creator', dept_id='', user_no='deleted-no', nick_name='历史创建人', password='', tenant_no='tenant-a', create_dept_id='', create_time=now, update_time=now, del_flag=1),
            SysUser(id='blank-creator', dept_id='', user_no='raw-creator-no', nick_name='', password='', tenant_no='tenant-a', create_dept_id='', create_time=now, update_time=now, del_flag=1),
        ]
    )
    first_sample, first_parent = _sample(db, 'creator-history')
    second_sample, second_parent = _sample(db, 'creator-blank')
    historical = _freight(
        db,
        'creator-history',
        [('cell_storage', first_sample, first_parent)],
        create_by='deleted-creator',
    )
    blank = _freight(
        db,
        'creator-blank',
        [('cell_storage', second_sample, second_parent)],
        create_by='blank-creator',
    )

    items, total = list_freights(db, ctx=_ctx(), user_info=_user())
    by_id = {item['id']: item for item in items}
    historical_detail = get_freight_detail(
        db, historical.id, ctx=_ctx(), user_info=_user()
    )
    blank_detail = get_freight_detail(
        db, blank.id, ctx=_ctx(), user_info=_user()
    )

    assert total == 2
    assert by_id[historical.id]['createByName'] == '历史创建人'
    assert historical_detail['createByName'] == '历史创建人'
    assert by_id[blank.id]['createByName'] == ''
    assert blank_detail['createByName'] == ''


def test_batch_without_visible_item_is_listed_and_linkable_like_web():
    """列表与 Web 一致：按运费单数据权限展示，不再要求现场可见明细。"""
    from app.services.market_freight import get_freight_detail, link_samples, list_freights

    db = _session()
    other, parent = _sample(db, 'other', sales='u2')
    own, own_parent = _sample(db, 'own')
    freight = _freight(db, 'hidden', [('cell_storage', other, parent)])
    rows, total = list_freights(db, ctx=_ctx(), user_info=_user(), include_items=True)
    detail = get_freight_detail(db, freight.id, ctx=_ctx(), user_info=_user())
    linked = link_samples(
        db,
        freight.id,
        ctx=_ctx(),
        user_info=_user(),
        items=_items(('cell_storage', own)),
    )

    assert total == 1
    assert rows[0]['id'] == freight.id
    assert rows[0]['items'] == []
    assert detail['id'] == freight.id
    assert detail['items'] == []
    assert detail['visibleItemCount'] == 0
    assert linked['visibleItemCount'] == 1
    assert [item['sampleCode'] for item in linked['items']] == ['C-own']


def test_keyword_and_source_filters_match_any_batch_items_like_web():
    from app.services.market_freight import list_freights
    db = _session()
    own, op = _sample(db, 'own')
    secret, sp = _sample(db, 'SECRET', source='research_donation', sales='u2')
    _freight(db, 'mixed', [('cell_storage', own, op), ('research_donation', secret, sp)])
    assert list_freights(db, ctx=_ctx(), user_info=_user(), keyword='SECRET')[1] == 1
    assert list_freights(db, ctx=_ctx(), user_info=_user(), sample_source='research_donation')[1] == 1
    assert list_freights(db, ctx=_ctx(), user_info=_user(), sample_source='cell_storage')[1] == 1


def test_freight_list_filters_create_time_range():
    from app.services.market_freight import list_freights

    db = _session()
    first, first_parent = _sample(db, 'created-in-range')
    second, second_parent = _sample(db, 'created-outside')
    in_range = _freight(
        db,
        'created-in-range',
        [('cell_storage', first, first_parent)],
    )
    outside = _freight(
        db,
        'created-outside',
        [('cell_storage', second, second_parent)],
    )
    in_range.create_time = datetime(2026, 10, 11, 8, 0)
    outside.create_time = datetime(2026, 9, 1, 8, 0)
    db.commit()

    items, total = list_freights(
        db,
        ctx=_ctx(),
        user_info=_user(),
        create_time_start='2026-10-01',
        create_time_end='2026-10-13',
    )

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


def test_manager_scope_is_region_union_document_manager_for_both_sources():
    from app.services.market_freight import get_freight_detail
    db = _session()
    db.add(BizCollectionRegionStaff(id='staff', region_id='managed', user_id='m1', is_leader=1, tenant_no='tenant-a', create_dept_id='', del_flag=0))
    db.commit()
    region_sample, rp = _sample(db, 'region', sales='u9', manager='x', region='managed')
    assigned, ap = _sample(db, 'assigned', source='research_donation', sales='u9', manager='m1', region='other')
    hidden, hp = _sample(db, 'hidden', sales='m1', manager='x', region='other')
    freight = _freight(db, 'manager', [('cell_storage', region_sample, rp), ('research_donation', assigned, ap), ('cell_storage', hidden, hp)])
    detail = get_freight_detail(db, freight.id, ctx=_ctx('manager', 'm1'), user_info=_user('m1'))
    assert {item['sampleCode'] for item in detail['items']} == {'C-region', 'R-assigned'}
    assert detail['visibleItemCount'] == 2
    assert all('sampleTypeName' in item and 'motherName' in item and 'newbornName' in item for item in detail['items'])


def test_dual_role_list_uses_data_scope_not_field_scope():
    """列表与 Web 一致仅走数据权限；专员/经理路由不再按现场范围裁剪批次。"""
    from app.services.market_freight import list_freights
    db = _session()
    specialist, sp = _sample(db, 'specialist', sales='m1', manager='x', region='other')
    manager, mp = _sample(db, 'manager', sales='u9', manager='m1', region='other')
    _freight(db, 'specialist', [('cell_storage', specialist, sp)])
    _freight(db, 'manager', [('cell_storage', manager, mp)])
    specialist_rows, _ = list_freights(db, ctx=_ctx('specialist', 'm1'), user_info=_user('m1'))
    manager_rows, _ = list_freights(db, ctx=_ctx('manager', 'm1'), user_info=_user('m1'))
    assert {row['freightNo'] for row in specialist_rows} == {'YF-specialist', 'YF-manager'}
    assert {row['freightNo'] for row in manager_rows} == {'YF-specialist', 'YF-manager'}


def test_cross_tenant_freight_is_invisible_while_orphan_item_batch_remains_listed():
    from app.services.market_freight import list_freights
    db = _session()
    cross, cp = _sample(db, 'cross', tenant='tenant-b')
    deleted, dp = _sample(db, 'deleted', deleted=1)
    _freight(db, 'cross', [('cell_storage', cross, cp)], tenant='tenant-b')
    orphan = _freight(db, 'deleted', [('cell_storage', deleted, dp)])
    rows, total = list_freights(db, ctx=_ctx(), user_info=_user())
    assert total == 1
    assert rows[0]['id'] == orphan.id


def test_sample_options_are_scoped_sql_paginated_stable_and_derive_link_state():
    from app.services.market_freight import list_sample_options
    db = _session()
    base = datetime(2026, 10, 10, 8, 0)
    first, fp = _sample(db, 'first', created=base)
    _sample(db, 'second', source='research_donation', created=base + timedelta(hours=1))
    _sample(db, 'third', created=base + timedelta(hours=2))
    _sample(db, 'hidden', sales='u2', created=base + timedelta(hours=3))
    _freight(db, 'linked', [('cell_storage', first, fp)])
    page1, total = list_sample_options(db, ctx=_ctx(), user_info=_user(), page=1, page_size=2)
    page2, total2 = list_sample_options(db, ctx=_ctx(), user_info=_user(), page=2, page_size=2)
    assert total == total2 == 3
    assert [x['sampleCode'] for x in page1] == ['C-third', 'R-second']
    assert [x['sampleCode'] for x in page2] == ['C-first']
    assert page2[0]['freightId'] == 'f-linked'
    assert page2[0]['freightNo'] == 'YF-linked'
    assert set(('protocolNo', 'customerName', 'newbornName', 'sampleTypeName')).issubset(page2[0])


def test_create_derives_snapshots_and_decimal_string_and_requires_visible_item():
    from app.services.market_freight import create_freight
    db = _session()
    own, parent = _sample(db, 'own')
    result = create_freight(db, ctx=_ctx(), user_info=_user(), transport_mode='sf', freight_amount=Decimal('12.30'), items=_items(('cell_storage', own)))
    item = db.query(BizSampleFreightItem).filter_by(freight_id=result['id']).one()
    assert result['freightAmount'] == '12.30'
    assert item.sample_code == 'C-own' and item.protocol_id == parent.id
    with pytest.raises(ValueError, match='至少'):
        create_freight(db, ctx=_ctx(), user_info=_user(), transport_mode='sf', freight_amount='1.00', items=[])


def test_create_prevalidates_every_item_and_rolls_back_without_partial_batch():
    from app.services.market_freight import create_freight
    db = _session()
    own, _ = _sample(db, 'own')
    other, _ = _sample(db, 'other', sales='u2')
    before = db.query(BizSampleFreight).count()
    with pytest.raises(HTTPException):
        create_freight(db, ctx=_ctx(), user_info=_user(), transport_mode='sf', freight_amount='1.00', items=_items(('cell_storage', own), ('cell_storage', other)))
    assert db.query(BizSampleFreight).count() == before


def test_link_rejects_hidden_and_existing_other_batch_without_silent_move():
    from app.services.market_freight import link_samples
    db = _session()
    own, op = _sample(db, 'own')
    other, _ = _sample(db, 'other', sales='u2')
    linked, lp = _sample(db, 'linked')
    target = _freight(db, 'target', [('cell_storage', own, op)])
    source = _freight(db, 'source', [('cell_storage', linked, lp)])
    for sample in (other, linked):
        with pytest.raises((HTTPException, ValueError)):
            link_samples(db, target.id, ctx=_ctx(), user_info=_user(), items=_items(('cell_storage', sample)))
    assert db.query(BizSampleFreightItem).filter_by(sample_id=linked.id, del_flag=0).one().freight_id == source.id


def test_unlink_prevalidates_all_items_and_rejects_hidden_or_nonexistent():
    from app.services.market_freight import unlink_samples
    db = _session()
    own, op = _sample(db, 'own')
    other, hp = _sample(db, 'hidden', sales='u2')
    freight = _freight(db, 'mixed', [('cell_storage', own, op), ('cell_storage', other, hp)])
    missing, _ = _sample(db, 'missing')
    for sample in (other, missing):
        with pytest.raises((HTTPException, ValueError)):
            unlink_samples(db, freight.id, ctx=_ctx(), user_info=_user(), items=_items(('cell_storage', sample)))
    assert db.query(BizSampleFreightItem).filter_by(freight_id=freight.id, del_flag=0).count() == 2


def test_unlink_last_visible_item_disappears_but_batch_remains_listed():
    from app.services.market_freight import list_freights, unlink_samples
    db = _session()
    own, op = _sample(db, 'own')
    other, hp = _sample(db, 'hidden', sales='u2')
    freight = _freight(db, 'mixed', [('cell_storage', own, op), ('cell_storage', other, hp)])
    result = unlink_samples(db, freight.id, ctx=_ctx(), user_info=_user(), items=_items(('cell_storage', own)))
    assert result == {'id': freight.id, 'visibleItemCount': 0, 'disappeared': True}
    assert list_freights(db, ctx=_ctx(), user_info=_user())[1] == 1
    assert db.query(BizSampleFreight).filter_by(id=freight.id).one().del_flag == 0


@pytest.mark.parametrize('kind,user_id', [('specialist', 'u1'), ('manager', 'm1')])
def test_shared_outside_scope_header_is_immutable_for_both_roles(kind, user_id):
    from app.services.market_freight import update_freight
    db = _session()
    visible, vp = _sample(db, 'visible', sales=user_id if kind == 'specialist' else 'u9', manager=user_id if kind == 'manager' else 'x')
    hidden, hp = _sample(db, 'hidden', sales='other', manager='other')
    freight = _freight(db, 'mixed', [('cell_storage', visible, vp), ('cell_storage', hidden, hp)])
    with pytest.raises(PermissionError, match='全部'):
        update_freight(db, freight.id, ctx=_ctx(kind, user_id), user_info=_user(user_id), freight_amount='99.00')
    assert db.query(BizSampleFreight).filter_by(id=freight.id).one().freight_amount == Decimal('12.30')


def test_header_update_allowed_when_all_active_items_are_visible():
    from app.services.market_freight import update_freight
    db = _session()
    own, op = _sample(db, 'own')
    freight = _freight(db, 'own', [('cell_storage', own, op)])
    result = update_freight(
        db, freight.id, ctx=_ctx(), user_info=_user(),
        freight_amount='99.10', amount_supplied=True,
    )
    assert result['freightAmount'] == '99.10'


@pytest.mark.parametrize('kind,user_id', [('specialist', 'u1'), ('manager', 'm1')])
@pytest.mark.parametrize('mixed', [False, True])
def test_detail_edit_action_requires_every_active_item_visible_without_leaking_totals(
    kind, user_id, mixed
):
    from app.services.market_freight import get_freight_detail
    from app.services.market_support import decorate_allowed_actions

    db = _session()
    visible, visible_parent = _sample(
        db,
        f'action-visible-{kind}-{mixed}',
        sales=user_id if kind == 'specialist' else 'other-sales',
        manager=user_id if kind == 'manager' else 'other-manager',
    )
    rows = [('cell_storage', visible, visible_parent)]
    if mixed:
        hidden, hidden_parent = _sample(
            db,
            f'action-hidden-{kind}',
            sales='hidden-sales',
            manager='hidden-manager',
        )
        rows.append(('cell_storage', hidden, hidden_parent))
    freight = _freight(db, f'action-{kind}-{mixed}', rows)

    detail = get_freight_detail(
        db,
        freight.id,
        ctx=_ctx(kind, user_id),
        user_info=_user(user_id),
    )
    rendered = decorate_allowed_actions(
        'freight', detail, scope_kind=kind, detail=True
    )

    expected = ['link', 'unlink'] if mixed else ['edit', 'delete', 'link', 'unlink']
    assert detail['allowedActions'] == expected
    assert rendered['allowedActions'] == expected
    for hidden_key in (
        'allItemsVisible',
        'hiddenItemCount',
        'totalItemCount',
        'activeItemCount',
    ):
        assert hidden_key not in detail


def test_dto_forbids_scope_snapshots_status_and_non_decimal_amounts():
    from app.schemas.market_mobile import MarketFreightCreateBody, MarketFreightItemsBody
    with pytest.raises(ValidationError):
        MarketFreightItemsBody.model_validate({'items': [{'sampleSource': 'cell_storage', 'sampleId': 's1', 'sampleCode': 'forged'}]})
    for extra in ('tenantNo', 'scope', 'status'):
        with pytest.raises(ValidationError):
            MarketFreightCreateBody.model_validate({'transportMode': 'sf', 'freightAmount': '1.00', 'items': [{'sampleSource': 'cell_storage', 'sampleId': 's1'}], extra: 'x'})
    with pytest.raises(ValidationError):
        MarketFreightCreateBody.model_validate({'transportMode': 'sf', 'freightAmount': 'NaN', 'items': [{'sampleSource': 'cell_storage', 'sampleId': 's1'}]})


def test_fixed_routes_and_precise_permissions_are_registered():
    from app.api.market_mobile import router
    paths = [(route.path, route.methods) for route in router.routes]
    for kind in ('specialist', 'manager'):
        prefix = f'/mini/market/{kind}/freights'
        assert any(path == prefix and 'GET' in methods for path, methods in paths)
        assert any(path == prefix and 'POST' in methods for path, methods in paths)
        assert any(path == f'{prefix}/sample-options' and 'GET' in methods for path, methods in paths)
    source = (Path(__file__).parents[1] / 'app/api/market_mobile.py').read_text()
    assert "f'mini:market:{scope_kind}:freight'" in source


def test_migration_has_mysql57_indexes_unique_active_link_and_duplicate_preflight():
    sql = (Path(__file__).parents[2] / 'docs/sql/migrations/2026-10-11-market-freight-indexes.sql').read_text()
    lowered = sql.lower()
    assert 'information_schema.statistics' in lowered
    assert 'generated always as' in lowered
    assert 'unique' in lowered and 'active' in lowered
    assert 'group by tenant_no, sample_source, sample_id' in lowered
    assert 'having count(*) > 1' in lowered
    assert 'create procedure' in lowered and 'signal sqlstate' in lowered
    assert 'character_maximum_length' in lowered
    assert 'data_type' in lowered and 'is_nullable' in lowered
    assert 'normalized_generation_expression' in lowered
    assert "uk_sample_freight_tenant_no" in lowered
    assert 'exact_freight_unique_indexes' in lowered


def test_write_service_contains_locking_postlock_validation_and_integrity_retry():
    source = (Path(__file__).parents[1] / 'app/services/market_freight.py').read_text()
    assert source.count('with_for_update') >= 3
    assert 'IntegrityError' in source
    assert '_lock_batch_header' in source
    assert '_lock_sample_graph' in source
    assert '_lock_active_items' in source


def test_multiple_party_rows_do_not_duplicate_visible_item_or_count():
    from app.services.market_freight import get_freight_detail
    db = _session()
    sample, parent = _sample(db, 'one')
    db.add(BizProtocolPartyCustomer(id='pc-one-duplicate', collection_protocol_id=parent.id, mother_name='重复客户行', tenant_no='tenant-a', create_dept_id='', del_flag=0))
    db.commit()
    freight = _freight(db, 'one', [('cell_storage', sample, parent)])
    detail = get_freight_detail(db, freight.id, ctx=_ctx(), user_info=_user())
    assert detail['visibleItemCount'] == 1
    assert len(detail['items']) == 1


def test_deleted_batch_is_invisible_even_with_visible_active_item():
    from app.services.market_freight import list_freights
    db = _session()
    sample, parent = _sample(db, 'one')
    _freight(db, 'deleted-batch', [('cell_storage', sample, parent)], deleted=1)
    assert list_freights(db, ctx=_ctx(), user_info=_user())[1] == 0


def test_context_user_mismatch_fails_closed():
    from app.services.market_freight import list_sample_options
    db = _session()
    _sample(db, 'one')
    with pytest.raises(PermissionError):
        list_sample_options(db, ctx=_ctx(), user_info=_user('u2'))


def test_manager_can_link_sample_in_managed_region():
    from app.services.market_freight import link_samples
    db = _session()
    db.add(BizCollectionRegionStaff(id='staff-link', region_id='managed', user_id='m1', is_leader=1, tenant_no='tenant-a', create_dept_id='', del_flag=0))
    db.commit()
    existing, ep = _sample(db, 'existing', sales='u9', manager='m1')
    new_sample, _ = _sample(db, 'new', sales='u9', manager='x', region='managed')
    freight = _freight(db, 'manager-link', [('cell_storage', existing, ep)])
    detail = link_samples(db, freight.id, ctx=_ctx('manager', 'm1'), user_info=_user('m1'), items=_items(('cell_storage', new_sample)))
    assert {item['sampleCode'] for item in detail['items']} == {'C-existing', 'C-new'}


def test_batch_paging_is_stable_and_total_counts_data_scoped_batches():
    from app.services.market_freight import list_freights
    db = _session()
    first, fp = _sample(db, 'first')
    second, sp = _sample(db, 'second')
    hidden, hp = _sample(db, 'hidden-page', sales='u2')
    f1 = _freight(db, 'first', [('cell_storage', first, fp)])
    f2 = _freight(db, 'second', [('cell_storage', second, sp)])
    f3 = _freight(db, 'hidden-page', [('cell_storage', hidden, hp)])
    f1.create_time = f2.create_time
    f3.create_time = f2.create_time
    db.commit()
    page1, total1 = list_freights(db, ctx=_ctx(), user_info=_user(), page=1, page_size=1)
    page2, total2 = list_freights(db, ctx=_ctx(), user_info=_user(), page=2, page_size=1)
    page3, total3 = list_freights(db, ctx=_ctx(), user_info=_user(), page=3, page_size=1)
    assert total1 == total2 == total3 == 3
    assert [page1[0]['id'], page2[0]['id'], page3[0]['id']] == [
        'f-second',
        'f-hidden-page',
        'f-first',
    ]


@pytest.mark.parametrize('source', ['cell_storage', 'research_donation'])
def test_sample_options_choose_latest_party_once_before_count_and_paging(source):
    from app.services.market_freight import list_sample_options
    db = _session()
    base = datetime(2026, 10, 10, 8, 0)
    target, parent = _sample(db, f'duplicate-{source}', source=source, created=base)
    _sample(db, f'other-{source}', source=source, created=base + timedelta(hours=1))
    if source == 'cell_storage':
        db.add(BizProtocolPartyCustomer(
            id='zz-latest-cell-party', collection_protocol_id=parent.id,
            mother_name='最新细胞客户', tenant_no='tenant-a', create_dept_id='',
            create_time=base + timedelta(hours=2), del_flag=0,
        ))
        latest_name = '最新细胞客户'
    else:
        db.add(BizResearchDonationParty(
            id='zz-latest-donation-party', research_donation_id=parent.id,
            mother_name='最新科研客户', tenant_no='tenant-a', create_dept_id='',
            create_time=base + timedelta(hours=2), del_flag=0,
        ))
        latest_name = '最新科研客户'
    db.commit()

    page1, total1 = list_sample_options(
        db, ctx=_ctx(), user_info=_user(), page=1, page_size=1,
        sample_source=source,
    )
    page2, total2 = list_sample_options(
        db, ctx=_ctx(), user_info=_user(), page=2, page_size=1,
        sample_source=source,
    )
    rows = page1 + page2

    assert total1 == total2 == 2
    assert sum(row['sampleId'] == target.id for row in rows) == 1
    target_row = next(row for row in rows if row['sampleId'] == target.id)
    assert target_row['customerName'] == latest_name


def test_mysql_lock_queries_are_current_reads_with_stable_global_order():
    from app.services.market_freight import (
        _active_item_lock_query,
        _collection_lock_query,
        _leader_lock_query,
        _sample_parent_lock_query,
    )
    db = _session()
    ctx = _ctx()
    sample_sql = str(
        _sample_parent_lock_query(
            db, 'cell_storage', ['z', 'a'], ctx
        ).statement.compile(dialect=mysql.dialect())
    ).lower()
    collection_sql = str(
        _collection_lock_query(
            db, 'cell_storage', ['p-z', 'p-a'], ctx
        ).statement.compile(dialect=mysql.dialect())
    ).lower()
    item_sql = str(
        _active_item_lock_query(
            db, ctx, freight_id='f1', keys=[('cell_storage', 'z')]
        ).statement.compile(dialect=mysql.dialect())
    ).lower()
    leader_sql = str(
        _leader_lock_query(db, _ctx('manager', 'm1')).statement.compile(
            dialect=mysql.dialect()
        )
    ).lower()

    for sql in (sample_sql, collection_sql, item_sql, leader_sql):
        assert ' for update' in sql
        assert ' order by ' in sql
        assert 'tenant_no' in sql and 'del_flag' in sql
    assert 'sales_user_id' in collection_sql
    assert 'biz_protocol_collection_info' in collection_sql
    assert 'biz_collection_region_staff' in leader_sql


def test_all_write_paths_use_batch_graph_item_lock_order():
    source = (Path(__file__).parents[1] / 'app/services/market_freight.py').read_text()
    for start_name, end_name in (
        ('def update_freight', 'def link_samples'),
        ('def link_samples', 'def unlink_samples'),
        ('def unlink_samples', 'def delete_freight'),
        ('def delete_freight', None),
    ):
        block = source[source.index(start_name):]
        if end_name:
            block = block[:block.index(end_name)]
        assert block.index('_lock_batch_header') < block.index('_lock_sample_graph')
        assert block.index('_lock_sample_graph') < block.index('_lock_active_items')
    create_block = source[source.index('def create_freight'):source.index('def update_freight')]
    assert create_block.index('_lock_sample_graph') < create_block.index('_lock_active_items')


def test_delete_freight_cascades_when_all_items_visible():
    from app.services.market_freight import delete_freight, list_freights, list_sample_options
    db = _session()
    own, op = _sample(db, 'del-own')
    freight = _freight(db, 'del-own', [('cell_storage', own, op)])
    result = delete_freight(db, freight.id, ctx=_ctx(), user_info=_user())
    assert result == {'id': freight.id, 'visibleItemCount': 0, 'disappeared': True}
    assert db.query(BizSampleFreight).filter_by(id=freight.id).one().del_flag == 1
    assert db.query(BizSampleFreightItem).filter_by(freight_id=freight.id, del_flag=0).count() == 0
    assert list_freights(db, ctx=_ctx(), user_info=_user())[1] == 0
    options, total = list_sample_options(db, ctx=_ctx(), user_info=_user(), page=1, page_size=20)
    assert total >= 1
    assert any(row['sampleId'] == own.id and not row.get('freightId') for row in options)


def test_delete_freight_rejects_when_outside_scope_items_exist():
    from app.services.market_freight import delete_freight
    db = _session()
    own, op = _sample(db, 'del-vis')
    other, hp = _sample(db, 'del-hid', sales='u2')
    freight = _freight(db, 'del-mixed', [('cell_storage', own, op), ('cell_storage', other, hp)])
    with pytest.raises(PermissionError, match='全部'):
        delete_freight(db, freight.id, ctx=_ctx(), user_info=_user())
    assert db.query(BizSampleFreight).filter_by(id=freight.id).one().del_flag == 0
    assert db.query(BizSampleFreightItem).filter_by(freight_id=freight.id, del_flag=0).count() == 2


def test_web_admin_delete_freight_cascades_items_and_allows_relink():
    from app.crud import sample_freight as freight_crud
    db = _session()
    own, op = _sample(db, 'web-del')
    freight = _freight(db, 'web-del', [('cell_storage', own, op)])
    empty = _freight(db, 'web-empty', [])
    result = freight_crud.delete_freight(
        db, freight.id, tenant_no='tenant-a', user_id='u1'
    )
    assert result == {'id': freight.id, 'deleted': True, 'linkedSampleCount': 0}
    assert db.query(BizSampleFreight).filter_by(id=freight.id).one().del_flag == 1
    assert db.query(BizSampleFreightItem).filter_by(freight_id=freight.id, del_flag=0).count() == 0
    empty_result = freight_crud.delete_freight(
        db, empty.id, tenant_no='tenant-a', user_id='u1'
    )
    assert empty_result['deleted'] is True
    assert db.query(BizSampleFreight).filter_by(id=empty.id).one().del_flag == 1
    created = freight_crud.create_freight(
        db,
        tenant_no='tenant-a',
        user_id='u1',
        transport_mode_raw='sf',
        freight_amount='8.00',
        items=[{'sampleSource': 'cell_storage', 'sampleId': own.id, 'protocolId': op.id}],
    )
    assert created['id']
    assert any(
        item.get('sampleId') == own.id for item in (created.get('items') or [])
    )


def test_header_update_rejects_assignment_changed_after_prior_visible_read():
    from app.services.market_freight import get_freight_detail, update_freight
    db = _session()
    sample, parent = _sample(db, 'scope-race')
    freight = _freight(db, 'scope-race', [('cell_storage', sample, parent)])
    assert get_freight_detail(db, freight.id, ctx=_ctx(), user_info=_user())['visibleItemCount'] == 1

    collection = db.query(BizProtocolCollectionInfo).filter_by(
        collection_protocol_id=parent.id
    ).one()
    db.query(BizProtocolCollectionInfo).filter_by(id=collection.id).update(
        {'sales_user_id': 'u2'}, synchronize_session=False
    )
    assert collection.sales_user_id == 'u1'

    with pytest.raises(HTTPException) as exc:
        update_freight(
            db, freight.id, ctx=_ctx(), user_info=_user(), freight_amount='55.00'
        )
    assert exc.value.status_code == 404
    assert db.query(BizSampleFreight).filter_by(id=freight.id).one().freight_amount == Decimal('12.30')


def test_final_item_lock_rejects_participant_snapshot_change():
    from app.services.market_freight import _lock_active_items
    db = _session()
    first, fp = _sample(db, 'snapshot-first')
    second, sp = _sample(db, 'snapshot-second')
    freight = _freight(db, 'snapshot', [('cell_storage', first, fp)])
    expected = [('cell_storage', first.id)]
    db.add(BizSampleFreightItem(
        id='fi-snapshot-race', freight_id=freight.id,
        sample_source='cell_storage', sample_id=second.id,
        sample_code=second.sample_code, protocol_id=sp.id,
        tenant_no='tenant-a', create_dept_id='', del_flag=0,
    ))
    db.commit()

    with pytest.raises(PermissionError, match='发生变化'):
        _lock_active_items(
            db, _ctx(), freight_id=freight.id,
            expected_batch_keys=expected,
        )


def test_lock_graph_does_not_rely_on_exists_scope_for_current_field_values():
    source = (Path(__file__).parents[1] / 'app/services/market_freight.py').read_text()
    block = source[source.index('def _lock_sample_graph'):source.index('def _lock_active_items')]
    assert 'apply_cell_protocol_scope' not in block
    assert 'apply_donation_scope' not in block
    assert '_current_field_visible_parent_ids' in block
    assert '_current_data_scope_parent_ids' in block


def test_manager_header_update_rejects_current_leader_revocation():
    from app.services.market_freight import get_freight_detail, update_freight
    db = _session()
    leader = BizCollectionRegionStaff(
        id='leader-race', region_id='managed-race', user_id='m1', is_leader=1,
        tenant_no='tenant-a', create_dept_id='', del_flag=0,
    )
    db.add(leader)
    db.commit()
    sample, parent = _sample(
        db, 'leader-race', sales='u9', manager='other', region='managed-race'
    )
    freight = _freight(db, 'leader-race', [('cell_storage', sample, parent)])
    ctx = _ctx('manager', 'm1')
    assert get_freight_detail(db, freight.id, ctx=ctx, user_info=_user('m1'))['visibleItemCount'] == 1
    db.query(BizCollectionRegionStaff).filter_by(id=leader.id).update(
        {'is_leader': 0}, synchronize_session=False
    )
    assert leader.is_leader == 1

    with pytest.raises(HTTPException) as exc:
        update_freight(
            db, freight.id, ctx=ctx, user_info=_user('m1'), freight_amount='88.00'
        )
    assert exc.value.status_code == 404


def test_header_update_uses_current_role_data_scope_not_prior_snapshot():
    from app.services.market_freight import get_freight_detail, update_freight
    db = _session()
    now = datetime(2026, 10, 11, 8, 0)
    db.add_all([
        SysUser(
            id='scope-user', dept_id='dept-current', user_no='scope-user',
            nick_name='范围用户', password='', tenant_no='tenant-a',
            create_dept_id='', create_time=now, update_time=now, del_flag=0,
        ),
        SysRole(
            id='scope-role', role_name='范围角色', data_scope='1', status='1',
            tenant_no='tenant-a', create_dept_id='', create_time=now,
            update_time=now, del_flag=0,
        ),
        SysUserRole(
            id='scope-user-role', user_id='scope-user', role_id='scope-role',
            tenant_no='tenant-a', create_dept_id='', create_time=now,
            update_time=now, del_flag=0,
        ),
    ])
    db.commit()
    sample, parent = _sample(db, 'data-scope', sales='scope-user')
    parent.create_by = 'another-user'
    parent.create_dept_id = 'another-dept'
    db.commit()
    freight = _freight(db, 'data-scope', [('cell_storage', sample, parent)])
    ctx = _ctx('specialist', 'scope-user')
    user = {'user_id': 'scope-user', 'tenant_no': 'tenant-a', 'role': 'admin'}
    assert get_freight_detail(db, freight.id, ctx=ctx, user_info=user)['visibleItemCount'] == 1
    cached_role = db.query(SysRole).filter_by(id='scope-role').one()
    db.query(SysRole).filter_by(id='scope-role').update(
        {'data_scope': '4'}, synchronize_session=False
    )
    assert cached_role.data_scope == '1'

    with pytest.raises(HTTPException) as exc:
        update_freight(
            db, freight.id, ctx=ctx, user_info=user, freight_amount='77.00'
        )
    assert exc.value.status_code == 404


def test_current_data_scope_preserves_post_roles_custom_scope_and_union():
    from app.services.market_freight import _current_data_scope_parent_ids
    db = _session()
    now = datetime(2026, 10, 11, 8, 0)
    db.add_all([
        SysUser(
            id='union-user', dept_id='union-home', user_no='union-user',
            nick_name='并集用户', password='', tenant_no='tenant-a',
            create_dept_id='', create_time=now, update_time=now, del_flag=0,
        ),
        SysRole(
            id='custom-role', role_name='自定义范围', data_scope='5', status='1',
            tenant_no='tenant-a', create_dept_id='', create_time=now,
            update_time=now, del_flag=0,
        ),
        SysRole(
            id='self-role', role_name='本人范围', data_scope='4', status='1',
            tenant_no='tenant-a', create_dept_id='', create_time=now,
            update_time=now, del_flag=0,
        ),
        SysUserRole(
            id='union-direct', user_id='union-user', role_id='custom-role',
            tenant_no='tenant-a', create_dept_id='', create_time=now,
            update_time=now, del_flag=0,
        ),
        SysRoleDataScopeDetail(
            id='union-custom-detail', role_id='custom-role', scope_type='dept',
            scope_value='custom-dept', tenant_no='tenant-a', create_dept_id='',
            create_time=now, update_time=now, del_flag=0,
        ),
        SysPost(
            id='union-post', post_name='继承职位', post_code='union-post',
            status='1', tenant_no='tenant-a', create_dept_id='',
            create_time=now, update_time=now, del_flag=0,
        ),
        SysUserPost(
            id='union-user-post', user_id='union-user', post_id='union-post',
            tenant_no='tenant-a', create_dept_id='', create_time=now,
            update_time=now, del_flag=0,
        ),
        SysPostRole(
            id='union-post-role', post_id='union-post', role_id='self-role',
            tenant_no='tenant-a', create_dept_id='', create_time=now,
            update_time=now, del_flag=0,
        ),
    ])
    db.commit()
    _, custom_parent = _sample(db, 'union-custom')
    _, self_parent = _sample(db, 'union-self')
    _, denied_parent = _sample(db, 'union-denied')
    custom_parent.create_dept_id = 'custom-dept'
    self_parent.create_by = 'union-user'
    denied_parent.create_dept_id = 'other-dept'
    denied_parent.create_by = 'other-user'
    db.commit()

    visible = _current_data_scope_parent_ids(
        db,
        [custom_parent, self_parent, denied_parent],
        _ctx('specialist', 'union-user'),
        {'user_id': 'union-user', 'tenant_no': 'tenant-a', 'role': 'admin'},
    )

    assert visible == {custom_parent.id, self_parent.id}


def test_current_data_scope_preserves_department_descendant_scope():
    from app.services.market_freight import _current_data_scope_parent_ids
    db = _session()
    now = datetime(2026, 10, 11, 8, 0)
    db.add_all([
        SysUser(
            id='dept-user', dept_id='dept-root', user_no='dept-user',
            nick_name='部门用户', password='', tenant_no='tenant-a',
            create_dept_id='', create_time=now, update_time=now, del_flag=0,
        ),
        SysRole(
            id='dept-role', role_name='部门以下', data_scope='3', status='1',
            tenant_no='tenant-a', create_dept_id='', create_time=now,
            update_time=now, del_flag=0,
        ),
        SysUserRole(
            id='dept-user-role', user_id='dept-user', role_id='dept-role',
            tenant_no='tenant-a', create_dept_id='', create_time=now,
            update_time=now, del_flag=0,
        ),
        SysDept(
            id='dept-root', parent_id='0', dept_name='根部门', status='1',
            tenant_no='tenant-a', create_dept_id='', create_time=now,
            update_time=now, del_flag=0,
        ),
        SysDept(
            id='dept-child', parent_id='dept-root', dept_name='子部门', status='1',
            tenant_no='tenant-a', create_dept_id='', create_time=now,
            update_time=now, del_flag=0,
        ),
    ])
    db.commit()
    _, child_parent = _sample(db, 'dept-child')
    _, denied_parent = _sample(db, 'dept-denied')
    child_parent.create_dept_id = 'dept-child'
    denied_parent.create_dept_id = 'other-dept'
    db.commit()

    visible = _current_data_scope_parent_ids(
        db,
        [child_parent, denied_parent],
        _ctx('specialist', 'dept-user'),
        {'user_id': 'dept-user', 'tenant_no': 'tenant-a', 'role': 'admin'},
    )

    assert visible == {child_parent.id}


def test_locked_data_scope_ignores_payload_superadmin_across_tenants():
    from app.services.market_freight import (
        _current_data_scope_parent_ids,
        get_freight_detail,
    )
    db = _session()
    now = datetime(2026, 10, 11, 8, 0)
    db.add_all([
        SysUser(
            id='cross-user', dept_id='cross-dept', user_no='cross-user',
            nick_name='跨租户用户', password='', tenant_no='tenant-a',
            create_dept_id='', create_time=now, update_time=now, del_flag=0,
        ),
        SysRole(
            id='default-super-role', role_name='超级管理员', data_scope='1',
            status='1', tenant_no='zkxw', create_dept_id='', create_time=now,
            update_time=now, del_flag=0,
        ),
        SysUserRole(
            id='default-super-link', user_id='cross-user',
            role_id='default-super-role', tenant_no='zkxw', create_dept_id='',
            create_time=now, update_time=now, del_flag=0,
        ),
        SysRole(
            id='business-self-role', role_name='业务专员', data_scope='4',
            status='1', tenant_no='tenant-a', create_dept_id='', create_time=now,
            update_time=now, del_flag=0,
        ),
        SysUserRole(
            id='business-self-link', user_id='cross-user',
            role_id='business-self-role', tenant_no='tenant-a',
            create_dept_id='', create_time=now, update_time=now, del_flag=0,
        ),
    ])
    db.commit()
    sample, parent = _sample(
        db, 'cross-tenant-scope', sales='cross-user'
    )
    parent.create_by = 'another-user'
    parent.create_dept_id = 'another-dept'
    db.commit()
    ctx = _ctx('specialist', 'cross-user')
    forged_default_tenant_payload = {
        'user_id': 'cross-user', 'tenant_no': 'tenant-a', 'role': 'super_admin'
    }
    freight = _freight(
        db, 'cross-tenant-scope', [('cell_storage', sample, parent)]
    )

    assert _current_data_scope_parent_ids(
        db, [parent], ctx, forged_default_tenant_payload
    ) == set()
    with pytest.raises(HTTPException) as exc:
        get_freight_detail(
            db, freight.id, ctx=ctx, user_info=forged_default_tenant_payload
        )
    assert exc.value.status_code == 404

    db.add_all([
        SysRole(
            id='business-super-role', role_name='超级管理员', data_scope='4',
            status='1', tenant_no='tenant-a', create_dept_id='', create_time=now,
            update_time=now, del_flag=0,
        ),
        SysPost(
            id='business-super-post', post_name='业务租户超级管理职位',
            post_code='business-super-post', status='1', tenant_no='tenant-a',
            create_dept_id='', create_time=now, update_time=now, del_flag=0,
        ),
        SysUserPost(
            id='business-super-user-post', user_id='cross-user',
            post_id='business-super-post', tenant_no='tenant-a',
            create_dept_id='', create_time=now, update_time=now, del_flag=0,
        ),
        SysPostRole(
            id='business-super-post-role', post_id='business-super-post',
            role_id='business-super-role', tenant_no='tenant-a',
            create_dept_id='', create_time=now, update_time=now, del_flag=0,
        ),
    ])
    db.commit()

    assert _current_data_scope_parent_ids(
        db, [parent], ctx, forged_default_tenant_payload
    ) == {parent.id}
    assert get_freight_detail(
        db, freight.id, ctx=ctx, user_info=forged_default_tenant_payload
    )['visibleItemCount'] == 1