Files
zkxw/backend/tests/test_market_protocol_api.py
T
zlwuandCursor 5e563a6669 fix: 子表保存改为就地同步,避免整批换 id 打断外键。
协议样本、生产产出和盘点明细按行更新并保留原 id;补齐存储证明质检预填、离职人员显示名,以及样本采集通知模板字段。

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-10-28 10:51:53 +08:00

4559 lines
155 KiB
Python

import json
from datetime import datetime, timedelta
from decimal import Decimal
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 Query, sessionmaker
from app.models.base import Base
from app.models.models import (
BizAttachment,
BizCellSampleType,
BizCollectionHospital,
BizCollectionProtocol,
BizCollectionRegion,
BizCollectionRegionStaff,
BizCustomer,
BizItemUnit,
BizProtocolCollectionInfo,
BizProtocolPartyCustomer,
BizProtocolPayment,
BizProtocolPaymentDetail,
BizProtocolPaymentSchedule,
BizProtocolPaymentTransfer,
BizProtocolSamplePayment,
BizProtocolSample,
BizProtocolSampleTransport,
BizPresignOrder,
BizPresignTransfer,
BizProtocolAudit,
BizProtocolStorageCert,
BizProtocolFullmoonTodo,
BizSampleCode,
BizSampleFreight,
BizSampleFreightItem,
BizSampleReception,
SysPost,
SysPostRole,
SysRole,
SysRoleDataScopeDetail,
SysDictData,
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)
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-b', sample_code='S-002', tenant_no='tenant-a', create_dept_id='', del_flag=0
),
BizSampleCode(
id='code-c', sample_code='S-003', 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
),
BizCollectionRegionStaff(
id='staff-u1-region-a', region_id='region-a', user_id='u1', is_leader=1,
tenant_no='tenant-a', create_dept_id='', del_flag=0,
),
BizCollectionRegionStaff(
id='staff-manager-a-region-a', region_id='region-a', user_id='manager-a', is_leader=0,
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_allows_web_parity_edit_without_status_change():
from app.services.market_protocol import update_protocol
db = _session()
_masters(db)
row = _cell(db, 'terminal')
row.protocol_status = 'yzf'
db.commit()
update_protocol(
db,
source='cell_storage',
protocol_id=row.id,
body=_body(protocolNo='CELL-terminal'),
ctx=_ctx('specialist'),
user_info=_user(),
)
db.expire_all()
saved = db.get(BizCollectionProtocol, row.id)
assert saved.protocol_status == 'yzf'
assert saved.protocol_no == 'CELL-terminal'
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_protocol_detail_resolves_web_display_names_for_mobile_fields():
from app.services.market_protocol import get_protocol
db = _session()
_masters(db)
row = _cell(db, 'named')
row.collector_user_id = 'u1'
collection = db.query(BizProtocolCollectionInfo).filter_by(
collection_protocol_id=row.id
).one()
collection.hospital_id = 'hospital-a'
db.add(
BizProtocolSample(
id='sample-named',
collection_protocol_id=row.id,
sample_type_id='type-a',
sample_code='S-NAMED',
unit_id='unit-a',
tenant_no='tenant-a',
create_dept_id='',
del_flag=0,
)
)
db.commit()
result = get_protocol(
db,
source='cell_storage',
protocol_id=row.id,
ctx=_ctx('specialist'),
user_info=_user(),
)
assert result['basic']['collectorUserName'] == '当前用户'
assert result['collection']['regionName'] == 'A区'
assert result['collection']['hospitalName'] == 'A院'
assert result['collection']['salesUserName'] == '当前用户'
assert result['collection']['regionManagerUserName'] == '经理'
assert result['samples'][0]['sampleTypeName'] == '样本A'
assert result['samples'][0]['unitName'] == '毫升'
def test_protocol_detail_resolves_soft_deleted_master_names_with_tenant_scope():
from app.services.market_protocol import get_protocol
db = _session()
_masters(db)
row = _cell(db, 'historical-master')
collection = db.query(BizProtocolCollectionInfo).filter_by(
collection_protocol_id=row.id
).one()
collection.hospital_id = 'hospital-a'
db.get(BizCollectionRegion, 'region-a').del_flag = 1
db.get(BizCollectionHospital, 'hospital-a').del_flag = 1
db.get(BizCellSampleType, 'type-a').del_flag = 1
db.get(BizItemUnit, 'unit-a').del_flag = 1
db.add_all(
[
BizCellSampleType(id='type-global-history', type_code='GLOBAL', type_name='历史全局样本', tenant_no='', create_dept_id='', del_flag=1),
BizCellSampleType(id='type-cross-history', type_code='CROSS', type_name='跨租户样本名称', tenant_no='tenant-b', create_dept_id='', del_flag=1),
BizItemUnit(id='unit-cross-history', unit_code='cross', unit_name='跨租户单位名称', tenant_no='tenant-b', create_dept_id='', del_flag=1),
BizProtocolSample(id='sample-history-local', collection_protocol_id=row.id, sample_type_id='type-a', sample_code='S-HISTORY-LOCAL', unit_id='unit-a', tenant_no='tenant-a', create_dept_id='', del_flag=0),
BizProtocolSample(id='sample-history-global', collection_protocol_id=row.id, sample_type_id='type-global-history', sample_code='S-HISTORY-GLOBAL', unit_id='', tenant_no='tenant-a', create_dept_id='', del_flag=0),
BizProtocolSample(id='sample-history-cross', collection_protocol_id=row.id, sample_type_id='type-cross-history', sample_code='S-HISTORY-CROSS', unit_id='unit-cross-history', tenant_no='tenant-a', create_dept_id='', del_flag=0),
]
)
db.commit()
result = get_protocol(
db,
source='cell_storage',
protocol_id=row.id,
ctx=_ctx('specialist'),
user_info=_user(),
)
samples = {item['sampleCode']: item for item in result['samples']}
assert result['collection']['regionName'] == 'A区'
assert result['collection']['hospitalName'] == 'A院'
assert samples['S-HISTORY-LOCAL']['sampleTypeName'] == '样本A'
assert samples['S-HISTORY-LOCAL']['unitName'] == '毫升'
assert samples['S-HISTORY-GLOBAL']['sampleTypeName'] == '历史全局样本'
assert samples['S-HISTORY-CROSS']['sampleTypeName'] == ''
assert samples['S-HISTORY-CROSS']['unitName'] == ''
def test_protocol_names_use_exact_historical_nicknames_without_raw_fallbacks():
from app.services.market_protocol import get_protocol, list_protocols
db = _session()
_masters(db)
row = _cell(db, 'historical-names')
row.collector_user_id = 'blank-name-user'
collection = db.query(BizProtocolCollectionInfo).filter_by(
collection_protocol_id=row.id
).one()
collection.region_manager_user_id = 'manager-a'
current_user = db.get(SysUser, 'u1')
current_user.nick_name = '历史删除专员'
current_user.del_flag = 1
manager = db.get(SysUser, 'manager-a')
manager.nick_name = '历史停用经理'
manager.status = '0'
db.add(
SysUser(
id='blank-name-user',
user_no='raw-collector-no',
nick_name='',
status='0',
tenant_no='tenant-a',
create_dept_id='',
del_flag=1,
)
)
sample_type = db.get(BizCellSampleType, 'type-a')
sample_type.type_name = ''
unit = db.get(BizItemUnit, 'unit-a')
unit.unit_name = ''
db.add(
BizProtocolSample(
id='sample-historical-names',
collection_protocol_id=row.id,
sample_type_id='type-a',
sample_code='S-HISTORY',
unit_id='unit-a',
tenant_no='tenant-a',
create_dept_id='',
del_flag=0,
)
)
db.commit()
items, total = list_protocols(
db, ctx=_ctx('specialist'), user_info=_user()
)
detail = get_protocol(
db,
source='cell_storage',
protocol_id=row.id,
ctx=_ctx('specialist'),
user_info=_user(),
)
assert total == 1
assert items[0]['salesUserName'] == '历史删除专员'
assert items[0]['collectorUserName'] == ''
assert items[0]['managerName'] == '历史停用经理'
assert items[0]['sampleTypeName'] == ''
assert detail['basic']['collectorUserName'] == ''
assert detail['collection']['salesUserName'] == '历史删除专员'
assert detail['collection']['regionManagerUserName'] == '历史停用经理'
assert detail['samples'][0]['sampleTypeName'] == ''
assert detail['samples'][0]['unitName'] == ''
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.protocol_status == 'sxz'
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_mobile_protocol_status_follows_web_options_and_transitions():
from app.services.market_protocol import create_protocol, update_protocol
db = _session()
_masters(db)
created = create_protocol(
db,
body=_body(protocolStatus='dsx'),
ctx=_ctx('specialist'),
user_info=_user(),
)
protocol_id = created['basic']['id']
assert db.get(BizCollectionProtocol, protocol_id).protocol_status == 'dsx'
updated = update_protocol(
db,
source='cell_storage',
protocol_id=protocol_id,
body=_body(protocolStatus='yzf'),
ctx=_ctx('specialist'),
user_info=_user(),
)
assert updated['basic']['protocolStatus'] == 'yzf'
with pytest.raises(ValueError, match='不可转为'):
update_protocol(
db,
source='cell_storage',
protocol_id=protocol_id,
body=_body(protocolStatus='sxz'),
ctx=_ctx('specialist'),
user_info=_user(),
)
donation = create_protocol(
db,
body=_body(
'research_donation',
donationNo='D-STATUS',
protocolNo=None,
donationStatus='sdfq',
attachments={},
samples=[
{
'sampleTypeId': 'type-a',
'sampleCodeId': 'code-b',
'sampleCode': 'S-002',
'unitId': 'unit-a',
'collectAmount': 10,
}
],
),
ctx=_ctx('specialist'),
user_info=_user(),
)
assert db.get(BizResearchDonation, donation['basic']['id']).donation_status == 'sdfq'
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.donation_status == 'sxz'
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_mobile_create_schedules_sample_collect_wx_notify_for_cell_and_donation():
from app.services.market_protocol import create_protocol
db_cell = _session()
_masters(db_cell)
with patch(
'app.services.market_protocol.schedule_sample_collect_notify'
) as schedule_cell:
cell = create_protocol(
db_cell, body=_body(), ctx=_ctx('specialist'), user_info=_user()
)
cell_kw = schedule_cell.call_args.kwargs
assert cell_kw['tenant_no'] == 'tenant-a'
assert cell_kw['protocol_id'] == cell['basic']['id']
assert cell_kw['protocol_kind'] == 'cell_storage'
assert cell_kw['before_codes'] == []
assert 'S-001' in cell_kw['submitted_codes']
db_don = _session()
_masters(db_don)
with patch(
'app.services.market_protocol.schedule_sample_collect_notify'
) as schedule_don:
donation = create_protocol(
db_don,
body=_body(
'research_donation',
donationNo='D-NOTIFY',
protocolNo=None,
samples=[
{
'sampleTypeId': 'type-a',
'unitId': 'unit-a',
'collectAmount': 10,
}
],
),
ctx=_ctx('specialist'),
user_info=_user(),
)
don_kw = schedule_don.call_args.kwargs
assert don_kw['protocol_id'] == donation['basic']['id']
assert don_kw['protocol_kind'] == 'research_donation'
assert don_kw['before_codes'] == []
assert don_kw['submitted_codes']
def test_mobile_update_schedules_notify_only_when_samples_submitted():
from app.schemas.market_mobile import MarketProtocolBody
from app.services.market_protocol import create_protocol, update_protocol
db = _session()
_masters(db)
with patch(
'app.services.market_protocol.schedule_sample_collect_notify'
):
created = create_protocol(
db, body=_body(), ctx=_ctx('specialist'), user_info=_user()
)
protocol_id = created['basic']['id']
with patch(
'app.services.market_protocol.schedule_sample_collect_notify'
) as schedule:
update_protocol(
db,
source='cell_storage',
protocol_id=protocol_id,
body=MarketProtocolBody.model_validate(
{'source': 'cell_storage', 'remark': '仅改备注'}
),
ctx=_ctx('specialist'),
user_info=_user(),
)
update_protocol(
db,
source='cell_storage',
protocol_id=protocol_id,
body=_body(
samples=[
{
'sampleTypeId': 'type-a',
'sampleCodeId': 'code-a',
'sampleCode': 'S-001',
'unitId': 'unit-a',
'collectAmount': 10,
},
{
'sampleTypeId': 'type-a',
'sampleCodeId': 'code-b',
'sampleCode': 'S-002',
'unitId': 'unit-a',
'collectAmount': 5,
},
]
),
ctx=_ctx('specialist'),
user_info=_user(),
)
assert schedule.call_count == 2
assert schedule.call_args_list[0].kwargs['submitted_codes'] is None
second = schedule.call_args_list[1].kwargs
assert 'S-001' in second['before_codes']
assert set(second['submitted_codes']) >= {'S-001', 'S-002'}
assert second['protocol_kind'] == 'cell_storage'
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': 'u1'}),
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'
@pytest.mark.parametrize('source', ['cell_storage', 'research_donation'])
def test_manager_create_cannot_self_assign_an_unmanaged_region(source):
from app.services.market_protocol import create_protocol
db = _session()
_masters(db)
body = _body(
source,
**(
{'donationNo': 'D-UNMANAGED', 'protocolNo': None}
if source == 'research_donation'
else {}
),
collection={
'regionId': 'region-b',
'regionManagerUserId': 'u1',
'salesUserId': 'u2',
},
attachments=None,
)
with pytest.raises(ValueError, match='范围'):
create_protocol(
db,
body=body,
ctx=_ctx('manager'),
user_info=_user(),
)
@pytest.mark.parametrize('source', ['cell_storage', 'research_donation'])
@pytest.mark.parametrize('operation', ['update', 'delete'])
def test_protocol_mutations_lock_the_scoped_parent_row(
monkeypatch, source, operation
):
from app.schemas.market_mobile import MarketProtocolBody
from app.services.market_protocol import delete_protocol, update_protocol
db = _session()
_masters(db)
row_id = f'{source}-{operation}-lock'
if source == 'cell_storage':
_cell(db, row_id)
else:
_donation(db, row_id)
db.commit()
lock_calls = []
original = Query.with_for_update
def record_lock(query, *args, **kwargs):
lock_calls.append(True)
return original(query, *args, **kwargs)
monkeypatch.setattr(Query, 'with_for_update', record_lock)
if operation == 'update':
update_protocol(
db,
source=source,
protocol_id=row_id,
body=MarketProtocolBody.model_validate(
{'source': source, 'remark': '锁后更新'}
),
ctx=_ctx('specialist'),
user_info=_user(),
)
else:
delete_protocol(
db,
source=source,
protocol_id=row_id,
ctx=_ctx('specialist'),
user_info=_user(),
)
assert lock_calls
def test_v1_cell_protocol_mobile_update_keeps_version_and_saves_legacy_payment():
from app.schemas.market_mobile import MarketProtocolBody
from app.services.market_protocol import update_protocol
db = _session()
_masters(db)
row = _cell(db, 'legacy-v1')
row.payment_model_version = 1
db.commit()
result = update_protocol(
db,
source='cell_storage',
protocol_id=row.id,
body=MarketProtocolBody.model_validate(
{
'source': 'cell_storage',
'remark': '移动端可编辑历史协议',
'payment': {
'payMode': 'dj',
'contractAmount': '999999999999.99',
'lumpSumDeadline': '20261231',
},
'paymentDetails': [
{
'installmentNo': 1,
'installmentDueDate': '20261231',
'dueAmount': '999999999999.99',
'transferType': 'wx',
'transferAmount': '1.01',
'transferTime': '2026-10-14 10:00:00',
'transferAttachments': 'protocol/a.pdf',
'remark': '移动草稿',
}
],
}
),
ctx=_ctx('specialist'),
user_info=_user(),
)
db.expire_all()
saved = db.get(BizCollectionProtocol, row.id)
assert saved.payment_model_version == 1
assert saved.remark == '移动端可编辑历史协议'
assert result['paymentModelVersion'] == 1
assert result['payment']['contractAmount'] == '999999999999.99'
assert result['paymentDetails'][0]['transferAmount'] == '1.01'
def test_v1_market_update_accepts_complete_generated_installment_amounts():
from app.schemas.market_mobile import MarketProtocolBody
from app.services.market_protocol import update_protocol
db = _session()
_masters(db)
row = _cell(db, 'legacy-v1-generated-installments')
row.payment_model_version = 1
db.commit()
update_protocol(
db,
source='cell_storage',
protocol_id=row.id,
body=MarketProtocolBody.model_validate(
{
'source': 'cell_storage',
'payment': {
'payMode': 'fq',
'contractAmount': '100.01',
'installmentCount': 2,
'installmentIntervalMonths': '1',
'installmentStartDate': '20260131',
'installmentAdvanceDays': 0,
'installmentRatios': ['50.00', '50.01'],
},
'paymentDetails': [
{
'installmentNo': 1,
'installmentDueDate': '20260131',
'dueAmount': '50.00',
},
{
'installmentNo': 2,
'installmentDueDate': '20260228',
'dueAmount': '50.01',
},
],
}
),
ctx=_ctx('specialist'),
user_info=_user(),
)
db.expire_all()
payment = db.query(BizProtocolPayment).filter_by(
collection_protocol_id=row.id, del_flag=0
).one()
assert payment.installment_ratios == '["50.00", "50.01"]'
details = db.query(BizProtocolPaymentDetail).filter_by(
payment_id=payment.id, del_flag=0
).order_by(BizProtocolPaymentDetail.installment_no).all()
assert [detail.due_amount for detail in details] == [
Decimal('50.00'), Decimal('50.01')
]
assert [detail.installment_due_date for detail in details] == [
'20260131', '20260228'
]
def test_protocol_body_legacy_payment_is_typed_and_rejects_sensitive_detail_fields():
from app.schemas.market_mobile import MarketProtocolBody
from fastapi.encoders import jsonable_encoder
body = MarketProtocolBody.model_validate(
{
'source': 'cell_storage',
'payment': {
'payMode': 'fq',
'contractAmount': '1000.01',
'installmentCount': 2,
'installmentIntervalMonths': '12.50',
'installmentStartDate': '20261014',
'installmentAdvanceDays': 30,
'installmentRatios': ['400.00', '600.01'],
},
'paymentDetails': [
{
'id': 'detail-a',
'installmentNo': 1,
'dueAmount': '400.00',
'transferType': 'bank',
'transferAmount': '1.01',
'transferTime': '2026-10-14 10:00:00',
'transferAttachments': 'proof.pdf',
'remark': '市场备注',
}
],
}
)
assert body.payment.contract_amount == Decimal('1000.01')
assert body.payment.installment_ratios == [Decimal('400.00'), Decimal('600.01')]
assert body.payment_details[0].transfer_amount == Decimal('1.01')
attachment_alias = MarketProtocolBody.model_validate(
{
'source': 'cell_storage',
'paymentDetails': [{'attachments': 'protocol/proof.pdf'}],
}
)
assert attachment_alias.payment_details[0].transfer_attachments == 'protocol/proof.pdf'
max_amount = MarketProtocolBody.model_validate(
{
'source': 'cell_storage',
'payment': {
'payMode': 'dj',
'contractAmount': '9999999999999999.99',
},
}
)
assert max_amount.payment.contract_amount == Decimal('9999999999999999.99')
with pytest.raises(ValidationError):
MarketProtocolBody.model_validate(
{
'source': 'cell_storage',
'payment': {'payMode': 'evil', 'contractAmount': '1.00'},
}
)
from app.crud import collection_protocol as cell_crud
serialized_payment = cell_crud._payment_to_dict(
BizProtocolPayment(
id='max-payment', collection_protocol_id='max-protocol',
contract_amount=Decimal('9999999999999999.99'),
installment_ratios='["4999999999999999.99", "5000000000000000.00"]',
tenant_no='tenant-a', create_dept_id='', del_flag=0,
)
)
serialized_detail = cell_crud._detail_to_dict(
BizProtocolPaymentDetail(
id='max-detail', payment_id='max-payment',
due_amount=Decimal('4999999999999999.99'),
transfer_amount=Decimal('0.01'), tenant_no='tenant-a',
create_dept_id='', del_flag=0,
)
)
assert serialized_payment['contractAmount'] == '9999999999999999.99'
assert jsonable_encoder(serialized_payment)['installmentRatios'] == [
'4999999999999999.99', '5000000000000000.00'
]
assert serialized_detail['dueAmount'] == '4999999999999999.99'
assert serialized_detail['transferAmount'] == '0.01'
for forbidden in (
{'payStatus': 'dfj'},
{'financeRemark': '越权'},
{'refundAmount': '1.00'},
{'financeConfirmAttachments': 'finance.pdf'},
):
with pytest.raises(ValidationError):
MarketProtocolBody.model_validate(
{
'source': 'cell_storage',
'paymentDetails': [{'dueAmount': '1.00', **forbidden}],
}
)
@pytest.mark.parametrize(
('section', 'field', 'value'),
[
('payment', 'lumpSumDeadline', '20261399'),
('payment', 'installmentStartDate', '20230229'),
('paymentDetails', 'installmentDueDate', '20260230'),
],
)
def test_protocol_body_legacy_payment_rejects_invalid_calendar_dates(
section, field, value
):
from app.schemas.market_mobile import MarketProtocolBody
payload = {'source': 'cell_storage'}
if section == 'paymentDetails':
payload[section] = [{field: value}]
else:
payload[section] = {'payMode': 'dj', field: value}
with pytest.raises(ValidationError, match='YYYYMMDD'):
MarketProtocolBody.model_validate(payload)
def test_protocol_body_legacy_payment_accepts_month_end_and_leap_dates():
from app.schemas.market_mobile import MarketProtocolBody
body = MarketProtocolBody.model_validate(
{
'source': 'cell_storage',
'payment': {
'payMode': 'fq',
'lumpSumDeadline': '20260228',
'installmentStartDate': '20240229',
},
'paymentDetails': [{'installmentDueDate': '20260228'}],
}
)
assert body.payment.installment_start_date == '20240229'
assert body.payment_details[0].installment_due_date == '20260228'
def test_protocol_body_blank_customer_numbers_are_unset():
from app.schemas.market_mobile import MarketProtocolBody
body = MarketProtocolBody.model_validate(
{
'source': 'cell_storage',
'customer': {
'motherName': '张敏',
'birthOrder': '',
'newbornWeightGram': ' ',
'apgar1min': '',
'apgar5min': ' 9 ',
},
}
)
assert body.customer.birth_order is None
assert body.customer.newborn_weight_gram is None
assert body.customer.apgar_1min is None
assert body.customer.apgar_5min == 9
whole = MarketProtocolBody.model_validate(
{
'source': 'research_donation',
'customer': {'birthOrder': '2.0', 'newbornWeightGram': '3200'},
}
)
assert whole.customer.birth_order == 2
assert whole.customer.newborn_weight_gram == 3200
with pytest.raises(ValidationError, match='新生儿体重须为整数'):
MarketProtocolBody.model_validate(
{
'source': 'cell_storage',
'customer': {'birthOrder': '1', 'newbornWeightGram': '3200.5'},
}
)
def test_protocol_body_normalizes_dashed_register_and_donation_dates():
from app.schemas.market_mobile import MarketProtocolBody
donation = MarketProtocolBody.model_validate(
{'source': 'research_donation', 'donationDate': '2026-06-17'}
)
assert donation.donation_date == '20260617'
cell = MarketProtocolBody.model_validate(
{'source': 'cell_storage', 'registerDate': '2026/06/17'}
)
assert cell.register_date == '20260617'
def test_protocol_payment_models_are_mutually_exclusive_by_stored_version():
from app.schemas.market_mobile import MarketProtocolBody
from app.services.market_protocol import update_protocol
db = _session()
_masters(db)
v1 = _cell(db, 'legacy-mixed')
v1.payment_model_version = 1
v2 = _cell(db, 'sample-mixed')
v2.payment_model_version = 2
db.commit()
with pytest.raises(ValueError, match='v1.*samplePayments|旧版.*样本缴费'):
update_protocol(
db,
source='cell_storage',
protocol_id=v1.id,
body=MarketProtocolBody.model_validate(
{
'source': 'cell_storage',
'samplePayments': [
{'sampleCode': 'S-001', 'payMode': 'dj'}
],
}
),
ctx=_ctx('specialist'),
user_info=_user(),
)
with pytest.raises(ValueError, match='v2.*payment|新版.*整单缴费'):
update_protocol(
db,
source='cell_storage',
protocol_id=v2.id,
body=MarketProtocolBody.model_validate(
{
'source': 'cell_storage',
'payment': {'payMode': 'dj', 'contractAmount': '1.00'},
'paymentDetails': [],
}
),
ctx=_ctx('specialist'),
user_info=_user(),
)
donation = _donation(db, 'donation-payment-mixed')
db.commit()
with pytest.raises(ValueError, match='科研捐赠.*缴费'):
update_protocol(
db,
source='research_donation',
protocol_id=donation.id,
body=MarketProtocolBody.model_validate(
{
'source': 'research_donation',
'payment': {'payMode': 'dj', 'contractAmount': '1.00'},
}
),
ctx=_ctx('specialist'),
user_info=_user(),
)
def test_v1_market_update_preserves_locked_rows_and_mutates_only_drafts():
from app.schemas.market_mobile import MarketProtocolBody
from app.services.market_protocol import update_protocol
db = _session()
_masters(db)
row = _cell(db, 'legacy-status-guard')
row.payment_model_version = 1
payment = BizProtocolPayment(
id='legacy-payment', collection_protocol_id=row.id, pay_mode='dj',
contract_amount=Decimal('1000.00'), tenant_no='tenant-a',
create_dept_id='', del_flag=0,
)
details = [
BizProtocolPaymentDetail(
id=f'detail-{status_name}', payment_id=payment.id,
pay_status=status, due_amount=Decimal('100.00'),
transfer_amount=Decimal('10.00'), remark='原值',
tenant_no='tenant-a', create_dept_id='', del_flag=0,
)
for status_name, status in (
('paid', 'yjf'), ('confirming', 'jfqrz'),
('empty', ''), ('spaces', ' '), ('draft', 'dfj'),
('padded-draft', ' dfj '), ('delete', 'dfj'),
)
]
db.add_all([payment, *details])
db.commit()
update_protocol(
db,
source='cell_storage',
protocol_id=row.id,
body=MarketProtocolBody.model_validate(
{
'source': 'cell_storage',
'payment': {'payMode': 'dj', 'contractAmount': '1000.00'},
'paymentDetails': [
{'id': 'detail-paid', 'dueAmount': '1.00', 'remark': '篡改'},
{'id': 'detail-confirming', 'dueAmount': '1.00', 'remark': '篡改'},
{'id': 'detail-empty', 'dueAmount': '1.00', 'remark': '篡改'},
{'id': 'detail-spaces', 'dueAmount': '1.00', 'remark': '篡改'},
{'id': 'detail-draft', 'dueAmount': '250.00', 'remark': '已更新'},
{'dueAmount': '350.00', 'remark': '新增草稿'},
],
}
),
ctx=_ctx('specialist'),
user_info=_user(),
)
db.expire_all()
for locked_id in (
'detail-paid', 'detail-confirming', 'detail-empty',
'detail-spaces', 'detail-padded-draft',
):
locked = db.get(BizProtocolPaymentDetail, locked_id)
assert locked.due_amount == Decimal('100.00')
assert locked.remark == '原值'
assert locked.del_flag == 0
assert db.get(BizProtocolPaymentDetail, 'detail-draft').due_amount == Decimal('250.00')
assert db.get(BizProtocolPaymentDetail, 'detail-delete').del_flag == 1
created = db.query(BizProtocolPaymentDetail).filter_by(
payment_id=payment.id, remark='新增草稿', del_flag=0
).one()
assert created.pay_status == 'dfj'
def test_v1_market_update_rolls_back_protocol_and_payment_on_detail_failure():
from app.schemas.market_mobile import MarketProtocolBody
from app.services.market_protocol import update_protocol
db = _session()
_masters(db)
row = _cell(db, 'legacy-atomic')
row.payment_model_version = 1
payment = BizProtocolPayment(
id='legacy-atomic-payment', collection_protocol_id=row.id,
pay_mode='fq', contract_amount=Decimal('2000.00'),
installment_count=2, installment_ratios='[1000, 1000]',
remark='原方案', tenant_no='tenant-a', create_dept_id='', del_flag=0,
)
locked = BizProtocolPaymentDetail(
id='legacy-atomic-locked', payment_id=payment.id, pay_status='yjf',
installment_no=2, due_amount=Decimal('1000.00'),
tenant_no='tenant-a', create_dept_id='', del_flag=0,
)
db.add_all([payment, locked])
db.commit()
with pytest.raises(ValueError, match='明细条数'):
update_protocol(
db,
source='cell_storage',
protocol_id=row.id,
body=MarketProtocolBody.model_validate(
{
'source': 'cell_storage',
'remark': '不应提交',
'payment': {
'payMode': 'fq', 'contractAmount': '700.00',
'installmentCount': 1, 'installmentRatios': ['700.00'],
},
'paymentDetails': [],
}
),
ctx=_ctx('specialist'),
user_info=_user(),
)
db.expire_all()
assert db.get(BizCollectionProtocol, row.id).remark is None
saved_payment = db.get(BizProtocolPayment, payment.id)
assert saved_payment.contract_amount == Decimal('2000.00')
assert saved_payment.installment_count == 2
assert saved_payment.remark == '原方案'
assert db.get(BizProtocolPaymentDetail, locked.id).del_flag == 0
def test_v1_market_update_rejects_locked_plan_inconsistency_and_rolls_back_everything():
from app.schemas.market_mobile import MarketProtocolBody
from app.services.market_protocol import update_protocol
db = _session()
_masters(db)
row = _cell(db, 'legacy-locked-plan')
row.payment_model_version = 1
payment = BizProtocolPayment(
id='legacy-locked-plan-payment', collection_protocol_id=row.id,
pay_mode='fq', contract_amount=Decimal('2000.00'),
installment_count=2, installment_interval_months=Decimal('12'),
installment_start_date='20261001', installment_advance_days=30,
installment_ratios='[1000, 1000]', tenant_no='tenant-a',
create_dept_id='', del_flag=0,
)
locked = BizProtocolPaymentDetail(
id='legacy-locked-plan-paid', payment_id=payment.id, pay_status='yjf',
installment_no=1, due_amount=Decimal('1000.00'),
tenant_no='tenant-a', create_dept_id='', del_flag=0,
)
db.add_all([payment, locked])
db.commit()
with pytest.raises(ValueError, match='已缴|缴清|锁定|合同金额|缴费计划'):
update_protocol(
db,
source='cell_storage',
protocol_id=row.id,
body=MarketProtocolBody.model_validate(
{
'source': 'cell_storage',
'remark': '不应提交',
'payment': {'payMode': 'dj', 'contractAmount': '1.00'},
'paymentDetails': [
{
'id': locked.id,
'installmentNo': 1,
'dueAmount': '1000.00',
}
],
}
),
ctx=_ctx('specialist'),
user_info=_user(),
)
db.expire_all()
assert db.get(BizCollectionProtocol, row.id).remark is None
saved_payment = db.get(BizProtocolPayment, payment.id)
assert saved_payment.pay_mode == 'fq'
assert saved_payment.contract_amount == Decimal('2000.00')
assert saved_payment.installment_count == 2
assert db.get(BizProtocolPaymentDetail, locked.id).del_flag == 0
def test_v1_market_update_blocks_plan_changes_during_finance_confirmation():
from app.schemas.market_mobile import MarketProtocolBody
from app.services.market_protocol import update_protocol
db = _session()
_masters(db)
row = _cell(db, 'legacy-confirming-plan')
row.payment_model_version = 1
payment = BizProtocolPayment(
id='legacy-confirming-plan-payment', collection_protocol_id=row.id,
pay_mode='fq', contract_amount=Decimal('2000.00'),
installment_count=2, installment_interval_months=Decimal('12'),
installment_start_date='20261001', installment_advance_days=30,
installment_ratios='[1000, 1000]', tenant_no='tenant-a',
create_dept_id='', del_flag=0,
)
confirming = BizProtocolPaymentDetail(
id='legacy-confirming-plan-locked', payment_id=payment.id,
pay_status='jfqrz', installment_no=1, due_amount=Decimal('1000.00'),
tenant_no='tenant-a', create_dept_id='', del_flag=0,
)
draft = BizProtocolPaymentDetail(
id='legacy-confirming-plan-draft', payment_id=payment.id,
pay_status='dfj', installment_no=2, due_amount=Decimal('1000.00'),
tenant_no='tenant-a', create_dept_id='', del_flag=0,
)
db.add_all([payment, confirming, draft])
db.commit()
with pytest.raises(ValueError, match='确认中|锁定|缴费计划'):
update_protocol(
db,
source='cell_storage',
protocol_id=row.id,
body=MarketProtocolBody.model_validate(
{
'source': 'cell_storage',
'remark': '不应提交',
'payment': {
'payMode': 'fq',
'contractAmount': '2000.00',
'installmentCount': 2,
'installmentIntervalMonths': '24',
'installmentStartDate': '20261001',
'installmentAdvanceDays': 30,
'installmentRatios': ['1000.00', '1000.00'],
},
'paymentDetails': [
{
'id': confirming.id,
'installmentNo': 1,
'dueAmount': '1000.00',
},
{
'id': draft.id,
'installmentNo': 2,
'dueAmount': '1000.00',
},
],
}
),
ctx=_ctx('specialist'),
user_info=_user(),
)
db.expire_all()
assert db.get(BizCollectionProtocol, row.id).remark is None
assert db.get(BizProtocolPayment, payment.id).installment_interval_months == Decimal('12')
def test_v1_market_update_rejects_stale_draft_schedule_after_paid_plan_change():
from app.schemas.market_mobile import MarketProtocolBody
from app.services.market_protocol import update_protocol
db = _session()
_masters(db)
row = _cell(db, 'legacy-stale-plan')
row.payment_model_version = 1
payment = BizProtocolPayment(
id='legacy-stale-plan-payment', collection_protocol_id=row.id,
pay_mode='fq', contract_amount=Decimal('2000.00'),
installment_count=2, installment_interval_months=Decimal('12'),
installment_start_date='20261001', installment_advance_days=30,
installment_ratios='[1000, 1000]', tenant_no='tenant-a',
create_dept_id='', del_flag=0,
)
paid = BizProtocolPaymentDetail(
id='legacy-stale-plan-paid', payment_id=payment.id, pay_status='yjf',
installment_no=1, installment_due_date='20261001',
due_amount=Decimal('1000.00'), tenant_no='tenant-a',
create_dept_id='', del_flag=0,
)
draft = BizProtocolPaymentDetail(
id='legacy-stale-plan-draft', payment_id=payment.id, pay_status='dfj',
installment_no=2, installment_due_date='20270901',
due_amount=Decimal('1000.00'), tenant_no='tenant-a',
create_dept_id='', del_flag=0,
)
db.add_all([payment, paid, draft])
db.commit()
with pytest.raises(ValueError, match='缴费计划|应缴时间'):
update_protocol(
db,
source='cell_storage',
protocol_id=row.id,
body=MarketProtocolBody.model_validate(
{
'source': 'cell_storage',
'remark': '不应提交',
'payment': {
'payMode': 'fq', 'contractAmount': '2000.00',
'installmentCount': 2, 'installmentIntervalMonths': '24',
'installmentStartDate': '20261001',
'installmentAdvanceDays': 30,
'installmentRatios': ['1000.00', '1000.00'],
},
'paymentDetails': [
{
'id': paid.id, 'installmentNo': 1,
'installmentDueDate': '20261001',
'dueAmount': '1000.00',
},
{
'id': draft.id, 'installmentNo': 2,
'installmentDueDate': '20270901',
'dueAmount': '1000.00',
},
],
}
),
ctx=_ctx('specialist'),
user_info=_user(),
)
db.expire_all()
assert db.get(BizCollectionProtocol, row.id).remark is None
assert db.get(BizProtocolPayment, payment.id).installment_interval_months == Decimal('12')
assert db.get(BizProtocolPaymentDetail, draft.id).installment_due_date == '20270901'
result = update_protocol(
db,
source='cell_storage',
protocol_id=row.id,
body=MarketProtocolBody.model_validate(
{
'source': 'cell_storage',
'remark': '已对齐新计划',
'payment': {
'payMode': 'fq', 'contractAmount': '2000.00',
'installmentCount': 2, 'installmentIntervalMonths': '24',
'installmentStartDate': '20261001',
'installmentAdvanceDays': 30,
'installmentRatios': ['1000.00', '1000.00'],
},
'paymentDetails': [
{
'id': paid.id, 'installmentNo': 1,
'installmentDueDate': '20261001',
'dueAmount': '1000.00',
},
{
'id': draft.id, 'installmentNo': 2,
'installmentDueDate': '20280901',
'dueAmount': '1000.00',
},
],
}
),
ctx=_ctx('specialist'),
user_info=_user(),
)
assert result['payment']['installmentIntervalMonths'] == 24.0
db.expire_all()
assert db.get(BizCollectionProtocol, row.id).remark == '已对齐新计划'
assert db.get(BizProtocolPaymentDetail, draft.id).installment_due_date == '20280901'
def test_v1_market_update_requires_rebuilt_lump_sum_after_paid_installment_switch():
from app.schemas.market_mobile import MarketProtocolBody
from app.services.market_protocol import update_protocol
db = _session()
_masters(db)
row = _cell(db, 'legacy-switch-to-lump-sum')
row.payment_model_version = 1
payment = BizProtocolPayment(
id='legacy-switch-to-lump-sum-payment', collection_protocol_id=row.id,
pay_mode='fq', contract_amount=Decimal('2000.00'),
installment_count=2, installment_interval_months=Decimal('12'),
installment_start_date='20261001', installment_advance_days=30,
installment_ratios='[1000, 1000]', tenant_no='tenant-a',
create_dept_id='', del_flag=0,
)
paid = BizProtocolPaymentDetail(
id='legacy-switch-to-lump-sum-paid', payment_id=payment.id,
pay_status='yjf', installment_no=1, installment_due_date='20261001',
due_amount=Decimal('1000.00'), tenant_no='tenant-a',
create_dept_id='', del_flag=0,
)
old_draft = BizProtocolPaymentDetail(
id='legacy-switch-to-lump-sum-draft', payment_id=payment.id,
pay_status='dfj', installment_no=2, installment_due_date='20270901',
due_amount=Decimal('1000.00'), tenant_no='tenant-a',
create_dept_id='', del_flag=0,
)
db.add_all([payment, paid, old_draft])
db.commit()
payment_payload = {
'payMode': 'dj', 'contractAmount': '2000.00',
'lumpSumDeadline': '20261231',
}
with pytest.raises(ValueError, match='趸交|待缴|期号|应缴时间'):
update_protocol(
db,
source='cell_storage',
protocol_id=row.id,
body=MarketProtocolBody.model_validate(
{
'source': 'cell_storage',
'remark': '不应提交',
'payment': payment_payload,
'paymentDetails': [
{
'id': paid.id, 'installmentNo': 1,
'installmentDueDate': '20261001',
'dueAmount': '1000.00',
},
{
'id': old_draft.id, 'installmentNo': 2,
'installmentDueDate': '20270901',
'dueAmount': '1000.00',
},
],
}
),
ctx=_ctx('specialist'),
user_info=_user(),
)
db.expire_all()
assert db.get(BizCollectionProtocol, row.id).remark is None
assert db.get(BizProtocolPayment, payment.id).pay_mode == 'fq'
assert db.get(BizProtocolPaymentDetail, old_draft.id).del_flag == 0
result = update_protocol(
db,
source='cell_storage',
protocol_id=row.id,
body=MarketProtocolBody.model_validate(
{
'source': 'cell_storage',
'remark': '转趸交',
'payment': payment_payload,
'paymentDetails': [
{
'id': paid.id, 'installmentNo': 1,
'installmentDueDate': '20261001',
'dueAmount': '1000.00',
},
{
'installmentNo': None,
'installmentDueDate': '20261231',
'dueAmount': '1000.00',
},
],
}
),
ctx=_ctx('specialist'),
user_info=_user(),
)
editable = [item for item in result['paymentDetails'] if item['payStatus'] == 'dfj']
assert len(editable) == 1
assert editable[0]['installmentNo'] is None
assert editable[0]['installmentDueDate'] == '20261231'
assert editable[0]['dueAmount'] == '1000.00'
db.expire_all()
assert db.get(BizProtocolPaymentDetail, old_draft.id).del_flag == 1
def test_v1_market_update_rejects_details_without_payment_and_keeps_attachment_staging():
from app.schemas.market_mobile import MarketProtocolBody
from app.services.market_protocol import update_protocol
db = _session()
_masters(db)
row = _cell(db, 'legacy-details-without-parent')
row.payment_model_version = 1
db.commit()
with pytest.raises(ValueError, match='缴费主表|缴费方案'):
update_protocol(
db,
source='cell_storage',
protocol_id=row.id,
body=MarketProtocolBody.model_validate(
{
'source': 'cell_storage',
'remark': '不应提交',
'paymentDetails': [
{
'dueAmount': '100.00',
'attachments': 'protocol/a.pdf',
}
],
}
),
ctx=_ctx('specialist'),
user_info=_user(),
)
db.expire_all()
assert db.get(BizCollectionProtocol, row.id).remark is None
assert db.query(BizProtocolPayment).count() == 0
assert db.query(BizProtocolPaymentDetail).count() == 0
attachment = db.get(BizAttachment, 'att-protocol-a')
assert attachment.business_type == 'market:specialist:protocol:staging'
assert attachment.business_id in (None, '')
@pytest.mark.parametrize(
'attachment',
[
{
'id': 'unknown-ref', 'file_url': 'protocol/unknown.pdf',
'business_type': 'market:specialist:protocol:staging',
'tenant_no': 'tenant-a', 'create_by': 'u1',
},
{
'id': 'cross-tenant-ref', 'file_url': 'protocol/cross.pdf',
'business_type': 'market:specialist:protocol:staging',
'tenant_no': 'tenant-b', 'create_by': 'u1',
},
{
'id': 'wrong-module-ref', 'file_url': 'payment/wrong-module.pdf',
'business_type': 'market:specialist:payment:staging',
'tenant_no': 'tenant-a', 'create_by': 'u1',
},
{
'id': 'other-user-ref', 'file_url': 'protocol/other-user.pdf',
'business_type': 'market:specialist:protocol:staging',
'tenant_no': 'tenant-a', 'create_by': 'u2',
},
],
)
def test_v1_market_update_rejects_untrusted_payment_detail_attachment_and_rolls_back(attachment):
from app.schemas.market_mobile import MarketProtocolBody
from app.services.market_protocol import update_protocol
db = _session()
_masters(db)
row = _cell(db, f"legacy-{attachment['id']}")
row.payment_model_version = 1
payment = BizProtocolPayment(
id=f"payment-{attachment['id']}", collection_protocol_id=row.id,
pay_mode='dj', contract_amount=Decimal('100.00'),
tenant_no='tenant-a', create_dept_id='', del_flag=0,
)
detail = BizProtocolPaymentDetail(
id=f"detail-{attachment['id']}", payment_id=payment.id,
pay_status='dfj', due_amount=Decimal('100.00'),
tenant_no='tenant-a', create_dept_id='', del_flag=0,
)
if attachment['id'] != 'unknown-ref':
db.add(
BizAttachment(
id=attachment['id'], business_type=attachment['business_type'],
original_name='proof.pdf', file_ext='pdf',
file_url=attachment['file_url'], file_size=10,
file_type='application/pdf', tenant_no=attachment['tenant_no'],
create_by=attachment['create_by'], create_dept_id='', del_flag=0,
)
)
db.add_all([payment, detail])
db.commit()
with pytest.raises(ValueError, match='附件'):
update_protocol(
db,
source='cell_storage',
protocol_id=row.id,
body=MarketProtocolBody.model_validate(
{
'source': 'cell_storage',
'remark': '不应提交',
'paymentDetails': [
{
'id': detail.id,
'dueAmount': '100.00',
'transferAttachments': attachment['file_url'],
}
],
}
),
ctx=_ctx('specialist'),
user_info=_user(),
)
db.expire_all()
assert db.get(BizCollectionProtocol, row.id).remark is None
assert db.get(BizProtocolPaymentDetail, detail.id).transfer_attachments is None
def test_v1_market_update_binds_payment_detail_attachment_to_protocol_business():
from app.schemas.market_mobile import MarketProtocolBody
from app.services.market_protocol import update_protocol
db = _session()
_masters(db)
row = _cell(db, 'legacy-payment-attachment')
row.payment_model_version = 1
payment = BizProtocolPayment(
id='legacy-payment-attachment-parent', collection_protocol_id=row.id,
pay_mode='dj', contract_amount=Decimal('100.00'),
tenant_no='tenant-a', create_dept_id='', del_flag=0,
)
detail = BizProtocolPaymentDetail(
id='legacy-payment-attachment-detail', payment_id=payment.id,
pay_status='dfj', due_amount=Decimal('100.00'),
tenant_no='tenant-a', create_dept_id='', del_flag=0,
)
db.add_all([payment, detail])
db.commit()
update_protocol(
db,
source='cell_storage',
protocol_id=row.id,
body=MarketProtocolBody.model_validate(
{
'source': 'cell_storage',
'paymentDetails': [
{
'id': detail.id,
'dueAmount': '100.00',
'attachments': 'protocol/a.pdf',
}
],
}
),
ctx=_ctx('specialist'),
user_info=_user(),
)
db.expire_all()
saved_detail = db.get(BizProtocolPaymentDetail, detail.id)
saved_attachment = db.get(BizAttachment, 'att-protocol-a')
assert saved_detail.transfer_attachments == 'protocol/a.pdf'
assert saved_attachment.business_type == 'market:specialist:protocol:bound'
assert saved_attachment.business_id == row.id
def test_protocol_attachment_download_authorizes_exact_payment_detail_relation_only():
from app.services.market_support import get_market_attachment_download_url
class FakeStorage:
@staticmethod
def presigned_download_url(file_url, file_name):
return f'https://download.test/{file_url}?name={file_name}'
db = _session()
own = _cell(db, 'payment-attachment-own')
hidden = _cell(db, 'payment-attachment-hidden', sales='u2')
own_payment = BizProtocolPayment(
id='payment-attachment-own-parent', collection_protocol_id=own.id,
pay_mode='dj', tenant_no='tenant-a', create_dept_id='', del_flag=0,
)
hidden_payment = BizProtocolPayment(
id='payment-attachment-hidden-parent', collection_protocol_id=hidden.id,
pay_mode='dj', tenant_no='tenant-a', create_dept_id='', del_flag=0,
)
db.add_all(
[
own_payment,
hidden_payment,
BizProtocolPaymentDetail(
id='payment-attachment-own-detail', payment_id=own_payment.id,
transfer_attachments='protocol/exact.pdf', pay_status='dfj',
tenant_no='tenant-a', create_dept_id='', del_flag=0,
),
BizProtocolPaymentDetail(
id='payment-attachment-hidden-detail', payment_id=hidden_payment.id,
transfer_attachments='protocol/hidden.pdf', pay_status='dfj',
tenant_no='tenant-a', create_dept_id='', del_flag=0,
),
BizProtocolPaymentDetail(
id='payment-attachment-substring-detail', payment_id=own_payment.id,
transfer_attachments='protocol/exact.pdf.evil', pay_status='dfj',
tenant_no='tenant-a', create_dept_id='', del_flag=0,
),
BizAttachment(
id='payment-attachment-exact',
business_type='market:specialist:protocol:bound', business_id=own.id,
original_name='exact.pdf', file_ext='pdf', file_url='protocol/exact.pdf',
file_size=10, file_type='application/pdf', tenant_no='tenant-a',
create_by='u1', create_dept_id='', del_flag=0,
),
BizAttachment(
id='payment-attachment-hidden',
business_type='market:specialist:protocol:bound', business_id=hidden.id,
original_name='hidden.pdf', file_ext='pdf', file_url='protocol/hidden.pdf',
file_size=10, file_type='application/pdf', tenant_no='tenant-a',
create_by='u2', create_dept_id='', del_flag=0,
),
BizAttachment(
id='payment-attachment-substring',
business_type='market:specialist:protocol:bound', business_id=own.id,
original_name='exact-short.pdf', file_ext='pdf', file_url='protocol/exact.pdf.evil',
file_size=10, file_type='application/pdf', tenant_no='tenant-a',
create_by='u1', create_dept_id='', del_flag=0,
),
]
)
db.commit()
result = get_market_attachment_download_url(
db,
file_url='protocol/exact.pdf',
file_name=None,
module='protocol',
ctx=_ctx('specialist'),
user_info=_user(),
storage_backend=FakeStorage(),
)
assert result['downloadUrl'].startswith('https://download.test/protocol/exact.pdf?')
with pytest.raises(PermissionError):
get_market_attachment_download_url(
db,
file_url='protocol/hidden.pdf',
file_name=None,
module='protocol',
ctx=_ctx('specialist'),
user_info=_user(),
storage_backend=FakeStorage(),
)
db.query(BizProtocolPaymentDetail).filter_by(
id='payment-attachment-own-detail'
).delete()
db.commit()
with pytest.raises(PermissionError):
get_market_attachment_download_url(
db,
file_url='protocol/exact.pdf',
file_name=None,
module='protocol',
ctx=_ctx('specialist'),
user_info=_user(),
storage_backend=FakeStorage(),
)
@pytest.mark.parametrize('boundary', ['tenant', 'scope'])
def test_v1_market_update_cannot_cross_tenant_or_manager_scope(boundary):
from app.schemas.market_mobile import MarketProtocolBody
from app.services.market_protocol import update_protocol
db = _session()
_masters(db)
row = _cell(
db,
f'legacy-{boundary}-hidden',
sales='u2',
manager='manager-a',
region='region-b' if boundary == 'scope' else 'region-a',
)
row.payment_model_version = 1
payment = BizProtocolPayment(
id=f'legacy-{boundary}-payment',
collection_protocol_id=row.id,
pay_mode='dj',
contract_amount=Decimal('100.00'),
tenant_no='tenant-a',
create_dept_id='',
del_flag=0,
)
if boundary == 'tenant':
row.tenant_no = 'tenant-b'
db.query(BizProtocolCollectionInfo).filter_by(
collection_protocol_id=row.id
).update({'tenant_no': 'tenant-b'})
payment.tenant_no = 'tenant-b'
db.add(payment)
db.commit()
with pytest.raises(HTTPException):
update_protocol(
db,
source='cell_storage',
protocol_id=row.id,
body=MarketProtocolBody.model_validate(
{
'source': 'cell_storage',
'payment': {'payMode': 'dj', 'contractAmount': '999.00'},
'paymentDetails': [],
}
),
ctx=_ctx('specialist' if boundary == 'tenant' else 'manager'),
user_info=_user(),
)
db.expire_all()
assert db.get(BizProtocolPayment, payment.id).contract_amount == Decimal('100.00')
def test_cell_sample_partial_edit_preserves_hidden_lifecycle_fields():
from app.schemas.market_mobile import MarketProtocolBody
from app.services.market_protocol import update_protocol
db = _session()
_masters(db)
row = _cell(db, 'sample-merge')
db.add(
BizProtocolSample(
id='sample-merge-old',
collection_protocol_id=row.id,
sample_type_id='type-a',
sample_code_id='code-a',
sample_code='S-001',
collect_amount=12.5,
unit_id='unit-a',
price=88,
original_price=99,
storage_years=20,
sample_status='yrk',
storage_in_date='20261001',
detect_start_date='20260929',
void_date='20261002',
storage_expire_date='20461001',
is_special=1,
special_remark='历史特殊说明',
remark='历史样本备注',
tenant_no='tenant-a',
create_dept_id='',
del_flag=0,
)
)
db.commit()
with patch(
'app.crud.collection_protocol._migrate_sample_child_refs_on_replace'
):
update_protocol(
db,
source='cell_storage',
protocol_id=row.id,
body=MarketProtocolBody.model_validate(
{
'source': 'cell_storage',
'samples': [
{
'sampleTypeId': 'type-a',
'sampleCodeId': 'code-a',
'sampleCode': 'S-001',
'price': 77,
}
],
}
),
ctx=_ctx('specialist'),
user_info=_user(),
)
saved = db.query(BizProtocolSample).filter_by(
collection_protocol_id=row.id, del_flag=0
).one()
assert float(saved.collect_amount) == 12.5
assert saved.unit_id == 'unit-a'
assert saved.remark == '历史样本备注'
assert saved.storage_in_date == '20261001'
assert saved.detect_start_date == '20260929'
assert saved.void_date == '20261002'
assert saved.storage_expire_date == '20461001'
assert float(saved.price) == 77
def test_donation_sample_partial_edit_preserves_hidden_lifecycle_fields():
from app.schemas.market_mobile import MarketProtocolBody
from app.services.market_protocol import update_protocol
db = _session()
_masters(db)
row = _donation(db, 'donation-sample-merge')
db.add(
BizResearchDonationSample(
id='donation-sample-merge-old',
research_donation_id=row.id,
sample_type_id='type-a',
sample_code_id='code-a',
sample_code='S-001',
collect_amount=9.5,
unit_id='unit-a',
storage_years=10,
sample_status='jcz',
storage_in_date='20261001',
detect_start_date='20260929',
void_date='20261002',
storage_expire_date='20361001',
is_special=1,
special_remark='历史捐赠说明',
remark='历史捐赠备注',
tenant_no='tenant-a',
create_dept_id='',
del_flag=0,
)
)
db.commit()
with patch(
'app.crud.research_donation._migrate_donation_sample_child_refs_on_replace'
):
update_protocol(
db,
source='research_donation',
protocol_id=row.id,
body=MarketProtocolBody.model_validate(
{
'source': 'research_donation',
'samples': [
{
'sampleTypeId': 'type-a',
'sampleCodeId': 'code-a',
'sampleCode': 'S-001',
'storageYears': 11,
}
],
}
),
ctx=_ctx('specialist'),
user_info=_user(),
)
saved = db.query(BizResearchDonationSample).filter_by(
research_donation_id=row.id, del_flag=0
).one()
assert float(saved.collect_amount) == 9.5
assert saved.unit_id == 'unit-a'
assert saved.remark == '历史捐赠备注'
assert saved.storage_in_date == '20261001'
assert saved.detect_start_date == '20260929'
assert saved.void_date == '20261002'
assert saved.storage_expire_date == '20361001'
assert saved.storage_years == 11
def _protocol_sample_row(source, *, row_id, protocol_id, code_id, code, amount, remark):
common = {
'id': row_id,
'sample_type_id': 'type-a',
'sample_code_id': code_id,
'sample_code': code,
'collect_amount': amount,
'unit_id': 'unit-a',
'sample_status': 'yrk',
'remark': remark,
'tenant_no': 'tenant-a',
'create_dept_id': '',
'del_flag': 0,
}
if source == 'cell_storage':
return BizProtocolSample(collection_protocol_id=protocol_id, **common)
return BizResearchDonationSample(research_donation_id=protocol_id, **common)
def _active_protocol_samples(db, source, protocol_id):
if source == 'cell_storage':
return (
db.query(BizProtocolSample)
.filter_by(collection_protocol_id=protocol_id, del_flag=0)
.order_by(BizProtocolSample.order_num.asc())
.all()
)
return (
db.query(BizResearchDonationSample)
.filter_by(research_donation_id=protocol_id, del_flag=0)
.order_by(BizResearchDonationSample.order_num.asc())
.all()
)
@pytest.mark.parametrize('source', ['cell_storage', 'research_donation'])
def test_protocol_sample_update_uses_id_for_insert_and_reorder(source):
from app.schemas.market_mobile import MarketProtocolBody
from app.services.market_protocol import update_protocol
db = _session()
_masters(db)
row_id = f'{source}-sample-reorder'
(_cell if source == 'cell_storage' else _donation)(db, row_id)
db.add_all(
[
_protocol_sample_row(
source,
row_id='sample-a',
protocol_id=row_id,
code_id='code-a',
code='S-001',
amount=11,
remark='隐藏-A',
),
_protocol_sample_row(
source,
row_id='sample-b',
protocol_id=row_id,
code_id='code-b',
code='S-002',
amount=22,
remark='隐藏-B',
),
]
)
db.commit()
migrate_target = (
'app.crud.collection_protocol._migrate_sample_child_refs_on_replace'
if source == 'cell_storage'
else 'app.crud.research_donation._migrate_donation_sample_child_refs_on_replace'
)
with patch(migrate_target):
update_protocol(
db,
source=source,
protocol_id=row_id,
body=MarketProtocolBody.model_validate(
{
'source': source,
'samples': [
{
'sampleTypeId': 'type-a',
'sampleCodeId': 'code-c',
'sampleCode': 'S-003',
},
{
'id': 'sample-b',
'sampleTypeId': 'type-a',
'sampleCodeId': 'code-b',
'sampleCode': 'S-002',
},
{
'id': 'sample-a',
'sampleTypeId': 'type-a',
'sampleCodeId': 'code-a',
'sampleCode': 'S-001',
},
],
}
),
ctx=_ctx('specialist'),
user_info=_user(),
)
saved = _active_protocol_samples(db, source, row_id)
assert [row.sample_code for row in saved] == ['S-003', 'S-002', 'S-001']
assert [row.remark for row in saved] == [None, '隐藏-B', '隐藏-A']
assert [float(row.collect_amount) if row.collect_amount is not None else None for row in saved] == [
None,
22.0,
11.0,
]
@pytest.mark.parametrize('source', ['cell_storage', 'research_donation'])
def test_protocol_sample_update_uses_id_when_replacing_code(source):
from app.schemas.market_mobile import MarketProtocolBody
from app.services.market_protocol import update_protocol
db = _session()
_masters(db)
row_id = f'{source}-sample-recode'
(_cell if source == 'cell_storage' else _donation)(db, row_id)
db.add(
_protocol_sample_row(
source,
row_id='sample-recode',
protocol_id=row_id,
code_id='code-a',
code='S-001',
amount=15,
remark='必须保留',
)
)
db.commit()
migrate_target = (
'app.crud.collection_protocol._migrate_sample_child_refs_on_replace'
if source == 'cell_storage'
else 'app.crud.research_donation._migrate_donation_sample_child_refs_on_replace'
)
with patch(migrate_target):
update_protocol(
db,
source=source,
protocol_id=row_id,
body=MarketProtocolBody.model_validate(
{
'source': source,
'samples': [
{
'sampleId': 'sample-recode',
'sampleTypeId': 'type-a',
'sampleCodeId': 'code-c',
'sampleCode': 'S-003',
}
],
}
),
ctx=_ctx('specialist'),
user_info=_user(),
)
saved = _active_protocol_samples(db, source, row_id)[0]
assert saved.sample_code_id == 'code-c'
assert saved.sample_code == 'S-003'
assert float(saved.collect_amount) == 15
assert saved.remark == '必须保留'
@pytest.mark.parametrize('source', ['cell_storage', 'research_donation'])
@pytest.mark.parametrize('duplicate_field', ['sampleCode', 'sampleCodeId'])
def test_protocol_sample_write_rejects_duplicate_identity(source, duplicate_field):
from app.services.market_protocol import create_protocol
db = _session()
_masters(db)
if duplicate_field == 'sampleCode':
samples = [
{'sampleTypeId': 'type-a', 'sampleCode': 'DUP-001'},
{'sampleTypeId': 'type-a', 'sampleCode': 'DUP-001'},
]
else:
samples = [
{'sampleTypeId': 'type-a', 'sampleCodeId': 'code-a', 'sampleCode': 'S-001'},
{'sampleTypeId': 'type-a', 'sampleCodeId': 'code-a', 'sampleCode': 'S-ALT'},
]
overrides = {'samples': samples}
if source == 'research_donation':
overrides.update({'donationNo': 'D-DUP', 'protocolNo': None})
with pytest.raises(ValueError, match=f'{duplicate_field}.*重复'):
create_protocol(
db,
body=_body(source, **overrides),
ctx=_ctx('specialist'),
user_info=_user(),
)
@pytest.mark.parametrize('source', ['cell_storage', 'research_donation'])
def test_protocol_sample_update_rejects_duplicate_sample_id(source):
from app.schemas.market_mobile import MarketProtocolBody
from app.services.market_protocol import update_protocol
db = _session()
_masters(db)
row_id = f'{source}-duplicate-sample-id'
(_cell if source == 'cell_storage' else _donation)(db, row_id)
db.add(
_protocol_sample_row(
source,
row_id='sample-duplicate-id',
protocol_id=row_id,
code_id='code-a',
code='S-001',
amount=10,
remark='不能双重继承',
)
)
db.commit()
migrate_target = (
'app.crud.collection_protocol._migrate_sample_child_refs_on_replace'
if source == 'cell_storage'
else 'app.crud.research_donation._migrate_donation_sample_child_refs_on_replace'
)
with patch(migrate_target), pytest.raises(ValueError, match='sampleId.*重复'):
update_protocol(
db,
source=source,
protocol_id=row_id,
body=MarketProtocolBody.model_validate(
{
'source': source,
'samples': [
{
'id': 'sample-duplicate-id',
'sampleTypeId': 'type-a',
'sampleCodeId': 'code-b',
'sampleCode': 'S-002',
},
{
'sampleId': 'sample-duplicate-id',
'sampleTypeId': 'type-a',
'sampleCodeId': 'code-c',
'sampleCode': 'S-003',
},
],
}
),
ctx=_ctx('specialist'),
user_info=_user(),
)
@pytest.mark.parametrize('source', ['cell_storage', 'research_donation'])
def test_protocol_sample_update_rejects_sample_id_from_another_protocol(source):
from app.schemas.market_mobile import MarketProtocolBody
from app.services.market_protocol import update_protocol
db = _session()
_masters(db)
owner_id = f'{source}-owner'
other_id = f'{source}-other'
creator = _cell if source == 'cell_storage' else _donation
creator(db, owner_id)
creator(db, other_id)
db.add_all(
[
_protocol_sample_row(
source,
row_id='sample-owner',
protocol_id=owner_id,
code_id='code-a',
code='S-001',
amount=10,
remark='本协议',
),
_protocol_sample_row(
source,
row_id='sample-foreign',
protocol_id=other_id,
code_id='code-b',
code='S-002',
amount=20,
remark='其他协议',
),
]
)
db.commit()
with pytest.raises(ValueError, match='样本.*不属于当前协议'):
update_protocol(
db,
source=source,
protocol_id=owner_id,
body=MarketProtocolBody.model_validate(
{
'source': source,
'samples': [
{
'id': 'sample-foreign',
'sampleTypeId': 'type-a',
'sampleCodeId': 'code-a',
'sampleCode': 'S-001',
}
],
}
),
ctx=_ctx('specialist'),
user_info=_user(),
)
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_can_rebind_same_tenant_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()
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()
if source == 'cell_storage':
party = (
db.query(BizProtocolPartyCustomer)
.filter_by(collection_protocol_id='customer-idor', del_flag=0)
.one()
)
else:
party = (
db.query(BizResearchDonationParty)
.filter_by(research_donation_id='customer-idor', del_flag=0)
.one()
)
assert party.customer_id == 'unrelated-c'
assert party.mother_name == '无关客户'
@pytest.mark.parametrize('source', ['cell_storage', 'research_donation'])
def test_protocol_update_rejects_missing_or_cross_tenant_customer_id(source):
from app.schemas.market_mobile import MarketProtocolBody
from app.services.market_protocol import update_protocol
db = _session()
_masters(db)
db.add(
BizCustomer(
id='other-tenant-c',
customer_name='外租户',
sex='1',
tenant_no='tenant-b',
create_dept_id='',
del_flag=0,
)
)
if source == 'cell_storage':
_cell(db, 'customer-cross')
db.add(
BizProtocolPartyCustomer(
id='cross-party',
collection_protocol_id='customer-cross',
mother_name='原姓名',
tenant_no='tenant-a',
create_dept_id='',
del_flag=0,
)
)
else:
_donation(db, 'customer-cross')
db.add(
BizResearchDonationParty(
id='cross-party',
research_donation_id='customer-cross',
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-cross',
body=MarketProtocolBody.model_validate(
{
'source': source,
'customer': {'customerId': 'other-tenant-c', 'motherName': '外租户'},
}
),
ctx=_ctx('specialist'),
user_info=_user(),
)
@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_allows_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()
created = 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(),
)
assert (created.get('customer') or {}).get('customerId') == 'existing-c'
def test_market_protocol_body_accepts_sample_payments():
from app.schemas.market_mobile import MarketProtocolBody
body = MarketProtocolBody.model_validate(
{
'source': 'cell_storage',
'protocolNo': 'P-PAY-1',
'customer': {'motherName': '李四'},
'samplePayments': [
{
'sampleCode': 'S-001',
'payMode': 'dj',
'contractAmount': 20000,
'lumpSumDeadline': '20261231',
'schedules': [
{
'scheduleNo': 1,
'dueAmount': 20000,
'dueDate': '20261231',
}
],
}
],
}
)
assert body.sample_payments is not None
assert len(body.sample_payments) == 1
assert body.sample_payments[0].sample_code == 'S-001'
assert body.sample_payments[0].pay_mode == 'dj'
def test_market_protocol_money_json_keeps_decimal_18_2_through_service_payload():
from app.schemas.market_mobile import MarketProtocolBody
from app.services.market_protocol import (
_market_sample_payment_payload,
_persist_sample_payments_if_any,
)
amount = '9999999999999999.99'
body = MarketProtocolBody.model_validate_json(json.dumps({
'source': 'cell_storage',
'protocolTotalAmount': amount,
'samplePayments': [{
'sampleCode': 'S-001',
'payMode': 'fq',
'contractAmount': amount,
'installmentRatios': [amount],
'schedules': [{
'scheduleNo': 1,
'dueAmount': amount,
'transfers': [{'transferAmount': amount, 'transferType': 'bank'}],
}],
}],
}))
assert body.protocol_total_amount == Decimal(amount)
assert body.sample_payments[0].contract_amount == Decimal(amount)
assert body.sample_payments[0].schedules[0].due_amount == Decimal(amount)
assert body.sample_payments[0].schedules[0].transfers[0].transfer_amount == Decimal(amount)
db = _session()
row = _cell(db, 'decimal-boundary')
row.payment_model_version = 2
sample = BizProtocolSample(
id='sample-decimal', collection_protocol_id=row.id,
sample_code='S-001', tenant_no='tenant-a', del_flag=0,
)
db.add(sample)
db.commit()
payload = _market_sample_payment_payload(
db,
row=row,
body=body,
ctx=_ctx('specialist'),
saved_samples=[sample],
)
assert payload[0]['contractAmount'] == Decimal(amount)
assert payload[0]['schedules'][0]['dueAmount'] == Decimal(amount)
with patch(
'app.services.market_protocol.psp_crud.save_sample_payments_for_protocol'
) as save:
_persist_sample_payments_if_any(
db,
row=row,
body=body,
ctx=_ctx('specialist'),
saved_samples=[sample],
)
crud_payload = save.call_args.kwargs['sample_payments']
assert crud_payload[0]['contractAmount'] == Decimal(amount)
assert crud_payload[0]['schedules'][0]['dueAmount'] == Decimal(amount)
def test_mobile_payment_response_serializes_decimal_as_fixed_string():
from app.crud.protocol_sample_payment import _transfer_to_dict
transfer = BizProtocolPaymentTransfer(
id='t-boundary',
transfer_amount=Decimal('9999999999999999.99'),
tenant_no='tenant-a',
del_flag=0,
)
response = _transfer_to_dict(_session(), transfer, 'tenant-a', money_as_string=True)
encoded = json.dumps(response)
assert response['transferAmount'] == '9999999999999999.99'
assert '1e+16' not in encoded.lower()
def test_protocol_create_persists_sample_payments():
from app.services.market_protocol import create_protocol, get_protocol
db = _session()
_masters(db)
body = _body(
samplePayments=[
{
'sampleCode': 'S-001',
'payMode': 'dj',
'contractAmount': 20000,
'lumpSumDeadline': '20261231',
'schedules': [
{
'scheduleNo': 1,
'dueAmount': 20000,
'dueDate': '20261231',
'scheduleStatus': 'open',
'transfers': [],
}
],
}
]
)
created = create_protocol(
db, body=body, ctx=_ctx('specialist'), user_info=_user()
)
protocol_id = created['basic']['id']
pays = created.get('samplePayments') or []
assert len(pays) == 1
assert pays[0]['sampleCode'] == 'S-001'
assert pays[0]['payMode'] == 'dj'
assert pays[0]['contractAmount'] == '20000.00'
assert pays[0]['schedules'][0]['dueAmount'] == '20000.00'
assert str(pays[0]['schedules'][0].get('dueDate') or '').replace('-', '')[:8] == '20261231'
detail = get_protocol(
db,
source='cell_storage',
protocol_id=protocol_id,
ctx=_ctx('specialist'),
user_info=_user(),
)
assert len(detail.get('samplePayments') or []) == 1
assert detail['samplePayments'][0]['sampleCode'] == 'S-001'
assert detail['samplePayments'][0]['contractAmount'] == '20000.00'
@pytest.mark.parametrize('scope_kind', ['specialist', 'manager'])
def test_protocol_update_rejects_foreign_sample_payment_id(scope_kind):
from app.services.market_protocol import _market_sample_payment_payload
db = _session()
_masters(db)
owner = _cell(db, 'pay-owner')
_cell(db, 'pay-foreign')
owner_sample = BizProtocolSample(
id='sample-owner', collection_protocol_id='pay-owner',
sample_type_id='type-a', sample_code='S-001', tenant_no='tenant-a',
create_dept_id='', del_flag=0,
)
db.add_all(
[
owner_sample,
BizProtocolSamplePayment(
id='payment-foreign', collection_protocol_id='pay-foreign',
protocol_sample_id='sample-foreign', sample_code='S-FOREIGN',
pay_mode='dj', contract_amount=999, tenant_no='tenant-a',
create_dept_id='', del_flag=0,
),
]
)
db.commit()
body = _body(
samples=None,
attachments=None,
samplePayments=[
{
'id': 'payment-foreign',
'sampleCode': 'S-001',
'protocolSampleId': 'sample-foreign',
'payMode': 'dj',
'contractAmount': 100,
'schedules': [
{'scheduleNo': 1, 'dueAmount': 100, 'dueDate': '20261031'}
],
}
]
)
with pytest.raises(ValueError, match='缴费方案'):
_market_sample_payment_payload(
db,
row=owner,
body=body,
ctx=_ctx(scope_kind, 'u1' if scope_kind == 'specialist' else 'manager-a'),
saved_samples=[owner_sample],
)
db.expire_all()
foreign = db.get(BizProtocolSamplePayment, 'payment-foreign')
assert foreign.collection_protocol_id == 'pay-foreign'
assert foreign.sample_code == 'S-FOREIGN'
assert float(foreign.contract_amount) == 999
def test_protocol_payment_payload_uses_saved_sample_and_preserves_submitted_transfer():
from app.crud.protocol_sample_payment import save_sample_payments_for_protocol
from app.services.market_protocol import _market_sample_payment_payload
db = _session()
_masters(db)
owner = _cell(db, 'pay-safe')
sample = BizProtocolSample(
id='sample-safe', collection_protocol_id='pay-safe',
sample_type_id='type-a', sample_code='S-SAFE', tenant_no='tenant-a',
create_dept_id='', del_flag=0,
)
payment = BizProtocolSamplePayment(
id='payment-safe', collection_protocol_id='pay-safe',
protocol_sample_id='sample-safe-old', sample_code='S-SAFE', pay_mode='dj',
contract_amount=100, tenant_no='tenant-a', create_dept_id='', del_flag=0,
)
schedule = BizProtocolPaymentSchedule(
id='schedule-safe', sample_payment_id='payment-safe',
collection_protocol_id='pay-safe', sample_code='S-SAFE', schedule_no=1,
due_amount=100, due_date='20261031', schedule_status='open',
tenant_no='tenant-a', create_dept_id='', del_flag=0,
)
transfer = BizProtocolPaymentTransfer(
id='transfer-safe', schedule_id='schedule-safe',
sample_payment_id='payment-safe', collection_protocol_id='pay-safe',
sample_code='S-SAFE', transfer_amount=20, transfer_type='bank',
transfer_attachments='payment/existing.jpg', pay_status='jfqrz',
tenant_no='tenant-a', create_dept_id='', del_flag=0,
)
db.add_all([sample, payment, schedule, transfer])
db.commit()
body = _body(
samples=None,
attachments=None,
samplePayments=[
{
'id': 'payment-safe', 'sampleCode': 'S-SAFE',
'protocolSampleId': 'foreign-sample', 'payMode': 'dj',
'contractAmount': 100,
'schedules': [
{
'id': 'foreign-schedule', 'scheduleNo': 1,
'dueAmount': 100, 'dueDate': '20261031',
'scheduleStatus': 'void',
'transfers': [
{
'id': 'transfer-safe', 'transferAmount': 999,
'transferAttachments': 'evil.jpg',
}
],
}
],
}
],
)
payload = _market_sample_payment_payload(
db,
row=owner,
body=body,
ctx=_ctx('specialist'),
saved_samples=[sample],
)
assert payload[0]['protocolSampleId'] == 'sample-safe'
assert payload[0]['schedules'][0]['scheduleStatus'] == 'open'
assert payload[0]['schedules'][0]['transfers'] == [{'id': 'transfer-safe'}]
save_sample_payments_for_protocol(
db,
collection_protocol_id='pay-safe',
tenant_no='tenant-a',
user_id='u1',
sample_payments=payload,
)
db.expire_all()
saved_transfer = db.get(BizProtocolPaymentTransfer, 'transfer-safe')
assert float(saved_transfer.transfer_amount) == 20
assert saved_transfer.transfer_attachments == 'payment/existing.jpg'
assert saved_transfer.pay_status == 'jfqrz'
assert db.get(BizProtocolSamplePayment, 'payment-safe').protocol_sample_id == 'sample-safe'
def test_save_sample_payments_serializes_decimal_installment_ratios_as_fixed_strings():
from app.crud.protocol_sample_payment import save_sample_payments_for_protocol
db = _session()
_cell(db, 'decimal-ratios')
save_sample_payments_for_protocol(
db,
collection_protocol_id='decimal-ratios',
tenant_no='tenant-a',
user_id='u1',
sample_payments=[
{
'sampleCode': 'S-DECIMAL',
'payMode': 'fq',
'contractAmount': Decimal('100.00'),
'installmentCount': 2,
'installmentRatios': [
Decimal('20.00'),
Decimal('80.00'),
Decimal('9999999999999999.99'),
],
'schedules': [
{'scheduleNo': 1, 'dueAmount': Decimal('20.00'), 'dueDate': '20261101'},
{'scheduleNo': 2, 'dueAmount': Decimal('80.00'), 'dueDate': '20261201'},
],
},
{
'sampleCode': 'S-STRING',
'payMode': 'fq',
'contractAmount': Decimal('100.00'),
'installmentCount': 1,
'installmentRatios': 'legacy-string',
'schedules': [
{'scheduleNo': 1, 'dueAmount': Decimal('100.00'), 'dueDate': '20261101'},
],
},
],
)
saved = (
db.query(BizProtocolSamplePayment)
.filter(
BizProtocolSamplePayment.collection_protocol_id == 'decimal-ratios',
BizProtocolSamplePayment.sample_code == 'S-DECIMAL',
)
.one()
)
assert json.loads(saved.installment_ratios) == [
'20.00',
'80.00',
'9999999999999999.99',
]
legacy = (
db.query(BizProtocolSamplePayment)
.filter(
BizProtocolSamplePayment.collection_protocol_id == 'decimal-ratios',
BizProtocolSamplePayment.sample_code == 'S-STRING',
)
.one()
)
assert legacy.installment_ratios == 'legacy-string'
def test_protocol_update_rejects_unscoped_freight_target():
from app.services.market_protocol import update_protocol
db = _session()
_masters(db)
_cell(db, 'freight-owner')
_cell(db, 'freight-hidden', sales='u2', manager='manager-other', region='region-b')
db.add_all(
[
BizProtocolSample(
id='sample-owner', collection_protocol_id='freight-owner',
sample_type_id='type-a', sample_code='S-001', tenant_no='tenant-a',
create_dept_id='', del_flag=0,
),
BizProtocolSample(
id='sample-hidden', collection_protocol_id='freight-hidden',
sample_type_id='type-a', sample_code='S-HIDDEN', tenant_no='tenant-a',
create_dept_id='', del_flag=0,
),
BizSampleFreight(
id='freight-hidden', freight_no='F-HIDDEN', transport_mode='zxc',
freight_amount=80, tenant_no='tenant-a', create_dept_id='', del_flag=0,
),
BizSampleFreightItem(
id='freight-item-hidden', freight_id='freight-hidden',
sample_source='cell_storage', sample_id='sample-hidden',
protocol_id='freight-hidden', sample_code='S-HIDDEN',
tenant_no='tenant-a', create_dept_id='', del_flag=0,
),
]
)
db.commit()
body = _body(
protocolNo='CELL-freight-owner',
samples=[
{
'sampleTypeId': 'type-a', 'sampleCodeId': 'code-a',
'sampleCode': 'S-001', 'unitId': 'unit-a',
'collectAmount': 10, 'freightId': 'freight-hidden',
}
]
)
with pytest.raises(HTTPException):
update_protocol(
db,
source='cell_storage',
protocol_id='freight-owner',
body=body,
ctx=_ctx('specialist'),
user_info=_user(),
)
db.expire_all()
assert db.get(BizProtocolSample, 'sample-owner').del_flag == 0
hidden_item = db.get(BizSampleFreightItem, 'freight-item-hidden')
assert hidden_item.sample_id == 'sample-hidden'
def test_protocol_body_rejects_status_finance_and_conversion_fields():
from app.schemas.market_mobile import MarketProtocolBody
forbidden = [
{'auditStatus': 'shtg'},
{'protocolStatus': 'zckc'},
{'donationStatus': 'zky'},
{'delFlag': 1},
{'refundAmount': 100},
{'financeRemark': '通过'},
{'toResearchTime': '2026-10-10'},
]
for field in forbidden:
with pytest.raises(ValidationError):
MarketProtocolBody.model_validate({'source': 'cell_storage', **field})
allowed = MarketProtocolBody.model_validate(
{'source': 'cell_storage', 'protocolStatus': 'sxz', 'donationStatus': 'dsx'}
)
assert allowed.protocol_status == 'sxz'
assert allowed.donation_status == 'dsx'
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(
('field_label', 'scope_kind', 'collection'),
[
(
'市场专员',
'manager',
{
'regionId': 'region-a',
'hospitalId': 'hospital-a',
'salesUserId': 'u2',
'regionManagerUserId': 'u1',
},
),
(
'地区经理',
'specialist',
{
'regionId': 'region-a',
'hospitalId': 'hospital-a',
'salesUserId': 'u1',
'regionManagerUserId': 'u2',
},
),
],
)
def test_protocol_create_allows_assignment_without_target_region_staff_binding(
source, field_label, scope_kind, collection
):
from app.services.market_protocol import create_protocol
db = _session()
_masters(db)
overrides = {'collection': collection, 'attachments': {}}
if source == 'research_donation':
overrides.update({'donationNo': 'D-OUTSIDER', 'protocolNo': None})
created = create_protocol(
db,
body=_body(source, **overrides),
ctx=_ctx(scope_kind),
user_info=_user(),
)
saved = created['collection']
if field_label == '市场专员':
assert saved['salesUserId'] == 'u2'
else:
assert saved['regionManagerUserId'] == 'u2'
@pytest.mark.parametrize('source', ['cell_storage', 'research_donation'])
@pytest.mark.parametrize(
('field_name', 'scope_kind'),
[('salesUserId', 'manager'), ('regionManagerUserId', 'specialist')],
)
def test_protocol_update_allows_changed_assignment_without_target_region_staff_binding(
source, field_name, scope_kind
):
from app.schemas.market_mobile import MarketProtocolBody
from app.services.market_protocol import update_protocol
db = _session()
_masters(db)
row_id = f'{source}-{field_name}-outsider'
if source == 'cell_storage':
_cell(
db,
row_id,
sales='u1',
manager='u1' if scope_kind == 'manager' else 'manager-a',
)
else:
_donation(
db,
row_id,
sales='u1',
manager='u1' if scope_kind == 'manager' else 'manager-a',
)
db.commit()
updated = update_protocol(
db,
source=source,
protocol_id=row_id,
body=MarketProtocolBody.model_validate(
{'source': source, 'collection': {field_name: 'u2'}}
),
ctx=_ctx(scope_kind),
user_info=_user(),
)
assert updated['collection'][field_name] == 'u2'
@pytest.mark.parametrize('source', ['cell_storage', 'research_donation'])
@pytest.mark.parametrize(
('historical_field', 'scope_kind'),
[('sales_user_id', 'manager'), ('region_manager_user_id', 'specialist')],
)
def test_protocol_update_keeps_unchanged_historical_assignment_without_binding(
source, historical_field, scope_kind
):
from app.schemas.market_mobile import MarketProtocolBody
from app.services.market_protocol import update_protocol
db = _session()
_masters(db)
row_id = f'{source}-{historical_field}-legacy'
kwargs = {'sales': 'u1', 'manager': 'manager-a'}
if historical_field == 'sales_user_id':
kwargs.update({'sales': 'u2', 'manager': 'u1'})
else:
kwargs['manager'] = 'u2'
if source == 'cell_storage':
_cell(db, row_id, **kwargs)
else:
_donation(db, row_id, **kwargs)
db.commit()
result = update_protocol(
db,
source=source,
protocol_id=row_id,
body=MarketProtocolBody.model_validate(
{'source': source, 'remark': '仅修改非归属字段'}
),
ctx=_ctx(scope_kind),
user_info=_user(),
)
assert result['basic']['remark'] == '仅修改非归属字段'
@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': 'u1',
},
}
),
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_expose_delete_but_no_web_only_admin_mutations():
from app.api.market_mobile import router
for scope in ('specialist', 'manager'):
prefix = f'/mini/market/{scope}/protocols/{{protocol_id}}'
routes = {
route.path: route.methods
for route in router.routes
if route.path.startswith(f'/mini/market/{scope}/protocols')
}
assert 'DELETE' in routes[prefix]
assert f'{prefix}/audit' not in routes
assert all(not any(word in path for word in ('void', 'refund', 'convert', 'to-research', 'to-storage')) for path in routes)
assert not any(
route.path.startswith(f'/mini/market/{scope}/samples')
or route.path.startswith(f'/mini/market/{scope}/sample/attachments')
for route in router.routes
)
def test_protocol_delete_rechecks_scoped_access_before_crud(monkeypatch):
from app.services.market_protocol import delete_protocol
db = _session()
_masters(db)
own = _cell(db, 'own', sales='u1')
other = _cell(db, 'other', sales='u2')
db.commit()
delete_calls = []
monkeypatch.setattr(
'app.services.market_protocol.cell_crud.soft_delete_collection_protocol',
lambda *args, **kwargs: delete_calls.append((args, kwargs)),
)
delete_protocol(
db,
source='cell_storage',
protocol_id=own.id,
ctx=_ctx('specialist'),
user_info=_user(),
)
assert delete_calls[0][0][1] == own.id
with pytest.raises(HTTPException, match='无权'):
delete_protocol(
db,
source='cell_storage',
protocol_id=other.id,
ctx=_ctx('specialist'),
user_info=_user(),
)
def test_protocol_list_contains_mobile_card_fields():
from app.services.market_protocol import list_protocols
db = _session()
_masters(db)
row = _cell(db, 'card')
row.protocol_total_amount = 20000
db.add_all([
BizProtocolPartyCustomer(
id='party-card', collection_protocol_id=row.id, mother_name='李安然',
mother_phone='13800138000', mother_id_card='340100', father_name='王先生',
newborn_sex='1', delivery_mode='sc', newborn_weight_gram=3200,
tenant_no='tenant-a', create_dept_id='', del_flag=0,
),
BizProtocolSample(
id='sample-card', collection_protocol_id=row.id, sample_type_id='type-a',
sample_code='S-CARD', tenant_no='tenant-a', create_dept_id='', del_flag=0,
),
])
db.commit()
items, total = list_protocols(db, ctx=_ctx('specialist'), user_info=_user())
assert total == 1
assert items[0]['motherName'] == '李安然'
assert items[0]['sampleCode'] == 'S-CARD'
assert items[0]['sampleTypeName'] == '样本A'
assert items[0]['regionName'] == 'A区'
assert items[0]['motherPhone'] == '13800138000'
assert items[0]['motherIdCard'] == '340100'
assert items[0]['fatherName'] == '王先生'
assert items[0]['newbornSex'] == '1'
assert items[0]['deliveryMode'] == 'sc'
assert items[0]['newbornWeightGram'] == 3200
assert items[0]['salesUserName'] == '当前用户'
by_customer, customer_total = list_protocols(
db, ctx=_ctx('specialist'), user_info=_user(), keyword='李安然'
)
by_sample, sample_total = list_protocols(
db, ctx=_ctx('specialist'), user_info=_user(), keyword='S-CARD'
)
by_phone, phone_total = list_protocols(
db, ctx=_ctx('specialist'), user_info=_user(), keyword='13800138000'
)
by_id_card, id_card_total = list_protocols(
db, ctx=_ctx('specialist'), user_info=_user(), keyword='340100'
)
assert customer_total == sample_total == phone_total == id_card_total == 1
assert {
by_customer[0]['id'],
by_sample[0]['id'],
by_phone[0]['id'],
by_id_card[0]['id'],
} == {row.id}
@pytest.mark.parametrize('source', ['cell_storage', 'research_donation'])
def test_protocol_list_and_detail_resolve_tenant_void_type_label(source):
from app.services.market_protocol import get_protocol, list_protocols
db = _session()
_masters(db)
db.add(
SysDictData(
id=f'void-dict-{source}',
dict_type='xyzflx',
dict_value='tenant-custom',
dict_label='租户自定义作废',
tenant_no='tenant-a',
create_dept_id='',
del_flag=0,
)
)
if source == 'cell_storage':
row = _cell(db, 'void-label-cell')
row.protocol_status = 'yzf'
row.void_type = 'tenant-custom'
else:
row = _donation(db, 'void-label-donation')
row.donation_status = 'yzf'
row.void_type = 'tenant-custom'
db.commit()
items, total = list_protocols(
db,
ctx=_ctx('specialist'),
user_info=_user(),
source=source,
)
detail = get_protocol(
db,
source=source,
protocol_id=row.id,
ctx=_ctx('specialist'),
user_info=_user(),
)
assert total == 1
assert items[0]['voidType'] == 'tenant-custom'
assert items[0]['voidTypeLabel'] == '租户自定义作废'
assert detail['basic']['voidType'] == 'tenant-custom'
assert detail['basic']['voidTypeLabel'] == '租户自定义作废'
def test_unknown_void_type_label_preserves_original_code():
from app.services.market_protocol import list_protocols
db = _session()
_masters(db)
row = _cell(db, 'void-label-unknown')
row.protocol_status = 'yzf'
row.void_type = 'legacy-unknown'
db.commit()
items, _ = list_protocols(
db,
ctx=_ctx('specialist'),
user_info=_user(),
source='cell_storage',
)
assert items[0]['voidType'] == 'legacy-unknown'
assert items[0]['voidTypeLabel'] == 'legacy-unknown'
def test_protocol_list_filters_status_create_time_and_sample_collect_time():
from app.services.market_protocol import list_protocols
db = _session()
_masters(db)
hit = _cell(db, 'hit', created=datetime(2026, 10, 8, 9, 0))
hit.protocol_status = 'sxz'
db.query(BizProtocolCollectionInfo).filter_by(
collection_protocol_id=hit.id
).update({'sample_collect_time': datetime(2026, 10, 9, 10, 0)})
miss_status = _cell(db, 'miss-status', created=datetime(2026, 10, 8, 9, 0))
miss_status.protocol_status = 'dsx'
db.query(BizProtocolCollectionInfo).filter_by(
collection_protocol_id=miss_status.id
).update({'sample_collect_time': datetime(2026, 10, 9, 10, 0)})
miss_date = _cell(db, 'miss-date', created=datetime(2026, 9, 1, 9, 0))
miss_date.protocol_status = 'sxz'
db.query(BizProtocolCollectionInfo).filter_by(
collection_protocol_id=miss_date.id
).update({'sample_collect_time': datetime(2026, 9, 2, 10, 0)})
db.commit()
items, total = list_protocols(
db,
ctx=_ctx('specialist'),
user_info=_user(),
source='cell_storage',
protocol_status='sxz',
create_time_start='2026-10-01',
create_time_end='2026-10-13',
sample_collect_time_start='2026-10-01',
sample_collect_time_end='2026-10-13',
)
assert total == 1
assert [item['id'] for item in items] == ['hit']
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
def test_cell_protocol_detail_includes_storage_insurance_and_audits():
from app.services.market_protocol import get_protocol
db = _session()
_masters(db)
_cell(db, 'cell-enrich')
db.add(
BizProtocolStorageCert(
id='sc-1',
collection_protocol_id='cell-enrich',
make_status='1',
make_date='20261001',
tenant_no='tenant-a',
create_dept_id='',
del_flag=0,
)
)
db.add(
BizProtocolFullmoonTodo(
id='fm-1',
collection_protocol_id='cell-enrich',
insurance_company_name='测试保险',
insurance_policy_no='POL-1',
tenant_no='tenant-a',
create_dept_id='',
del_flag=0,
)
)
db.add(
BizProtocolAudit(
id='au-1',
collection_protocol_id='cell-enrich',
audit_status='shtg',
audit_remark='通过',
auditor_user_id='u1',
tenant_no='tenant-a',
create_dept_id='',
del_flag=0,
)
)
db.commit()
detail = get_protocol(
db,
source='cell_storage',
protocol_id='cell-enrich',
ctx=_ctx('specialist'),
user_info=_user(),
)
assert detail['storageCert']['makeStatus'] == '1'
assert detail['fullmoonTodo']['insurancePolicyNo'] == 'POL-1'
assert len(detail['audits']) == 1
assert detail['audits'][0]['auditStatus'] == 'shtg'
def test_cell_protocol_detail_hides_cross_tenant_handler_and_keeps_deleted_name():
from app.services.market_protocol import get_protocol
db = _session()
_masters(db)
_cell(db, 'cell-handler-isolation')
db.add_all(
[
SysUser(
id='cross-tenant-handler',
user_no='cross-handler',
nick_name='异租户经办人',
status='1',
tenant_no='tenant-b',
create_dept_id='',
del_flag=0,
),
SysUser(
id='deleted-handler',
user_no='deleted-handler',
nick_name='已删除经办人',
status='1',
tenant_no='tenant-a',
create_dept_id='',
del_flag=1,
),
BizProtocolStorageCert(
id='sc-handler-isolation',
collection_protocol_id='cell-handler-isolation',
handler_user_id='cross-tenant-handler',
tenant_no='tenant-a',
create_dept_id='',
del_flag=0,
),
BizProtocolFullmoonTodo(
id='fm-handler-isolation',
collection_protocol_id='cell-handler-isolation',
handler_user_id='deleted-handler',
tenant_no='tenant-a',
create_dept_id='',
del_flag=0,
),
]
)
db.commit()
detail = get_protocol(
db,
source='cell_storage',
protocol_id='cell-handler-isolation',
ctx=_ctx('specialist'),
user_info=_user(),
)
assert detail['storageCert']['handlerUserName'] == ''
assert detail['fullmoonTodo']['handlerUserName'] == '已删除经办人'
def test_cell_protocol_detail_includes_presign_deposit_hint_and_transfers():
from app.services.market_protocol import get_protocol
db = _session()
_masters(db)
row = _cell(db, 'cell-deposit')
row.source_presign_order_id = 'presign-d1'
db.add(
BizPresignOrder(
id='presign-d1',
protocol_no='PS-D1',
mother_name='张敏',
order_status='yzh',
tenant_no='tenant-a',
create_dept_id='',
del_flag=0,
)
)
db.add(
BizPresignTransfer(
id='tr-d1',
presign_order_id='presign-d1',
transfer_amount=1000,
transfer_type='wx',
pay_status='yjf',
tenant_no='tenant-a',
create_dept_id='',
del_flag=0,
)
)
db.add(
SysDictData(
id='dict-deposit-wx',
dict_label='微信支付',
dict_value='wx',
dict_type='zzfs',
tenant_no='tenant-a',
create_dept_id='',
del_flag=0,
)
)
db.commit()
detail = get_protocol(
db,
source='cell_storage',
protocol_id='cell-deposit',
ctx=_ctx('specialist'),
user_info=_user(),
)
assert detail['presignDepositHint']['confirmedDepositAmount'] == '1000.00'
assert detail['presignDepositHint']['confirmedTransferCount'] == 1
assert len(detail['depositTransfers']) == 1
assert detail['depositTransfers'][0]['id'] == 'tr-d1'
assert detail['depositTransfers'][0]['transferAmount'] == '1000.00'
assert detail['depositTransfers'][0]['transferTypeLabel'] == '微信支付'
def test_protocol_customer_search_empty_keyword_returns_recent_list():
"""空关键字仍返回近期客户列表,供移动端「选档案」弹窗默认展示。"""
from app.services.market_support import search_customers_for_protocol
db = _session()
db.add(
BizCustomer(
id='c-empty-kw',
customer_name='可见客户',
sex='1',
tenant_no='tenant-a',
create_dept_id='',
del_flag=0,
)
)
db.commit()
data = search_customers_for_protocol(
db, ctx=_ctx('specialist'), user_info=_user(), keyword=' '
)
assert [item['id'] for item in data['list']] == ['c-empty-kw']
assert data['total'] == 1
def test_protocol_customer_search_matches_name_phone_and_isolates_tenant():
from app.services.market_support import search_customers_for_protocol
db = _session()
db.add_all(
[
BizCustomer(
id='c-hit',
customer_name='张敏',
phone_no='13800001111',
id_card_no='ID-HIT',
sex='1',
tenant_no='tenant-a',
create_dept_id='',
del_flag=0,
),
BizCustomer(
id='c-other-tenant',
customer_name='张敏',
phone_no='13800001111',
sex='1',
tenant_no='tenant-b',
create_dept_id='',
del_flag=0,
),
BizCustomer(
id='c-miss',
customer_name='李四',
phone_no='13900002222',
sex='1',
tenant_no='tenant-a',
create_dept_id='',
del_flag=0,
),
]
)
db.commit()
by_name = search_customers_for_protocol(
db, ctx=_ctx('specialist'), user_info=_user(), keyword='张敏'
)
assert [item['id'] for item in by_name['list']] == ['c-hit']
assert by_name['list'][0]['phoneNo'] == '13800001111'
assert by_name['list'][0]['idCardNo'] == 'ID-HIT'
by_phone = search_customers_for_protocol(
db, ctx=_ctx('manager'), user_info=_user(), keyword='1380000'
)
assert [item['id'] for item in by_phone['list']] == ['c-hit']
def test_protocol_routes_expose_customers_search():
from app.api.market_mobile import router
for scope in ('specialist', 'manager'):
paths = {
route.path
for route in router.routes
if route.path.startswith(f'/mini/market/{scope}/protocols')
}
assert f'/mini/market/{scope}/protocols/customers' in paths
def test_protocol_sample_body_accepts_status_and_freight_fields():
from app.schemas.market_mobile import MarketProtocolBody
body = MarketProtocolBody.model_validate(
{
'source': 'cell_storage',
'samples': [
{
'sampleTypeId': 'type-a',
'sampleCode': 'S-001',
'sampleStatus': 'ycj',
'etaTime': '2026-10-13 12:00:00',
'freightId': 'fr-1',
'freightDraft': {
'clientKey': 'fd_1',
'transportMode': 'zxc',
'freightAmount': 120,
},
}
],
}
)
sample = body.samples[0]
assert sample.sample_status == 'ycj'
assert sample.freight_id == 'fr-1'
assert sample.freight_draft.client_key == 'fd_1'
assert sample.freight_draft.transport_mode == 'zxc'
assert sample.freight_draft.freight_amount == 120
def test_cell_protocol_detail_reuses_web_sample_lifecycle_bundle(monkeypatch):
from app.services.market_protocol import get_protocol
db = _session()
_masters(db)
_cell(db, 'cell-lifecycle')
db.add(
BizProtocolSample(
id='ps-lifecycle',
collection_protocol_id='cell-lifecycle',
sample_type_id='type-a',
sample_code='S-LIFE-1',
sample_status='yrk',
tenant_no='tenant-a',
create_dept_id='',
del_flag=0,
)
)
db.commit()
lifecycle_sample = {
'id': 'ps-lifecycle',
'sampleCode': 'S-LIFE-1',
'sampleStatus': 'yrk',
'sampleReception': {'receptionTime': '2026-10-01', 'receiveAttachmentFiles': [{'fileUrl': 'receive.pdf'}]},
'productionReception': {'processTime': '2026-10-02'},
'qcInfo': {'workOrderNo': 'QC-1'},
'processQcInfoList': [{'workOrderNo': 'PQC-1'}],
'productionInfo': {'workOrderNo': 'PR-1'},
'storageReleaseInfo': {'storageStatus': 'yrk'},
}
monkeypatch.setattr(
'app.services.market_protocol.cell_crud.get_protocol_detail_bundle',
lambda *args, **kwargs: {'samples': [lifecycle_sample]},
)
detail = get_protocol(
db,
source='cell_storage',
protocol_id='cell-lifecycle',
ctx=_ctx('specialist'),
user_info=_user(),
)
sample = detail['samples'][0]
for key in [
'sampleReception', 'productionReception', 'qcInfo',
'processQcInfoList', 'productionInfo', 'storageReleaseInfo',
]:
assert sample[key] == lifecycle_sample[key]
def test_cell_v1_detail_reuses_web_payment_details_and_transfer_type_dict(monkeypatch):
from app.services.market_protocol import get_protocol
db = _session()
_masters(db)
row = _cell(db, 'cell-v1-payment')
row.payment_model_version = 1
db.add(
SysDictData(
id='dict-wx',
dict_label='微信支付',
dict_value='wx',
dict_type='zzfs',
tenant_no='tenant-a',
create_dept_id='',
del_flag=0,
)
)
db.commit()
monkeypatch.setattr(
'app.services.market_protocol.cell_crud.get_protocol_detail_bundle',
lambda *args, **kwargs: {
'samples': [],
'paymentDetails': [{
'id': 'pd-1',
'installmentNo': 1,
'dueAmount': 1000,
'transferAmount': 800,
'installmentDueDate': '20261031',
'transferTime': '2026-10-13 12:00:00',
'transferType': 'wx',
'transferAttachments': 'proof.pdf',
'payStatus': 'yjf',
'financeRemark': '已确认',
}],
},
)
detail = get_protocol(
db,
source='cell_storage',
protocol_id='cell-v1-payment',
ctx=_ctx('specialist'),
user_info=_user(),
)
assert detail['paymentModelVersion'] == 1
assert detail['samplePayments'] == []
assert detail['paymentDetails'][0]['installmentNo'] == 1
assert detail['paymentDetails'][0]['transferTypeLabel'] == '微信支付'
def test_cell_v2_detail_labels_all_transfer_types_with_unknown_fallback(monkeypatch):
from app.services.market_protocol import get_protocol
db = _session()
_masters(db)
row = _cell(db, 'cell-v2-labels')
row.payment_model_version = 2
db.add(
SysDictData(
id='dict-v2-zfb',
dict_label='支付宝',
dict_value='zfb',
dict_type='zzfs',
tenant_no='tenant-a',
create_dept_id='',
del_flag=0,
)
)
db.commit()
monkeypatch.setattr(
'app.services.market_protocol.cell_crud.get_protocol_detail_bundle',
lambda *args, **kwargs: {'samples': [], 'paymentDetails': []},
)
monkeypatch.setattr(
'app.services.market_protocol.psp_crud.get_sample_payments_bundle',
lambda *args, **kwargs: [{
'id': 'sp-1',
'schedules': [{
'id': 'sch-1',
'transfers': [
{'id': 'tr-1', 'transferType': 'zfb'},
{'id': 'tr-2', 'transferType': 'future-code'},
],
}],
}],
)
detail = get_protocol(
db,
source='cell_storage',
protocol_id='cell-v2-labels',
ctx=_ctx('specialist'),
user_info=_user(),
)
transfers = detail['samplePayments'][0]['schedules'][0]['transfers']
assert transfers[0]['transferTypeLabel'] == '支付宝'
assert transfers[1]['transferTypeLabel'] == 'future-code'
def test_donation_detail_reuses_web_bundle_lifecycle_audits_and_convert_aliases(monkeypatch):
from app.services.market_protocol import get_protocol
db = _session()
_masters(db)
_donation(db, 'don-web-detail')
db.commit()
lifecycle_sample = {
'id': 'ds-1',
'sampleCode': 'D-S-1',
'sampleReception': {'receptionTime': '2026-10-01'},
'productionReception': {'processTime': '2026-10-02'},
'qcInfo': {'workOrderNo': 'D-QC-1'},
'processQcInfoList': [{'workOrderNo': 'D-PQC-1'}],
'productionInfo': {'workOrderNo': 'D-PR-1'},
'storageReleaseInfo': {'storageStatus': 'yrk'},
}
monkeypatch.setattr(
'app.services.market_protocol.donation_crud.get_donation_detail',
lambda *args, **kwargs: {
'donation': {
'id': 'don-web-detail',
'donationNo': 'DON-WEB',
'donationStatus': 'zckc',
'toStorageTime': '2026-10-10 10:00:00',
'toStorageDesc': '已转细胞存储',
'consentFormUrls': 'consent.pdf',
},
'party': {'motherName': '张敏'},
'collection': {'regionName': 'A区'},
'samples': [lifecycle_sample],
},
)
monkeypatch.setattr(
'app.services.market_protocol.donation_crud.list_audits',
lambda *args, **kwargs: {
'list': [{'id': 'da-1', 'auditStatus': 'shtg', 'auditRemark': '通过'}]
},
)
detail = get_protocol(
db,
source='research_donation',
protocol_id='don-web-detail',
ctx=_ctx('specialist'),
user_info=_user(),
)
assert detail['samples'][0] == lifecycle_sample
assert detail['audits'][0]['id'] == 'da-1'
assert detail['basic']['convertTime'] == '2026-10-10 10:00:00'
assert detail['basic']['convertDesc'] == '已转细胞存储'
def test_cell_protocol_detail_attaches_sample_freight():
from app.services.market_protocol import get_protocol
db = _session()
_masters(db)
_cell(db, 'cell-fr')
db.add(
BizProtocolSample(
id='ps-fr',
collection_protocol_id='cell-fr',
sample_type_id='type-a',
sample_code='S-FR-1',
sample_status='ycj',
tenant_no='tenant-a',
create_dept_id='',
del_flag=0,
)
)
db.add(
SysDictData(
id='dict-zxc',
dict_label='专线车',
dict_value='zxc',
dict_type='sample_transport_mode',
tenant_no='tenant-a',
create_dept_id='',
del_flag=0,
)
)
db.add(
BizSampleFreight(
id='fr-1',
freight_no='F-001',
transport_mode='zxc',
freight_amount=88,
tenant_no='tenant-a',
create_dept_id='',
del_flag=0,
)
)
db.add(
BizSampleFreightItem(
id='fri-1',
freight_id='fr-1',
sample_source='cell_storage',
sample_id='ps-fr',
protocol_id='cell-fr',
sample_code='S-FR-1',
tenant_no='tenant-a',
create_dept_id='',
del_flag=0,
)
)
db.commit()
detail = get_protocol(
db,
source='cell_storage',
protocol_id='cell-fr',
ctx=_ctx('specialist'),
user_info=_user(),
)
sample = detail['samples'][0]
assert sample['freightId'] == 'fr-1'
assert sample['freight']['transportMode'] == 'zxc'
assert sample['freight']['freightAmount'] == '88.00'
def test_protocol_update_replaces_freight_item_without_duplicate_active_links():
from app.services.market_protocol import update_protocol
db = _session()
_masters(db)
_cell(db, 'cell-edit')
_cell(db, 'cell-sibling')
db.add_all(
[
BizProtocolSample(
id='sample-old', collection_protocol_id='cell-edit',
sample_type_id='type-a', sample_code_id='code-a', sample_code='S-001',
tenant_no='tenant-a', create_dept_id='', del_flag=0,
),
BizProtocolSample(
id='sample-sibling', collection_protocol_id='cell-sibling',
sample_type_id='type-a', sample_code_id='code-b', sample_code='S-002',
tenant_no='tenant-a', create_dept_id='', del_flag=0,
),
BizSampleFreight(
id='freight-shared', freight_no='F-SHARED', transport_mode='zxc',
freight_amount=88, tenant_no='tenant-a', create_dept_id='', del_flag=0,
),
BizSampleFreightItem(
id='item-old', freight_id='freight-shared', sample_source='cell_storage',
sample_id='sample-old', protocol_id='cell-edit', sample_code='S-001',
tenant_no='tenant-a', create_dept_id='', del_flag=0,
),
BizSampleFreightItem(
id='item-sibling', freight_id='freight-shared', sample_source='cell_storage',
sample_id='sample-sibling', protocol_id='cell-sibling', sample_code='S-002',
tenant_no='tenant-a', create_dept_id='', del_flag=0,
),
]
)
db.commit()
body = _body(
protocolNo='CELL-cell-edit',
samples=[{
'sampleTypeId': 'type-a', 'sampleCodeId': 'code-a',
'sampleCode': 'S-001', 'unitId': 'unit-a',
'collectAmount': 10, 'freightId': 'freight-shared',
}],
)
first = update_protocol(
db, source='cell_storage', protocol_id='cell-edit', body=body,
ctx=_ctx('specialist'), user_info=_user(),
)
second = update_protocol(
db, source='cell_storage', protocol_id='cell-edit', body=body,
ctx=_ctx('specialist'), user_info=_user(),
)
active_sample = db.query(BizProtocolSample).filter_by(
collection_protocol_id='cell-edit', tenant_no='tenant-a', del_flag=0,
).one()
active_items = db.query(BizSampleFreightItem).filter_by(
freight_id='freight-shared', tenant_no='tenant-a', del_flag=0,
).all()
assert {item.sample_id for item in active_items} == {
active_sample.id, 'sample-sibling'
}
assert db.get(BizSampleFreightItem, 'item-old').del_flag == 1
assert first['samples'][0]['id'] != second['samples'][0]['id']
assert second['samples'][0]['freight']['linkedSampleCount'] == 2
def test_protocol_update_cancels_then_relinks_freight_without_touching_other_samples():
from app.services.market_protocol import update_protocol
db = _session()
_masters(db)
_cell(db, 'cell-move')
_cell(db, 'cell-other-a')
_cell(db, 'cell-other-b')
samples = [
BizProtocolSample(
id='sample-move', collection_protocol_id='cell-move',
sample_type_id='type-a', sample_code_id='code-a', sample_code='S-001',
tenant_no='tenant-a', create_dept_id='', del_flag=0,
),
BizProtocolSample(
id='sample-other-a', collection_protocol_id='cell-other-a',
sample_type_id='type-a', sample_code_id='code-b', sample_code='S-002',
tenant_no='tenant-a', create_dept_id='', del_flag=0,
),
BizProtocolSample(
id='sample-other-b', collection_protocol_id='cell-other-b',
sample_type_id='type-a', sample_code_id='code-c', sample_code='S-003',
tenant_no='tenant-a', create_dept_id='', del_flag=0,
),
]
freights = [
BizSampleFreight(
id='freight-a', freight_no='F-A', transport_mode='zxc', freight_amount=10,
tenant_no='tenant-a', create_dept_id='', del_flag=0,
),
BizSampleFreight(
id='freight-b', freight_no='F-B', transport_mode='zxc', freight_amount=20,
tenant_no='tenant-a', create_dept_id='', del_flag=0,
),
]
items = [
BizSampleFreightItem(
id='item-move', freight_id='freight-a', sample_source='cell_storage',
sample_id='sample-move', protocol_id='cell-move', sample_code='S-001',
tenant_no='tenant-a', create_dept_id='', del_flag=0,
),
BizSampleFreightItem(
id='item-other-a', freight_id='freight-a', sample_source='cell_storage',
sample_id='sample-other-a', protocol_id='cell-other-a', sample_code='S-002',
tenant_no='tenant-a', create_dept_id='', del_flag=0,
),
BizSampleFreightItem(
id='item-other-b', freight_id='freight-b', sample_source='cell_storage',
sample_id='sample-other-b', protocol_id='cell-other-b', sample_code='S-003',
tenant_no='tenant-a', create_dept_id='', del_flag=0,
),
]
role = SysRole(
id='role-all', role_name='市场人员', data_scope='1', status='1',
tenant_no='tenant-a', create_dept_id='', del_flag=0,
)
user_role = SysUserRole(
id='user-role-all', user_id='u1', role_id='role-all',
tenant_no='tenant-a', create_dept_id='', del_flag=0,
)
db.add_all([*samples, *freights, *items, role, user_role])
db.commit()
base_sample = {
'sampleTypeId': 'type-a', 'sampleCodeId': 'code-a',
'sampleCode': 'S-001', 'unitId': 'unit-a', 'collectAmount': 10,
}
cancelled = update_protocol(
db, source='cell_storage', protocol_id='cell-move',
body=_body(protocolNo='CELL-cell-move', samples=[base_sample]),
ctx=_ctx('specialist'), user_info=_user(),
)
assert cancelled['samples'][0]['freightId'] == ''
assert db.get(BizSampleFreightItem, 'item-move').del_flag == 1
assert db.get(BizSampleFreightItem, 'item-other-a').del_flag == 0
relinked = update_protocol(
db, source='cell_storage', protocol_id='cell-move',
body=_body(
protocolNo='CELL-cell-move',
samples=[{**base_sample, 'freightId': 'freight-b'}],
),
ctx=_ctx('specialist'), user_info=_user(),
)
current_id = relinked['samples'][0]['id']
assert relinked['samples'][0]['freight']['linkedSampleCount'] == 2
assert {
item.sample_id
for item in db.query(BizSampleFreightItem).filter_by(
freight_id='freight-b', tenant_no='tenant-a', del_flag=0,
).all()
} == {'sample-other-b', current_id}
assert db.get(BizSampleFreightItem, 'item-other-a').del_flag == 0
def test_protocol_update_freight_reconciliation_rolls_back_on_later_failure(monkeypatch):
from app.services.market_protocol import update_protocol
db = _session()
_masters(db)
_cell(db, 'cell-rollback')
db.add_all(
[
BizProtocolSample(
id='sample-rollback', collection_protocol_id='cell-rollback',
sample_type_id='type-a', sample_code_id='code-a', sample_code='S-001',
tenant_no='tenant-a', create_dept_id='', del_flag=0,
),
BizSampleFreight(
id='freight-rollback', freight_no='F-ROLLBACK', transport_mode='zxc',
freight_amount=88, tenant_no='tenant-a', create_dept_id='', del_flag=0,
),
BizSampleFreightItem(
id='item-rollback', freight_id='freight-rollback',
sample_source='cell_storage', sample_id='sample-rollback',
protocol_id='cell-rollback', sample_code='S-001',
tenant_no='tenant-a', create_dept_id='', del_flag=0,
),
]
)
db.commit()
monkeypatch.setattr(
'app.services.market_protocol._persist_sample_payments_if_any',
lambda *args, **kwargs: (_ for _ in ()).throw(RuntimeError('later failure')),
)
with pytest.raises(RuntimeError, match='later failure'):
update_protocol(
db, source='cell_storage', protocol_id='cell-rollback',
body=_body(
protocolNo='CELL-cell-rollback',
samples=[{
'sampleTypeId': 'type-a', 'sampleCodeId': 'code-a',
'sampleCode': 'S-001', 'unitId': 'unit-a',
'collectAmount': 10, 'freightId': 'freight-rollback',
}],
),
ctx=_ctx('specialist'), user_info=_user(),
)
db.expire_all()
assert db.get(BizProtocolSample, 'sample-rollback').del_flag == 0
assert db.get(BizSampleFreightItem, 'item-rollback').del_flag == 0
assert db.query(BizProtocolSample).filter_by(
collection_protocol_id='cell-rollback', tenant_no='tenant-a', del_flag=0,
).count() == 1
assert db.query(BizSampleFreightItem).filter_by(
protocol_id='cell-rollback', tenant_no='tenant-a', del_flag=0,
).count() == 1