zkxw/backend/tests/test_web_freight_permissions.py

"""Web 运费数据权限与写入边界，使用内存数据库避免改动业务数据。"""
import asyncio
from datetime import datetime
from decimal import Decimal

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

from app.api import sample_freight as api
from app.crud.sample_freight import _parse_amount
from app.models.base import Base
from app.models.models import (
    BizCollectionProtocol, BizProtocolSample, BizSampleFreight,
    BizSampleFreightItem, SysDictData, SysDictType, SysRole,
    SysRoleDataScopeDetail, SysUser, SysUserRole,
)

USER = {'user_id': 'u1', 'tenant_no': 'tenant-a', 'role': 'admin', 'dept_id': 'dept-a'}


@pytest.fixture
def db():
    engine = create_engine('sqlite:///:memory:')
    Base.metadata.create_all(engine)
    session = sessionmaker(bind=engine)()
    now = datetime.now()
    common = dict(tenant_no='tenant-a', create_dept_id='dept-a', create_time=now, del_flag=0)
    session.add_all([
        SysUser(id='u1', user_no='u1', nick_name='用户', password='', dept_id='dept-a', **common),
        SysRole(id='self', role_name='普通人员', data_scope='4', status='1', **common),
        SysUserRole(id='ur', user_id='u1', role_id='self', **common),
        SysDictType(id='dt', dict_name='运输方式', dict_type='sample_transport_mode', **common),
        SysDictData(id='dd', dict_label='顺丰', dict_value='sf', dict_type='sample_transport_mode', **common),
    ])
    for owner in ('u1', 'u2'):
        session.add_all([
            BizCollectionProtocol(id='p-' + owner, protocol_no='P-' + owner, create_by=owner, **common),
            BizProtocolSample(id='s-' + owner, collection_protocol_id='p-' + owner, sample_code='S-' + owner, create_by=owner, **common),
            BizSampleFreight(id='f-' + owner, freight_no='F-' + owner, transport_mode='sf', freight_amount=Decimal('10'), create_by=owner, **common),
        ])
    session.commit()
    yield session
    session.close()
    engine.dispose()


def call(endpoint, db, **kwargs):
    return asyncio.run(endpoint(db=db, user=USER, **kwargs))


def test_list_obeys_self_scope(db):
    result = call(api.list_sample_freights, db, page=1, pageSize=10, includeItems=False)
    assert [r['id'] for r in result['data']['list']] == ['f-u1']
    assert result['data']['total'] == 1


def test_multiple_roles_union_preserves_allowed_records(db):
    common = dict(tenant_no='tenant-a', create_dept_id='', del_flag=0)
    db.add_all([
        SysRole(id='custom', role_name='扩展权限', data_scope='5', status='1', **common),
        SysUserRole(id='ur2', user_id='u1', role_id='custom', **common),
        SysRoleDataScopeDetail(id='detail', role_id='custom', scope_type='user', scope_value='u2', **common),
    ])
    db.commit()
    result = call(api.list_sample_freights, db, page=1, pageSize=10, includeItems=False)
    assert {r['id'] for r in result['data']['list']} == {'f-u1', 'f-u2'}


@pytest.mark.parametrize('operation', ['detail', 'update', 'delete', 'link', 'unlink'])
def test_outside_scope_batch_is_not_addressable(db, operation):
    endpoints = {
        'detail': (api.get_sample_freight, {}),
        'update': (api.update_sample_freight, {'body': api.SampleFreightUpdateBody(freightAmount=999)}),
        'delete': (api.delete_sample_freight, {}),
        'link': (api.link_sample_freight, {'body': api.SampleFreightItemsBody(items=[{'sampleSource': 'cell_storage', 'sampleId': 's-u1'}])}),
        'unlink': (api.unlink_sample_freight, {'body': api.SampleFreightItemsBody(items=[])}),
    }
    endpoint, kwargs = endpoints[operation]
    result = call(endpoint, db, freight_id='f-u2', **kwargs)
    assert result['code'] in (403, 404)
    obj = db.query(BizSampleFreight).filter_by(id='f-u2').one()
    assert obj.del_flag == 0
    assert obj.freight_amount == Decimal('10')


def test_sample_options_obey_parent_scope(db):
    result = call(api.list_sample_freight_options, db, page=1, pageSize=20, sampleSource='cell_storage')
    assert [r['sampleId'] for r in result['data']['list']] == ['s-u1']


@pytest.mark.parametrize('sample_id', ['s-u2', 'does-not-exist'])
@pytest.mark.parametrize('operation', ['create', 'link'])
def test_cannot_link_hidden_or_missing_sample(db, sample_id, operation):
    items = [{'sampleSource': 'cell_storage', 'sampleId': sample_id}]
    if operation == 'create':
        result = call(api.create_sample_freight, db, body=api.SampleFreightCreateBody(transportMode='sf', freightAmount=10, items=items))
    else:
        result = call(api.link_sample_freight, db, freight_id='f-u1', body=api.SampleFreightItemsBody(items=items))
    assert result['code'] in (400, 403, 404)
    assert db.query(BizSampleFreightItem).count() == 0
    assert db.query(BizSampleFreight).count() == 2


def test_created_batch_has_server_department_and_sample_snapshot(db):
    result = call(api.create_sample_freight, db, body=api.SampleFreightCreateBody(
        transportMode='sf', freightAmount=10,
        items=[{'sampleSource': 'cell_storage', 'sampleId': 's-u1', 'sampleCode': 'forged', 'protocolId': 'p-u2'}],
    ))
    assert result['code'] == 200
    row = db.query(BizSampleFreight).filter_by(id=result['data']['id']).one()
    assert row.create_dept_id == 'dept-a'
    assert result['data']['items'][0]['sampleCode'] == 'S-u1'
    assert result['data']['items'][0]['protocolId'] == 'p-u1'


@pytest.mark.parametrize('amount', ['NaN', 'Infinity', '-Infinity'])
def test_nonfinite_freight_amount_is_validation_error(amount):
    with pytest.raises(ValueError):
        _parse_amount(amount)


@pytest.mark.parametrize('operation', ['create', 'link'])
def test_link_cannot_remove_sample_from_an_inaccessible_batch(db, operation):
    db.add(BizSampleFreightItem(
        id='existing-link', freight_id='f-u2', sample_source='cell_storage',
        sample_id='s-u1', sample_code='S-u1', protocol_id='p-u1',
        tenant_no='tenant-a', create_dept_id='', del_flag=0,
    ))
    db.commit()
    items = [{'sampleSource': 'cell_storage', 'sampleId': 's-u1'}]
    if operation == 'create':
        result = call(api.create_sample_freight, db, body=api.SampleFreightCreateBody(transportMode='sf', freightAmount=10, items=items))
    else:
        result = call(api.link_sample_freight, db, freight_id='f-u1', body=api.SampleFreightItemsBody(items=items))
    assert result['code'] in (403, 404)
    assert db.query(BizSampleFreightItem).filter_by(id='existing-link').one().freight_id == 'f-u2'