zkxw/backend/tests/test_market_support_api.py

import asyncio
from decimal import Decimal
from io import BytesIO
from types import SimpleNamespace
from zipfile import ZipFile

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,
    BizCollectionProtocol,
    BizCollectionRegion,
    BizCollectionRegionStaff,
    BizItemUnit,
    BizProtocolCollectionInfo,
    BizProtocolFullmoonTodo,
    BizProtocolPayment,
    BizProtocolPaymentDetail,
    BizProtocolSample,
    BizProtocolStorageCert,
    BizSampleCode,
    SysPost,
    SysPostRole,
    SysRole,
    SysRoleDataScopeDetail,
    SysDictData,
    SysEthnic,
    SysUser,
    SysUserPost,
    SysUserRole,
)
from app.models.research_donation import (
    BizResearchDonation,
    BizResearchDonationCollection,
    BizResearchDonationSample,
)
from app.services import market_support
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__,
            BizCollectionProtocol.__table__,
            BizCollectionRegion.__table__,
            BizCollectionRegionStaff.__table__,
            BizItemUnit.__table__,
            BizProtocolSample.__table__,
            BizResearchDonationSample.__table__,
            BizSampleCode.__table__,
            SysDictData.__table__,
            SysEthnic.__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',
                storage_price=Decimal('25880.00'),
                storage_years=20,
                tenant_no='tenant-a',
                create_dept_id='',
                del_flag=0,
            ),
            BizCellSampleType(
                id='type-x',
                type_code='X',
                type_name='跨租户秘密样本',
                storage_price=Decimal('9999999999999999.99'),
                storage_years=99,
                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),
            BizSampleCode(id='sample-code-a', sample_code='S-001', tenant_no='tenant-a', create_dept_id='', del_flag=0),
            SysEthnic(id='ethnic-han', ethnic_name='汉族', ethnic_code='01', tenant_no='', 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_list_all_tenant_regions_like_web(monkeypatch):
    """市场专员采集地区选项应与 Web 一致：本租户全部地区，不依赖人员绑定。"""
    from app.services.market_support import get_form_options

    db = _session()
    _seed_options(db)
    # 未绑定任何采集地区的专员仍应能选到租户内地区（对齐 Web 预签表单）
    unbound = get_form_options(
        db,
        module='presign',
        ctx=_ctx('specialist', 'staff-unbound'),
        user_info=_user('staff-unbound'),
    )
    assert unbound['regions'] == [
        {'value': 'ra', 'label': 'A区'},
        {'value': 'rb', 'label': 'B区'},
    ]
    assert '跨租户' not in str(unbound)

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

    assert out['regions'] == [
        {'value': 'ra', 'label': 'A区'},
        {'value': 'rb', 'label': 'B区'},
    ]
    assert out['hospitals'] == [
        {'value': 'ha', 'label': 'A院'},
        {'value': 'hb', 'label': 'B院'},
    ]
    assert {row['value'] for row in out['staff']} >= {
        'leader-a',
        'staff-a',
        'user-a',
        'user-b',
    }
    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 out['currentUser'] == {'value': 'user-a', 'label': '本人'}
    # 当前登录人关联采集地区（新增协议单默认采集地区用）
    assert out['currentUserRegions'] == [{'value': 'ra', 'label': 'A区'}]
    assert '跨租户' not in str(out)

    scoped = get_form_options(
        db,
        module='presign',
        ctx=_ctx(),
        user_info=_user(),
        region_id='ra',
    )
    assert scoped['regions'] == [
        {'value': 'ra', 'label': 'A区'},
        {'value': 'rb', 'label': 'B区'},
    ]
    assert scoped['hospitals'] == [{'value': 'ha', 'label': 'A院'}]
    assert scoped['staff'] == [
        {'value': 'leader-a', 'label': 'A负责人', 'isLeader': True},
        {'value': 'staff-a', 'label': 'A员工', 'isLeader': False},
        {'value': 'user-a', 'label': '本人', 'isLeader': False},
    ]

    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_form_options_current_user_regions_keep_staff_binding_order():
    """多地区绑定时 currentUserRegions 按人员地区表顺序返回，供前端默认第一个。"""
    from app.services.market_support import get_form_options

    db = _session()
    _seed_options(db)
    db.add(
        BizCollectionRegionStaff(
            id='rsa-b',
            region_id='rb',
            user_id='user-a',
            is_leader=0,
            order_num=20,
            tenant_no='tenant-a',
            create_dept_id='',
            del_flag=0,
        )
    )
    # 已有 rsa 绑定 ra；显式调高 order_num 保证 ra 在前
    rsa = db.query(BizCollectionRegionStaff).filter_by(id='rsa').one()
    rsa.order_num = 10
    db.commit()

    out = get_form_options(db, module='protocol', ctx=_ctx(), user_info=_user())
    assert out['currentUserRegions'] == [
        {'value': 'ra', 'label': 'A区'},
        {'value': 'rb', 'label': 'B区'},
    ]

    unbound = get_form_options(
        db,
        module='protocol',
        ctx=_ctx('specialist', 'staff-unbound'),
        user_info=_user('staff-unbound'),
    )
    assert unbound['currentUserRegions'] == []


def test_manager_form_options_list_all_tenant_regions_like_web():
    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': 'ra', 'label': 'A区'},
        {'value': 'rb', 'label': 'B区'},
    ]
    assert out['hospitals'] == [{'value': 'hb', 'label': 'B院'}]
    assert out['ethnicities'] == [{'value': '汉族', 'label': '汉族'}]
    assert out['sampleCodes'] == [
        {'value': 'sample-code-a', 'label': 'S-001', 'sampleCode': 'S-001'}
    ]

    # 级联医院可按任意本租户地区加载；跨租户地区仍拒绝
    other = get_form_options(
        db,
        module='protocol',
        ctx=_ctx('manager', 'user-b'),
        user_info=_user('user-b'),
        region_id='ra',
    )
    assert other['hospitals'] == [{'value': 'ha', 'label': 'A院'}]

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


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

    db = _session()
    _seed_options(db)
    max_type = BizCellSampleType(
        id='type-max',
        type_code='MAX',
        type_name='最大金额样本',
        storage_price=Decimal('9999999999999999.99'),
        storage_years=30,
        tenant_no='tenant-a',
        create_dept_id='',
        del_flag=0,
    )
    null_type = BizCellSampleType(
        id='type-null',
        type_code='NULL',
        type_name='未配置样本',
        storage_price=None,
        storage_years=None,
        tenant_no='tenant-a',
        create_dept_id='',
        del_flag=0,
    )
    db.add_all([max_type, null_type])
    # SQLite 会把超出双精度安全范围的 NUMERIC 值以 REAL 回读；保持本次
    # flush 后的 Decimal 实例在 identity map 中，测试生产 MySQL 返回 Decimal
    # 时的实际 API 序列化路径，而不是 SQLite 方言自身的精度损失。
    db.flush()

    out = get_form_options(
        db,
        module='protocol',
        ctx=_ctx('manager', 'user-b'),
        user_info=_user('user-b'),
        region_id='rb',
    )
    sample_types = {item['value']: item for item in out['sampleTypes']}

    assert sample_types == {
        'type-a': {
            'value': 'type-a',
            'label': '样本A',
            'storagePrice': '25880.00',
            'storageYears': 20,
        },
        'type-max': {
            'value': 'type-max',
            'label': '最大金额样本',
            'storagePrice': '9999999999999999.99',
            'storageYears': 30,
        },
        'type-null': {
            'value': 'type-null',
            'label': '未配置样本',
            'storagePrice': None,
            'storageYears': None,
        },
    }
    assert 'type-x' not in sample_types


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

    db = _session()
    _seed_options(db)
    db.add(BizSampleCode(id='sample-code-b', sample_code='S-002', tenant_no='tenant-a', create_dept_id='', del_flag=0))
    db.add(
        BizCollectionProtocol(
            id='proto-used',
            protocol_no='CP-USED',
            tenant_no='tenant-a',
            create_dept_id='',
            del_flag=0,
        )
    )
    db.add(
        BizProtocolSample(
            id='ps-used',
            collection_protocol_id='proto-used',
            sample_code='S-001',
            tenant_no='tenant-a',
            create_dept_id='',
            del_flag=0,
        )
    )
    db.commit()

    out = get_form_options(
        db,
        module='protocol',
        ctx=_ctx('manager', 'user-b'),
        user_info=_user('user-b'),
        region_id='rb',
    )
    assert out['sampleCodes'] == [
        {'value': 'sample-code-b', 'label': 'S-002', 'sampleCode': 'S-002'}
    ]
    assert out['transferTypes'] == [{'value': 'wx', 'label': '微信'}]
    assert out['transportModes'] == [{'value': 'zxc', 'label': '专线车'}]


def test_presign_form_options_search_active_tenant_users_without_pii():
    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(),
        user_keyword='员工',
    )

    assert out['users'] == [{'value': 'staff-a', 'label': 'A员工'}]
    assert '跨租户' not in str(out['users'])
    assert '停用' not in str(out['users'])
    assert 'userNo' not in str(out['users'])


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_and_protocol_upload_allow_20_mib_while_payment_remains_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)
    protocol_result = save_market_attachment(
        _session(),
        content=content,
        filename='protocol-proof.pdf',
        content_type='application/pdf',
        module='protocol',
        ctx=_ctx(),
        user_info=_user(),
        storage_backend=_Storage(),
    )
    assert protocol_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(),
        )


def _zip_payload(path):
    output = BytesIO()
    with ZipFile(output, 'w') as archive:
        archive.writestr('[Content_Types].xml', '<Types/>')
        archive.writestr(path, 'payload')
    return output.getvalue()


@pytest.mark.parametrize(
    ('content', 'filename', 'content_type'),
    [
        (_zip_payload('ppt/presentation.xml'), 'slides.pptx', 'application/vnd.openxmlformats-officedocument.presentationml.presentation'),
        (b'plain text', 'notes.txt', 'text/plain'),
        (_zip_payload('folder/readme.txt'), 'archive.zip', 'application/zip'),
        (b'Rar!\x1a\x07\x01\x00' + b'a' * 8, 'archive.rar', 'application/vnd.rar'),
        (b'\x00\x00\x00\x18ftypmp42' + b'a' * 12, 'video.mp4', 'video/mp4'),
    ],
)
def test_protocol_upload_accepts_web_protocol_attachment_formats(content, filename, content_type):
    from app.services.market_support import save_market_attachment

    result = save_market_attachment(
        _session(),
        content=content,
        filename=filename,
        content_type=content_type,
        module='protocol',
        ctx=_ctx(),
        user_info=_user(),
        storage_backend=_Storage(),
    )

    assert result['fileName'] == filename


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

    detail = {
        'source': 'cell_storage',
        'basic': {'protocolStatus': 'sxz', 'paymentModelVersion': 2},
        'samplePayments': [{
            'id': 'pay-1',
            'schedules': [{
                'id': 'schedule-1',
                'scheduleStatus': 'open',
                'arrearsAmount': 100,
                'transfers': [
                    {'id': 'draft', 'payStatus': 'dfj'},
                    {'id': 'submitted', 'payStatus': 'jfqrz'},
                    {'id': 'confirmed', 'payStatus': 'yjf'},
                ],
            }],
        }],
    }

    readonly = decorate_allowed_actions(
        'protocol', detail, scope_kind='specialist', has_payment_permission=False
    )
    assert readonly['samplePayments'][0]['schedules'][0]['allowedActions'] == []
    assert [row['allowedActions'] for row in readonly['samplePayments'][0]['schedules'][0]['transfers']] == [[], [], []]

    writable = decorate_allowed_actions(
        'protocol', detail, scope_kind='specialist', has_payment_permission=True
    )
    schedule = writable['samplePayments'][0]['schedules'][0]
    assert schedule['allowedActions'] == ['register-transfer']
    assert [row['allowedActions'] for row in schedule['transfers']] == [
        ['edit-transfer', 'delete-payment-transfer', 'submit-finance'],
        ['withdraw-transfer'],
        [],
    ]


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


class _DownloadStorage:
    def __init__(self):
        self.requests = []

    def presigned_download_url(self, file_url, file_name):
        self.requests.append((file_url, file_name))
        return 'https://files.test/signed-download'


def _protocol_attachment_session():
    engine = create_engine('sqlite:///:memory:')
    Base.metadata.create_all(
        engine,
        tables=[
            BizAttachment.__table__,
            BizCollectionProtocol.__table__,
            BizCollectionRegionStaff.__table__,
            BizProtocolCollectionInfo.__table__,
            BizProtocolFullmoonTodo.__table__,
            BizProtocolPayment.__table__,
            BizProtocolPaymentDetail.__table__,
            BizProtocolStorageCert.__table__,
            BizResearchDonation.__table__,
            BizResearchDonationCollection.__table__,
            SysPost.__table__,
            SysPostRole.__table__,
            SysRole.__table__,
            SysRoleDataScopeDetail.__table__,
            SysUserPost.__table__,
            SysUserRole.__table__,
        ],
    )
    db = sessionmaker(bind=engine)()
    db.add(
        SysRole(
            id='market-role',
            role_name='市场移动端',
            data_scope='1',
            status='1',
            tenant_no='tenant-a',
            create_dept_id='',
            del_flag=0,
        )
    )
    for user_id in ('sales-a', 'sales-b', 'manager-a'):
        db.add(
            SysUserRole(
                id=f'user-role-{user_id}',
                user_id=user_id,
                role_id='market-role',
                tenant_no='tenant-a',
                create_dept_id='',
                del_flag=0,
            )
        )
    db.add(
        BizCollectionRegionStaff(
            id='manager-a-region-a',
            region_id='region-a',
            user_id='manager-a',
            is_leader=1,
            tenant_no='tenant-a',
            create_dept_id='',
            del_flag=0,
        )
    )
    db.commit()
    return db


def _market_user(user_id):
    return {
        'user_id': user_id,
        'tenant_no': 'tenant-a',
        'role': 'market_user',
        'dept_id': 'dept-a',
    }


def _add_protocol_attachment_reference(
    db,
    protocol_id,
    *,
    field,
    value,
    sales_user_id='sales-a',
    region_id='region-a',
    manager_user_id='manager-b',
    parent_tenant_no='tenant-a',
    parent_del_flag=0,
    collection_tenant_no=None,
    collection_del_flag=0,
    child_tenant_no='tenant-a',
    child_del_flag=0,
):
    db.add_all(
        [
            BizCollectionProtocol(
                id=protocol_id,
                protocol_no=f'P-{protocol_id}',
                tenant_no=parent_tenant_no,
                create_by=sales_user_id,
                create_dept_id='dept-a',
                del_flag=parent_del_flag,
            ),
            BizProtocolCollectionInfo(
                id=f'collection-{protocol_id}',
                collection_protocol_id=protocol_id,
                region_id=region_id,
                sales_user_id=sales_user_id,
                region_manager_user_id=manager_user_id,
                tenant_no=(
                    collection_tenant_no
                    if collection_tenant_no is not None
                    else parent_tenant_no
                ),
                create_dept_id='dept-a',
                del_flag=collection_del_flag,
            ),
        ]
    )
    model = (
        BizProtocolStorageCert
        if field == 'cert_image_urls'
        else BizProtocolFullmoonTodo
    )
    db.add(
        model(
            id=f'child-{protocol_id}',
            collection_protocol_id=protocol_id,
            tenant_no=child_tenant_no,
            create_dept_id='dept-a',
            del_flag=child_del_flag,
            **{field: value},
        )
    )
    db.commit()


def _protocol_download_ctx(scope_kind, user_id):
    return MarketFieldContext('tenant-a', user_id, scope_kind)


def test_protocol_cert_and_insurance_references_use_real_scope_and_exact_urls():
    db = _protocol_attachment_session()
    _add_protocol_attachment_reference(
        db,
        'own-cert',
        field='cert_image_urls',
        value=' files/cert.pdf.evil , files/cert-exact.pdf ',
    )
    _add_protocol_attachment_reference(
        db,
        'own-insurance',
        field='insurance_attachments',
        value='files/insurance.pdf',
    )

    assert market_support._has_accessible_market_attachment_reference(
        db,
        module='protocol',
        candidates=['files/cert-exact.pdf'],
        ctx=_protocol_download_ctx('specialist', 'sales-a'),
        user_info=_market_user('sales-a'),
    )
    assert market_support._has_accessible_market_attachment_reference(
        db,
        module='protocol',
        candidates=['files/insurance.pdf'],
        ctx=_protocol_download_ctx('manager', 'manager-a'),
        user_info=_market_user('manager-a'),
    )
    assert not market_support._has_accessible_market_attachment_reference(
        db,
        module='protocol',
        candidates=['files/cert.pdf'],
        ctx=_protocol_download_ctx('specialist', 'sales-a'),
        user_info=_market_user('sales-a'),
    )
    assert not market_support._has_accessible_market_attachment_reference(
        db,
        module='protocol',
        candidates=['files/cert-exact.pdf.evil'],
        ctx=_protocol_download_ctx('specialist', 'sales-a'),
        user_info=_market_user('sales-a'),
    )
    assert not market_support._has_accessible_market_attachment_reference(
        db,
        module='protocol',
        candidates=['files/cert-exact.pdf'],
        ctx=_protocol_download_ctx('specialist', 'sales-b'),
        user_info=_market_user('sales-b'),
    )
    db.add_all(
        [
            _attachment(
                id='short-url',
                file_url='files/cert.pdf',
                business_type='market:specialist:protocol:bound',
            ),
            _attachment(
                id='long-url',
                file_url='files/cert-exact.pdf.evil',
                business_type='market:specialist:protocol:bound',
            ),
            _attachment(
                id='other-specialist-scope',
                file_url='files/cert-exact.pdf',
                business_type='market:specialist:protocol:bound',
            ),
        ]
    )
    db.commit()
    for file_url in ('files/cert.pdf', 'files/cert-exact.pdf.evil'):
        with pytest.raises(PermissionError, match='附件'):
            market_support.get_market_attachment_download_url(
                db,
                file_url=file_url,
                file_name=None,
                module='protocol',
                ctx=_protocol_download_ctx('specialist', 'sales-a'),
                user_info=_market_user('sales-a'),
                storage_backend=_DownloadStorage(),
            )
    with pytest.raises(PermissionError, match='附件'):
        market_support.get_market_attachment_download_url(
            db,
            file_url='files/cert-exact.pdf',
            file_name=None,
            module='protocol',
            ctx=_protocol_download_ctx('specialist', 'sales-b'),
            user_info=_market_user('sales-b'),
            storage_backend=_DownloadStorage(),
        )


@pytest.mark.parametrize(
    ('attachment_id', 'file_url', 'business_type', 'field', 'scope_kind', 'user_id'),
    [
        ('legacy-cert', 'files/legacy-cert.pdf', '', 'cert_image_urls', 'specialist', 'sales-a'),
        (
            'protocol-cert',
            'files/protocol-cert.pdf',
            'market:specialist:protocol:bound',
            'cert_image_urls',
            'specialist',
            'sales-a',
        ),
        (
            'protocol-insurance',
            'files/protocol-insurance.pdf',
            'market:manager:protocol:bound',
            'insurance_attachments',
            'manager',
            'manager-a',
        ),
        (
            'sample-cert',
            'files/sample-cert.pdf',
            'market:specialist:sample:bound',
            'cert_image_urls',
            'specialist',
            'sales-a',
        ),
        (
            'sample-insurance',
            'files/sample-insurance.pdf',
            'market:manager:sample:bound',
            'insurance_attachments',
            'manager',
            'manager-a',
        ),
    ],
)
def test_protocol_download_allows_accessible_cert_and_insurance_attachment_history(
    attachment_id,
    file_url,
    business_type,
    field,
    scope_kind,
    user_id,
):
    db = _protocol_attachment_session()
    _add_protocol_attachment_reference(
        db,
        f'protocol-{attachment_id}',
        field=field,
        value=file_url,
    )
    db.add(
        _attachment(
            id=attachment_id,
            file_url=file_url,
            business_type=business_type,
            business_id=f'protocol-{attachment_id}',
            create_by='historical-uploader',
        )
    )
    db.commit()

    result = market_support.get_market_attachment_download_url(
        db,
        file_url=file_url,
        file_name=None,
        module='protocol',
        ctx=_protocol_download_ctx(scope_kind, user_id),
        user_info=_market_user(user_id),
        storage_backend=_DownloadStorage(),
    )

    assert result == {'downloadUrl': 'https://files.test/signed-download'}


def test_protocol_download_does_not_use_plain_protocol_reference_for_sample_bound():
    db = _protocol_attachment_session()
    _add_protocol_attachment_reference(
        db,
        'plain-protocol-reference',
        field='cert_image_urls',
        value='files/different-cert.pdf',
    )
    protocol = db.get(BizCollectionProtocol, 'plain-protocol-reference')
    protocol.protocol_attachments = 'files/plain-protocol.pdf'
    db.add(
        _attachment(
            id='plain-protocol-sample-bound',
            file_url='files/plain-protocol.pdf',
            business_type='market:specialist:sample:bound',
            business_id='plain-protocol-reference',
        )
    )
    db.commit()

    with pytest.raises(PermissionError, match='附件'):
        market_support.get_market_attachment_download_url(
            db,
            file_url='files/plain-protocol.pdf',
            file_name=None,
            module='protocol',
            ctx=_protocol_download_ctx('specialist', 'sales-a'),
            user_info=_market_user('sales-a'),
            storage_backend=_DownloadStorage(),
        )


@pytest.mark.parametrize(
    (
        'case',
        'parent_tenant_no',
        'parent_del_flag',
        'collection_tenant_no',
        'collection_del_flag',
        'child_tenant_no',
        'child_del_flag',
    ),
    [
        ('cross-parent-tenant', 'tenant-b', 0, 'tenant-b', 0, 'tenant-a', 0),
        ('deleted-parent', 'tenant-a', 1, 'tenant-a', 0, 'tenant-a', 0),
        ('cross-collection-tenant', 'tenant-a', 0, 'tenant-b', 0, 'tenant-a', 0),
        ('deleted-collection', 'tenant-a', 0, 'tenant-a', 1, 'tenant-a', 0),
        ('cross-child-tenant', 'tenant-a', 0, 'tenant-a', 0, 'tenant-b', 0),
        ('deleted-child', 'tenant-a', 0, 'tenant-a', 0, 'tenant-a', 1),
    ],
)
def test_protocol_cert_reference_rejects_tenant_scope_and_soft_delete_boundaries(
    case,
    parent_tenant_no,
    parent_del_flag,
    collection_tenant_no,
    collection_del_flag,
    child_tenant_no,
    child_del_flag,
):
    db = _protocol_attachment_session()
    file_url = f'files/{case}.pdf'
    _add_protocol_attachment_reference(
        db,
        case,
        field='cert_image_urls',
        value=file_url,
        parent_tenant_no=parent_tenant_no,
        parent_del_flag=parent_del_flag,
        collection_tenant_no=collection_tenant_no,
        collection_del_flag=collection_del_flag,
        child_tenant_no=child_tenant_no,
        child_del_flag=child_del_flag,
    )

    assert not market_support._has_accessible_market_attachment_reference(
        db,
        module='protocol',
        candidates=[file_url],
        ctx=_protocol_download_ctx('specialist', 'sales-a'),
        user_info=_market_user('sales-a'),
    )
    db.add(
        _attachment(
            id=f'attachment-{case}',
            file_url=file_url,
            business_type='market:specialist:protocol:bound',
        )
    )
    db.commit()
    with pytest.raises(PermissionError, match='附件'):
        market_support.get_market_attachment_download_url(
            db,
            file_url=file_url,
            file_name=None,
            module='protocol',
            ctx=_protocol_download_ctx('specialist', 'sales-a'),
            user_info=_market_user('sales-a'),
            storage_backend=_DownloadStorage(),
        )


@pytest.mark.parametrize(
    ('attachment_id', 'file_url', 'business_type', 'attachment_tenant', 'attachment_del_flag'),
    [
        ('sample-staging', 'files/sample-staging.pdf', 'market:specialist:sample:staging', 'tenant-a', 0),
        ('other-staging', 'files/other-staging.pdf', 'market:manager:protocol:staging', 'tenant-a', 0),
        ('cross-tenant', 'files/cross-tenant.pdf', '', 'tenant-b', 0),
        ('deleted-attachment', 'files/deleted-attachment.pdf', '', 'tenant-a', 1),
        ('unreferenced', 'files/unreferenced.pdf', '', 'tenant-a', 0),
    ],
)
def test_protocol_download_rejects_staging_cross_tenant_deleted_and_unreferenced_rows(
    attachment_id,
    file_url,
    business_type,
    attachment_tenant,
    attachment_del_flag,
):
    db = _protocol_attachment_session()
    referenced_url = file_url if attachment_id != 'unreferenced' else 'files/other.pdf'
    _add_protocol_attachment_reference(
        db,
        f'protocol-{attachment_id}',
        field='cert_image_urls',
        value=referenced_url,
    )
    db.add(
        _attachment(
            id=attachment_id,
            file_url=file_url,
            business_type=business_type,
            tenant_no=attachment_tenant,
            del_flag=attachment_del_flag,
        )
    )
    db.commit()

    with pytest.raises(PermissionError, match='附件'):
        market_support.get_market_attachment_download_url(
            db,
            file_url=file_url,
            file_name=None,
            module='protocol',
            ctx=_protocol_download_ctx('specialist', 'sales-a'),
            user_info=_market_user('sales-a'),
            storage_backend=_DownloadStorage(),
        )


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

    db = _session()
    db.add(_attachment())
    db.commit()
    backend = _DownloadStorage()

    result = get_market_attachment_download_url(
        db,
        file_url='market/specialist/payment/a.pdf',
        file_name='spoofed.exe',
        module='payment',
        ctx=_ctx(),
        user_info=_user(),
        storage_backend=backend,
    )

    assert result == {'downloadUrl': 'https://files.test/signed-download'}
    assert backend.requests == [
        ('market/specialist/payment/a.pdf', 'proof.pdf')
    ]


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

    db = _session()
    db.add_all(
        [
            _attachment(id='cross', file_url='cross.pdf', tenant_no='tenant-b'),
            _attachment(
                id='wrong-module',
                file_url='wrong-module.pdf',
                business_type='market:specialist:sample:bound',
                business_id='sample-a',
            ),
            _attachment(
                id='other-staging',
                file_url='other-staging.pdf',
                create_by='user-b',
            ),
        ]
    )
    db.commit()

    for file_url in ('cross.pdf', 'wrong-module.pdf', 'other-staging.pdf'):
        with pytest.raises(PermissionError, match='附件'):
            get_market_attachment_download_url(
                db,
                file_url=file_url,
                file_name=None,
                module='payment',
                ctx=_ctx(),
                user_info=_user(),
                storage_backend=_DownloadStorage(),
            )


def test_market_attachment_download_rejects_unreferenced_bound_and_legacy_rows(
    monkeypatch,
):
    from app.services import market_support

    db = _session()
    db.add_all(
        [
            _attachment(
                id='manager-bound',
                file_url='manager-bound.pdf',
                business_type='market:manager:payment:bound',
                business_id='transfer-a',
            ),
            _attachment(
                id='legacy',
                file_url='legacy.pdf',
                business_type=None,
                create_by='admin',
            ),
        ]
    )
    db.commit()
    monkeypatch.setattr(
        market_support,
        '_has_accessible_market_attachment_reference',
        lambda *args, **kwargs: False,
    )

    for file_url in ('manager-bound.pdf', 'legacy.pdf'):
        with pytest.raises(PermissionError, match='附件'):
            market_support.get_market_attachment_download_url(
                db,
                file_url=file_url,
                file_name=None,
                module='payment',
                ctx=_ctx(),
                user_info=_user(),
                storage_backend=_DownloadStorage(),
            )


def test_market_attachment_download_allows_accessible_referenced_legacy_row(monkeypatch):
    from app.services import market_support

    db = _session()
    db.add(
        _attachment(
            id='legacy',
            file_url='legacy.pdf',
            business_type=None,
            create_by='admin',
        )
    )
    db.commit()
    checked = []

    def allow_reference(_db, *, module, candidates, ctx, user_info):
        checked.append((module, candidates, ctx.user_id, user_info['user_id']))
        return True

    monkeypatch.setattr(
        market_support,
        '_has_accessible_market_attachment_reference',
        allow_reference,
    )

    result = market_support.get_market_attachment_download_url(
        db,
        file_url='legacy.pdf',
        file_name=None,
        module='payment',
        ctx=_ctx(),
        user_info=_user(),
        storage_backend=_DownloadStorage(),
    )

    assert result['downloadUrl'] == 'https://files.test/signed-download'
    assert checked == [('payment', ['legacy.pdf'], 'user-a', 'user-a')]


def test_protocol_attachment_reference_respects_specialist_donation_scope():
    from app.models.models import (
        BizCollectionProtocol,
        BizProtocolCollectionInfo,
        BizProtocolPayment,
        BizProtocolPaymentDetail,
    )
    from app.models.research_donation import (
        BizResearchDonation,
        BizResearchDonationCollection,
    )
    from app.services import market_support

    engine = create_engine('sqlite:///:memory:')
    Base.metadata.create_all(
        engine,
        tables=[
            BizCollectionProtocol.__table__,
            BizProtocolCollectionInfo.__table__,
            BizProtocolStorageCert.__table__,
            BizProtocolFullmoonTodo.__table__,
            BizProtocolPayment.__table__,
            BizProtocolPaymentDetail.__table__,
            BizResearchDonation.__table__,
            BizResearchDonationCollection.__table__,
        ],
    )
    db = sessionmaker(bind=engine)()
    db.add_all(
        [
            BizResearchDonation(
                id='donation-a',
                donation_no='D-1',
                donation_status='sxz',
                source_type='direct',
                attachment_urls='legacy.pdf',
                tenant_no='tenant-a',
                create_dept_id='',
                del_flag=0,
            ),
            BizResearchDonationCollection(
                id='collection-a',
                research_donation_id='donation-a',
                sales_user_id='user-a',
                tenant_no='tenant-a',
                create_dept_id='',
                del_flag=0,
            ),
        ]
    )
    db.commit()
    super_admin_a = {**_user('user-a'), 'role': 'super_admin'}
    super_admin_b = {**_user('user-b'), 'role': 'super_admin'}

    assert market_support._has_accessible_market_attachment_reference(
        db,
        module='protocol',
        candidates=['legacy.pdf'],
        ctx=_ctx(user_id='user-a'),
        user_info=super_admin_a,
    )
    assert not market_support._has_accessible_market_attachment_reference(
        db,
        module='protocol',
        candidates=['legacy.pdf'],
        ctx=_ctx(user_id='user-b'),
        user_info=super_admin_b,
    )


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', 'delete']
    assert allowed_actions('protocol', status='yzz') == ['edit', 'delete']
    assert allowed_actions('protocol', status='yzf') == ['edit', 'delete']
    assert allowed_actions('payment-transfer', status='dfj') == [
        'edit-transfer',
        'delete-payment-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) == []
    assert allowed_actions('sample', status='', source='research_donation', has_sample_code=True) == []
    assert allowed_actions('sample', status='', source='cell_storage', has_sample_code=True) == []
    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_legacy_v1_protocol_detail_keeps_status_scoped_edit_and_delete():
    from app.services.market_support import decorate_allowed_actions

    result = decorate_allowed_actions(
        'protocol',
        {
            'source': 'cell_storage',
            'basic': {
                'protocolStatus': 'sxz',
                'paymentModelVersion': 1,
            },
        },
        scope_kind='specialist',
        detail=True,
    )

    assert result['allowedActions'] == ['edit', 'delete']


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',
                'arrearsAmount': '80.00',
                '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', 'delete-payment-transfer', 'submit-finance'],
        ['withdraw-transfer'],
        [],
    ]


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

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

    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', 'delete-payment-transfer', 'submit-finance'],
        ['withdraw-transfer'],
    ]


@pytest.mark.parametrize(
    ('schedule_status', 'arrears_amount', 'expected_schedule_actions'),
    [('cleared', '80.00', ['register-transfer']), ('open', '0.00', []), ('void', '80.00', [])],
)
def test_payment_schedule_actions_require_non_void_status_and_positive_arrears(
    schedule_status, arrears_amount, expected_schedule_actions
):
    from app.services.market_support import decorate_allowed_actions

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

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

    assert out['allowedActions'] == expected_schedule_actions
    assert out['schedules'][0]['allowedActions'] == expected_schedule_actions
    assert out['schedules'][0]['transfers'][0]['allowedActions'] == [
        'edit-transfer',
        'delete-payment-transfer',
        'submit-finance',
    ]


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', 'delete', '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', 'delete', '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', 'delete']
    assert terminal['allowedActions'] == ['edit', 'delete']


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}/freights/form-options',
        ):
            assert path in paths
        for module in ('presign', 'protocol', 'payment'):
            upload_path = f'/mini/market/{scope}/{module}/attachments'
            download_path = f'{upload_path}/download-url'
            assert upload_path in paths
            assert download_path in paths
            upload_route = next(route for route in router.routes if route.path == upload_path)
            download_route = next(route for route in router.routes if route.path == download_path)
            assert [param.name for param in upload_route.dependant.query_params] == []
            assert download_route.methods == {'POST'}
        assert f'/mini/market/{scope}/sample/attachments' not in paths
        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 f'/mini/market/{scope}/samples/form-options' not in paths


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_market_list_routes_map_invalid_filters_to_validation_response(monkeypatch):
    from app.api import market_mobile

    route_paths = {
        '/mini/market/specialist/protocols': market_mobile.market_protocol,
        '/mini/market/specialist/payments': market_mobile.market_payment,
        '/mini/market/specialist/freights': market_mobile.market_freight,
    }
    routes = {
        route.path: route
        for route in market_mobile.router.routes
        if route.path in route_paths and route.methods == {'GET'}
    }
    for path, service in route_paths.items():
        method_name = {
            '/mini/market/specialist/protocols': 'list_protocols',
            '/mini/market/specialist/payments': 'list_payments',
            '/mini/market/specialist/freights': 'list_freights',
        }[path]
        monkeypatch.setattr(
            service,
            method_name,
            lambda *args, **kwargs: (_ for _ in ()).throw(ValueError('invalid filter')),
        )
        result = asyncio.run(
            routes[path].endpoint(user=_user(), db=SimpleNamespace())
        )
        assert result['code'] == 400
        assert result['message'] == 'invalid filter'


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


def test_payment_transfer_response_carries_server_authorized_capabilities():
    draft = market_support.decorate_allowed_actions(
        'payment-transfer',
        {'id': 't1', 'payStatus': 'dfj'},
        scope_kind='specialist',
    )
    pending = market_support.decorate_allowed_actions(
        'payment-transfer',
        {'id': 't2', 'payStatus': 'jfqrz'},
        scope_kind='specialist',
    )
    assert draft['allowedActions'] == [
        'edit-transfer',
        'delete-payment-transfer',
        'submit-finance',
    ]
    assert pending['allowedActions'] == ['withdraw-transfer']