1042 lines
36 KiB
Python
1042 lines
36 KiB
Python
import asyncio
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
from app.models.base import Base
|
|
from app.models.models import (
|
|
BizAttachment,
|
|
BizCellSampleType,
|
|
BizCollectionHospital,
|
|
BizCollectionRegion,
|
|
BizCollectionRegionStaff,
|
|
BizItemUnit,
|
|
SysDictData,
|
|
SysUser,
|
|
)
|
|
from app.services.market_field_scope import MarketFieldContext
|
|
|
|
|
|
def _session():
|
|
engine = create_engine('sqlite:///:memory:')
|
|
Base.metadata.create_all(
|
|
engine,
|
|
tables=[
|
|
BizAttachment.__table__,
|
|
BizCellSampleType.__table__,
|
|
BizCollectionHospital.__table__,
|
|
BizCollectionRegion.__table__,
|
|
BizCollectionRegionStaff.__table__,
|
|
BizItemUnit.__table__,
|
|
SysDictData.__table__,
|
|
SysUser.__table__,
|
|
],
|
|
)
|
|
return sessionmaker(bind=engine)()
|
|
|
|
|
|
def _ctx(scope='specialist', user_id='user-a'):
|
|
return MarketFieldContext('tenant-a', user_id, scope)
|
|
|
|
|
|
def _user(user_id='user-a'):
|
|
return {
|
|
'user_id': user_id,
|
|
'tenant_no': 'tenant-a',
|
|
'region_id': 'dept-a',
|
|
}
|
|
|
|
|
|
def _seed_options(db):
|
|
db.add_all(
|
|
[
|
|
BizCollectionRegion(id='ra', region_name='A区', tenant_no='tenant-a', create_dept_id='', del_flag=0),
|
|
BizCollectionRegion(id='rb', region_name='B区', tenant_no='tenant-a', create_dept_id='', del_flag=0),
|
|
BizCollectionRegion(id='rx', region_name='跨租户秘密区', tenant_no='tenant-b', create_dept_id='', del_flag=0),
|
|
BizCollectionHospital(id='ha', region_id='ra', hospital_name='A院', tenant_no='tenant-a', create_dept_id='', del_flag=0),
|
|
BizCollectionHospital(id='hb', region_id='rb', hospital_name='B院', tenant_no='tenant-a', create_dept_id='', del_flag=0),
|
|
BizCollectionHospital(id='hx', region_id='rx', hospital_name='跨租户秘密院', tenant_no='tenant-b', create_dept_id='', del_flag=0),
|
|
BizCollectionRegionStaff(id='rsa', region_id='ra', user_id='user-a', is_leader=0, tenant_no='tenant-a', create_dept_id='', del_flag=0),
|
|
BizCollectionRegionStaff(id='rsb', region_id='rb', user_id='user-b', is_leader=1, tenant_no='tenant-a', create_dept_id='', del_flag=0),
|
|
BizCollectionRegionStaff(id='rsa2', region_id='ra', user_id='staff-a', is_leader=0, tenant_no='tenant-a', create_dept_id='', del_flag=0),
|
|
BizCollectionRegionStaff(id='rsa3', region_id='ra', user_id='leader-a', is_leader=1, tenant_no='tenant-a', create_dept_id='', del_flag=0),
|
|
BizCollectionRegionStaff(id='rsx', region_id='rx', user_id='cross-user', is_leader=0, tenant_no='tenant-b', create_dept_id='', del_flag=0),
|
|
SysUser(id='user-a', user_no='ua', nick_name='本人', status='1', tenant_no='tenant-a', create_dept_id='', del_flag=0),
|
|
SysUser(id='user-b', user_no='ub', nick_name='B经理', status='1', tenant_no='tenant-a', create_dept_id='', del_flag=0),
|
|
SysUser(id='staff-a', user_no='sa', nick_name='A员工', status='1', tenant_no='tenant-a', create_dept_id='', del_flag=0),
|
|
SysUser(id='leader-a', user_no='la', nick_name='A负责人', status='1', tenant_no='tenant-a', create_dept_id='', del_flag=0),
|
|
SysUser(id='disabled', user_no='off', nick_name='停用秘密', status='0', tenant_no='tenant-a', create_dept_id='', del_flag=0),
|
|
SysUser(id='cross-user', user_no='xx', nick_name='跨租户秘密用户', status='1', tenant_no='tenant-b', create_dept_id='', del_flag=0),
|
|
BizCellSampleType(id='type-a', type_code='A', type_name='样本A', tenant_no='tenant-a', create_dept_id='', del_flag=0),
|
|
BizCellSampleType(id='type-x', type_code='X', type_name='跨租户秘密样本', tenant_no='tenant-b', create_dept_id='', del_flag=0),
|
|
BizItemUnit(id='unit-a', unit_code='ml', unit_name='毫升', tenant_no='tenant-a', create_dept_id='', del_flag=0),
|
|
SysDictData(id='dict-pay', dict_label='微信', dict_value='wx', dict_type='zzfs', tenant_no='tenant-a', create_dept_id='', del_flag=0),
|
|
SysDictData(id='dict-freight', dict_label='专线车', dict_value='zxc', dict_type='sample_transport_mode', tenant_no='tenant-a', create_dept_id='', del_flag=0),
|
|
SysDictData(id='dict-secret', dict_label='不应泄漏', dict_value='secret', dict_type='unrelated', tenant_no='tenant-a', create_dept_id='', del_flag=0),
|
|
]
|
|
)
|
|
db.commit()
|
|
|
|
|
|
def test_specialist_form_options_are_region_scoped_and_minimal(monkeypatch):
|
|
from app.services.market_support import get_form_options
|
|
|
|
db = _session()
|
|
_seed_options(db)
|
|
|
|
out = get_form_options(db, module='presign', ctx=_ctx(), user_info=_user())
|
|
|
|
assert out['regions'] == [{'value': 'ra', 'label': 'A区'}]
|
|
assert out['hospitals'] == [{'value': 'ha', 'label': 'A院'}]
|
|
assert out['staff'] == [
|
|
{'value': 'leader-a', 'label': 'A负责人', 'isLeader': True},
|
|
{'value': 'staff-a', 'label': 'A员工', 'isLeader': False},
|
|
{'value': 'user-a', 'label': '本人', 'isLeader': False},
|
|
]
|
|
assert out['sampleTypes'] == [{'value': 'type-a', 'label': '样本A'}]
|
|
assert out['units'] == [{'value': 'unit-a', 'label': '毫升'}]
|
|
assert out['payModes'] == [
|
|
{'label': '趸交', 'value': 'dj'},
|
|
{'label': '分期', 'value': 'fq'},
|
|
]
|
|
assert out['orderStatuses'] == [
|
|
{'label': '草稿', 'value': 'draft'},
|
|
{'label': '进行中', 'value': 'active'},
|
|
{'label': '已转正式', 'value': 'converted'},
|
|
{'label': '已作废', 'value': 'void'},
|
|
]
|
|
assert out['transferTypes'] == [{'value': 'wx', 'label': '微信'}]
|
|
assert '跨租户' not in str(out)
|
|
|
|
monkeypatch.setattr(
|
|
'app.services.market_support._business_region_id',
|
|
lambda *args, **kwargs: 'rb',
|
|
)
|
|
supplemented = get_form_options(
|
|
db,
|
|
module='presign',
|
|
ctx=_ctx(),
|
|
user_info=_user(),
|
|
business_id='own-document',
|
|
)
|
|
assert supplemented['regions'] == [
|
|
{'value': 'ra', 'label': 'A区'},
|
|
{'value': 'rb', 'label': 'B区'},
|
|
]
|
|
|
|
|
|
def test_manager_form_options_only_include_led_regions_and_requested_region_must_be_authorized():
|
|
from app.services.market_support import get_form_options
|
|
|
|
db = _session()
|
|
_seed_options(db)
|
|
|
|
out = get_form_options(
|
|
db,
|
|
module='protocol',
|
|
ctx=_ctx('manager', 'user-b'),
|
|
user_info=_user('user-b'),
|
|
region_id='rb',
|
|
)
|
|
assert out['regions'] == [{'value': 'rb', 'label': 'B区'}]
|
|
assert out['hospitals'] == [{'value': 'hb', 'label': 'B院'}]
|
|
|
|
with pytest.raises(PermissionError, match='地区'):
|
|
get_form_options(
|
|
db,
|
|
module='protocol',
|
|
ctx=_ctx('manager', 'user-b'),
|
|
user_info=_user('user-b'),
|
|
region_id='ra',
|
|
)
|
|
|
|
|
|
def test_payment_and_freight_options_use_fixed_dictionary_allowlist():
|
|
from app.services.market_support import get_form_options
|
|
|
|
db = _session()
|
|
_seed_options(db)
|
|
payment = get_form_options(db, module='payment', ctx=_ctx(), user_info=_user())
|
|
freight = get_form_options(db, module='freight', ctx=_ctx(), user_info=_user())
|
|
|
|
assert payment == {'transferTypes': [{'value': 'wx', 'label': '微信'}]}
|
|
assert freight == {'transportModes': [{'value': 'zxc', 'label': '专线车'}]}
|
|
assert '不应泄漏' not in str(payment) + str(freight)
|
|
|
|
|
|
class _Storage:
|
|
def __init__(self):
|
|
self.uploaded = []
|
|
self.deleted = []
|
|
|
|
def upload_file(self, content, filename, folder):
|
|
self.uploaded.append((content, filename, folder))
|
|
return f'https://files.test/{folder}/generated.pdf'
|
|
|
|
def delete_file(self, path):
|
|
self.deleted.append(path)
|
|
return True
|
|
|
|
|
|
def test_upload_uses_fixed_folder_and_records_current_tenant_and_uploader():
|
|
from app.services.market_support import save_market_attachment
|
|
|
|
db = _session()
|
|
backend = _Storage()
|
|
result = save_market_attachment(
|
|
db,
|
|
content=b'%PDF-1.7 safe',
|
|
filename='proof.pdf',
|
|
content_type='application/pdf',
|
|
module='payment',
|
|
ctx=_ctx(),
|
|
user_info=_user(),
|
|
storage_backend=backend,
|
|
)
|
|
|
|
assert backend.uploaded[0][2] == 'market/specialist/payment'
|
|
row = db.query(BizAttachment).one()
|
|
assert (row.tenant_no, row.create_by, row.create_dept_id) == ('tenant-a', 'user-a', 'dept-a')
|
|
assert row.business_type == 'market:specialist:payment:staging'
|
|
assert result['id'] == row.id
|
|
assert result['fileUrl'] == row.file_url
|
|
|
|
|
|
def test_presign_upload_allows_20_mib_while_other_modules_remain_at_10_mib():
|
|
from app.services.market_support import save_market_attachment
|
|
|
|
content = b'%PDF-' + b'a' * (10 * 1024 * 1024)
|
|
backend = _Storage()
|
|
|
|
result = save_market_attachment(
|
|
_session(),
|
|
content=content,
|
|
filename='presign-proof.pdf',
|
|
content_type='application/pdf',
|
|
module='presign',
|
|
ctx=_ctx(),
|
|
user_info=_user(),
|
|
storage_backend=backend,
|
|
)
|
|
|
|
assert result['fileSize'] == len(content)
|
|
with pytest.raises(ValueError, match='10MiB'):
|
|
save_market_attachment(
|
|
_session(),
|
|
content=content,
|
|
filename='payment-proof.pdf',
|
|
content_type='application/pdf',
|
|
module='payment',
|
|
ctx=_ctx(),
|
|
user_info=_user(),
|
|
storage_backend=_Storage(),
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
('content', 'filename', 'content_type'),
|
|
[
|
|
(b'', 'empty.pdf', 'application/pdf'),
|
|
(b'<svg/>', 'bad.svg', 'image/svg+xml'),
|
|
(b'MZ', 'bad.exe', 'application/octet-stream'),
|
|
(b'not an image', 'fake.jpg', 'image/jpeg'),
|
|
(b'%PDF-1.7', 'fake.jpg', 'image/jpeg'),
|
|
(b'\x00\x00\x00\x18junk-ftyp-heic-data', 'fake.heic', 'image/heic'),
|
|
(b'\x00\x00\x00\x18ftypzzzz\x00\x00\x00\x00zzzz', 'fake.heic', 'image/heic'),
|
|
(b'\x00\x00\x00\xffftypheic\x00\x00\x00\x00', 'fake.heic', 'image/heic'),
|
|
(b'\x00\x00\x00\x0cftypheic', 'short.heic', 'image/heic'),
|
|
(b'\x00\x00\x00\x0fftypheicabc', 'short.heic', 'image/heic'),
|
|
(b'\x00\x00\x00\x15ftypheic\x00\x00\x00\x00mif1x', 'misaligned.heic', 'image/heic'),
|
|
(b'\x00\x00\x00\x10ftypmif1\x00\x00\x00\x00', 'generic-only.heic', 'image/heic'),
|
|
(b'PK\x03\x04not-a-docx', 'fake.docx', 'application/vnd.openxmlformats-officedocument.wordprocessingml.document'),
|
|
(b'a' * (10 * 1024 * 1024 + 1), 'huge.pdf', 'application/pdf'),
|
|
],
|
|
)
|
|
def test_upload_rejects_empty_oversized_or_unsafe_files(content, filename, content_type):
|
|
from app.services.market_support import save_market_attachment
|
|
|
|
db = _session()
|
|
backend = _Storage()
|
|
with pytest.raises(ValueError):
|
|
save_market_attachment(
|
|
db,
|
|
content=content,
|
|
filename=filename,
|
|
content_type=content_type,
|
|
module='payment',
|
|
ctx=_ctx(),
|
|
user_info=_user(),
|
|
storage_backend=backend,
|
|
)
|
|
assert backend.uploaded == []
|
|
|
|
|
|
def test_upload_deletes_storage_object_when_database_commit_fails(monkeypatch):
|
|
from app.services.market_support import save_market_attachment
|
|
|
|
db = _session()
|
|
backend = _Storage()
|
|
monkeypatch.setattr(db, 'commit', lambda: (_ for _ in ()).throw(RuntimeError('db down')))
|
|
|
|
with pytest.raises(RuntimeError, match='db down'):
|
|
save_market_attachment(
|
|
db,
|
|
content=b'%PDF-1.7 safe',
|
|
filename='proof.pdf',
|
|
content_type='application/pdf',
|
|
module='protocol',
|
|
ctx=_ctx(),
|
|
user_info=_user(),
|
|
storage_backend=backend,
|
|
)
|
|
assert backend.deleted == ['https://files.test/market/specialist/protocol/generated.pdf']
|
|
|
|
|
|
def test_upload_never_refreshes_after_commit_or_deletes_committed_object(monkeypatch):
|
|
from app.services.market_support import save_market_attachment
|
|
|
|
db = _session()
|
|
backend = _Storage()
|
|
committed = False
|
|
original_commit = db.commit
|
|
original_refresh = db.refresh
|
|
|
|
def tracked_commit():
|
|
nonlocal committed
|
|
original_commit()
|
|
committed = True
|
|
|
|
def refresh_only_before_commit(row):
|
|
if committed:
|
|
raise RuntimeError('refresh after commit is unsafe')
|
|
return original_refresh(row)
|
|
|
|
monkeypatch.setattr(db, 'commit', tracked_commit)
|
|
monkeypatch.setattr(db, 'refresh', refresh_only_before_commit)
|
|
|
|
result = save_market_attachment(
|
|
db,
|
|
content=b'%PDF-1.7 safe',
|
|
filename='proof.pdf',
|
|
content_type='application/pdf',
|
|
module='protocol',
|
|
ctx=_ctx(),
|
|
user_info=_user(),
|
|
storage_backend=backend,
|
|
)
|
|
|
|
assert result['id']
|
|
assert backend.deleted == []
|
|
|
|
|
|
def test_upload_accepts_well_formed_heif_ftyp_box_with_allowed_brand():
|
|
from app.services.market_support import save_market_attachment
|
|
|
|
db = _session()
|
|
backend = _Storage()
|
|
content = b'\x00\x00\x00\x14ftypheic\x00\x00\x00\x00mif1'
|
|
|
|
result = save_market_attachment(
|
|
db,
|
|
content=content,
|
|
filename='photo.heic',
|
|
content_type='image/heic',
|
|
module='sample',
|
|
ctx=_ctx(),
|
|
user_info=_user(),
|
|
storage_backend=backend,
|
|
)
|
|
|
|
assert result['fileType'] == 'image/heic'
|
|
assert backend.uploaded[0][0] == content
|
|
|
|
|
|
def test_heif_extension_accepts_generic_heif_brand():
|
|
from app.services.market_support import save_market_attachment
|
|
|
|
db = _session()
|
|
backend = _Storage()
|
|
content = b'\x00\x00\x00\x10ftypmif1\x00\x00\x00\x00'
|
|
|
|
result = save_market_attachment(
|
|
db,
|
|
content=content,
|
|
filename='photo.heif',
|
|
content_type='image/heif',
|
|
module='sample',
|
|
ctx=_ctx(),
|
|
user_info=_user(),
|
|
storage_backend=backend,
|
|
)
|
|
|
|
assert result['fileType'] == 'image/heif'
|
|
|
|
|
|
def _attachment(**overrides):
|
|
values = dict(
|
|
id='att-a',
|
|
business_type='market:specialist:payment:staging',
|
|
business_id=None,
|
|
original_name='proof.pdf',
|
|
file_ext='pdf',
|
|
file_url='market/specialist/payment/a.pdf',
|
|
file_size=12,
|
|
file_type='application/pdf',
|
|
tenant_no='tenant-a',
|
|
create_by='user-a',
|
|
create_dept_id='',
|
|
del_flag=0,
|
|
)
|
|
values.update(overrides)
|
|
return BizAttachment(**values)
|
|
|
|
|
|
def test_create_attachment_refs_require_current_uploader_staging_exact_match():
|
|
from app.services.market_support import validate_attachment_refs
|
|
|
|
db = _session()
|
|
db.add_all(
|
|
[
|
|
_attachment(),
|
|
_attachment(id='att-other', file_url='market/specialist/payment/other.pdf', create_by='other'),
|
|
_attachment(id='att-cross', file_url='market/specialist/payment/cross.pdf', tenant_no='tenant-b'),
|
|
]
|
|
)
|
|
db.commit()
|
|
|
|
assert validate_attachment_refs(
|
|
db,
|
|
'att-a,market/specialist/payment/a.pdf',
|
|
module='payment',
|
|
ctx=_ctx(),
|
|
creating=True,
|
|
) == 'market/specialist/payment/a.pdf'
|
|
for invalid in ('att-other', 'att-cross', 'https://evil.test/a.pdf', 'a.pdf'):
|
|
with pytest.raises(ValueError, match='附件'):
|
|
validate_attachment_refs(db, invalid, module='payment', ctx=_ctx(), creating=True)
|
|
|
|
|
|
def test_staging_attachment_must_match_current_scope_role_exactly():
|
|
from app.services.market_support import validate_attachment_refs
|
|
|
|
db = _session()
|
|
db.add(
|
|
_attachment(
|
|
id='manager-staging',
|
|
business_type='market:manager:payment:staging',
|
|
file_url='market/manager/payment/proof.pdf',
|
|
)
|
|
)
|
|
db.commit()
|
|
|
|
with pytest.raises(ValueError, match='附件'):
|
|
validate_attachment_refs(
|
|
db,
|
|
'manager-staging',
|
|
module='payment',
|
|
ctx=_ctx('specialist'),
|
|
creating=True,
|
|
business_id='pay-a',
|
|
)
|
|
|
|
|
|
def test_attachment_refs_bind_once_to_exact_scope_module_and_business():
|
|
from app.services.market_support import validate_attachment_refs
|
|
|
|
db = _session()
|
|
row = _attachment()
|
|
db.add(row)
|
|
db.commit()
|
|
|
|
result = validate_attachment_refs(
|
|
db,
|
|
'att-a,market/specialist/payment/a.pdf',
|
|
module='payment',
|
|
ctx=_ctx(),
|
|
creating=True,
|
|
business_id='transfer-a',
|
|
)
|
|
|
|
assert result == 'market/specialist/payment/a.pdf'
|
|
assert (row.business_type, row.business_id) == (
|
|
'market:specialist:payment:bound',
|
|
'transfer-a',
|
|
)
|
|
assert validate_attachment_refs(
|
|
db,
|
|
'att-a',
|
|
module='payment',
|
|
ctx=_ctx(),
|
|
creating=False,
|
|
business_id='transfer-a',
|
|
) == 'market/specialist/payment/a.pdf'
|
|
with pytest.raises(ValueError, match='附件'):
|
|
validate_attachment_refs(
|
|
db,
|
|
'att-a',
|
|
module='payment',
|
|
ctx=_ctx(),
|
|
creating=False,
|
|
business_id='transfer-b',
|
|
)
|
|
with pytest.raises(ValueError, match='附件'):
|
|
validate_attachment_refs(
|
|
db,
|
|
'att-a',
|
|
module='payment',
|
|
ctx=_ctx('manager'),
|
|
creating=False,
|
|
business_id='transfer-a',
|
|
)
|
|
|
|
|
|
def test_attachment_ref_resolution_uses_one_batched_query_and_deduplicates_aliases():
|
|
from sqlalchemy import event
|
|
|
|
from app.services.market_support import validate_attachment_refs
|
|
|
|
db = _session()
|
|
db.add_all(
|
|
[
|
|
_attachment(id='att-1', file_url='market/specialist/payment/1.pdf'),
|
|
_attachment(id='att-2', file_url='market/specialist/payment/2.pdf'),
|
|
]
|
|
)
|
|
db.commit()
|
|
statements = []
|
|
|
|
def capture(_conn, _cursor, statement, _parameters, _context, _many):
|
|
if 'biz_attachment' in statement.lower() and statement.lstrip().lower().startswith('select'):
|
|
statements.append(statement)
|
|
|
|
event.listen(db.get_bind(), 'before_cursor_execute', capture)
|
|
try:
|
|
result = validate_attachment_refs(
|
|
db,
|
|
'att-1,market/specialist/payment/1.pdf,att-2',
|
|
module='payment',
|
|
ctx=_ctx(),
|
|
creating=True,
|
|
business_id='transfer-batch',
|
|
)
|
|
finally:
|
|
event.remove(db.get_bind(), 'before_cursor_execute', capture)
|
|
|
|
assert result == 'market/specialist/payment/1.pdf,market/specialist/payment/2.pdf'
|
|
assert len(statements) == 1
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
'refs',
|
|
[','.join(f'att-{index}' for index in range(21)), 'x' * 4001],
|
|
)
|
|
def test_attachment_refs_reject_excessive_count_or_total_length(refs):
|
|
from app.services.market_support import validate_attachment_refs
|
|
|
|
with pytest.raises(ValueError, match='附件'):
|
|
validate_attachment_refs(
|
|
_session(),
|
|
refs,
|
|
module='payment',
|
|
ctx=_ctx(),
|
|
creating=True,
|
|
business_id='transfer-limit',
|
|
)
|
|
|
|
|
|
def test_all_mobile_attachment_schemas_cap_total_reference_length():
|
|
from pydantic import ValidationError
|
|
|
|
from app.schemas.market_mobile import (
|
|
MarketPaymentTransferBody,
|
|
MarketPresignTransferBody,
|
|
MarketProtocolAttachmentsBody,
|
|
MarketSampleStorageCertBody,
|
|
)
|
|
|
|
too_long = 'x' * 4001
|
|
payloads = (
|
|
(MarketPresignTransferBody, {'transferAttachments': too_long}),
|
|
(MarketProtocolAttachmentsBody, {'protocolAttachments': too_long}),
|
|
(MarketPaymentTransferBody, {'transferAttachments': too_long}),
|
|
(MarketSampleStorageCertBody, {'attachments': too_long}),
|
|
)
|
|
for schema, payload in payloads:
|
|
with pytest.raises(ValidationError):
|
|
schema.model_validate(payload)
|
|
|
|
|
|
def test_update_attachment_refs_allow_exact_existing_refs_and_current_user_new_only():
|
|
from app.services.market_support import validate_attachment_refs
|
|
|
|
db = _session()
|
|
db.add_all(
|
|
[
|
|
_attachment(
|
|
id='existing',
|
|
file_url='market/specialist/payment/existing.pdf',
|
|
create_by='legacy',
|
|
business_type='market:specialist:payment:bound',
|
|
business_id='transfer-existing',
|
|
),
|
|
_attachment(id='new', file_url='market/specialist/payment/new.pdf'),
|
|
]
|
|
)
|
|
db.commit()
|
|
|
|
out = validate_attachment_refs(
|
|
db,
|
|
'existing,new',
|
|
module='payment',
|
|
ctx=_ctx(),
|
|
creating=False,
|
|
existing_refs='market/specialist/payment/existing.pdf',
|
|
business_id='transfer-existing',
|
|
)
|
|
assert out == 'market/specialist/payment/existing.pdf,market/specialist/payment/new.pdf'
|
|
|
|
|
|
@pytest.mark.parametrize('metadata_kind', ['missing', 'legacy-unbound', 'other-role-bound'])
|
|
def test_update_may_retain_exact_server_supplied_existing_url_without_current_metadata(
|
|
metadata_kind,
|
|
):
|
|
from app.services.market_support import validate_attachment_refs
|
|
|
|
db = _session()
|
|
existing_url = 'legacy/payment/existing.pdf'
|
|
if metadata_kind == 'legacy-unbound':
|
|
db.add(
|
|
_attachment(
|
|
id='legacy-unbound',
|
|
file_url=existing_url,
|
|
business_type='protocol',
|
|
create_by='legacy-user',
|
|
)
|
|
)
|
|
elif metadata_kind == 'other-role-bound':
|
|
db.add(
|
|
_attachment(
|
|
id='specialist-bound',
|
|
file_url=existing_url,
|
|
business_type='market:specialist:payment:bound',
|
|
business_id='transfer-existing',
|
|
create_by='specialist-user',
|
|
)
|
|
)
|
|
db.commit()
|
|
|
|
assert validate_attachment_refs(
|
|
db,
|
|
existing_url,
|
|
module='payment',
|
|
ctx=_ctx('manager'),
|
|
creating=False,
|
|
existing_refs=existing_url,
|
|
business_id='transfer-existing',
|
|
) == existing_url
|
|
|
|
|
|
def test_existing_ref_exception_allows_removal_but_not_arbitrary_url_smuggling():
|
|
from app.services.market_support import validate_attachment_refs
|
|
|
|
db = _session()
|
|
assert validate_attachment_refs(
|
|
db,
|
|
'',
|
|
module='payment',
|
|
ctx=_ctx(),
|
|
creating=False,
|
|
existing_refs='legacy/payment/existing.pdf',
|
|
business_id='transfer-existing',
|
|
) == ''
|
|
|
|
with pytest.raises(ValueError, match='附件'):
|
|
validate_attachment_refs(
|
|
db,
|
|
'https://evil.test/smuggled.pdf',
|
|
module='payment',
|
|
ctx=_ctx(),
|
|
creating=False,
|
|
existing_refs='legacy/payment/existing.pdf',
|
|
business_id='transfer-existing',
|
|
)
|
|
|
|
|
|
def test_allowed_actions_are_state_scope_and_permission_aware():
|
|
from app.services.market_support import allowed_actions
|
|
|
|
assert allowed_actions('presign', status='active') == ['edit', 'convert', 'delete']
|
|
assert allowed_actions('presign', status='void') == ['edit', 'delete']
|
|
assert allowed_actions('presign', status='converted') == ['edit']
|
|
assert allowed_actions('presign-transfer', status='dfj') == ['edit', 'delete', 'submit']
|
|
assert allowed_actions('presign-transfer', status='jfqrz') == ['withdraw']
|
|
assert allowed_actions('presign-transfer', status='yjf') == []
|
|
assert allowed_actions('protocol', status='dsx') == ['edit']
|
|
assert allowed_actions('protocol', status='yzz') == []
|
|
assert allowed_actions('payment-transfer', status='dfj') == ['edit-transfer', 'submit-finance']
|
|
assert allowed_actions('payment-transfer', status='jfqrz') == ['withdraw-transfer']
|
|
assert allowed_actions('payment-chase', status='', has_payment_permission=False) == []
|
|
assert allowed_actions('payment-chase', status='', has_payment_permission=True) == ['register-transfer']
|
|
assert allowed_actions('sample', status='', source='research_donation', has_sample_code=True) == []
|
|
assert allowed_actions('sample', status='', source='cell_storage', has_sample_code=True) == ['storage-cert', 'insurance']
|
|
assert allowed_actions('freight', status='', scope_kind='specialist') == []
|
|
assert allowed_actions('freight', status='', scope_kind='manager') == ['edit']
|
|
assert allowed_actions('presign-list', status='') == ['create']
|
|
assert allowed_actions('protocol-list', status='') == ['create']
|
|
assert allowed_actions('freight-list', status='', scope_kind='specialist') == ['create']
|
|
|
|
|
|
def test_presign_actions_include_delete_and_nested_transfer_actions():
|
|
from app.services.market_support import decorate_allowed_actions
|
|
|
|
active = decorate_allowed_actions(
|
|
'presign',
|
|
{
|
|
'orderStatus': 'active',
|
|
'transfers': [
|
|
{'id': 't1', 'payStatus': 'dfj'},
|
|
{'id': 't2', 'payStatus': 'jfqrz'},
|
|
{'id': 't3', 'payStatus': 'yjf'},
|
|
],
|
|
},
|
|
scope_kind='specialist',
|
|
detail=True,
|
|
)
|
|
|
|
assert active['allowedActions'] == ['edit', 'convert', 'delete']
|
|
assert [row['allowedActions'] for row in active['transfers']] == [
|
|
['edit', 'delete', 'submit'],
|
|
['withdraw'],
|
|
[],
|
|
]
|
|
|
|
for parent_status, parent_actions in (
|
|
('converted', ['edit']),
|
|
('void', ['edit', 'delete']),
|
|
):
|
|
locked = decorate_allowed_actions(
|
|
'presign',
|
|
{
|
|
'orderStatus': parent_status,
|
|
'transfers': [
|
|
{'id': 't1', 'payStatus': 'dfj'},
|
|
{'id': 't2', 'payStatus': 'jfqrz'},
|
|
],
|
|
},
|
|
scope_kind='specialist',
|
|
detail=True,
|
|
)
|
|
assert locked['allowedActions'] == parent_actions
|
|
assert [row['allowedActions'] for row in locked['transfers']] == [[], []]
|
|
|
|
unknown = decorate_allowed_actions(
|
|
'presign',
|
|
{
|
|
'orderStatus': 'legacy-locked',
|
|
'transfers': [{'id': 't1', 'payStatus': 'dfj'}],
|
|
},
|
|
scope_kind='specialist',
|
|
detail=True,
|
|
)
|
|
assert unknown['transfers'][0]['allowedActions'] == []
|
|
|
|
|
|
def test_form_options_do_not_leak_children_of_soft_deleted_regions():
|
|
from app.services.market_support import get_form_options
|
|
|
|
db = _session()
|
|
_seed_options(db)
|
|
db.add_all(
|
|
[
|
|
BizCollectionRegion(
|
|
id='deleted-region',
|
|
region_name='已删除地区',
|
|
tenant_no='tenant-a',
|
|
create_dept_id='',
|
|
del_flag=1,
|
|
),
|
|
BizCollectionHospital(
|
|
id='deleted-region-hospital',
|
|
region_id='deleted-region',
|
|
hospital_name='不应泄漏医院',
|
|
tenant_no='tenant-a',
|
|
create_dept_id='',
|
|
del_flag=0,
|
|
),
|
|
BizCollectionRegionStaff(
|
|
id='deleted-region-staff',
|
|
region_id='deleted-region',
|
|
user_id='user-a',
|
|
is_leader=0,
|
|
tenant_no='tenant-a',
|
|
create_dept_id='',
|
|
del_flag=0,
|
|
),
|
|
]
|
|
)
|
|
db.commit()
|
|
|
|
out = get_form_options(db, module='presign', ctx=_ctx(), user_info=_user())
|
|
|
|
assert '已删除地区' not in str(out)
|
|
assert '不应泄漏医院' not in str(out)
|
|
with pytest.raises(PermissionError, match='地区'):
|
|
get_form_options(
|
|
db,
|
|
module='presign',
|
|
ctx=_ctx(),
|
|
user_info=_user(),
|
|
region_id='deleted-region',
|
|
)
|
|
|
|
|
|
def test_decorate_allowed_actions_covers_nested_payment_transfers():
|
|
from app.services.market_support import decorate_allowed_actions
|
|
|
|
detail = {
|
|
'id': 'pay-a',
|
|
'schedules': [
|
|
{
|
|
'id': 'schedule-a',
|
|
'scheduleStatus': 'open',
|
|
'transfers': [
|
|
{'id': 'draft', 'payStatus': 'dfj'},
|
|
{'id': 'submitted', 'payStatus': 'jfqrz'},
|
|
{'id': 'done', 'payStatus': 'yjf'},
|
|
],
|
|
}
|
|
],
|
|
}
|
|
out = decorate_allowed_actions('payment', detail, scope_kind='specialist')
|
|
|
|
assert out['allowedActions'] == ['register-transfer']
|
|
assert out['schedules'][0]['allowedActions'] == ['register-transfer']
|
|
assert [row['allowedActions'] for row in out['schedules'][0]['transfers']] == [
|
|
['edit-transfer', 'submit-finance'],
|
|
['withdraw-transfer'],
|
|
[],
|
|
]
|
|
|
|
|
|
@pytest.mark.parametrize('protocol_status', ['yzf', 'sdfq', 'zky'])
|
|
def test_payment_terminal_protocol_removes_all_nested_write_actions(protocol_status):
|
|
from app.services.market_support import decorate_allowed_actions
|
|
|
|
detail = {
|
|
'id': 'pay-terminal',
|
|
'protocolStatus': protocol_status,
|
|
'schedules': [
|
|
{
|
|
'id': 'schedule-open',
|
|
'scheduleStatus': 'open',
|
|
'transfers': [
|
|
{'id': 'draft', 'payStatus': 'dfj'},
|
|
{'id': 'submitted', 'payStatus': 'jfqrz'},
|
|
],
|
|
}
|
|
],
|
|
}
|
|
|
|
out = decorate_allowed_actions('payment', detail, scope_kind='specialist')
|
|
|
|
assert out['allowedActions'] == []
|
|
assert out['schedules'][0]['allowedActions'] == []
|
|
assert [row['allowedActions'] for row in out['schedules'][0]['transfers']] == [[], []]
|
|
|
|
|
|
@pytest.mark.parametrize('schedule_status', ['cleared', 'void'])
|
|
def test_non_open_payment_schedule_removes_schedule_and_transfer_actions(schedule_status):
|
|
from app.services.market_support import decorate_allowed_actions
|
|
|
|
detail = {
|
|
'id': 'pay-closed',
|
|
'protocolStatus': 'sxz',
|
|
'schedules': [
|
|
{
|
|
'id': 'schedule-closed',
|
|
'scheduleStatus': schedule_status,
|
|
'transfers': [{'id': 'draft', 'payStatus': 'dfj'}],
|
|
}
|
|
],
|
|
}
|
|
|
|
out = decorate_allowed_actions('payment', detail, scope_kind='specialist')
|
|
|
|
assert out['allowedActions'] == []
|
|
assert out['schedules'][0]['allowedActions'] == []
|
|
assert out['schedules'][0]['transfers'][0]['allowedActions'] == []
|
|
|
|
|
|
def test_freight_detail_exposes_only_scope_safe_link_actions():
|
|
from app.services.market_support import decorate_allowed_actions
|
|
|
|
specialist = decorate_allowed_actions(
|
|
'freight',
|
|
{'id': 'f1', 'allowedActions': ['edit', 'link', 'unlink']},
|
|
scope_kind='specialist',
|
|
detail=True,
|
|
)
|
|
manager = decorate_allowed_actions(
|
|
'freight',
|
|
{'id': 'f1', 'allowedActions': ['link', 'unlink']},
|
|
scope_kind='manager',
|
|
detail=True,
|
|
)
|
|
assert specialist['allowedActions'] == ['edit', 'link', 'unlink']
|
|
assert manager['allowedActions'] == ['link', 'unlink']
|
|
|
|
|
|
def test_protocol_detail_action_reads_nested_basic_status():
|
|
from app.services.market_support import decorate_allowed_actions
|
|
|
|
active = decorate_allowed_actions(
|
|
'protocol', {'basic': {'protocolStatus': 'dsx'}}, scope_kind='specialist'
|
|
)
|
|
terminal = decorate_allowed_actions(
|
|
'protocol', {'basic': {'protocolStatus': 'yzf'}}, scope_kind='specialist'
|
|
)
|
|
assert active['allowedActions'] == ['edit']
|
|
assert terminal['allowedActions'] == []
|
|
|
|
|
|
def test_market_support_routes_are_fixed_and_registered_before_dynamic_details():
|
|
from app.api.market_mobile import router
|
|
|
|
paths = [route.path for route in router.routes]
|
|
for scope in ('specialist', 'manager'):
|
|
for path in (
|
|
f'/mini/market/{scope}/presign-orders/form-options',
|
|
f'/mini/market/{scope}/protocols/form-options',
|
|
f'/mini/market/{scope}/payments/form-options',
|
|
f'/mini/market/{scope}/samples/form-options',
|
|
f'/mini/market/{scope}/freights/form-options',
|
|
):
|
|
assert path in paths
|
|
for module in ('presign', 'protocol', 'payment', 'sample'):
|
|
upload_path = f'/mini/market/{scope}/{module}/attachments'
|
|
assert upload_path in paths
|
|
upload_route = next(route for route in router.routes if route.path == upload_path)
|
|
assert [param.name for param in upload_route.dependant.query_params] == []
|
|
assert paths.index(f'/mini/market/{scope}/presign-orders/form-options') < paths.index(
|
|
f'/mini/market/{scope}/presign-orders/{{order_id}}'
|
|
)
|
|
assert paths.index(f'/mini/market/{scope}/protocols/form-options') < paths.index(
|
|
f'/mini/market/{scope}/protocols/{{protocol_id}}'
|
|
)
|
|
assert paths.index(f'/mini/market/{scope}/samples/form-options') < paths.index(
|
|
f'/mini/market/{scope}/samples/{{sample_id}}'
|
|
)
|
|
|
|
|
|
def test_presign_write_routes_use_fixed_paths_and_methods():
|
|
from app.api.market_mobile import router
|
|
|
|
for scope in ('specialist', 'manager'):
|
|
expected = {
|
|
f'/mini/market/{scope}/presign-orders/{{order_id}}': {'GET', 'PUT', 'DELETE'},
|
|
f'/mini/market/{scope}/presign-orders/{{order_id}}/transfers': {'POST'},
|
|
f'/mini/market/{scope}/presign-transfers/{{transfer_id}}': {'PUT', 'DELETE'},
|
|
f'/mini/market/{scope}/presign-transfers/{{transfer_id}}/submit': {'POST'},
|
|
f'/mini/market/{scope}/presign-transfers/{{transfer_id}}/withdraw': {'POST'},
|
|
}
|
|
for path, methods in expected.items():
|
|
registered = {
|
|
method
|
|
for route in router.routes
|
|
if route.path == path
|
|
for method in route.methods
|
|
}
|
|
assert registered == methods
|
|
|
|
|
|
def test_presign_write_routes_map_permission_and_validation_errors(monkeypatch):
|
|
from app.api import market_mobile
|
|
from app.schemas.market_mobile import MarketPresignTransferBody
|
|
|
|
delete_route = next(
|
|
route
|
|
for route in market_mobile.router.routes
|
|
if route.path == '/mini/market/specialist/presign-orders/{order_id}'
|
|
and route.methods == {'DELETE'}
|
|
)
|
|
create_transfer_route = next(
|
|
route
|
|
for route in market_mobile.router.routes
|
|
if route.path
|
|
== '/mini/market/specialist/presign-orders/{order_id}/transfers'
|
|
)
|
|
monkeypatch.setattr(
|
|
market_mobile.market_presign,
|
|
'delete_presign_order',
|
|
lambda *args, **kwargs: (_ for _ in ()).throw(PermissionError('forbidden')),
|
|
)
|
|
permission_result = asyncio.run(
|
|
delete_route.endpoint(order_id='o1', user=_user(), db=SimpleNamespace())
|
|
)
|
|
assert permission_result['code'] == 403
|
|
|
|
monkeypatch.setattr(
|
|
market_mobile.market_presign,
|
|
'create_presign_transfer',
|
|
lambda *args, **kwargs: (_ for _ in ()).throw(ValueError('invalid')),
|
|
)
|
|
validation_result = asyncio.run(
|
|
create_transfer_route.endpoint(
|
|
order_id='o1',
|
|
body=MarketPresignTransferBody(transferAmount=100),
|
|
user=_user(),
|
|
db=SimpleNamespace(),
|
|
)
|
|
)
|
|
assert validation_result['code'] == 400
|
|
|
|
|
|
def test_attachment_routes_read_only_their_module_upload_limit(monkeypatch):
|
|
from app.api import market_mobile
|
|
|
|
class TrackingUpload:
|
|
filename = 'proof.pdf'
|
|
content_type = 'application/pdf'
|
|
|
|
def __init__(self):
|
|
self.read_size = None
|
|
|
|
async def read(self, size):
|
|
self.read_size = size
|
|
return b'%PDF-safe'
|
|
|
|
async def close(self):
|
|
return None
|
|
|
|
monkeypatch.setattr(
|
|
market_mobile.market_support,
|
|
'save_market_attachment',
|
|
lambda *args, **kwargs: {'id': 'attachment'},
|
|
)
|
|
routes = {
|
|
route.path: route
|
|
for route in market_mobile.router.routes
|
|
if route.path in {
|
|
'/mini/market/specialist/presign/attachments',
|
|
'/mini/market/specialist/payment/attachments',
|
|
}
|
|
}
|
|
presign_file = TrackingUpload()
|
|
payment_file = TrackingUpload()
|
|
|
|
asyncio.run(
|
|
routes['/mini/market/specialist/presign/attachments'].endpoint(
|
|
file=presign_file, user=_user(), db=SimpleNamespace()
|
|
)
|
|
)
|
|
asyncio.run(
|
|
routes['/mini/market/specialist/payment/attachments'].endpoint(
|
|
file=payment_file, user=_user(), db=SimpleNamespace()
|
|
)
|
|
)
|
|
|
|
assert presign_file.read_size == 20 * 1024 * 1024 + 1
|
|
assert payment_file.read_size == 10 * 1024 * 1024 + 1
|