"""样本编码全链路纠错服务测试。""" from datetime import datetime from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from app.models.base import Base from app.models.models import ( BizCellProduct, BizProductionTaskInput, BizProductionWorkOrder, BizProtocolPaymentSchedule, BizProtocolPaymentTransfer, BizProtocolSample, BizProtocolSamplePayment, BizProtocolSampleTransport, BizQcTask, BizQcWorkOrder, BizSampleCode, BizSampleCodeCorrectionLog, BizSampleCustomerService, BizSampleFreightItem, BizSampleInsurance, BizSampleProductionReception, BizSampleReception, BizSampleStorageCert, BizSampleStorageCertQc, BizStorageReleaseReview, ) from app.models.research_donation import ( BizResearchDonationSample, BizResearchDonationSampleTransport, ) from app.services import sample_code_correction as svc def _session(): engine = create_engine("sqlite:///:memory:") Base.metadata.create_all( engine, tables=[ BizProtocolSample.__table__, BizResearchDonationSample.__table__, BizSampleCode.__table__, BizProtocolSamplePayment.__table__, BizProtocolPaymentSchedule.__table__, BizProtocolPaymentTransfer.__table__, BizProtocolSampleTransport.__table__, BizResearchDonationSampleTransport.__table__, BizSampleFreightItem.__table__, BizSampleReception.__table__, BizQcWorkOrder.__table__, BizQcTask.__table__, BizStorageReleaseReview.__table__, BizSampleStorageCertQc.__table__, BizSampleProductionReception.__table__, BizProductionWorkOrder.__table__, BizProductionTaskInput.__table__, BizSampleStorageCert.__table__, BizSampleInsurance.__table__, BizSampleCustomerService.__table__, BizSampleCodeCorrectionLog.__table__, BizCellProduct.__table__, ], ) return sessionmaker(bind=engine)() def _add_protocol_sample(db, *, code="OLD-001", sample_code_id=None, sid="ps-1"): now = datetime.now() row = BizProtocolSample( id=sid, collection_protocol_id="cp-1", sample_code_id=sample_code_id, sample_code=code, tenant_no="tenant-a", create_dept_id="", create_time=now, update_time=now, del_flag=0, ) db.add(row) return row def test_preview_and_apply_cascades_downstream(): db = _session() now = datetime.now() pool = BizSampleCode( id="pool-old", sample_code="OLD-001", issued=1, tenant_no="tenant-a", create_dept_id="", create_time=now, update_time=now, del_flag=0, ) db.add(pool) _add_protocol_sample(db, code="OLD-001", sample_code_id="pool-old") db.add( BizSampleReception( id="recv-1", protocol_sample_id="ps-1", collection_protocol_id="cp-1", reception_time=now, accept_conclusion="js", sample_code="OLD-001", reception_source="cell_storage", tenant_no="tenant-a", create_dept_id="", create_time=now, update_time=now, del_flag=0, ) ) db.add( BizQcWorkOrder( id="qc-1", work_order_no="QC-001", sample_code="OLD-001", reception_source="cell_storage", tenant_no="tenant-a", create_dept_id="", create_time=now, update_time=now, del_flag=0, ) ) db.add( BizQcTask( id="qct-1", work_order_id="qc-1", task_no="QCT-001", task_type="sysz", source_line_id="line-1", item_code="ITEM", item_name="项目", sample_code="OLD-001", tenant_no="tenant-a", create_dept_id="", create_time=now, update_time=now, del_flag=0, ) ) db.add( BizProductionWorkOrder( id="pwo-1", work_order_no="PWO-001", sample_code="OLD-001", reception_source="cell_storage", tenant_no="tenant-a", create_dept_id="", create_time=now, update_time=now, del_flag=0, ) ) db.add( BizProductionTaskInput( id="pti-1", task_id="pt-1", input_type="sample", input_code="OLD-001", tenant_no="tenant-a", create_dept_id="", create_time=now, update_time=now, del_flag=0, ) ) db.add( BizSampleStorageCert( id="cert-1", sample_code="OLD-001", tenant_no="tenant-a", create_dept_id="", create_time=now, update_time=now, del_flag=0, ) ) db.commit() preview = svc.preview_correction( db, tenant_no="tenant-a", reception_source="cell_storage", old_sample_code="OLD-001", new_sample_code="NEW-001", ) assert preview["ok"] is True assert preview["totalAffectedRows"] >= 5 tables = {i["table"] for i in preview["impacts"]} assert "biz_protocol_sample" in tables assert "biz_sample_reception" in tables assert "biz_qc_work_order" in tables assert "biz_production_task_input" in tables result = svc.apply_correction( db, tenant_no="tenant-a", reception_source="cell_storage", old_sample_code="OLD-001", new_sample_code="NEW-001", reason="填错编码", user_id="u1", ) assert result["ok"] is True ps = ( db.query(BizProtocolSample) .filter(BizProtocolSample.id == "ps-1") .one() ) assert ps.sample_code == "NEW-001" pool_row = db.query(BizSampleCode).filter(BizSampleCode.id == "pool-old").one() assert pool_row.sample_code == "NEW-001" assert ( db.query(BizSampleReception) .filter(BizSampleReception.id == "recv-1") .one() .sample_code == "NEW-001" ) assert ( db.query(BizQcTask).filter(BizQcTask.id == "qct-1").one().sample_code == "NEW-001" ) assert ( db.query(BizProductionTaskInput) .filter(BizProductionTaskInput.id == "pti-1") .one() .input_code == "NEW-001" ) assert ( db.query(BizSampleStorageCert) .filter(BizSampleStorageCert.id == "cert-1") .one() .sample_code == "NEW-001" ) log = db.query(BizSampleCodeCorrectionLog).one() assert log.old_sample_code == "OLD-001" assert log.new_sample_code == "NEW-001" assert log.reason == "填错编码" def test_preview_blocks_when_new_code_occupied(): db = _session() _add_protocol_sample(db, code="OLD-001", sid="ps-1") _add_protocol_sample(db, code="NEW-001", sid="ps-2") db.commit() preview = svc.preview_correction( db, tenant_no="tenant-a", reception_source="cell_storage", old_sample_code="OLD-001", new_sample_code="NEW-001", ) assert preview["ok"] is False assert any("占用" in b for b in preview["blockers"]) def test_donation_source_updates_donation_sample(): db = _session() now = datetime.now() db.add( BizResearchDonationSample( id="ds-1", research_donation_id="rd-1", sample_code="JZ2026010001", tenant_no="tenant-a", create_dept_id="", create_time=now, update_time=now, del_flag=0, ) ) db.add( BizSampleReception( id="recv-d", protocol_sample_id="", collection_protocol_id="", donation_sample_id="ds-1", research_donation_id="rd-1", reception_time=now, accept_conclusion="js", sample_code="JZ2026010001", reception_source="research_donation", tenant_no="tenant-a", create_dept_id="", create_time=now, update_time=now, del_flag=0, ) ) db.commit() result = svc.apply_correction( db, tenant_no="tenant-a", reception_source="research_donation", old_sample_code="JZ2026010001", new_sample_code="JZ2026010099", reason="捐赠码纠错", user_id="u1", ) assert result["ok"] is True assert ( db.query(BizResearchDonationSample) .filter(BizResearchDonationSample.id == "ds-1") .one() .sample_code == "JZ2026010099" ) assert ( db.query(BizSampleReception) .filter(BizSampleReception.id == "recv-d") .one() .sample_code == "JZ2026010099" )