938 lines
32 KiB
Python
938 lines
32 KiB
Python
from datetime import datetime, timedelta
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
from pydantic import ValidationError
|
|
from sqlalchemy import create_engine, event
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
from app.models.base import Base
|
|
from app.models.models import (
|
|
BizAttachment,
|
|
BizCellSampleType,
|
|
BizCollectionHospital,
|
|
BizCollectionProtocol,
|
|
BizCollectionRegion,
|
|
BizCollectionRegionStaff,
|
|
BizCustomer,
|
|
BizItemUnit,
|
|
BizProtocolCollectionInfo,
|
|
BizProtocolPartyCustomer,
|
|
BizProtocolPayment,
|
|
BizProtocolSample,
|
|
BizSampleCode,
|
|
SysPost,
|
|
SysPostRole,
|
|
SysRole,
|
|
SysRoleDataScopeDetail,
|
|
SysUser,
|
|
SysUserPost,
|
|
SysUserRole,
|
|
)
|
|
from app.models.research_donation import (
|
|
BizProtocolDonationConvert,
|
|
BizResearchDonation,
|
|
BizResearchDonationCollection,
|
|
BizResearchDonationParty,
|
|
BizResearchDonationSample,
|
|
BizResearchDonationSampleTransport,
|
|
)
|
|
|
|
|
|
def _session():
|
|
engine = create_engine('sqlite:///:memory:')
|
|
Base.metadata.create_all(
|
|
engine,
|
|
tables=[
|
|
BizCollectionProtocol.__table__,
|
|
BizProtocolPartyCustomer.__table__,
|
|
BizProtocolCollectionInfo.__table__,
|
|
BizProtocolSample.__table__,
|
|
BizProtocolPayment.__table__,
|
|
BizSampleCode.__table__,
|
|
BizResearchDonation.__table__,
|
|
BizResearchDonationParty.__table__,
|
|
BizResearchDonationCollection.__table__,
|
|
BizResearchDonationSample.__table__,
|
|
BizResearchDonationSampleTransport.__table__,
|
|
BizProtocolDonationConvert.__table__,
|
|
BizCollectionRegionStaff.__table__,
|
|
BizCollectionRegion.__table__,
|
|
BizCollectionHospital.__table__,
|
|
BizCellSampleType.__table__,
|
|
BizItemUnit.__table__,
|
|
BizCustomer.__table__,
|
|
BizAttachment.__table__,
|
|
SysUser.__table__,
|
|
SysRole.__table__,
|
|
SysRoleDataScopeDetail.__table__,
|
|
SysUserRole.__table__,
|
|
SysPost.__table__,
|
|
SysPostRole.__table__,
|
|
SysUserPost.__table__,
|
|
],
|
|
)
|
|
return sessionmaker(bind=engine)()
|
|
|
|
|
|
def _ctx(scope_kind, user_id='u1'):
|
|
from app.services.market_field_scope import MarketFieldContext
|
|
|
|
return MarketFieldContext('tenant-a', user_id, scope_kind)
|
|
|
|
|
|
def _user(user_id='u1'):
|
|
return {
|
|
'user_id': user_id,
|
|
'tenant_no': 'tenant-a',
|
|
'role': 'super_admin',
|
|
'region_id': 'dept-from-auth',
|
|
}
|
|
|
|
|
|
def _cell(db, row_id, *, sales='u1', manager='manager-a', region='region-a', created=None):
|
|
row = BizCollectionProtocol(
|
|
id=row_id,
|
|
protocol_no=f'CELL-{row_id}',
|
|
protocol_status='dsx',
|
|
tenant_no='tenant-a',
|
|
create_dept_id='dept-a',
|
|
create_time=created or datetime.now(),
|
|
del_flag=0,
|
|
)
|
|
collection = BizProtocolCollectionInfo(
|
|
id=f'ci-{row_id}',
|
|
collection_protocol_id=row_id,
|
|
region_id=region,
|
|
sales_user_id=sales,
|
|
region_manager_user_id=manager,
|
|
tenant_no='tenant-a',
|
|
create_dept_id='dept-a',
|
|
del_flag=0,
|
|
)
|
|
db.add_all([row, collection])
|
|
return row
|
|
|
|
|
|
def _donation(db, row_id, *, sales='u1', manager='manager-a', region='region-a', created=None):
|
|
row = BizResearchDonation(
|
|
id=row_id,
|
|
donation_no=f'DON-{row_id}',
|
|
donation_status='dsx',
|
|
tenant_no='tenant-a',
|
|
create_dept_id='dept-a',
|
|
create_time=created or datetime.now(),
|
|
del_flag=0,
|
|
)
|
|
collection = BizResearchDonationCollection(
|
|
id=f'dci-{row_id}',
|
|
research_donation_id=row_id,
|
|
region_id=region,
|
|
sales_user_id=sales,
|
|
region_manager_user_id=manager,
|
|
tenant_no='tenant-a',
|
|
create_dept_id='dept-a',
|
|
del_flag=0,
|
|
)
|
|
db.add_all([row, collection])
|
|
return row
|
|
|
|
|
|
def _masters(db):
|
|
db.add_all(
|
|
[
|
|
BizCollectionRegion(
|
|
id='region-a', region_name='A区', tenant_no='tenant-a', create_dept_id='', del_flag=0
|
|
),
|
|
BizCollectionRegion(
|
|
id='region-b', region_name='B区', tenant_no='tenant-a', create_dept_id='', del_flag=0
|
|
),
|
|
BizCollectionHospital(
|
|
id='hospital-a', region_id='region-a', hospital_name='A院', tenant_no='tenant-a', 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
|
|
),
|
|
BizSampleCode(
|
|
id='code-a', sample_code='S-001', tenant_no='tenant-a', create_dept_id='', del_flag=0
|
|
),
|
|
BizSampleCode(
|
|
id='code-cross', sample_code='S-CROSS', tenant_no='tenant-b', create_dept_id='', del_flag=0
|
|
),
|
|
BizSampleCode(
|
|
id='code-deleted', sample_code='S-DELETED', tenant_no='tenant-a', create_dept_id='', del_flag=1
|
|
),
|
|
BizItemUnit(
|
|
id='unit-a', unit_code='ml', unit_name='毫升', tenant_no='tenant-a', create_dept_id='', del_flag=0
|
|
),
|
|
SysUser(
|
|
id='u1', user_no='u1', nick_name='当前用户', status='1', tenant_no='tenant-a', create_dept_id='', del_flag=0
|
|
),
|
|
SysUser(
|
|
id='manager-a', user_no='manager-a', nick_name='经理', status='1', tenant_no='tenant-a', create_dept_id='', del_flag=0
|
|
),
|
|
SysUser(
|
|
id='u2', user_no='u2', nick_name='其他专员', status='1', tenant_no='tenant-a', create_dept_id='', del_flag=0
|
|
),
|
|
BizAttachment(
|
|
id='att-protocol-a',
|
|
business_type='market:specialist:protocol:staging',
|
|
original_name='a.pdf',
|
|
file_ext='pdf',
|
|
file_url='protocol/a.pdf',
|
|
file_size=10,
|
|
file_type='application/pdf',
|
|
tenant_no='tenant-a',
|
|
create_by='u1',
|
|
create_dept_id='',
|
|
del_flag=0,
|
|
),
|
|
]
|
|
)
|
|
db.commit()
|
|
|
|
|
|
def _body(source='cell_storage', **overrides):
|
|
from app.schemas.market_mobile import MarketProtocolBody
|
|
|
|
payload = {
|
|
'source': source,
|
|
'protocolNo': 'P-001',
|
|
'registerDate': '20261010',
|
|
'attachments': {'protocolAttachments': 'protocol/a.pdf'},
|
|
'customer': {'motherName': '张三', 'motherPhone': '13800000000'},
|
|
'collection': {
|
|
'regionId': 'region-a',
|
|
'hospitalId': 'hospital-a',
|
|
'salesUserId': 'someone-else',
|
|
'regionManagerUserId': 'manager-a',
|
|
},
|
|
'samples': [
|
|
{
|
|
'sampleTypeId': 'type-a',
|
|
'sampleCodeId': 'code-a',
|
|
'sampleCode': 'S-001',
|
|
'unitId': 'unit-a',
|
|
'collectAmount': 10,
|
|
}
|
|
],
|
|
'remark': '移动端登记',
|
|
}
|
|
payload.update(overrides)
|
|
return MarketProtocolBody.model_validate(payload)
|
|
|
|
|
|
def test_protocol_create_rejects_unknown_attachment_reference():
|
|
from app.services.market_protocol import create_protocol
|
|
|
|
db = _session()
|
|
_masters(db)
|
|
body = _body(
|
|
attachments={'protocolAttachments': 'https://evil.test/a.pdf'}
|
|
)
|
|
|
|
with pytest.raises(ValueError, match='附件'):
|
|
create_protocol(
|
|
db,
|
|
body=body,
|
|
ctx=_ctx('specialist'),
|
|
user_info=_user(),
|
|
)
|
|
|
|
|
|
def test_terminal_protocol_rejects_hidden_update_action():
|
|
from app.services.market_protocol import update_protocol
|
|
|
|
db = _session()
|
|
_masters(db)
|
|
row = _cell(db, 'terminal')
|
|
row.protocol_status = 'yzf'
|
|
db.commit()
|
|
|
|
with pytest.raises(ValueError, match='不可修改'):
|
|
update_protocol(
|
|
db,
|
|
source='cell_storage',
|
|
protocol_id=row.id,
|
|
body=_body(protocolNo='CELL-terminal'),
|
|
ctx=_ctx('specialist'),
|
|
user_info=_user(),
|
|
)
|
|
|
|
|
|
def test_protocol_list_combines_cell_and_donation_without_scope_leak():
|
|
from app.services.market_protocol import list_protocols
|
|
|
|
db = _session()
|
|
now = datetime.now()
|
|
_cell(db, 'cell-own', created=now - timedelta(minutes=2))
|
|
_cell(db, 'cell-other', sales='u2', created=now)
|
|
_donation(db, 'don-own', created=now - timedelta(minutes=1))
|
|
_donation(db, 'don-other', sales='u2', created=now + timedelta(minutes=1))
|
|
db.commit()
|
|
|
|
items, total = list_protocols(db, ctx=_ctx('specialist'), user_info=_user())
|
|
|
|
assert total == 2
|
|
assert [(item['source'], item['id']) for item in items] == [
|
|
('research_donation', 'don-own'),
|
|
('cell_storage', 'cell-own'),
|
|
]
|
|
|
|
|
|
def test_protocol_list_large_page_uses_bounded_source_queries_and_global_order():
|
|
from app.services.market_protocol import list_protocols
|
|
|
|
db = _session()
|
|
base = datetime(2026, 10, 10, 12, 0, 0)
|
|
expected = []
|
|
for index in range(24):
|
|
created = base + timedelta(seconds=index)
|
|
cell_id = f'cell-{index:02d}'
|
|
donation_id = f'don-{index:02d}'
|
|
_cell(db, cell_id, created=created)
|
|
_donation(db, donation_id, created=created + timedelta(microseconds=1))
|
|
expected.extend([(created, cell_id), (created + timedelta(microseconds=1), donation_id)])
|
|
db.commit()
|
|
statements = []
|
|
|
|
def capture(_conn, _cursor, statement, _parameters, _context, _many):
|
|
if 'FROM biz_collection_protocol ' in statement or 'FROM biz_research_donation ' in statement:
|
|
statements.append(statement)
|
|
|
|
event.listen(db.get_bind(), 'before_cursor_execute', capture)
|
|
try:
|
|
items, total = list_protocols(
|
|
db,
|
|
ctx=_ctx('specialist'),
|
|
user_info=_user(),
|
|
page=4,
|
|
page_size=7,
|
|
)
|
|
finally:
|
|
event.remove(db.get_bind(), 'before_cursor_execute', capture)
|
|
|
|
ordered_ids = [item[1] for item in sorted(expected, key=lambda item: (item[0], item[1]), reverse=True)]
|
|
assert total == 48
|
|
assert [item['id'] for item in items] == ordered_ids[21:28]
|
|
page_selects = [sql for sql in statements if 'count(' not in sql.lower()]
|
|
assert len(page_selects) == 2
|
|
assert all(' LIMIT ' in sql.upper() for sql in page_selects)
|
|
|
|
|
|
def test_protocol_list_stable_tie_breaker_prevents_page_overlap():
|
|
from app.services.market_protocol import list_protocols
|
|
|
|
db = _session()
|
|
same_time = datetime(2026, 10, 10, 12, 0, 0)
|
|
for row_id in ('a1', 'a3', 'a5', 'a7'):
|
|
_cell(db, row_id, created=same_time)
|
|
for row_id in ('a2', 'a4', 'a6', 'a8'):
|
|
_donation(db, row_id, created=same_time)
|
|
db.commit()
|
|
|
|
page_one, total = list_protocols(
|
|
db, ctx=_ctx('specialist'), user_info=_user(), page=1, page_size=3
|
|
)
|
|
page_two, _ = list_protocols(
|
|
db, ctx=_ctx('specialist'), user_info=_user(), page=2, page_size=3
|
|
)
|
|
|
|
assert total == 8
|
|
assert [item['id'] for item in page_one] == ['a8', 'a7', 'a6']
|
|
assert [item['id'] for item in page_two] == ['a5', 'a4', 'a3']
|
|
assert {item['id'] for item in page_one}.isdisjoint(
|
|
{item['id'] for item in page_two}
|
|
)
|
|
|
|
|
|
def test_protocol_detail_rechecks_source_specific_idor():
|
|
from app.services.market_protocol import get_protocol
|
|
|
|
db = _session()
|
|
_cell(db, 'cell-other', sales='u2')
|
|
_donation(db, 'don-other', sales='u2')
|
|
db.commit()
|
|
|
|
with pytest.raises(HTTPException) as cell_error:
|
|
get_protocol(
|
|
db,
|
|
source='cell_storage',
|
|
protocol_id='cell-other',
|
|
ctx=_ctx('specialist'),
|
|
user_info=_user(),
|
|
)
|
|
with pytest.raises(HTTPException) as donation_error:
|
|
get_protocol(
|
|
db,
|
|
source='research_donation',
|
|
protocol_id='don-other',
|
|
ctx=_ctx('specialist'),
|
|
user_info=_user(),
|
|
)
|
|
assert cell_error.value.status_code == 404
|
|
assert donation_error.value.status_code == 404
|
|
|
|
|
|
def test_protocol_detail_can_resolve_source_from_scoped_record():
|
|
from app.services.market_protocol import get_protocol
|
|
|
|
db = _session()
|
|
_donation(db, 'don-own')
|
|
db.commit()
|
|
|
|
result = get_protocol(
|
|
db,
|
|
source=None,
|
|
protocol_id='don-own',
|
|
ctx=_ctx('specialist'),
|
|
user_info=_user(),
|
|
)
|
|
|
|
assert result['source'] == 'research_donation'
|
|
assert result['basic']['id'] == 'don-own'
|
|
|
|
|
|
def test_specialist_cell_create_forces_current_collection_and_collector_users():
|
|
from app.services.market_protocol import create_protocol
|
|
|
|
db = _session()
|
|
_masters(db)
|
|
|
|
result = create_protocol(
|
|
db, body=_body(), ctx=_ctx('specialist'), user_info=_user()
|
|
)
|
|
|
|
main = db.get(BizCollectionProtocol, result['basic']['id'])
|
|
collection = db.query(BizProtocolCollectionInfo).filter_by(collection_protocol_id=main.id, del_flag=0).one()
|
|
party = db.query(BizProtocolPartyCustomer).filter_by(collection_protocol_id=main.id, del_flag=0).one()
|
|
sample = db.query(BizProtocolSample).filter_by(collection_protocol_id=main.id, del_flag=0).one()
|
|
assert main.collector_user_id == 'u1'
|
|
assert main.create_dept_id == 'dept-from-auth'
|
|
assert collection.sales_user_id == 'u1'
|
|
assert collection.create_dept_id == 'dept-from-auth'
|
|
assert party.collector_user_id == 'u1'
|
|
assert party.customer_id is None
|
|
assert db.query(BizCustomer).count() == 0
|
|
assert sample.sample_code == 'S-001'
|
|
|
|
|
|
def test_specialist_donation_create_forces_current_users_and_has_no_payment_section():
|
|
from app.services.market_protocol import create_protocol
|
|
|
|
db = _session()
|
|
_masters(db)
|
|
|
|
result = create_protocol(
|
|
db,
|
|
body=_body('research_donation', donationNo='D-001', protocolNo=None),
|
|
ctx=_ctx('specialist'),
|
|
user_info=_user(),
|
|
)
|
|
|
|
donation = db.get(BizResearchDonation, result['basic']['id'])
|
|
collection = db.query(BizResearchDonationCollection).filter_by(research_donation_id=donation.id, del_flag=0).one()
|
|
party = db.query(BizResearchDonationParty).filter_by(research_donation_id=donation.id, del_flag=0).one()
|
|
sample = db.query(BizResearchDonationSample).filter_by(research_donation_id=donation.id, del_flag=0).one()
|
|
assert donation.collector_user_id == 'u1'
|
|
assert donation.create_dept_id == 'dept-from-auth'
|
|
assert collection.sales_user_id == 'u1'
|
|
assert collection.create_dept_id == 'dept-from-auth'
|
|
assert party.collector_user_id == 'u1'
|
|
assert party.create_dept_id == 'dept-from-auth'
|
|
assert sample.create_dept_id == 'dept-from-auth'
|
|
assert party.customer_id is None
|
|
assert db.query(BizCustomer).count() == 0
|
|
assert 'payment' not in result
|
|
|
|
|
|
def test_manager_update_cannot_move_document_out_of_scope_and_rolls_back():
|
|
from app.services.market_protocol import update_protocol
|
|
|
|
db = _session()
|
|
_masters(db)
|
|
_cell(db, 'cell-manager', manager='u1')
|
|
db.commit()
|
|
|
|
with pytest.raises(ValueError, match='范围'):
|
|
update_protocol(
|
|
db,
|
|
source='cell_storage',
|
|
protocol_id='cell-manager',
|
|
body=_body(collection={'regionId': 'region-b', 'regionManagerUserId': 'manager-a'}),
|
|
ctx=_ctx('manager'),
|
|
user_info=_user(),
|
|
)
|
|
|
|
db.expire_all()
|
|
collection = db.query(BizProtocolCollectionInfo).filter_by(collection_protocol_id='cell-manager', del_flag=0).one()
|
|
assert collection.region_id == 'region-a'
|
|
assert collection.region_manager_user_id == 'u1'
|
|
|
|
|
|
def test_partial_collection_update_validates_against_resulting_region():
|
|
from app.schemas.market_mobile import MarketProtocolBody
|
|
from app.services.market_protocol import update_protocol
|
|
|
|
db = _session()
|
|
_masters(db)
|
|
_cell(db, 'cell-partial', sales='u1')
|
|
db.commit()
|
|
|
|
update_protocol(
|
|
db,
|
|
source='cell_storage',
|
|
protocol_id='cell-partial',
|
|
body=MarketProtocolBody.model_validate(
|
|
{
|
|
'source': 'cell_storage',
|
|
'collection': {'hospitalId': 'hospital-a'},
|
|
}
|
|
),
|
|
ctx=_ctx('specialist'),
|
|
user_info=_user(),
|
|
)
|
|
|
|
collection = db.query(BizProtocolCollectionInfo).filter_by(
|
|
collection_protocol_id='cell-partial', del_flag=0
|
|
).one()
|
|
assert collection.region_id == 'region-a'
|
|
assert collection.hospital_id == 'hospital-a'
|
|
|
|
|
|
def test_cell_partial_customer_update_preserves_ids_and_unspecified_fields():
|
|
from app.schemas.market_mobile import MarketProtocolBody
|
|
from app.services.market_protocol import update_protocol
|
|
|
|
db = _session()
|
|
_masters(db)
|
|
_cell(db, 'cell-party')
|
|
db.add_all(
|
|
[
|
|
BizCustomer(id='mother-c', customer_name='母客户', sex='1', tenant_no='tenant-a', create_dept_id='', del_flag=0),
|
|
BizCustomer(id='father-c', customer_name='父客户', sex='0', tenant_no='tenant-a', create_dept_id='', del_flag=0),
|
|
BizCustomer(id='newborn-c', customer_name='子客户', sex='2', tenant_no='tenant-a', create_dept_id='', del_flag=0),
|
|
BizProtocolPartyCustomer(
|
|
id='party-cell', collection_protocol_id='cell-party',
|
|
customer_id='mother-c', father_customer_id='father-c', newborn_customer_id='newborn-c',
|
|
mother_name='母亲', mother_phone='旧电话', mother_id_card='ID-M',
|
|
father_name='父亲', father_phone='父电话', newborn_name='宝宝',
|
|
tenant_no='tenant-a', create_dept_id='', del_flag=0,
|
|
),
|
|
]
|
|
)
|
|
db.commit()
|
|
|
|
update_protocol(
|
|
db,
|
|
source='cell_storage',
|
|
protocol_id='cell-party',
|
|
body=MarketProtocolBody.model_validate(
|
|
{
|
|
'source': 'cell_storage',
|
|
'customer': {'motherPhone': '新电话', 'remark': '只改电话'},
|
|
}
|
|
),
|
|
ctx=_ctx('specialist'),
|
|
user_info=_user(),
|
|
)
|
|
|
|
party = db.get(BizProtocolPartyCustomer, 'party-cell')
|
|
assert (party.customer_id, party.father_customer_id, party.newborn_customer_id) == (
|
|
'mother-c', 'father-c', 'newborn-c'
|
|
)
|
|
assert party.mother_name == '母亲'
|
|
assert party.mother_id_card == 'ID-M'
|
|
assert party.father_name == '父亲'
|
|
assert party.father_phone == '父电话'
|
|
assert party.newborn_name == '宝宝'
|
|
assert party.mother_phone == '新电话'
|
|
assert party.remark == '只改电话'
|
|
|
|
|
|
@pytest.mark.parametrize('source', ['cell_storage', 'research_donation'])
|
|
def test_protocol_update_rejects_same_tenant_unlinked_customer_id(source):
|
|
from app.schemas.market_mobile import MarketProtocolBody
|
|
from app.services.market_protocol import update_protocol
|
|
|
|
db = _session()
|
|
_masters(db)
|
|
db.add_all(
|
|
[
|
|
BizCustomer(id='linked-c', customer_name='已关联', sex='1', tenant_no='tenant-a', create_dept_id='', del_flag=0),
|
|
BizCustomer(id='unrelated-c', customer_name='无关客户', sex='1', tenant_no='tenant-a', create_dept_id='', del_flag=0),
|
|
]
|
|
)
|
|
if source == 'cell_storage':
|
|
_cell(db, 'customer-idor')
|
|
db.add(BizProtocolPartyCustomer(id='linked-party', collection_protocol_id='customer-idor', customer_id='linked-c', mother_name='原姓名', tenant_no='tenant-a', create_dept_id='', del_flag=0))
|
|
else:
|
|
_donation(db, 'customer-idor')
|
|
db.add(BizResearchDonationParty(id='linked-party', research_donation_id='customer-idor', customer_id='linked-c', mother_name='原姓名', tenant_no='tenant-a', create_dept_id='', del_flag=0))
|
|
db.commit()
|
|
|
|
with pytest.raises(ValueError, match='客户'):
|
|
update_protocol(
|
|
db,
|
|
source=source,
|
|
protocol_id='customer-idor',
|
|
body=MarketProtocolBody.model_validate(
|
|
{
|
|
'source': source,
|
|
'customer': {'customerId': 'unrelated-c', 'motherName': '被篡改'},
|
|
}
|
|
),
|
|
ctx=_ctx('specialist'),
|
|
user_info=_user(),
|
|
)
|
|
|
|
db.expire_all()
|
|
assert db.get(BizCustomer, 'unrelated-c').customer_name == '无关客户'
|
|
assert db.get(BizCustomer, 'linked-c').customer_name == '已关联'
|
|
|
|
|
|
@pytest.mark.parametrize('source', ['cell_storage', 'research_donation'])
|
|
def test_protocol_update_does_not_mutate_unrelated_customer_by_id_card(source):
|
|
from app.schemas.market_mobile import MarketProtocolBody
|
|
from app.services.market_protocol import update_protocol
|
|
|
|
db = _session()
|
|
_masters(db)
|
|
db.add(BizCustomer(id='unrelated-card', customer_name='身份证客户', sex='1', id_card_no='CARD-1', tenant_no='tenant-a', create_dept_id='', del_flag=0))
|
|
if source == 'cell_storage':
|
|
_cell(db, 'card-idor')
|
|
db.add(BizProtocolPartyCustomer(id='card-party', collection_protocol_id='card-idor', mother_name='原姓名', tenant_no='tenant-a', create_dept_id='', del_flag=0))
|
|
else:
|
|
_donation(db, 'card-idor')
|
|
db.add(BizResearchDonationParty(id='card-party', research_donation_id='card-idor', mother_name='原姓名', tenant_no='tenant-a', create_dept_id='', del_flag=0))
|
|
db.commit()
|
|
|
|
update_protocol(
|
|
db,
|
|
source=source,
|
|
protocol_id='card-idor',
|
|
body=MarketProtocolBody.model_validate(
|
|
{
|
|
'source': source,
|
|
'customer': {'motherName': '协议姓名', 'motherIdCard': 'CARD-1'},
|
|
}
|
|
),
|
|
ctx=_ctx('specialist'),
|
|
user_info=_user(),
|
|
)
|
|
|
|
db.expire_all()
|
|
assert db.get(BizCustomer, 'unrelated-card').customer_name == '身份证客户'
|
|
|
|
|
|
@pytest.mark.parametrize('source', ['cell_storage', 'research_donation'])
|
|
def test_protocol_create_forbids_existing_customer_ids(source):
|
|
from app.services.market_protocol import create_protocol
|
|
|
|
db = _session()
|
|
_masters(db)
|
|
db.add(BizCustomer(id='existing-c', customer_name='已有客户', sex='1', tenant_no='tenant-a', create_dept_id='', del_flag=0))
|
|
db.commit()
|
|
|
|
with pytest.raises(ValueError, match='客户'):
|
|
create_protocol(
|
|
db,
|
|
body=_body(
|
|
source,
|
|
**({'donationNo': 'D-CUSTOMER', 'protocolNo': None} if source == 'research_donation' else {}),
|
|
customer={'customerId': 'existing-c', 'motherName': '已有客户'},
|
|
),
|
|
ctx=_ctx('specialist'),
|
|
user_info=_user(),
|
|
)
|
|
|
|
|
|
def test_protocol_body_rejects_status_finance_and_conversion_fields():
|
|
from app.schemas.market_mobile import MarketProtocolBody
|
|
|
|
forbidden = [
|
|
{'auditStatus': 'shtg'},
|
|
{'protocolStatus': 'yzf'},
|
|
{'delFlag': 1},
|
|
{'refundAmount': 100},
|
|
{'financeRemark': '通过'},
|
|
{'toResearchTime': '2026-10-10'},
|
|
]
|
|
for field in forbidden:
|
|
with pytest.raises(ValidationError):
|
|
MarketProtocolBody.model_validate({'source': 'cell_storage', **field})
|
|
|
|
|
|
def test_cross_tenant_or_disabled_referenced_user_is_rejected():
|
|
from app.services.market_protocol import create_protocol
|
|
|
|
db = _session()
|
|
_masters(db)
|
|
db.add(
|
|
SysUser(
|
|
id='disabled', user_no='disabled', nick_name='停用', status='0', tenant_no='tenant-a', create_dept_id='', del_flag=0
|
|
)
|
|
)
|
|
db.commit()
|
|
|
|
with pytest.raises(ValueError, match='用户'):
|
|
create_protocol(
|
|
db,
|
|
body=_body(collection={'regionId': 'region-a', 'hospitalId': 'hospital-a', 'regionManagerUserId': 'disabled'}),
|
|
ctx=_ctx('specialist'),
|
|
user_info=_user(),
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize('source', ['cell_storage', 'research_donation'])
|
|
@pytest.mark.parametrize(
|
|
('sample_code_id', 'sample_code'),
|
|
[
|
|
('code-cross', 'S-CROSS'),
|
|
('code-deleted', 'S-DELETED'),
|
|
('code-a', 'S-WRONG'),
|
|
],
|
|
)
|
|
def test_protocol_rejects_invalid_or_inconsistent_sample_code_reference(
|
|
source, sample_code_id, sample_code
|
|
):
|
|
from app.services.market_protocol import create_protocol
|
|
|
|
db = _session()
|
|
_masters(db)
|
|
payload = {
|
|
'samples': [
|
|
{
|
|
'sampleTypeId': 'type-a',
|
|
'sampleCodeId': sample_code_id,
|
|
'sampleCode': sample_code,
|
|
'unitId': 'unit-a',
|
|
}
|
|
]
|
|
}
|
|
if source == 'research_donation':
|
|
payload.update({'donationNo': 'D-CODE', 'protocolNo': None})
|
|
|
|
with pytest.raises(ValueError, match='样本编码'):
|
|
create_protocol(
|
|
db,
|
|
body=_body(source, **payload),
|
|
ctx=_ctx('specialist'),
|
|
user_info=_user(),
|
|
)
|
|
|
|
|
|
def test_cell_update_rolls_back_main_and_existing_children_after_late_failure():
|
|
from app.schemas.market_mobile import MarketProtocolBody
|
|
from app.services.market_protocol import update_protocol
|
|
|
|
db = _session()
|
|
_masters(db)
|
|
_cell(db, 'cell-update')
|
|
db.add_all(
|
|
[
|
|
BizProtocolPartyCustomer(
|
|
id='party-old', collection_protocol_id='cell-update', mother_name='旧姓名', tenant_no='tenant-a', create_dept_id='', del_flag=0
|
|
),
|
|
BizProtocolSample(
|
|
id='sample-old', collection_protocol_id='cell-update', sample_code='OLD-CODE', tenant_no='tenant-a', create_dept_id='', del_flag=0
|
|
),
|
|
]
|
|
)
|
|
db.commit()
|
|
body = MarketProtocolBody.model_validate(
|
|
{
|
|
'source': 'cell_storage',
|
|
'remark': '新备注',
|
|
'customer': {'motherName': '新姓名'},
|
|
'collection': {'medicalStaff': '新医护'},
|
|
'samples': [{'sampleTypeId': 'type-a', 'sampleCodeId': 'code-a', 'sampleCode': 'S-001'}],
|
|
}
|
|
)
|
|
|
|
with patch(
|
|
'app.services.market_protocol.cell_crud.save_samples',
|
|
side_effect=RuntimeError('late failure'),
|
|
):
|
|
with pytest.raises(RuntimeError, match='late'):
|
|
update_protocol(
|
|
db,
|
|
source='cell_storage',
|
|
protocol_id='cell-update',
|
|
body=body,
|
|
ctx=_ctx('specialist'),
|
|
user_info=_user(),
|
|
)
|
|
|
|
db.expire_all()
|
|
assert db.get(BizCollectionProtocol, 'cell-update').remark is None
|
|
assert db.get(BizProtocolPartyCustomer, 'party-old').mother_name == '旧姓名'
|
|
assert db.get(BizProtocolPartyCustomer, 'party-old').del_flag == 0
|
|
assert db.get(BizProtocolSample, 'sample-old').sample_code == 'OLD-CODE'
|
|
assert db.get(BizProtocolCollectionInfo, 'ci-cell-update').medical_staff is None
|
|
|
|
|
|
def test_donation_update_rolls_back_main_and_old_children_after_soft_delete():
|
|
from app.schemas.market_mobile import MarketProtocolBody
|
|
from app.services.market_protocol import update_protocol
|
|
|
|
db = _session()
|
|
_masters(db)
|
|
_donation(db, 'don-update')
|
|
db.add_all(
|
|
[
|
|
BizResearchDonationParty(
|
|
id='dparty-old', research_donation_id='don-update', mother_name='旧姓名', tenant_no='tenant-a', create_dept_id='', del_flag=0
|
|
),
|
|
BizResearchDonationSample(
|
|
id='dsample-old', research_donation_id='don-update', sample_code='OLD-CODE', tenant_no='tenant-a', create_dept_id='', del_flag=0
|
|
),
|
|
]
|
|
)
|
|
db.commit()
|
|
body = MarketProtocolBody.model_validate(
|
|
{
|
|
'source': 'research_donation',
|
|
'donationNo': 'DON-don-update',
|
|
'remark': '新备注',
|
|
'customer': {'motherName': '新姓名'},
|
|
'collection': {'medicalStaff': '新医护'},
|
|
'samples': [{'sampleTypeId': 'type-a', 'sampleCodeId': 'code-a', 'sampleCode': 'S-001'}],
|
|
}
|
|
)
|
|
|
|
with patch(
|
|
'app.crud.research_donation._create_samples',
|
|
side_effect=RuntimeError('late failure'),
|
|
):
|
|
with pytest.raises(RuntimeError, match='late'):
|
|
update_protocol(
|
|
db,
|
|
source='research_donation',
|
|
protocol_id='don-update',
|
|
body=body,
|
|
ctx=_ctx('specialist'),
|
|
user_info=_user(),
|
|
)
|
|
|
|
db.expire_all()
|
|
assert db.get(BizResearchDonation, 'don-update').remark is None
|
|
assert db.get(BizResearchDonationParty, 'dparty-old').mother_name == '旧姓名'
|
|
assert db.get(BizResearchDonationParty, 'dparty-old').del_flag == 0
|
|
assert db.get(BizResearchDonationSample, 'dsample-old').del_flag == 0
|
|
assert db.get(BizResearchDonationCollection, 'dci-don-update').medical_staff is None
|
|
|
|
|
|
def test_manager_can_update_donation_in_scope_but_cannot_move_it_out():
|
|
from app.schemas.market_mobile import MarketProtocolBody
|
|
from app.services.market_protocol import update_protocol
|
|
|
|
db = _session()
|
|
_masters(db)
|
|
_donation(db, 'don-manager', sales='u2', manager='u1')
|
|
db.add(
|
|
BizResearchDonationParty(
|
|
id='dparty-manager', research_donation_id='don-manager', mother_name='旧姓名', tenant_no='tenant-a', create_dept_id='', del_flag=0
|
|
)
|
|
)
|
|
db.commit()
|
|
|
|
result = update_protocol(
|
|
db,
|
|
source='research_donation',
|
|
protocol_id='don-manager',
|
|
body=MarketProtocolBody.model_validate(
|
|
{
|
|
'source': 'research_donation',
|
|
'donationNo': 'DON-don-manager',
|
|
'remark': '经理已修改',
|
|
'customer': {'motherName': '新姓名'},
|
|
'collection': {'medicalStaff': '新医护'},
|
|
}
|
|
),
|
|
ctx=_ctx('manager'),
|
|
user_info=_user(),
|
|
)
|
|
assert result['basic']['remark'] == '经理已修改'
|
|
assert result['customer']['motherName'] == '新姓名'
|
|
assert result['collection']['medicalStaff'] == '新医护'
|
|
|
|
with pytest.raises(ValueError, match='范围'):
|
|
update_protocol(
|
|
db,
|
|
source='research_donation',
|
|
protocol_id='don-manager',
|
|
body=MarketProtocolBody.model_validate(
|
|
{
|
|
'source': 'research_donation',
|
|
'collection': {
|
|
'regionId': 'region-b',
|
|
'regionManagerUserId': 'manager-a',
|
|
},
|
|
}
|
|
),
|
|
ctx=_ctx('manager'),
|
|
user_info=_user(),
|
|
)
|
|
db.expire_all()
|
|
collection = db.query(BizResearchDonationCollection).filter_by(
|
|
research_donation_id='don-manager', del_flag=0
|
|
).one()
|
|
assert collection.region_id == 'region-a'
|
|
assert collection.region_manager_user_id == 'u1'
|
|
|
|
|
|
def test_protocol_routes_do_not_expose_audit_void_refund_or_conversion():
|
|
from app.api.market_mobile import router
|
|
|
|
paths = {route.path for route in router.routes}
|
|
disallowed = ('audit', 'void', 'refund', 'convert', 'to-research', 'to-storage')
|
|
protocol_paths = [path for path in paths if '/protocols' in path]
|
|
assert protocol_paths
|
|
assert all(not any(word in path for word in disallowed) for path in protocol_paths)
|
|
|
|
|
|
def test_cell_create_rolls_back_main_customer_and_children_on_child_failure():
|
|
from app.services.market_protocol import create_protocol
|
|
|
|
db = _session()
|
|
_masters(db)
|
|
|
|
with patch(
|
|
'app.services.market_protocol.cell_crud.upsert_collection_info',
|
|
side_effect=RuntimeError('injected child failure'),
|
|
):
|
|
with pytest.raises(RuntimeError, match='injected'):
|
|
create_protocol(
|
|
db, body=_body(), ctx=_ctx('specialist'), user_info=_user()
|
|
)
|
|
|
|
assert db.query(BizCollectionProtocol).count() == 0
|
|
assert db.query(BizProtocolPartyCustomer).count() == 0
|
|
assert db.query(BizCustomer).count() == 0
|
|
|
|
def test_donation_create_rolls_back_customer_and_children_on_child_failure():
|
|
from app.services.market_protocol import create_protocol
|
|
|
|
db = _session()
|
|
_masters(db)
|
|
|
|
with patch(
|
|
'app.crud.research_donation._create_collection',
|
|
side_effect=RuntimeError('injected child failure'),
|
|
):
|
|
with pytest.raises(RuntimeError, match='injected'):
|
|
create_protocol(
|
|
db,
|
|
body=_body(
|
|
'research_donation', donationNo='D-ROLLBACK', protocolNo=None
|
|
),
|
|
ctx=_ctx('specialist'),
|
|
user_info=_user(),
|
|
)
|
|
|
|
assert db.query(BizResearchDonation).count() == 0
|
|
assert db.query(BizResearchDonationParty).count() == 0
|
|
assert db.query(BizCustomer).count() == 0
|