zkxw/backend/tests/test_market_support_api.py

import asyncio
from types import SimpleNamespace

import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from app.models.base import Base
from app.models.models import (
    BizAttachment,
    BizCellSampleType,
    BizCollectionHospital,
    BizCollectionRegion,
    BizCollectionRegionStaff,
    BizItemUnit,
    SysDictData,
    SysUser,
)
from app.services.market_field_scope import MarketFieldContext


def _session():
    engine = create_engine('sqlite:///:memory:')
    Base.metadata.create_all(
        engine,
        tables=[
            BizAttachment.__table__,
            BizCellSampleType.__table__,
            BizCollectionHospital.__table__,
            BizCollectionRegion.__table__,
            BizCollectionRegionStaff.__table__,
            BizItemUnit.__table__,
            SysDictData.__table__,
            SysUser.__table__,
        ],
    )
    return sessionmaker(bind=engine)()


def _ctx(scope='specialist', user_id='user-a'):
    return MarketFieldContext('tenant-a', user_id, scope)


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


def _seed_options(db):
    db.add_all(
        [
            BizCollectionRegion(id='ra', region_name='A区', tenant_no='tenant-a', create_dept_id='', del_flag=0),
            BizCollectionRegion(id='rb', region_name='B区', tenant_no='tenant-a', create_dept_id='', del_flag=0),
            BizCollectionRegion(id='rx', region_name='跨租户秘密区', tenant_no='tenant-b', create_dept_id='', del_flag=0),
            BizCollectionHospital(id='ha', region_id='ra', hospital_name='A院', tenant_no='tenant-a', create_dept_id='', del_flag=0),
            BizCollectionHospital(id='hb', region_id='rb', hospital_name='B院', tenant_no='tenant-a', create_dept_id='', del_flag=0),
            BizCollectionHospital(id='hx', region_id='rx', hospital_name='跨租户秘密院', tenant_no='tenant-b', create_dept_id='', del_flag=0),
            BizCollectionRegionStaff(id='rsa', region_id='ra', user_id='user-a', is_leader=0, tenant_no='tenant-a', create_dept_id='', del_flag=0),
            BizCollectionRegionStaff(id='rsb', region_id='rb', user_id='user-b', is_leader=1, tenant_no='tenant-a', create_dept_id='', del_flag=0),
            BizCollectionRegionStaff(id='rsa2', region_id='ra', user_id='staff-a', is_leader=0, tenant_no='tenant-a', create_dept_id='', del_flag=0),
            BizCollectionRegionStaff(id='rsa3', region_id='ra', user_id='leader-a', is_leader=1, tenant_no='tenant-a', create_dept_id='', del_flag=0),
            BizCollectionRegionStaff(id='rsx', region_id='rx', user_id='cross-user', is_leader=0, tenant_no='tenant-b', create_dept_id='', del_flag=0),
            SysUser(id='user-a', user_no='ua', nick_name='本人', status='1', tenant_no='tenant-a', create_dept_id='', del_flag=0),
            SysUser(id='user-b', user_no='ub', nick_name='B经理', status='1', tenant_no='tenant-a', create_dept_id='', del_flag=0),
            SysUser(id='staff-a', user_no='sa', nick_name='A员工', status='1', tenant_no='tenant-a', create_dept_id='', del_flag=0),
            SysUser(id='leader-a', user_no='la', nick_name='A负责人', status='1', tenant_no='tenant-a', create_dept_id='', del_flag=0),
            SysUser(id='disabled', user_no='off', nick_name='停用秘密', status='0', tenant_no='tenant-a', create_dept_id='', del_flag=0),
            SysUser(id='cross-user', user_no='xx', nick_name='跨租户秘密用户', status='1', tenant_no='tenant-b', create_dept_id='', del_flag=0),
            BizCellSampleType(id='type-a', type_code='A', type_name='样本A', tenant_no='tenant-a', create_dept_id='', del_flag=0),
            BizCellSampleType(id='type-x', type_code='X', type_name='跨租户秘密样本', tenant_no='tenant-b', create_dept_id='', del_flag=0),
            BizItemUnit(id='unit-a', unit_code='ml', unit_name='毫升', tenant_no='tenant-a', create_dept_id='', del_flag=0),
            SysDictData(id='dict-pay', dict_label='微信', dict_value='wx', dict_type='zzfs', tenant_no='tenant-a', create_dept_id='', del_flag=0),
            SysDictData(id='dict-freight', dict_label='专线车', dict_value='zxc', dict_type='sample_transport_mode', tenant_no='tenant-a', create_dept_id='', del_flag=0),
            SysDictData(id='dict-secret', dict_label='不应泄漏', dict_value='secret', dict_type='unrelated', tenant_no='tenant-a', create_dept_id='', del_flag=0),
        ]
    )
    db.commit()


def test_specialist_form_options_are_region_scoped_and_minimal(monkeypatch):
    from app.services.market_support import get_form_options

    db = _session()
    _seed_options(db)

    out = get_form_options(db, module='presign', ctx=_ctx(), user_info=_user())

    assert out['regions'] == [{'value': 'ra', 'label': 'A区'}]
    assert out['hospitals'] == [{'value': 'ha', 'label': 'A院'}]
    assert out['staff'] == [
        {'value': 'leader-a', 'label': 'A负责人', 'isLeader': True},
        {'value': 'staff-a', 'label': 'A员工', 'isLeader': False},
        {'value': 'user-a', 'label': '本人', 'isLeader': False},
    ]
    assert out['sampleTypes'] == [{'value': 'type-a', 'label': '样本A'}]
    assert out['units'] == [{'value': 'unit-a', 'label': '毫升'}]
    assert out['payModes'] == [
        {'label': '趸交', 'value': 'dj'},
        {'label': '分期', 'value': 'fq'},
    ]
    assert out['orderStatuses'] == [
        {'label': '草稿', 'value': 'draft'},
        {'label': '进行中', 'value': 'active'},
        {'label': '已转正式', 'value': 'converted'},
        {'label': '已作废', 'value': 'void'},
    ]
    assert out['transferTypes'] == [{'value': 'wx', 'label': '微信'}]
    assert '跨租户' not in str(out)

    monkeypatch.setattr(
        'app.services.market_support._business_region_id',
        lambda *args, **kwargs: 'rb',
    )
    supplemented = get_form_options(
        db,
        module='presign',
        ctx=_ctx(),
        user_info=_user(),
        business_id='own-document',
    )
    assert supplemented['regions'] == [
        {'value': 'ra', 'label': 'A区'},
        {'value': 'rb', 'label': 'B区'},
    ]


def test_manager_form_options_only_include_led_regions_and_requested_region_must_be_authorized():
    from app.services.market_support import get_form_options

    db = _session()
    _seed_options(db)

    out = get_form_options(
        db,
        module='protocol',
        ctx=_ctx('manager', 'user-b'),
        user_info=_user('user-b'),
        region_id='rb',
    )
    assert out['regions'] == [{'value': 'rb', 'label': 'B区'}]
    assert out['hospitals'] == [{'value': 'hb', 'label': 'B院'}]

    with pytest.raises(PermissionError, match='地区'):
        get_form_options(
            db,
            module='protocol',
            ctx=_ctx('manager', 'user-b'),
            user_info=_user('user-b'),
            region_id='ra',
        )


def test_payment_and_freight_options_use_fixed_dictionary_allowlist():
    from app.services.market_support import get_form_options

    db = _session()
    _seed_options(db)
    payment = get_form_options(db, module='payment', ctx=_ctx(), user_info=_user())
    freight = get_form_options(db, module='freight', ctx=_ctx(), user_info=_user())

    assert payment == {'transferTypes': [{'value': 'wx', 'label': '微信'}]}
    assert freight == {'transportModes': [{'value': 'zxc', 'label': '专线车'}]}
    assert '不应泄漏' not in str(payment) + str(freight)


class _Storage:
    def __init__(self):
        self.uploaded = []
        self.deleted = []

    def upload_file(self, content, filename, folder):
        self.uploaded.append((content, filename, folder))
        return f'https://files.test/{folder}/generated.pdf'

    def delete_file(self, path):
        self.deleted.append(path)
        return True


def test_upload_uses_fixed_folder_and_records_current_tenant_and_uploader():
    from app.services.market_support import save_market_attachment

    db = _session()
    backend = _Storage()
    result = save_market_attachment(
        db,
        content=b'%PDF-1.7 safe',
        filename='proof.pdf',
        content_type='application/pdf',
        module='payment',
        ctx=_ctx(),
        user_info=_user(),
        storage_backend=backend,
    )

    assert backend.uploaded[0][2] == 'market/specialist/payment'
    row = db.query(BizAttachment).one()
    assert (row.tenant_no, row.create_by, row.create_dept_id) == ('tenant-a', 'user-a', 'dept-a')
    assert row.business_type == 'market:specialist:payment:staging'
    assert result['id'] == row.id
    assert result['fileUrl'] == row.file_url


def test_presign_upload_allows_20_mib_while_other_modules_remain_at_10_mib():
    from app.services.market_support import save_market_attachment

    content = b'%PDF-' + b'a' * (10 * 1024 * 1024)
    backend = _Storage()

    result = save_market_attachment(
        _session(),
        content=content,
        filename='presign-proof.pdf',
        content_type='application/pdf',
        module='presign',
        ctx=_ctx(),
        user_info=_user(),
        storage_backend=backend,
    )

    assert result['fileSize'] == len(content)
    with pytest.raises(ValueError, match='10MiB'):
        save_market_attachment(
            _session(),
            content=content,
            filename='payment-proof.pdf',
            content_type='application/pdf',
            module='payment',
            ctx=_ctx(),
            user_info=_user(),
            storage_backend=_Storage(),
        )


@pytest.mark.parametrize(
    ('content', 'filename', 'content_type'),
    [
        (b'', 'empty.pdf', 'application/pdf'),
        (b'<svg/>', 'bad.svg', 'image/svg+xml'),
        (b'MZ', 'bad.exe', 'application/octet-stream'),
        (b'not an image', 'fake.jpg', 'image/jpeg'),
        (b'%PDF-1.7', 'fake.jpg', 'image/jpeg'),
        (b'\x00\x00\x00\x18junk-ftyp-heic-data', 'fake.heic', 'image/heic'),
        (b'\x00\x00\x00\x18ftypzzzz\x00\x00\x00\x00zzzz', 'fake.heic', 'image/heic'),
        (b'\x00\x00\x00\xffftypheic\x00\x00\x00\x00', 'fake.heic', 'image/heic'),
        (b'\x00\x00\x00\x0cftypheic', 'short.heic', 'image/heic'),
        (b'\x00\x00\x00\x0fftypheicabc', 'short.heic', 'image/heic'),
        (b'\x00\x00\x00\x15ftypheic\x00\x00\x00\x00mif1x', 'misaligned.heic', 'image/heic'),
        (b'\x00\x00\x00\x10ftypmif1\x00\x00\x00\x00', 'generic-only.heic', 'image/heic'),
        (b'PK\x03\x04not-a-docx', 'fake.docx', 'application/vnd.openxmlformats-officedocument.wordprocessingml.document'),
        (b'a' * (10 * 1024 * 1024 + 1), 'huge.pdf', 'application/pdf'),
    ],
)
def test_upload_rejects_empty_oversized_or_unsafe_files(content, filename, content_type):
    from app.services.market_support import save_market_attachment

    db = _session()
    backend = _Storage()
    with pytest.raises(ValueError):
        save_market_attachment(
            db,
            content=content,
            filename=filename,
            content_type=content_type,
            module='payment',
            ctx=_ctx(),
            user_info=_user(),
            storage_backend=backend,
        )
    assert backend.uploaded == []


def test_upload_deletes_storage_object_when_database_commit_fails(monkeypatch):
    from app.services.market_support import save_market_attachment

    db = _session()
    backend = _Storage()
    monkeypatch.setattr(db, 'commit', lambda: (_ for _ in ()).throw(RuntimeError('db down')))

    with pytest.raises(RuntimeError, match='db down'):
        save_market_attachment(
            db,
            content=b'%PDF-1.7 safe',
            filename='proof.pdf',
            content_type='application/pdf',
            module='protocol',
            ctx=_ctx(),
            user_info=_user(),
            storage_backend=backend,
        )
    assert backend.deleted == ['https://files.test/market/specialist/protocol/generated.pdf']


def test_upload_never_refreshes_after_commit_or_deletes_committed_object(monkeypatch):
    from app.services.market_support import save_market_attachment

    db = _session()
    backend = _Storage()
    committed = False
    original_commit = db.commit
    original_refresh = db.refresh

    def tracked_commit():
        nonlocal committed
        original_commit()
        committed = True

    def refresh_only_before_commit(row):
        if committed:
            raise RuntimeError('refresh after commit is unsafe')
        return original_refresh(row)

    monkeypatch.setattr(db, 'commit', tracked_commit)
    monkeypatch.setattr(db, 'refresh', refresh_only_before_commit)

    result = save_market_attachment(
        db,
        content=b'%PDF-1.7 safe',
        filename='proof.pdf',
        content_type='application/pdf',
        module='protocol',
        ctx=_ctx(),
        user_info=_user(),
        storage_backend=backend,
    )

    assert result['id']
    assert backend.deleted == []


def test_upload_accepts_well_formed_heif_ftyp_box_with_allowed_brand():
    from app.services.market_support import save_market_attachment

    db = _session()
    backend = _Storage()
    content = b'\x00\x00\x00\x14ftypheic\x00\x00\x00\x00mif1'

    result = save_market_attachment(
        db,
        content=content,
        filename='photo.heic',
        content_type='image/heic',
        module='sample',
        ctx=_ctx(),
        user_info=_user(),
        storage_backend=backend,
    )

    assert result['fileType'] == 'image/heic'
    assert backend.uploaded[0][0] == content


def test_heif_extension_accepts_generic_heif_brand():
    from app.services.market_support import save_market_attachment

    db = _session()
    backend = _Storage()
    content = b'\x00\x00\x00\x10ftypmif1\x00\x00\x00\x00'

    result = save_market_attachment(
        db,
        content=content,
        filename='photo.heif',
        content_type='image/heif',
        module='sample',
        ctx=_ctx(),
        user_info=_user(),
        storage_backend=backend,
    )

    assert result['fileType'] == 'image/heif'


def _attachment(**overrides):
    values = dict(
        id='att-a',
        business_type='market:specialist:payment:staging',
        business_id=None,
        original_name='proof.pdf',
        file_ext='pdf',
        file_url='market/specialist/payment/a.pdf',
        file_size=12,
        file_type='application/pdf',
        tenant_no='tenant-a',
        create_by='user-a',
        create_dept_id='',
        del_flag=0,
    )
    values.update(overrides)
    return BizAttachment(**values)


def test_create_attachment_refs_require_current_uploader_staging_exact_match():
    from app.services.market_support import validate_attachment_refs

    db = _session()
    db.add_all(
        [
            _attachment(),
            _attachment(id='att-other', file_url='market/specialist/payment/other.pdf', create_by='other'),
            _attachment(id='att-cross', file_url='market/specialist/payment/cross.pdf', tenant_no='tenant-b'),
        ]
    )
    db.commit()

    assert validate_attachment_refs(
        db,
        'att-a,market/specialist/payment/a.pdf',
        module='payment',
        ctx=_ctx(),
        creating=True,
    ) == 'market/specialist/payment/a.pdf'
    for invalid in ('att-other', 'att-cross', 'https://evil.test/a.pdf', 'a.pdf'):
        with pytest.raises(ValueError, match='附件'):
            validate_attachment_refs(db, invalid, module='payment', ctx=_ctx(), creating=True)


def test_staging_attachment_must_match_current_scope_role_exactly():
    from app.services.market_support import validate_attachment_refs

    db = _session()
    db.add(
        _attachment(
            id='manager-staging',
            business_type='market:manager:payment:staging',
            file_url='market/manager/payment/proof.pdf',
        )
    )
    db.commit()

    with pytest.raises(ValueError, match='附件'):
        validate_attachment_refs(
            db,
            'manager-staging',
            module='payment',
            ctx=_ctx('specialist'),
            creating=True,
            business_id='pay-a',
        )


def test_attachment_refs_bind_once_to_exact_scope_module_and_business():
    from app.services.market_support import validate_attachment_refs

    db = _session()
    row = _attachment()
    db.add(row)
    db.commit()

    result = validate_attachment_refs(
        db,
        'att-a,market/specialist/payment/a.pdf',
        module='payment',
        ctx=_ctx(),
        creating=True,
        business_id='transfer-a',
    )

    assert result == 'market/specialist/payment/a.pdf'
    assert (row.business_type, row.business_id) == (
        'market:specialist:payment:bound',
        'transfer-a',
    )
    assert validate_attachment_refs(
        db,
        'att-a',
        module='payment',
        ctx=_ctx(),
        creating=False,
        business_id='transfer-a',
    ) == 'market/specialist/payment/a.pdf'
    with pytest.raises(ValueError, match='附件'):
        validate_attachment_refs(
            db,
            'att-a',
            module='payment',
            ctx=_ctx(),
            creating=False,
            business_id='transfer-b',
        )
    with pytest.raises(ValueError, match='附件'):
        validate_attachment_refs(
            db,
            'att-a',
            module='payment',
            ctx=_ctx('manager'),
            creating=False,
            business_id='transfer-a',
        )


def test_attachment_ref_resolution_uses_one_batched_query_and_deduplicates_aliases():
    from sqlalchemy import event

    from app.services.market_support import validate_attachment_refs

    db = _session()
    db.add_all(
        [
            _attachment(id='att-1', file_url='market/specialist/payment/1.pdf'),
            _attachment(id='att-2', file_url='market/specialist/payment/2.pdf'),
        ]
    )
    db.commit()
    statements = []

    def capture(_conn, _cursor, statement, _parameters, _context, _many):
        if 'biz_attachment' in statement.lower() and statement.lstrip().lower().startswith('select'):
            statements.append(statement)

    event.listen(db.get_bind(), 'before_cursor_execute', capture)
    try:
        result = validate_attachment_refs(
            db,
            'att-1,market/specialist/payment/1.pdf,att-2',
            module='payment',
            ctx=_ctx(),
            creating=True,
            business_id='transfer-batch',
        )
    finally:
        event.remove(db.get_bind(), 'before_cursor_execute', capture)

    assert result == 'market/specialist/payment/1.pdf,market/specialist/payment/2.pdf'
    assert len(statements) == 1


@pytest.mark.parametrize(
    'refs',
    [','.join(f'att-{index}' for index in range(21)), 'x' * 4001],
)
def test_attachment_refs_reject_excessive_count_or_total_length(refs):
    from app.services.market_support import validate_attachment_refs

    with pytest.raises(ValueError, match='附件'):
        validate_attachment_refs(
            _session(),
            refs,
            module='payment',
            ctx=_ctx(),
            creating=True,
            business_id='transfer-limit',
        )


def test_all_mobile_attachment_schemas_cap_total_reference_length():
    from pydantic import ValidationError

    from app.schemas.market_mobile import (
        MarketPaymentTransferBody,
        MarketPresignTransferBody,
        MarketProtocolAttachmentsBody,
        MarketSampleStorageCertBody,
    )

    too_long = 'x' * 4001
    payloads = (
        (MarketPresignTransferBody, {'transferAttachments': too_long}),
        (MarketProtocolAttachmentsBody, {'protocolAttachments': too_long}),
        (MarketPaymentTransferBody, {'transferAttachments': too_long}),
        (MarketSampleStorageCertBody, {'attachments': too_long}),
    )
    for schema, payload in payloads:
        with pytest.raises(ValidationError):
            schema.model_validate(payload)


def test_update_attachment_refs_allow_exact_existing_refs_and_current_user_new_only():
    from app.services.market_support import validate_attachment_refs

    db = _session()
    db.add_all(
        [
            _attachment(
                id='existing',
                file_url='market/specialist/payment/existing.pdf',
                create_by='legacy',
                business_type='market:specialist:payment:bound',
                business_id='transfer-existing',
            ),
            _attachment(id='new', file_url='market/specialist/payment/new.pdf'),
        ]
    )
    db.commit()

    out = validate_attachment_refs(
        db,
        'existing,new',
        module='payment',
        ctx=_ctx(),
        creating=False,
        existing_refs='market/specialist/payment/existing.pdf',
        business_id='transfer-existing',
    )
    assert out == 'market/specialist/payment/existing.pdf,market/specialist/payment/new.pdf'


@pytest.mark.parametrize('metadata_kind', ['missing', 'legacy-unbound', 'other-role-bound'])
def test_update_may_retain_exact_server_supplied_existing_url_without_current_metadata(
    metadata_kind,
):
    from app.services.market_support import validate_attachment_refs

    db = _session()
    existing_url = 'legacy/payment/existing.pdf'
    if metadata_kind == 'legacy-unbound':
        db.add(
            _attachment(
                id='legacy-unbound',
                file_url=existing_url,
                business_type='protocol',
                create_by='legacy-user',
            )
        )
    elif metadata_kind == 'other-role-bound':
        db.add(
            _attachment(
                id='specialist-bound',
                file_url=existing_url,
                business_type='market:specialist:payment:bound',
                business_id='transfer-existing',
                create_by='specialist-user',
            )
        )
    db.commit()

    assert validate_attachment_refs(
        db,
        existing_url,
        module='payment',
        ctx=_ctx('manager'),
        creating=False,
        existing_refs=existing_url,
        business_id='transfer-existing',
    ) == existing_url


def test_existing_ref_exception_allows_removal_but_not_arbitrary_url_smuggling():
    from app.services.market_support import validate_attachment_refs

    db = _session()
    assert validate_attachment_refs(
        db,
        '',
        module='payment',
        ctx=_ctx(),
        creating=False,
        existing_refs='legacy/payment/existing.pdf',
        business_id='transfer-existing',
    ) == ''

    with pytest.raises(ValueError, match='附件'):
        validate_attachment_refs(
            db,
            'https://evil.test/smuggled.pdf',
            module='payment',
            ctx=_ctx(),
            creating=False,
            existing_refs='legacy/payment/existing.pdf',
            business_id='transfer-existing',
        )


def test_allowed_actions_are_state_scope_and_permission_aware():
    from app.services.market_support import allowed_actions

    assert allowed_actions('presign', status='active') == ['edit', 'convert', 'delete']
    assert allowed_actions('presign', status='void') == ['edit', 'delete']
    assert allowed_actions('presign', status='converted') == ['edit']
    assert allowed_actions('presign-transfer', status='dfj') == ['edit', 'delete', 'submit']
    assert allowed_actions('presign-transfer', status='jfqrz') == ['withdraw']
    assert allowed_actions('presign-transfer', status='yjf') == []
    assert allowed_actions('protocol', status='dsx') == ['edit']
    assert allowed_actions('protocol', status='yzz') == []
    assert allowed_actions('payment-transfer', status='dfj') == ['edit-transfer', 'submit-finance']
    assert allowed_actions('payment-transfer', status='jfqrz') == ['withdraw-transfer']
    assert allowed_actions('payment-chase', status='', has_payment_permission=False) == []
    assert allowed_actions('payment-chase', status='', has_payment_permission=True) == ['register-transfer']
    assert allowed_actions('sample', status='', source='research_donation', has_sample_code=True) == []
    assert allowed_actions('sample', status='', source='cell_storage', has_sample_code=True) == ['storage-cert', 'insurance']
    assert allowed_actions('freight', status='', scope_kind='specialist') == []
    assert allowed_actions('freight', status='', scope_kind='manager') == ['edit']
    assert allowed_actions('presign-list', status='') == ['create']
    assert allowed_actions('protocol-list', status='') == ['create']
    assert allowed_actions('freight-list', status='', scope_kind='specialist') == ['create']


def test_presign_actions_include_delete_and_nested_transfer_actions():
    from app.services.market_support import decorate_allowed_actions

    active = decorate_allowed_actions(
        'presign',
        {
            'orderStatus': 'active',
            'transfers': [
                {'id': 't1', 'payStatus': 'dfj'},
                {'id': 't2', 'payStatus': 'jfqrz'},
                {'id': 't3', 'payStatus': 'yjf'},
            ],
        },
        scope_kind='specialist',
        detail=True,
    )

    assert active['allowedActions'] == ['edit', 'convert', 'delete']
    assert [row['allowedActions'] for row in active['transfers']] == [
        ['edit', 'delete', 'submit'],
        ['withdraw'],
        [],
    ]

    for parent_status, parent_actions in (
        ('converted', ['edit']),
        ('void', ['edit', 'delete']),
    ):
        locked = decorate_allowed_actions(
            'presign',
            {
                'orderStatus': parent_status,
                'transfers': [
                    {'id': 't1', 'payStatus': 'dfj'},
                    {'id': 't2', 'payStatus': 'jfqrz'},
                ],
            },
            scope_kind='specialist',
            detail=True,
        )
        assert locked['allowedActions'] == parent_actions
        assert [row['allowedActions'] for row in locked['transfers']] == [[], []]

    unknown = decorate_allowed_actions(
        'presign',
        {
            'orderStatus': 'legacy-locked',
            'transfers': [{'id': 't1', 'payStatus': 'dfj'}],
        },
        scope_kind='specialist',
        detail=True,
    )
    assert unknown['transfers'][0]['allowedActions'] == []


def test_form_options_do_not_leak_children_of_soft_deleted_regions():
    from app.services.market_support import get_form_options

    db = _session()
    _seed_options(db)
    db.add_all(
        [
            BizCollectionRegion(
                id='deleted-region',
                region_name='已删除地区',
                tenant_no='tenant-a',
                create_dept_id='',
                del_flag=1,
            ),
            BizCollectionHospital(
                id='deleted-region-hospital',
                region_id='deleted-region',
                hospital_name='不应泄漏医院',
                tenant_no='tenant-a',
                create_dept_id='',
                del_flag=0,
            ),
            BizCollectionRegionStaff(
                id='deleted-region-staff',
                region_id='deleted-region',
                user_id='user-a',
                is_leader=0,
                tenant_no='tenant-a',
                create_dept_id='',
                del_flag=0,
            ),
        ]
    )
    db.commit()

    out = get_form_options(db, module='presign', ctx=_ctx(), user_info=_user())

    assert '已删除地区' not in str(out)
    assert '不应泄漏医院' not in str(out)
    with pytest.raises(PermissionError, match='地区'):
        get_form_options(
            db,
            module='presign',
            ctx=_ctx(),
            user_info=_user(),
            region_id='deleted-region',
        )


def test_decorate_allowed_actions_covers_nested_payment_transfers():
    from app.services.market_support import decorate_allowed_actions

    detail = {
        'id': 'pay-a',
        'schedules': [
            {
                'id': 'schedule-a',
                'scheduleStatus': 'open',
                'transfers': [
                    {'id': 'draft', 'payStatus': 'dfj'},
                    {'id': 'submitted', 'payStatus': 'jfqrz'},
                    {'id': 'done', 'payStatus': 'yjf'},
                ],
            }
        ],
    }
    out = decorate_allowed_actions('payment', detail, scope_kind='specialist')

    assert out['allowedActions'] == ['register-transfer']
    assert out['schedules'][0]['allowedActions'] == ['register-transfer']
    assert [row['allowedActions'] for row in out['schedules'][0]['transfers']] == [
        ['edit-transfer', 'submit-finance'],
        ['withdraw-transfer'],
        [],
    ]


@pytest.mark.parametrize('protocol_status', ['yzf', 'sdfq', 'zky'])
def test_payment_terminal_protocol_removes_all_nested_write_actions(protocol_status):
    from app.services.market_support import decorate_allowed_actions

    detail = {
        'id': 'pay-terminal',
        'protocolStatus': protocol_status,
        'schedules': [
            {
                'id': 'schedule-open',
                'scheduleStatus': 'open',
                'transfers': [
                    {'id': 'draft', 'payStatus': 'dfj'},
                    {'id': 'submitted', 'payStatus': 'jfqrz'},
                ],
            }
        ],
    }

    out = decorate_allowed_actions('payment', detail, scope_kind='specialist')

    assert out['allowedActions'] == []
    assert out['schedules'][0]['allowedActions'] == []
    assert [row['allowedActions'] for row in out['schedules'][0]['transfers']] == [[], []]


@pytest.mark.parametrize('schedule_status', ['cleared', 'void'])
def test_non_open_payment_schedule_removes_schedule_and_transfer_actions(schedule_status):
    from app.services.market_support import decorate_allowed_actions

    detail = {
        'id': 'pay-closed',
        'protocolStatus': 'sxz',
        'schedules': [
            {
                'id': 'schedule-closed',
                'scheduleStatus': schedule_status,
                'transfers': [{'id': 'draft', 'payStatus': 'dfj'}],
            }
        ],
    }

    out = decorate_allowed_actions('payment', detail, scope_kind='specialist')

    assert out['allowedActions'] == []
    assert out['schedules'][0]['allowedActions'] == []
    assert out['schedules'][0]['transfers'][0]['allowedActions'] == []


def test_freight_detail_exposes_only_scope_safe_link_actions():
    from app.services.market_support import decorate_allowed_actions

    specialist = decorate_allowed_actions(
        'freight',
        {'id': 'f1', 'allowedActions': ['edit', 'link', 'unlink']},
        scope_kind='specialist',
        detail=True,
    )
    manager = decorate_allowed_actions(
        'freight',
        {'id': 'f1', 'allowedActions': ['link', 'unlink']},
        scope_kind='manager',
        detail=True,
    )
    assert specialist['allowedActions'] == ['edit', 'link', 'unlink']
    assert manager['allowedActions'] == ['link', 'unlink']


def test_protocol_detail_action_reads_nested_basic_status():
    from app.services.market_support import decorate_allowed_actions

    active = decorate_allowed_actions(
        'protocol', {'basic': {'protocolStatus': 'dsx'}}, scope_kind='specialist'
    )
    terminal = decorate_allowed_actions(
        'protocol', {'basic': {'protocolStatus': 'yzf'}}, scope_kind='specialist'
    )
    assert active['allowedActions'] == ['edit']
    assert terminal['allowedActions'] == []


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

    paths = [route.path for route in router.routes]
    for scope in ('specialist', 'manager'):
        for path in (
            f'/mini/market/{scope}/presign-orders/form-options',
            f'/mini/market/{scope}/protocols/form-options',
            f'/mini/market/{scope}/payments/form-options',
            f'/mini/market/{scope}/samples/form-options',
            f'/mini/market/{scope}/freights/form-options',
        ):
            assert path in paths
        for module in ('presign', 'protocol', 'payment', 'sample'):
            upload_path = f'/mini/market/{scope}/{module}/attachments'
            assert upload_path in paths
            upload_route = next(route for route in router.routes if route.path == upload_path)
            assert [param.name for param in upload_route.dependant.query_params] == []
        assert paths.index(f'/mini/market/{scope}/presign-orders/form-options') < paths.index(
            f'/mini/market/{scope}/presign-orders/{{order_id}}'
        )
        assert paths.index(f'/mini/market/{scope}/protocols/form-options') < paths.index(
            f'/mini/market/{scope}/protocols/{{protocol_id}}'
        )
        assert paths.index(f'/mini/market/{scope}/samples/form-options') < paths.index(
            f'/mini/market/{scope}/samples/{{sample_id}}'
        )


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

    for scope in ('specialist', 'manager'):
        expected = {
            f'/mini/market/{scope}/presign-orders/{{order_id}}': {'GET', 'PUT', 'DELETE'},
            f'/mini/market/{scope}/presign-orders/{{order_id}}/transfers': {'POST'},
            f'/mini/market/{scope}/presign-transfers/{{transfer_id}}': {'PUT', 'DELETE'},
            f'/mini/market/{scope}/presign-transfers/{{transfer_id}}/submit': {'POST'},
            f'/mini/market/{scope}/presign-transfers/{{transfer_id}}/withdraw': {'POST'},
        }
        for path, methods in expected.items():
            registered = {
                method
                for route in router.routes
                if route.path == path
                for method in route.methods
            }
            assert registered == methods


def test_presign_write_routes_map_permission_and_validation_errors(monkeypatch):
    from app.api import market_mobile
    from app.schemas.market_mobile import MarketPresignTransferBody

    delete_route = next(
        route
        for route in market_mobile.router.routes
        if route.path == '/mini/market/specialist/presign-orders/{order_id}'
        and route.methods == {'DELETE'}
    )
    create_transfer_route = next(
        route
        for route in market_mobile.router.routes
        if route.path
        == '/mini/market/specialist/presign-orders/{order_id}/transfers'
    )
    monkeypatch.setattr(
        market_mobile.market_presign,
        'delete_presign_order',
        lambda *args, **kwargs: (_ for _ in ()).throw(PermissionError('forbidden')),
    )
    permission_result = asyncio.run(
        delete_route.endpoint(order_id='o1', user=_user(), db=SimpleNamespace())
    )
    assert permission_result['code'] == 403

    monkeypatch.setattr(
        market_mobile.market_presign,
        'create_presign_transfer',
        lambda *args, **kwargs: (_ for _ in ()).throw(ValueError('invalid')),
    )
    validation_result = asyncio.run(
        create_transfer_route.endpoint(
            order_id='o1',
            body=MarketPresignTransferBody(transferAmount=100),
            user=_user(),
            db=SimpleNamespace(),
        )
    )
    assert validation_result['code'] == 400


def test_attachment_routes_read_only_their_module_upload_limit(monkeypatch):
    from app.api import market_mobile

    class TrackingUpload:
        filename = 'proof.pdf'
        content_type = 'application/pdf'

        def __init__(self):
            self.read_size = None

        async def read(self, size):
            self.read_size = size
            return b'%PDF-safe'

        async def close(self):
            return None

    monkeypatch.setattr(
        market_mobile.market_support,
        'save_market_attachment',
        lambda *args, **kwargs: {'id': 'attachment'},
    )
    routes = {
        route.path: route
        for route in market_mobile.router.routes
        if route.path in {
            '/mini/market/specialist/presign/attachments',
            '/mini/market/specialist/payment/attachments',
        }
    }
    presign_file = TrackingUpload()
    payment_file = TrackingUpload()

    asyncio.run(
        routes['/mini/market/specialist/presign/attachments'].endpoint(
            file=presign_file, user=_user(), db=SimpleNamespace()
        )
    )
    asyncio.run(
        routes['/mini/market/specialist/payment/attachments'].endpoint(
            file=payment_file, user=_user(), db=SimpleNamespace()
        )
    )

    assert presign_file.read_size == 20 * 1024 * 1024 + 1
    assert payment_file.read_size == 10 * 1024 * 1024 + 1