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