138 lines
6.3 KiB
Python
138 lines
6.3 KiB
Python
"""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'
|