zkxw/backend/tests/test_replace_step_outputs.py

"""保存步骤产出物时保留原行 ID，避免质检工单等外键被整表替换打断。"""

import unittest
from datetime import datetime

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from app.crud.production_task import replace_step_outputs
from app.models.models import BizProductionTaskStepOutput


def _session():
    engine = create_engine("sqlite://")
    BizProductionTaskStepOutput.__table__.create(bind=engine)
    return sessionmaker(bind=engine)()


def _output(**kwargs):
    now = datetime(2026, 9, 2, 9, 1, 8)
    data = dict(
        step_id="step-1",
        tenant_no="tenant-a",
        output_type_id="type-valid",
        output_name="细胞悬液",
        output_barcode="NK-UBMC202603009",
        create_dept_id="",
        create_time=now,
        update_time=now,
        del_flag=0,
    )
    data.update(kwargs)
    return BizProductionTaskStepOutput(**data)


class TestReplaceStepOutputsKeepsIdentity(unittest.TestCase):
    def test_existing_row_keeps_id_when_fields_change(self):
        db = _session()
        db.add(_output(id="out-keep"))
        db.commit()

        saved = replace_step_outputs(
            db,
            "step-1",
            "tenant-a",
            "user-1",
            [
                {
                    "id": "out-keep",
                    "output_type_id": "type-valid",
                    "output_name": "细胞悬液",
                    "output_barcode": "NK-UBMC202603009-NEW",
                    "remark": "",
                }
            ],
        )

        self.assertEqual(saved[0]["id"], "out-keep")
        self.assertEqual(saved[0]["outputBarcode"], "NK-UBMC202603009-NEW")
        alive = (
            db.query(BizProductionTaskStepOutput)
            .filter(BizProductionTaskStepOutput.del_flag == 0)
            .all()
        )
        self.assertEqual([row.id for row in alive], ["out-keep"])

    def test_insert_new_update_old_and_soft_delete_only_removed(self):
        db = _session()
        db.add(_output(id="out-keep", output_name="细胞悬液", output_barcode="NK-OLD"))
        db.add(_output(id="out-drop", output_name="细胞上清", output_barcode="NK-DROP"))
        db.commit()

        saved = replace_step_outputs(
            db,
            "step-1",
            "tenant-a",
            "user-1",
            [
                {
                    "id": "out-keep",
                    "output_type_id": "type-valid",
                    "output_name": "细胞悬液",
                    "output_barcode": "NK-UPDATED",
                    "remark": "改条码",
                },
                {
                    "output_type_id": "type-valid",
                    "output_name": "废液",
                    "output_barcode": "NK-NEW",
                    "remark": "",
                },
            ],
        )

        self.assertEqual(saved[0]["id"], "out-keep")
        self.assertEqual(saved[0]["outputBarcode"], "NK-UPDATED")
        self.assertTrue(saved[1]["id"])
        self.assertNotEqual(saved[1]["id"], "out-keep")
        self.assertEqual(saved[1]["outputName"], "废液")

        rows = {row.id: row for row in db.query(BizProductionTaskStepOutput).all()}
        self.assertEqual(rows["out-keep"].del_flag, 0)
        self.assertEqual(rows["out-keep"].output_barcode, "NK-UPDATED")
        self.assertEqual(rows["out-drop"].del_flag, 1)
        self.assertEqual(rows[saved[1]["id"]].del_flag, 0)
        self.assertEqual(rows[saved[1]["id"]].output_barcode, "NK-NEW")

    def test_removed_row_is_soft_deleted_and_kept_row_stays(self):
        db = _session()
        db.add(_output(id="out-keep", output_name="细胞悬液"))
        db.add(_output(id="out-drop", output_name="细胞上清"))
        db.commit()

        replace_step_outputs(
            db,
            "step-1",
            "tenant-a",
            "user-1",
            [
                {
                    "id": "out-keep",
                    "output_type_id": "type-valid",
                    "output_name": "细胞悬液",
                    "output_barcode": "NK-UBMC202603009",
                    "remark": "",
                }
            ],
        )

        rows = {row.id: row for row in db.query(BizProductionTaskStepOutput).all()}
        self.assertEqual(rows["out-keep"].del_flag, 0)
        self.assertEqual(rows["out-drop"].del_flag, 1)

    def test_id_from_another_step_is_rejected(self):
        db = _session()
        db.add(_output(id="out-other", step_id="step-2"))
        db.commit()

        with self.assertRaises(ValueError):
            replace_step_outputs(
                db,
                "step-1",
                "tenant-a",
                "user-1",
                [
                    {
                        "id": "out-other",
                        "output_type_id": "type-valid",
                        "output_name": "细胞悬液",
                        "output_barcode": "NK-UBMC202603009",
                        "remark": "",
                    }
                ],
            )
        other = db.query(BizProductionTaskStepOutput).filter_by(id="out-other").one()
        self.assertEqual(other.step_id, "step-2")
        self.assertEqual(other.del_flag, 0)


if __name__ == "__main__":
    unittest.main()