2160 lines
70 KiB
Python
2160 lines
70 KiB
Python
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']
|