zkxw/backend/tests/test_sample_code_correction.py

"""样本编码全链路纠错服务测试。"""
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"
    )